代码语言

知识点思维导图

35 个知识节点

参考资料

生产工程（03） - 流式响应

读完后，你应能完成以下任务：

绘制“生产工程（03） - 流式响应 / token 流：模型本来就是一个个字生成的”的关键对象与数据流，解释“很多人以为模型是「先想好整句、再一次性返回」，其实不是。”，并用源码位置、日志或 Trace 标注证据。

为“生产工程（03） - 流式响应 / SSE：把 token 流通过 HTTP 推给前端”设计正常与异常输入，验证“普通 JSON 响应是「一锤子买卖」，不适合持续推送。”，输出首个偏差位置与回归测试结果。

实现“生产工程（03） - 流式响应 / 停止生成：用户的中断权”的最小代码或配置，检验“并且保留已经生成的部分（不是丢弃）。”，输出命令、结果与 Diff，并说明不适用边界。

一、与进阶篇的分工

本篇保留为流式响应基础：重点讲 SSE、打字机效果和前端状态。进阶实现请读 70《Nest + LangChain 实现基于 SSE 的流式 ai 接口》、73《给 Agent 加上语音交互》、74《AGUI 协议》，它们分别覆盖后端流式接口、语音流式和组件事件流。

二、流式响应的真实应用场景

你做的 AI 助手，用户问一句话，模型要 5 秒才能生成完整回答。如果等全部生成完再一次性返回，用户盯着转圈的 loading 干等 5 秒——体感很差，还会怀疑是不是卡死了。

ChatGPT 给的答案是：边生成边显示，字一个个往外蹦。用户在第 0.3 秒就看到第一个字，虽然总时长没变，但「在动」这件事让等待变得可接受。这就是流式响应。它解决的不是速度问题，是等待体验问题。

对前端来说，这意味着接口不再是「发一个请求、收一个完整响应」，而是「发一个请求、持续收到一串小片段」。承接方式也得变。

三、token 流：模型本来就是一个个字生成的

很多人以为模型是「先想好整句、再一次性返回」，其实不是。模型是自回归的：生成完一个 token，把它接到输入后面，再生成下一个，循环往复。所以「流式」不是额外加工，而是模型本来的工作方式——只是非流式接口帮你攒齐了再给。

在 Python 里，这种「产出一个、暂停、再产出一个」的语义，天生就该用生成器：

yield 的妙处：函数执行到这里会暂停，把一个 token 交出去，调用方处理完再回来要下一个。真实模型 API 开 stream=True 时，返回的就是这样一个可迭代对象，你 for token in stream 挨个取就行。

四、SSE：把 token 流通过 HTTP 推给前端

后端有了 token 流，怎么传给浏览器？普通 JSON 响应是「一锤子买卖」，不适合持续推送。最常用的方案是 SSE（Server-Sent Events），一种基于 HTTP 的单向推送协议（服务器→浏览器）。

SSE 的报文格式很简单，每个事件三部分：

event: token
data: 报

event: token
data: 销

event: done
data: {"finish_reason":"stop"}

规则：event: 一行写事件类型， data: 一行写内容，空行表示一个事件结束。前端用浏览器内置的 EventSource 监听，按 event 类型分发：

事件类型	前端动作
`token`	追加到当前回答气泡，制造打字机效果
`done`	标记本轮结束，把完整回答存进历史
`error`	停掉 loading，显示可重试的错误提示

为什么用 SSE 而不是 WebSocket？因为这里是单向推送（只有服务器往客户端发）， SSE 更轻、自带断线重连、走标准 HTTP。要双向实时（比如协同编辑）才上 WebSocket。

五、停止生成：用户的中断权

长回答场景，用户经常生成到一半发现跑题了，想点「停止」。后端要能响应这个中断：停止从模型拉后续 token、释放连接，并且保留已经生成的部分（不是丢弃）。

注意两点：一是已生成的 collected 要留着，用户可能还想看；二是真实项目里中断时要主动关掉到模型的连接，否则模型那边还在生成、还在计费。

