#SSE 与 WebSocket：AI 应用中的流式传输选型

阅读提示 本笔记面向已有 HTTP 与 FastAPI 基础、正在做 LLM / Agent 应用的后端开发者，不假设你写过流式接口。

主线是一个问题：同样是「把模型输出实时推给前端」，什么时候 SSE 够用，什么时候必须上 WebSocket？ 顺序是：先讲 AI 为什么必须流式 → 两种协议各自怎么跑起来 → 在 AI 场景下它们到底在哪几个维度分道扬镳 → 踩过的坑 → 决策规则。

相关笔记：Python 异步编程（async for 与事件循环）、FastAPI 请求全链路原理（响应对象与中间件顺序）。

符号约定：⚠️ = 常见陷阱　🆚 = 对比辨析　💡 = 选择建议

#目录

• 一、为什么 AI 应用绕不开流式
• 二、三种做法的坐标系
• 三、SSE：报文格式与 FastAPI 实现
• 四、WebSocket：握手、帧与 FastAPI 实现
• 五、AI 场景下真正拉开差距的六个维度
• 六、常见陷阱
• 七、怎么选
• 八、动手练习
• 速查表
• 复习重点

核心概念 SSE（Server-Sent Events）本质是「一个迟迟不结束的 HTTP 响应」：请求发出去之后服务端不关闭响应体，持续往里写文本，靠空行分帧。它没有发明新协议，所以代理、CDN、鉴权、日志这些既有设施全都认。

WebSocket 本质是「用 HTTP 握手换来一条裸 TCP 双工通道」：101 Switching Protocols 之后，报文不再是 HTTP，而是 WS 帧，双方随时可发，且支持二进制。

在 AI 应用里，两者的分界线不是「快不快」（都够快），而是三件事：会话中要不要持续上行、要不要传二进制、服务端要不要推与本次请求无关的消息。三个都不要 → SSE；任意一个要 → WebSocket。

#一、为什么 AI 应用绕不开流式

一句话：LLM 的首字延迟（TTFT）远小于全文生成时间，不流式就是在白白浪费这段差值。

一次典型的 8B 模型生成 500 字回答：

差别不只是体感。流式还带来三个工程收益：

1. 用户能提前判断跑偏并打断，省下后面 400 个 token 的钱；
2. 不容易撞上网关超时 —— 很多 LB 的默认响应超时是 30/60 秒，非流式的长回答会被直接掐断；
3. Agent 可以把中间步骤（正在检索 / 正在调用工具）实时暴露出来，否则用户面对 40 秒白屏只会以为挂了。

所以问题从来不是「要不要流式」，而是「用什么姿势流」。

#二、三种做法的坐标系

在选型之前，先把三种做法放在同一张时序图上看，重点看连接活了多久和箭头朝哪边：

读图顺序：从左到右三栏，每栏两条竖虚线代表浏览器和服务端，竖线越长代表连接活得越久。

• 轮询每次都是一条新连接，所以每次都要重付 TCP + TLS + 鉴权的开销，而且延迟下限就是轮询间隔；
• SSE 只有开头一个上行箭头，之后全是下行 —— 这就是「单向」的准确含义：不是不能上行，而是连接建立之后不能再上行；
• WebSocket 握手线以下双向箭头混杂，那个橙色的 {"cmd":"stop"} 是 SSE 做不到的动作。

「SSE 是单向的」这句话要说准确 发起 SSE 请求时你完全可以用 POST 带一个很大的 body（整段对话历史、RAG 上下文都塞得下）。受限的是建立之后：这条连接的上行方向已经用完了，想再说话只能另开一个 HTTP 请求。

很多人把「单向」误解成「只能 GET、参数只能放 URL」，进而以为长 prompt 必须用 WebSocket —— 这是错误迁移。原生 EventSource 确实只能发 GET，但那是浏览器 API 的限制，不是 SSE 协议的限制（见 5.2 鉴权：两个对称的坑）。

#三、SSE：报文格式与 FastAPI 实现

#3.1 报文长什么样

SSE 的全部规范可以浓缩成一句话：纯文本，一行一个字段，空行代表「这一帧完了，交给前端」。

