Spring AI 流式輸出：stream 與 SseEmitter

一句話先說結論：流式輸出靠 .stream() 拿到 Flux，MVC 裡用 SseEmitter 訂閱推送，WebFlux 裡直接返回 Flux。最隱蔽的坑是“看起來是流式、實則一次性”——某些版本（2.0.0-M5/M6）的 OpenAI stream() 內部把整條流 collectList() 了再發，直接退化成非流式；另外 SseEmitter 不設超時會掛死，produces 設定漏了會導致連接提前關閉。

背景

LLM 生成一句話可能要幾秒，.call() 會等你拿到完整答案才返回，用戶體驗是“轉圈幾秒後一整塊文字蹦出來”。.stream() 改為把生成的每個 token 都即時吐出來，用戶幾百毫秒就開始讀。總耗時其實沒變，但“首 token 時間”大幅提前——這是整個功能的核心價值。

現象

SseEmitter 推流時最常見的報錯原文是寫流失敗：

org.springframework.web.context.request.async.AsyncRequestNotUsableException:
    ServletOutputStream failed to write: java.io.IOException: Broken pipe

或者客戶端中途斷開、服務端還在往已關閉的連接上 send：

java.io.IOException: 你的主机中的软件中止了一个已建立的连接。

帶 reactor 訂閱回調的完整堆疊長這樣，最內層是你自己的 lambda：

java.io.IOException: Broken pipe
    at com.example.controller.ChatController.lambda$stream$0(ChatController.java:38)
    at reactor.core.publisher.LambdaSubscriber.onNext(LambdaSubscriber.java:160)
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:114)
    at reactor.core.publisher.FluxReplay$SizeBoundReplayBuffer.replayNormal(FluxReplay.java:1061)
    at reactor.core.publisher.FluxReplay$ReplaySubscriber.onNext(FluxReplay.java:425)
Caused by: org.apache.catalina.connector.ClientAbortException: java.io.IOException: Broken pipe
    at org.springframework.web.servlet.mvc.method.annotation.SseEmitter.send(SseEmitter.java:180)
    at org.apache.catalina.connector.OutputBuffer.realWriteBytes(OutputBuffer.java:354)
    at org.apache.catalina.connector.OutputBuffer.flushByteBuffer(OutputBuffer.java:829)
    at org.apache.catalina.connector.OutputBuffer.append(OutputBuffer.java:752)
    at org.apache.catalina.connector.Response.doWrite(Response.java:654)

流到一半突然停住、沒有任何例外拋出的情況，則是 publishOn 缺 prefetch 導致的 demand 耗盡，日誌裡往往乾淨得讓你無從下手——只能靠現象反推。

基本用法

流式最簡一行，用 .stream() 替代 .call()：

Flux<String> tokens = chatClient.prompt()
    .user("给我讲个笑话")
    .stream()
    .content();          // Flux<String>，文本增量

stream() 提供三種返回：

返回類型	內容
`Flux<String> content()`	文本增量（95% 的場景夠用）
`Flux<ChatResponse> chatResponse()`	每塊響應 + 元數據
`Flux<ChatClientResponse> chatClientResponse()`	帶執行上下文，能看到 advisor/RAG 檢索了啥

Spring MVC：SseEmitter

@GetMapping(path = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter stream(@RequestParam String message) {
    SseEmitter emitter = new SseEmitter(120_000L);   // 显式超时

    Flux<String> tokens = chatClient.prompt().user(message).stream().content();

    tokens.subscribe(
        token -> {
            try {
                emitter.send(token);
            } catch (IOException e) {
                emitter.completeWithError(e);   // 客户端已断开，别硬发
            }
        },
        emitter::completeWithError,
        emitter::complete);

    return emitter;
}

Spring WebFlux：直接返回 Flux

@GetMapping(value = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> stream(@RequestParam String message) {
    return chatClient.prompt().user(message).stream().content();
}

根因分析

坑 1：stream 被內部緩衝，退化成一次性返回

Spring AI 2.0.0-M5/M6 的 OpenAiChatModel.internalStream() 一度為了判斷“有沒有 tool calls”，先 collectList() 把整條流收齊了再發。結果是：訂閱者啥都收不到，直到模型生成完才一次性吐一批——流式名存實亡（issue #5987 / #6183，已在後續版本修復）。升級到 2.0 早期的同學很容易撞上，表現就是“加了 stream 還是整塊輸出”。

坑 2：SseEmitter 不設超時掛死

new SseEmitter() 預設超時很長，客戶端中途斷開、服務端又一直等，servlet 執行緒就被佔住。務必顯式給超時（如上 120_000L）。

坑 3：SSE 設定缺失導致連接提前關

MVC 裡必須 produces = MediaType.TEXT_EVENT_STREAM_VALUE；少了它，或者返回類型/狀態碼不對（比如預設 204），瀏覽器會立刻斷掉 SSE 連接。前端就會“連一下就沒下文”。

坑 4：advisor 的 publishOn 無 prefetch 導致凍結

某些帶 ChatModelStreamAdvisor 的場景，publishOn(Schedulers.boundedElastic()) 沒設 prefetch（預設 256），流吐滿 256 個元素後會因為 demand 耗盡而凍結（issue #5651）。自測時碰“流到中途突然停”，可以往這個方向查。

排查與防線

前端用瀏覽器內建 EventSource 時注意：它只能發 GET、不能自定義 header 或 POST body。要 POST 或帶 token，用 fetch + 手動解析 SSE 流。
EventSource 在服務端正常結束時也會觸發 onerror（收到關閉事件），處理完記得 source.close()，別當成真錯誤。
走 Nginx 要加 X-Accel-Buffering: no，否則反向代理會把流緩衝起來，又變回一次性。
長回答中間留白、以及 reasoning 模型“先想再答”的首 token 延遲，容易和“流斷了”混淆，必要時加心跳 Flux.interval(...) 保活。

小結

.stream().content() 拿 Flux<String>，MVC 用 SseEmitter、WebFlux 直接返 Flux。
流式“沒流起來”先想兩件事：是不是撞上了 2.0 早期的 stream 緩衝 bug；produces 和超時配沒配。
SseEmitter 一定顯式超時；前端 EventSource 只能 GET，要 POST 用 fetch。
反向代理記得關緩衝，否則前面全白做。