大模型流式响应的设计与实现

可视化演示

动态演示大模型流式响应的设计与实现的核心机制，通过采集/分析/决策/反馈四种模式展示关键概念与交互流程。

流式响应概念：从寄信到打电话

你给远方的朋友寄了一封信，问“我该不该换工作？”。朋友很认真，花了一整天把回信写好再寄回来。三天后你拆开信，一篇完整的分析跃然纸上——内容是好，可这三天里你什么也做不了。非流式的大模型接口就是这种感觉：用户提问后，服务端要等整段回答全部生成完毕，才一次性把结果返回，屏幕上只有一个转圈的光标。

流式响应（Streaming Response）改变了交付方式：服务端每生成一小段内容（token），就立刻通过保持开启的连接推给前端。用户第一秒就能看到“好的，我在想……”，随后内容像打字机一样持续涌出。它并没有缩短“生成完整个回答”的总时间，却极大缩短了“从提问到看到第一个字”的时间。

把用户的感知等待拆开看，它由两部分组成：第一个字之前的等待，以及剩余内容逐步到达的时间。非流式下，首字延迟等于全部生成耗时；流式下单字延迟可以压到几百毫秒。

$$\displaystyle T_{\text{感知}} = T_{\text{TTFT}} + \frac{D}{v_{\text{显示}}}$$

1. T感知（总感知等待）：用户从发出问题到读完整个回答所感受到的总耗时。

$$\displaystyle T_{\text{感知}}$$

2. T_TTFT（首字延迟）：从请求发出到收到第一个 token 的时间，流式下决定产品的“第一印象”。

$$\displaystyle T_{\text{TTFT}}$$

3. D（剩余内容量）：第一个 token 之后还需要输出的字符数。

$$\displaystyle \frac{D}{v_{\text{显示}}}$$

4. v显示（显示速度）：内容到达并渲染到屏幕上的速率，流式下由生成速度决定。

$$\displaystyle v_{\text{显示}}$$

明白了这个公式，就理解了为什么“打电话”比“寄信”体验好：电话的第一声“喂”只需一瞬间，而信件的第一行字要等上三天。流式要做的事，就是让这声“喂”尽快传到用户耳边，然后让余下的话源源不断。接下来，我们看看这条“电话线”底层的协议——SSE。

SSE协议原理：电话线如何保持畅通

电话能一直说下去，前提是两边都不挂断。传统 HTTP 请求是“一问一答”：客户端发请求，服务端返回响应，连接就关闭了。而 SSE（Server-Sent Events，服务器发送事件）复用普通 HTTP 协议，服务端把响应头里的 Content-Type 设为 text/event-stream，再加上 Connection: keep-alive，就等于对客户端说：“喂，我不挂电话，你也别挂。”

SSE 为什么能用普通 HTTP 实现流式？因为 HTTP 响应体本质上是一个可以持续写入的字节流。只要服务端不结束响应，连接就一直开着，数据就能一块一块往下发。前端通过 EventSource 或 fetch 读取这个流，每收到一段数据就触发一次处理逻辑——这正是 ChatBot 打字机效果的底层机制。

SSE 把数据组织成一个个“帧”，帧与帧之间用空行分隔。每个帧可以带四类字段：data 放内容、id 放序号、event 放事件类型、retry 告诉客户端断线后多久重连。多个帧按顺序排列，就构成一条“事件流”：

$$\displaystyle \mathcal{F} = [ f_1, f_2, \dots, f_n ], \qquad f_i = (\mathrm{id}_i,\ \mathrm{event}_i,\ \mathrm{data}_i)$$

1. F（事件流）：从连接建立到关闭期间收到的全部帧的集合。

$$\displaystyle \mathcal{F} = [ f_1, f_2, \dots, f_n ]$$

2. fi（单个帧）：SSE 传输的基本单位，对应一次“推送”。

$$\displaystyle f_i = (\mathrm{id}_i,\ \mathrm{event}_i,\ \mathrm{data}_i)$$

3. id_i（帧序号）：由服务端递增分配，用于断线后告诉服务端“我收到哪了”。

$$\displaystyle \mathrm{id}_i$$

4. data_i（数据载荷）：帧携带的实际内容，通常是 JSON 字符串。

$$\displaystyle \mathrm{data}_i$$

