OpenAI 流式响应实战用 streamTrue 和 delta 解析让答案实时输出【免费下载链接】openai-pythonThe official Python library for the OpenAI API项目地址: https://gitcode.com/GitHub_Trending/op/openai-python聊天界面里用户盯着空白屏 10 秒才看到第一个字——这就是非流式调用的体验。换成 openai-python只要多传一个streamTrueOpenAI 流式响应就会把答案按 SSE 分块送达边收边用。读完你能让 Chat Completions 实时输出答案还会攒出完整文本、随时掐断流。⚡ 流式响应为什么快SSE 分块传输原理非流式请求的逻辑是模型把整篇回答生成完才打包成一个大 JSON 交给你。流式请求的逻辑反过来服务端一边生成一边切块通过 SSEServer-Sent Events长连接把小块持续推给客户端先收先处理。openai-python 把这个细节包住了。源码里的Stream/AsyncStream类见src/openai/_streaming.py负责逐行读取 SSE 事件、还原成 Python 对象你拿到的只是一个可迭代的stream对象不用自己拼 HTTP 长连接。对比维度非流式streamFalse流式streamTrue首个可见结果耗时等整篇生成完秒级起步首个分块到达即显示数据形态一次性完整 JSON一串 ChatCompletionChunk 小对象内存与等待感全量驻留用户干等边处理边释放用户看着逐字出现适合场景短回答、批处理脚本实时对话、长文生成、打字机展示接入复杂度其实没差别——同一套create调用只是多一个布尔参数和一层 for 循环。跑通第一个流式请求streamTrue 同步与异步API key 放在环境变量OPENAI_API_KEY里即可不必写进代码。# 同步版拿到 stream 后直接 for 遍历 from openai import OpenAI client OpenAI() # key 读自环境变量 OPENAI_API_KEY stream client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 一句话解释 SSE}], streamTrue, ) for chunk in stream: piece chunk.choices[0].delta.content # 取本块新增内容 if piece: print(piece, end, flushTrue)异步版把for换成async forcreate前面多一个await其余逻辑一致# 异步版最短可用形态 import asyncio from openai import AsyncOpenAI client AsyncOpenAI() async def main(): stream await client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 你好}], streamTrue, ) async for chunk in stream: print(chunk.choices[0].delta.content or , end) asyncio.run(main())注意asyncio.run(main())必须放在函数外——这是异步流式调用里最常见的报错点。拆开一个 chunkChatCompletionChunk 与 delta 字段流里吐出的每一块都是ChatCompletionChunk对象顶层有id、model、created这些元信息核心在choices列表。每个 choice 里有三个关键字段——index、delta、finish_reason字段定义在src/openai/types/chat/chat_completion_chunk.py。delta才是增量本身内部字段role身份标记通常只在第一个chunk 出现如assistantcontent本块新增的文本可能为 Nonetool_calls模型发起工具调用时参数会分块出现在这里refusal被拒答时的说明文本。为什么必须对content判空因为很多 chunk 根本不带正文首块只交代 role收尾块的 delta 几乎为空、只带finish_reasonstop、length、tool_calls等。遇到这些块直接取content会拿到None拼接和打印都会报错或混入 None 字样。所以上面代码里才有一行if piece:——它不是可有可无的防御而是流式解析的标配。流式解析的三种落地姿势姿势一攒出完整答案界面要展示分块后端逻辑却往往需要完整文本存库或再加工。把打印换成追加就行stream 同上不再重复贴pieces [] for chunk in stream: piece chunk.choices[0].delta.content if piece: pieces.append(piece) answer .join(pieces) # 完整的模型回答姿势二打字机效果首段同步示例里的print(piece, end, flushTrue)已经在做这件事了。关键是flushTrue不加它输出会先进缓冲区攒一批逐字蹦出就变成一顿一顿观感大打折扣。终端演示、Web 控制台打字机靠的就是这一行。姿势三兜底与止损流式过程中网络抖动、额度用尽都可能抛错有时你也只想看前几块就提前离场。异常用APIError接住中断用stream.close()from openai import APIError try: for i, chunk in enumerate(stream): print(chunk.choices[0].delta.content or , end) if i 3: stream.close() # 拿到 3 块即掐断 break except APIError as e: print(f请求失败{e})close()会关掉底层连接避免服务端继续白推数据。✅ 上线前自检清单delta 判空与事件循环自检项现象原因处理空 deltacontent是 None首块只带 role、尾块只带 finish_reason用前判空或or 事件循环no running event loop报错Jupyter 里直接跑协程包一层asyncio.run中文乱码输出方块或生僻符号终端编码非 UTF-8终端设为 UTF-8连接不收尾提前 break 后仍占连接只 break 没关流记得stream.close()收尾一行streamTrue把等全文变成实时输出判空、攒文本、关流就是流式解析的全部骨架。下一步不妨给create传上 tools看看tool_calls在 delta 里是怎么一块块拼出来的。【免费下载链接】openai-pythonThe official Python library for the OpenAI API项目地址: https://gitcode.com/GitHub_Trending/op/openai-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考