OpenAI SDK 流式输出实战

让 AI 应用"显得快"的秘密，多半不在模型生成得多快，而在第一个字多早出现。同样一段长回答：一种交互是进度圈转上十几秒、文字整段砸出来；另一种是不到一秒第一行就冒头，随后文字匀速流出。两者总耗时几乎一样，体感却天差地别——差别只在一件事：流式输出。

本篇是《OpenAI Python SDK 从入门到实战》系列第 3 篇：第 1 篇装好环境跑通首次调用，第 2 篇理清消息结构与模型选型，本篇把 Responses API 的流式用法一次讲透。文中事件类型清单核实自官方 developers.openai.com 文档（2026 年 10 月）。全文按"为什么流式 → 最小示例 → 事件导航 → 收尾统计 → 同步异步 → 打字机实战 → 总结"组织。

1、为什么要流式输出

默认情况下，Responses API 是"一口气"交付：官方流式指南 的描述是，服务端会先生成模型的完整输出，再装进单个 HTTP 响应一次性返回。把一次调用的等待拆开看，由三段构成：网络往返与排队、首 token 延迟（TTFT，Time To First Token）、剩余 token 的生成时间。非流式必须三段全部走完才见第一个字，流式只需前两段。两种模式的对比如下：

SSE 指 Server-Sent Events（服务器推送事件），是 HTTP 长连接上由服务端持续推送数据的协议，stream=True 之后底层走的就是它。流式不改变生成速度，改的是交付节奏，因此它优化的是感知延迟；输出越长，改善越明显。它还有一个工程红利：可以中途放弃——消费是逐事件进行的，看到模型跑偏直接 break 跳出循环，省下后续 token 的等待时间与费用，非流式则只能陪它写完再整体丢弃。

流式也不是万能答案，它生效的前提是"有人或程序在消费中间态"：聊天界面有人盯着，打字机有终端看着。而后台批处理、定时摘要、批量抽取这类任务，拿到完整结果才能继续下一步，中间态无人问津，流式就纯属多余，还会把错误处理搞复杂——收尾信息被拆散进事件流里，不如直接判断一个完整响应对象来得省事。

2、最小流式示例

在第 1 篇的首次调用上加一个参数，就得到流式版本：responses.create 传入 stream=True，返回值从完整响应变成可迭代的事件流，第 1 篇装好的 openai 3.x 无需任何额外依赖。完整示例如下：

from openai import OpenAI

client = OpenAI()

stream = client.responses.create(
    model="gpt-6.1-terra",
    input="用两句话解释什么是流式输出。",
    stream=True,               # 唯一的新参数
)

for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)  # 增量即时上屏
print()                       # 收尾补一个换行

运行示例能看到文字分批跳出，每批几个字到十几个字不等，这就是 delta 的粒度，随网络与模型负载浮动。

解读：stream=True 让 create 不再阻塞到生成结束，而是立刻返回流对象；for 循环每次取出一个事件，event.type 是类型标签；response.output_text.delta 是文本增量的主通道，event.delta 只含本次新增的一小段文字，按到达顺序拼接才是完整回复。注意与非流式的差别：第 1 篇的 response.output_text 便捷属性在流式下帮不上忙——SDK 无法预知未来的增量，拼接必须自己做。这个写法与 官方 SDK 仓库 的流式示例一致，可以直接对照阅读。

3、事件类型导航

Responses API 的流式事件是语义化的：每个事件都有预定义类型与字段结构，官方原话是这条 API "designed with streaming in mind"（为流式而设计）。事件总量有几十种，纯文本场景只需认清一条主线，核心事件如下：

一次纯文本响应的事件顺序是一条链：response.created 打头，response.in_progress 跟进，随后 output_item.added、content_part.added 逐层打开容器，中间夹着若干次 output_text.delta，容器再逐层关闭，最后以 response.completed 收束。可以把它想象成四层嵌套：响应包着条目、条目包着部件、部件里流出增量——工具调用与多模态输出沿用同一套骨架，只是条目与增量的种类更多。每个事件都带 sequence_number 用于保序，常规场景按到达顺序消费即可。

顺带认识"条目"这个概念：纯文本响应里新增的条目是一条 role 为 assistant 的 message；除此之外，output 数组还可能装着推理条目、函数调用条目、网页搜索条目，收尾事件里能拿到它们的完整形态，系列的工具调用篇会正式展开。其余事件按能力族扩展，本系列后续篇章会陆续遇到：

• 网页搜索：response.web_search_call.* 的进行、搜索、完成三态
• 函数调用：response.function_call_arguments.delta 流式拼参数
• 推理摘要：response.reasoning_summary_text.delta 一族

完整清单见流式事件参考，用到对应能力时回来查表即可，不必提前背全。

4、收尾信息与用量统计

增量只负责把字打出来，有账可查的信息全在收尾事件里：response.completed 的 response 字段就是完整响应对象，与非流式返回同构，status、usage、output 都从这里取。给第 2 节的循环加一个分支即可：

final = None
for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
    elif event.type == "response.completed":
        final = event.response      # 接住终局事件

usage = final.usage
print(f"\n消耗：输入 {usage.input_tokens}"
      f" + 输出 {usage.output_tokens} tokens")