四个字段里，日常只会用到两个半：

以 : 开头的行是注释，浏览器会忽略但连接保持活跃 —— 这就是标准的心跳写法：: ping\n\n。

#3.2 FastAPI 最小实现

import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
 
app = FastAPI()
 
class ChatIn(BaseModel):
    prompt: str
 
async def event_stream(request: Request, prompt: str):
    """把 LLM 的 token 流包装成 SSE 帧。"""
    async for chunk in llm.astream(prompt):        # 你的模型客户端
        # ① 每轮都检查客户端是否已经跑了
        if await request.is_disconnected():
            break
        # ② JSON 编码是刚需，不是讲究 —— 见 3.3
        payload = json.dumps({"t": chunk}, ensure_ascii=False)
        yield f"data: {payload}\n\n"               # ③ 结尾必须是两个 \n
    yield "data: [DONE]\n\n"
 
@app.post("/chat")
async def chat(request: Request, body: ChatIn):
    return StreamingResponse(
        event_stream(request, body.prompt),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",             # ④ 让 Nginx 别缓冲，见第六节
        },
    )

四个标号里，③ 和 ④ 是新手最常漏的：少写一个 \n 前端就一个字都收不到，少写 X-Accel-Buffering 在本地一切正常、上了线全变成「转圈 20 秒然后一次性刷出来」。

#3.3 为什么必须 JSON 编码，以及大厂为什么都用 [DONE]

这是 SSE 在 AI 场景最容易出事的地方，单独拎出来说。

🔴 直接把 token 拼进 data: 会劈坏帧 错误操作：

yield f"data: {chunk}\n\n"     # ❌ chunk 直接进去

实际结果： 平时都好好的，一旦模型开始输出代码块或分段落，前端就收到半截 JSON、乱序内容，甚至连接直接结束。

原因： 分帧规则是「空行结束一帧」。当模型吐出的 chunk 本身含有 \n\n（写代码、列清单时必然发生），拼出来的字节流就变成：

data: def foo():⏎
⏎                    ← 这里被浏览器当成帧结束了
    return 1⏎        ← 这行不以 data: 开头，直接被丢弃

内容不但被截断，还被静默吞掉一部分 —— 没有任何报错，最难查。

正确做法： 用 json.dumps() 把 chunk 编码成单行字符串，\n 会被转义成字面量 \\n，天然不会出现裸空行。前端 JSON.parse(e.data) 还原。

至于结尾为什么是 data: [DONE] 而不是直接关连接 —— 因为前端分不清「正常结束」和「网线被拔了」。两种情况在 EventSource 那里都表现为 onerror + 自动重连。发一个显式终止标记，前端就能先 close() 再收尾。OpenAI 兼容格式定这个约定就是这个原因，不是为了好看。

#3.4 自动重连：省事，也最容易变成事故

EventSource 断线后会自己重连，这是 SSE 相对 WebSocket 的最大便利 —— 但在 LLM 场景，它默认的行为是危险的：

⚠️ 自动重连导致 token 双倍计费 错误操作： 前端 new EventSource('/chat')，服务端不管 Last-Event-ID，收到请求就从头跑一遍模型。

实际结果： 用户在地铁里断网 3 秒，浏览器悄悄重连，服务端又完整生成了一遍 500 token 的回答。用户毫无感知，账单翻倍；如果重连发生多次，费用线性增长。

原因： SSE 的重连语义是「重发这个 HTTP 请求」，而不是「恢复上次的流」。是否续传完全靠服务端配合 Last-Event-ID 请求头，浏览器只负责把它带上。

正确做法（三选一，按成本递增）：

1. 禁用自动重连 —— 用 fetch + ReadableStream 代替 EventSource，重连逻辑自己写（这也是 5.2 鉴权问题的解法，一举两得）；
2. 让重连幂等 —— 服务端按 会话 ID + 消息 ID 把已生成内容写进 Redis（参考 redis 的 Cache-Aside），重连时直接回放缓存而不再调模型；
3. 实现真正的续传 —— 发帧时带 id:，重连时读 request.headers.get("last-event-id")，从该序号之后继续。

