别再只用replaceAll了！手把手教你用Reactor的scan操作符处理AI流式响应的跨分块标签

别再只用replaceAll了！手把手教你用Reactor的scan操作符处理AI流式响应的跨分块标签

在构建现代AI对话系统时，开发者常常会遇到一个棘手的问题：如何处理流式响应中被拆分的结构化标签。想象这样一个场景——当AI返回包含<think>标签的响应时（这类标签通常用于隐藏AI的推理过程），如果标签内容恰好被分割到不同的数据块中，简单的字符串替换就会彻底失效。这正是本文要解决的核心痛点。

1. 为什么传统字符串处理会失败

当AI服务以流式方式返回响应时，数据会被拆分成多个小块（chunk）逐步传输。这种机制虽然能显著提升用户体验，却给标签处理带来了独特挑战。假设我们收到以下两个连续的数据块：

第一个chunk: "人类可以<think>这是" 第二个chunk: "一个测试</think>理解吗？"

使用replaceAll("<think>.*</think>", "")这样的正则表达式会完全失效，因为：

1. 跨块匹配问题：正则表达式无法跨多个chunk进行匹配
2. 状态丢失：简单的字符串处理无法记住前一个chunk是否开启了标签
3. 内容截断风险：直接替换可能导致有效内容被意外删除

更糟糕的是，这类问题往往在测试中难以发现，因为标签拆分取决于网络条件和响应内容长度，属于典型的"边界条件"问题。

2. Reactor的scan操作符与状态机模式

2.1 scan操作符的核心优势

在响应式编程中，scan操作符提供了一种优雅的解决方案。与map不同，scan会维护一个累积状态，对每个输入元素应用转换函数后，既产生输出又更新状态。这种特性使其特别适合处理需要"记忆"的场景。

Flux<String> processed = rawFlux.scan(initialState, (state, chunk) -> { // 处理当前chunk并返回新状态 return processChunk(state, chunk); });

对比常见操作符：

2.2 状态机设计模式

状态机是解决这类问题的经典模式。对于标签处理，我们可以定义两种基本状态：

1. NORMAL：正常处理文本内容
2. INSIDE_TAG：处于标签内部（如<think>和</think>之间）

状态转换规则如下：

NORMAL -> 遇到<think> -> INSIDE_TAG INSIDE_TAG -> 遇到</think> -> NORMAL

提示：状态机应该设计为不可变对象，确保线程安全

3. 完整实现方案

3.1 定义状态对象

首先创建一个专门处理标签状态的对象：

public class TagProcessorState { private final boolean insideTag; private final StringBuilder buffer; // 初始化方法等... public TagProcessorState process(String chunk) { StringBuilder output = new StringBuilder(); String remaining = chunk; while (!remaining.isEmpty()) { if (!insideTag) { int tagStart = remaining.indexOf("<think>"); if (tagStart == -1) { output.append(remaining); break; } output.append(remaining.substring(0, tagStart)); remaining = remaining.substring(tagStart + "<think>".length()); insideTag = true; } else { int tagEnd = remaining.indexOf("</think>"); if (tagEnd == -1) break; remaining = remaining.substring(tagEnd + "</think>".length()); insideTag = false; } } return new TagProcessorState(insideTag, buffer.append(output)); } }

3.2 构建处理管道

将状态机与scan操作符结合：

Flux<ResponseDTO> processedStream = aiService.streamRequest(prompt) .scan(new TagProcessorState(false, new StringBuilder()), (state, response) -> state.process(response.getText())) .filter(state -> !state.getContent().isEmpty()) .map(state -> { ResponseDTO dto = new ResponseDTO(); dto.setText(state.getContent()); return dto; });

关键点说明：

1. 初始状态：创建未进入标签的状态实例
2. 状态传递：每个chunk处理后会返回新状态
3. 内容过滤：过滤掉空内容避免无效传输
4. 结果转换：将处理后的状态转换为响应DTO

4. 高级应用与优化

4.1 处理多种标签类型

当系统需要处理多种标签（如<think>、<reasoning>等）时，可以扩展状态机：

enum TagType { THINK, REASONING, NONE } class MultiTagState { private TagType currentTag; private Map<TagType, StringBuilder> buffers; // 处理方法需要根据不同类型标签进行调整 }

4.2 性能优化技巧

1. 缓冲区复用：避免频繁创建StringBuilder
2. 索引优化：使用KMP算法加速标签查找
3. 并行处理：对于CPU密集型操作可使用parallel()

.parallel() .runOn(Schedulers.parallel()) .scan(...) .sequential()

4.3 错误处理策略

为增强鲁棒性，应该添加：

1. 标签嵌套检测：防止<think><think>这样的错误
2. 未闭合标签超时：长时间未闭合的标签应自动关闭
3. 异常恢复：在错误状态后能恢复正常处理

5. 替代方案对比

虽然scan+状态机组合功能强大，但也要根据场景选择合适方案：

在实际项目中，我通常会先评估标签出现的频率和拆分概率。对于低频出现的标签，有时简单的服务端预处理反而是更经济的方案。但对于核心业务场景，scan+状态机的可靠性往往值得额外的实现成本。