一、概述

用过 ChatGPT 的都体验过"一个字一个字往外蹦"的感觉——相比等上十几秒才看到一整段回复，流式输出让用户感觉"AI 已经在思考了"。

从产品角度讲，用户对 AI 响应的耐心阈值只有 2-3 秒。超过这个时间，用户会以为程序卡死了。流式响应的本质不是技术炫技，而是用时间换体验——让用户在等待的同时已经在消费内容。

但从技术角度看，在 Spring AI 中实现流式调用，跟同步调用完全是两套逻辑。Spring AI 从接口层就把流式拆分出来了。

本文将从头到尾拆解 Spring AI 流式调用的每一层源码，涵盖：

• StreamingChatModel 为什么是独立接口（而非 ChatModel 的 boolean 参数）
• DefaultChatClient 的流式链路是怎么串起来的
• OpenAiChatModel.stream() 的 filter/map 链做了什么防御
• OpenAiApi 底层如何用 WebClient 消费 SSE 事件流
• 背压（Backpressure）策略在 LLM 场景下的最佳实践
• 生产环境的超时、重试、降级三板斧

---

二、接口设计：为什么 StreamingChatModel 是单独的接口？

2.1 两种接口，两种范式

先看 Spring AI 定义的两套接口：

// 同步调用 —— 一次性返回完整响应
public interface ChatModel {
    ChatResponse call(Prompt prompt);
}

// 流式调用 —— 返回响应式流，逐个发射 token
public interface StreamingChatModel {
    Flux<ChatResponse> stream(Prompt prompt);
}

一个返回 ChatResponse（完整响应），一个返回 Flux<ChatResponse>（响应流）。注意返回类型是 Flux 而非 Mono——因为流式场景下会发射多个 chunk，每个 chunk 都是一个部分响应。

2.2 为什么不合二为一？

如果只有一个 ChatModel 接口，用 boolean 参数区分：

// 反模式：混在一起的写法
public interface ChatModel {
    // stream=true → Flux, stream=false → ChatResponse
    <T> T call(Prompt prompt, boolean stream);
}

这种设计有三个问题：

1. 泛型擦除：返回类型 T 运行时被擦除，调用方必须手动 cast，编译器帮不了你
2. 分支逻辑：实现类里全是 if (stream) { ... } else { ... }，每加一个逻辑要改两处
3. 接口隔离原则：调用方拿到的对象能调 call 也能调 stream，但有些厂商根本不支持流式

实际上，Spring AI 早期版本就是这么干的。后来社区反馈代码可读性太差，重构拆成了两个独立接口。拆完之后，ChatModel 的实现者不需要关心流式，StreamingChatModel 的实现者不需要关心同步，各管各的。

2.3 OpenAiChatModel 同时实现两个接口

public class OpenAiChatModel implements ChatModel, StreamingChatModel {

    @Override
    public ChatResponse call(Prompt prompt) {
        // 同步调用：RestClient 直连，拿到完整响应后返回
        ChatCompletionRequest request = createRequest(prompt, false);
        ResponseEntity<ChatCompletion> response = 
            openAiApi.chatCompletionEntity(request);
        return toChatResponse(response.getBody());
    }

    @Override
    public Flux<ChatResponse> stream(Prompt prompt) {
        // 流式调用：WebClient 直连，逐 chunk 发射
        ChatCompletionRequest request = createRequest(prompt, true);
        Flux<ChatCompletionChunk> chunks = 
            openAiApi.chatCompletionStream(request);
        return chunks
            .filter(this::isValidChunk)
            .map(this::toChatResponse);
    }
}

关键设计：StreamingChatModel 是可选接口。如果某个 AI 服务商不支持流式（比如一些本地部署的模型），直接不实现它。上层 ChatClient 会做能力检测，流式不可用时降级为同步调用。

这是一种典型的**接口隔离原则（ISP）**应用：不要强迫客户端依赖它不使用的方法。

---

三、调用链路：从 ChatClient 到底层 SSE

3.1 用户侧入口

chatClient.prompt()
    .user("用三句话介绍 Spring AI")
    .stream()
    .content()
    .subscribe(token -> System.out.print(token));

三行代码，底层发生了什么？逐层展开：

3.2 DefaultChatClient.content()：map 管道

// DefaultChatClient 的流式方法
@Override
public Flux<String> content() {
    Prompt prompt = buildPrompt();
    return chatModel.stream(prompt)       // Flux<ChatResponse>
        .map(ChatResponse::getResult)      // Flux<Generation>
        .map(Generation::getOutput)        // Flux<AssistantMessage>
        .map(AssistantMessage::getContent);// Flux<String>
}

三层 map 剥离：

步骤	输入类型	输出类型	含义
map(getResult)	ChatResponse	Generation	从响应中提取当前代（Generation）
map(getOutput)	Generation	AssistantMessage	从代中提取模型输出消息
map(getContent)	AssistantMessage	String	从消息中提取纯文本内容

