流式响应 SSE 与 WebSocket

本章你将学会

• 为什么需要流式：大文件下载、慢生成任务、AI 逐字输出——一次性响应撑不住的三种场面
• StreamingResponse 全套用法：同步/异步生成器、media_type、Content-Disposition 文件下载
• SSE 协议的规矩：text/event-stream、data:/event:/id:/retry: 四类字段、消息边界那个空行
• 断线重连：浏览器自动重连的机制、Last-Event-ID 断点续传怎么接
• sse-starlette 的 EventSourceResponse：比手写 SSE 多白送的三样货
• AI 场景实战：给墨集写一个「AI 写作助手逐字生成」端点——本课程 AI 定位的落地样板
• WebSocket：@app.websocket、生命周期 accept/receive/send/close、ConnectionManager 广播、前端 WebSocket() 对照
• SSE vs WebSocket 选型对比表：单向还是双向、要不要自动重连、代理兼容性

上一章（第 13 章）用 loguru 把墨集的日志盘顺了，观测性的活儿到此收齐。这一章——全课程的最后一章——来还两笔欠账：第 01 章说过「高并发、WebSocket、长连接这些新需求，都得靠 ASGI 才好使」，当时一笔带过；第 10 章的信封中间件遇到 SSE 流要专门放行，当时说「第 14 章见」。本章把流式响应、SSE、WebSocket 一次讲透，主题只有一个：不再一次性给完。

1. 为什么需要流式：一次性响应的三种翻车现场

作用：给「为什么需要流式」建立直觉。先看墨集 inkhub 的三个真实场面：

1. 大文件下载。运营要把全站文章打包导出，500 MB。一次性响应意味着服务端得把 500 MB 在内存里攒成完整 body 才能开始发——内存直接被吃爆；就算攒得齐，客户端也要等到最后一个字节生成完，才收到第一个字节。
2. 慢生成。年度报告的统计接口要跑 30 秒。一次性响应下，客户端就白屏干等 30 秒，网关的超时阈值（常见 30~60 秒）一过，连接直接被掐。
3. AI 逐字输出。墨集的 AI 写作助手生成一篇 500 字的短文要 15 秒。用过 ChatGPT 类产品都知道那种「打字机效果」——字一个一个往外冒。一次性响应 15 秒憋出一整篇，体验上是两个时代的产品。

关键认知：HTTP 响应从来不要求「全部准备好才发第一个字节」。第 02 章那种普通端点之所以一次性，是因为框架替你把返回值攒成一个完整 body 再发。HTTP/1.1 自带分块传输（Transfer-Encoding: chunked）：服务器可以把响应体切成任意多块，边生成边发，最后发一个终止块收尾。

打个比方：一次性响应是食堂大锅菜——一桌菜全做好才端上来，菜越多等越久，后厨还得有一张摆得下整桌的案板（内存）；流式响应是回转寿司——师傅捏好一碟推出来一碟，客人三秒就能入口第一碟，师傅手边永远只需要一个碟子的位置。

第 01 章那句「ASGI 才好使」在这兑现：流式的本质是生成器 yield 一段、服务器 send 一段，发送间隙事件循环还能去伺候别的请求。同步的 WSGI 世界里，框架只能把生成器从头跑到尾攒齐了再给。

常见坑（认知层面先排三个雷）：

• 「要流式就得先上 WebSocket」——不用。普通 HTTP 就能流式，SSE 本质就是一条流式 HTTP 响应。
• 流式响应头里没有 content-length，curl -i 看到的是 transfer-encoding: chunked——这是分块传输的标志，不是 bug。
• 流式省的是等待感和服务端内存，不省总耗时：生成要 15 秒的活，客户端总归 15 秒才收完，区别只是「憋 15 秒一次性吐」变成「边产边收」。

2. StreamingResponse：边做边给的基本功

作用：把一个生成器交给框架，框架边从生成器取、边往网络写。

签名：

from fastapi.responses import StreamingResponse