解读：usage.input_tokens 与 usage.output_tokens 就是第 1 篇提过的计费口径；流式不影响统计，费用按实际生成的 token 计算，与非流式完全一致。截断同样有账可查：撞上 max_output_tokens 上限时，收尾事件换成 response.incomplete，判断口径与本系列第 2 篇一致——response.status 为 incomplete，incomplete_details.reason 为 max_output_tokens。

注意：别拿"for 循环跑完"当成功标志。流中途出错时循环同样会结束，只是收尾事件变成 response.failed 或 error；生产代码至少要给正常、截断、失败三类收尾各留一个分支。

顺带一句 Chat Completions 对照：它的流式文本在 chunk 的 choices[0].delta.content，结束信息在最后一个 chunk 的 finish_reason 里，没有语义化的事件分层——这也是第 1 篇选型时偏向 Responses 的原因之一。

5、同步与异步消费姿势

事件流的消费有两种写法，区别只在客户端用 OpenAI 还是 AsyncOpenAI。先看全貌，一次流式调用的时间线如下：

从时序图看三方分工：SDK 是转接头，把 SSE 字节流解析成带类型的事件对象逐个递给你；服务端只管生成与推送；你的脚本每收到一个增量就能立即处理，不必等收尾。同步与异步的差异只发生在"脚本与 SDK"这一段。同步写法就是第 2 节的 for 循环；异步写法两个要点——调用要 await、迭代要 async for：

import asyncio
from openai import AsyncOpenAI

client = AsyncOpenAI()

async def main():
    stream = await client.responses.create(  # 先 await 拿到流
        model="gpt-6.1-terra",
        input="用两句话解释什么是流式输出。",
        stream=True,
    )
    async for event in stream:               # 再异步迭代
        if event.type == "response.output_text.delta":
            print(event.delta, end="", flush=True)

asyncio.run(main())

两种写法的选型对照如下：

为什么 Web 服务偏爱异步？一次流式响应动辄持续几秒到几十秒，同步调用会把整个线程占满；async 写法挂在事件循环上，等待期间让出执行权，同一个进程能同时喂几十条流，这也是 FastAPI 网关做实时转发的标配姿势。

最常见的翻车是漏写 await：拿到的是协程对象而不是流，迭代时直接报类型错误；同步客户端配 async for（或反过来）同样必挂。两套客户端两套协议，不要混用。判断标准很简单：代码跑在 FastAPI 这类异步框架里就用异步；独立脚本与终端工具，老实用同步。

6、终端打字机实战

把前面的零件组装起来：事件循环取增量、逐字上屏、收尾报账，一个完整的终端打字机如下：

import time

from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()
client = OpenAI()

def typewriter(text, interval=0.03):
    for ch in text:                     # 逐字输出
        print(ch, end="", flush=True)   # 关键：立即刷出
        time.sleep(interval)            # 打字节奏

stream = client.responses.create(
    model="gpt-6.1-terra",
    instructions="你是惜字如金的终端助手，回答不超过两句。",
    input="用一句话解释什么是流式输出。",
    stream=True,
)

for event in stream:
    if event.type == "response.output_text.delta":
        typewriter(event.delta)         # 增量逐字打出
    elif event.type == "response.completed":
        u = event.response.usage
        print(f"\n[tokens] 输入 {u.input_tokens} / 输出 {u.output_tokens}")

解读分三点。其一，flush=True 是整个效果的灵魂：终端输出默认有缓冲，不带 flush 的 print 会把文字攒到缓冲区满或程序结束才一次性出现，打字机瞬间退化成"卡半天、蹦一行"。其二，interval 控制节奏，0.03 秒约合每分钟两千字的敲击手感；delta 到达速度通常快于这个节奏，typewriter 相当于一个匀速节流阀，网络抖动被自然抹平。其三，想忠实呈现网络速度，把 typewriter(event.delta) 换成 print(event.delta, end="", flush=True) 即可——那就是第 2 节的最小示例，快与慢之间只差一个节流阀。

7、总结

流式输出不改变模型写一个字的时间，改变的是第一个字何时到达：stream=True 一个参数，等待从全量时长缩短到首 token 延迟。Responses API 的流式是语义化事件流：response.output_text.delta 拼文本，response.completed 接收尾，截断与失败各有专属事件，usage 口径与非流式一致。消费姿势跟着宿主走：脚本用同步 for，异步框架用 AsyncOpenAI 加 async for。对于一切"人盯着屏幕等输出"的场景，建议默认上流式——代价只是一个参数加一层循环，换来的是体感上的巨大提升。

留一道练习：给第 6 节的打字机加耗时统计——用 time.perf_counter() 记录请求发出、第一个 delta 到达、收尾事件到达三个时刻，打印首 token 延迟与总耗时；再把同一个问题用非流式跑一遍对比。三组数字一摆，"流式优化的是感知延迟"就从结论变成你亲手测过的事实。想加难度，可以把逐字节奏升级成"遇标点多停半秒"，模拟真人敲键盘的呼吸感。

下一篇起进入核心能力阶段，解决"让模型稳定返回可解析 JSON"的结构化输出——它既是流水线上最实用的能力，也是工具调用的地基。