OpenAI API 流式传输

OpenAI API 流式传输教程 🌊

本教程将详细解释 OpenAI API 如何进行数据流式传输，从基本的文本块到复杂的工具调用指令。流式传输允许你逐步从模型接收数据，这对于构建响应灵敏的用户界面和处理长输出非常有用。

---

1. 基础知识：Server-Sent Events (SSE)

OpenAI 使用 Server-Sent Events (SSE) 作为其流式传输的底层协议。当你发起流式请求时，API 会保持 HTTP 连接打开，并以特定格式发送数据。

• Content-Type：服务器将响应 Content-Type: text/event-stream。
• 事件结构：数据以数据块的形式发送，每个数据块通常以 data: 开头，后跟一个 JSON 有效负载，并以两个换行符 (\n\n) 结束。

  data: {"id":"chatcmpl-xxxx", "choices": [{"delta": {"content": "你好"}}]}data: {"id":"chatcmpl-xxxx", "choices": [{"delta": {"content": "世界"}}]}

• 流结束标记：数据流由一个特殊消息终止：

  data: [DONE]

---

2. 流式传输基本文本响应 📝

这是流式传输最常见的用例——逐块获取模型生成的文本。

请求流式传输：
要在你的 Chat Completions API 请求中启用流式传输，请设置 stream: true。

文本块（Chunk）的结构：
每个 data: 块将包含一个 JSON 对象。对于文本，关键部分包括：

• id: 聊天补全的唯一 ID。
• object: 对象类型 (例如, chat.completion.chunk)。
• created: 创建时间戳。
• model: 使用的模型 (例如, gpt-4o-mini)。
• choices: 一个包含选项的数组。对于流式传输，你通常关注 choices[0]。
  • choices[0].delta: 此对象包含此块中的变更或新数据。
    • delta.role: 通常在助手回合开始时出现一次 (例如, {"role": "assistant"})。
    • delta.content: 这是实际的文本片段。它可能是一个词、词的一部分或几个词。
  • choices[0].finish_reason: 在该选项的流结束前，此字段将为 null。当模型因“自然”停止而完成文本生成时，它将是 stop。其他原因包括 length (达到最大 token 数) 或 tool_calls。

文本流示例流程：

1. 请求：

   {"model": "gpt-4o-mini","messages": [{"role": "user", "content": "给我讲个简短的故事。"}],"stream": true
   }
   

2. 流式响应 (简化的 data: 有效负载序列)：

   • 块 1 (角色和内容开始):

     {"id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 1717500000, "model": "gpt-4o-mini","choices": [{"index": 0, "delta": {"role": "assistant", "content": "从前"}, "finish_reason": null}]
     }
     

   • 块 2 (更多内容):

     {"id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 1717500000, "model": "gpt-4o-mini","choices": [{"index": 0, "delta": {"content": "有个"}, "finish_reason": null}]
     }
     

   • 块 3 (更多内容):

     {"id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 1717500000, "model": "gpt-4o-mini","choices": [{"index": 0, "delta": {"content": "小村庄..."}, "finish_reason": null}]
     }
     

   • 块 4 (生成结束):

     {"id": "chatcmpl-123", "object": "chat.completion.chunk", "created": 1717500000, "model": "gpt-4o-mini","choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]
     }
     

   • 流结束标记:

     data: [DONE]
     

客户端文本处理逻辑：
你需要累积来自每个块的 delta.content 片段以重建完整的消息。

---

3. 流式传输工具调用 (高级) 🛠️

当你希望模型输出结构化数据以调用外部函数或工具时，你会使用“工具调用”(Tool Calling)。这也支持流式传输。

请求工具调用：
在你的 API 请求中包含 tools (描述你可用工具/函数的数组) 和可选的 tool_choice。

工具调用在流中的显示方式：
除了 delta.content（或代替它），你会看到 delta.tool_calls。

tool_call 块的结构：
delta.tool_calls 字段是一个数组，因为模型可能决定调用多个工具。此数组中的每个对象代表特定工具调用的信息块。

• choices[0].delta.tool_calls: 一个数组。每个元素通常具有：
  • index: 一个整数，标识此块属于哪个工具调用 (如果模型并行调用多个工具，则第一个为 0，第二个为 1，依此类推)。
  • id: 此特定工具调用实例的唯一 ID (例如, call_abc123)。此 ID 对于将结果匹配回调用非常重要。
  • type: 总是 "function"。
  • function: 一个包含以下内容的对象：
    • name: 模型想要调用的函数的名称 (例如, get_current_weather)。这可能分部分流式传输，但通常会完整提供。
    • arguments: 一个包含函数 JSON 参数的字符串 (例如, {"location": "波士顿"})。这部分最常以片段形式流式传输。

工具调用流示例流程：

