# AI应用开发中的流式输出：从协议原理到工程实战的完整指南 > **本文目标**：让读完这篇的人能彻底搞懂——流式输出到底是什么、为什么 AI 应用离不开它、底层跑的是什么协议、在 Java 和 Python 里分别怎么落地。不是 API 文档搬运，是讲清楚"为什么"和"怎么做"。 ![image-20260714170729889.png](https://pic.code-nav.cn/post_picture/1609766978631761921/7nXzo6MLIQDIHXg6.webp) ## 一、从一个真实的痛点说起你有没有想过一个问题：为什么 ChatGPT 的回答是一个字一个字"蹦"出来的，而不是等个十秒钟"唰"一下全部弹出来？如果你觉得这只是"炫酷的视觉效果"，那你就想错了。这是一种被精心设计的工程选择，背后牵扯到用户体验、网络协议、服务器资源、AI 模型的工作方式——是一整套系统设计。我来讲一个真实场景。假设你用 AI 写一篇 3000 字的技术文档。大模型生成这段文字大约需要 15 秒。如果走传统的 HTTP 请求-响应模式，整个流程是这样的：用户点击"生成"按钮 → 浏览器发一个 HTTP 请求 → 服务器把请求转发给 AI 模型 → AI 模型吭哧吭哧算了 15 秒 → 把 3000 字完整结果一次性返回给服务器 → 服务器返回给浏览器 → 浏览器渲染。这 15 秒里，用户看到的是什么？一个转圈的 loading 动画。用户不知道系统是在工作还是卡死了，不知道还要等多久，也不知道结果是否会符合预期。在产品经理的眼里，这 15 秒叫做"用户流失窗口"——很大概率用户会不耐烦，刷新页面，甚至关掉网页。但大模型的工作方式其实不是这样的。大模型在生成文本时，是**一个 token 一个 token 地往外吐**的（你可以粗略地把 token 理解成一个词或几个字）。也就是说，在用户点击"生成"后的第 0.5 秒，模型的第一个词其实已经生成出来了；第 1 秒，第二个词出来了……到第 15 秒，最后一词出来了。传统 HTTP 模式把所有这些词攒在服务器端，等全部生成完了才一次性返回。**这意味着 14.5 秒的可用数据被白白浪费了**。而流式输出做的事情就是：模型每生成一个词，服务器就立刻把这个词推给浏览器，浏览器立刻渲染出来。用户看到的就是打字机一样的效果——文字一个一个地出现，就像有人在实时打字。 **用户从"干等 15 秒"变成了"0.5 秒就看到第一个字"**，体验上是从"等待"到"见证"的根本转变。这不是优化，这是质变。 ```mermaid graph LR subgraph "传统HTTP模式" A1[用户请求] --> A2[等待...] A2 --> A3["等待..."] A3 --> A4["等待... (15秒)"] A4 --> A5["一次性返回全部内容"] end subgraph "流式输出模式" B1[用户请求] --> B2["0.5s: 第一个词"] B2 --> B3["1.0s: 第二个词"] B3 --> B4["1.5s: 第三个词"] B4 --> B5["...持续输出..."] B5 --> B6["15s: 最后一个词"] end style A2 fill:#FFE0E0,stroke:#D32F2F style A3 fill:#FFE0E0,stroke:#D32F2F style A4 fill:#FFE0E0,stroke:#D32F2F style B2 fill:#E0F2E0,stroke:#388E3C style B3 fill:#E0F2E0,stroke:#388E3C style B4 fill:#E0F2E0,stroke:#388E3C ``` 好，问题清楚了。接下来的关键问题是：**技术上怎么实现"服务器持续往浏览器推数据"这件事？** 这就是接下来几章要讲的内容——各种网络通信协议的登场。 --- ## 二、通信协议全景图：谁能做流式输出？要实现"服务器持续推送数据给客户端"，有不止一种技术方案。在 AI 应用开发中，最常被讨论的有四种：HTTP 长轮询、SSE（Server-Sent Events）、WebSocket、gRPC 流。让我用通俗的方式一个一个讲清楚。 ### 2.1 HTTP：最老实的快递员 HTTP 是互联网最基础的协议，它的本质是"一问一答"——客户端发一个请求，服务器回一个响应，然后连接关闭。就像你打电话叫快递，快递员把东西送到你手上就走了，你不开门再喊一次，他不会再来。那 HTTP 能不能做"持续推送"呢？严格说不能，但有个变通方法——**长轮询（Long Polling）**。客户端发一个请求，服务器不急着回复，而是把连接"挂"在那儿，等有新数据了再回复。客户端收到回复后，立刻再发一个新请求，继续挂着等。这样循环往复，就模拟出了"服务器主动推送"的效果。打个比方：你不打电话叫快递了，而是直接搬到快递站门口坐着不走，快递员一有东西就给你。但你每隔几分钟就得重新搬一次凳子（重新发请求），每次搬凳子都要消耗体力（网络开销）。长轮询的问题很明显： - 每次轮询都有 HTTP 头开销，大量无效请求浪费带宽 - 服务器维护大量挂起的连接，内存压力大 - 数据有延迟——新数据来了，但客户端还没发新请求来接收 - 实现复杂，容易出错在 AI 流式输出的场景下，长轮询几乎不会被选择，因为大模型每个 token 之间的间隔可能只有几十毫秒，用长轮询意味着几十毫秒内就得重连一次，根本不现实。 ```mermaid sequenceDiagram participant C as 客户端 participant S as 服务器 Note over C,S: 长轮询模式 C->>S: HTTP请求1 Note over S: 挂起请求，等待数据... S-->>C: 响应1（有数据了） C->>S: HTTP请求2 Note over S: 挂起请求，等待数据... S-->>C: 响应2（有数据了） C->>S: HTTP请求3 Note over S: 挂起请求，等待数据... Note over C,S: 每次都需要重新建连接，开销大 ``` ### 2.2 SSE：专门为"服务器推"设计的协议 SSE 全称 Server-Sent Events，从名字就能看出来——这是专门为"服务器发送事件给客户端"设计的协议。如果说 HTTP 是"快递员送完就走"，那 SSE 就是"快递员在你家门口装了一个管子，有包裹就从管子里滑进来，你不用反复开门"。 SSE 的底层还是 HTTP——它就是 HTTP 的一种特殊用法。客户端发一个普通的 HTTP GET 请求，但带上了 `Accept: text/event-stream` 这个请求头，告诉服务器："我想要的是一个事件流，你别一次性返回完。" 服务器收到后，保持连接不断开，把响应的 `Content-Type` 设为 `text/event-stream`，然后就开始一段一段地往响应体里写数据。每写一段就是一个"事件"。 SSE 的数据格式极其简单，就是纯文本，每个事件用空行分隔： ``` data: 你好 data: ，我是 data: ChatGPT ``` 就这。没有复杂的二进制编码，没有 XML 包装，就是 `data:` 后面跟文本内容，然后空行结束一个事件。简单到不能再简单。而且 SSE 还有几个超棒的特性： 1. **自动重连**：浏览器内置了断线重连机制，连接断了会自动重新连上。 2. **事件ID**：每个事件可以带一个 `id` 字段，断线重连时浏览器会通过 `Last-Event-ID` 请求头告诉服务器"上次我收到了第几条，从下一条开始给我"。 3. **自定义事件类型**：可以给事件分类，客户端只监听感兴趣的事件类型。 4. **纯文本**：调试方便，用浏览器开发者工具就能看到原始数据流。 ```mermaid sequenceDiagram participant B as 浏览器 participant S as 服务器 B->>S: GET /chat?q=你好
Accept: text/event-stream Note over S: 保持连接不关闭 S-->>B: HTTP 200
Content-Type: text/event-stream S-->>B: data: 你好\n\n Note over B: 渲染"你好" S-->>B: data: ，我是\n\n Note over B: 追加"，我是" S-->>B: data: ChatGPT\n\n Note over B: 追加"ChatGPT" Note over S: 连接保持，可继续发送 Note over B: 断开时自动重连 ``` ### 2.3 WebSocket：全双工的"电话" WebSocket 是一个完全独立的协议（虽然握手时借用了 HTTP）。它的特点是**全双工通信**——服务器和客户端可以随时互发消息，就像两个人打电话，双方都能说话和听话。 SSE 是单向的（服务器→客户端），WebSocket 是双向的（服务器↔客户端）。这就像是 SSE 是"广播站往外发信号"，而 WebSocket 是"打电话互相对话"。 WebSocket 建立连接的过程是这样的：客户端先发一个 HTTP 请求，但带上 `Upgrade: websocket` 头，服务器如果同意"升级"，就返回 `101 Switching Protocols` 响应，之后这条 TCP 连接就从 HTTP 协议切换到了 WebSocket 协议，双方可以随时互发消息帧（frame）。 WebSocket 的优势是真正的全双工、低延迟、支持二进制数据。它的劣势是： - 协议比 SSE 复杂得多 - 需要单独的心跳保活机制（SSE 的 HTTP 连接天然有心跳） - 没有内置断线重连（需要自己实现） - 需要专门的消息格式设计（不像 SSE 的纯文本那么简单） - 在某些企业网络环境/代理服务器下可能被拦截 ```mermaid graph TB subgraph "SSE — 单向" direction TB S1[服务器] -->|持续推送事件| C1[客户端] C1 -.->|HTTP GET请求一次| S1 end subgraph "WebSocket — 全双工" direction TB S2[服务器] -->|随时推送消息| C2[客户端] C2 -->|随时发送消息| S2 end style S1 fill:#E3F2FD,stroke:#1565C0 style S2 fill:#E3F2FD,stroke:#1565C0 ``` ### 2.4 gRPC 流：二进制的高效通道 gRPC 是 Google 推出的高性能 RPC 框架，底层使用 HTTP/2 和 Protocol Buffers 序列化。gRPC 支持三种流模式：服务端流（Server Streaming）、客户端流（Client Streaming）、双向流（Bidirectional Streaming）。对于 AI 场景来说，最常用的是服务端流——客户端发一次请求，服务器持续返回多条响应消息。 gRPC 流的优势在于二进制序列化的极高效率和 HTTP/2 的多路复用。劣势在于：浏览器原生不支持，需要引入 gRPC-Web 代理层，对前端开发来说门槛较高；调试不如纯文本直观。 ### 2.5 四种方案横向对比 | 维度 | HTTP 长轮询 | SSE | WebSocket | gRPC 流 | |------|-----------|-----|-----------|---------| | **通信方向** | 单向（模拟） | 单向（服务器→客户端） | 双向 | 双向 | | **底层协议** | HTTP | HTTP | 独立协议（WS） | HTTP/2 | | **数据格式** | 任意 | 纯文本 | 任意（含二进制） | Protobuf（二进制） | | **断线重连** | 手动实现 | **浏览器内置** | 手动实现 | 手动实现 | | **浏览器原生支持** | ✅ | ✅（EventSource API） | ✅ | ❌（需 gRPC-Web） | | **连接数限制** | 每域名6个（HTTP/1.1） | 每域名6个（HTTP/1.1） | 无限制 | 多路复用 | | **适用场景** | 低频更新 | 服务器推送（日志/通知/AI流） | 实时双向（聊天室/游戏） | 微服务间通信 | | **实现复杂度** | 中 | **低** | 高 | 中高 | | **代理/防火墙友好** | ✅ | ✅（标准HTTP） | ⚠️（可能被拦） | ⚠️（需HTTP/2） | ```mermaid quadrantChart title "协议选择决策象限" x-axis "低复杂度" --> "高复杂度" y-axis "单向推送" --> "双向通信" quadrant-1 "适合双向实时通信" quadrant-2 "适合单向推送" quadrant-3 "简单单向场景" quadrant-4 "复杂双向场景" "HTTP长轮询": [0.25, 0.2] "SSE": [0.2, 0.3] "WebSocket": [0.75, 0.85] "gRPC流": [0.8, 0.7] ``` --- ## 三、为什么 SSE 在 AI 时代成了主流？看完上面的对比，你可能会想：WebSocket 功能最强，为什么不直接用 WebSocket？这个问题问得好。让我从三个维度来回答：**AI 场景的通信特征、SSE 的工程优势、以及生态层面的原因**。 ### 3.1 AI 对话的通信特征：天生就是单向的 AI 聊天的通信模式极其简单： 1. 用户发一句话（一次请求） 2. AI 一个字一个字地回复（持续推送） 3. 回复完了，等用户下一条消息你看出来了吗？**这是一个典型的"客户端发一次，服务器持续推"的单向通信模式。** 用户不需要在 AI 回复的过程中打断它、给它发消息。WebSocket 的全双工能力在这种场景下完全是用不上的——就像你给一台只会单向广播的收音机装了个双向对讲模块，多余。 SSE 的"服务器→客户端"单向推送模型与 AI 对话场景的匹配度是 100%。不需要双向通道，就不该为不需要的能力付出复杂度代价。 ### 3.2 SSE 的工程优势：简单就是王道 **部署友好性**：SSE 底层是标准 HTTP，任何反向代理（Nginx、Apache、CDN）都天然支持，不需要任何特殊配置。WebSocket 的握手过程需要代理支持 `Upgrade` 头，不少企业网络的代理会拦截或篡改这些非标准 HTTP 头。gRPC 则需要 HTTP/2 支持，不少老旧基础设施还停在 HTTP/1.1。SSE 的"标准 HTTP 身份"让它在部署层面畅通无阻。 **调试友好性**：SSE 的数据是纯文本，打开浏览器开发者工具的 Network 面板，选中那条 `text/event-stream` 类型的请求，就能看到原始数据一行行流过来。不需要任何专用工具。WebSocket 和 gRPC 的二进制帧让人头大。 **浏览器内置 API**：浏览器原生提供了 `EventSource` 对象来消费 SSE 流，几行代码就能跑。虽然后续我们会讲到 `fetch + ReadableStream` 的方式更灵活，但 `EventSource` 的简洁性让入门成本极低。 **断线重连内置**：AI 生成可能需要 30 秒甚至更长时间。网络一抖动，连接断了。WebSocket 断了你得自己写重连逻辑、自己记录上下文、自己恢复状态。SSE 的浏览器实现自动帮你重连，还会带上 `Last-Event-ID` 告诉服务器从哪里继续。 **基础设施兼容**：SSE 可以无缝穿越各种 CDN、负载均衡器、API 网关，不需要特殊配置。这在内网部署和云部署中都极为重要。 ```mermaid mindmap root((SSE 流行原因)) 场景匹配 AI对话是单向推送用户无需中途打断匹配度100% 工程优势基于标准HTTP 部署零障碍调试简单纯文本浏览器原生支持自动断线重连基础设施友好生态原因 OpenAI使用SSE Claude使用SSE 几乎所有LLM API用SSE 形成事实标准 Spring/FastAPI原生支持简单即正确不需要双向就不引入双向复杂度与需求匹配少一个故障点 ``` ### 3.3 生态决定：大厂都用 SSE OpenAI 的流式 API 用的是 SSE。Anthropic Claude 的流式 API 用的是 SSE。Google Gemini 的流式 API 用的也是 SSE。几乎所有主流大模型的流式接口都选择了 SSE 作为传输协议。这不是巧合。当 OpenAI 在 2022 年发布 ChatGPT 时，选择了 SSE 作为流式输出的传输方案，这个选择随后被整个行业效仿。各大模型的 SDK（Python SDK、Java SDK、JavaScript SDK）都内置了对 SSE 的解析支持。Spring AI、LangChain4j、LangChain Python 等框架全部围绕 SSE 构建了流式响应管道。 **事实标准一旦形成，后入场的玩家没有理由不用它**。用 SSE 意味着你的系统可以直接和所有主流 AI 服务商对接，不需要做协议适配。这就是生态的力量。 ### 3.4 一个朴素的设计原则在软件工程中，有一个被反复验证的原则：**用最简单的方案解决问题。** 如果你只需要服务器推送数据，就别用 WebSocket。WebSocket 的全双工能力是你的需求不需要的，但它带来的心跳维护、消息帧格式、自定义重连逻辑、代理穿透配置……每一个都是额外的复杂度和潜在的 bug 来源。 SSE 就像一把专用螺丝刀——形状对了，大小对了，拧就完了。WebSocket 像一把瑞士军刀——功能多，但拧一个螺丝的时候你用不上那些刀片和开瓶器，反而碍事。 --- ## 四、深入理解 SSE 协议前面说了 SSE 的好处，现在是时候真正钻进协议内部，看看它到底是怎么运作的。这一章我会讲得比较细，因为只有理解了协议的细节，后面写代码的时候你才知道每一行为什么那么写。 ### 4.1 SSE 的 HTTP 协商过程 SSE 的建立过程就是一个普通的 HTTP GET 请求，但有两个特殊之处： **请求侧**，浏览器发送的 HTTP 头： ``` GET /api/chat/stream?message=你好 HTTP/1.1 Host: api.example.com Accept: text/event-stream Cache-Control: no-cache ``` 关键点： - `Accept: text/event-stream` —— 告诉服务器我要的是事件流 - `Cache-Control: no-cache` —— 别给我缓存，我需要实时数据 **响应侧**，服务器返回的 HTTP 头： ``` HTTP/1.1 200 OK Content-Type: text/event-stream; charset=utf-8 Cache-Control: no-cache Connection: keep-alive Transfer-Encoding: chunked ``` 关键点： - `Content-Type: text/event-stream` —— 确认返回的是事件流 - `Cache-Control: no-cache` —— 不缓存 - `Connection: keep-alive` —— 保持 TCP 连接 - `Transfer-Encoding: chunked` —— 分块传输，不是一次性返回完 `Transfer-Encoding: chunked` 这一点很重要。HTTP 的传统模式需要服务器在响应头里声明 `Content-Length`（响应体多大），浏览器收到这么多字节就认为响应结束了。但 SSE 的特点就是"不知道总共多大、什么时候结束"，所以用 chunked 编码——数据一块一块地发，每块前面标明这块多大，最后一块用 `0\r\n\r\n` 表示结束。 ```mermaid sequenceDiagram participant B as 浏览器 participant S as 服务器 B->>S: GET /api/stream
Accept: text/event-stream
Cache-Control: no-cache Note over S: 准备响应头 S-->>B: HTTP/1.1 200 OK
Content-Type: text/event-stream
Transfer-Encoding: chunked Note over S: 生成第一块数据 S->>B: chunk: "data: 你好\n\n" Note over B: 收到第一个事件 Note over S: 生成第二块数据 S->>B: chunk: "data: ，我是\n\n" Note over B: 收到第二个事件 Note over S: 生成第三块数据 S->>B: chunk: "data: AI助手\n\n" Note over B: 收到第三个事件 Note over S: 数据发送完毕 S->>B: chunk: 0 (结束标记) Note over B: 连接关闭 ``` ### 4.2 SSE 的消息格式：简单到令人发指 SSE 的消息格式规范定义在 HTML5 标准中。一条 SSE 消息由若干个字段行组成，以一个空行（两个换行符 `\n\n`）结束。一个完整的事件长这样： ``` id: 42 event: token retry: 3000 data: {"text": "你好", "index": 0} ``` 逐行解释每个字段： - **`id:`** —— 事件 ID。客户端会记住最后收到的 ID，断线重连时通过 `Last-Event-ID` 请求头发给服务器，让服务器知道从哪里续传。这个在 AI 流式输出中特别有用——如果生到一半连接断了，重连后可以从断点继续。 - **`event:`** —— 事件类型。默认是 `message`，你可以自定义，比如 `token`、`thinking`、`done`。客户端可以只监听特定类型的事件。 - **`retry:`** —— 重连等待时间（毫秒）。告诉浏览器："如果连接断了，等这么多毫秒再重连。" 默认值通常是 3 秒。 - **`data:`** —— 实际数据。这是最重要的字段。可以是多行（多个 `data:` 行会被 `\n` 拼接），但最终是一个字符串。 **一个事件必须以空行结束**（即 `\n\n`）。浏览器看到空行才会把之前攒的数据当作一个完整事件处理。如果你忘了空行，浏览器会一直攒着不触发——这是新手最常踩的坑。还有几个特殊规则： 1. **冒号开头的行是注释**：以 `:` 开头的行被忽略，通常用来发送心跳（保持连接活跃）。比如 `: keep-alive`。 2. **`data:` 后面有一个空格**：规范的写法是 `data: 你好`，冒号后跟一个空格。但实际上浏览器对空格很宽容，有没有都能解析。 3. **多行 `data:` 自动拼接**： ``` data: 第一行 data: 第二行 ``` 客户端收到的 `event.data` 是 `"第一行\n第二行"`，中间用换行符拼接。 ### 4.3 SSE 事件类型在 AI 场景中的设计在 AI 流式输出中，我们通常需要区分不同类型的事件。比如大模型可能同时输出正文文本、思考过程、工具调用信息——这些不应该混在一起。SSE 的 `event:` 字段就是为此设计的。一个设计良好的 AI 流式响应长这样： ``` event: thinking data: {"content": "我需要先分析用户的问题..."} event: thinking data: {"content": "这个问题涉及到..."} event: token data: {"content": "根据"} event: token data: {"content": "您的描述"} event: tool_call data: {"name": "search", "arguments": "{\"query\": \"相关资料\"}"} event: tool_result data: {"name": "search", "result": "找到了3条相关结果"} event: token data: {"content": "，我找到了以下信息"} event: done data: {"totalTokens": 142, "finishReason": "stop"} ``` 这样前端可以根据 `event` 类型分别渲染——思考过程用灰色斜体、正文用正常黑色、工具调用用一个可展开的卡片。所有信息在同一条 SSE 连接上有序传输，互不干扰。 ```mermaid graph LR subgraph "SSE 事件流" E1["event: thinking
data: 思考内容..."] E2["event: thinking
data: 继续思考..."] E3["event: token
data: 正文片段1"] E4["event: tool_call
data: 工具调用信息"] E5["event: tool_result
data: 工具执行结果"] E6["event: token
data: 正文片段2"] E7["event: done
data: 完成信息"] end E1 --> E2 --> E3 --> E4 --> E5 --> E6 --> E7 subgraph "前端渲染" T1["💭 思考区
灰色斜体"] T2["📝 正文区
正常文本"] T3["🔧 工具区
可展开卡片"] T4["✅ 完成标记"] end E1 -.-> T1 E2 -.-> T1 E3 -.-> T2 E4 -.-> T3 E5 -.-> T3 E6 -.-> T2 E7 -.-> T4 ``` ### 4.4 SSE 的生命周期管理 SSE 连接的完整生命周期是这样的： ```mermaid stateDiagram-v2 [*] --> Connecting: 客户端发起请求 Connecting --> Connected: 收到200响应头 Connected --> Receiving: 开始接收数据 Receiving --> Receiving: 收到新事件 Receiving --> Error: 网络异常 Receiving --> Completed: 服务器关闭连接 Error --> Connecting: 自动重连(retry毫秒后) Connecting --> Connected: 重连成功 Connecting --> Failed: 重连失败超过限制 Completed --> [*] Failed --> [*] ``` 这里有几个关键点需要注意： **连接保持**：SSE 是长连接，服务器端需要确保这个连接不被中间的代理服务器超时关闭。通常的做法是定期发送注释行（`: heartbeat\n\n`）作为心跳。Nginx 默认的 `proxy_read_timeout` 是 60 秒，如果 60 秒内没有任何数据流过，Nginx 会断开连接。所以如果你的 AI 生成可能超过 60 秒（深度思考模型经常这样），一定要在服务器端加心跳。 **连接关闭**：当 AI 生成完毕后，服务器应该主动关闭连接。在 HTTP chunked 编码中，关闭就是发送最后的 `0\r\n\r\n` 标记。在 Spring Boot 中，当 `Flux` 发出 `onComplete` 信号时，框架会自动处理连接关闭。 **连接数限制**：HTTP/1.1 规定浏览器对同一域名的并发连接数限制是 6 个。如果你在一个页面里打开了 6 个 SSE 连接，第 7 个连接会被阻塞。HTTP/2 没有这个限制（多路复用），所以如果你可能有多条 SSE 连接，确保启用了 HTTP/2。 --- ## 五、最佳实践：Spring Boot 中的 SSE 流式输出理论讲完了，现在进入实操。这一章会从零开始，带你在一个 Spring Boot 项目中完整实现 AI 流式输出的后端。 ### 5.1 什么是 Flux？—— Reactor 核心概念在 Spring Boot 中实现 SSE 流式输出，你绕不开一个东西：**Flux**。它是 Project Reactor 框架的核心类，也是 Spring WebFlux 的基础。让我用一个故事来解释。想象你有一个水管。水管的一头连着水源（数据生产者），另一头连着水龙头（数据消费者）。 **同步模式**就像一个大水桶——水源把所有水灌满整个水桶，然后消费者一次性把水桶端走。你得等水桶装满才能喝到水。这就是传统的 `String` 返回模式——方法执行完毕，全部数据准备好，才返回。 **Flux 模式**就像一根水管——水源每产生一滴水就顺着管子流过来，消费者拧开水龙头就能接到水，不需要等管子装满。这就是流式——数据产生和消费是同时进行的，是"推"（Push）模型而非"拉"（Pull）模型。用更技术的话说： - `String` / `List` 是**同步容器**——你拿到它的时候，里面的数据已经全部就位了。 - `Flux` 是**异步数据流**——它代表的是"未来将会陆续产生的 0 到 N 个数据的序列"。你拿到 `Flux` 的时候，数据还没开始产生呢。当你订阅（subscribe）它的时候，数据才会开始流动。 `Flux` 和 `Mono` 是 Reactor 的两个核心类型： | 类型 | 含义 | 对比 | |------|------|------| | `Mono` | 0 或 1 个数据的异步序列 | 相当于 `CompletableFuture` 的增强版 | | `Flux` | 0 到 N 个数据的异步序列 | 相当于"异步的 `List`"，但元素是逐个产生的 | ```mermaid graph TB subgraph "同步模式 String" S1["调用方法"] --> S2["方法内部执行
等AI全部生成完"] S2 --> S3["返回完整字符串"] S3 --> S4["客户端拿到全部数据"] end subgraph "Flux 模式 Flux" F1["调用方法"] --> F2["立即返回 Flux 对象
数据还没产生"] F2 --> F3["客户端订阅 Flux"] F3 --> F4["AI每生成一个token
Flux发出一个元素"] F4 --> F5["客户端逐个接收
逐个渲染"] F5 --> F6["全部生成完
Flux发出 onComplete"] end style S2 fill:#FFE0E0,stroke:#D32F2F style F4 fill:#E0F2E0,stroke:#388E3C style F5 fill:#E0F2E0,stroke:#388E3C ``` **为什么 SSE 和 Flux 天生一对？** 因为它们在概念上是完全对应的。SSE 是"服务器持续推送事件"的协议，Flux 是"持续产生元素"的数据结构。Spring WebFlux 内置了对这两者配合的支持——当你的 Controller 方法返回 `Flux` 且指定 `produces = MediaType.TEXT_EVENT_STREAM_VALUE` 时，Spring 会自动把 Flux 的每个元素包装成一个 SSE 事件发给客户端。不需要你手写任何 SSE 格式代码。 ```mermaid graph LR subgraph "Spring Boot 内部" A["Controller 方法
返回 Flux"] B["Spring WebFlux
自动桥接"] C["SSE 编码器
text/event-stream"] end subgraph "网络传输" D["HTTP 响应
data: 第一个token\n\n
data: 第二个token\n\n
..."] end subgraph "浏览器" E["EventSource / fetch
接收SSE流"] F["逐个渲染"] end A --> B --> C --> D --> E --> F style B fill:#E8EAF6,stroke:#3F51B5 ``` ### 5.2 项目搭建：依赖与配置先创建一个 Spring Boot 项目，引入必要依赖： ```xml org.springframework.boot spring-boot-starter-webflux org.springframework.ai spring-ai-starter-model-openai ``` > **为什么用 WebFlux 而不是传统 MVC？** 传统的 `spring-boot-starter-web` 基于 Servlet（每个请求占用一个线程），而 `spring-boot-starter-webflux` 基于 Netty（少量线程处理大量连接）。SSE 是长连接，如果用传统 MVC，100 个并发 SSE 连接会占用 100 个线程，很容易把线程池耗尽。WebFlux 的非阻塞模型可以用极少线程支撑大量长连接，非常适合 SSE 场景。 `application.yml` 配置： ```yaml spring: ai: openai: api-key: ${OPENAI_API_KEY} chat: options: model: gpt-4o-mini temperature: 0.7 # WebFlux 响应式配置 server: port: 8080 ``` ### 5.3 最简单的 SSE 流式接口先看一个最简单的例子——不调用 AI，纯粹用 `Flux` 模拟流式输出，帮你理解 Flux 和 SSE 的配合： ```java import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Flux; import java.time.Duration; @RestController public class StreamController { /** * 最简单的 SSE 流式接口 * 浏览器访问 http://localhost:8080/hello 即可看到效果 */ @GetMapping(value = "/hello", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux streamHello() { // Flux.interval 每隔 500ms 产生一个数字（0, 1, 2, 3...） // map 把数字转成文字 // take(10) 只取前 10 个 return Flux.interval(Duration.ofMillis(500)) .map(i -> "这是第 " + (i + 1) + " 条消息") .take(10) .doOnNext(s -> System.out.println("发送: " + s)) .doOnComplete(() -> System.out.println("流结束")); } } ``` 这段代码做的事情： 1. `Flux.interval(Duration.ofMillis(500))` —— 每 500 毫秒产生一个数字 2. `.map(...)` —— 把数字转成字符串 3. `.take(10)` —— 只取前 10 个，取完自动完成 4. `produces = MediaType.TEXT_EVENT_STREAM_VALUE` —— 告诉 Spring 这是 SSE 响应 Spring 会自动把每个 `Flux` 元素编码成 SSE 格式发送给浏览器。你用浏览器访问 `http://localhost:8080/hello`，会看到文字一条一条地出现。 ### 5.4 接入 AI 模型的流式输出现在接入真正的大模型。Spring AI 的 `ChatClient` 内置了流式调用支持： ```java import org.springframework.ai.chat.client.ChatClient; import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.*; import reactor.core.publisher.Flux; @RestController @RequestMapping("/api/chat") public class ChatController { private final ChatClient chatClient; // Spring AI 自动注入 ChatClient.Builder public ChatController(ChatClient.Builder builder) { this.chatClient = builder .defaultSystem("你是一个友好的AI助手，用简洁的中文回答问题。") .build(); } /** * SSE 流式对话接口 * 浏览器访问: /api/chat/stream?q=你好 */ @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux streamChat(@RequestParam String q) { // chatClient.prompt().stream().content() 返回 Flux // 每个 Flux 元素就是 AI 生成的一个文本片段（通常是 1-2 个字） return chatClient.prompt() .user(q) .stream() .content(); } } ``` 就这么简单。`chatClient.prompt().user(q).stream().content()` 这一行做了所有事情： - 构建 prompt（把用户消息发给模型） - 调用模型的流式 API（底层走 SSE 连接到 OpenAI/其他模型） - 将模型返回的 token 流转成 `Flux` 返回 - Spring 自动把 Flux 编码成 SSE 格式发给浏览器 **这里发生了一个精妙的"双重 SSE"**：你的后端通过 SSE 连接到 AI 模型的 API，拿到 token 流后封装成 `Flux`，然后你的 Spring Boot 又把这个 Flux 通过另一条 SSE 连接推给浏览器。数据从 AI 模型 → 你的后端 → 浏览器，全程流式，全程不攒数据。 ```mermaid sequenceDiagram participant B as 浏览器 participant SB as Spring Boot participant AI as AI模型API B->>SB: GET /api/chat/stream?q=你好
Accept: text/event-stream SB->>AI: POST /v1/chat/completions
stream: true
(SSE连接) Note over AI: 开始生成token AI-->>SB: SSE: data: {"token": "你"} SB-->>B: SSE: data: 你 AI-->>SB: SSE: data: {"token": "好"} SB-->>B: SSE: data: 好 AI-->>SB: SSE: data: {"token": "，"} SB-->>B: SSE: data: ， AI-->>SB: SSE: data: {"token": "我是"} SB-->>B: SSE: data: 我是 AI-->>SB: SSE: data: [DONE] Note over SB: Flux onComplete SB-->>B: 连接关闭 Note over B: 浏览器渲染: 你好，我是... ``` ### 5.5 封装后端流式数据：结构化 SSE 事件前面说了，AI 的流式输出不仅有正文文本，还可能有思考过程、工具调用、错误信息等。直接返回 `Flux` 只能传输纯文本，无法区分事件类型。我们需要对数据进行封装。方案是：自定义一个 SSE 事件对象，用 `ServerSentEvent` 包装： ```java import org.springframework.http.MediaType; import org.springframework.http.codec.ServerSentEvent; import org.springframework.web.bind.annotation.*; import reactor.core.publisher.Flux; import java.time.LocalDateTime; /** * 统一的流式响应事件封装 */ public record ChatStreamEvent( String event, // 事件类型: token / thinking / tool_call / done / error String content, // 内容 LocalDateTime timestamp ) { // 快速构造方法 public static ChatStreamEvent token(String content) { return new ChatStreamEvent("token", content, LocalDateTime.now()); } public static ChatStreamEvent thinking(String content) { return new ChatStreamEvent("thinking", content, LocalDateTime.now()); } public static ChatStreamEvent done(int totalTokens) { return new ChatStreamEvent("done", "{\"totalTokens\":" + totalTokens + "}", LocalDateTime.now()); } public static ChatStreamEvent error(String message) { return new ChatStreamEvent("error", message, LocalDateTime.now()); } } ``` 然后在 Controller 中用 `ServerSentEvent` 包装： ```java import org.springframework.ai.chat.client.ChatClient; import org.springframework.http.MediaType; import org.springframework.http.codec.ServerSentEvent; import org.springframework.web.bind.annotation.*; import reactor.core.publisher.Flux; @RestController @RequestMapping("/api/chat") public class ChatStreamController { private final ChatClient chatClient; public ChatStreamController(ChatClient.Builder builder) { this.chatClient = builder .defaultSystem("你是一个友好的AI助手。") .build(); } /** * 结构化 SSE 流式接口 * 返回 Flux> * Spring 会把每个 ServerSentEvent 编码成带 event: 和 data: 的 SSE 消息 */ @GetMapping(value = "/v2/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux> streamChatV2(@RequestParam String q) { // 前置事件：告诉客户端"开始生成" Flux> startEvent = Flux.just( ServerSentEvent.builder() .event("start") .data(ChatStreamEvent.thinking("正在思考你的问题...")) .build() ); // AI token 流：每个 token 包装成一个 SSE 事件 Flux> tokenStream = chatClient.prompt() .user(q) .stream() .content() .map(token -> ServerSentEvent.builder() .event("token") .data(ChatStreamEvent.token(token)) .build() ); // 结束事件：告诉客户端"生成完毕" Flux> endEvent = Flux.just( ServerSentEvent.builder() .event("done") .data(ChatStreamEvent.done(0)) .build() ); // 拼接：start + tokens + done // concat 按顺序执行：先发 start，再流式发 tokens，最后发 done // onErrorResume：如果 AI 调用失败，发送 error 事件而不是让连接异常断开 return Flux.concat(startEvent, tokenStream, endEvent) .onErrorResume(e -> Flux.just( ServerSentEvent.builder() .event("error") .data(ChatStreamEvent.error(e.getMessage())) .build() )) // 心跳：每 15 秒发一个注释行，防止代理超时断开 .mergeWith( Flux.interval(Duration.ofSeconds(15)) .map(i -> ServerSentEvent.builder() .comment("keep-alive") .build() ) .takeUntilOther(tokenStream.then()) ); } } ``` 这段代码做了几件关键的事： **1. 结构化事件**：用 `ServerSentEvent` 包装每个事件，可以指定 `event`（事件类型）和 `data`（数据内容）。Spring 会把这些编码成带 `event:` 前缀的 SSE 格式。 **2. 三段式拼接**：`Flux.concat(startEvent, tokenStream, endEvent)` 先发一个"开始"事件，然后流式发出 AI 的每个 token，最后发一个"完成"事件。这样前端可以清晰知道生成状态。 **3. 错误处理**：`onErrorResume` 确保即使 AI 调用失败，也会通过 SSE 发送一个 error 事件，而不是让连接突然断开——前端可以收到错误信息并展示给用户。 **4. 心跳保活**：`.mergeWith(heartbeat)` 每 15 秒发一个注释行，防止 Nginx 等代理因超时关闭连接。`takeUntilOther` 确保在 token 流结束后心跳也自动停止。这个设计是一个生产级 AI 流式接口的骨架。让我用一张完整的时序图来展示数据流向： ```mermaid sequenceDiagram participant FE as 前端 participant CT as ChatStreamController participant CC as ChatClient participant AI as AI模型API FE->>CT: GET /api/chat/v2/stream?q=你好 Note over CT: 发送 start 事件 CT-->>FE: event: start
data: {"event":"thinking","content":"正在思考..."} CT->>CC: prompt().user("你好").stream() CC->>AI: SSE连接到模型 loop 持续接收token AI-->>CC: SSE: token "你" CC-->>CT: Flux元素 "你" CT-->>FE: event: token
data: {"event":"token","content":"你"} AI-->>CC: SSE: token "好" CC-->>CT: Flux元素 "好" CT-->>FE: event: token
data: {"event":"token","content":"好"} end Note over AI: 生成完毕 [DONE] CC-->>CT: Flux onComplete Note over CT: 发送 done 事件 CT-->>FE: event: done
data: {"event":"done","content":"{\"totalTokens\":0}"} Note over CT: Flux onComplete → 连接关闭 FE->>FE: 渲染完成，关闭EventSource ``` ### 5.6 配合 LangChain4j 的 Flux 流式输出如果你用的是 LangChain4j 而不是 Spring AI，实现方式也非常类似。LangChain4j 的 `@AiService` 可以直接声明返回 `Flux`： ```java import dev.langchain4j.service.AiServices; import dev.langchain4j.service.spring.AiService; import dev.langchain4j.service.spring.SystemMessage; import org.springframework.web.bind.annotation.*; import reactor.core.publisher.Flux; // 声明式 AI 服务接口 @AiService public interface Assistant { @SystemMessage("你是一个友好的AI助手，用简洁的中文回答。") Flux streamChat(String userMessage); } ``` ```java @RestController @RequestMapping("/api/lc4j") public class LangChain4jController { private final Assistant assistant; public LangChain4jController(Assistant assistant) { this.assistant = assistant; } @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux stream(@RequestParam String q) { // assistant.streamChat() 返回 Flux // 每个元素是 AI 生成的文本片段 return assistant.streamChat(q); } } ``` LangChain4j 的 `langchain4j-reactor` 模块在底层做了 Flux 适配——它把 `TokenStream` 的 `onPartialResponse` 回调转成了 `Flux` 的元素发射。你拿到的 `Flux` 和 Spring AI 的 `Flux` 在使用方式上完全一致，Spring WebFlux 都能自动编码成 SSE。如果你需要更精细的控制（比如区分思考过程和正文），可以用 `TokenStream` 直接处理，然后手动包装成 `ServerSentEvent`： ```java @GetMapping(value = "/stream-rich", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux> streamRich(@RequestParam String q) { return Flux.create(sink -> { assistant.chat(q) // 返回 TokenStream .onPartialThinking(thinking -> { sink.next(ServerSentEvent.builder() .event("thinking") .data(thinking.text()) .build()); }) .onPartialResponse(token -> { sink.next(ServerSentEvent.builder() .event("token") .data(token) .build()); }) .onCompleteResponse(response -> { sink.next(ServerSentEvent.builder() .event("done") .data("{\"tokens\":" + response.tokenUsage().totalTokenCount() + "}") .build()); sink.complete(); }) .onError(error -> { sink.next(ServerSentEvent.builder() .event("error") .data(error.getMessage()) .build()); sink.complete(); }) .start(); }); } ``` 这里用 `Flux.create()` 手动桥接——TokenStream 的每个回调都对应发射一个 `ServerSentEvent`。这种方式让你可以完全控制每个事件的内容和类型。 ### 5.7 Flux 常用操作符速查在 AI 流式开发中，你会反复用到以下 Flux 操作符。理解它们对于灵活处理流式数据至关重要： ```mermaid mindmap root((Flux 操作符)) 数据转换 map 一对一转换 flatMap 一对多展开 mapNotNull 过滤null 流控制 "take n 只取前n个" "takeUntil 取到条件停止" timeout 超时控制 delayElements 延迟发射流组合 concat 顺序拼接 merge 并行合并 zip 一一配对 combineLatest 取最新错误处理 onErrorResume 降级处理 onErrorReturn 返回默认值 retry 重试 onErrorMap 转换异常副作用 doOnNext 每个元素到达时 doOnComplete 流结束时 doOnError 出错时 doOnSubscribe 订阅时 ``` 最常用的几个： ```java // map：把每个 token 转成大写 flux.map(token -> token.toUpperCase()) // filter：过滤空内容 flux.filter(token -> !token.isEmpty()) // concat：先发 "思考开始"，再发 token 流，最后发 "结束" Flux.concat( Flux.just("思考开始"), aiTokenStream, Flux.just("结束") ) // onErrorResume：AI 调用失败时返回错误提示 aiTokenStream .onErrorResume(e -> Flux.just("[错误] AI 服务暂时不可用: " + e.getMessage())) // timeout：30 秒没数据就超时 aiTokenStream.timeout(Duration.ofSeconds(30)) // scan：累加所有 token（实现"到目前为止的完整文本"） aiTokenStream.scan("", (acc, token) -> acc + token) ``` --- ## 六、最佳实践：前端如何消费 SSE 流式数据后端搞定了，前端怎么接收和渲染？有两种方式：`EventSource` API 和 `fetch + ReadableStream`。 ### 6.1 EventSource：最简单的方式 `EventSource` 是浏览器原生 API，专门用于消费 SSE 流。代码极简： ```javascript // 建立连接 const eventSource = new EventSource('/api/chat/stream?q=你好'); // 监听默认的 message 事件 eventSource.onmessage = function(event) { // event.data 就是 SSE 中 data: 后面的内容 // 每收到一个事件，这个回调就会触发一次 document.getElementById('output').textContent += event.data; }; // 监听自定义事件类型（对应后端 ServerSentEvent.event("thinking")） eventSource.addEventListener('thinking', function(event) { const thinkingDiv = document.getElementById('thinking'); thinkingDiv.textContent += event.data; thinkingDiv.style.color = 'gray'; thinkingDiv.style.fontStyle = 'italic'; }); // 监听 token 事件 eventSource.addEventListener('token', function(event) { const data = JSON.parse(event.data); document.getElementById('output').textContent += data.content; }); // 监听完成事件 eventSource.addEventListener('done', function(event) { console.log('生成完成'); eventSource.close(); // 主动关闭连接 }); // 监听错误事件 eventSource.addEventListener('error', function(event) { console.error('SSE 错误'); eventSource.close(); }); // 错误处理（连接层面） eventSource.onerror = function(event) { console.error('连接错误，浏览器会自动重连...'); // 如果不想自动重连，调用 eventSource.close() }; ``` `EventSource` 的优点是简单、自带重连。缺点是只支持 GET 请求——如果你想用 POST 传一个很大的 prompt，或者带复杂的请求体（比如多轮对话历史），`EventSource` 做不到。 ### 6.2 fetch + ReadableStream：更灵活的方式现代 AI 应用更常用 `fetch` + `ReadableStream` 来消费 SSE，因为它支持 POST 请求和自定义请求体： ```javascript async function streamChat(messages) { const response = await fetch('/api/chat/v3/stream', { method: 'POST', headers: { 'Content-Type': 'application/json', 'Accept': 'text/event-stream' // 声明我要的是SSE流 }, body: JSON.stringify({ messages: messages }) }); if (!response.ok) { throw new Error(`HTTP ${response.status}`); } // 获取可读流的 reader const reader = response.body.getReader(); const decoder = new TextDecoder('utf-8'); let buffer = ''; try { while (true) { // 每次读取一块数据 const { done, value } = await reader.read(); if (done) { console.log('流结束'); break; } // value 是 Uint8Array，解码成字符串 buffer += decoder.decode(value, { stream: true }); // SSE 事件以 \n\n 分隔 // 从 buffer 中切出完整的事件 const lines = buffer.split('\n'); buffer = lines.pop() || ''; // 最后可能是不完整的，留着下次拼 let currentEvent = 'message'; let currentData = ''; for (const line of lines) { if (line.startsWith('event:')) { currentEvent = line.slice(6).trim(); } else if (line.startsWith('data:')) { currentData += (currentData ? '\n' : '') + line.slice(5).trim(); } else if (line === '' && currentData) { // 空行 = 事件结束，处理这个事件 handleSSEEvent(currentEvent, currentData); currentEvent = 'message'; currentData = ''; } } } } finally { reader.releaseLock(); } } function handleSSEEvent(eventType, data) { switch (eventType) { case 'token': const token = JSON.parse(data); appendToChat(token.content); break; case 'thinking': appendToThinking(JSON.parse(data).content); break; case 'done': console.log('生成完成', data); break; case 'error': console.error('错误:', data); break; } } // 使用 streamChat([ { role: 'user', content: '你好，介绍一下你自己' } ]); ``` 这段代码做的事情：用 `fetch` 发一个 POST 请求，拿到响应后，通过 `response.body.getReader()` 获取一个字节流 reader。然后循环调用 `reader.read()` 读取每一块数据，解码成字符串，手动解析 SSE 格式（按 `\n\n` 分割事件，按 `event:` / `data:` 提取字段）。这种方式更灵活，但代码也更复杂。在实际项目中，通常会封装成一个工具函数或使用第三方库（如 `@microsoft/fetch-event-source`）来简化。 ```mermaid graph TB subgraph "前端SSE消费流程" A["fetch POST 请求
Accept: text/event-stream"] --> B["获取 response.body"] B --> C["调用 getReader()"] C --> D["循环 reader.read()"] D --> E{"done?"} E -->|否| F["解码 Uint8Array → String"] F --> G["按 \\n\\n 分割事件"] G --> H["解析 event: 和 data:"] H --> I["根据事件类型渲染"] I --> D E -->|是| J["流结束"] end style D fill:#E3F2FD,stroke:#1565C0 style G fill:#FFF3E0,stroke:#E65100 style I fill:#E8F5E9,stroke:#2E7D32 ``` ### 6.3 封装一个通用的 SSE 客户端工具类把上面的逻辑封装一下，在实际项目中复用： ```javascript /** * SSE 流式客户端工具类 * 支持 GET 和 POST，支持自定义事件类型，支持中断 */ class SSEClient { constructor(url, options = {}) { this.url = url; this.options = options; this.controller = null; // AbortController，用于手动中断 this.eventHandlers = {}; // 事件处理器映射 } // 注册事件处理器 on(eventType, handler) { this.eventHandlers[eventType] = handler; return this; } // 启动流式请求 async start() { this.controller = new AbortController(); const response = await fetch(this.url, { method: this.options.method || 'GET', headers: { 'Accept': 'text/event-stream', ...(this.options.headers || {}) }, body: this.options.body ? JSON.stringify(this.options.body) : undefined, signal: this.controller.signal // 支持中断 }); if (!response.ok) { throw new Error(`HTTP ${response.status}: ${response.statusText}`); } const reader = response.body.getReader(); const decoder = new TextDecoder(); let buffer = ''; try { while (true) { const { done, value } = await reader.read(); if (done) break; buffer += decoder.decode(value, { stream: true }); // 解析完整的 SSE 事件 const events = buffer.split('\n\n'); buffer = events.pop() || ''; for (const eventText of events) { if (!eventText.trim()) continue; let eventType = 'message'; let data = ''; for (const line of eventText.split('\n')) { if (line.startsWith('event:')) { eventType = line.slice(6).trim(); } else if (line.startsWith('data:')) { data += (data ? '\n' : '') + line.slice(5).trim(); } } // 调用对应的事件处理器 const handler = this.eventHandlers[eventType]; if (handler) { handler(data); } // 默认处理器 if (eventType === 'message' && this.eventHandlers['message']) { this.eventHandlers['message'](data); } } } } finally { reader.releaseLock(); } } // 中断请求 abort() { if (this.controller) { this.controller.abort(); } } } // 使用示例 const client = new SSEClient('/api/chat/v3/stream', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: { message: '你好，介绍一下AI的发展' } }); client .on('token', (data) => { const parsed = JSON.parse(data); document.getElementById('output').textContent += parsed.content; }) .on('thinking', (data) => { const parsed = JSON.parse(data); document.getElementById('thinking').textContent += parsed.content; }) .on('done', () => { console.log('生成完成'); }) .on('error', (data) => { console.error('错误:', data); }); client.start(); // 用户点击"停止生成"时 // document.getElementById('stop').onclick = () => client.abort(); ``` 这个工具类封装了 SSE 解析的完整逻辑，支持： - POST 请求（携带 JSON body） - 自定义事件类型监听 - 主动中断（AbortController） - 错误处理在 Vue 或 React 项目中，你可以把它放到一个 composable 或 hook 中配合响应式状态使用。 ### 6.4 前端流式实时渲染：Markdown 与代码高亮组件到这里你已经有了一个能接收 SSE token 流的客户端工具类。但拿到 token 只是第一步——你怎么把它渲染成漂亮的 Markdown？怎么让代码块在流式过程中就有语法高亮？这是 AI 流式前端开发中**最容易被低估的难点**。让我讲清楚问题在哪，以及怎么解。 #### 问题：Markdown 解析器无法处理"半截文本" 传统的 Markdown 渲染器（如 `marked`、`markdown-it`、`react-markdown`）的设计前提是：给我一段**完整的** Markdown 文本，我返回渲染好的 HTML。但流式输出时，你拿到的是一个 token 一个 token 地追加。比如 AI 正在生成一段代码： ``` // 第1个token ``` ```python # 第2个token def # 第3个token hello # 第4个token (): # 第5个token print # 第6个token ("hello") # 第7个token ``` 在前 6 个 token 到达时，这段 Markdown 是**不完整的**——代码块的开头 `` ```python `` 有了，但结束的 `` ``` `` 还没来。如果你把这段不完整的文本丢给 Markdown 解析器，它要么报错，要么把未闭合的代码块当成普通文本渲染——结果就是用户看到一大坨没有格式的乱码，直到最后一个 token 到来时才突然变好看。这种"先丑后美"的渲染跳变严重影响体验。更糟糕的是，每次新 token 到来都重新解析全部文本，会导致整段内容闪烁重绘——代码块在"闭合→未闭合→闭合"之间反复跳变，用户根本没法看。 ```mermaid graph TB subgraph "朴素方案：每次全量重新解析" T1["token: ```python"] --> P1["解析 → 代码块开始
但未闭合，渲染为普通文本"] T2["token: def"] --> P2["重新解析全文 → 仍未闭合"] T3["token: hello():"] --> P3["重新解析 → 仍未闭合"] T4["token: print('hi')"] --> P4["重新解析 → 仍未闭合"] T5["token: ```"] --> P5["重新解析 → 代码块闭合！
突然变好看了"] end subgraph "问题" Q1["1. 前4步渲染为普通文本（丑）"] Q2["2. 每步全量重解析（慢）"] Q3["3. 最后一步格式跳变（闪）"] end P5 --> Q1 P5 --> Q2 P5 --> Q3 style P1 fill:#FFE0E0,stroke:#D32F2F style P2 fill:#FFE0E0,stroke:#D32F2F style P3 fill:#FFE0E0,stroke:#D32F2F style P4 fill:#FFE0E0,stroke:#D32F2F style P5 fill:#E0F2E0,stroke:#388E3C ``` #### 解决思路：增量渲染 + 防抖 + 容错解析业界主流的解法是三管齐下： **1. 防抖（Debounce）**：不是每个 token 到了就立刻渲染，而是攒一个小批（比如 50ms 内的 token），然后一次性渲染。这样减少了重解析频率，也避免了 token 间隔极短时的高频重绘。 **2. 容错解析**：让 Markdown 解析器对不完整的语法"宽容"一些。比如遇到未闭合的代码块，就假设它会在后面闭合，先按代码块渲染。遇到未闭合的加粗 `**`，先按加粗渲染。 **3. 智能切换**：在检测到正在写代码块（遇到过 `` ``` `` 但还没遇到闭合的 `` ``` ``）时，用纯文本模式追加渲染代码内容，不做 Markdown 解析；代码块闭合后一次性做语法高亮。 ```mermaid graph TB A["SSE token 流"] --> B["缓冲区追加 token"] B --> C{"距上次渲染>50ms?"} C -->|否| B C -->|是| D["取出缓冲区内容"] D --> E{"是否在代码块内?"} E -->|是| F["代码块内：追加纯文本
不做Markdown解析"] E -->|否| G["代码块外：增量Markdown解析"] F --> H["渲染到DOM"] G --> H H --> B style C fill:#FFF3E0,stroke:#E65100 style E fill:#E3F2FD,stroke:#1565C0 ``` #### 主流渲染组件选型下面是目前前端 AI 流式渲染最常用的方案： | 组件 / 库 | 技术栈 | 特点 | 适用场景 | |-----------|--------|------|---------| | **react-markdown** + rehype | React | 生态成熟，支持插件（rehype-highlight/shiki） | React 项目首选 | | **markdown-it** + highlight.js | 框架无关 | 轻量灵活，可自定义渲染规则 | Vue/原生 JS 项目 | | **streamdown** | React | 专为流式设计，内置增量解析 | 专注 AI 聊天 UI | | **Shiki** | 框架无关 | VS Code 同款高亮引擎，效果最佳 | 对代码高亮质量要求高 | | **rehype-pretty-code** | React | 基于 Shiki，支持行高亮、diff | 技术博客/文档类 AI 应用 | | **marked** + DOMPurify | 框架无关 | 极简，性能好 | 轻量级需求 | #### React 实战：react-markdown 流式渲染 ```jsx import React, { useState, useEffect, useRef, useCallback } from 'react'; import ReactMarkdown from 'react-markdown'; import remarkGfm from 'remark-gfm'; // GitHub Flavored Markdown import rehypeHighlight from 'rehype-highlight'; // 代码语法高亮 import 'highlight.js/styles/github-dark.css'; // 高亮主题 /** * AI 流式 Markdown 渲染组件 * 接收 SSE token 流，增量渲染 Markdown */ export function StreamMarkdown({ tokenStream$, isStreaming }) { const [content, setContent] = useState(''); const bufferRef = useRef(''); const renderTimerRef = useRef(null); const inCodeBlockRef = useRef(false); // 是否在代码块内 // 防抖渲染：50ms 内的 token 攒一批再渲染 const scheduleRender = useCallback(() => { if (renderTimerRef.current) return; // 已有定时器在等 renderTimerRef.current = setTimeout(() => { renderTimerRef.current = null; setContent(bufferRef.current); }, 50); }, []); useEffect(() => { if (!tokenStream$) return; const subscription = tokenStream$.subscribe({ next: (token) => { bufferRef.current += token; // 检测是否进入了/退出了代码块 // 简单策略：统计 ``` 出现次数，奇数=在代码块内 const fenceCount = (bufferRef.current.match(/```/g) || []).length; inCodeBlockRef.current = fenceCount % 2 === 1; scheduleRender(); }, complete: () => { // 流结束，确保最终内容完整渲染 if (renderTimerRef.current) { clearTimeout(renderTimerRef.current); renderTimerRef.current = null; } setContent(bufferRef.current); } }); return () => subscription.unsubscribe(); }, [tokenStream$, scheduleRender]); // 流结束后清理缓冲区（为下次对话准备） useEffect(() => { if (!isStreaming && content) { bufferRef.current = content; // 保留当前内容 } }, [isStreaming]); return (

{children}

); }, // 自定义链接渲染：新窗口打开 a({ href, children }) { return {children}; } }} > {content} {/* 流式进行中的光标动画 */} {isStreaming && ( ▊ )} ); } ``` 关键设计点解析： 1. **`bufferRef` + 防抖**：token 先进 buffer，50ms 定时器触发渲染。不是每个 token 都重新解析 Markdown，避免高频重绘的性能问题。 2. **代码块检测**：通过统计 `` ``` `` 的出现次数判断当前是否在代码块内。奇数次=进入了代码块但还没出来。这个信息可以用来在代码块内切换为纯文本追加模式（不做 Markdown 解析），代码块闭合后一次性做语法高亮。 3. **`rehypeHighlight` 的 `ignoreMissing: true`**：流式过程中代码块可能不完整（语言标识可能有，但代码内容只有半句），`ignoreMissing: true` 让高亮器遇到无法识别的语言时静默跳过而不是报错。 4. **`detect: true`**：让 highlight.js 自动检测代码语言，因为流式过程中 AI 可能还没写完 `` ```python `` 这个语言标识。 5. **流式光标**：`isStreaming` 状态下显示一个闪烁光标，告诉用户"还在生成"。 #### Vue 实战：markdown-it 流式渲染 ```vue ``` #### 代码高亮方案对比：Shiki vs highlight.js | 维度 | highlight.js | Shiki | |------|-------------|-------| | **高亮引擎** | 正则匹配 | TextMate 语法（VS Code 同款） | | **高亮质量** | 好 | **极好**（精确到 token 类型） | | **包体积** | 小（可按需引入语言） | 较大（每个语言一份 JSON 语法文件） | | **运行时性能** | 快 | 首次加载稍慢（需异步加载语法） | | **流式友好度** | 高（同步、容错好） | 中（异步加载可能闪烁） | | **主题支持** | CSS 主题 | 内置 VS Code 所有主题 | | **推荐场景** | AI 聊天流式渲染 | 技术文档/博客最终渲染 | 在 AI 流式输出场景下，**highlight.js 更合适**——它是同步的、容错性好、包体积小，适合在流式过程中频繁调用。Shiki 的异步特性在流式场景下可能导致"先无色后有色"的跳变。如果最终结果需要 Shiki 级别的高亮质量，可以采用**流式阶段用 highlight.js、流结束后用 Shiki 重新高亮**的混合策略。 #### 一个完整的流式渲染流程 ```mermaid graph LR subgraph "数据层" SSE["SSE token 流"] --> BUF["缓冲区
累加 token"] end subgraph "渲染控制层" BUF -->|"50ms 防抖"| REND["触发渲染"] REND --> CB{"在代码块内?"} CB -->|是| TXT["纯文本追加模式
不做MD解析"] CB -->|否| MD["Markdown 解析"] end subgraph "渲染层" TXT --> CODE["代码块 DOM
追加文本"] MD --> HTML["渲染 HTML
react-markdown/markdown-it"] CODE --> HL["highlight.js 高亮"] end subgraph "交互层" HTML --> CUR["显示流式光标 ▊"] CODE --> CUR CUR --> COPY["代码复制按钮"] end SSE -.->|"流结束"| FINAL["最终渲染
全量Markdown解析
可选Shiki重高亮"] style BUF fill:#E3F2FD,stroke:#1565C0 style REND fill:#FFF3E0,stroke:#E65100 style CB fill:#E8EAF6,stroke:#3F51B5 style FINAL fill:#E8F5E9,stroke:#2E7D32 ``` --- ## 七、最佳实践：Python 中的 SSE 流式输出 AI 开发不只是 Java 的专利，Python 是 AI 领域的原生语言。在 Python 生态中实现 SSE 流式输出，最常用的框架是 **FastAPI**——它原生支持异步和流式响应。 ### 7.1 FastAPI SSE 基础先安装依赖： ```bash pip install fastapi uvicorn openai sse-starlette ``` > `sse-starlette` 是一个为 Starlette/FastAPI 提供 SSE 支持的库，封装了 `text/event-stream` 的格式化逻辑。最简单的 SSE 接口： ```python from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app = FastAPI() async def event_generator(): """生成 SSE 事件流的异步生成器""" for i in range(10): # 每隔 500ms 产生一个事件 await asyncio.sleep(0.5) # yield 出去的每一项就是一个 SSE 事件 yield f"data: 这是第 {i + 1} 条消息\n\n" @app.get("/hello") async def stream_hello(): """SSE 流式接口""" return StreamingResponse( event_generator(), media_type="text/event-stream" ) ``` 这段代码做的事情和 Java 版本完全一致。`StreamingResponse` 接收一个异步生成器，每当生成器 `yield` 一个值，FastAPI 就把它当作一个 SSE 数据块发送给客户端。`media_type="text/event-stream"` 告诉浏览器这是 SSE 流。 ### 7.2 接入 OpenAI 流式 API ```python from fastapi import FastAPI from fastapi.responses import StreamingResponse from openai import OpenAI import json app = FastAPI() client = OpenAI() # 从环境变量读取 OPENAI_API_KEY async def chat_stream_generator(message: str): """AI 聊天的 SSE 事件流生成器""" # 1. 发送开始事件 yield f"event: start\ndata: {json.dumps({'status': 'thinking'})}\n\n" try: # 2. 调用 OpenAI 流式 API # stream=True 让 SDK 返回一个迭代器，逐个 yield token stream = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "你是一个友好的AI助手。"}, {"role": "user", "content": message} ], stream=True # 关键参数：开启流式 ) total_tokens = 0 # 3. 遍历流式响应，每个 chunk 就是一个 token 片段 for chunk in stream: if chunk.choices[0].delta.content is not None: token = chunk.choices[0].delta.content total_tokens += 1 # 把每个 token 包装成 SSE 事件 event_data = json.dumps({ "content": token, "index": total_tokens }) yield f"event: token\ndata: {event_data}\n\n" # 4. 发送完成事件 yield f"event: done\ndata: {json.dumps({'totalTokens': total_tokens})}\n\n" except Exception as e: # 5. 错误处理：发送错误事件而不是让连接异常断开 yield f"event: error\ndata: {json.dumps({'message': str(e)})}\n\n" @app.get("/api/chat/stream") async def stream_chat(message: str): """SSE 流式聊天接口""" return StreamingResponse( chat_stream_generator(message), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", # Nginx 禁用缓冲，确保实时推送 } ) ``` 这段代码的流程： 1. `client.chat.completions.create(..., stream=True)` 调用 OpenAI 的流式 API。`stream=True` 让 SDK 不等待完整响应，而是返回一个迭代器。 2. `for chunk in stream` 遍历每个 token 片段。每个 `chunk` 包含一个 `delta.content`，就是 AI 当前生成的一小段文字。 3. 每个 token 被包装成 SSE 格式（`event: token\ndata: {...}\n\n`）后 `yield` 出去，FastAPI 立即推送给客户端。 4. 生成完毕后发送 `done` 事件，包含 token 统计。 5. 任何异常都通过 `error` 事件通知前端，而不是让连接静默断开。 ```mermaid sequenceDiagram participant FE as 前端 participant FA as FastAPI participant OA as OpenAI API FE->>FA: GET /api/chat/stream?message=你好 Note over FA: yield start 事件 FA-->>FE: event: start
data: {"status":"thinking"} FA->>OA: chat.completions.create(stream=True) loop 遍历 stream OA-->>FA: chunk: {"delta": {"content": "你"}} FA-->>FE: event: token
data: {"content":"你","index":1} OA-->>FA: chunk: {"delta": {"content": "好"}} FA-->>FE: event: token
data: {"content":"好","index":2} end OA-->>FA: 流结束 Note over FA: yield done 事件 FA-->>FE: event: done
data: {"totalTokens": 2} Note over FA: 生成器结束 → 连接关闭 ``` ### 7.3 使用 sse-starlette 简化代码 `sse-starlette` 库提供了 `EventSourceResponse`，可以更优雅地处理 SSE 事件： ```python from fastapi import FastAPI from sse_starlette.sse import EventSourceResponse from openai import OpenAI import json import asyncio app = FastAPI() client = OpenAI() async def chat_event_generator(message: str): """使用 sse-starlette 的事件生成器 yield 的是 dict，由 EventSourceResponse 自动编码成 SSE 格式 """ # 开始事件 yield {"event": "start", "data": json.dumps({"status": "thinking"})} try: stream = client.chat.completions.create( model="gpt-4o-mini", messages=[ {"role": "system", "content": "你是一个友好的AI助手。"}, {"role": "user", "content": message} ], stream=True ) token_count = 0 for chunk in stream: content = chunk.choices[0].delta.content if content is not None: token_count += 1 # yield dict，EventSourceResponse 自动编码 yield { "event": "token", "data": json.dumps({"content": content, "index": token_count}) } yield { "event": "done", "data": json.dumps({"totalTokens": token_count}) } except Exception as e: yield { "event": "error", "data": json.dumps({"message": str(e)}) } @app.get("/api/chat/stream") async def stream_chat(message: str): """使用 EventSourceResponse 替代 StreamingResponse""" return EventSourceResponse( chat_event_generator(message), # ping=15 表示每 15 秒发一个心跳注释，防止代理超时 ping=15 ) ``` `EventSourceResponse` 的优势： - 自动把 dict 编码成 SSE 格式（`event: xxx\ndata: yyy\n\n`） - 内置心跳支持（`ping=15` 每 15 秒发注释行） - 自动处理 `retry` 字段 - 错误处理更优雅 ### 7.4 使用 LangChain 的流式输出如果你用 LangChain 而不是直接调 OpenAI SDK，流式输出同样简单： ```python from fastapi import FastAPI from fastapi.responses import StreamingResponse from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage import json app = FastAPI() # LangChain 模型，开启流式 model = ChatOpenAI( model="gpt-4o-mini", streaming=True, # 关键：开启流式 temperature=0.7 ) async def langchain_stream_generator(message: str): """LangChain 流式 SSE 生成器""" yield f"event: start\ndata: {json.dumps({'status': 'thinking'})}\n\n" try: # LangChain 的 stream() 方法返回一个迭代器 # 每次迭代产出一个 AIMessageChunk，包含一小段文本 messages = [ SystemMessage(content="你是一个友好的AI助手。"), HumanMessage(content=message) ] token_count = 0 for chunk in model.stream(messages): token = chunk.content if token: token_count += 1 yield f"event: token\ndata: {json.dumps({'content': token, 'index': token_count})}\n\n" yield f"event: done\ndata: {json.dumps({'totalTokens': token_count})}\n\n" except Exception as e: yield f"event: error\ndata: {json.dumps({'message': str(e)})}\n\n" @app.get("/api/chat/langchain") async def langchain_stream(message: str): return StreamingResponse( langchain_stream_generator(message), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", } ) ``` LangChain 的 `model.stream()` 方法和 OpenAI SDK 的 `stream=True` 参数本质上是同一个东西——都是把大模型的流式 API 封装成 Python 的迭代器接口。区别在于 LangChain 额外提供了 `astream()`（异步版本）和 `astream_events()`（更细粒度的事件流，包括工具调用、检索等）。 ### 7.5 Python 和 Java 的对比 | 维度 | Java (Spring Boot) | Python (FastAPI) | |------|-------------------|------------------| | **流式核心类型** | `Flux` (Reactor) | `AsyncGenerator` (Python) | | **SSE 编码** | Spring WebFlux 自动编码 | `StreamingResponse` / `EventSourceResponse` | | **AI SDK 流式调用** | `ChatClient.stream()` / `TokenStream` | `client.chat.completions.create(stream=True)` / `model.stream()` | | **数据流模型** | 响应式流（背压支持） | 异步迭代器 | | **并发模型** | Netty 少量线程 | asyncio 事件循环 | | **SSE 格式化** | `ServerSentEvent` 包装 | 手动 `f"data: ...\n\n"` 或 `sse-starlette` | 两种语言的实现思路是一致的： 1. **AI SDK 提供流式迭代器**——不管是 `Flux` 还是 `for chunk in stream`，底层都是通过 SSE 连接到 AI 模型 API，逐 token 接收。 2. **Web 框架把迭代器转成 SSE 推给浏览器**——Spring WebFlux 自动把 `Flux` 编码，FastAPI 通过 `StreamingResponse` 把生成器输出推出去。 3. **数据封装**——用 `ServerSentEvent`（Java）或 dict/字符串（Python）封装事件类型和数据。 --- ## 八、完整架构图：从浏览器到 AI 模型的全链路最后，让我们用一张完整的架构图把所有环节串联起来，看看一个生产级 AI 流式输出系统的全貌： ```mermaid graph TB subgraph "用户终端" UI["前端页面
fetch + ReadableStream
逐token渲染"] end subgraph "接入层" N["Nginx / 负载均衡
proxy_buffering off
proxy_read_timeout 300s"] end subgraph "应用层 Spring Boot / FastAPI" CT["Controller
返回 Flux 或 StreamingResponse"] SV["Service 层
封装 SSE 事件
事件类型: token/thinking/done/error"] SDK["AI SDK
Spring AI / LangChain4j / LangChain"] end subgraph "AI 模型层" AI["OpenAI / Claude / Gemini
通过 SSE 返回 token 流"] end subgraph "基础设施" RD["Redis
会话管理 / 限流"] LG["日志 / 监控
Prometheus + Grafana"] end UI -->|"HTTP SSE 连接
Accept: text/event-stream"| N N -->|"转发（不缓冲）"| CT CT --> SV SV --> SDK SDK -->|"SSE 连接到 AI API
stream: true"| AI AI -.->|"token 流"| SDK SDK -.->|"Flux / Iterator"| SV SV -.->|"ServerSentEvent 包装"| CT CT -.->|"text/event-stream"| N N -.->|"SSE 推送"| UI SV --- RD SV --- LG style UI fill:#E8F5E9,stroke:#2E7D32 style AI fill:#FFF3E0,stroke:#E65100 style SV fill:#E3F2FD,stroke:#1565C0 ``` **关键配置要点**： 1. **Nginx 必须关闭缓冲**：`proxy_buffering off` 和 `proxy_cache off`，否则 Nginx 会把 SSE 数据攒在缓冲区里，等攒够一批再发给浏览器，完全破坏流式效果。还要设 `proxy_read_timeout` 足够长（AI 深度思考可能需要几分钟）。 2. **HTTP/2 优先**：如果可能，启用 HTTP/2 避免浏览器对同一域名 6 个连接的限制。 3. **超时配置**：AI 模型生成可能需要很长时间（尤其是深度思考模式），确保每一层的超时设置都足够——网关、负载均衡、应用服务器。 4. **限流和会话管理**：SSE 是长连接，每个用户占用一个连接。在高并发场景下需要通过 Redis 管理会话、限制单用户并发连接数。 --- ## 九、常见问题与排查指南 ### 9.1 为什么我的 SSE 事件不触发？ **最常见原因：忘了事件后面的空行（`\n\n`）。** SSE 规范规定，一个事件必须以空行结束。如果只有 `data: hello` 后面没有 `\n\n`，浏览器会一直等着，不触发 `onmessage`。 ```python # ❌ 错误：只有 \n，没有空行 yield f"data: hello\n" # ✅ 正确：\n\n 结束事件 yield f"data: hello\n\n" ``` ### 9.2 为什么数据一次性返回而不是流式？ **常见原因 1：Nginx 缓冲。** Nginx 默认开启 `proxy_buffering`，会把后端的响应缓冲完整再发给客户端。SSE 场景下必须关闭： ```nginx location /api/chat/ { proxy_pass http://backend; proxy_buffering off; # 关闭响应缓冲 proxy_cache off; # 关闭缓存 proxy_read_timeout 300s; # 读超时设为5分钟 proxy_http_version 1.1; # 使用HTTP/1.1 } ``` **常见原因 2：Spring Boot 缓冲。** 如果你在 Spring MVC（非 WebFlux）中使用 `SseEmitter`，某些 HTTP 消息转换器可能会缓冲。确保使用 WebFlux 的 `Flux` 返回方式。 **常见原因 3：Python 中 `yield` 不在异步上下文中。** FastAPI 的 `StreamingResponse` 需要异步生成器。如果你用了同步函数，整个流会被阻塞到完成才返回。确保 `async def` 和 `await` 使用正确。 ### 9.3 连接总是很快断开？ **检查心跳**。如果 AI 生成超过 60 秒，Nginx 的默认 `proxy_read_timeout` 会断开连接。解决方法： - 增加 `proxy_read_timeout` 到 300 秒或更长 - 在服务器端发送心跳注释（Java 的 `.mergeWith(interval)` 或 Python 的 `ping=15`） ### 9.4 如何实现"停止生成"功能？前端通过 `AbortController` 中断 fetch 请求： ```javascript const controller = new AbortController(); fetch('/api/chat/stream', { signal: controller.signal }); // 用户点击"停止" controller.abort(); ``` 后端在 Java 中可以通过 `Flux` 的 `doOnCancel` 感知到客户端断开，并取消上游 AI 请求： ```java return chatClient.prompt() .user(q) .stream() .content() .doOnCancel(() -> { log.info("客户端取消了流式请求"); // 可以在这里释放资源 }); ``` 在 Python 中，FastAPI 会在客户端断开时让生成器收到 `CancelledError`： ```python async def chat_stream_generator(message: str): try: # ... 生成逻辑 ... except asyncio.CancelledError: # 客户端断开连接 print("客户端取消了请求") # 清理资源 raise # 重新抛出，让 FastAPI 正常处理 ``` ### 9.5 踩坑 #1：空格和换行被后端"吃掉"了 **这是流式输出开发中最高频的坑，没有之一。** 现象：AI 模型明明返回了 `"你好\n\n这是第二段"`，前端收到的却变成了 `"你好这是第二段"`——换行没了。或者模型返回了 `" 缩进代码"`，前端收到的是 `"缩进代码"`——前导空格没了。Markdown 的段落分隔（两个换行）全部失效，所有文字挤成一坨。这个坑有多个元凶，让我一个一个揪出来。 #### 元凶 1：JSON 序列化中的 `.trim()` 很多开发者在封装 SSE 数据时会这样做： ```python # ❌ Python 常见错误 event_data = json.dumps({ "content": token.strip() # ← 罪魁祸首！.strip() 会去掉首尾空格和换行 }) yield f"event: token\ndata: {event_data}\n\n" ``` ```java // ❌ Java 常见错误 return chatClient.prompt() .user(q) .stream() .content() .map(token -> token.trim()); // ← 同样的错误！.trim() 吃掉了空格和换行 ``` AI 模型返回的 token 经常以空格或换行开头——比如 `" 你"` 或 `"\n\n"``。这些空白字符是 Markdown 格式的核心组成部分：`\n\n` 是段落分隔，` ` 开头是代码缩进，`\n` 是行内换行。`trim()` 把它们干掉了，格式自然全乱。 **解决：绝对不要对 AI 的 token 做 trim。** token 里是什么就传什么，一个空格都不能少。 ```python # ✅ 正确：原样传递 event_data = json.dumps({ "content": token # 不做任何处理 }) ``` ```java // ✅ 正确：原样传递 return chatClient.prompt() .user(q) .stream() .content(); // 不加 .map(token -> token.trim()) ``` #### 元凶 2：SSE `data:` 字段的 `.trim()` 解析这个坑更隐蔽。SSE 规范说 `data:` 后面的内容会被浏览器解析，但很多 SSE 解析库（包括一些前端代码）在解析时会自动 `.trim()` 掉 `data:` 后面的内容。来看你自己写的前端 SSE 解析代码： ```javascript // ❌ 很多教程里的写法 const data = line.slice(5).trim(); // ← .trim() 又来了！ ``` `line.slice(5)` 取的是 `data:` 后面的内容，如果内容是 `" 你好"`（前面有空格），`.trim()` 就会把空格干掉。而且这个坑更恶心：**SSE 规范本身允许 `data:` 后面有一个可选的空格**。规范说的是"冒号后面如果有一个空格，这个空格不算是数据的一部分"。但很多实现把**所有**空格都 trim 了。 **解决：手动控制 trim 的范围。** 按规范，只去掉 `data:` 后面的**第一个**空格（如果有的话），后面的内容原样保留。 ```javascript // ✅ 正确的 SSE data 字段解析 function parseSSEDataField(line) { // data: 后面的内容 let data = line.slice(5); // "data:".length === 5 // SSE 规范：如果第一个字符是空格，跳过它 // 但只跳过一个空格，不做全量 trim if (data.startsWith(' ')) { data = data.slice(1); } // 注意：不要用 .trim()！只去掉开头的一个空格 return data; } ``` #### 元凶 3：JSON 中的换行符转义当你用 JSON 包装 token 时，`\n` 会被 JSON 编码成 `\\n`（即字符串 `\n`）。这本身没问题——`json.dumps()` 会正确处理，`JSON.parse()` 也会正确还原。但如果你**手动拼接** JSON 字符串（不使用序列化库），换行符会导致 JSON 格式错误： ```python # ❌ 手动拼接 JSON，换行符会破坏 JSON 格式 token = "第一行\n第二行" yield f'data: {{"content": "{token}"}}\n\n' # 实际发送：data: {"content": "第一行 # 第二行"}\n\n ← JSON 换行了，格式错误！ ``` ```python # ✅ 使用 json.dumps()，换行符会被正确转义为 \n import json yield f'event: token\ndata: {json.dumps({"content": token})}\n\n' # 实际发送：data: {"content": "\u7b2c\u4e00\u884c\n\u7b2c\u4e8c\u884c"}\n\n # 换行符变成了 \n（两个字符），JSON 格式正确 ``` #### 元凶 4：Spring WebFlux 的 `Flux` 自动 JSON 序列化如果你在 Spring WebFlux 中直接返回 `Flux`（而不是 `Flux>`），Spring 会把每个 String 元素当作 SSE 的 `data` 字段值。但这里有一个**隐藏行为**：Spring 默认使用 Jackson 序列化器，它会把 String 用双引号包裹——`你好` 变成 `"你好"`（带引号）。如果你的前端直接用 `event.data`，拿到的会是带引号的字符串。 ```java // ❌ 可能出现问题：Flux 直接返回 @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux stream(@RequestParam String q) { return chatClient.prompt().user(q).stream().content(); // SSE 输出：data: "你好"\n\n ← 注意多了双引号！ } ``` ```java // ✅ 使用 ServerSentEvent 或指定不序列化 @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public Flux stream(@RequestParam String q) { return chatClient.prompt().user(q).stream().content() .map(token -> token); // 明确 map，避免自动序列化干扰 } // 或者在 WebFlux 配置中禁用 String 的 JSON 序列化 ``` 如果前端拿到了带引号的字符串，记得 `JSON.parse(event.data).replace(/^"|"$/g, '')` 或者直接检查是否需要 parse。 ```mermaid graph TB subgraph "空格/换行丢失的四个元凶" E1[".trim() / .strip()
后端处理token时"] E2["SSE data: 解析时 trim
前端解析时"] E3["手动拼接JSON
换行符破坏格式"] E4["Spring自动JSON序列化
String被加引号"] end subgraph "解决方案" S1["后端：不做任何 trim/strip
token 原样传递"] S2["前端：只跳过 data: 后第一个空格
不做全量 trim"] S3["使用 json.dumps/JSON.stringify
不要手动拼JSON"] S4["用 ServerSentEvent 明确指定
或检查是否需要 parse"] end E1 -.-> S1 E2 -.-> S2 E3 -.-> S3 E4 -.-> S4 style E1 fill:#FFCDD2,stroke:#D32F2F style E2 fill:#FFCDD2,stroke:#D32F2F style E3 fill:#FFCDD2,stroke:#D32F2F style E4 fill:#FFCDD2,stroke:#D32F2F style S1 fill:#C8E6C9,stroke:#388E3C style S2 fill:#C8E6C9,stroke:#388E3C style S3 fill:#C8E6C9,stroke:#388E3C style S4 fill:#C8E6C9,stroke:#388E3C ``` ### 9.6 踩坑 #2：Markdown 流式渲染闪烁和格式跳变这个坑在前面 6.4 节提到过，这里补充更完整的分析和解决方案。 **现象**：流式过程中，已经渲染好的 Markdown 突然"塌缩"——一段格式漂亮的文字突然变回纯文本，然后再重新变好。或者代码块在"有高亮→无高亮→有高亮"之间反复跳变。 **根因**：Markdown 解析器对**不完整语法**的处理方式不可预测。比如： ``` AI 正在生成：**这是加粗文字** ``` 当 token 流到达 `**这是` 时，Markdown 解析器看到一个未闭合的 `**`，可能把它渲染成普通文本 `**这是`。下一个 token `加粗` 到达后，文本变成 `**这是加粗`，还是未闭合。再下一个 token `文字**` 到达，终于闭合了——这时整段文字突然从"纯文本带星号"变成"加粗无星号"，视觉上就是一个明显的跳变。 **解决方案：增量解析 + 状态保持** 更高级的方案是不全量重新解析，而是只解析**新增的部分**，与之前已经解析好的 AST（抽象语法树）合并。这样已渲染的部分不会因为新增内容而重绘。 ```javascript /** * 增量 Markdown 渲染器 * 只解析新增的 token，避免全量重解析导致的闪烁 */ class IncrementalMarkdownRenderer { constructor(parser) { this.parser = parser; // markdown-it 或 marked 实例 this.fullText = ''; // 完整文本 this.lastRenderedIndex = 0; // 上次渲染到的位置 this.renderedBlocks = []; // 已渲染的块列表 } /** * 追加新 token */ append(token) { const prevText = this.fullText; this.fullText += token; // 检测是否跨越了块级边界（段落/代码块/列表等） // 块级边界 = 两个连续换行 \n\n const newText = this.fullText.slice(this.lastRenderedIndex); const blockBoundary = newText.lastIndexOf('\n\n'); if (blockBoundary !== -1) { // 有新的完整块可以渲染了 const completeBlock = newText.slice(0, blockBoundary + 2); const html = this.parser.render(completeBlock); this.renderedBlocks.push(html); this.lastRenderedIndex = prevText.length + blockBoundary + 2; } // 追加"进行中"的部分（最后一个未闭合的块） const inProgress = this.fullText.slice(this.lastRenderedIndex); const inProgressHtml = this.parser.render(inProgress); return this.renderedBlocks.join('') + inProgressHtml; } /** * 流结束时，确保最终完整渲染 */ finalize() { const html = this.parser.render(this.fullText); this.renderedBlocks = [html]; this.lastRenderedIndex = this.fullText.length; return html; } } ``` 这个方案的核心思想：**以 `\n\n`（段落分隔符）为边界，把文本切成块。** 已经完整的块渲染一次就不再重新渲染。只有最后一个"进行中"的块会频繁更新。这样已渲染的内容不会闪烁。 ```mermaid graph TB subgraph "增量渲染策略" A["token 流"] --> B["追加到 fullText"] B --> C["检测 \\n\\n 块边界"] C --> D{"有新完整块?"} D -->|是| E["渲染新块
加入 renderedBlocks"] D -->|否| F["只重新渲染
最后一个未闭合块"] E --> G["输出: 已完成块 + 进行中块"] F --> G G --> B H["流结束"] --> I["finalize: 全量重新渲染
确保最终格式正确"] end style E fill:#C8E6C9,stroke:#388E3C style F fill:#FFF9C4,stroke:#F9A825 style I fill:#C8E6C9,stroke:#388E3C ``` ### 9.7 踩坑 #3：数据被"攒着"不实时推送 **现象**：AI 模型确实在流式生成，但前端要等好几秒才突然收到一大段文字，然后又等好几秒——完全不是"打字机"效果。这个问题通常不是 AI 模型的问题，而是传输链路上某个环节在缓冲数据。让我列出所有可能的缓冲点： ```mermaid graph TB subgraph "数据缓冲的 6 个可能位置" P1["1. AI SDK 内部缓冲
Spring AI / LangChain
某些SDK会攒token"] P2["2. Java Flux 操作符
buffer/window/bufferTimeout"] P3["3. Spring WebFlux 编码器
某些Codec会缓冲"] P4["4. Nginx proxy_buffering
默认开启"] P5["5. CDN / 云网关
可能缓存响应"] P6["6. 浏览器 TextDecoder
某些编码下会攒"] end P1 --> P2 --> P3 --> P4 --> P5 --> P6 style P4 fill:#FFCDD2,stroke:#D32F2F style P5 fill:#FFCDD2,stroke:#D32F2F ``` **排查方法**：在浏览器 Network 面板选中 SSE 请求，看 EventStream 标签页。如果这里的数据是一条一条实时到达的，说明后端没问题，问题在渲染层（防抖太长？Markdown 解析太慢？）。如果这里的数据也是一大段一大段到达的，说明是传输层在缓冲。 **最常见的两个缓冲元凶**： **元凶 1：Nginx `proxy_buffering`（最常见）** Nginx 默认开启响应缓冲——它会把后端的响应攒到一定大小或超时后才发给客户端。对于普通 HTTP 响应这是优化，对于 SSE 这是灾难。 ```nginx # ❌ 默认配置——SSE 数据被 Nginx 攒着 location /api/ { proxy_pass http://backend; } # ✅ SSE 专用配置——关闭所有缓冲 location /api/chat/ { proxy_pass http://backend; proxy_buffering off; # 关闭响应缓冲（最关键！） proxy_cache off; # 关闭缓存 proxy_request_buffering off; # 关闭请求缓冲（POST body） proxy_http_version 1.1; # HTTP/1.1 支持长连接 proxy_set_header Connection ""; # 清除 Connection 头 proxy_read_timeout 600s; # 超时设长（AI 可能很久） chunked_transfer_encoding on; # 确保 chunked 编码 add_header X-Accel-Buffering no; # 额外保险：告诉Nginx不缓冲 } ``` **元凶 2：Spring WebFlux 的 `Flux.buffer()` 或 `window()`** 有些开发者为了减少前端渲染频率，在 Flux 上加了 `buffer` 操作符把多个 token 攒成一批： ```java // ❌ 这样会把 token 攒成 List，破坏了实时性 return chatClient.prompt() .user(q) .stream() .content() .buffer(10) // 攒够10个token才发一次！ .map(list -> String.join("", list)); ``` 如果确实需要降低前端渲染频率，应该在**前端用防抖**（前面 6.4 节的 50ms 防抖方案），而不是在后端攒数据。后端应该每个 token 就立即推送出去，前端的防抖只影响渲染频率不影响数据到达。 **元凶 3：Python 中的 `text/event-stream` 缺少 `X-Accel-Buffering: no`** ```python # ❌ 没有禁用缓冲头 return StreamingResponse( generator(), media_type="text/event-stream" ) # ✅ 显式禁用 Nginx 缓冲 return StreamingResponse( generator(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # 关键！告诉 Nginx 不要缓冲这个响应 } ) ``` ### 9.8 踩坑 #4：中文字符被截断成乱码 **现象**：前端偶尔收到乱码，如 `\u4f60\u597`（`你` 的 Unicode 被截断）或者显示为 `?` 的方块。 **根因**：SSE 数据是字节流，中文字符在 UTF-8 编码下占 3 个字节。如果网络传输时一个中文字符的字节被分在两个 chunk 里（比如第一个 chunk 包含前 2 个字节，第二个 chunk 包含第 3 个字节），`TextDecoder` 在解码时如果不知道这是不完整的字符，就会出错。 **解决：`TextDecoder` 的 `stream: true` 参数** ```javascript const decoder = new TextDecoder('utf-8'); // ❌ 不传 stream 参数，遇到不完整字符会输出替换字符 const text = decoder.decode(chunk); // ✅ 传入 stream: true，遇到不完整字符会保留在内部缓冲区 // 等下次 decode 时拼接完整 const text = decoder.decode(chunk, { stream: true }); ``` `{ stream: true }` 告诉解码器："这次解码可能不是最后一次，遇到不完整的 UTF-8 序列先存着，等下次输入来了再拼。" 这样中文字符即使跨 chunk 也不会乱码。 **Python 端也需注意**：FastAPI 的 `StreamingResponse` 在发送字符串时会自动编码为 UTF-8，但如果你的 yield 内容中有非 UTF-8 字符（比如 GBK 编码的旧数据），也会导致乱码。确保所有字符串都是 UTF-8： ```python # ✅ 确保字符串是 UTF-8 content = token.encode('utf-8').decode('utf-8') yield f"data: {json.dumps({'content': content}, ensure_ascii=False)}\n\n" # ensure_ascii=False 让中文不被转义为 \uXXXX，直接输出 UTF-8 中文 ``` ### 9.9 踩坑速查表把上面所有踩坑场景整理成一张速查表，方便开发时对照检查： | 症状 | 可能原因 | 解决方案 | |------|---------|---------| | 换行/空格丢失 | 后端 `.trim()` / `.strip()` | token 原样传递，不做任何 trim | | 换行/空格丢失 | 前端 SSE 解析 `.trim()` | 只跳过 `data:` 后第一个空格 | | 段落挤在一起 | JSON 中 `\n` 被转义 | 确保前端 `JSON.parse()` 还原换行符 | | 代码块无高亮 | rehype-highlight 遇到不完整代码报错 | `ignoreMissing: true` + `detect: true` | | 渲染闪烁跳变 | 每个 token 全量重解析 | 防抖 50ms + 增量解析（按 `\n\n` 分块） | | 数据一大段到达 | Nginx `proxy_buffering` | `proxy_buffering off` + `X-Accel-Buffering: no` | | 数据一大段到达 | Spring Flux `buffer()` | 去掉 `buffer`/`window`，前端防抖 | | 中文乱码 | `TextDecoder` 未开 stream | `decoder.decode(chunk, { stream: true })` | | 中文显示为 \uXXXX | Python `json.dumps` 默认转义 | `json.dumps(..., ensure_ascii=False)` | | 连接频繁断开 | 代理超时 / 无心跳 | `proxy_read_timeout 300s` + 心跳 | | EventSource 不触发 | 事件后没有 `\n\n` | 确保每个事件以空行结束 | | String 带了多余引号 | Spring 自动 JSON 序列化 | 使用 `ServerSentEvent` 明确指定 | ```mermaid graph TD BUG["流式输出有问题"] --> Q1{"数据到达是实时的吗?
看 Network EventStream"} Q1 -->|否,一大段到达| BUF["缓冲问题"] Q1 -->|是,但渲染不对| Q2{"格式正确吗?
换行/空格在吗?"} Q2 -->|格式错误| FMT["格式丢失问题"] Q2 -->|格式正确| Q3{"有乱码吗?"} Q3 -->|有乱码| ENC["编码问题"] Q3 -->|无乱码| Q4{"渲染闪烁吗?"} Q4 -->|闪烁| REND["渲染策略问题"] BUF --> S1["检查Nginx proxy_buffering
检查Flux buffer操作符
检查X-Accel-Buffering头"] FMT --> S2["检查后端 trim/strip
检查前端 data: 解析
检查 JSON 序列化方式"] ENC --> S3["TextDecoder stream:true
json.dumps ensure_ascii=False"] REND --> S4["防抖50ms
增量Markdown解析
按\\n\\n分块渲染"] style BUF fill:#FFCDD2,stroke:#D32F2F style FMT fill:#FFCDD2,stroke:#D32F2F style ENC fill:#FFCDD2,stroke:#D32F2F style REND fill:#FFCDD2,stroke:#D32F2F ``` --- ## 十、知识体系总结 ```mermaid mindmap root((流式输出
知识体系)) 协议层 HTTP 长轮询模拟推送开销大延迟高 SSE 服务器单向推送基于标准HTTP 自动重连纯文本格式浏览器原生支持 WebSocket 全双工通信协议复杂需手动重连 gRPC流二进制高效浏览器不支持原生 SSE为什么流行 AI对话天生单向工程简单部署友好自动重连生态标准 SSE协议细节 Content-Type: text/event-stream Transfer-Encoding: chunked 事件格式: data/event/id/retry 空行分隔事件注释行做心跳 Java实现 Spring WebFlux Flux 响应式流 ServerSentEvent 包装自动SSE编码 Spring AI ChatClient.stream LangChain4j TokenStream 回调链 Flux 适配 Flux操作符 map/filter/concat onErrorResume/timeout Python实现 FastAPI StreamingResponse 异步生成器 sse-starlette EventSourceResponse 内置心跳 OpenAI SDK stream=True for chunk in stream LangChain model.stream astream_events 前端消费 EventSource 简单 GET 自动重连 fetch + ReadableStream 支持 POST 灵活控制 AbortController 中断实时渲染组件 react-markdown + rehype markdown-it + highlight.js Shiki vs highlight.js 增量解析防闪烁防抖渲染策略工程实践 Nginx关闭缓冲心跳保活错误降级处理超时配置连接数管理踩坑实录空格换行丢失后端trim/strip 前端data解析trim JSON转义换行符 Markdown渲染闪烁不完整语法解析增量分块渲染数据不实时推送 Nginx缓冲 Flux buffer操作符中文截断乱码 TextDecoder stream模式 ensure_ascii=False ``` --- ## 写在最后流式输出这件事，表面上看是"服务器一个字一个字地推给浏览器"，但往深了挖，它涉及网络协议选型、响应式编程范式、前后端协作模式、AI 模型 API 对接、反向代理配置、错误处理策略…… 这篇文章试图从最底层的"为什么传统 HTTP 不行"开始，一路讲到协议对比、SSE 原理细节、Java 和 Python 的实战代码、前端消费方式、Nginx 配置。如果你一路读到这里，应该已经具备了从零搭建一个生产级 AI 流式输出系统的全部知识。记住几个核心判断： - **场景是单向推送 → 选 SSE，别上 WebSocket** - **Java 后端 → Flux + Spring WebFlux，天然和 SSE 配合** - **Python 后端 → FastAPI + 异步生成器 + StreamingResponse** - **前端 → fetch + ReadableStream 比 EventSource 更灵活** - **渲染 → react-markdown + 防抖 + 增量解析，别每个 token 全量重绘** - **踩坑 → token 不 trim、data 只跳一个空格、Nginx 关缓冲、TextDecoder 开 stream 模式** - **部署 → Nginx 关缓冲、加心跳、设长超时** 把这几条记住，剩下的就是根据具体业务场景做封装和调整了。

herongwei