diff --git a/src/main/java/com/wok/supportbot/app/AssistantApp.java b/src/main/java/com/wok/supportbot/app/AssistantApp.java index 8be4ec6..de5f309 100644 --- a/src/main/java/com/wok/supportbot/app/AssistantApp.java +++ b/src/main/java/com/wok/supportbot/app/AssistantApp.java @@ -18,7 +18,9 @@ import org.springframework.ai.vectorstore.VectorStore; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; +import reactor.core.Disposable; import reactor.core.publisher.Flux; +import reactor.core.publisher.FluxSink; import java.util.ArrayList; import java.util.Collections; @@ -97,6 +99,9 @@ public class AssistantApp { private static final String SYSTEM_PROMPT = ""; + /** 尾部空白缓冲上限:超过后强制发出,避免纯空白输出导致 buffer 无界增长 */ + private static final int MAX_TRAILING_WHITESPACE_BUFFER = 256; + /** * 初始化 ChatClient * @@ -235,7 +240,8 @@ public class AssistantApp { } ChatRequest req = chatPipeline.buildRequest(ctx); if (req.faqHit()) { - // FAQ 命中:直接返回答案文本(不附加 SSE 协议行,Spring 会自动做 data: 帧包装) + // FAQ 命中:整段答案原样输出,由 SSE 编码器处理内部换行, + // 后端不做任何格式增删(不拆行、不加换行、不补空格)。 return Flux.just(req.faqAnswer().get()); } McpToolCallback.resetEvents(); @@ -249,7 +255,7 @@ public class AssistantApp { } // 原始文本流;推荐问题已不再由主回复同步生成,改由 SuggestionGenerator 异步按需生成 Flux rawStream = spec.stream().content(); - return rawStream + return preserveTrailingWhitespace(rawStream) .doOnComplete(() -> aiCircuitBreaker.recordSuccess(AI_CIRCUIT_KEY)) .doOnError(e -> { aiCircuitBreaker.recordFailure(AI_CIRCUIT_KEY); @@ -263,6 +269,62 @@ public class AssistantApp { .onErrorResume(e -> Flux.just("抱歉,AI 服务调用失败:" + e.getMessage())); } + /** + * 缓冲以空白字符结尾的 chunk,将其与下一个 chunk 合并后再发出。 + *

+ * 原因:前端 SSE 解析会对每行执行 trim(),如果 chunk 以空白结尾(如 Markdown 标题 "## "), + * 行尾空白会被削掉,导致 "## " + "一、..." 拼成 "##一、...",破坏 Markdown 渲染。 + * 本方法不改变大模型输出的文本内容,只调整 chunk 边界以避开 SSE 的空白截断。 + * + * @param source 原始大模型输出流 + * @return 调整后的流,每个元素均不以空白字符结尾(流末尾除外) + */ + private static Flux preserveTrailingWhitespace(Flux source) { + return Flux.create(sink -> { + StringBuilder buffer = new StringBuilder(); + // 捕获内部订阅,下游取消/释放时同步取消上游,避免资源泄漏 + Disposable subscription = source.subscribe( + chunk -> { + if (chunk == null) { + return; + } + buffer.append(chunk); + String current = buffer.toString(); + if (current.isEmpty()) { + return; + } + char last = current.charAt(current.length() - 1); + // 当前累积内容不以空白结尾,可以安全发出 + if (!Character.isWhitespace(last)) { + sink.next(current); + buffer.setLength(0); + } else if (buffer.length() >= MAX_TRAILING_WHITESPACE_BUFFER) { + // 尾部空白过长(如纯空白输出),强制发出以限制内存占用 + sink.next(current); + buffer.setLength(0); + } + // 若以空白结尾,继续缓存,等下一个 chunk + }, + e -> { + // 出错前先把已缓冲内容发出,避免已生成文本丢失 + if (buffer.length() > 0) { + sink.next(buffer.toString()); + buffer.setLength(0); + } + sink.error(e); + }, + () -> { + if (buffer.length() > 0) { + sink.next(buffer.toString()); + } + sink.complete(); + } + ); + sink.onCancel(subscription); + sink.onDispose(subscription); + }, FluxSink.OverflowStrategy.BUFFER); + } + /** * 统一检索引用来源(新入口,委托 {@link ChatPipeline#retrieveSources})。 *