#四、WebSocket：握手、帧与 FastAPI 实现

#4.1 从 HTTP「升级」上来

WebSocket 的开头仍然是一个普通 HTTP 请求，只是带了几个特殊头：

GET /ws/chat HTTP/1.1
Host: api.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13

服务端同意后返回 101 Switching Protocols。从这一刻起，这条 TCP 连接上跑的不再是 HTTP，而是 WebSocket 帧（带 opcode，区分文本帧 / 二进制帧 / ping / pong / close）。

这个「换协议」的动作，正是它一切优点和一切麻烦的根源：

• 优点：不再受 HTTP 请求-响应模型约束，双向、二进制、低开销（每帧头部只有 2~14 字节，对比 HTTP 每次几百字节的头）；
• 麻烦：中间设备（老旧代理、部分企业防火墙、某些 CDN 配置）只认 HTTP，看到 Upgrade 要么不转发要么直接断，而且这类问题只在客户网络出现，本地永远复现不了。

#4.2 FastAPI 最小实现：注意这里有个结构性陷阱

import asyncio
from fastapi import WebSocket, WebSocketDisconnect
 
@app.websocket("/ws/chat")
async def ws_chat(ws: WebSocket):
    await ws.accept()
    task: asyncio.Task | None = None
 
    async def generate(prompt: str):
        try:
            async for chunk in llm.astream(prompt):
                await ws.send_json({"type": "token", "t": chunk})
            await ws.send_json({"type": "done"})
        except asyncio.CancelledError:
            await ws.send_json({"type": "cancelled"})   # 打断后给个交代
            raise
 
    try:
        while True:
            msg = await ws.receive_json()
            if msg.get("cmd") == "stop":
                if task and not task.done():
                    task.cancel()                        # ← SSE 做不到的事
                continue
            # 关键：生成放进后台 task，接收循环立刻回到 await receive_json
            task = asyncio.create_task(generate(msg["ask"]))
    except WebSocketDisconnect:
        if task and not task.done():
            task.cancel()                                # 客户端跑了，别再烧 token

⚠️ 在接收循环里直接 await 生成过程，「停止」按钮就永远失灵 错误操作：

while True:
    msg = await ws.receive_json()
    async for chunk in llm.astream(msg["ask"]):   # ❌ 直接在循环里生成
        await ws.send_json({"t": chunk})

实际结果： token 能正常推，但用户点「停止」后毫无反应，非要等整段生成完才响应 —— 表现得和 SSE 一模一样，白上了 WebSocket。

原因： 这个协程被 async for 占住了，根本没有回到 await ws.receive_json()。{"cmd":"stop"} 只是躺在 TCP 缓冲区里没人读。连接是双工的，但你的代码是单线的。

正确做法： 用 asyncio.create_task() 把生成挪到后台，让接收循环始终处于「空闲待命」状态 —— 上面代码里的写法。这一点和 Python 异步编程 里「协程并发靠的是让出控制权」是同一个道理。

#4.3 心跳不是可选项

# 服务端定时 ping，也可由客户端发
async def heartbeat(ws: WebSocket, interval: int = 25):
    while True:
        await asyncio.sleep(interval)
        await ws.send_json({"type": "ping"})

原因：Nginx / ALB / 各类企业网关对空闲连接的默认超时普遍在 60 秒。而 AI 场景恰恰有大段空闲 —— 用户读完回答思考两分钟很正常。没有心跳，连接会被中间设备静默掐断，前端下次发消息才发现连接已死。

SSE 对应的心跳写法是注释行：yield ": ping\n\n"，同样需要。

#五、AI 场景下真正拉开差距的六个维度

前面都是通用知识。这一节才是这篇笔记的重点：同样做 AI 应用，这两个协议在哪些地方会真的把你逼到墙角。

#5.1 打断生成：区分「一次性取消」和「持续交互」

这是最常被拿来论证「AI 必须用 WebSocket」的理由，但它论证不成立。

用户点「停止生成」，本质是发一次性的控制信号。SSE 完全可以这样做：

