聊聊SpringAI流式輸出的底層實現？

作者：磊哥 2025-04-24 00:00:00

人工智能

Spring AI 中的流式輸出有兩種實現，而通過查看這兩種流式輸出的實現源碼可知，Spring AI 中的流式輸出是通過 Reactor Streams 技術實現的，和 Spring WebFlux 的底層實現技術一樣。

在 Spring AI 中，流式輸出（Streaming Output）是一種逐步返回 AI 模型生成結果的技術，允許服務器將響應內容分批次實時傳輸給客戶端，而不是等待全部內容生成完畢后再一次性返回。

這種機制能顯著提升用戶體驗，尤其適用于大模型響應較慢的場景（如生成長文本或復雜推理結果）。

技術實現

在 Spring AI 中流式輸出的實現有以下兩種方式：

通過 ChatModel 實現流式輸出。
通過 ChatClient 實現流式輸出。

ChatModel 流式輸出

Spring AI 中的流式輸出實現非常簡單，使用 ChatModel 中的 stream 即可實現：

@RequestMapping(value = "/streamChat", produces = "text/event-stream")
public Flux<String> streamChat(@RequestParam(value = "msg") String msg) {
    return chatModel.stream(msg);
}

ChatClient 流式輸出

ChatClient 流式輸出實現也很簡單，也是調用 stream().content() 返回 Flux 對象即可：

@RequestMapping("/stream")
public Flux<String> stream(String question) {
    return chatClient.prompt(question)
            .stream()
            .content();
}

底層實現

那么問題來了流式輸出的底層實現究竟是啥呢？

根據以往的經驗我們知道，流式輸出的實現技術基本有兩種：

Spring MVC（Servlet）+ SSE 實現流式輸出。
Spring WebFlux Reactor 模型實現流式輸出。

SSE 介紹

SSE（Server-Sent Events）是一種允許服務器向瀏覽器或其他客戶端推送實時更新的技術。它是一種單向通信機制，服務器可以主動向客戶端發送數據，而客戶端無需頻繁輪詢服務器請求數據。SSE 是基于 HTTP 協議的，使用標準的 text/event-stream MIME 類型來傳輸數據。

SSE 主要特點

單向通信：SSE 僅支持服務器到客戶端的單向通信，客戶端不能向服務器發送消息。如果需要雙向通信，可以結合 WebSocket 或其他技術。
基于 HTTP：SSE 使用標準的 HTTP 協議，不需要額外的協議支持，因此兼容性較好。
自動重連：客戶端在連接中斷后會自動嘗試重新連接。
數據格式：SSE 數據以特定的格式發送，每條消息以 data: 開頭，以兩個換行符 \n\n 結尾。
事件類型：可以為每條消息指定事件類型，客戶端可以通過監聽特定事件類型來處理不同的消息。

Spring MVC（Spring Web）底層是基于 Servlet 實現的，它是使用 SseEmitter 技術實現 SSE 協議實現流式輸出的。

SseEmitter 基本用法

這里提供一個 SseEmitter 的簡單使用案例，實現流式輸出，讓大家更好的理解這個技術點：

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import java.io.IOException;

@RestController
public class SseDemoController {
    @GetMapping(value = "/sse-demo", produces = "text/event-stream")
    public SseEmitter streamData() {
        // 設置超時時間（單位：毫秒）
        SseEmitter emitter = new SseEmitter(30_000L); // 30秒超時

        // 異步任務模擬流式輸出
        new Thread(() -> {
            try {
                for (int i = 1; i <= 5; i++) {
                    String message = "第 " + i + " 條消息";
                    emitter.send(message);
                    Thread.sleep(1000); // 每秒發送一次
                }
                emitter.complete(); // 完成推送
            } catch (IOException | InterruptedException e) {
                emitter.completeWithError(e); // 異常處理
            }
        }).start();

        return emitter;
    }
}