尧图网站建设 
OpenAI 流式响应实战Python 里把等待压到首字符【免费下载链接】openai-pythonThe official Python library for the OpenAI API项目地址: https://gitcode.com/GitHub_Trending/op/openai-python让模型写一段 500 字的分析用户盯着转圈的光标 8 秒——这时你分不清是连接挂了还是模型在思考。解法很直接在 openai-python 的客户端上开启streamTrueOpenAI与AsyncOpenAI均支持请求发出后首字符通常 1~2 秒内就能到达而不是等整段生成完毕才一次性返回。下面从最小实现讲到可以直接上生产的写法。机制速览SSE 分块传输与 Stream 封装传了streamTrue之后服务端不再把整个 JSON 攒完再返回而是改用 SSEServer-Sent Events格式响应体变成一连串data: {...}\n\n的行每行携带一段增量结果最后一行是data: [DONE]作为结束信号。openai-python 里的Stream/AsyncStream实现源码把这条字节流封装成可迭代对象按 SSE 事件切分、把每行 JSON 解析成ChatCompletionChunk抛给你迭代器内部用finally保证连接被释放。你只面对对象分帧细节不用关心。⚡同步流式调用for 循环逐块打印先写同步版create(streamTrue)拿到流for 循环里逐块取增量边打印边累积。from openai import OpenAI client OpenAI() stream client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 用三句话介绍你自己}], streamTrue, ) full for chunk in stream: delta chunk.choices[0].delta.content if delta: full delta print(delta, end, flushTrue) print()想验证首字到达时间可以用first next(stream)在循环外先取一块从请求发出到拿到它的时间差就是你的 TTFTtime to first token。examples/streaming.py 里的next(response)演示的就是这个手动取块用法。异步流式调用AsyncOpenAI 只差两个 await异步版逻辑与上面完全相同差异只有两处create要await循环换成async for——因为AsyncOpenAI底层是异步 HTTP 客户端阻塞点被让给了事件循环。import asyncio from openai import AsyncOpenAI client AsyncOpenAI() async def main() - None: stream await client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 用三句话介绍你自己}], streamTrue, ) async for chunk in stream: if chunk.choices and chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end, flushTrue) asyncio.run(main())chunk → choices → delta 三层嵌套与 content 为空的三种情况拿到的每个对象是一个ChatCompletionChunk类型定义三层嵌套chunk→chunk.choices列表→choice.delta本块增量。delta.content只是本轮新增的文本把所有块拼起来才是完整回答。第一个块长这样{ id: chatcmpl-9abc, object: chat.completion.chunk, choices: [ { index: 0, delta: { role: assistant, content: null }, finish_reason: null } ] }content为空或为None不是异常常见有三种情况首块delta 只携带role: assistant宣告身份正文还没开始工具调用块模型在拼参数时content是None增量在delta.tool_calls里用量统计块传stream_options{include_usage: True}时最后一个块的choices是空数组里面只有usage。第三种最危险——直接chunk.choices[0]会抛IndexError所以下文生产代码的防御写法是if chunk.choices and ...。生产三件事错误捕获、中途 close() 与超时兜底错误捕获库里所有 API 异常都继承自APIError包括APIConnectionError、APITimeoutError、RateLimitError异常层级。注意流式请求的错误可能发生在连接建立阶段也可能发生在迭代中途——某个块携带error字段时Stream会直接抛APIError所以 try 必须包住 for 循环而不是只包住create。中途取消用户点了停止就break然后调用stream.close()释放连接。其实迭代器结束含break跳出时finally也会兜底关闭连接但显式调用close()语义更清楚异步版对应await stream.aclose()。超时兜底OpenAI(timeout30.0)全局设置或在create(..., timeout30.0)按请求覆盖触发后抛APITimeoutError同样能被APIError接住。from openai import OpenAI, APIError client OpenAI(timeout30.0) stream None try: stream client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 写一首短诗}], streamTrue, max_tokens200, ) for i, chunk in enumerate(stream): if chunk.choices and chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end, flushTrue) if i 20: # 用户打断或预算到顶 break except APIError as err: print(f流式请求异常: {err.message}) finally: if stream is not None: stream.close()到这里从首字符到异常路径的最小闭环就齐了流式 边收边处理抓住三条——按 chunk 逐块解析、跳过空 delta、异常时保证连接关闭。往下延伸最自然的场景是工具调用把累积delta.content换成按index累积delta.tool_calls的function.arguments增量拼出完整 JSON 再执行本地函数流式框架本身一行不用改。【免费下载链接】openai-pythonThe official Python library for the OpenAI API项目地址: https://gitcode.com/GitHub_Trending/op/openai-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 OpenAI 流式响应实战：Python 里把等待压到首字符
OpenAI 流式响应实战Python 里把等待压到首字符【免费下载链接】openai-pythonThe official Python library for the OpenAI API项目地址: https://gitcode.com/GitHub_Trending/op/openai-python让模型写一段 500 字的分析用户盯着转圈的光标 8 秒——这时你分不清是连接挂了还是模型在思考。解法很直接在 openai-python 的客户端上开启streamTrueOpenAI与AsyncOpenAI均支持请求发出后首字符通常 1~2 秒内就能到达而不是等整段生成完毕才一次性返回。下面从最小实现讲到可以直接上生产的写法。机制速览SSE 分块传输与 Stream 封装传了streamTrue之后服务端不再把整个 JSON 攒完再返回而是改用 SSEServer-Sent Events格式响应体变成一连串data: {...}\n\n的行每行携带一段增量结果最后一行是data: [DONE]作为结束信号。openai-python 里的Stream/AsyncStream实现源码把这条字节流封装成可迭代对象按 SSE 事件切分、把每行 JSON 解析成ChatCompletionChunk抛给你迭代器内部用finally保证连接被释放。你只面对对象分帧细节不用关心。⚡同步流式调用for 循环逐块打印先写同步版create(streamTrue)拿到流for 循环里逐块取增量边打印边累积。from openai import OpenAI client OpenAI() stream client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 用三句话介绍你自己}], streamTrue, ) full for chunk in stream: delta chunk.choices[0].delta.content if delta: full delta print(delta, end, flushTrue) print()想验证首字到达时间可以用first next(stream)在循环外先取一块从请求发出到拿到它的时间差就是你的 TTFTtime to first token。examples/streaming.py 里的next(response)演示的就是这个手动取块用法。异步流式调用AsyncOpenAI 只差两个 await异步版逻辑与上面完全相同差异只有两处create要await循环换成async for——因为AsyncOpenAI底层是异步 HTTP 客户端阻塞点被让给了事件循环。import asyncio from openai import AsyncOpenAI client AsyncOpenAI() async def main() - None: stream await client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 用三句话介绍你自己}], streamTrue, ) async for chunk in stream: if chunk.choices and chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end, flushTrue) asyncio.run(main())chunk → choices → delta 三层嵌套与 content 为空的三种情况拿到的每个对象是一个ChatCompletionChunk类型定义三层嵌套chunk→chunk.choices列表→choice.delta本块增量。delta.content只是本轮新增的文本把所有块拼起来才是完整回答。第一个块长这样{ id: chatcmpl-9abc, object: chat.completion.chunk, choices: [ { index: 0, delta: { role: assistant, content: null }, finish_reason: null } ] }content为空或为None不是异常常见有三种情况首块delta 只携带role: assistant宣告身份正文还没开始工具调用块模型在拼参数时content是None增量在delta.tool_calls里用量统计块传stream_options{include_usage: True}时最后一个块的choices是空数组里面只有usage。第三种最危险——直接chunk.choices[0]会抛IndexError所以下文生产代码的防御写法是if chunk.choices and ...。生产三件事错误捕获、中途 close() 与超时兜底错误捕获库里所有 API 异常都继承自APIError包括APIConnectionError、APITimeoutError、RateLimitError异常层级。注意流式请求的错误可能发生在连接建立阶段也可能发生在迭代中途——某个块携带error字段时Stream会直接抛APIError所以 try 必须包住 for 循环而不是只包住create。中途取消用户点了停止就break然后调用stream.close()释放连接。其实迭代器结束含break跳出时finally也会兜底关闭连接但显式调用close()语义更清楚异步版对应await stream.aclose()。超时兜底OpenAI(timeout30.0)全局设置或在create(..., timeout30.0)按请求覆盖触发后抛APITimeoutError同样能被APIError接住。from openai import OpenAI, APIError client OpenAI(timeout30.0) stream None try: stream client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 写一首短诗}], streamTrue, max_tokens200, ) for i, chunk in enumerate(stream): if chunk.choices and chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end, flushTrue) if i 20: # 用户打断或预算到顶 break except APIError as err: print(f流式请求异常: {err.message}) finally: if stream is not None: stream.close()到这里从首字符到异常路径的最小闭环就齐了流式 边收边处理抓住三条——按 chunk 逐块解析、跳过空 delta、异常时保证连接关闭。往下延伸最自然的场景是工具调用把累积delta.content换成按index累积delta.tool_calls的function.arguments增量拼出完整 JSON 再执行本地函数流式框架本身一行不用改。【免费下载链接】openai-pythonThe official Python library for the OpenAI API项目地址: https://gitcode.com/GitHub_Trending/op/openai-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考