OpenAI 流式响应实战:回复像打字机实时蹦出来
【免费下载链接】openai-pythonThe official Python library for the OpenAI API项目地址: https://gitcode.com/GitHub_Trending/op/openai-python
问一句话,页面转圈半分钟,答案才憋出第一行——这种 AI 交互体验很难让人有耐心。OpenAI 流式响应处理解决的就是这个问题:模型把回答切成碎片陆续发出,页面边收边显示,用户看着答案一个字一个字长出来。下面基于官方 openai-python 库的 Chat Completions 流式调用,讲清楚这个效果怎么做出来。
流式和非流式到底差在哪
先打个比方。非流式像烧水壶,水烧开了才整壶倒进杯子,你得干等;流式像开水龙头,细水流持续往外走,杯子没满也能先接上来用。
两种模式的区别,核心就一个字:首字什么时候到。
| 观察点 | 非流式 | 流式 |
|---|---|---|
| 第一个字什么时候到 | 整段回答生成完才拿到 | 请求发出后几百毫秒 |
| 用户体感 | 干等转圈,毫无反馈 | 打字机效果,逐字出现 |
| 数据到达方式 | 一个完整 JSON | 连续的小碎片(SSE) |
| 适合的场景 | 短问答、批量跑任务 | 长文生成、实时对话 |
流式在传输层用的是 Server-Sent Events,服务端把回答切成一个个事件往外发。这部分细节 openai-python 已经封装成Stream对象,不用我们去碰 SSE 的底层。
最小可运行的流式调用:stream=True 之后发生了什么
这段代码要解决一件事:让模型的回答在终端里实时蹦出来。核心只有一个stream参数。
from openai import OpenAI client = OpenAI() stream = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "用一句话解释什么是TCP"}], stream=True, ) for chunk in stream: content = chunk.choices[0].delta.content if content: print(content, end="", flush=True)三行关键代码记牢:stream=True告诉 API「请分片给我」;chunk.choices[0].delta.content是每片新文字待的地方;if content判空,因为不是每片都有内容。
异步流式调用一句话带过:把客户端换成AsyncOpenAI,create前加await,for换成async for,其余照抄。
把碎片拼成完整回复:delta 内容的取值路径
每次迭代拿到的chunk是一个ChatCompletionChunk,取值路径固定:chunk→choices[0](取第一个分支)→delta(本片的增量)→content(文字)。流式解析的核心动作,就是把这些增量累加起来:
full = "" for chunk in stream: delta = chunk.choices[0].delta full += delta.content or "" print(full)补一个细节:delta.role通常只在第一个 chunk 里出现,值是assistant,之后就没有了。想记录「发言者」时,判断首片即可。
两个最容易踩的坑
坑一:delta.content 可能是 None
现象:直接print(chunk.choices[0].delta.content),要么打印出一堆None,要么后续拼接报错——只有元信息的片(首片、收尾片)content 就是空的。解法:一行判空,跳过它。
content = chunk.choices[0].delta.content if content: print(content, end="", flush=True)坑二:想提前停,或流中断了
现象:用户只想扫一眼,或者页面已经关了,for循环还在傻等剩余数据,白白占着时间和带宽。解法:把迭代器存成变量,想停就主动调stream.close()断开连接。
for i, chunk in enumerate(stream): print(chunk.choices[0].delta.content or "", end="", flush=True) if i >= 5: stream.close() break拿到五片就立刻断开。如果是连接自己断了,异常会从迭代那一行抛出,外层套try接住就行。
SSE 解析、[DONE]结束信号这些实现的细节,都在 src/openai/_streaming.py 里,想深挖可以直接读。手痒想跑起来,examples/streaming.py 是官方示例;版本间的变动可以看 CHANGELOG.md。
【免费下载链接】openai-pythonThe official Python library for the OpenAI API项目地址: https://gitcode.com/GitHub_Trending/op/openai-python
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考