【ChatGPT】OpenAI 如何使用流模式进行回答

当你向 OpenAI 请求完成时，默认情况下，整个回复会在一次性响应中全部生成并返回给你。如果你正在生成的回复内容较长，等待完整回复的时间可能会让人觉得有点漫长——好几秒钟呢！为了能更快地获取到部分回复，你可以选择“流式”接收这些正在生成的回复。这样做的话，你就可以在完整的回复还没准备好之前就开始展示或处理部分内容了。

要开启流式回复，只需要在调用 chat completions 或 completions 接口时设置 stream=True。这将返回一个对象，它以 data-only server-sent events 的形式逐步发送回应数据。你应该从 delta 字段而不是 message 字段提取数据块。

缺点

需要注意的是，在生产应用中使用 stream=True 会使管理回复内容变得更加复杂，因为部分回复可能更难以评估。这对批准的使用情况可能会产生影响。

示例代码

下面这个笔记本展示了：

一个典型的聊天回复看起来是什么样子
流式聊天回复又是什么模样
流式传输聊天回复能节省多少时间
如何获取流式聊天回复的令牌使用数据

# !pip install openai

# 导入必要的库
import time  # 用于测量API调用的时间
from openai import OpenAI
import os
client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY", "<你的OpenAI API密钥，如果没有设置为环境变量>"))

1. 典型的聊天回复长什么样

通常的 ChatCompletions API 调用会先计算回复，然后一次性返回所有内容。

# 这是一个OpenAI ChatCompletion请求的例子
# https://platform.openai.com/docs/guides/text-generation/chat-completions-api# 记录请求发送前的时间
start_time = time.time()# 发送一个ChatCompletion请求来数到100
response = client.chat.completions.create(model='gpt-4o-mini',messages=[{'role': 'user', 'content': '从1数到100，每个数字之间用逗号隔开，不要换行。例如：1, 2, 3, ...'}],temperature=0,
)
# 计算接收到回复所花费的时间
response_time = time.time() - start_time# 打印延迟时间和收到的文本
print(f"完整回复在请求发出后 {response_time:.2f} 秒内收到")
print(f"完整回复内容:\n{response}")

完整回复在请求发出后 1.88 秒内收到

提取的回复:

ChatCompletionMessage(content='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, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42, 43, 44, 45, 46, 47, 48, 49, 50, 51, 52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62, 63, 64, 65, 66, 67, 68, 69, 70, 71, 72, 73, 74, 75, 76, 77, 78, 79, 80, 81, 82, 83, 84, 85, 86, 87, 88, 89, 90, 91, 92, 93, 94, 95, 96, 97, 98, 99, 100', role='assistant', function_call=None, tool_calls=None)

2. 如何流式传输聊天回复

通过流式API调用，回复是通过一个事件流分块增量发送回来的。在Python中，你可以用 for 循环遍历这些事件。

让我们来看看这是什么样子的：

# 这是一个带有 stream=True 的OpenAI ChatCompletion请求例子
# https://platform.openai.com/docs/api-reference/streaming#chat/create-stream# 一个ChatCompletion请求
response = client.chat.completions.create(model='gpt-4o-mini',messages=[{'role': 'user', 'content': "1+1等于多少？用一个词回答。"}],temperature=0,stream=True  # 这次我们设置了 stream=True
)for chunk in response:print(chunk)print(chunk.choices[0].delta.content)print("****************")

如上所示，流式回复有一个 delta 字段，而不是 message 字段。delta 可以包含如下内容：

角色令牌（例如 {"role": "assistant"}）
内容令牌（例如 {"content": "\n\n"}）
空值（例如 {}），当流结束时

3. 流式传输聊天回复能省下多少时间

现在让 gpt-4o-mini 再次数到100，看看需要多长时间。

# 带有 stream=True 的OpenAI ChatCompletion请求例子
# https://platform.openai.com/docs/api-reference/streaming#chat/create-stream# 记录请求发送前的时间
start_time = time.time()# 发送一个ChatCompletion请求来数到100
response = client.chat.completions.create(model='gpt-4o-mini',messages=[{'role': 'user', 'content': '从1数到100，每个数字之间用逗号隔开，不要换行。例如：1, 2, 3, ...'}],temperature=0,stream=True  # 再次设置 stream=True
)
# 创建变量来收集流式的块
collected_chunks = []
collected_messages = []
# 遍历流式事件
for chunk in response:chunk_time = time.time() - start_time  # 计算块的延迟时间collected_chunks.append(chunk)  # 保存事件回复chunk_message = chunk.choices[0].delta.content  # 提取消息collected_messages.append(chunk_message)  # 保存消息print(f"在请求发出后 {chunk_time:.2f} 秒收到消息: {chunk_message}")  # 打印延迟时间和文本# 打印延迟时间和收到的文本
print(f"完整回复在请求发出后 {chunk_time:.2f} 秒内收到")
# 清除 collected_messages 中的 None
collected_messages = [m for m in collected_messages if m is not None]
full_reply_content = ''.join(collected_messages)
print(f"完整对话内容: {full_reply_content}")

时间对比

在上面的例子中，两个请求都大约用了4到5秒才完全完成。实际的请求时间会根据负载和其他随机因素有所不同。

然而，对于流式请求来说，我们在0.1秒后就收到了第一个令牌，随后的令牌则每隔约0.01-0.02秒就会收到一次。

4. 如何获取流式聊天回复的令牌使用数据

你可以通过设置 stream_options={"include_usage": True} 来获取流式回复的令牌使用统计信息。当你这样做时，作为最后一个块会额外流式传输一个块。你可以通过该块上的 usage 字段访问整个请求的使用数据。当你设置 stream_options={"include_usage": True} 时，有几点需要注意：

除了最后一个块外，所有块的 usage 字段的值都将为 null。
最后一个块的 usage 字段包含整个请求的令牌使用统计数据。
最后一个块的 choices 字段将始终是一个空数组 []。

让我们看看它是如何工作的，还是用第2个例子中的方式。

# 带有 stream=True 和 stream_options={"include_usage": True} 的OpenAI ChatCompletion请求例子# 一个ChatCompletion请求
response = client.chat.completions.create(model='gpt-4o-mini',messages=[{'role': 'user', 'content': "1+1等于多少？用一个词回答。"}],temperature=0,stream=True,stream_options={"include_usage": True}, # 获取流式回复的令牌使用情况
)for chunk in response:print(f"choices: {chunk.choices}\nusage: {chunk.usage}")print("****************")