还在用WebSocket做LLM流式传输？FastAPI + SSE让你少踩一半坑

你以为实时通信必须用WebSocket？其实SSE才是隐藏的王者

“这个LLM对话页面的流式输出怎么老断？”
“负载均衡又报WebSocket升级失败……”
“粘性会话没配好，用户被路由到不同节点，对话上下文丢了！”

如果你正在用WebSocket给LLM应用做token流式传输，上面这些坑你大概率踩过。WebSocket确实能干活，但它带来的麻烦也不少：连接升级、负载均衡配置、代理拦截、断线重连……尤其是在生产环境，每一样都可能让你半夜爬起来修bug。

但你可能不知道，绝大多数LLM流式传输场景，根本不需要WebSocket。

今天我要给你介绍一个更简单、更稳定的方案：Server-Sent Events（SSE）。它基于普通HTTP，浏览器原生支持，还能自动重连。配合FastAPI的异步能力，几行代码就能实现丝滑的token流式传输。更重要的是，它能让你的部署架构回归简单，告别那些烦人的WebSocket基础设施问题。

本文会带你从零实现一个生产可用的SSE流式接口，并深入探讨SSE在实时通知、性能优化等方面的最佳实践。读完你会发现，原来实时通信可以这么清爽。

SSE是什么？用一个外卖订单的比喻让你秒懂

想象你在外卖App上点了一份餐。WebSocket相当于你骑手的专属电话——你可以随时打电话问骑手到哪了，骑手也可以随时打电话告诉你“快到了”。双向沟通，非常灵活，但每个骑手都需要一个电话线路，管理起来麻烦。

而SSE（Server-Sent Events） 更像是一个订单状态推送服务：你下单后，系统会不断给你发短信“商家已接单”“骑手已取餐”“距您还有1公里”……你只需要收短信，不需要回复。如果信号不好漏掉一条，系统会自动重发。

正式定义：SSE是HTML5标准的一部分，允许服务器通过HTTP连接向客户端推送数据。客户端通过JavaScript的EventSource接口监听事件，连接断开时会自动重连，服务器可以给每个事件指定ID，确保断点续传。

相比WebSocket，SSE最大的特点就是单向：只能服务器→客户端。但这恰恰完美匹配LLM token流式传输和实时通知的场景——客户端只需要接收服务器吐出来的文字。

对比一下：

FastAPI + SSE实战：流式输出LLM token

环境准备

确保你安装了Python 3.10+和FastAPI 0.115+（本文代码基于这些版本测试）。还需要一个ASGI服务器，比如Uvicorn：

pip install fastapi uvicorn

核心实现：StreamingResponse + 异步生成器

FastAPI的StreamingResponse可以直接返回一个异步生成器，这正是SSE需要的——持续不断地输出数据。我们按SSE协议格式组装消息：

• 响应头：Content-Type: text/event-stream
• 每个事件由若干行组成，以空行分隔
• 常用字段：event: 事件类型，data: 数据（JSON格式），id: 事件ID
• 注释行（以: 开头）可用来发送心跳，不会触发客户端事件

下面是一个完整的SSE端点，模拟LLM流式输出：

import asyncio
import json
import time
from typing import AsyncGenerator
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
app = FastAPI()
defformat_sse(event: str, data: dict) -> str:
"""将Python对象格式化为SSE消息"""
returnf"event: {event}\n" + f"data: {json.dumps(data, ensure_ascii=False)}\n\n"
asyncdeffake_llm_stream(prompt: str) -> AsyncGenerator[str, None]:
"""模拟LLM流式生成token，实际可替换为OpenAI、智谱等API"""
# 这里是一个模拟的token流
for token in ["好的", "，", "这是", "你", "的", "回答", "……"]:
await asyncio.sleep(0.1)  # 模拟生成延迟
yield token
@app.get("/stream")
asyncdefstream(prompt: str, request: Request):
asyncdefevent_generator():
# 可选：发送一个元事件，让客户端知道连接已建立
yield format_sse("meta", {"status": "started", "ts": time.time()})

        last_ping = time.time()

try:
asyncfor token in fake_llm_stream(prompt):
# ⚠️ 关键检查：如果客户端断开连接，立即停止生成
ifawait request.is_disconnected():
break

yield format_sse("token", {"t": token})

# 心跳：每15秒发送一个注释行，防止代理超时
                now = time.time()
if now - last_ping > 15:
yield": ping\n\n"# 注释行，不会触发事件
                    last_ping = now