六、工程上真正会踩的坑

SSE 少了那个空行，前端 EventSource 收不到完整事件，表现为「一直没消息然后突然全来了」。data 后面必须跟空行。
data 内容里有换行，会把一个事件拆成多个。多行内容要么转义，要么每行都加 data: 前缀，传 JSON 时尤其注意。
断连后前端重复追加 token。SSE 自动重连后如果后端从头重发，前端会把已显示的内容再追加一遍。要么后端记录发送位置续传，要么前端按事件 id 去重。
流式了就不记日志。token 是流式发的，但完整答案、token 用量、耗时这些最后还是要落库，否则坏 case 没法复盘。流式和日志不冲突，流结束时统一记一条。

七、一句话面试答法

流式响应怎么实现，要注意什么？ 模型本身是一个 token 一个 token 自回归生成的，流式就是不攒齐、边生成边返回。后端用生成器逐个产出 token，通过 SSE 协议推给前端——SSE 是基于 HTTP 的单向推送，格式是 event/data 加空行，前端用 EventSource 按事件类型分发。要处理几件事：用户停止生成时中断并保留已生成部分、断连重连时防止 token 重复追加、流结束后仍要把完整答案和 token 用量落日志。单向推送用 SSE 就够，双向才上 WebSocket。

八、动手实践：13 流式响应

用 Python 生成器模拟大模型的 token 流，做出打字机效果，并展示 SSE 事件的真实报文格式、以及「停止生成」怎么中断。终端里能直接看到字一个个蹦出来。

8.1 在线运行

零依赖，纯标准库。终端运行时场景 1 和 3 有逐字动画，下面是去掉动画后的最终输出。

8.2 预期输出

=== 场景 1：打字机效果（逐 token 显示）===
助手：报销需在费用产生后 30 天内提交，附发票和审批单。
（流结束，后端拿到完整答案落库：报销需在费用产生后 30 天内提交，附发票和审批单。）

=== 场景 2：底层 SSE 事件长什么样（前 3 个 token + done）===
'event: token\ndata: 报\n\n'
'event: token\ndata: 销\n\n'
'event: token\ndata: 需\n\n'
'event: done\ndata: {"finish_reason":"stop"}\n\n'

=== 场景 3：中途停止生成（用户不想等了）===
助手：报销需在费用产生 [用户点击停止生成]
（已生成部分被保留：报销需在费用产生）

场景 2 用 repr 把换行符显出来，能看清 SSE 的格式：event: 一行、data: 一行、空行收尾。前端 EventSource 就靠这个结构分发事件。

8.3 代码↔概念对应

概念	在 main.py 哪里
用生成器模拟 token 流（边生成边产出）	`stream_tokens`
SSE 事件格式（event/data/空行）	`to_sse_event`
打字机渲染：收到一个 token 就追加显示	`typewriter`
流结束后拼成完整答案（落库）	`typewriter` 的返回值
中途停止生成 + 保留已生成部分	`stream_with_stop`

8.4 动手改

把 stream_tokens 的 delay 调大到 0.2，打字机效果更明显；调成 0 就是「秒回」。
把 FULL_ANSWER 换成多行长文本，观察打字机怎么逐行铺开。
真实项目里 stream_tokens 换成模型 API 的 stream=True 返回的迭代器，to_sse_event 那套格式原样可用，前端代码不用动。

8.5 可运行源码：流式响应

main.py

九、总结

token 流：模型本来就是一个个字生成的：很多人以为模型是「先想好整句、再一次性返回」，其实不是。
SSE：把 token 流通过 HTTP 推给前端：普通 JSON 响应是「一锤子买卖」，不适合持续推送。
停止生成：用户的中断权：并且保留已经生成的部分（不是丢弃）。
一句话面试答法：模型本身是一个 token 一个 token 自回归生成的，流式就是不攒齐、边生成边返回。