# 前端：EventSource 之外，再发一个普通请求
# fetch('/chat/cancel', {method: 'POST', body: JSON.stringify({sid})})
 
CANCELLED: set[str] = set()          # 生产环境用 Redis，多实例才共享
 
@app.post("/chat/cancel")
async def cancel(sid: str):
    CANCELLED.add(sid)
    return {"ok": True}
 
async def event_stream(request: Request, prompt: str, sid: str):
    async for chunk in llm.astream(prompt):
        if sid in CANCELLED or await request.is_disconnected():
            break
        yield f"data: {json.dumps({'t': chunk}, ensure_ascii=False)}\n\n"

事实上还有更省事的做法：前端直接 es.close()。连接一断，服务端下一次 await request.is_disconnected() 就返回 True，生成自然终止。绝大多数场景这就够了。

💡 判断标准：控制信号是「一次性」还是「会话级」

• 一次性（停止、点赞、重新生成）→ 独立 HTTP 请求即可，别为它上 WebSocket；
• 会话级持续交互（Agent 执行到一半要人工审批，用户改主意后补充条件，实时语音打断）→ 每次都新建连接的开销和状态同步复杂度会失控，这时 WebSocket 才真正值回票价。

#5.2 鉴权：两个对称的坑

这是实践中最先撞上的问题，而且两边都有。

🔴 浏览器的 EventSource 和 WebSocket 都不能带自定义请求头 错误操作：

new EventSource('/chat', { headers: { Authorization: `Bearer ${token}` } })  // ❌ 参数被忽略
new WebSocket('wss://api/ws', { headers: {...} })                            // ❌ 浏览器端无此重载

实际结果： 没有任何报错，请求照发，只是头没带上，服务端一律 401。用 Postman 或 Python 客户端测试全部正常（因为它们能自定义头），只有浏览器不行 —— 极易误判成后端问题。

原因： 这两个都是浏览器的高层 API，规范里就没开放自定义头的口子（EventSource 只有一个 withCredentials 选项，WebSocket 只有一个 protocols 参数）。和协议本身无关，服务端用 httpx / websockets 库调时随便带头。

正确做法：

方案	SSE	WebSocket	评价
Cookie	withCredentials: true + CORS 允许凭据	同源自动带上	同源架构下最省事，跨域要配 SameSite
Query 参数	?token=xxx	wss://...?token=xxx	⚠️ token 会进入 Nginx access log 和浏览器历史，只能用短时效一次性 token
子协议头	不适用	new WebSocket(url, ['bearer', token])	借 Sec-WebSocket-Protocol 夹带，能带头但语义是滥用
换 API	fetch + ReadableStream	无对应方案	✅ SSE 的推荐解

SSE 这一栏的最后一行是关键：放弃 EventSource，改用 fetch 手动读流。因为 SSE 就是普通 HTTP 响应，fetch 完全能收，还顺带解决了「只能 GET」「自动重连不可控」两个问题。Vercel AI SDK、OpenAI 官方 JS SDK 走的都是这条路。

// 用 fetch 收 SSE：能带 header、能 POST、重连自己说了算
const res = await fetch('/chat', {
  method: 'POST',
  headers: { 'Authorization': `Bearer ${token}`, 'Content-Type': 'application/json' },
  body: JSON.stringify({ prompt }),
  signal: controller.signal,          // 顺便拿到了原生的取消能力
});
const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
let buf = '';
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  buf += value;
  const frames = buf.split('\n\n');   // 自己按空行分帧
  buf = frames.pop();                 // 最后一段可能不完整，留着
  for (const f of frames) {
    const line = f.split('\n').find(l => l.startsWith('data: '));
    if (!line) continue;
    const data = line.slice(6);
    if (data === '[DONE]') return;
    handleToken(JSON.parse(data));
  }
}

用了 fetch，还算 SSE 吗 算。SSE 定义的是报文格式和 text/event-stream 这个 MIME 类型，不是 EventSource 这个 API。服务端代码一个字都不用改，换的只是客户端读法。代价是自动重连、Last-Event-ID 回传这些浏览器代劳的事，现在得自己写 —— 但在 LLM 场景，这些恰恰是你本来就想自己控制的（见 3.4）。