# 流结束事件
yield format_sse("done", {"status": "completed", "ts": time.time()})
except Exception as e:
# 错误处理：不要把敏感信息发出去
yield format_sse("error", {"message": "stream_failed", "detail": str(e)[:200]})

    headers = {
"Cache-Control": "no-cache",
"Connection": "keep-alive",
# 关键：告诉nginx等代理不要缓冲
"X-Accel-Buffering": "no",
    }
return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers=headers
    )

代码解释：

• format_sse 辅助函数确保格式正确，注意最后的两个换行符不可省略。
• event_generator 内部使用异步生成器，每次yield都会立刻通过网络发送出去。
• request.is_disconnected() 是FastAPI提供的方法，用于检测客户端是否已关闭连接。如果不检查，用户关闭页面后服务器还在拼命生成token，白白浪费CPU。
• 心跳注释行 : ping\n\n 是SSE协议允许的，它可以保持连接活跃，而不会触发客户端的onmessage事件。
• X-Accel-Buffering: no 是给nginx的指令，禁止它对响应进行缓冲。如果不加，nginx可能会攒一堆数据再一起发，流式效果就没了。

浏览器客户端：简单到难以置信

SSE的客户端API是浏览器内置的，无需任何第三方库：

<preid="output"></pre>
<script>
const prompt = encodeURIComponent("写一首关于代码的诗");
const es = new EventSource(`/stream?prompt=${prompt}`);
const output = document.getElementById('output');
let text = '';
  es.addEventListener('token', (e) => {
const { t } = JSON.parse(e.data);
    text += t;
    output.textContent = text;
  });
  es.addEventListener('done', () => {
    es.close(); // 关闭连接，释放资源
console.log('流式输出完成');
  });
  es.addEventListener('error', (e) => {
console.error('连接出错，自动重连中...', e);
// 如果不想自动重连，可以调用 es.close()
  });
</script>

对比WebSocket：不需要管理重连逻辑，不需要处理二进制帧，不需要担心握手失败。EventSource在连接断开后默认会等待几秒后自动重连，而且如果服务器发送了id字段，重连时会自动带上Last-Event-ID头，让服务器可以从断点继续发送，避免消息丢失。

扩展思考：不止LLM，实时通知也能轻松搞定

SSE的应用场景远不止LLM流式传输。任何需要服务器主动推送的场景，比如实时通知、仪表盘数据更新、股票行情，都可以用SSE优雅实现。

实战：基于Redis Pub/Sub的广播通知

假设我们有一个需要向所有在线用户推送通知的系统。当后台任务完成、新消息到达时，我们希望所有订阅了通知的用户实时收到。架构如下：

[后台任务] → Redis发布消息 → [所有FastAPI进程订阅] → [通过SSE推送给各自连接的客户端]

使用FastAPI+Redis实现：

import aioredis
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
app = FastAPI()
# 假设已创建Redis连接池
redis = aioredis.from_url("redis://localhost")
asyncdefevent_stream(user_id: str):
    pubsub = redis.pubsub()
await pubsub.subscribe("global_notifications")  # 订阅全局频道
try:
asyncfor message in pubsub.listen():
if message['type'] == 'message':
                data = json.loads(message['data'])
# 可以按user_id过滤，也可以广播给所有人
yield format_sse("notification", data)
finally:
await pubsub.unsubscribe("global_notifications")
@app.get("/notifications/{user_id}")
asyncdefnotifications(user_id: str):
return StreamingResponse(event_stream(user_id), media_type="text/event-stream")

任何地方需要推送通知时：

await redis.publish("global_notifications", json.dumps({"text": "新订单！", "user_id": 123}))

所有连接到/notifications/的客户端都会立即收到这条消息。你可以用类似的方式实现更精细的按用户隔离（每个用户订阅自己的频道）。

⚠️ 注意：多进程部署时，每个进程都有自己的Redis订阅，因此所有进程都能收到消息并推送给自己的客户端。这要求你的Redis是共享的，且进程数不能太多（否则Redis连接数暴涨）。大规模时可考虑用Kafka或NATS替代。

生产环境避坑指南

SSE虽然简单，但在生产环境中仍有几个容易踩的坑，我帮你提前列出来：

1. 代理缓冲会毁掉流式效果

Nginx默认会缓冲后端响应，直到攒够一定大小才发给客户端。这对于SSE是灾难性的——用户会看到文字“一顿一顿”地出现。解决方案：