代码看着简洁，但有一个坑：有些 SSE chunk 的 delta.content 是 null。比如 OpenAI 在工具调用（function calling）场景下，某些 chunk 只携带 tool_calls 信息，content 为空。如果下游消费逻辑不做判空，String 变成 null，UI 端可能直接崩。

Spring AI 在 toChatResponse 方法里做了 deltaContent == null ? "" : deltaContent 的防御，但如果你的下游还做二次处理，建议也加上判空。

3.3 OpenAiChatModel.stream()：核心转换逻辑

@Override
public Flux<ChatResponse> stream(Prompt prompt) {
    // 注意这里的 stream=false 参数——它指的是 prompt 的元数据标记，
    // 真正的 stream 标记在 API 层设置
    ChatCompletionRequest request = createRequest(prompt, false);
    
    // 底层 API 调用，返回的是原始 chunk 流
    Flux<ChatCompletionChunk> chunks = openAiApi.chatCompletionStream(request);
    
    return chunks
        // 防御一：过滤掉 choices 为 null 或空的 chunk
        .filter(chunk -> chunk.choices() != null && !chunk.choices().isEmpty())
        // 防御二：chunk → ChatResponse 转换（内部做了 null content 防御）
        .map(this::toChatResponse)
        // 防御三：错误日志，但不中断流
        .doOnError(e -> log.error("Streaming error: {}", e.getMessage()))
        // 防御四：流完成日志
        .doOnComplete(() -> log.debug("Stream completed successfully"));
}

四层防御机制：

• filter：过滤掉代表 [DONE] 的空 chunk
• map：转换期间做 null content 判空
• doOnError：错误不打穿流，打印日志后继续
• doOnComplete：流正常结束时打印调试信息

3.4 全景流程图

---

四、SSE 底层实现：WebClient 如何消费事件流

4.1 SSE 协议简介

SSE（Server-Sent Events）是一种基于 HTTP 的单向推送协议。客户端发起一个普通的 HTTP 请求，服务器保持连接打开，持续推送事件：

POST /v1/chat/completions HTTP/1.1
Content-Type: application/json
Accept: text/event-stream

{"model":"gpt-4","messages":[...],"stream":true}

← HTTP/1.1 200 OK
← Content-Type: text/event-stream

data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":"Spring"},"index":0}]}

data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":" AI"},"index":0}]}

data: {"id":"chatcmpl-xxx","choices":[{"delta":{"content":" 是"},"index":0}]}

data: [DONE]

关键细节：

• Content-Type: text/event-stream 告诉客户端这是 SSE 流
• 每个事件以 data: 开头，后跟 JSON 或纯文本
• 事件之间用两个换行符隔开
• data: [DONE] 是 OpenAI 的流结束标记

4.2 OpenAiApi 的 SSE 消费实现

public Flux<ChatCompletionChunk> chatCompletionStream(ChatCompletionRequest request) {
    // 核心一步：把 stream 设为 true
    request.setStream(true);
    
    return webClient.post()
        .uri("/v1/chat/completions")
        .contentType(MediaType.APPLICATION_JSON)
        .accept(MediaType.TEXT_EVENT_STREAM)  // 关键：告诉服务器返回 SSE
        .body(Mono.just(request), ChatCompletionRequest.class)
        .retrieve()
        .bodyToFlux(ChatCompletionChunk.class); // 关键：自动反序列化
}

同步调用用 RestClient，流式调用用 WebClient——这是 Spring AI 的双客户端策略。

为什么流式必须用 WebClient？因为 bodyToFlux() 是 Spring WebFlux 独有的能力。它底层做了三件事：

1. 读取 SSE 事件行：识别 data: 前缀，提取 JSON 内容
2. 反序列化：将每个 JSON chunk 反序列化为 ChatCompletionChunk 对象
3. 组合成 Flux：将反序列化后的对象逐个发射到响应式流中

RestClient 做不到这一点——它是同步阻塞的，必须等整个响应体返回才能处理。

4.3 Chunk 到 ChatResponse 的转换细节

private ChatResponse toChatResponse(ChatCompletionChunk chunk) {
    ChatCompletionChunk.ChunkChoice choice = chunk.choices().get(0);
    String deltaContent = choice.delta().content();
    
    // 关键判空：某些 chunk 的 delta content 是 null
    if (deltaContent == null) {
        deltaContent = "";
    }
    
    Generation generation = new Generation(new AssistantMessage(deltaContent));
    
    return new ChatResponse(
        List.of(generation),
        Map.of(
            "id", chunk.id(),
            "model", chunk.model(),
            "created", chunk.created()
        )
    );
}

什么情况下 delta.content() 会是 null？三种常见场景：