打电话时，你不需要先写封信确认对方在听，开口说话对方就收到了。SSE 也一样：客户端只需发起一次 HTTP 请求，之后服务端就可以不断推送帧。相比 WebSocket 需要握手升级、是全双工通信，SSE 基于普通 HTTP、单向推送、自动重连，是“服务端给客户端单向报话”场景的最简解。

后端流式输出：后厨的分菜窗口

想象一家餐厅。传统做法是：后厨把所有菜全部炒完，才一次性端上桌，客人饿着肚子干等。流式做法呢？后厨每一道菜出锅就立刻从出菜窗口递出去，客人边吃边等，永远有东西在嘴里。

对应到代码，后端不再“拼接完整字符串后 return”，而是把一个生成函数改造成生成器（generator）：每生成一个 token，就用 yield 交出来。Python 里的常见做法是：业务函数不断 yield 文本块，Web 框架拿到生成器后，把它包装成 StreamingResponse，或者直接向响应体写入并 flush。

大模型推理接口普遍支持流式参数。以 OpenAI 兼容接口为例，请求体带 stream: true 时，返回的不再是单条 JSON，而是一串以 data: 开头的数据帧；每个帧里的 delta.content 携带一小段增量文本。后端把这些增量一一拆出来、再包装成 SSE 帧推给前端。这中间的“搬运速度”可以用吞吐率衡量：

$$\displaystyle \rho = \frac{N_{\text{token}}}{T_{\text{总}}}$$

1. ρ（吞吐率）：单位时间内完成生成的 token 数量，衡量接口的“出菜速度”。

$$\displaystyle \rho$$

2. N_token（总 token 数）：回答总共拆成多少个 token，中文场景下每个字大约对应 1~2 个 token。

$$\displaystyle N_{\text{token}}$$

3. T总（总耗时）：从请求开始到最后一个 token 生成完毕的时间。

$$\displaystyle T_{\text{总}}$$

出菜窗口的快慢决定了客人能否“一直有菜吃”。后端流式输出的核心，就是把“一锅端的整段 return”改造成“一道一道出菜的 yield”，再配合 HTTP 流的 flush，让增量内容像热菜一样源源不断送往前台。接下来看前端那侧如何“接菜”。

前端接收方案：拿起听筒听到话

电话那头开口了，你这边得把听筒贴到耳朵上，而不是每听一句就挂断重拨。前端接收流式响应，同样要“持续保持连接地听”。SSE 官方推荐的写法是 EventSource——一个专门用来“听”事件的浏览器 API，收到一帧就触发一次回调。

但 EventSource 有两个短板：只能用 GET 请求，且不能自定义请求头。而主流大模型接口大多要求 POST + Authorization 鉴权，于是更通用的方案是 fetch + resp.body.getReader()：把响应体当作可读流，循环读取一个个数据块，再把字节拼成文本、按行切出 data: 帧。

内容拿到之后，剩下的是“怎么显示”。如果每来一个字节就更新一次 DOM，浏览器可能吃不消；如果攒太多再更新，又退回“寄信”体验。工程上通常做增量渲染：把增量文本追加到缓冲，再配合节流，控制在每帧最多更新一次，形成平滑的“打字机效果”：

$$\displaystyle v_{\text{渲染}} = \min(v_{\text{生成}},\ v_{\text{帧上限}})$$

1. v渲染（渲染速度）：前端把增量文本写入页面并显示的速度。

$$\displaystyle v_{\text{渲染}}$$

2. v生成（生成速度）：服务端推送 token 的速度，由模型推理决定。

$$\displaystyle v_{\text{生成}}$$

3. v帧上限（屏幕显示上限）：受浏览器帧率与 DOM 更新成本约束的显示速度上限。

$$\displaystyle v_{\text{帧上限}}$$

4. min（取较小值）：保证渲染速度不会超过任何一个上限。

$$\displaystyle \min(\cdot, \cdot)$$

听筒贴稳了，声音自然连贯。前端要做的事总结起来就三件：选对“听”的通道（EventSource 或 fetch 流）、把字节流还原成文本帧、用节流的增量渲染把文字“打”出来。但电话总有信号不好的时候，用户也可能突然不想听了——下一节聊聊挂断与重拨。

中断与重连：遇到忙音怎么办

