SSE流式输出：让大模型回答像打字一样实时呈现

可视化演示

展示SSE流式输出中Token逐字推送过程、SSE与WebSocket单向/双向通信对比，以及EventSource事件流格式解析。

为什么需要流式输出

你有没有注意过，在 ChatGPT 问世之前，几乎所有的“AI 问答”功能都是一个死板的流程：你点击“提交”按钮，页面转圈，三五秒甚至十几秒后，一整段文字“啪”地出现在屏幕上。可 ChatGPT 不一样——它像一位打字飞快的秘书，你在屏幕上看着一个字一个字被打出来。这两种体验的差别，就像“在餐厅里饿着肚子等一桌菜上齐”和“坐在吧台前看着寿司师傅现捏现送”。

这种体验上的革命，背后就是流式输出技术。传统 HTTP 请求是一次性的：客户端发一个请求，服务器把所有数据处理完，再一次性返回完整响应。在这个模型下，大模型生成 500 字回答需要 10 秒，用户就必须对着转圈的空屏等 10 秒。注意，这 10 秒里用户其实完全无事可做——而这 10 秒恰恰是“对话感”消失的元凶。

流式输出彻底改变了这个体验。服务器一旦开始生成内容，就立刻把已经生成的片段发送给客户端，用户看到的不是“一张白纸等 10 秒”，而是“屏幕上第 1 秒就开始蹦字”。心理学上有个概念叫“感知性能”——用户对等待的忍受度，往往取决于有没有进度反馈。流式输出让“等待”变成了“观看过程”，把 10 秒的煎熬变成 10 秒的期待。

所以，流式输出解决的不是“生成速度问题”，而是“等待体验问题”——它把后厨悄悄备菜，变成了开放厨房里的现场料理秀。对你未来的 AI 产品而言，这可能就是用户留存率的分水岭。

SSE 协议核心机制

想象这条传送带本身：它看似一直在动，但物理结构只是一条固定在吧台上的滚轴带，不需要每次传菜都重新架一条。SSE 就是这条固定的传送带——客户端和服务器之间只建立一个 HTTP 连接，然后这个连接一直保持着，服务器源源不断地把数据放在上面推给客户端。

SSE 全称 Server-Sent Events（服务器发送事件），名字直白：就是“服务器发送过来的事件”。它基于普通的 HTTP 协议，但使用了 Content-Type: text/event-stream 这个特殊的 MIME 类型来标记响应。一旦浏览器看到这个类型，就知道“这不是一次性响应，而是会持续推送的流”。连接建立后，服务器可以随时向客户端发送消息——注意，是“随时”，不需要客户端再次请求。

SSE 的消息格式非常有规律，像传送带上的寿司碟子，每碟都有固定的摆盘规则。一条 SSE 消息由若干字段行组成，最常见的是 data: 字段；消息之间用空行分隔。举个最简例子：

这一行 data: 你好 加一个空行，就是一条完整的 SSE 消息。服务器可以持续发送多个这样的块。浏览器端通过 EventSource API 接收，每收到一条 data 消息，就会触发一个 onmessage 回调。一个用起来异常简单的 API——不需要处理复杂的状态码，不需要握手升级协议。

SSE 还内置了两个贴心机制：自动重连和 Last-Event-ID。如果连接意外断开，浏览器会自动重新发起连接；服务器还可以通过 id: 字段告诉客户端“你断到哪了”，重连时客户端带上这个 ID，服务器就能从断点续传。这让 SSE 在弱网环境下依然有不错的鲁棒性——传送带断了，拼接好又继续转。

data: 你好

SSE 与 WebSocket 对比

选通信协议，有点像选择用餐方式：如果你只是坐在吧台前等主厨一道道把菜递给你，SSE 的“单向传送带”绰绰有余；但如果你需要跟主厨讨论“这道菜少放盐、那道菜多加点芥末”，就需要一种双向对话机制——这正是 WebSocket 的用武之地，它是一条允许双方随时往对方盘子里夹菜的双向通道。