1. 请求 (简化)：

   {"model": "gpt-4o-mini","messages": [{"role": "user", "content": "波士顿今天天气怎么样？"}],"tools": [{"type": "function","function": {"name": "get_current_weather","description": "获取指定地点的当前天气","parameters": {"type": "object","properties": {"location": {"type": "string", "description": "城市和州，例如 San Francisco, CA"}},"required": ["location"]}}}],"stream": true
   }
   

2. 流式响应 (简化的 data: 有效负载序列，重点关注工具调用)：

   • 块 1 (工具调用开始，可能包含 role):

     {"choices": [{"index": 0,"delta": {"role": "assistant", // 可能在这里或之前的块中"content": null,     // 通常为 null，因为模型直接进行工具调用"tool_calls": [{"index": 0// id, type, function name 和 arguments 的开头可能在这里或后续块中}]}}]// ... 其他字段省略 ...
     }
     

   • 块 2 (提供工具调用的 id, type, function.name):

     {"choices": [{"index": 0,"delta": {"tool_calls": [{"index": 0,"id": "call_abc123","type": "function","function": {"name": "get_current_weather","arguments": "{" // 参数开始}}]}}]// ... 其他字段省略 ...
     }
     

   • 块 3 (流式传输 function.arguments 片段):

     {"choices": [{"index": 0,"delta": {"tool_calls": [{"index": 0,"function": {"arguments": "\"location\":\"波" // 参数片段}}]}}]// ... 其他字段省略 ...
     }
     

   • 块 4 (继续流式传输 function.arguments):

     {"choices": [{"index": 0,"delta": {"tool_calls": [{"index": 0,"function": {"arguments": "士顿\"}" // 参数片段结束}}]}}]// ... 其他字段省略 ...
     }
     

   • 块 5 (工具调用信息发送完毕，finish_reason 为 tool_calls):

     {"choices": [{"index": 0,"delta": {}, // delta 可能为空"finish_reason": "tool_calls"}]// ... 其他字段省略 ...
     }
     

   • 流结束标记:

     data: [DONE]
     

处理并行工具调用：
如果模型决定并行调用多个工具，delta.tool_calls 数组中将包含多个具有不同 index 值的对象。客户端需要能够同时累积每个 index 的工具调用信息。

客户端工具调用处理逻辑：

1. 初始化一个数据结构 (例如，一个对象或字典) 来按 index 存储每个工具调用的累积信息 (id, name, arguments 片段)。
2. 当收到包含 delta.tool_calls 的块时，遍历数组中的每个条目。
3. 根据其 index，将 id, function.name 和 function.arguments 的片段附加到相应的累积对象中。
4. 持续拼接 arguments 字符串片段。
5. 当 finish_reason 为 "tool_calls" 时，表示模型已完成指定所有工具调用。此时，你应该已经为每个请求的工具调用累积了完整的 name 和 arguments 字符串。
6. 将累积的 arguments 字符串解析为 JSON 对象，以便执行工具。

---

4. 整合：客户端逻辑要点

1. 初始化累加器：为 choices[0] 维护一个对象，用于累积 role, content。如果涉及工具调用，则需要一个更复杂的结构来按 index 存储每个 tool_call 的 id, name 和 arguments。
2. 处理每个 SSE 事件：
   • 检查是否为 data: [DONE]，如果是则停止处理。
   • 解析 data: 后面的 JSON 字符串。
   • 根据 delta 中的字段更新累加器：
     • 如果存在 delta.role，则设置角色。
     • 如果存在 delta.content，则追加到累积的文本内容中。
     • 如果存在 delta.tool_calls，则遍历数组，根据 index 更新或创建工具调用对象，并追加 id, function.name 和 function.arguments 的片段。
3. 处理 finish_reason：
   • stop: 文本生成完成。
   • length: 因达到 max_tokens 而停止。
   • tool_calls: 模型请求执行工具调用。此时，你需要将收集到的完整工具调用信息用于执行相应的函数，并将结果作为新的消息发送回 API 以继续对话。
   • 其他 finish_reason (如 content_filter)。

---

5. 关键注意事项和最佳实践 ✨

• 使用官方库：OpenAI 提供了 Python 和 Node.js/TypeScript 的官方客户端库，它们极大地简化了流式响应（包括工具调用）的处理，内置了累积和解析逻辑。强烈建议使用这些库。
• 错误处理：网络连接可能会中断，或者 API 可能返回错误。确保你的代码能够妥善处理这些情况。
• 稳健的 JSON 解析：特别是对于流式传输的 function.arguments，在将其解析为 JSON 对象之前，务必确保已收到并正确拼接了所有片段。
• API 版本和模型差异：虽然核心流式结构相对稳定，但始终关注 OpenAI 的官方文档，了解特定模型或 API 版本的任何细微差别或新增功能。

希望这份教程能帮助你更好地理解和使用 OpenAI 的流式 API！