Day 12 / 共 68 天 · 阶段 2 大模型基础
流式输出
昨天(Day 11)你打通了第一通"云端顾问电话",但回答要等它全想完才一次性蹦出来,长答案干等好几秒。今天学 流式输出:让文字像 ChatGPT 那样一个字一个字往外冒。你会搞懂背后的 SSE、怎么用 stream=True 逐块接收拼接。学完你的 AI 应用体验就"专业"了;明天(Day 13)算清一次调用到底花多少钱、快不快。
📍 你在阶段 2(大模型基础 D10-15)的位置
D10 大模型是什么→
D11 调 LLM API→
D12 流式输出→
D13 参数与成本→
D14 结构化输出
💡 用一套类比兜住今天(今天全程沿用「餐厅上菜」的世界观)
非流式 = 大厨把整桌菜全做好,一次性推出来——你干坐着等,桌上空空如也;流式 = 做好一道先端一道,你边吃边等下一道,全程有事看、有的吃。SSE = 后厨到餐桌的传菜通道(单向、一趟一趟送);chunk(数据块) = 一次端来的一小碟;拼接 = 你把陆续端来的小碟攒成完整一餐。今天你学会"边做边上菜"。
L01
为什么要流式:别让用户干等
🤔 痛点昨天的代码问一个长问题,屏幕一片空白,三五秒后才"啪"地把整段答案甩出来。用户以为卡死了,早就没耐心——这是所有 AI 应用体验的头号杀手。
💡 本质模型本来就是一个词一个词生成的(回忆 Day10)。非流式是"等它全部生成完再一次给你";流式是"生成一点就送你一点"。就像餐厅:与其让你空等一桌菜,不如做好一道先端一道。
图注:总时长其实差不多,但流式让用户"立刻看到反应",感知快得多——体验是主观的。
👶 关键点流式并没有让模型算得更快,改变的是你什么时候开始看到结果。首字出现时间(TTFT)从几秒降到零点几秒,用户的"焦虑感"就没了。这是纯体验优化,却是产品好用与否的分水岭。
L02
流式的本质:SSE 一趟一趟送
🤔 痛点普通请求是"我问一句、它答一句、连接就断"。可流式要在一次连接里连续送来很多小段——这靠什么技术做到?
💡 本质靠 SSE(Server-Sent Events,服务器推送事件):连接建立后先不挂断,服务器一有新内容就沿着这条通道推一小块过来,直到发完一个结束信号才收工。就是后厨到餐桌那条一直开着的传菜通道,一碟一碟往外送。
👶 小白:这跟我 Day07 学的普通 HTTP 请求有啥不一样?
👨🏫 老师:普通 HTTP 是"一问一答一挂断"——请求出去、完整响应回来、结束。SSE 是在 HTTP 之上的一种玩法:响应不一次性给完,而是让连接开着、分很多次往回推,每次推一小块(chunk),最后推一个"结束"标记。好消息是——这些底层细节 SDK 全帮你封装好了,你只要用一个 for 循环去"接"这些块就行(下一讲就写)。
👶 名词别怕你会听到 SSE、chunk、delta、token 这些词。一句话对应:SSE=传菜通道,chunk=端来的一碟,delta=这一碟里"新增"的那口菜(增量文本),token=模型眼里的最小食材单位(回忆 Day10)。记不住先混脸熟,写两次代码就懂。
L03
stream=True:一个开关切成流式
🤔 痛点听起来流式很高级,是不是要重写一大堆代码、学新库?
💡 本质不用!在昨天的调用里加一个参数
stream=True,返回的就不再是"一整段回答",而是一个可以用 for 循环遍历的"数据块流"。改动极小。from openai import OpenAI
client = OpenAI()
stream = client.chat.completions.create( # 注意:返回的不是回答,是一个"流"
model="gpt-4o-mini",
messages=[{"role": "user", "content": "用三句话介绍长城"}],
stream=True, # ← 唯一的关键改动:打开流式开关
)
for chunk in stream: # 像遍历列表一样,一块一块接
delta = chunk.choices[0].delta.content # delta.content = 这一块新增的文字
if delta: # 有些块是空的(比如开头的元信息),跳过
print(delta, end="", flush=True) # end="" 不换行,flush 立刻刷到屏幕
print() # 全部结束后补一个换行
👶
end="" flush=True 是干嘛的?print 默认每次换行、且会攒着一起显示。流式要"贴着上一块往后接、且立刻显示",所以 end="" 取消换行、flush=True 强制马上刷到屏幕——这样才有一个字一个字冒出来的效果。L04
逐块拼接:把小碟攒成一整餐
🤔 痛点流式一块块蹦出来,看是看到了,可我程序里想拿到"完整的那段回答"(存进历史、写进数据库),总不能只在屏幕上飘过就没了吧?
💡 本质一边显示、一边用一个字符串把每块 delta 累加起来,流结束时这个字符串就是完整回答。像你把陆续端来的小碟,攒成桌上完整的一餐。
full = "" # 准备一个空字符串,当"攒菜的盘子"
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
print(delta, end="", flush=True) # 一边实时显示
full += delta # 一边累加,攒成完整回答
print()
# 现在 full 里就是完整答案,可以正常使用了:
history.append({"role": "assistant", "content": full}) # 存进对话历史(回忆 Day11)
print("字数:", len(full))
📝 举个例子:delta 是怎么拼起来的
模型分块送来:
"长" → "城" → "是" → "中国" → …… 每块只是"新增的一小截"。你的 full += delta 就把它们首尾相连:"长"→"长城"→"长城是"→"长城是中国"……流结束时自然拼成完整句子。流式的核心就这一步累加,不神秘。📝 举个例子:多轮对话里必须拼接
回忆 Day11:多轮对话要把 assistant 的回答塞回 history。用流式时,你屏幕上飘过的文字必须靠
full 攒下来,才能存进 history 供下一轮使用。只显示不累加,模型下一轮就"失忆"了。L05
打字机效果 & 什么时候该用流式
🤔 痛点网页版 ChatGPT 那种"打字机"效果是怎么来的?我做网页/App 前端时也想要,但流式是不是所有场景都该用?
💡 本质前端"打字机"= 后端流式把每块 delta 通过 SSE 推给浏览器,前端收到一块就往页面上追加一块文字。至于用不用流式:给人看的长文本用流式,程序自己要用的结构化结果通常不用。
| 场景 | 建议 | 原因 |
|---|---|---|
| 聊天机器人、写作助手 | ✅ 用流式 | 长回答,用户需要即时反馈 |
| 命令行工具、Web 前端展示 | ✅ 用流式 | 打字机效果,体验专业 |
| 要解析成 JSON 再给程序用(Day14) | ❌ 常不用 | 要等完整内容才能解析,边流边解析易碎 |
| Agent 内部一步步调工具的中间结果 | 看情况 | 给人看则流,纯机器流转则不必 |
👶 前端怎么接?(了解即可)后端用 FastAPI(Day09 学过)返回一个
StreamingResponse,把每块 delta 按 SSE 格式吐出去;前端用浏览器的 EventSource 或 fetch 的流式读取,收到一块就 元素.textContent += 那块。现在知道"前端追加 = 后端流式的镜像"就够,真做全栈时再深究。L06
流式的坑:中途断了 & 取消
🤔 痛点流式连接开着好几秒,万一网络中途抖了、或者用户不想看了想停——一次性调用没这些烦恼,流式怎么处理?
💡 本质两条实战守则:① 用
try/except(Day05)把整个 for 循环兜住,中途断了也能把已攒到的 full 保住而不是全丢;② 想中途停,直接 break 跳出循环即可,连接会随之关闭。full = ""
try:
for i, chunk in enumerate(stream): # enumerate 顺便数第几块
delta = chunk.choices[0].delta.content
if delta:
print(delta, end="", flush=True)
full += delta
if len(full) > 2000: # 举例:太长了主动喊停,省钱省时
print("\n[已达长度上限,停止]")
break # 跳出循环 → 流被关闭
except Exception as e: # 网络中途出错也不至于全盘皆输
print("\n[流中断]", e)
print("\n已收到内容长度:", len(full)) # 哪怕断了,full 里也留着已到的部分
👶 流式的报错更"分散"非流式调用出错,一上来就报了;流式可能先正常吐了几块、跑到一半才断。所以务必把累加放在 try 里、且随时保住
full——这在后面写 Agent 长时间运行时尤其重要(Day54 会讲失败闸门与降级)。L07
今日小结 + 动手 10 分钟
🧠 今天你应该能回答
- 流式解决了什么体验痛点?它让模型算得更快了吗?(没有,只是更早看到)
- SSE 是什么?和普通 HTTP 一问一答有啥区别?
- 开启流式的关键参数是什么?返回的东西怎么遍历?
- chunk / delta 分别是啥?怎么拼出完整回答?
- 哪些场景该用流式、哪些不该?中途想停怎么做?
✋ 动手 10 分钟:把昨天的机器人改成打字机
在 Day11 的 chat-demo 环境里(已装好 openai / python-dotenv),新建 stream_chat.py:
from openai import OpenAI
from dotenv import load_dotenv
load_dotenv()
client = OpenAI()
history = [{"role": "system", "content": "你是友好的中文助手"}]
print("流式聊天(输入 quit 退出)")
while True:
q = input("\n你:")
if q == "quit":
break
history.append({"role": "user", "content": q})
stream = client.chat.completions.create(
model="gpt-4o-mini", messages=history, stream=True) # 打开流式
print("AI:", end="", flush=True)
full = ""
for chunk in stream: # 一块块接
delta = chunk.choices[0].delta.content
if delta:
print(delta, end="", flush=True) # 实时打字机显示
full += delta # 同时累加成完整回答
history.append({"role": "assistant", "content": full}) # 存历史,供下轮
跑起来对比昨天:同样的问题,今天文字是"冒"出来的,体验立刻不一样。
明日预告 · Day 13:会调、会流式了,接下来是每个真实项目都绕不开的现实问题——钱和速度。明天讲
temperature/top_p/max_tokens 这些参数各调什么、输入输出 token 怎么计费、一次对话到底花几分钱,以及一堆立竿见影的省钱技巧。学完你就能对老板/自己说清"这个功能一个月大概要花多少钱"。