先说协议层面。WebSocket 是一个独立的协议（ws:// 和 wss://），它需要在 HTTP 之上进行一次“升级握手”，把协议从 HTTP 切换为 WebSocket 协议。而 SSE 本质还是 HTTP——它没有协议升级，只是把连接的响应方式从“一次性”变成了“持续推送”。这个差异带来一个关键后果：SSE 天然兼容现有的 HTTP 基础设施（比如负载均衡、防火墙通常都放行 HTTP），而 WebSocket 在某些代理环境里可能会被拦。

再说通信模式。WebSocket 是双向的，客户端和服务器都可以随时发起消息；SSE 是单向的，只能服务器向客户端推。这带来一个微妙的工程权衡：SSE 的模型极简——你不需要处理来自客户端的随时消息，少了一整类并发问题。但代价是：如果你的应用场景需要客户端频繁向服务器发消息（比如多人实时协作编辑器），SSE 就得再开一个普通 HTTP 通道配合使用，绕了一个弯。

WebSocket 就像一个高级的双方对讲机，什么都好，但你要是用它来单向喊话，就杀鸡用了牛刀。SSE 用最小的机制做最合适的事——对 LLM 应用而言，模型生成回答本来就是单向的，SSE 是性价比最高的选择。

Python 实现 SSE 接口

纸上谈兵容易，真正的问题来了：怎么在你的 Python 后端里端出这条传送带？现代 Web 框架几乎都支持 SSE。这里我们用 FastAPI 来演示——它原生支持异步流式响应，语法异常简洁。FastAPI 是一个基于 Python 类型注解的现代 Web 框架，相当于餐厅的中央厨房管理系统，你只需要告诉它“这道菜走传送带上”。

SSE 的原理不复杂，实现也谈不上尖端，但真正的难点在于正确设置响应头和用生成器产出数据。核心点：生成器是一种惰性求值的函数对象——普通函数是一次性把所有结果算完返回，而生成器一次只产出（yield）一个值，然后暂停。当你在生成器里循环遍历大模型的输出时，每个 yield 都会立即把内容推送到客户端，这就是 SSE 流式的魔法所在。

服务端代码的核心骨架如下：定义一个生成器函数，在循环中 yield 出符合 SSE 格式的字符串（即 data: ...\n\n），然后用 StreamingResponse 包装给 FastAPI——这个类就像一个贴好标签的传菜窗口，它知道要用 text/event-stream 来标识这次响应。客户端收到这个响应头后，就知道“这条连接会持续上菜，不会等一次送完”。

就这么简单——SSE 接口的实现本质是一个生成器 + 一个响应头。但要注意，这只是“能用”的起点。实际生产环境还需要处理客户端断开、代理缓冲、并发控制等问题，这些在后面的章节展开。先把这个骨架搭出来跑通，再持续加固。

SSE 在 LLM 应用实战

现在你已经掌握了传送带的原理，是时候看看它在真实餐厅里的运转了。最经典的场景当然是大模型对话应用——ChaiGPT、Claude、文心一言的网页界面，你输入一句话后看到的逐字输出，底层几乎都是 SSE 在发力。这条传送带从 Llama 或 GPT 模型推理服务开始，把一个个 token（最小文本单位）传送到你的浏览器上。

具体流程拆解：前端用 EventSource 或 fetch 的流式模式发起请求；后端收到请求后向 LLM 推理服务发出调用，两者之间通常也走 SSE；后端拿到模型的 token 流后，用生成器逐字取出内容，再以 SSE 格式转发给前端——这就像传送带在中转站接驳，一路直达用户。前端每收到一条 data: 消息，就追加一段文字到页面上，于是用户看到的就是“AI 在思考中打字”。

SSE 的另一个实用战场是实时进度上报。想象你有一个“文章摘要批处理系统”，用户提交 100 篇文档，系统要逐一处理。传统做法是提交后页面上一个进度条旋转半天；用 SSE，服务器每处理完一篇文档就推送一条进度消息，前端实时更新百分比，用户体验瞬间提升一个档次。这种用法不需要大模型，但 SSE 的单向推送特性在这里恰如其分。