StreamingResponse(
    content,                       # 同步或异步迭代器，逐块产出 str / bytes
    status_code: int = 200,
    headers: dict[str, str] | None = None,
    media_type: str | None = None, # 不写就没有 content-type，见 2.3
    background: BackgroundTask | None = None,
)

2.1 异步生成器（首选）

最小示例——墨集的定时任务日志边产生边推：

import asyncio
from collections.abc import AsyncIterator

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI(title="墨集 inkhub")


@app.get("/api/logs/stream")
async def stream_logs() -> StreamingResponse:
    async def gen() -> AsyncIterator[str]:
        for line in ("INFO  定时任务启动：扫描全站死链", "WARN  发现 1 条失效外链", "INFO  任务结束"):
            await asyncio.sleep(0.4)  # 模拟日志慢慢产生
            yield f"{line}\n"

    return StreamingResponse(gen(), media_type="text/plain")

实测（curl -i http://127.0.0.1:8000/api/logs/stream，约每 0.4 秒蹦一行）：

HTTP/1.1 200 OK
content-type: text/plain; charset=utf-8
transfer-encoding: chunked        ← 没有 content-length，分块传输

INFO  定时任务启动：扫描全站死链
INFO  已扫描 1280 篇文章
WARN  《夜航船里的旧灯》存在 1 条失效外链
INFO  死链报告已生成，等待客户端拉取
INFO  本次任务结束

2.2 同步生成器：也能跑，但要知道它在哪跑

普通函数生成器一样能交给 StreamingResponse——Starlette 会把它丢进线程池里迭代，不堵事件循环：

from collections.abc import Iterator


def chunks() -> Iterator[bytes]:
    for line in report_text.splitlines(keepends=True):
        yield line.encode("utf-8")

return StreamingResponse(chunks(), media_type="text/markdown")

两个必须知道的点：

• 同步生成器里的 time.sleep 占的是线程池线程，池子默认只有 40 个——大量慢同步生成器同时在线，会把线程池耗干，别的请求跟着遭殃。能 await 的场景一律用异步生成器 + asyncio.sleep。
• 同步生成器里写不了 await——要消费异步资源（比如第 04 章的异步数据库游标），就必须换异步生成器。

2.3 media_type：不写就没有 content-type