1. 流结束标记：[DONE] 被反序列化后，choices 数组为空（已被 filter 过滤）
2. Function Calling：模型决定调用工具时，delta 只包含 tool_calls，content 为 null
3. 角色切换：某些模型在切换 system/user/assistant 角色时，delta 只有 role 信息

4.4 同步 vs 流式对比表

维度	同步调用 (ChatModel)	流式调用 (StreamingChatModel)
返回类型	ChatResponse（单次）	Flux<ChatResponse>（多次）
HTTP 客户端	RestClient（阻塞）	WebClient（非阻塞）
反序列化	bodyToMono()	bodyToFlux()
首字节时间	等待全部生成完成	第一个 token 到达即可消费
内存占用	完整响应一次性加载	每个 chunk 即时消费，常驻内存小
适用场景	短回复、批量处理	长文本生成、聊天 UI、实时交互
线程模型	阻塞当前线程	事件循环，不阻塞

---

五、背压（Backpressure）：当消费跟不上生产

5.1 什么是背压

背压是响应式编程的核心概念。简单说：生产者发射数据的速度 > 消费者处理数据的速度，中间的差距需要一种机制来协调。

在 LLM 流式场景下：

• 生产者（OpenAI）：每秒生成约 20-50 个 token
• 消费者（你的代码）：每个 token 可能需要做数据存储、二次处理、推送到前端…

如果消费者处理每个 token 需要 50ms，而生产者每 20ms 发射一个 token，积压很快就会堆积。

5.2 Spring AI 的背压策略

.bodyToFlux(ChatCompletionChunk.class)
    .onBackpressureBuffer(256)    // 开 256 缓冲区
    .onBackpressureDrop(dropped -> {
        log.warn("Chunk dropped due to backpressure: {}", dropped);
    });

三种策略对比：

策略	行为	适用场景
onBackpressureBuffer(N)	在内存中缓存最多 N 个元素	消费者速度偶尔慢于生产者
onBackpressureDrop()	丢弃超出消费能力的数据	实时性要求高，可以丢数据
onBackpressureLatest()	只保留最新的一个元素	只关心最新状态，不关心中间过程

LLM 场景下应该用 Buffer 而不是 Drop。因为每个 token 都是流式文本的一部分，如果丢弃中间的 token，输出内容就会不完整，用户看到的就是"断句"。

5.3 缓冲区大小的选择

// 普通对话：256 够用
.onBackpressureBuffer(256)

// 长文本生成（数千 token）：需要更大缓冲区
.onBackpressureBuffer(1024)

如何判断自己的场景需要多大缓冲区？

缓冲区需求 ≈ Token生成速率(个/秒) × 单Token处理耗时(秒) × 安全系数(2-3)

例：50 tokens/s × 0.05s 处理耗时 × 3 安全系数 ≈ 8
→ 实际上 256 绰绰有余，不需要过度优化

实际经验：大多数场景 256 够了。但如果你遇到频繁的 BufferOverflowException，优先检查是不是消费者处理逻辑太重了——比如每个 token 都同步写数据库，这才是真正的瓶颈。

---

六、错误处理：流断了怎么办

6.1 流式调用的三种失败模式

模式	现象	典型原因
连接失败	流根本建立不起来	API Key 无效、网络不通、400 错误
中途断开	流跑了一半突然没数据了	网络抖动、服务端限流、Token 超限
超时	长时间没有任何 chunk	模型卡住、生成内容过长触发了超时

6.2 Spring AI 的三板斧

return openAiApi.chatCompletionStream(request)
    .filter(this::isValidChunk)
    .map(this::toChatResponse)
    
    // 第一板斧：超时控制
    .timeout(Duration.ofSeconds(60))
    
    // 第二板斧：临时性异常重试
    .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
        .filter(throwable -> throwable instanceof TransientException)
        .onRetryExhaustedThrow((retrySpec, retrySignal) -> 
            new StreamExhaustedException("Max retries reached")))
    
    // 第三板斧：优雅降级，不炸整个流
    .onErrorResume(e -> {
        log.error("Stream error, falling back to error signal");
        return Flux.just(createErrorResponse(e));
    });

超时（timeout）

为什么是 60 秒？因为 LLM 正常生成时，token 间隔只有几十到几百毫秒。超过 1 分钟没动静，基本可以判定连接异常。注意这个超时是 token 间超时而非总会话超时——只要模型持续在吐 token，就不会触发。

重试（retryWhen）

Retry.backoff(3, Duration.ofSeconds(1))

关键点：只重试 TransientException。比如网络抖动、临时限流（429）——这些重试有意义。但如果是 400（参数错误）或 401（认证失败），重试一万次也没用，应该在第一次就快速失败。

降级（onErrorResume）

.onErrorResume(e -> Flux.just(createErrorResponse(e)))