#5.3 二进制与语音：SSE 的硬边界

这是真正无法绕过的分界。SSE 的响应体是 text/event-stream，规范层面只能传 UTF-8 文本。要传音频：

所以但凡涉及实时语音对话 Agent（麦克风 PCM 上行 + TTS 音频下行），WebSocket 几乎是唯一务实选择。OpenAI Realtime API 用 WebSocket / WebRTC 而不是 SSE，就是这个原因。

反过来，纯文本的 Chat / RAG 应用，这一条完全不构成理由。

#5.4 断线恢复：谁的语义更适合 LLM

结论有点反直觉：SSE 的自动重连在普通 Web 应用里是优点，在 LLM 应用里是风险（3.4 那个坑）。而 WebSocket 什么都不帮你做，反而没有意外扣费 —— 但你要自己写一堆重连逻辑。

#5.5 横向扩容：无状态 vs 有状态

这条在从「demo 跑通」走向「上线扛量」时会突然变得很重要。

• SSE 是无状态的：每条流就是一个 HTTP 请求，任何一个实例都能处理。加机器、滚动发布、蓝绿部署，跟普通接口没区别。发布时正在跑的流会断，浏览器自动重连到新实例 —— 只要你按 3.4 做了幂等。
• WebSocket 是有状态的：连接绑死在某个实例上。于是引出一串附加工程：
  • 负载均衡要配粘性会话或按连接哈希；
  • 多实例间要广播消息，通常得引入 Redis Pub/Sub 之类的中间层；
  • 滚动发布时所有连接一起断，得设计优雅下线 + 客户端重连风暴的削峰；
  • 每个实例的连接数成为新的容量指标（每条连接常驻内存），而不只是 QPS。

💡 这条经常被低估 一个只有几十人用的内部工具，WebSocket 的运维成本约等于零。但当你需要多实例部署时，「上 WebSocket」实际上意味着「顺带要引入一套连接状态管理」。选型时要把这部分算进成本。

#5.6 可观测性与调试

排查线上「某个用户的回答生成到一半断了」这类问题时，这个差距非常真实。

#六、常见陷阱

前面各节内嵌的坑不重复。这里是剩下几个高频的。

🔴 Nginx 缓冲：本地完美，上线变成「转圈然后一次性刷出」 错误操作： 服务端 SSE 写得完全正确，直接上 Nginx 反代。

实际结果： 本地 uvicorn 直连时逐字输出正常；过了 Nginx 之后前端 20 秒白屏，然后整段文字「啪」地一次性出现。

原因： Nginx 的 proxy_buffering 默认是 on，它会先把上游响应攒进缓冲区，攒满或结束了才转发给客户端 —— 对普通页面这是优化，对流式响应是灾难。

正确做法（二选一）：

location /chat {
    proxy_pass http://backend;
    proxy_buffering off;          # 关缓冲
    proxy_cache off;
    proxy_read_timeout 3600s;     # 顺手把超时放宽，否则长回答会被掐
    proxy_set_header Connection '';
    proxy_http_version 1.1;
}

或者由应用侧下发 X-Accel-Buffering: no 响应头（3.2 代码里那行），Nginx 会针对该响应关闭缓冲 —— 这个方式更推荐，因为改一处代码就行，不用协调运维改配置。

⚠️ 同类问题还会出现在 CDN、云厂商 API 网关、Serverless 平台上，每家的开关名字都不一样，上线前必须在真实链路上验证一次，别只信本地。

⚠️ GZip 中间件让流「一顿一顿」 错误操作： 全局挂 app.add_middleware(GZipMiddleware, minimum_size=1000)。

实际结果： token 不是平滑出现，而是攒一批蹦一下。

原因： 压缩器有自己的内部缓冲区，几个字节的 token 块进去后不会立刻产出压缩输出，要攒到一定量才刷出，等于在你的流上又加了一层缓冲。

正确做法： 让 SSE 路径跳过压缩。最简单是判断响应类型，或把流式路由挂在不经过该中间件的子应用上：