掌握了这条链路，你就能自己动手做一个和大厂体验相当的 AI 对话应用了。传送带就是消息通道，LLM 就是主厨，而你要做的就是把两者接得溜光顺滑。

SSE 局限与最佳实践

传送带也不是万能的。如果有一天客人想跟主厨实时对话、调整口味，单向的传送带就乏力了——因为 SSE 只能从服务器到客户端单向流动，客户端无法通过同一条连接给服务器发消息。这意味着你不能用 SSE 做聊天室、多人协作白板这类需要双向高频交互的应用。要双向，还是得用 WebSocket，别硬用 SSE。

另一个隐蔽的坑在网络代理的缓冲问题。很多企业级代理/Nginx 默认会缓冲 HTTP 响应，等积累到一定量才转发给客户端——这在普通响应时没问题，但对 SSE 来说，等于传送带被中间人囤货了，前端迟迟收到不到 data 消息。解决方法是在 Nginx 中显式关闭缓冲：proxy_buffering off;。这提醒我们：SSE 依赖“边生成边推送”，任何环节的缓冲都可能导致体验毁于一旦。

最后，说一下断线重连策略和心跳机制。浏览器 EventSource 自带自动重连，但服务器端需要配合：客户端重连时，服务器应该知道它断到了哪条消息，这靠 SSE 的 id: 字段实现。同时，如果连接长期空闲（比如大模型在思考），某些中间设备可能把空闲连接当废连接掐掉，所以需要给连接发送心跳注释（:\n\n），告诉代理“我还在活着呢”。这跟寿司师傅为防止客人离席，时不时往传送带上放一碟姜片传递“营业中”信号是一个道理。

要在生产环境用好 SSE，核心原则就是尊重传输路径上的每一个环节。代理要关缓冲，负载均衡要支持长连接超时配置，服务端要发心跳；只有把这些螺丝都拧紧，传送带才能昼夜不息地旋转。

AI 应用场景

Python 演示

import jsonimport timefrom fastapi import FastAPIfrom fastapi.responses import StreamingResponse app = FastAPI()  def mock_llm_stream(prompt: str):    """模拟大模型流式生成, 按 token 逐个产出回答片段"""    full_answer = f"你好, 你刚才问的是: {prompt}。让我为你详细解答..."    for char in full_answer:        time.sleep(0.05)  # 模拟 50ms 生成一个字        yield char  def format_sse(data: str) -> str:    """将数据封装成 SSE 格式"""    return f"data: {data}\n\n"  @app.get("/chat/stream")def chat_stream(prompt: str):    """SSE 流式对话接口"""    generator = mock_llm_stream(prompt)     def event_stream():        for chunk in generator:            payload = json.dumps({"token": chunk}, ensure_ascii=False)            yield format_sse(payload)     return StreamingResponse(        event_stream(),        media_type="text/event-stream",        headers={            "Cache-Control": "no-cache",            "Connection": "keep-alive",            "X-Accel-Buffering": "no",        },    )  if __name__ == "__main__":    import uvicorn     uvicorn.run(app, host="0.0.0.0", port=8000)

1. mock_llm_stream(prompt)模拟大模型推理输出流，先拼出完整句子再用 time.sleep(0.05) 逐字吐出，真实场景中这里是调用 vLLM/OpenAI 的流式接口。
2. format_sse(data)SSE 格式的封装函数，核心是 data: 内容\n\n 格式。这是整个协议的字面核心，前端依赖空行识别一条完整消息。
3. event_stream()逐字取出生成器内容，包装为 JSON 格式后按 SSE 格式产出。
4. StreamingResponse(...)FastAPI 的关键封装：接收生成器对象并保持 HTTP 连接不断开，返回流式响应。
5. media_type="text/event-stream"设置 SSE 特有的 MIME 类型，浏览器和前端库据此知道该用流式方式解析。
6. "Cache-Control": "no-cache"禁用缓存，防止代理或浏览器缓存 SSE 响应导致数据延误。
7. "X-Accel-Buffering": "no"专门针对 Nginx 的响应头，关闭代理缓冲，让数据立即传送，不被囤积。

常见误区

面试题