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. + *
  3. 非流式也为空:发出一条明确提示的内容分片,保证客户端拿到的一定是非空内容。
  4. + *
+ * 注意不能只做第 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 合并后再发出。 *