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)。非流式是"等它全部生成完再一次给你";流式是"生成一点就送你一点"。就像餐厅:与其让你空等一桌菜,不如做好一道先端一道。
非流式 vs 流式:用户的等待体验 ❌ 非流式(一次性) 0-5 秒:空白屏,干等 第 5 秒:整段轰出 "是不是卡死了?" ✅ 流式(边生成边送) 第 0.3 秒:文字开始蹦 持续:一个字一个字来 "它在认真答,我等得住"
图注:总时长其实差不多,但流式让用户"立刻看到反应",感知快得多——体验是主观的。
👶 关键点流式并没有让模型算得更快,改变的是你什么时候开始看到结果。首字出现时间(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 格式吐出去;前端用浏览器的 EventSourcefetch 的流式读取,收到一块就 元素.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 怎么计费、一次对话到底花几分钱,以及一堆立竿见影的省钱技巧。学完你就能对老板/自己说清"这个功能一个月大概要花多少钱"。
← Day 11 · 调用大模型 API Day 13 · 参数 / 成本 / 延迟 →