class SkipSSEGZip(GZipMiddleware):
    async def __call__(self, scope, receive, send):
        if scope["type"] == "http" and scope["path"].startswith("/chat"):
            return await self.app(scope, receive, send)
        return await super().__call__(scope, receive, send)

SSE 是纯文本，确实很适合压缩，但实时性优先级更高。真想压，用 WebSocket 的 permessage-deflate 扩展，它按消息压缩不会跨消息攒。

⚠️ 同域 SSE 连接数撞上浏览器 6 连接上限 错误操作： HTTP/1.1 环境下，页面里给每个会话卡片各开一条常驻 SSE。

实际结果： 开到第 7 个标签页（或第 7 条流）时，新连接一直 pending，页面里其他普通 AJAX 请求也跟着卡住。

原因： HTTP/1.1 下浏览器对每个域名的并发连接上限是 6。SSE 是长连接，占着不放，会把额度吃光，连累同域的所有请求。

正确做法：

1. 上 HTTP/2（最优解）—— 多路复用后同域只要一条 TCP 连接，上限变成 100+ 并发流，问题自动消失；
2. 全局只维持一条 SSE 通道，用 event: 字段区分不同会话的消息；
3. 实在不行，把流式接口放到独立子域分摊额度。

💡 WebSocket 不受这个限制（浏览器对 WS 的上限通常是每域 255），这算是它一个少被提及的实际优势。

⚠️ 用户关了页面，服务端还在烧 token 错误操作： 生成器里只管 async for ... yield，不检查连接状态。

实际结果： 账单比预期高一截，且和用户实际读到的内容对不上。压测时尤其明显。

原因： 客户端断开后，yield 的数据写进一个已经没人收的连接。ASGI 层会发 http.disconnect 事件，但你不主动查就不会知道，async for 会一路跑到模型生成结束。

正确做法： SSE 用 await request.is_disconnected()（3.2 代码 ①），WebSocket 捕获 WebSocketDisconnect 后 task.cancel()（4.2 代码）。

⚠️ 注意 is_disconnected() 只在你调用它的那一刻检查，所以要放在循环里每轮都查，不能只在开头查一次。

⚠️ 在 async def 生成器里调同步阻塞的 SDK 错误操作：

async def event_stream(prompt):
    for chunk in openai_sync_client.chat.completions.create(..., stream=True):  # ❌ 同步
        yield f"data: ...\n\n"

实际结果： 单人测试完全正常。一旦有第二个用户进来，所有人的响应一起卡死，包括健康检查接口。

原因： 同步客户端的网络等待会阻塞整个事件循环，这是 Python 异步编程 里最经典的错误 —— 在 async 函数里做同步 I/O，等于让整个进程停摆。流式场景下更致命，因为阻塞时间是整段生成的时长（几十秒）而不是几百毫秒。

正确做法： 用异步客户端（AsyncOpenAI、llm.astream()）；实在只有同步库，用 asyncio.to_thread() 包一层丢进线程池。

#七、怎么选

三个问题按顺序问，任何一个答「是」就右转 WebSocket，全「否」就落到 SSE。读图时注意 Q1 底下那行小字 —— 「点停止」不算持续上行，这是最常见的过度选型理由。

落到具体场景：

💡 一个务实的默认策略 先上 SSE。 它能覆盖绝大多数 AI 应用，而且几乎不欠技术债 —— 因为它就是普通 HTTP，鉴权、限流、日志、灰度全部复用现成设施。

等到你的产品真的要做语音、或者真的要做会话中人机协同时，再引入 WebSocket，而且只给那一条链路用，不要为了「统一技术栈」把整个 Chat 接口迁过去。两者并存是常态，不是妥协。

#八、动手练习

前面 3.4 给了三种「防止重连重复计费」的思路，笔记里只写了方向，没有给完整实现 —— 这个决策点有多种合理答案，值得你自己权衡一次。

代码骨架在下面，注释标出的位置就是需要你补的地方：

# resume_stream.py —— SSE 断线续传骨架
 
import json
from fastapi import Request
from redis.asyncio import Redis
 
redis: Redis = ...   # 参考 [redis](/posts/redis) 的连接池封装
 
