企业官网建设流程全解析

OpenAI 流式响应实战：Python 里把等待压到首字符

【免费下载链接】openai-pythonThe official Python library for the OpenAI API项目地址: https://gitcode.com/GitHub_Trending/op/openai-python

让模型写一段 500 字的分析，用户盯着转圈的光标 8 秒——这时你分不清是连接挂了还是模型在思考。解法很直接：在 openai-python 的客户端上开启stream=True（OpenAI与AsyncOpenAI均支持），请求发出后首字符通常 1~2 秒内就能到达，而不是等整段生成完毕才一次性返回。下面从最小实现讲到可以直接上生产的写法。

机制速览：SSE 分块传输与 Stream 封装

传了stream=True之后，服务端不再把整个 JSON 攒完再返回，而是改用 SSE（Server-Sent Events）格式：响应体变成一连串data: {...}\n\n的行，每行携带一段增量结果，最后一行是data: [DONE]作为结束信号。openai-python 里的Stream/AsyncStream（实现源码）把这条字节流封装成可迭代对象：按 SSE 事件切分、把每行 JSON 解析成ChatCompletionChunk抛给你，迭代器内部用finally保证连接被释放。你只面对对象，分帧细节不用关心。⚡

同步流式调用：for 循环逐块打印

先写同步版：create(stream=True)拿到流，for 循环里逐块取增量，边打印边累积。

from openai import OpenAI client = OpenAI() stream = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "用三句话介绍你自己"}], stream=True, ) full = "" for chunk in stream: delta = chunk.choices[0].delta.content if delta: full += delta print(delta, end="", flush=True) print()

想验证首字到达时间，可以用first = next(stream)在循环外先取一块，从请求发出到拿到它的时间差就是你的 TTFT（time 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( model="gpt-4o-mini", messages=[{"role": "user", "content": "用三句话介绍你自己"}], stream=True, ) async for chunk in stream: if chunk.choices and chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="", flush=True) 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(timeout=30.0)全局设置，或在create(..., timeout=30.0)按请求覆盖；触发后抛APITimeoutError，同样能被APIError接住。

from openai import OpenAI, APIError client = OpenAI(timeout=30.0) stream = None try: stream = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "写一首短诗"}], stream=True, max_tokens=200, ) for i, chunk in enumerate(stream): if chunk.choices and chunk.choices[0].delta.content: print(chunk.choices[0].delta.content, end="", flush=True) 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