前面几篇的请求方式都是”发出去，等全部生成完，一次性拿到完整回复”。这在模型要写几百字时意味着等待好几秒，屏幕上却毫无动静。本篇介绍流式输出：开启 stream=True 后，模型每生成一小段内容就立刻推送给客户端，效果就像文字正在被逐字打出来。它不改变回复内容，只改变内容的到达方式。

什么是流式输出

先对比两种模式的差别。非流式请求中，服务端要把整段回复生成完毕才返回，客户端拿到的是一个完整的响应对象；而流式请求中，连接始终保持打开，服务端一边生成一边推送，客户端收到的是一个内容分片（chunk）的序列。

对比项 非流式 流式
首次拿到内容的时间 全部生成完成后 生成第一个字后
响应形式 一个完整对象 一串分片对象
体验 长时间等待、突然出全文 打字机效果、实时可见
适用场景 批量处理、后台任务 聊天界面、需要进度反馈

流式输出并不会让总耗时显著变短，它的价值在于首字延迟：用户几乎立刻看到内容开始滚动，感知上的等待时间大大缩短。OpenAI 兼容接口中，只需要在请求里加一个参数：stream=True。

最小流式示例

在之前的最小聊天程序基础上加一行参数，就能看到打字机效果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import os

from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()

client = OpenAI(
    api_key=os.getenv("OPENAI_API_KEY"),
    base_url=os.getenv("OPENAI_BASE_URL"),
)
model_name = os.getenv("MODEL_NAME", "gpt-4o-mini")

stream = client.chat.completions.create(
    model=model_name,
    messages=[{"role": "user", "content": "用三句话介绍流式输出。"}],
    stream=True,  # 开启流式
)

for chunk in stream:
    piece = chunk.choices[0].delta.content
    if piece:
        print(piece, end="", flush=True)
print()

运行后，文字会逐字逐句地在终端里”打”出来。关键点有两个：

  • stream=True 让 create() 不再返回完整响应，而是返回一个可迭代的分片流，循环里每次取到一个 chunk。
  • chunk.choices[0].delta.content 是这一片新增的文本。注意它可能是 None（比如第一个分片只携带角色信息 delta.role，最后一个分片只携带结束原因），所以要先判断再使用。
  • print(piece, end="", flush=True) 中，end="" 去掉默认换行，flush=True 强制立即输出到屏幕，否则终端会缓冲。

把分片拼成完整文本

流式循环中每个分片只是”一小块”，如果后续要把这段回复存进对话历史、写入文件或做进一步处理，就需要把它们累加成一个完整字符串：

1
2
3
4
5
6
7
8
9
full_text = ""
for chunk in stream:
    piece = chunk.choices[0].delta.content
    if piece:
        full_text += piece
        print(piece, end="", flush=True)

print()
print("完整文本长度：", len(full_text))

这个累加逻辑和前面”多轮对话”文章里保存 assistant 回复的做法一致：流式模式下模型回复不会自动出现在响应对象里，你必须自己收集。最简单可靠的写法就是上面的”边收边拼”。

获取 token 用量

非流式响应里 response.usage 直接可用，但流式模式默认不返回用量信息。如果确实需要，可以在请求中开启：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
stream = client.chat.completions.create(
    model=model_name,
    messages=[{"role": "user", "content": "用一句话解释什么是 token。"}],
    stream=True,
    stream_options={"include_usage": True},  # 请求返回用量
)

usage = None
for chunk in stream:
    if chunk.usage:            # 用量只出现在最后一个分片
        usage = chunk.usage
    piece = chunk.choices[0].delta.content
    if piece:
        print(piece, end="", flush=True)

print()
if usage:
    print(f"总 token 数：{usage.total_tokens}")

include_usage 开启后，最后一个分片会携带 usage 字段，其余分片为 None。需要注意：部分第三方兼容服务商不支持该参数或返回空值，代码里要做好判空。

什么时候用流式

  • 聊天界面、命令行交互：首选流式，用户反馈实时，体验明显更好。
  • 后台批量任务、结构化输出：非流式更简单，一个完整响应对象直接取用，无需手动拼接。
  • 两者兼顾：可以始终用流式请求，在循环里累加完整文本，再交给下游处理——这样既拿到实时反馈，又得到完整结果。

常见问题

输出全挤在一行或迟迟不显示。 多半是 print 的换行与缓冲问题。确认使用了 end="" 去掉换行、flush=True 关闭缓冲。

delta.content 为 None 导致报错或漏字。 流式分片并非每片都有内容：首片通常只有 role，末片通常只有 finish_reason。务必先判断再处理，这也正是示例里 if piece: 的原因。

流式中途报错怎么办？ 异常发生在生成过程中时，你可能已经收到一部分文本。此时没有”断点续传”，重试只能重新发起整个请求。所以流式循环应配合错误处理，把已收到的部分和异常信息一并记录下来，便于排查。

为什么开了流式还是感觉慢？ 总生成时间基本不变，流式改善的是”看到第一个字”的时间。如果连首字都很慢，通常是网络延迟或模型本身推理慢，可以检查 base_url 指向的节点是否离你较远。

第三方接口的流式格式不一样？ 大多数兼容 OpenAI 接口的服务商都遵循相同的 SSE 分片格式，字段结构一致；个别服务商可能不返回 usage 或首片不携带 role。代码按”字段可能缺失”来写，兼容性最好。

小结

  • 流式输出通过 stream=True 开启，返回分片流，让用户立即看到生成过程。
  • 每个分片通过 chunk.choices[0].delta.content 取增量文本，需判空；全文要靠循环累加。
  • 需要 token 用量时开启 stream_options={"include_usage": True}，只在最后一个分片返回。
  • 交互场景用流式提升体验，批量场景用非流式保持简单。下一步我们将学习结构化输出：让模型稳定返回 JSON，而不是自由文本。