不抛异常，而是发射一个 error 信号给前端。用户体验就是：流突然断了，UI 显示"抱歉，连接中断，请重试"。如果直接抛异常，前端可能白屏或无响应，体验更差。

---

七、实战：流式 API 的前后端对接

7.1 后端 Controller

@RestController
@RequestMapping("/api/chat")
public class ChatController {

    private final ChatClient chatClient;

    public ChatController(ChatClient.Builder builder) {
        this.chatClient = builder.build();
    }

    /**
     * 流式对话接口
     * 返回 SSE 格式：text/event-stream
     */
    @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<String> chatStream(@RequestParam String message) {
        return chatClient.prompt()
            .user(message)
            .stream()
            .content()
            // 防御：过滤可能的 null token
            .filter(Objects::nonNull)
            // 可以在这里做额外的处理
            .doOnNext(token -> log.debug("Emitted token: {}", token))
            .doOnError(e -> log.error("Stream error for message: {}", message, e));
    }
}

关键是 produces = MediaType.TEXT_EVENT_STREAM_VALUE，告诉 Spring MVC 这个接口返回的是 SSE 流。浏览器和前端框架（如 axios、fetch）会自动按照 SSE 协议解析。

7.2 前端消费：EventSource vs Fetch

方式一：EventSource（最简单）

const eventSource = new EventSource('/api/chat/stream?message=你好');

eventSource.onmessage = (event) => {
    // 每个 token 追加到 DOM
    document.getElementById('output').textContent += event.data;
};

eventSource.onerror = () => {
    // EventSource 不支持自定义请求头，
    // 错误时自动重连，需要手动处理关闭
    eventSource.close();
    showErrorToast('连接中断，请重试');
};

优点：浏览器原生支持，自动重连。缺点：不支持 POST 请求，不支持自定义 Header，无法传 JSON body。

方式二：Fetch + ReadableStream（推荐）

async function chatStream(message) {
    const response = await fetch('/api/chat/stream?message=' + encodeURIComponent(message), {
        headers: { 'Accept': 'text/event-stream' }
    });
    
    const reader = response.body.getReader();
    const decoder = new TextDecoder();
    let buffer = '';
    
    while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        
        buffer += decoder.decode(value, { stream: true });
        
        // 按 SSE 格式解析 data: 行
        const lines = buffer.split('\n');
        buffer = lines.pop(); // 保留不完整的行
        
        for (const line of lines) {
            if (line.startsWith('data:')) {
                const data = line.slice(5).trim();
                if (data && data !== '[DONE]') {
                    appendToken(data); // 追加到 UI
                }
            }
        }
    }
}

优点：完全控制请求配置，支持 POST、自定义 Header、AbortController 取消。缺点：需要手动解析 SSE 格式。

7.3 生产环境注意事项

// 配置线程池以支持高并发流式请求
@Bean
public Scheduler streamScheduler() {
    return Schedulers.boundedElastic()
        .withQueueCapacity(500)
        .withThreadNamePrefix("stream-");
}

// 流式调用时指定调度器
@GetMapping("/chat/stream")
public Flux<String> chatStream(@RequestParam String message) {
    return chatClient.prompt()
        .user(message)
        .stream()
        .content()
        .subscribeOn(streamScheduler()); // 不阻塞主线程
}

---

八、总结：流式调用的设计启示

8.1 核心设计决策回顾

设计决策	为什么这样做	违背了会怎样
接口拆分（ChatModel vs StreamingChatModel）	接口隔离，各管各的	if-else 地狱，不支持的厂商被迫实现空方法
双客户端（RestClient + WebClient）	同步用阻塞，流式用响应式	流式用 RestClient 根本无法消费 SSE
filter 过滤空 chunk	SSE 结束标记 [DONE] 反序列化后 choices 为空	下游 NPE
delta content 判空	Function Calling 等场景 content 可能为 null	下游 NPE
onBackpressureBuffer	消费速度跟不上生产速度	数据丢失，流式输出断断续续
timeout + retry + onErrorResume	网络不稳定、服务端限流	流中断后前端无响应

8.2 总结

流式调用比同步多一个数量级的复杂度，但 Spring AI 通过接口拆分、响应式管道、防御式编程，把这种复杂度封装在框架内部。理解这些设计决策，比记住 API 重要得多。

---

参考

• Spring AI 官方文档 - ChatClient
• Spring WebFlux - Reactor Core
• SSE 协议规范 (W3C)

---

本文为 Spring AI 源码解析系列第五篇。系列文章导航：

• Spring AI 源码解析（一）：整体架构与设计理念
• Spring AI 源码解析（二）：ChatClient 的构建与配置
• Spring AI 源码解析（三）：Prompt 模板与消息系统
• Spring AI 源码解析（四）：Tool Calling 机制与实现
• Spring AI 源码解析（五）：流式响应与 SSE 源码深度实现（本文）