文章目录
-
- 前言
- 1. 先唠透原理:模型本来就是一个字一个字挤的
-
- 1.1 自回归到底是啥
- 1.2 流式和非流式,体感差一个次元
- 2. 上代码:手把手教你接流式
-
- 2.1 逐行扒细节,坑都给你标出来
- 3. 怎么知道话讲完了?
-
- 3.1 两种结束信号,都得接住
- 4. 新手高频踩坑:字典形式的chunk
- 5. 流式和非流式,到底该用哪个?
- 6. 聊个和钱相关的冷知识
- 7. 动手做三个小实验
- 8. 最后收个尾

P.S. 无意间发现了一个巨牛的人工智能教程,非常通俗易懂,对AI感兴趣的朋友强烈推荐去看看,
传送门https://blog.csdn.net/qq_34419312
前言
咱之前聊大模型调用,全是“憋大招”模式——等半天,答案啪一下全出来。
但你平常用ChatGPT的时候,字都是一个一个往外跳的吧?这就是流式输出。别觉得这是啥黑科技,今天咱从根上给你讲明白,顺带把代码给你写得明明白白。
1. 先唠透原理:模型本来就是一个字一个字挤的
1.1 自回归到底是啥
大模型生成文本的方式,学名叫做自回归。说人话就是:给上文,猜下一个字。
举个例子:输入“今天天气真”,第一步猜下一个是“不”,句子变成“今天天气真不”;第二步再猜下一个是“错”,变成“今天天气真不错”;第三步补个句号。就这么一步步往下推。
模型一次只能产出一个token,每个字都得靠前面所有内容当上下文。所谓“模型想好了再一次性输出”根本就是错觉——服务器本来就是边生成边能往外发的。
1.2 流式和非流式,体感差一个次元
非流式是什么体验?服务器攒完整篇回答,打包成一个大包发回来。你就盯着空白屏幕干等20秒,跟在楼下等外卖还看不到骑手位置似的,越等越焦虑。
流式就贴心多了:每生成一个token立刻发过来。1秒内就能看到第一个字,剩下的慢慢蹦,跟有人在对面打字似的。
行业里有个关键指标叫TTFT,就是首token时间。流式不改变总生成时长,但把“20秒后看全文”变成“1秒后开始逐字看”,体感差距堪比从绿皮硬座换成高铁商务座。
2. 上代码:手把手教你接流式
代码其实超简单,和普通调用比,就多了一个参数。
from openai import OpenAI
client = OpenAI(api_key="sk-xxx", base_url="https://api.deepseek.com")
stream = client.chat.completions.create(
model="deepseek-chat",
messages=[{"role": "user", "content": "写一首四行诗,主题是debug"}],
stream=True, # 唯一的改动:打开流式开关
)
full_text = "" # 累积完整回答
for chunk in stream: # chunk = 一小片段,不是一个完整响应
delta = chunk.choices[0].delta # delta = "这次新增的内容"
if delta.content: # 有的chunk只有元信息,content为空,要判空
full_text += delta.content # 自己负责拼接
print(delta.content, end="", flush=True) # 立刻打印,不换行
print("\\n— 完整回答 —")
print(full_text)
2.1 逐行扒细节,坑都给你标出来
for chunk in stream:普通调用返回一个完整的响应对象,流式返回的是个迭代器,每次循环吐一个小片段。
chunk.choices[0].delta.content:这里必须敲黑板,是delta不是message。delta里只有增量内容,第一个chunk可能是“写”,第二个是“一”,第三个是“首”,拼起来才是完整的一句话。
if delta.content:流里面混着一些事件通知类的chunk,比如收尾的标识,它们的content是空的。不判空直接报错,新手十个人里有九个栽在这。
flush=True:让print立刻刷新到屏幕。默认会有缓冲区,容易卡一下蹦一串,那打字机的氛围感直接就没了。
full_text += …:流式只管给你传碎片,完整内容得自己动手拼。忘了拼的话,最后手里只剩一堆碎渣,啥用没有。
3. 怎么知道话讲完了?
总不能让循环一直跑下去吧,总得有个结束信号。
for chunk in stream:
if chunk.choices[0].finish_reason == "stop":
print("\\n[生成完毕]")
break
...
3.1 两种结束信号,都得接住
第一种:最后一个chunk的finish_reason变成"stop",这是正常结束;或者变成"length",说明被max_tokens给截断了。
第二种:迭代器自己跑完了,for循环自然退出。
工程上稳妥的写法是两者都处理:循环里检查finish_reason做收尾动作,比如通知前端生成完了;循环外面做资源清理。双保险,怎么都不会错。
4. 新手高频踩坑:字典形式的chunk
有些场景下,比如用FastAPI做转发,你会把chunk转成字典再发给前端。这时候取值方式就变了,很多人在这里翻车。
d = chunk.model_dump() # 转成普通字典
content = d["choices"][0]["delta"]["content"] # 用[]取键,不是.属性
对象用点访问,字典用中括号,这俩千万别搞混。一看到AttributeError: ‘dict’ object has no attribute ‘choices’,别想别的,九成九是这问题。就像你拿家门钥匙去开电动车锁,硬插肯定插不进去啊。
5. 流式和非流式,到底该用哪个?
不是说流式就一定高级,得分场景用。
| 程序内部调用:要完整结果做下一步处理 | 直接给用户看:聊天界面 |
| 简单,一行就能取结果 | 要处理chunk拼接、判空、结束信号 |
| 全程等待,最后一次性出结果 | 打字机效果,边生成边看 |
这里说个非常重要的经验:Agent内部的LLM调用,大多用非流式。因为Agent循环需要完整的回答来解析工具调用,总不能边蹦字边判断该调啥工具吧。只有面向用户的最后一公里,才用流式。后面咱聊前后端对接的时候,就会把这俩拼起来用。
6. 聊个和钱相关的冷知识
流式生成的过程中,服务器已经算出来的KV缓存,天然就是下一轮对话的前缀。
所以多轮对话加流式,是天生的缓存友好组合。现在听不懂没关系,等讲到KV Cache的时候你再回头看,绝对会拍大腿喊“原来如此”。省下来的,可都是真金白银的算力钱。
7. 动手做三个小实验
光说不练假把式,自己跑一遍,印象才深刻。
实验一:跑通上面的基础代码,亲眼看着文字逐字输出。
实验二:给每个chunk加上时间戳,看看它们的到达节奏。
import time
for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
print(f"[{time.time()%100:.2f}] {delta.content}")
你会发现chunk不是均匀到达的。模型遇到难抉择的地方,概率分布比较平坦,就会明显卡顿一下。这可是观察模型“思考过程”的绝佳小窗口。
实验三:故意把max_tokens设得很小,看看finish_reason什么时候变成"length"。
8. 最后收个尾
P.S. 无意间发现了一个巨牛的人工智能教程,非常通俗易懂,对AI感兴趣的朋友强烈推荐去看看,传送门https://blog.csdn.net/qq_34419312
网硕互联帮助中心





评论前必须登录!
注册