打电话最扫兴的是什么？聊到一半断线。也可能是你不想听了，主动挂断。流式场景里，“主动挂断”对应前端停止生成，“意外断线”对应网络抖动导致连接中断，这两件事都得有预案。

主动停止很简单：如果用 fetch + ReadableStream，调用 AbortController.abort() 即可中断请求，同时告诉后端“别再生成浪费算力了”，后端收到断开信号会停止迭代。如果用的是 EventSource，调用 close() 优雅关闭。这是“挂断电话”的标准动作。

意外断线则需要重连策略。EventSource 内置自动重连，并遵循服务端 retry 字段指定的间隔；而 fetch 方案要自己实现。业界常用指数退避：前几次失败快速重试，之后间隔越来越大，防止把服务端打爆：

$$\displaystyle t_k = \min\left(t_0 \times 2^{k}, \ t_{\max}\right),\qquad k = 0,1,2,\dots$$

1. tk（第 k 次重试等待）：第 k 次重连前要等待的时间。

$$\displaystyle t_k$$

2. t0 × 2^k（指数增长部分）：间隔随重试次数按 2 的幂次翻倍。

$$\displaystyle t_0 \times 2^{k}$$

3. tmax（上限）：防止无限指数增长设定的最大等待时间，到达后不再翻倍。

$$\displaystyle t_{\max}$$

4. min（取较小值）：让等待时间在上限处封顶，保持恒定。

$$\displaystyle \min(\cdot, \cdot)$$

挂断要爽快，重拨要聪明。把“主动停止”和“被动重连”两条路径都设计好，你的“电话”才算真正能打。但现实中还有一类更隐蔽的问题：你以为电话通了、对方也开口了，可声音被“缓冲”卡在半路，一个字都没传过来——这就是下一节的“杂音”。

常见踩坑：电话里的杂音

你有没有过这种经历：对方明明在说话，可话筒里一片死寂，隔了几秒突然炸出一大段话？这在电话里叫“信号延迟”，在流式项目里叫“缓冲陷阱”——数据其实已经生成了，却被某个环节攒着，迟迟没有推给用户。

缓冲陷阱的根源有三处：一是应用层忘了 flush，字节窝在语言运行时的输出缓冲区里；二是 Web 框架或反向代理（如 Nginx）默认开启缓冲，SSE 响应体会被攒成整块再放行；三是运营商或网关可能把长时间挂着的连接当成异常掐断。标准姿势是：服务端显式 flush，并在响应头加 Cache-Control: no-cache、X-Accel-Buffering: no 来关闭中间层缓冲。

用公式描述这个状态最直观：只要任何一环没有把“已生成的数据”交出来，首字节时间就趋近于无穷大，流式形同虚设：

$$\displaystyle T_{\text{首字节}} = \begin{cases} t_{\text{首个chunk}}, & \text{所有环节均刷新}\\ +\infty, & \text{存在缓冲未刷新} \end{cases}$$

1. T首字节（首字节时间）：用户实际收到第一个字节的时间。

$$\displaystyle T_{\text{首字节}}$$

2. t_chunk（生成首块时间）：服务端生成第一块数据的时间，通常只有几十毫秒。

$$\displaystyle t_{\text{首个chunk}}$$

3. ∞（无穷大）：代表缓冲没刷新时用户永远收不到内容，连接被“冻住”。

$$\displaystyle +\infty$$

4. 分段函数：一句话结论——好体验的前提是“每一环都在实时交付”。

$$\displaystyle \begin{cases} \cdots \\ \cdots \end{cases}$$

处理杂音最好的办法，是从源头消灭它：验证流是否生效、逐跳检查缓冲、给代理配上“放行”头。排查顺序可以靠这个公式倒推：先看服务端首块时间是否正常，再检查中间有没有“攒着不放”的环节。把坑填平之后，最后聊聊怎么让这条电话线更快更稳。

性能优化：让通话更流畅

同一个电话网络，有人三秒接通，有人等三十秒转人工。流式性能优化的目标，就是让“接通”更快、“通话”中听得更连贯。核心指标就两个：首字时间（TTFT）和整体吞吐。

降低 TTFT 的招数集中在第一个 token 出现之前：启用 prompt caching，相同前缀的 prompt 直接命中缓存，省去重复计算；服务端生成第一批内容后立刻 flush，让响应头和首帧尽快到达；模型推理侧优化 prefill 阶段，让第一个 token 尽快“算出”。

