From 5c75cb3f711b04b495ad2603c2fce58c6e305385 Mon Sep 17 00:00:00 2001 From: wanghanlin <1533525126@qq.com> Date: Thu, 24 Sep 2026 17:42:16 +0800 Subject: [PATCH] =?UTF-8?q?fix(ai):=20=E6=B5=81=E5=BC=8F=E5=9B=9E=E7=AD=94?= =?UTF-8?q?=E9=9B=B6=E5=86=85=E5=AE=B9=E6=97=B6=E4=B8=8D=E5=86=8D=E8=BF=94?= =?UTF-8?q?=E5=9B=9E=E7=A9=BA=E5=9B=9E=E7=AD=94=EF=BC=8C=E5=B9=B6=E6=A0=87?= =?UTF-8?q?=E8=AE=B0=20EMPTY=5FCOMPLETION?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 问题（与上一个提交同一个现场）：/ai/chat/stream 只发出「开场 role 帧 + finish_reason=stop 帧 + [DONE]」。这两个帧都是本地生成的、与模型无关（startWith / concatWith），所以这就是一个 **空回答**：客户端表现为一直停在加载态，而 llm_call_trace 把它记为 status=COMPLETE 且 error_type 为空，服务端全程没有告警痕迹 —— 线上持续 10 天、7/204 次，直到客户端报障才发现。 根因（数据库证据，非推断）：llm_call_trace 全库 204 条 COMPLETE 记录中，ai_response 为空的有 7 条，且这 7 条 **100% 都 completion_tokens >= max_tokens**（精确等于配置的 2000）。CHAT 活跃 模型 deepseek-v4-flash 会把推理 token 计入 completion_tokens；开启 RAG 后 prompt_tokens 涨到 1035~3557，推理阶段把 max_tokens=2000 的预算吃干，可见答案零 token 产出。Spring AI 只取 delta.content（推理内容进 metadata 的 reasoningContent，不拼进正文），于是上游分片全为空串， 被 preserveTrailingWhitespace 的 current.isEmpty() 分支静默吞掉（该分支无日志、无计数），最终 只剩本地生成的首尾帧。同一问题的同步兜底请求因 URL 不带 enableRag、prompt 仅 334 token， 预算有富余而返回了正常答案 —— 这也解释了「流式空、同步正常」的观感。 - chatStreamOpenAi 增加零内容兜底，帧序仍为 role → 内容… → 兜底内容 → stop → [DONE] （客户端解析器无需改动）：统计非空白内容分片数与上游原始分片数，内容分片为 0 时补一帧内容 ① 先复用同一 spec 做一次非流式调用（subscribeOn(boundedElastic)），成功则作为内容帧发出， 用户完全无感 —— 覆盖「provider 流式分片不带内容、但非流式正常」这类成因； ② 非流式也为空则发一条明确提示 —— 覆盖「预算被推理吃光」这类成因：非流式走的是同一套预算与 同样的 prompt，同样会返回空，所以两级都不可省。 - 可观测性：兜底发生时打 WARN（含上游分片数、内容分片数、兜底来源），并在 llm_call_trace 记 error_type=EMPTY_COMPLETION，后台「提示词追踪」可直接筛出这类记录。 - MyLoggerAdvisor：修 after() 的真实 NPE —— getResult() 在 generations 为空时返回 null （provider 在流末下发 "choices":[] 的用量分片即命中），相邻的 ContentSafetyAdvisor 判了空、 它没判，会抛 NPE 打断整条 advisor 链；同时补注释说明「流式下 AI Response: 为空是必然现象」 （BaseAdvisor.adviseStream 只在携带 finish_reason 的收尾分片回调 after，那片 delta 内容天然 为空），避免下次排查被这行日志误导。 - README：修正流格式契约 —— data: 后**无空格**（原文档写作 data: {...}，按文档实现的自研解析器 会一个字节都取不到），并补充首帧只含 delta.role、末帧 delta 为空对象 + finish_reason=stop、 全链路不发任何 event: 具名事件这三个特征。 注意：本提交不改变「推理吃光 max_tokens 预算」这一根因，只把「静默的空回答」变成「可见、不破坏 体验的失败」。真正恢复答案需要把 CHAT 的 max_tokens 从 2000 调大（配置层，另行处理）。 验证：mvn clean package -P prod 通过；编译产物已确认含 EMPTY_COMPLETION 与提示文案；用真实 SDK 回归新帧序 —— 兜底内容被正常渲染、无 stream_empty、客户端不再触发自身兜底（syncFallbackCalls=0）。 --- README.md | 2 +- .../supportbot/advisor/MyLoggerAdvisor.java | 17 ++- .../com/wok/supportbot/app/AssistantApp.java | 104 +++++++++++++++++- 3 files changed, 118 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 8680556..d6c234f 100644 --- a/README.md +++ b/README.md @@ -114,7 +114,7 @@ CREATE TABLE knowledge_document ( | 方法 | 路径 | 说明 | |------|------|------| | GET | `/ai/chat` | 同步对话（`enableRag=true` 时走 RAG 增强） | -| GET | `/ai/chat/stream` | SSE 流式对话（OpenAI Chat Completions 兼容格式，输出 `data: {"choices":[{"delta":{"content":"..."}}]}` 分片，结束 `data: [DONE]`） | +| GET | `/ai/chat/stream` | SSE 流式对话（OpenAI Chat Completions 兼容格式）。帧格式 `data:{"choices":[{"delta":{"content":"..."}}]}` —— **`data:` 后无空格**；首帧只含 `delta.role`（无 content，用于尽早送出首字节），末帧 `delta` 为空对象且 `finish_reason:"stop"`，结束帧 `data:[DONE]`。全链路不发任何 `event:` 具名事件 | | GET | `/ai/chat/sources` | RAG 引用来源 | | GET | `/ai/product_info_app/chat/sync` | 商品信息结构化提取 | diff --git a/src/main/java/com/wok/supportbot/advisor/MyLoggerAdvisor.java b/src/main/java/com/wok/supportbot/advisor/MyLoggerAdvisor.java index d3167ed..e4e76d1 100644 --- a/src/main/java/com/wok/supportbot/advisor/MyLoggerAdvisor.java +++ b/src/main/java/com/wok/supportbot/advisor/MyLoggerAdvisor.java @@ -10,6 +10,13 @@ import org.springframework.ai.chat.prompt.Prompt; /** * 自定义日志 Advisor（适配 Spring AI 1.0.1 新 Advisor API） * 打印 info 级别日志、只输出单次用户提示词和 AI 回复的文本 + * + *

排查提醒（流式下的日志语义）：Spring AI 的 {@code BaseAdvisor.adviseStream} 默认实现只在 + * 「携带 finish_reason 的分片」上回调 {@link #after}，而 OpenAI 标准把 finish_reason 放在独立的收尾分片里、 + * 其 delta 内容天然为空。因此**流式请求打印出 {@code AI Response: }（冒号后为空）属正常现象**， + * 绝不能据此判断「模型没有输出」或「回答为空」。 + * 判断流式回答的真实内容，请查 {@code llm_call_trace.ai_response}（聚合全文）及该表的 + * {@code status} / {@code error_type}（零内容会被标记 {@code EMPTY_COMPLETION}）。 */ @Slf4j public class MyLoggerAdvisor implements BaseAdvisor { @@ -36,7 +43,15 @@ public class MyLoggerAdvisor implements BaseAdvisor { @Override public ChatClientResponse after(ChatClientResponse response, AdvisorChain chain) { - String text = response.chatResponse().getResult().getOutput().getText(); + // 流式下本方法只拿到「收尾分片」，text 恒为空，见类注释 —— 不要用这行判断回答内容。 + // 必须逐级判空：getResult() 在 generations 为空时返回 null（例如 provider 在流末下发 + // "choices": [] 的用量分片），不判会在此抛 NPE 并打断整条 advisor 链。 + String text = null; + if (response != null && response.chatResponse() != null + && response.chatResponse().getResult() != null + && response.chatResponse().getResult().getOutput() != null) { + text = response.chatResponse().getResult().getOutput().getText(); + } log.info("AI Response: {}", text); return response; } diff --git a/src/main/java/com/wok/supportbot/app/AssistantApp.java b/src/main/java/com/wok/supportbot/app/AssistantApp.java index a3a4010..8a1b758 100644 --- a/src/main/java/com/wok/supportbot/app/AssistantApp.java +++ b/src/main/java/com/wok/supportbot/app/AssistantApp.java @@ -36,6 +36,7 @@ import reactor.core.Disposable; import reactor.core.publisher.Flux; import reactor.core.publisher.FluxSink; import reactor.core.publisher.SignalType; +import reactor.core.scheduler.Schedulers; import java.util.ArrayList; import java.util.Collections; @@ -44,7 +45,9 @@ import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import static org.springframework.ai.chat.memory.ChatMemory.CONVERSATION_ID; @@ -125,6 +128,13 @@ public class AssistantApp { /** AI 熔断时的降级提示语 */ private static final String CIRCUIT_OPEN_MESSAGE = "AI 服务暂时不可用，请稍后重试。"; + /** + * 流式零内容、且非流式重试同样为空时给用户的明确提示。 + * 兜底目的：避免「空回答」，否则客户端会一直停在加载态（SDK 需再发一次同步请求才可能出答案）。 + */ + private static final String ZERO_CONTENT_NOTICE = + "抱歉，本次未能生成有效回答（可能受模型输出长度限制影响）。请重试一次，或联系管理员检查模型配置。"; + private static final String SYSTEM_PROMPT = ""; /** 尾部空白缓冲上限：超过后强制发出，避免纯空白输出导致 buffer 无界增长 */ @@ -518,9 +528,33 @@ public class AssistantApp { }); // 聚合所有分片用于埋点（在 doFinally 时取完整回复文本） StringBuilder aggregated = new StringBuilder(); - return preserveTrailingWhitespace(rawStream) - .doOnNext(aggregated::append) + // 内容分片计数：只有「非空白文本」才算内容，用于识别整条流零内容。 + // 典型成因：推理模型把 max_tokens 预算耗在 reasoning 上，delta.content 全为空， + // 被 preserveTrailingWhitespace 静默吸收（该方法的空串分支没有任何日志与计数）。 + // rawChunks 记录上游原始分片数，用于区分「上游一个分片都没有」与「有分片但内容全空」。 + AtomicLong rawChunks = new AtomicLong(); + AtomicLong contentChunks = new AtomicLong(); + AtomicBoolean emptyCompletion = new AtomicBoolean(false); + AtomicReference emptyFallbackKind = new AtomicReference<>(); + // spec 在上面被多次重新赋值，不是 effectively final，无法直接在 lambda 中引用，故取最终引用 + ChatClient.ChatClientRequestSpec fallbackSpec = spec; + return preserveTrailingWhitespace(rawStream.doOnNext(x -> rawChunks.incrementAndGet())) + .doOnNext(chunk -> { + aggregated.append(chunk); + if (!chunk.isBlank()) { + contentChunks.incrementAndGet(); + } + }) .map(chunk -> buildOpenAiChunk(completionId, model, created, chunk, false, null)) + // 零内容兜底：见 zeroContentFallback 的注释。必须在 concatWith(stop, [DONE]) 之前， + // 最终帧序保持「role → 内容… → 兜底内容 → stop → [DONE]」，客户端解析器无需改动。 + .concatWith(Flux.defer(() -> { + if (contentChunks.get() > 0) { + return Flux.empty(); + } + return zeroContentFallback(completionId, model, created, fallbackSpec, ctx, + emptyCompletion, emptyFallbackKind, rawChunks.get()); + })) .doOnComplete(() -> aiCircuitBreaker.recordSuccess(AI_CIRCUIT_KEY)) .doOnError(e -> { aiCircuitBreaker.recordFailure(AI_CIRCUIT_KEY); @@ -533,8 +567,17 @@ public class AssistantApp { String status = signalType == SignalType.ON_COMPLETE ? "COMPLETE" : signalType == SignalType.ON_ERROR ? "ERROR" : "CANCEL"; Usage usage = usageRef.get(); + // 发生零内容兜底时补记错误类型与说明，让后台「提示词追踪」能直接筛出这类记录。 + // 真实异常优先于兜底标记：errorTypeRef 非空说明上游确实报错了。 + String traceErrorType = errorTypeRef.get(); + String traceErrorMessage = errorMessageRef.get(); + if (traceErrorType == null && emptyCompletion.get()) { + traceErrorType = "EMPTY_COMPLETION"; + traceErrorMessage = "流式响应零内容（上游分片 " + rawChunks.get() + + " 个、内容分片 0 个），已兜底为 " + emptyFallbackKind.get(); + } recordTrace(ctx, req, aggregated.toString(), elapsedMillis(startNanos), status, - new TraceMeta(errorTypeRef.get(), errorMessageRef.get(), + new TraceMeta(traceErrorType, traceErrorMessage, usage != null ? usage.getPromptTokens() : null, usage != null ? usage.getCompletionTokens() : null, usage != null ? usage.getTotalTokens() : null, @@ -612,6 +655,61 @@ public class AssistantApp { "[DONE]"); } + /** + * 流式「零内容」兜底：整条流的非空内容分片为 0 时调用。 + *

+ * 若不兜底，调用方只会收到「开场 role 分片 + finish_reason=stop 分片 + [DONE]」——注意这两个分片 + * 都是本地生成的、与模型无关，所以这就是一个**空回答**：客户端表现为一直停在加载态 + * （SDK 只能靠再发一次同步请求才可能拿到答案），而 {@code llm_call_trace} 会把它记为 + * {@code status=COMPLETE} 且 {@code error_type} 为空，服务端全程没有告警痕迹 + * （线上曾持续 10 天、7/204 次，直到客户端报障才发现）。 + *

+ * 两级兜底： + *

    + *
  1. 先试一次非流式调用（复用同一个 spec 的 messages/system/toolContext）：覆盖 + * 「provider 流式分片不带内容、但非流式正常」这类问题；成功则把完整答案作为一条内容分片发出， + * 用户完全无感。
    2. + *
  2. 非流式也为空：发出一条明确提示的内容分片，保证客户端拿到的一定是非空内容。
    3. + *
+ * 注意不能只做第 1 级：像「推理 token 吃光 max_tokens 预算」这类成因，非流式走的是同一套 + * 预算与同样的 prompt，同样会返回空 —— 两级都不可省。 + * + * @param id chunk 唯一 ID + * @param spec 原请求的 spec，用于非流式重试 + * @param ctx 对话上下文（仅用于日志） + * @param emptyCompletion 输出参数：已触发零内容兜底，供埋点标记 {@code EMPTY_COMPLETION} + * @param fallbackKind 输出参数：兜底来源描述，供埋点与日志 + * @param upstreamChunks 上游原始分片数，用于区分「上游一个分片都没有」与「有分片但内容全空」 + * @return 只含一条内容分片的流（stop 与 [DONE] 由调用链后续统一追加） + */ + private Flux zeroContentFallback(String id, String model, long created, + ChatClient.ChatClientRequestSpec spec, ChatContext ctx, + AtomicBoolean emptyCompletion, + AtomicReference fallbackKind, + long upstreamChunks) { + emptyCompletion.set(true); + // subscribeOn(boundedElastic)：非流式调用是阻塞的，不能占用当前事件线程 + return Flux.defer(() -> { + String text = null; + try { + text = spec.call().content(); + } catch (Exception e) { + log.warn("零内容兜底的非流式重试失败: chatId={}, error={}", ctx.chatId(), e.getMessage()); + } + String kind; + if (StringUtils.hasText(text)) { + kind = "非流式回答"; + } else { + kind = "明确提示（非流式重试同样为空）"; + text = ZERO_CONTENT_NOTICE; + } + fallbackKind.set(kind); + log.warn("流式响应零内容，已兜底: chatId={}, model={}, 上游分片={}, 内容分片=0, 兜底={}", + ctx.chatId(), model, upstreamChunks, kind); + return Flux.just(buildOpenAiChunk(id, model, created, text, false, null)); + }).subscribeOn(Schedulers.boundedElastic()); + } + /** * 缓冲以空白字符结尾的 chunk，将其与下一个 chunk 合并后再发出。 *