流式输出:让用户 0.5 秒就看到东西

实测:同一个问题,两种给法

① 阻塞式:等它全写完才返回
  用户盯着空白屏幕:4.16 秒
  然后一次性收到 487 字(输出 285 tok)

② 流式:边生成边吐
  首字出现:0.51 秒  ← 用户从这一刻起就"看见东西在动"
  全部写完:4.02 秒(共 228 个数据块,400 字)

③ 差距
  总耗时差不多(4.16s vs 4.02s),但"用户开始看到内容"的时刻:
    阻塞式 4.16s → 流式 0.51s   快了 8.2 倍

记住这个词:TTFT(首字延迟)——它才是用户感知的"快慢",不是总耗时。做产品优化时,盯 TTFT 比盯总时长有用得多。

流式怎么收:一行一个 data

流式接口返回的是 SSE(Server-Sent Events):一行一个 data:,每行一小块内容,最后一行 data: [DONE]。核心代码就这几行:

with urlopen(req) as resp:              # 请求体加 "stream": True
    for raw in resp:
        line = raw.decode("utf-8").strip()
        if not line.startswith("data:"): continue
        data = line[5:].strip()
        if data == "[DONE]": break
        delta = json.loads(data)["choices"][0]["delta"].get("content", "")
        if delta:
            print(delta, end="", flush=True)   # 真实产品里:推给前端

从后端到前端的完整链路(AP7 会做完整版):模型 SSE → 你的后端逐块转发(也是 SSE)→ 前端边收边渲染。中间任何一环做了缓冲,流式就白做了——这是最常见的翻车点。

流式的三个工程代价

爽是爽的,但要知道代价(AP6 会逐个治):

1. 错误可能发生在"已经吐了一半"之后——用户看到半句话然后断了。需要兜底:前端标记"生成中断",后端记录部分输出;
2. 用量统计要在流结束时收口——中途断开的请求也消耗了 token,别漏记(ap1.6 的成本仪表要处理);
3. 校验类功能不能流式——比如 AP3 的 JSON 校验、AP6 的内容安全过滤,都得等完整输出才能判断。结构化输出别用流式,这是一条硬规矩。

🔧 动手做:亲手感受 8 倍差距(10 分钟)

1. 跑本节代码 streaming.py,看你自己的两个数字:阻塞总耗时 vs 流式首字延迟。
2. 把输出改长:把问题从"300 字"改成"1200 字",两个版本各跑一次——阻塞式的等待时间涨了几倍?流式的首字延迟涨了吗?(这就是流式最值钱的地方:输出越长,优势越大。)
3. 在流式循环里加一行 print(delta, end="", flush=True),亲眼看字一个个蹦出来的感觉。
4. 制造中断:流式跑到一半按 Ctrl+C,想一想:这次消耗的 token 你的系统记账了吗?(记不上——这就是代价 2,ap1.6 要解决。)
5. 记一条决策进笔记:我的毕业产品里,哪些接口该流式(对话/长文生成),哪些不能(结构化提取/需要校验的)。

✅ 做到这里你应该有:你自己机器上的 TTFT 数字,和一份"哪些接口走流式"的清单。

为什么:流式是零成本的体验升级——不多花一分钱,用户感知速度快 8 倍。但它也是第一个"体验和工程复杂度"的权衡点:你多了断流处理、用量收口、不能中途校验三个问题。知道代价再选,才叫工程决策。

我要给【你的产品】做流式输出:后端【技术栈】,前端【技术栈】。请给出:1) 后端如何逐块转发 SSE(含 flush 要点);2) 前端如何边收边渲染;3) 中断/报错时两端各怎么兜底;4) 哪些中间层(代理/CDN/网关)可能缓冲流,怎么验证。