From 0205af7b72a4b17c3a35ce5fc1b221677cf6c4c6 Mon Sep 17 00:00:00 2001 From: wanghanlin <1533525126@qq.com> Date: Thu, 13 Aug 2026 09:57:57 +0800 Subject: [PATCH] =?UTF-8?q?fix(app):=20SSE=20=E6=B5=81=E5=BC=8F=E8=BE=93?= =?UTF-8?q?=E5=87=BA=E4=BF=9D=E7=95=99=E8=A1=8C=E5=B0=BE=E7=A9=BA=E7=99=BD?= =?UTF-8?q?=E9=81=BF=E5=85=8D=20Markdown=20=E6=A0=87=E9=A2=98=E6=8D=9F?= =?UTF-8?q?=E5=9D=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 preserveTrailingWhitespace 方法，缓冲以空白结尾的 chunk 与下一 chunk 合并后发出，规避前端 SSE 逐行 trim 导致的行尾空格丢失（如 "## " 标题标记被拼成 "##标题"）。 - 捕获内部订阅并注册 onCancel/onDispose，下游取消时同步取消上游，避免资源泄漏 - 尾部空白缓冲设 256 上限，防止纯空白输出导致 buffer 无界增长 - onError 先 flush 已缓冲内容再传播，避免已生成文本丢失 - 空白判定改用 Character.isWhitespace，对齐前端 trim 语义 --- .../com/wok/supportbot/app/AssistantApp.java | 66 ++++++++++++++++++- 1 file changed, 64 insertions(+), 2 deletions(-) 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}）。 *