提升长期吞吐则靠并发与批处理：把多个用户的请求拼成 batch 一起推理，GPU 利用率更高；合理设置 max_tokens，避免回答被截断；前端对高频 chunk 做合并渲染，减少 DOM 更新次数。综合起来，用户感知的总时间可以分解为：

$$\displaystyle T_{\text{用户感知}}=T_{\text{TTFT}}+\frac{N_{\text{总token}}-N_{\text{首块}}}{r_{\text{生成}}}+T_{\text{渲染}}$$

1. T_TTFT（首字时间）：请求发出到第一个字出现的时间，决定产品“第一印象”。

$$\displaystyle T_{\text{TTFT}}$$

2. N总 与 N首块（剩余内容量）：总 token 数减去首个数据块已包含的 token 数，就是剩下要生成的内容量。

$$\displaystyle \frac{N_{\text{总token}}-N_{\text{首块}}}{r_{\text{生成}}}$$

3. r生成（生成速率）：模型每秒生成的 token 数，由算力和 batch 大小决定。

$$\displaystyle r_{\text{生成}}$$

4. T渲染（渲染开销）：前端把内容画到屏幕上的时间，节流后可忽略不计。

$$\displaystyle T_{\text{渲染}}$$

一通流畅的电话，讲究“接得快”和“说得稳”。后端压 TTFT、提升 batch 吞吐，前端做增量渲染、控制更新频率，双管齐下，用户才能真正感到“丝滑”。至此，从寄信到打电话、从协议到前后端配合、再到中断与优化，一条完整的流式链路已经打通——剩下的，就是你在自己的项目里把这通电话打起来。

AI 应用场景

Python 演示

import jsonimport threadingimport timeimport urllib.requestfrom http.server import BaseHTTPRequestHandler, HTTPServer  def token_stream(prompt):    """模拟大模型逐 token 生成回答, 产出 SSE 帧"""    answer = "Hello! You asked: " + prompt + ". This is a streamed reply."    for ch in answer:        chunk = {"choices": [{"delta": {"content": ch}}]}        frame = "data: " + json.dumps(chunk, ensure_ascii=False) + "\n\n"        yield frame        time.sleep(0.03)  def read_sse(url):    """客户端逐帧读取 SSE 流并打印"""    resp = urllib.request.urlopen(url)    for line in resp:        text = line.decode("utf-8").strip()        if not text.startswith("data: "):            continue        payload = json.loads(text[len("data: "):])        delta = payload["choices"][0]["delta"]["content"]        print(delta, end="", flush=True)    print()  class SSEHandler(BaseHTTPRequestHandler):    def do_GET(self):        if self.path != "/chat":            self.send_response(404)            self.end_headers()            return        self.send_response(200)        self.send_header("Content-Type", "text/event-stream")        self.send_header("Cache-Control", "no-cache")        self.send_header("Connection", "keep-alive")        self.end_headers()        for frame in token_stream("streaming"):            self.wfile.write(frame.encode("utf-8"))            self.wfile.flush()     def log_message(self, *args):        pass  if __name__ == "__main__":    server = HTTPServer(("127.0.0.1", 8000), SSEHandler)    thread = threading.Thread(target=server.serve_forever, daemon=True)    thread.start()    time.sleep(0.2)    print("SSE demo: reading from http://127.0.0.1:8000/chat")    read_sse("http://127.0.0.1:8000/chat")    server.shutdown()

1. def token_stream(prompt):用生成器函数模拟大模型逐 token“开口”，yield 一帧就交付一帧，绝不攒着。
2. frame = "data: " + json.dumps(chunk, ensure_ascii=False) + "\n\n"按 SSE 规范组装帧：数据以 data: 开头、以空行结尾。
3. self.send_header("Content-Type", "text/event-stream")关键响应头，告诉客户端“这是事件流，不要按普通请求处理”。
4. self.wfile.flush()每写一帧立即冲刷缓冲区，防止字节窝在缓冲区里形成“寄信”。
5. text[len("data: "):]客户端剥掉 data: 前缀，取出真正的 JSON 载荷。
6. threading.Thread(target=server.serve_forever, daemon=True)把服务端放进后台线程，主线程同时扮演客户端，一条命令演示完整链路。

常见误区

面试题