async def resumable_stream(request: Request, sid: str, prompt: str):
    """支持断线续传的 SSE 流。
 
    约定：每帧带 id:（从 1 递增）；已生成内容按 sid 存进 Redis List。
    浏览器重连时会带上 Last-Event-ID 请求头。
    """
    last_id = int(request.headers.get("last-event-id") or 0)
 
    # TODO(human): 在这里实现续传决策
    #
    # 你需要决定：
    #   1. last_id > 0 时（说明是重连），先做什么？是回放缓存、
    #      还是直接拒绝并让前端提示用户？
    #   2. 回放的话，Redis 里存的历史帧怎么读、怎么发出去，
    #      发完之后要不要接着调模型继续生成剩下的部分？
    #   3. 如果重连时上一轮其实已经生成完了（缓存里有 [DONE] 标记），
    #      要怎么快速收尾，避免又调一次模型？
    #
    # 提示：redis.lrange(f"stream:{sid}", start, -1) 可以取某个位点之后的帧。
 
    # 首次请求（或续传决策之后）走这里，正常生成
    seq = last_id
    async for chunk in llm.astream(prompt):
        if await request.is_disconnected():
            break
        seq += 1
        payload = json.dumps({"t": chunk}, ensure_ascii=False)
        await redis.rpush(f"stream:{sid}", payload)
        await redis.expire(f"stream:{sid}", 600)
        yield f"id: {seq}\ndata: {payload}\n\n"
    yield f"id: {seq + 1}\ndata: [DONE]\n\n"

#速查表

必背的三行代码：

yield f"data: {json.dumps(x, ensure_ascii=False)}\n\n"   # 两个 \n，且必须 JSON 编码
headers={"X-Accel-Buffering": "no"}                       # 否则线上一次性刷出
if await request.is_disconnected(): break                 # 否则用户跑了还在烧钱

#复习重点

复习重点

1. 一句话主线： SSE 是「不结束的 HTTP 响应」，WebSocket 是「换掉 HTTP 的双工通道」；AI 应用默认选 SSE，只有会话中持续上行 / 二进制 / 服务端主动推无关消息这三件事才值得升级到 WebSocket。
2. 最隐蔽的 bug： token 直接拼进 data: 会被内容里的 \n\n 劈坏帧，且静默丢数据 —— 必须 json.dumps 成单行。
3. 最先撞上的墙： 浏览器的 EventSource 和 WebSocket 都不能带自定义请求头；SSE 的解法是改用 fetch 读流，这同时还解决了「只能 GET」和「自动重连不可控」。
4. 最贵的坑： SSE 自动重连会让服务端从头重跑模型，token 双倍计费；要么禁用自动重连，要么用 Last-Event-ID + Redis 做幂等。
5. 最容易上线才发现的坑： Nginx proxy_buffering 默认开着，本地正常线上一次性刷出 —— 下发 X-Accel-Buffering: no。
6. 「打断生成」不是选 WebSocket 的理由： 前端 es.close() 或另发一个 POST /cancel 就能解决。真正的理由是会话级的持续双向交互。
7. WebSocket 用了就要写的三件事： 心跳（避开网关 60 秒空闲超时）、把生成扔进 asyncio.create_task（否则停止按钮失灵）、多实例的粘性会话与 Pub/Sub。
8. 选型口诀： 默认 SSE，被语音和 HITL 逼着才上 WebSocket，且只给那一条链路用。

#延伸

• HTTP/2 与 HTTP/3 下的 SSE：多路复用消除了 6 连接限制，SSE 的主要短板之一自动消失，值得优先升级。
• WebTransport：基于 HTTP/3 / QUIC，兼具双向、二进制和多流，且没有队头阻塞，被认为是 WebSocket 的下一代替代品，浏览器支持仍在铺开。
• WebRTC DataChannel / 音频轨：实时语音场景比 WebSocket 更进一步（UDP、可容忍丢包、内建回声消除），OpenAI Realtime API 在浏览器端推荐用它。
• gRPC streaming：服务端到服务端的流式首选，浏览器需 gRPC-Web 且不支持双向流。