• **在FastAPI返回头中添加 X-Accel-Buffering: no**（仅对nginx有效）
• 或者在nginx配置中针对SSE路径关闭缓冲：

location /stream {
    proxy_buffering off;
    proxy_cache off;
    proxy_set_header X-Accel-Buffering no;
    # 其他代理设置...
}

对于其他代理（如ALB），也需要查阅文档禁用缓冲。

2. 超时问题：心跳+适当配置

许多负载均衡器有默认的60秒空闲超时。如果LLM思考时间超过60秒（比如复杂推理），连接就会被切断。解决：

• 发送心跳注释行，如上文代码中的 : ping\n\n，每隔15-30秒发一个。
• 调整代理的超时设置，例如nginx可以设置proxy_read_timeout 300s;。

3. 客户端断开检测

这是最容易忽视的一点。如果不检查request.is_disconnected()，用户关闭页面后，服务器可能还在后台继续生成token，直到整个流结束。这不仅浪费资源，在计费模式下还可能导致账单飙升。务必在每个token生成前检查连接状态。

4. 多进程共享状态

SSE连接是进程级别的。如果你有多个FastAPI worker进程，每个进程只能向自己维护的客户端推送消息。要实现跨进程广播，必须引入一个中央消息总线（Redis、Kafka等）。上面Redis Pub/Sub的例子就是为此设计的。

5. HTTP/2与连接数限制

在HTTP/1.1下，浏览器对同一域名的并发连接数有限制（通常是6个）。如果你的页面需要同时打开多个SSE连接（比如多个监控面板），可能会达到上限。解决方案：

• 使用HTTP/2，它支持多路复用，一个连接可以承载多个SSE流。
• 将不同业务的SSE端点分散到不同子域名（如notifications.example.com、stream.example.com）。

6. CORS与认证

如果前端和后端不同域，需要配置CORS。FastAPI的CORSMiddleware支持SSE：

from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://yourfrontend.com"],
    allow_credentials=True,  # 允许携带cookie
    allow_methods=["GET"],
)

客户端需要携带认证信息时，使用：

const es = new EventSource(url, { withCredentials: true });

这样浏览器会自动发送该域下的cookie，实现会话认证。

性能调优建议

如果你的SSE服务需要支撑大量并发连接（比如上万用户），以下几点可以帮助你榨干服务器性能：

1. 使用ASGI服务器：Uvicorn搭配--workers可以启动多个进程。注意每个进程能支撑的连接数受限于操作系统的文件描述符限制，适当调高ulimit -n。
2. 压缩响应：虽然SSE是流式输出，但每个消息仍可以启用gzip压缩。在nginx层配置压缩，或者使用FastAPI的GZipMiddleware，注意text/event-stream通常需要显式添加到压缩类型列表。
3. 批量发送：如果事件产生频率极高（如每秒几十次），可以考虑将多个数据打包成一个事件（JSON数组），减少HTTP帧开销。但要注意客户端解析逻辑需相应调整。
4. 避免内存泄漏：确保每个SSE生成器正确释放资源（如Redis订阅、数据库连接）。使用try...finally块或异步上下文管理器。

核心回顾

1. 原理：SSE基于普通HTTP，通过text/event-stream和分块传输实现服务器推送，浏览器EventSource自动处理重连和事件解析。
2. 实践：FastAPI的StreamingResponse配合异步生成器可以轻松实现SSE，关键点包括格式组装、心跳、断开检测、代理缓冲控制。
3. 避坑：代理缓冲、超时、客户端断开检测、多进程状态共享是生产环境最容易出问题的地方。

写在最后

过去几年，WebSocket几乎成了实时通信的代名词，以至于我们遇到需要推送的场景，第一反应就是上WebSocket。但技术选型应该回归本质：你的需求到底是什么？如果是服务器单向推送，SSE可能是更简单、更稳定的选择。它不需要额外协议，不需要复杂的连接管理，部署起来和普通HTTP接口无异。

我见过不少团队因为一开始用了WebSocket，后来被各种基础设施问题折磨，最后不得不重构。而SSE就像那个不起眼但永远可靠的同事，平时不显山露水，关键时刻从不掉链子。

当然，SSE不是银弹。如果你需要真正的双向通信（比如聊天、游戏），WebSocket依然是更好的选择。但对于LLM流式输出、实时通知这类场景，不妨试试SSE——你可能会发现，原来实时通信可以这么清爽。

你在项目中用过SSE吗？如果没用过，你会在哪些场景考虑用它替代WebSocket？欢迎留言分享你的看法。