第 02 章讲过 MIME 类型。这里的坑很朴素：media_type 不写，响应连 content-type 头都没有，浏览器只能瞎猜内容类型。写了 text/* 开头的类型，框架自动补 ; charset=utf-8（上面实测里的 text/plain; charset=utf-8 就是这么来的）。

2.4 文件下载：Content-Disposition

作用：告诉浏览器「这是要下载保存的文件，别当网页渲染」。

@app.get("/api/reports/download")
async def download_report() -> StreamingResponse:
    def chunks() -> Iterator[bytes]:
        for line in build_report().splitlines(keepends=True):
            yield line.encode("utf-8")

    return StreamingResponse(
        chunks(),
        media_type="text/markdown",
        headers={"Content-Disposition": 'attachment; filename="inkhub-report.md"'},
    )

实测响应头：

content-disposition: attachment; filename="inkhub-report.md"
content-type: text/markdown; charset=utf-8
transfer-encoding: chunked

• attachment = 提示浏览器下载保存；inline = 能渲染就渲染（PDF 常用）。
• 中文文件名要走 RFC 5987 扩展格式：filename*=UTF-8''%E5%A2%A8%E9%9B%86%E6%97%A5%E6%8A%A5.md，和 filename= 两段一起写最稳（老浏览器认前者，新浏览器认后者）。
• 磁盘上现成的文件优先用 FileResponse（第 10 章文章导出用过）——传个 filename 参数它自动帮你写好 Content-Disposition，发送效率也比手搓生成器高。StreamingResponse 留给「内存里现生成的数据」。

2.5 常见坑

• 流已开吐，状态码就改不了了。响应头（含状态码 200）一旦发出去，中途再抛异常也没法把状态码改成 500——第 09 章的异常处理器只能在「响应还没开始」时接管。所以流式端点的生成器内部要自己兜错误，通常以一条错误事件（SSE 场景）或提前终止流（下载场景）收场。
• 不能返回 dict。普通端点返回 dict 框架自动转 JSON；StreamingResponse 只认产 str/bytes 的迭代器，JSON 要自己 json.dumps。
• curl 看流式要加 -N（--no-buffer），否则 curl 自己攒够缓冲才往屏幕打，看起来像卡死。

3. SSE：HTTP 长连接上的单向广播

3.1 SSE 是什么

Server-Sent Events（服务器推送事件）：服务器向客户端单向持续推消息的协议。本质是一条保持不关的 HTTP 流式响应，Content-Type 固定为 text/event-stream。

打个比方：SSE 是收音机——电台（服务器）持续广播，你（客户端）只能听、不能插嘴；想点歌，得另外打个电话（发一条普通 HTTP 请求）。收音机断了电（断线），它自己会重新搜台（浏览器自动重连），还记得上次听到第几首（Last-Event-ID）。

第 10 章 demo 里 /api/posts/{id}/stream 吐的那些 data: 开头的行——那就是手写 SSE，本章把协议规矩补全。当时信封中间件按 content-type 判断放行它，现在你知道为什么必须放行：把流读光再重打包，实时性就没了，流式响应唯一的价值就是「边产边走」。

3.2 报文格式：四类字段 + 一个空行

一条最完整的 SSE 消息长这样：

id: 42
event: token
retry: 3000
data: 墨

字段手册：

三条铁律：

1. 每条消息以空行结尾——报文里就是两个换行 \n\n。空行是「一条消息到此为止」的边界，漏写的话客户端会把几条攒成一条、甚至永远收不到，这是 SSE 的头号大坑。
2. data 可以多行，每行都要带 data: 前缀，浏览器收到后用 \n 拼接。
3. 一行一个字段，格式 字段名: 值，冒号后的空格可有可无。

3.3 浏览器端：EventSource

const es = new EventSource("/api/ai/write");
es.onmessage = (e) => console.log("默认 message 事件:", e.data);
es.addEventListener("token", (e) => console.log("逐字:", e.data));  // 具名事件
es.addEventListener("done", (e) => es.close());                    // 服务器说完了，主动关

三个硬限制：

• 只能 GET——事件流全靠 URL（加 query 参数）表达，没法用 POST 带请求体。
• 不能自定义请求头——第 11 章放进 Authorization 头的 JWT 在这带不进去，token 只能走 cookie 或 URL query。
• 自动重连是浏览器送的；但 fetch/axios 没这待遇——想用 POST 享受流式，得自己用 fetch + ReadableStream 逐块读（前端进阶话题，知道方向即可）。

3.4 断线重连：Last-Event-ID 断点续传

SSE 断线后浏览器自动重连（默认约 3 秒，服务端可用 retry: 调整），重连时自动带上 Last-Event-ID 请求头，值就是最后收到的那条事件的 id。服务端读它，从下一个编号接着发即可：

客户端                                服务端
  │  GET /api/ai/write                 │
  │───────────────────────────────────▶│
  │◀── id: 0 ... id: 9（吐到第 10 个字）│   ← 网络断了
  │        ✗ 连接断开                   │
  │  （约 3 秒后，浏览器自动重连）        │
  │  GET /api/ai/write                 │
  │  Last-Event-ID: 9                  │
  │───────────────────────────────────▶│
  │◀── id: 10, 11, ...（从断点续着吐）──│

服务端就三行：

last_id = request.headers.get("last-event-id", "")
start = int(last_id) + 1 if last_id.isdigit() else 0
# 生成器从 start 开始发编号

注意：服务端不实现这个也没人报错——只是客户端重连后会把整个流从头再收一遍，AI 场景里就是打字机倒带重放。

3.5 EventSourceResponse：sse-starlette 替你把脏活干了

裸 StreamingResponse 手写 SSE，得自己拼 id:/event:/data: 行和空行。社区库 sse-starlette 的 EventSourceResponse 收一个「产 dict 的生成器」，报文自动拼。

安装：uv add sse-starlette

签名（节选）：

from sse_starlette.sse import EventSourceResponse

EventSourceResponse(
    content,                                  # 产 dict / str 的（异步）生成器
    status_code: int = 200,
    headers: dict[str, str] | None = None,
    media_type: str = "text/event-stream",    # 默认已是对的
    ping: float | None = 15,                  # 心跳间隔秒数；None 关闭
)

最小示例：

from collections.abc import AsyncIterator

from sse_starlette.sse import EventSourceResponse


@app.get("/api/feed")
async def feed() -> EventSourceResponse:
    async def gen() -> AsyncIterator[dict[str, str]]:
        yield {"event": "notice", "data": "墨集专栏更新了"}
        yield {"id": "1", "event": "notice", "data": "有人评论了你的文章"}

    return EventSourceResponse(gen())

dict 的 key 就是 3.2 表里的字段名（data/event/id/retry），用哪个写哪个。

白送的三样（实测响应头）：

cache-control: no-store                 ← 不许中间设备缓存
x-accel-buffering: no                   ← 告诉 nginx 别缓冲（下一节细说）
content-type: text/event-stream; charset=utf-8

外加默认每 15 秒一条注释行心跳（ping 参数可调可关）——长时间不发包的空闲连接，容易被中间设备当死连接掐掉，心跳就是「我还活着」的信号。

文档去哪看：https://github.com/sysid/sse-starlette ；EventSource 的 MDN 页面：https://developer.mozilla.org/zh-CN/docs/Web/API/EventSource

3.6 常见坑

• \n\n 漏写（手写版）——重申一遍，头号坑。
• 本地好使、上线卡死：开发环境 uvicorn 逐块直发不缓冲；上了 nginx，默认 proxy_buffering on 会把流攒够一批才转发，打字机效果变成「憋十秒、蹦一百个字」。解法：响应头 X-Accel-Buffering: no（EventSourceResponse 自动带）或 nginx 站点配置 proxy_buffering off。
• EventSource 断线自动重连是无条件的——服务端不支持 Last-Event-ID 时会整流重放；另外连不上时它会永远重连下去，前端记得在合适的时机 es.close()。
• EventSource 也受同源策略管，跨源要后端配 CORS（第 10 章的 CORSMiddleware），本章 demo 已放行。

4. AI 场景实战：墨集 AI 写作助手逐字生成

现在所有大模型 Web 界面的「打字机效果」，背后几乎都是 SSE 或 chunked 流；主流 LLM API 的 stream=True 返回的也正是 text/event-stream。本节的模拟端点吃透之后，接真模型只是把 asyncio.sleep 换成「从模型客户端的流里逐块取」。

4.1 端点设计

• GET /api/ai/write?topic=深夜书房：逐字吐出一篇短文；
• 先沉默约 1 秒——照实模拟大模型的「首字延迟」；
• 每个字一条 event: token 事件，id 递增；
• 结束发一条 event: done，data 直接带全文（前端不用自己攒字）；
• 读 Last-Event-ID，断线重连从断点续吐。

4.2 核心代码（节选自 demo/demo_stream.py）

@app.get("/api/ai/write")
async def ai_write(request: Request, topic: str = "深夜书房") -> EventSourceResponse:
    draft = AI_DRAFTS.get(topic, AI_DRAFTS["深夜书房"])

    # 断线重连：浏览器自动带上 Last-Event-ID，从下一个编号接着发
    last_id = request.headers.get("last-event-id", "")
    start = int(last_id) + 1 if last_id.isdigit() else 0

    async def gen() -> AsyncIterator[dict[str, str]]:
        await asyncio.sleep(1.0)              # 模拟首字延迟
        for seq in range(start, len(draft)):
            await asyncio.sleep(0.12)         # 模拟逐字节奏
            yield {"id": str(seq), "event": "token", "data": draft[seq]}
        yield {"id": str(len(draft)), "event": "done", "data": draft}

    return EventSourceResponse(gen())

4.3 实测

curl -N http://127.0.0.1:8000/api/ai/write

（沉默约 1 秒——首字延迟）
id: 0
event: token
data: 墨

id: 1
event: token
data: 集

……（逐字省略）……

id: 57
event: done
data: 墨集编辑部的深夜书房永远亮着一盏灯，键盘声此起彼伏，像一场不会散场的雨。有人写完最后一段，抬头发现窗外天已经亮了。

断点续传实测——手动模拟浏览器的重连请求：

curl -N -H "Last-Event-ID: 9" http://127.0.0.1:8000/api/ai/write

id: 10
event: token
data: 永

前 10 个字（id 0~9 的「墨集编辑部的深夜书房」）没有重发，从第 11 个字「永」接着走。

4.4 前端打字机

const box = document.querySelector("#draft");
const es = new EventSource("/api/ai/write?topic=雨巷");
es.addEventListener("token", (e) => { box.textContent += e.data; }); // 一字一格往外冒
es.addEventListener("done", (e) => { console.log("全文：", e.data); es.close(); });
es.addEventListener("error", () => es.close());  // 服务端真挂了就别无限重连

4.5 接真模型只差一步

把生成器里的「睡 0.12 秒吐一个字」换成「从模型流里取一块」，其余一行不动：

# 示意：以 OpenAI 兼容接口为例
async for chunk in await client.chat.completions.create(
    model="deepseek-chat", messages=[...], stream=True
):
    delta = chunk.choices[0].delta.content
    if delta:
        yield {"event": "token", "data": delta}

5. WebSocket：全双工对讲机

5.1 和 SSE 的本质区别

HTTP 的请求-响应模型是半双工：一问一答，问完才能答。SSE 已经把「答」拉长成一条持续不断的流，但方向仍然只有服务器 → 客户端。

WebSocket（RFC 6455）更进一步：先用一次普通 HTTP 请求带上 Upgrade: websocket 头握手，服务器同意后，这条 TCP 连接就升级成全双工通道——两边随时互发，谁也不用等谁。

比喻升级：SSE 是收音机，WebSocket 是对讲机——你一句我一句，还能同时喊。

两句欠账在这兑现：

• 第 01 章说 WebSocket 靠 ASGI——ASGI 规范里 scope["type"] 除了 "http" 还有 "websocket"，WebSocket 是 ASGI 的原生公民。
• 第 10 章说 @app.middleware("http") 管不到 WebSocket——那套中间件挂在 http 类型上，websocket 类型的流量从旁边直接穿过去。计时、信封、CORS 三件套对 WS 连接全部失效，需要在 WS 端点里自己干的活，后面坑列表里列。

5.2 签名与最小示例

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI(title="墨集 inkhub")


@app.websocket("/ws/echo")
async def echo(websocket: WebSocket) -> None:
    await websocket.accept()
    try:
        while True:
            text = await websocket.receive_text()
            await websocket.send_text(f"echo: {text}")
    except WebSocketDisconnect:
        pass  # 客户端断开，收摊

路由识别靠 websocket: WebSocket 这个类型标注——和 HTTP 端点用 q: str 声明查询参数是同一套机制。

5.3 连接生命周期：accept → receive/send → close

收发全家桶：

• 收：receive_text() / receive_json() / receive_bytes() / receive()（原始 dict）
• 发：send_text() / send_json() / send_bytes() / send()（原始 dict）

三个生命周期补充：

• WebSocketDisconnect 必须接住：客户端关页面、断网、杀进程，服务端的 receive 会抛它（.code 属性是关闭码）。不接住，日志里全是异常栈。常见关闭码：1000 正常关闭、1001 离开、1008 策略拒绝。
• 握手前拒绝：accept 之前直接 await ws.close(code=1008)，握手失败客户端拿 403——鉴权失败的推荐姿势（第 11 章的 token 可在握手时校验，token 走 URL query 或 cookie）。
• 并发收发可以，并发收不行：同一个连接上，一边 receive 一边 send 没问题（对讲机边听边说）；但两个协程同时 await receive() 会抢消息，一条连接配一个收协程。

文档去哪看：https://fastapi.tiangolo.com/zh/advanced/websockets/ ；协议本体 RFC 6455：https://www.rfc-editor.org/rfc/rfc6455.html

5.4 常见坑

• 忘写 accept——握手没完成就 receive/send，当场 RuntimeError: WebSocket is not connected. Need to call "accept" first.。
• /docs 里看不到 WebSocket 路由——OpenAPI 规范不覆盖 WebSocket，交互文档只管 HTTP；WS 接口的文档要自己维护（路径、消息格式、事件类型写进 README 或接口文档）。
• HTTP 中间件全失效——第 10 章三件套对 WS 不生效：计时要在端点里自己掐表；信封更没意义（WS 没有「响应」概念）；CORS 也不设防——浏览器对 WebSocket 不做预检，任何网页都能连你的 WS 服务，想限制就在 accept 前检查 ws.headers.get("origin")。
• 异常处理器也够不着——第 09 章那套住在 HTTP 洋葱最内层；WS 端点里抛出的异常没有统一出口，要自己 try/except 后 send_json({"error": ...})。

5.5 前端对照：new WebSocket()

浏览器四件套，和后端生命周期一一对应：

const ws = new WebSocket("ws://127.0.0.1:8001/ws/room/editors?name=沈墨");

ws.onopen = () => ws.send("编辑部的各位好");        // 对应后端 accept() 完成
ws.onmessage = (e) => console.log(e.data);          // 对应后端 send_text()
ws.onclose = (e) => console.log(e.code, e.reason);  // 对应后端 close() / 断线
ws.onerror = () => {};                              // 出错，随后多伴生一次 onclose

三个注意点：

• URL 协议用 ws://（https 站点用 wss://）。
• EventSource 的自动重连是送的，WebSocket 的重连要自己写——惯用姿势：onclose 里 setTimeout 递归重连，指数退避。
• 消息用 JSON 最常见：ws.send(JSON.stringify({type: "join", name: "沈墨"}))，后端 receive_json() 配对。

Python 侧客户端用 websockets 库（uv add websockets），第 12 章写 WS 测试也能用 TestClient 的 websocket_connect：

import asyncio
import websockets


async def main() -> None:
    async with websockets.connect("ws://127.0.0.1:8001/ws/room/editors?name=沈墨") as ws:
        await ws.send("大家好，我刚提交了《夜航船里的旧灯》初稿")
        while True:
            print(await ws.recv())


asyncio.run(main())

6. ConnectionManager：聊天室的广播台

6.1 为什么需要它

广播 = 拿着所有在线连接挨个 send 一遍。但 WebSocket 对象之间互相不认识，得有人管名单：谁在线（登记）、谁走了（划掉）、有事喊一嗓子（逐个通知）——这就是连接管理器 ConnectionManager。墨集的场景：编辑们连进「编辑室」，谁提交了稿件、哪篇文章过审、哪里来了新评论，全编辑室实时可见。

打个比方：编辑室门口的签到台。进门登记（connect），离开划掉（disconnect），前台有事拿起喇叭挨桌喊（broadcast）。

6.2 代码（节选自 demo/demo_ws.py）

class ConnectionManager:
    def __init__(self) -> None:
        self.rooms: dict[str, list[WebSocket]] = {}   # room_id -> 在线名单

    async def connect(self, room_id: str, websocket: WebSocket) -> None:
        await websocket.accept()                      # 谁登记谁握手
        self.rooms.setdefault(room_id, []).append(websocket)

    def disconnect(self, room_id: str, websocket: WebSocket) -> None:
        room = self.rooms.get(room_id)
        if room and websocket in room:
            room.remove(websocket)

    async def broadcast(self, room_id: str, message: str) -> None:
        for ws in list(self.rooms.get(room_id, [])):  # 快照遍历，见下
            try:
                await ws.send_text(message)
            except RuntimeError:
                self.disconnect(room_id, ws)          # 死连接踢出名单


manager = ConnectionManager()

三个细节：

• 快照遍历：list(...) 先拷一份名单再循环，循环体里删元素才安全——边遍历原列表边删，会跳过元素甚至抛错。
• 兜 RuntimeError：给已断开的连接 send 会抛异常，一条死连接不能毒死整轮广播。
• accept 放进 connect：连接管理的固定套路——「登记」和「握手」绑定在一起，业务代码里就不会忘。

6.3 聊天室主循环 + HTTP 配合

@app.websocket("/ws/room/{room_id}")
async def chat_room(
    websocket: WebSocket,
    room_id: str,
    name: str = Query(default="匿名编辑", max_length=20),
) -> None:
    await manager.connect(room_id, websocket)
    await manager.broadcast(room_id, f"【系统】{name} 进入了编辑室（当前 {manager.count(room_id)} 人在线）")
    try:
        while True:
            text = await websocket.receive_text()     # 没消息就挂起，不空转
            await manager.broadcast(room_id, f"{name}：{text}")
    except WebSocketDisconnect:
        manager.disconnect(room_id, websocket)
        await manager.broadcast(room_id, f"【系统】{name} 离开了编辑室（剩余 {manager.count(room_id)} 人在线）")

再配一个 HTTP 入口，演示「通知从 HTTP 进、从 WS 出」——业务系统照常走 HTTP，WS 只当广播喇叭，这是实时通知类功能的标准姿势：

@app.post("/api/notify")
async def notify(
    article: str = "夜航船里的旧灯", event: str = "published", room: str = "editors",
) -> dict[str, str]:
    message = f"【协作通知】《{article}》已发布"
    await manager.broadcast(room, message)
    return {"status": "ok", "message": message}

实测（两个 websockets 客户端各占一个终端，再 curl -X POST http://127.0.0.1:8001/api/notify）：

终端 A（沈墨）                                终端 B（阿蓝）
<- 【系统】阿蓝 进入了编辑室（当前 2 人在线）
<- 沈墨：大家好，我刚提交了《夜航船里的旧灯》初稿
                                             <- 沈墨：大家好，我刚提交了《夜航船里的旧灯》初稿
                                             <- 【协作通知】《夜航船里的旧灯》已发布      ← curl 触发
<- 【系统】阿蓝 离开了编辑室（剩余 1 人在线）   ← B 的终端被关掉，A 立刻知道

curl -X POST http://127.0.0.1:8001/api/notify 的响应：

{"status":"ok","message":"【协作通知】《夜航船里的旧灯》已发布"}

6.4 常见坑

• 内存名单只认单进程：uvicorn --workers 4 时每个进程一份 manager，A 进程的连接收不到 B 进程广播的 broadcast。多进程/多机部署要上 Redis Pub/Sub 之类的消息总线把广播跨进程转发——demo 不展开，知道边界在哪就行。
• 忘 disconnect——名单越积越长，全是幽灵连接，每轮广播挨个抛异常。
• 逐个 await 发送会互相拖累——一条慢连接能让整轮广播变慢；连接量大时可换 asyncio.gather 并发发。demo 量级小，顺序发足够且更好读。

7. 选型对比：SSE 还是 WebSocket

选型口诀：只需要服务器往外推，用 SSE；客户端也要频繁说话，才上 WebSocket。

AI 逐字输出是教科书级的 SSE 场景：你把提问发给模型，走一条普通 HTTP POST 就够了；模型吐字给你，走一条 SSE——「说」和「听」两条路各干各的，没必要为听单口相声开一条对讲机专线。反过来，墨集编辑室的多人协作，编辑随时要发言、系统随时要推通知，双向高频，WebSocket 才对路。

8. 跑一下本章 demo

两个文件、两个端口（故意错开，可以同时跑）：

uv add fastapi "uvicorn[standard]" sse-starlette
python demo/demo_stream.py   # 8000 端口：流式 + SSE + AI 逐字输出
python demo/demo_ws.py       # 8001 端口：WebSocket 聊天室

demo_stream.py 验证清单（以下均为实测输出，节选）：

# ① StreamingResponse 基本款：日志逐条蹦（响应头无 content-length）
curl -i http://127.0.0.1:8000/api/logs/stream

# ② 文件下载：Content-Disposition 触发浏览器「另存为」
curl -i http://127.0.0.1:8000/api/reports/download

# ③ 手写 SSE 报文：自己拼 id/event/data + 空行
curl -N http://127.0.0.1:8000/api/ai/write-raw

# ④ 本章主角：AI 逐字输出（-N 关掉 curl 缓冲，别忘）
curl -N http://127.0.0.1:8000/api/ai/write

# ⑤ 断点续传：从 id: 10 接着吐
curl -N -H "Last-Event-ID: 9" http://127.0.0.1:8000/api/ai/write

demo_ws.py 验证清单：

# HTTP 侧群推协作通知（配合任意 WebSocket 客户端验证）
curl -X POST http://127.0.0.1:8001/api/notify
curl -X POST "http://127.0.0.1:8001/api/notify?event=review"

WebSocket 客户端两种开法（demo 文件头注释里有完整可粘贴版本）：

• Python：websockets.connect(...) 开两个终端各连一个，体验「一条消息两边都收到」；
• 浏览器：任意页面 F12 控制台 new WebSocket("ws://127.0.0.1:8001/ws/room/editors?name=沈墨")，WebSocket 握手不受同源策略限制。

9. 小结速查

自测三问：

1. StreamingResponse 的响应头里为什么没有 content-length？流已经开吐之后端点函数抛了异常，客户端收到的状态码是什么、为什么改不了？
2. SSE 手写报文漏了 \n\n 会发生什么？Last-Event-ID 是谁放进请求头的、在什么时机？
3. 第 10 章的三个中间件为什么对 WebSocket 全部失效？由此推论，WS 端点里哪两件事得自己干？

10. 全课程收官

十四章前，你在 01 章敲下第一行 uvicorn.run(app, ...)，墨集 inkhub 开了张；十四章后的现在，把整条路串一遍：一个请求从 Uvicorn 进门（01），对上路由、拿到参数（02），被 Pydantic 过一遍安检（03），在 SQLAlchemy 的异步会话里读写（04~06），库表结构跟着 Alembic 演进（07），一路上依赖注入递工具（08）、异常有统一的病历单（09）、中间件在出口盖计时章、打包信封（10）、门口查 JWT 令牌（11）；上线之前 pytest 过一遍（12），跑起来之后 loguru 盯着日志（13）；最后，该一次给完的快速给完，不该等的——大文件、慢生成、AI 逐字输出——学会让它边做边给（14）。一套 API 服务的完整零件箱，你已经一件一件摸过了。

课程到这里正式完结，但你的路才刚开始。几个值得去的方向：

• 把这套东西用到一个真实项目里——真实需求永远是最好的下一章：给自己写个工具、给团队写个内部服务、接一个真模型的 API 都行，零件箱里的东西只有在真项目里拧过螺丝才算你的。
• AI 方向：把第 4 节的模拟端点接到真模型上（OpenAI 兼容接口、本地模型皆可），SSE 链路一模一样；再往后是 Function Calling、RAG，那些都是「FastAPI 当传送带」的活。
• 部署方向：nginx 反代下 SSE 的 proxy_buffering off、WebSocket 的 Upgrade 放行、多进程下的 Redis 广播总线——本章「常见坑」里点过名的账，实战里都会真实遇到。

从第一个 Hello World 到一条逐字流淌的流式响应，你走完了。去写点什么吧。

全课程完 · 返回 00 · 使用指南