数据截至 (上游 commit ceade4cbe9f2)
Responses API:服务端 agentic 编排循环(皇冠明珠)
30 秒导读: 普通的"chat completion"只回一次话;真正的 agent 要「问模型 → 模型想调工具 → 执行工具 → 把结果喂回模型 → 再问一次……」转好几圈才收尾。OGX 把这整个循环放进了服务端,客户端只发一次请求,就能以流式事件看到中间每一步,循环跑完再落库。这一章讲的就是这个循环怎么转。
本章是 OGX 的核心价值所在。前面几章讲了请求怎么进来(01)、provider 怎么装配(02)、一次模型调用怎么落到后端(03)。这一章往上走一层:多次模型调用 + 工具执行,如何被编排成一个自动循环。具体工具怎么执行(RAG / MCP / 函数)和后台响应、自动压缩,留给 05。
1. 这是什么(零基础也能懂)
先分清两个概念
- Chat Completion(对话补全):你给模型一段对话,模型回你一段话。一问一答,一次结束。这是最底层的原语。
- Response(响应):你给模型一个任务和一堆工具,模型可能"想一下 → 调个搜索工具 → 看结果 → 再调一个 → 最后才回答"。一个 Response 内部可能包含很多次 chat completion 和很多次工具执行。
OpenAI 在 2024 年推出的 Responses API 就是把后者标准化。OGX 实现了这套 API,而且是在自己的服务端完整跑这个循环,而不是把循环甩给客户端。
它解决什么问题 / 给谁用
假设你在写一个"能查资料再回答"的 agent。如果只有 chat completion,你得自己在客户端写这样的循环:
# 示意,非源码:客户端自己扛 agent 循环的痛苦
while True:
resp = call_model(messages, tools) # 问模型
if not resp.tool_calls: # 模型不想调工具了
break
for call in resp.tool_calls: # 逐个执行工具
result = run_tool(call)
messages.append(tool_result_message(result))
messages.append(resp.assistant_message) # 把这轮也记进历史
# 期间还得自己管:流式、用量统计、历史存储、并发工具……
这段循环又臭又长又容易写错:流式怎么拼?多轮的历史怎么攒?工具调用怎么分类(客户端函数 vs 服务端内置 vs MCP)?用量怎么累加?
OGX 的 Responses API 把这一整坨收 进服务端。客户端只需:
# 示意,非源码:客户端只发一次请求
response = client.responses.create(
model="gpt-4o",
input="东京今天天气如何?顺便查下汇率",
tools=[web_search_tool, mcp_tool],
stream=True, # 想看中间过程就开流
)
for event in response: # 服务端把循环的每一步作为事件推过来
print(event.type) # response.created / output_text.delta / ...
用户对象是后端工程师和应用开发者:他们已经在用 OpenAI SDK,想把代码指到 OGX 上"直接就能跑",还想换后端模型(vLLM / Ollama / Bedrock…)不改业务代码。
一句话直觉
把它想成一个餐厅后厨的传菜循环:客户(客户端)只下一次单;后厨(服务端编排器)自己在"问主厨(模型)→ 主厨说还差个食材 → 跑去拿(工具)→ 回来再问主厨"之间来回跑,期间不断把"上菜进度"(流式事件)端给客户看,直到主厨说"齐活了"才收尾结账(落库)。
2. 顶层全景(它大概怎么转)
谁负责什么
这套编排由几个角色协作,职责分得很清:
| 角色 | 干什么 | 在哪个文件 |
|---|---|---|
BuiltinResponsesImpl | Provider 外壳,把 API 请求解包成参数 | src/ogx/providers/inline/responses/builtin/impl.py:99 |
OpenAIResponsesImpl | 门面:输入拼接、落库、后台/压缩、同步会话 | src/ogx/providers/inline/responses/builtin/responses/openai_responses.py:125 |
StreamingResponseOrchestrator | 真正的循环引擎:调模型 ↔ 分离工具 ↔ 执行 ↔ 回喂 | src/ogx/providers/inline/responses/builtin/responses/streaming.py:249 |
ToolExecutor | 具体执行一个工具调用(RAG/MCP/内置),见 05 | src/ogx/providers/inline/responses/builtin/responses/tool_executor.py:117 |
ChatCompletionContext / ToolContext | 贯穿整个循环的共享状态(消息、工具、审批) | src/ogx/providers/inline/responses/builtin/responses/types.py:184 / :98 |
下文所有
.../均指src/ogx/providers/inline/responses/。
一次请求的高层走向
从请求进来到事件流出去,大致这样(从上到下是时间顺序):
客户端 responses.create(input, tools, stream)
│
▼
① BuiltinResponsesImpl.create_openai_response impl.py:125
│ 解包 CreateResponseRequest → 关键字参数
▼
② OpenAIResponsesImpl.create_openai_response openai_responses.py:619
│ 校验参数;background? → 走后台队列(见05)
│ 否则 → _create_streaming_response(...)
▼
③ _create_streaming_response openai_responses.py:1058
│ 输入拼接(接上一条响应/会话/prompt/instructions)
│ 组装 ChatCompletionContext
│ new StreamingResponseOrchestrator
▼
④ orchestrator.create_response() ←── 皇冠明珠的循环 ──→ streaming.py:406
│ 逐个 yield 事件
▼
⑤ 每个事件:_persist_streaming_state 落库 + yield 给上层 openai_responses.py:526
│ 终态事件时:_sync_response_to_conversation openai_responses.py:1567
▼
客户端逐事件收到(stream=True),或聚合成一个终态对象(stream=False)
一句话记住:②③是"准备",④是"跑循环",⑤是"边跑边存边推"。
为什么整条链都是"流式生成器"
注意 ②create_openai_response 无论客户端要不要流,内部总是先拿到 _create_streaming_response 的异步生成器(openai_responses.py:740)。区别只在最后:
stream=True:把生成器原样返回,事件逐个吐给客户端(openai_responses.py:773)。stream=False:服务端自己把生成器消费完,只挑出终态那一个response.completed返回(openai_responses.py:779-828)。
这样设计的好处:落库、会话同步这些副作用,不管客户端流不流都一定会发生——因为它们挂在生成器被消费的过程里,而不是循环之外一个单独的收尾步骤。
3. 核心原理:输入怎么拼(循环开始前)
模型是无状态的——它不记得"上一条响应"。所以循环开始前,服务端必须把完整历史拼成一串 messages 喂给模型。这一步在 _process_input_with_previous_response(openai_responses.py:268)。
三种历史来源
一个新请求的历史可能来自三处,代码用 if/elif 分三条路:
| 场景 | 触发条件 | 怎么拼 |
|---|---|---|
| 接上一条响应 | 传了 previous_response_id | 取旧响应的 input + output,再接新 input |
| 接一个会话 | 传了 conversation | 从 conversation 拉历史 items |
| 全新对话 | 都没传 | 只转换本次 input |
场景一:接上一条响应
这是 agent 多轮的主路径。核心是 _prepend_previous_response(openai_responses.py:252):把旧响应的输入项 + 输出项摊平成 一个列表,再把新 input 追加到尾部。
# 示意,非源码:把"上一条响应"整个接到新输入前面
new_items = list(previous_response.input) # 旧的输入
new_items.extend(previous_response.output) # 旧的输出(含工具调用、回答)
new_items.append(new_user_message) # 这次的新话
真实实现里有个关键优化:优先直接复用旧响应存好的 messages,只把新 input 转成 message 追加(openai_responses.py:294-301),避免每轮都从头重建整段 chat 历史。旧数据没有 messages 时才回退到全量重建(:302-304)。
同时,ToolContext.recover_tools_from_previous_response(types.py:123)会把上一条响应里已经列过的 MCP 工具清单捡回来复用——省掉重复的 tools/list 调用。
场景二 / 三
- 接会话(
conversation):从conversations_api.list_items拉出历史,同样优先用存储的 messages 作真源,新 input 追加(openai_responses.py:307-338)。 - 全新:直接
convert_response_input_to_chat_messages(input)(openai_responses.py:339-341)。
拼完还要"戴帽子":instructions 与 prompt
历史拼好后,_create_streaming_response 还会往 messages 最前面插系统消息:
instructions(本次请求的系统指令):messages.insert(0, system_msg)(openai_responses.py:1113)。- 可复用的 prompt 模板:
_prepend_prompt(openai_responses.py:345)。它从prompts_api取模板,用正则{{var}}把变量替换掉(:403-408),文本变量直接填、图片/文件变量则生成一句[Image: xxx]占位并把真实媒体作为新 user message 追加(:387-415)。
拼装完成后,所有状态被塞进 ChatCompletionContext(openai_responses.py:1122-1134)——这个对象会贯穿整个循环。
4. 核心原理:编排循环本身(皇冠明珠)
进入正题。StreamingResponseOrchestrator.create_response()(streaming.py:434)是一个 async 生成器,它 yield 出的每一个东西都是一个 OpenAI 风格的流式事件。
4.1 循环怎么读:一张图
先建立整体心智模型。怎么读这张图:从上往下是一次 create_response 的生命周期;中间的方框是一个 while True 循环,命中"无工具调用"才跳出。
create_response() streaming.py:406
│
├─ yield response.created 先告诉客户端"我开工了"
├─ yield response.in_progress
│
├─ [可选] 输入 guardrails 校验 run_guardrails(输入) streaming.py:422
│ └─ 命中 → yield 拒答 → return
│
├─ _process_tools() 准备工具清单 streaming.py:1717
│ └─ MCP tools/list、web_search、file_search 挂进 ctx.chat_tools
│
▼
┌──────────────── while True(主循环) ─────────────────┐ streaming.py:489
│ │
│ ① 组装 params,调模型(流式) │ streaming.py:540-584
│ inference_api.openai_chat_completion(stream) │
│ │
│ ② _process_streaming_chunks() │ streaming.py:1024
│ 把 chunk 拼成完整回答 + 累加 usage │
│ 期间 yield: output_text.delta / 工具参数 delta │
│ │
│ ③ _separate_tool_calls() │ streaming.py:773
│ 把本轮工具调用分成:函数(客户端) / │
│ 非函数(服务端内置+MCP) / 待审批 │
│ │
│ ④ 有 reasoning? → 作为 reasoning item 先 yield 出去 │ streaming.py:631
│ │
│ ⑤ _coordinate_tool_execution() │ streaming.py:1427
│ 执行"非函数"工具,结果作为 tool message │
│ 追加到 next_turn_messages(准备回喂) │
│ │
│ ⑥ 收尾判断: │ streaming.py:708-755
│ ├─ 没有任何工具调用 → break(收工) │
│ ├─ 有"函数"(客户端)调用 → break(交给客户端) │
│ └─ 否则 n_iter++,超 max_infer_iters → incomplete│
│ │
│ messages = next_turn_messages(把结果喂回,再转一圈) │ streaming.py:696
└───────────────────────────────────────────────────────┘
│
▼
yield response.completed / response.incomplete / response.failed streaming.py:754-771
4.2 循环三步,逐个拆
步骤 ① 调模型(流式)
每一圈,编排器都根据 ctx 组装一个 OpenAIChatCompletionRequestWithExtraBody(streaming.py:620),关键点:
stream=True+stream_options.include_usage=True:要流式,还要最后一块带上用量统计(streaming.py:616-618)。- 工具会按
allowed_tools过滤(streaming.py:594-600);parallel_tool_calls只在有工具时才传(streaming.py:611-613)。 - 需要推理(reasoning)时优先走
openai_chat_completions_with_reasoning,provider 不支持就降级回普通补全(streaming.py:650-665)。
步骤 ② 拼 chunk + 累加用 量
_process_streaming_chunks(streaming.py:1139)是循环里最长的一段。它做两件事:
- 把流式碎片拼成一个完整回答:文本内容累加进
chat_response_content,工具调用参数按index累加进chat_response_tool_calls(streaming.py:1302-1372)。同时把每一小块作为output_text.delta、function_call_arguments.delta、mcp_call_arguments.delta等事件yield出去,客户端能实时看到打字机效果。 - 累加用量:每个 chunk 调
_accumulate_usage(streaming.py:978)。因为一个 Response 可能调好几次模型,用量必须跨调用累加——accumulated_usage把每次的 prompt/completion/total tokens 加起来(streaming.py:1001-1015)。
这个方法最后不是普通地结束,而是 yield 一个 ChatCompletionResult(streaming.py:1530)——这是给循环主体用的"这一轮的完整结果",和给客户端的流式事件走同一个生成器、靠类型区分(streaming.py:669-674)。
步骤 ③ 分离工具调用
拿到完整回答后,_separate_tool_calls(streaming.py:859)把模型这一轮想调的工具分成三类,因为它们的后续处理完全不同:
| 类别 | 判定 | 谁执行 |
|---|---|---|
| 函数工具(function) | 是客户端注册的函数,或模型瞎编的名字 | 返回给客户端执行 |
| 非函数工具 | web_search / file_search / 已知 MCP 工具 | 服务端自己执行 |
| 待审批 | MCP 工具且 require_approval 命中 | 先发审批请求,等下一轮 |
一个巧妙的容错:模型如果调了一个既不是注册函数、也不是内置、也不是 MCP 的工具名(即幻觉出来的名字),不会让服务器崩,处置按场景分两种(streaming.py:905-938):请求里带了客户端函数工具时,当成客户端函数调用返回——把未知情况优雅地推给客户端(streaming.py:909-918);纯服务端循环里没人能替它执行,则把一条「工具不存在,可用工具有…」的 tool 错误消息回喂模型让它自纠(streaming.py:920-938),超过重试上限仍幻觉就标记 incomplete 退出(streaming.py:784-794)。
这一步还顺便把这一轮的 assistant 消息(含工具调用)追加进 next_turn_messages(streaming.py:889),这就是"回喂"的载体。
步骤 ⑤ 执行工具并回喂
_coordinate_tool_execution(streaming.py:1572)执行非函数(服务端)工具。它逐个调 tool_executor.execute_tool_call(streaming.py:1640,细节见 05),把流式事件转发出去,把工具的输出作为 tool 角色消息追加进 next_turn_messages(streaming.py:1673-1674)。
函数(客户端)工具则不执行,只作为 function_call 输出项 yield 出去(streaming.py:1679-1708)——因为服务端没法执行客户端的代码。
4.3 循环怎么收尾
循环末尾的四个分支决定这一圈之后干嘛(streaming.py:780-838):
- 没有任何工具调用 →
break,模型给出了最终答案,循环结束。 - 幻觉了工具名(纯服务端循环、错误已回喂)→ 计数
n_hallucinated_retries,超过_MAX_HALLUCINATED_TOOL_RETRIES上限就标记incomplete(max_iterations_exceeded)退出;没超则带着纠错消息再转一圈(streaming.py:784-794)。 - 有函数(客户端)工具调用 → 也
break,把球踢给客户端(它执行完会带着结果发新请求接上)。 - 否则(只有服务端工具)→
n_iter += 1,把工具结果messages = next_turn_messages回喂,再转一圈。转够max_infer_iters(默认 10)还没停,就标记为incomplete(原因max_iterations_exceeded)。
还有几个提前收尾的守卫:max_output_tokens 达到(streaming.py:570-581)、模型 finish_reason == "length"(streaming.py:819)都会让状态变 incomplete。
一个防死循环的细节:第一圈如果 tool_choice 是"强制调某工具",第二圈开始自动重置为 auto(streaming.py:806-807),否则会永远被逼着调工具、停不下来。
5. 核心原理:OpenAI 风格的事件序列
Responses API 的价值一半在这套流式事件协议——它必须和 OpenAI 逐字节对齐,客户端 SDK 才能"直接就能跑"。
事件的骨架
一次典型的(带一次工具调用的)响应,事件大致按这个顺序流出:
response.created ← 开工
response.in_progress
response.output_item.added (工具调用 item 出现)
response.function_call_arguments.delta × N (工具参数逐字节)
response.function_call_arguments.done
response.output_item.done
── 服务端执行工具、回喂、再调模型 ──
response.output_item.added (assistant 文本消息)
response.content_part.added
response.output_text.delta × N (回答逐字节)
response.content_part.done
response.output_item.done
response.completed ← 收工(带最终 response 对象 + usage)
每个事件都带一个单调递增的 sequence_number(streaming.py:321 初始化,全程 self.sequence_number += 1),客户端靠它保证顺序。
每次都发一个"响应快照"
response.created / in_progress / completed / incomplete / failed 这些"里程碑"事件,都带一个完整的 OpenAIResponseObject。它由 _snapshot_response(streaming.py:389)现造:根据当前状态和已累积的 output_messages、accumulated_usage,拍一张当前进度的快照。
这就是为什么客户端在流式中途 GET /v1/responses/{id} 也能看到"半成品"状态,而不是空结果。
安全阀:guardrails 在流中拦截
如果开了 guardrails,输出也要边流边查。_process_streaming_chunks 里,文本 delta 不直接 yield,而是先缓冲(streaming.py:1372-1377 的到期判断),每攒够 200 字符(_GUARDRAIL_BATCH_CHARS)就调一次 run_guardrails(streaming.py:1377-1397)。没问题才把缓冲的事件放出去;命中违规则清空缓冲、发一个拒答响应、return(streaming.py:1388-1394),后半段回答绝不外泄。
6. 核心原理:边跑边落库(增量持久化)
循环在 orchestrator.create_response() 里跑,但存储发生在上层 _create_streaming_response(openai_responses.py:1198-1236)。它一边消费编排器吐出的事件,一边在关键节点落库。
在哪些点存
_persist_streaming_state(openai_responses.py:526)按事件类型决定存不存:
| 事件 | 存什么 |
|---|---|
response.in_progress | 初次 INSERT,output 为空 |
response.output_item.done | 增量 UPDATE:当前累积的 output + messages |
response.completed / incomplete | 终态 UPDATE:完整状态 |
response.failed | 存错误状态,便于 GET 看到失败 |
好处很实在:客户端在流式过程中轮询 GET /v1/responses/{id} 就能看到进行中的 turn 状态,而不是等到全部结束(openai_responses.py:539-542 的注释点明了这个意图)。
一个省存储的技巧:增量输入
如果这次是接着 previous_response_id,历史已经存过了,没必要每轮再把整段历史重存一遍(那是 O(n²))。所以代码用 incremental 标志:接续且没触发压缩时,只存本轮新增的 input(openai_responses.py:1192-1196)。这把存储从 O(n²) 降到 O(n)。
终态时同步会话
如果请求带了 conversation,循环成功收尾时会把这一轮的输入 + 输出同步进会 话:_sync_response_to_conversation(openai_responses.py:1567)把 input 和 output items 组装好,调 conversations_api.add_items 落进会话(openai_responses.py:1229-1234)。注意压缩项(OpenAIResponseCompaction)会被过滤掉不入会话(openai_responses.py:1579)。
7. 贯穿状态:ChatCompletionContext 与 ToolContext
循环能转起来,靠两个"随身背包"式的状态对象(都在 types.py)。
ChatCompletionContext(types.py:184)
装一次响应 turn 内、跨多次 chat completion 的累积状态:
messages:当前完整对话(每轮回喂后更新)。chat_tools:转成 OpenAI 格式、真正传给模型的工具清单。response_tools/tool_context:原始工具定义与工具上下文。approval_requests/approval_responses:MCP 工具的审批记录——构造时就从 input 里把审批请求/回应分拣出来(types.py:227-231),approval_response()供循环查"这个工具调用是否已获批"(types.py:233)。
ToolContext(types.py:98)
专管"工具复用"的账本,四个字段分工明确:
| 字段 | 含义 |
|---|---|
current_tools | 本次请求传入的工具 |
previous_tools | 从上一条响应重建的"工具名 → MCP server"映射 |
previous_tool_listings | 上一条响应里可复用的 mcp-list-tools 对象 |
tools_to_process | 本次还需要重新处理的工具 |
它的核心方法 recover_tools_from_previous_response(types.py:123)做的就是:逐个比对当前 MCP 工具和上一条响应里的,allowed_tools 一样的就复用旧清单,不一样的才重新 tools/list(types.py:139-160)。这是 MCP 多轮对话省往返的关键。
8. 巧妙之处(可借鉴的技术)
-
一条生成器同时承载"给客户端的事件"和"给循环的结果"。
_process_streaming_chunks既 yield 流式事件、又 yield 一个ChatCompletionResult,上层靠isinstance分流(streaming.py:669-674)。省掉了"回调 + 返回值"两套机制,一个 async generator 全搞定。 -
流不流,副作用都发生。 因为落库/会话同步挂在生成器消费过程里,
stream=False时服务端 自己消费完(openai_responses.py:779),stream=True时客户端消费;无论哪种,存储都不会漏(openai_responses.py:1221-1223的注释特意说明:即使消费者提前 break 也已存好)。 -
幻觉工具名不崩服务。 模型编一个不存在的工具名:有客户端函数在就当客户端调用退回,纯服务端循环则回喂错误让模型自纠、超限才
incomplete(streaming.py:905-938),而不是抛异常。 -
增量输入把存储从 O(n²) 降到 O(n)。 接续响应时只存新增 input(
openai_responses.py:1189-1196)。 -
第二圈强制把 tool_choice 重置为 auto。 防止"强制调工具"导致的无限循环(
streaming.py:806-807)。 -
guardrails 攒批检查,违规不外泄。 输出 delta 先缓冲,每 200 字符查一次,命中就清缓冲发拒答(
streaming.py:1376-1392)。
9. 边界与局限(诚实)
truncation="auto"未实现。 只支持disabled;传auto直接返回failed,让 provider 自己拒绝超长上下文(streaming.py:279)。stream与background互斥。 同时传会报错(openai_responses.py:659);background 还必须store=True(:662)。- 函数(客户端)工具无法服务端执行。 遇到就 break,把控制权交回客户端(
streaming.py:796-798)— —这是设计使然,不是缺陷,但意味着客户端函数会打断服务端循环。 - 循环有硬上限。
max_infer_iters默认 10、max_output_tokens、max_tool_calls都会强制收尾为incomplete,而非无限跑。 - 用量统计依赖 provider 回传。
include_usage若 provider 不支持,accumulated_usage可能不准(代码尽力累加,见streaming.py:972-1015)。
10. 代码地图(导航索引)
按符号名可 grep 定位,比行号抗漂移。
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| Provider 外壳/解包请求 | src/ogx/providers/inline/responses/builtin/impl.py | BuiltinResponsesImpl.create_openai_response |
| 门面入口/参数校验/分流 | src/ogx/providers/inline/responses/builtin/responses/openai_responses.py | OpenAIResponsesImpl.create_openai_response |
| 组装 context 并启动编排 | src/ogx/providers/inline/responses/builtin/responses/openai_responses.py | _create_streaming_response |
| 输入拼接(三种历史来源) | src/ogx/providers/inline/responses/builtin/responses/openai_responses.py | _process_input_with_previous_response |
| 接上一条响应 | src/ogx/providers/inline/responses/builtin/responses/openai_responses.py | _prepend_previous_response |
| prompt 模板变量替换 | src/ogx/providers/inline/responses/builtin/responses/openai_responses.py | _prepend_prompt |
| 增量持久化(边跑边存) | src/ogx/providers/inline/responses/builtin/responses/openai_responses.py | _persist_streaming_state |
| 首次落库 | src/ogx/providers/inline/responses/builtin/responses/openai_responses.py | _store_response |
| 终态同步进会话 | src/ogx/providers/inline/responses/builtin/responses/openai_responses.py | _sync_response_to_conversation |
| 循环引擎(皇冠明珠) | src/ogx/providers/inline/responses/builtin/responses/streaming.py | StreamingResponseOrchestrator.create_response |
| 准备工具清单 | src/ogx/providers/inline/responses/builtin/responses/streaming.py | _process_tools |
| 拼 chunk + yield 事件 | src/ogx/providers/inline/responses/builtin/responses/streaming.py | _process_streaming_chunks |
| 跨调用累加用量 | src/ogx/providers/inline/responses/builtin/responses/streaming.py | _accumulate_usage |
| 分离工具调用(三类) | src/ogx/providers/inline/responses/builtin/responses/streaming.py | _separate_tool_calls |
| 执行工具并回喂 | src/ogx/providers/inline/responses/builtin/responses/streaming.py | _coordinate_tool_execution |
| 造响应快照 | src/ogx/providers/inline/responses/builtin/responses/streaming.py | _snapshot_response |
| 贯穿状态:对话上下文 | src/ogx/providers/inline/responses/builtin/responses/types.py | ChatCompletionContext |
| 贯穿状态:工具复用账本 | src/ogx/providers/inline/responses/builtin/responses/types.py | ToolContext |
| 一轮的完整结果载体 | src/ogx/providers/inline/responses/builtin/responses/types.py | ChatCompletionResult |
相邻章节:上一步"一次模型调用怎么落到后端"见 03;本章循环里"工具具体怎么执行"以及后台响应、自动压缩见 05;存储与多租户见 06。