默认情况下,当你向 OpenAI API 发起请求时,我们会先生成模型的全部输出,然后将其作为单一 HTTP 响应返回。在生成长文本时,等待完整响应可能会消耗较长时间。流式响应允许你在模型继续生成完整内容的同时,实时开始打印或处理输出的开头部分。
本指南重点介绍通过服务器发送事件 (SSE) 进行的 HTTP 流式传输(stream=true)。关于使用 previous_response_id 进行增量输入的持久 WebSocket 传输,请参阅 Responses API 的 WebSocket 模式。
启用流式传输
要开始流式传输响应,请在向 Responses 端点发起请求时设置 stream=True。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from openai import OpenAI
client = OpenAI()
stream = client.responses.create(
model="gpt-5.5",
input=[
{
"role": "user",
"content": "Say 'double bubble bath' ten times fast.",
},
],
stream=True,
)
for event in stream:
print(event)Responses API 使用语义化事件进行流式传输。每个事件都有预定义的模式(schema),因此你可以监听自己关心的事件。
如需完整的事件类型列表,请参阅 流式传输 API 参考。以下是一些示例:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
type StreamingEvent =
| ResponseCreatedEvent
| ResponseInProgressEvent
| ResponseFailedEvent
| ResponseCompletedEvent
| ResponseOutputItemAdded
| ResponseOutputItemDone
| ResponseContentPartAdded
| ResponseContentPartDone
| ResponseOutputTextDelta
| ResponseOutputTextAnnotationAdded
| ResponseTextDone
| ResponseRefusalDelta
| ResponseRefusalDone
| ResponseFunctionCallArgumentsDelta
| ResponseFunctionCallArgumentsDone
| ResponseFileSearchCallInProgress
| ResponseFileSearchCallSearching
| ResponseFileSearchCallCompleted
| ResponseCodeInterpreterInProgress
| ResponseCodeInterpreterCallCodeDelta
| ResponseCodeInterpreterCallCodeDone
| ResponseCodeInterpreterCallInterpreting
| ResponseCodeInterpreterCallCompleted
| Error流式传输 Chat Completions 非常直观。不过,我们建议使用 Responses API 进行流式传输,因为我们在设计时充分考虑了流式需求。Responses API 使用语义化事件进行流式传输,并且是类型安全的。
流式传输 Chat Completion
要流式传输完成内容,请在调用 Chat Completions 或旧版 Completions 端点时设置 stream=True。这将返回一个对象,以纯数据服务器发送事件的形式流式传输响应。
响应通过事件流以增量块的形式发送。你可以使用 for 循环迭代事件流,如下所示:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from openai import OpenAI
client = OpenAI()
stream = client.chat.completions.create(
model="gpt-5",
messages=[
{
"role": "user",
"content": "Say 'double bubble bath' ten times fast.",
},
],
stream=True,
)
for chunk in stream:
print(chunk)
print(chunk.choices[0].delta)
print("****************")读取响应
如果你使用我们的 SDK,每个事件都是一个类型化的实例。你还可以使用事件的 type 属性来识别单个事件。
一些关键的生命周期事件仅触发一次,而另一些则会随着响应的生成多次触发。流式传输文本时需要关注的常见事件包括:
- `response.created`
- `response.output_text.delta`
- `response.completed`
- `error`如需查看所有可监听的事件列表,请参阅 流式传输 API 参考。
当你流式传输 Chat Completion 时,响应包含 delta 字段,而不是 message 字段。delta 字段可能包含角色 token、内容 token 或为空。
{ role: 'assistant', content: '', refusal: null }
****************
{ content: 'Why' }
****************
{ content: " don't" }
****************
{ content: ' scientists' }
****************
{ content: ' trust' }
****************
{ content: ' atoms' }
****************
{ content: '?\n\n' }
****************
{ content: 'Because' }
****************
{ content: ' they' }
****************
{ content: ' make' }
****************
{ content: ' up' }
****************
{ content: ' everything' }
****************
{ content: '!' }
****************
{}
****************要仅流式传输 Chat Completion 的文本响应,你的代码应如下所示:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
from openai import OpenAI
client = OpenAI()
stream = client.chat.completions.create(
model="gpt-5",
messages=[
{
"role": "user",
"content": "Say 'double bubble bath' ten times fast.",
},
],
stream=True,
)
for chunk in stream:
if chunk.choices[0].delta.content is not None:
print(chunk.choices[0].delta.content, end="")进阶用例
对于更高级的用例(如流式传输工具调用),请查看以下专题指南:
审核风险
请注意,在生产应用程序中流式传输模型输出会增加内容审核的难度,因为部分完成的内容可能更难进行评估。这可能会对已批准的使用场景产生影响。