数据截 至 (上游 commit ceade4cbe9f2)
工具执行、RAG/file search、MCP,与后台响应/自动压缩
30 秒导读: 上一章(04)讲的 agentic 主循环,决定"要不要再调一次模型、模型这轮想调哪些工具"。但模型只会说"我要调
file_search(query=...)",它自己不会去搜。这一章讲那句话怎么被真正执行——内置 RAG(接向量库)、web 搜索、以及外部 MCP 工具怎么被发现、批准、调用;再讲两个让系统能扛"长任务、长对话"的高级机制:后台响应(请求立即返回、干活在后台队列里跑)和自动上下文压缩(对话太长就先摘要,省 token)。
1. 这章在整幅图里的位置
先用一句话把边界划清:04 讲"决策",本章讲"执行 + 两个扩展机制"。
主循环每转一圈,模型的输出里可能夹着若干"工具调用"(tool call)。04 的编排器负责把它们分类、排序、决定循环是否继续;真正"把工具跑起来、把结果塞回对话"的脏活,交给本章的两个主角:
| 角色 | 文件 | 职责一句话 |
|---|---|---|
ToolExecutor | providers/inline/responses/builtin/responses/tool_executor.py | 拿到一个 tool call,分派到 RAG/web/MCP/函数,执行,发流式进度,产出结果消息 |
| 编排 器的 MCP 方法 | providers/inline/responses/builtin/responses/streaming.py | 发现 MCP 工具、缓存工具清单、把"需人工批准"的工具挡在门外 |
MCPSessionManager | providers/utils/tools/mcp.py | 一次请求内复用 MCP 连接,避免每次调用都重连、重列工具 |
| 后台执行 + 自动压缩 | providers/inline/responses/builtin/responses/openai_responses.py | 长任务丢队列后台跑;长对话超阈值先摘要 |
本章不重复 04 的主循环骨架;需要主循环上下文时请回看 04。
2. 顶层全景:一次工具调用怎么落地
先看"决策 → 执行"这条缝在哪儿对上。下图从左到右是一次工具调用的生命周期,编排器负责左半(决定),ToolExecutor 负责右半(执行):
编排器主循环(04) ToolExecutor(本章)
┌───────────────────────────┐ ┌────── ────────────────────────┐
│ 模型返回 tool_calls │ │ execute_tool_call │
│ │ │ ① 发"开始"进度事件 │
│ _separate_tool_calls │ │ ② _execute_tool 分派执行 │
│ ├ 函数工具 → 交回客户端 │ │ ├ file_search → 向量库 │
│ ├ 内置/MCP → 服务端执行 │──每个──▶ │ ├ web_search → 工具运行时│
│ └ 需批准? → 挡下, 发批准请求│ tool_call │ └ MCP 名 → invoke_mcp │
└───────────────────────────┘ │ ③ 发"完成/失败"进度事件 │
▲ │ ④ _build_result_messages │
│ 把工具结果塞回 next_turn │ → 输出消息 + 喂回模型的消息│
└────────────────────────────────│ ⑤ yield 最 终结果 │
└──────────────────────────────┘
怎么读:一个 tool call 进 execute_tool_call,走 ①→⑤ 五步;第 ⑤ 步产出的"喂回模型的消息"会被编排器 append 进 next_turn_messages,于是下一轮模型就"看见"了工具结果。这正是 agentic 循环能自我推进的燃料。
入口签名在 tool_executor.py:135(ToolExecutor.execute_tool_call),它是个异步生成器——一边执行一边 yield 进度事件,最后 yield 带结果的 ToolExecutionResult。
3. ToolExecutor:工具调用怎么被真正执行
3.1 五步骨架
execute_tool_call(tool_executor.py:135)本身很短,像一条流水线,把活分给四个私有方法:
# 示意,非源码。重点看"一次调用被切成五步"
async def execute_tool_call(tool_call, ctx, ...):
kwargs = json.loads(tool_call.function.arguments) # 模型给的参数是 JSON 字符串
# ① 发"开始"事件(不同工具类型发不同事件)
async for ev in self._emit_progress_events(name, ...): yield ev
# ② 真正执行(唯一会失败的地方,错误被收进 error_exc)
error_exc, result = await self._execute_tool(name, kwargs, ctx, mcp_map)
# ③ 发"完成/失败"事件
async for ev in self._emit_completion_events(name, ..., has_error): yield ev
# ④ 把结果拼成两条消息(见 3.5)
out_msg, in_msg = await self._build_result_messages(...)
# ⑤ 交回最终结果 + 引用文件
yield ToolExecutionResult(final_output_message=out_msg, final_input_message=in_msg, ...)
一个关键设计:执行只发生在第 ② 步,且被 try/except 兜住——_execute_tool 从不抛异常给上层,而是返回 (error_exc, result) 二元组(tool_executor.py:383)。于是"工具挂了"不会掀翻整个响应流,只会变成一条 status="failed" 的输出消息。容错的根就扎在这里。
3.2 _execute_tool:一张分派表
_execute_tool(tool_executor.py:383)按工具名把调用路由到不同后端。看清这张表,就看清了 OGX 支持哪几类工具:
| 判定条件 | 走哪条路 | 落到哪个 API |
|---|---|---|
工具名在 mcp_tool_to_server 里 | invoke_mcp_tool(...) | 外部 MCP 服务器(见 §4) |
工具名是 knowledge_search / file_search | _execute_file_search_via_vector_store | 内置 RAG → VectorIO(见 §3.3) |
工具名是 web_search | tool_runtime_api.invoke_tool | 工具运行时 provider |
| 其它(兜底) | tool_runtime_api.invoke_tool | 工具运行时 provider |
判定顺序很重要:MCP 优先(tool_executor.py:395),因为 MCP 工具名是运行时动态注册的,不能和内置名撞车。web_search 分支还会把 response 级配置(allowed_domains、user_location、search_context_size)从 ctx.response_tools 抠出来塞进 kwargs(tool_executor.py:436-450),这样调用方在工具定义上写的过滤条件才真正生效。
每条真执行路径外面都裹了一层 OpenTelemetry span(如 tracer.start_as_current_span("invoke_mcp_tool", ...),tool_executor.py:408),用于分布式追踪工具耗时。
3.3 内置 RAG:file search 怎么接向量库
这是本章工程含量最高的一块。file_search / knowledge_search 不是简单转发,而是在服务端跑一遍完整的检索 + 组装 RAG 上下文。核心在 _execute_file_search_via_vector_store(tool_executor.py:194)。
它要解决的小问题: 模型说"帮我搜 X",系统得:在多个向量库里并行检索 → 把命中的 chunk 拼成一段带引用标注的文本 → 让模型能在回答里正确 cite 来源文件。
思路,分四步走:
① 并行检索:对每个 vector_store_id 起一个 search 任务
search_single_store(vid) ──┐
search_single_store(vid) ──┼─▶ asyncio.gather ─▶ 扁平化所有命中
search_single_store(vid) ──┘
② 拼装:header 模板 + 每个 chunk 套 annotation 模板 + footer 模板
③ 收集引用:{file_id → filename} 映射(citation_files)
④ 打包成 ToolInvocationResult(content=文本块, metadata=检索明细)
几个值得记住的细节:
- 并行 + 单库容错:每个库的检索包在自己的 try/except 里,某个库挂了只
logger.warning并返回空列表(tool_executor.py:219-221),不拖垮其它库。多库检索用asyncio.gather一把并发(tool_executor.py:225)。 - 检索走的是向量库的 search API,不是底层 chunk 接口:
self.vector_io_api.openai_search_vector_store(...)(tool_executor.py:207),这样才能支持filters和ranking_options。检索模式取自配置chunk_retrieval_params.default_search_mode。 - RAG 提示词全是模板化的:header/footer/context/annotation 四类模板都来自
VectorStoresConfig,注释开关关掉时回落到默认模板(tool_executor.py:241-250)。这让平台方能不改代码就定制" 检索结果长什么样、怎么要求模型 cite"。 - 兼容旧数据:
file_id可能在result_item.file_id,也可能藏在attributes["document_id"]里,代码两处都兜(tool_executor.py:259-261、227-232)。
产出的 metadata 里带了 document_ids / chunks / scores / attributes / citation_files(tool_executor.py:308-314),后面 _build_result_messages 会用它拼出结构化的 file search 结果给客户端。
3.4 流式进度事件:让客户端看见"工具在跑"
工具执行往往要几百毫秒到几秒,期间得给客户端发"心跳"。_emit_progress_events(tool_executor.py:317)和 _emit_completion_events(tool_executor.py:478)按工具类型发不同的语义事件:
| 工具类型 | 开始阶段发的事件 | 结束阶段发的事件 |
|---|---|---|
| MCP 工具 | McpCallInProgress | McpCallCompleted / McpCallFailed |
web_search | WebSearchCallInProgress → ...Searching | WebSearchCallCompleted |
file_search / knowledge_search | FileSearchCallInProgress → ...Searching | FileSearchCallCompleted |
注意 web/file search 在"开始"时会连发两个事件(InProgress 后紧跟 Searching,tool_executor.py:360-381),对应 UI 上"已发起 → 正在搜索"两个状态。每发一个事件 sequence_number += 1——这个全局递增序号是流式协议的排序锚点,04 已详述。
3.5 _build_result_messages:一次执行,两条消息
执行完得到 result,_build_result_messages(tool_executor.py:518)把它拼成两条用途不同的消息:
┌─ output_message ─▶ 放进 output_messages,是给【客户端】看的
result ── 拼装 ──┤ (结构化: MCPCall / WebSearchToolCall / FileSearchToolCall)
└─ input_message ─▶ 塞回 next_turn_messages,是【喂回模型】的
(OpenAIToolMessageParam, tool_call_id 对齐)
- output_message 按工具类型分三种结构。MCP 的走
OpenAIResponseOutputMessageMCPCall,出错时把错误码/信息塞进.error(tool_executor.py:548-555);file search 的会把metadata里的document_ids/chunks/scores展开成一条条FileSearchToolCallResults(tool_executor.py:583-598)。 - input_message 是给下一轮模型看的"工具返回了啥"。文本直接放,图片转成
data:image;base64,...的 image part(tool_executor.py:620-625);执行失败时回落成一句str(error_exc)或"Tool execution failed"(tool_executor.py:637-639)——模型于是能"看见"工具报错并自我纠正。这是 agentic 容错的最后一环。
4. MCP 集成:调用别人家的工具
MCP(Model Context Protocol,模型上下文协议——让模型能连到外部"工具服务器"的开放协议)是 OGX 工具能力的外部扩展口。它的复杂度分两层:编排器侧管"发现工具、缓存清单、人工批准",会话层管"连上服务器、复用连接"。
4.1 编排器侧:发现工具与缓存清单
当请求里 带了一个 type: "mcp" 的工具配置,编排器走 _process_mcp_tool(streaming.py:1761)。它做四件事:
_process_mcp_tool
① 若给了 connector_id,先解析成 server_url(resolve_mcp_connector_id)
② 发 mcp_list_tools.in_progress 事件
③ list_mcp_tools(server_url) 拉到服务器暴露的所有工具
④ 对每个工具:
├ 命中 never_allowed → 跳过
├ 不在 always_allowed → 跳过
└ 保留 → 注册进 3 张表(见下)+ 加进 mcp_list_message
保留的工具会被登记进三张映射表,后续执行和批准全靠它们:
| 映射表 | 键 → 值 | 用途 |
|---|---|---|
ctx.chat_tools | 追加一个工具定义 | 让模型"知道有这个工具可调" |
mcp_tool_to_server | 工具名 → MCP 服务器配置 | 执行时定位 server_url / 判批准 |
server_label_to_tools | 服务器标签 → 工具名列表 | tool_choice 按服务器筛选时用 |
同名工具会直接报错(streaming.py:1833,Duplicate tool name),避免两个服务器的同名工具互相覆盖。
缓存复用:多轮对话里,上一轮已经列过的 MCP 工具清单不必重列。_process_tools(streaming.py:1862)会先遍历 previous_tool_listings,对每个走 _reuse_mcp_list_tools(streaming.py:1953)——直接拿旧清单重建工具、重发 list_tools 事件,跳过一次真实的 tools/list 网络往返。
4.2 人工批准:把危险工具挡在门外
有些 MCP 工具(比如"删库""发邮件")不能让模型自作主张调用。OGX 用一道批准门:_approval_required(streaming.py:1878)按服务器配置的 require_approval 判定:
require_approval 配置 | 判定 |
|---|---|
"always" | 需要批准 |
"never" | 不需要 |
ApprovalFilter(always=[...], never=[...]) | 按工具名分别命中;都没命中则默认需要 |
判定发生在编排器分类工具调用时(_separate_tool_calls,streaming.py:940)。流程是这样的:
模型调用一个需批准的 MCP 工具
│
查 ctx.approval_response(找客户端此前是否回过批准)
├ 已批准(approve=True) → 放进 non_function_tool_calls,正常执行
├ 已拒绝(approve=False) → 丢弃,标记 has_deferred_or_denied
└ 还没回 → 加进 approvals 列表,发一个 MCP 批准请求给客户端,本轮到此为止
"还没批准"时,_add_mcp_approval_request(streaming.py:1893)生成一个 OpenAIResponseMCPApprovalRequest(带唯一 approval_... id),作为输出项发给客户端并结束本轮。客户端下次带着批准结果再发请求,循环才继续。
一个易漏的细节:当这一轮里有工具被挡下(has_deferred_or_denied),编排器会改写本轮的 assistant 消息——只保留真正执行了的工具调用,或者整条消息弹掉(streaming.py:961-968)。这样喂回模型的历史才自洽,不会出现"消息里挂着一个从没执行的工具调用"。
4.3 MCP 会话层:连接复用与协议自动探测
到了真调用,底层是 MCPSessionManager(mcp.py:81)。它解决的是 GitHub issue #4452:每次调工具前都重列一遍工具清单、重建一次连接,太浪费。
它的核心是按 (endpoint, headers_hash) 缓存已建立的会话。 get_session(mcp.py:120)是标准的"双检锁"缓存:
# 示意,非源码。重点看"快路径 + 按 key 加锁防并发重建"
def get_session(endpoint, headers):
key = make_key(endpoint, headers) # endpoint + headers 的 sha256 前16位
if key in self._sessions: # 快路径:已有会话直接返回
return self._sessions[key][0]
lock = get_lock(key) # 每个 key 一把锁
with lock:
if key in self._sessions: # 拿到锁后再检查一次(可能别人刚建好)
return self._sessions[key][0]
session = create_session(endpoint, headers) # 真正建连
self._sessions[key] = session
return session
同一次请求内,list_mcp_tools 和多次 invoke_mcp_tool 都传同一个 session_manager,于是共用一条连接。整个 session manager 的生命周期是一次流式响应——在 _create_streaming_response 里用 async with MCPSessionManager() as mcp_session_manager(openai_responses.py:1143)包住,请求结束时 __aexit__ 统一关闭所有会话。
协议自动探测 + 降级:_create_session(mcp.py:151)不知道对面服务器用哪种传输,就按顺序试:
默认顺序: [STREAMABLE_HTTP, SSE]
│
├ 成功 → 把命中的协议写进 protocol_cache(下次这个 endpoint 直接优先它)
└ 失败 → 按异常类型分流:
├ 401 → 立刻抛 AuthenticationRequiredError(不重试,要鉴权)
├ ConnectError → 还有下一个策略就 fallback,否则抛 ConnectionError
├ Timeout → 同上,最后抛 TimeoutError
└ McpError → 还有下一个策略就 fallback,否则抛
这里用了 Python 3.11 的 except*(异常组语法)——因为 MCP 客户端底层用 anyio task group,异常会被包成 ExceptionGroup。protocol_cache 是个 TTL 字典(见下),命中过的协议记 1 小时,避免每次都从头试。
4.4 TTLDict:带过期的协议缓存
protocol_cache(mcp.py:70)的类型是 TTLDict(ttl_dict.py:12)——一个"每个 key 独立过期"的字典。实现很朴素但线程安全:
__setitem__时记下_expires[key] = monotonic() + ttl(ttl_dict.py:32-35);__getitem__/__contains__时若已过期,惰性删除并抛KeyError(ttl_dict.py:42-49);- 全程用
RLock保护(可重入锁,get内部会再调__getitem__)。
用 time.monotonic() 而非 time.time(),是为了不受系统时钟回拨影响。协议探测结果缓存 1 小时(ttl_seconds=3600),既省了重复探测,又能在服务器换了传输方式后自动失效重探。
5. 后台执行:请求秒回,活在后台干
有些响应要跑很久(多轮工具调用、大检索)。让客户端连着 HTTP 干等几分钟不现实。后台模式让 create_openai_response(background=True) 立即返回一个 status="queued" 的对象,真正的处理丢进一个 worker 池异步跑;客户端之后靠 GET /v1/responses/{id} 轮询进度。
5.1 全景:一个队列 + 一池 worker
create_openai_response(background=True)
│ ①存一个 queued 响应,②入队一个 _BackgroundWorkItem
▼
_background_queue (asyncio.Queue, 最多 100 个)
│
├── worker ┐
├── worker ┼─ 10 个 _background_worker 协程,各自 get() 一个 item
└── worker ┘ │
▼
_run_background_response_loop(**item.kwargs)
复用 _create_streaming_response,把流"喝干"
途中周期性 查 status=="cancelled" 就中止
终态写回 responses_store(completed/failed/cancelled)
关键常量(openai_responses.py:93-95):队列上限 100(BACKGROUND_QUEUE_MAX_SIZE)、10 个 worker(BACKGROUND_NUM_WORKERS)、单个后台响应超时 300 秒(BACKGROUND_RESPONSE_TIMEOUT_SECONDS)。
5.2 入队:_create_background_response
_create_background_response(openai_responses.py:830)做三件事:
- 懒启动 worker:
_ensure_workers_started(openai_responses.py:158)在当前请求的事件循环里补齐 worker。注释解释了为什么不在initialize里启动——provider 初始化用的是临时事件循环,init 一结束就销毁,任务会被连带取消(openai_responses.py:150-155)。 - 先存一个
queued响应再入队,保证客户端立刻能查到这个 id。 put_nowait入队,满了直接抛ValueError("queue is full")(openai_responses.py:940-943),用背压保护服务器,而不是无限堆积。
入队的 _BackgroundWorkItem(openai_responses.py:99)不只装业务 kwargs,还装了 capture_request_context()——把当前请求的上下文(租户、认证等)一起打包,worker 里再用 activate_request_context 还原(openai_responses.py:185)。这样后台任务才不会丢失"我是替谁干活"的身份。多租户隔离细节见 06。
5.3 干活:worker 与三种终态
_background_worker(openai_responses.py:181)是个死循环,不停 get() 出 item。它把真处理包成一个可取消的 task,再套 asyncio.wait_for(..., timeout=300):
# 示意,非源码。重点看"三种异常 → 三种终态写回"
task = asyncio.create_task(wait_for(self._run_background_response_loop(**kwargs), timeout=300))
self._background_response_tasks[response_id] = task # 登记,便于外部取消
try:
await task
except CancelledError: 存 status="cancelled"
except TimeoutError: 存 status="failed", error="timed out after 300s"
except Exception as e: 存 status="failed", error=str(e)
finally: 从登记表移除, queue.task_done()
把 task 登记进 _background_response_tasks(openai_responses.py:197-198)是为了让 cancel_openai_response 能找到它并 .cancel()。每种失败都尽力写回终态,并且连"写终态也失败"都记了日志(openai_responses.py:226-230),提醒"轮询的客户端将看不到这次失败"。
_run_background_response_loop(openai_responses.py:947)本身很克制:它复用普通的 _create_streaming_response,把流式 chunk 一个个消费掉,每个 chunk 前都查一次 status=="cancelled"(openai_responses.py:1022-1026),实现协作式取消;只把 response.completed/incomplete/failed 的终态响应留下来最后写库。
6. 自动上下文压缩:长对话省 token
对话越滚越长,喂给模型的 token 越来越多——又贵又可能超窗口。**压缩(compaction)**的思路:把老历史交给模型摘要成一段,用摘要替换原文,只保留用户消息原文 + 一段"交接摘要"。
6.1 两个入口:手动 vs 自动
| 入口 | 触发 | 方法 |
|---|---|---|
| 手动压缩 | 客户端显式调 compact 接口 | compact_openai_response(openai_responses.py:1241) |
| 自动压缩 | 请求带 context_management,且历史超阈值 | _maybe_auto_compact(openai_responses.py:1487)→ 内部调上面那个 |
自动压缩挂在 _create_streaming_response 开头(openai_responses.py:1105):算一下当前输入的 token,超阈值就先把历史压成摘要再进主循环。压过之后 compacted_history_applied = True,后续存储会把"压缩后的完整历史"存成快照,而不是增量(openai_responses.py:1192),这样下一轮用 previous_response_id 才能接着压缩后的上下文走。
6.2 _maybe_auto_compact:阈值判定
# 示意,非源码。重点看"优先用 provider 报的真实 token 数,否则自己估"
for entry in context_management:
if entry.type != "compaction": continue
threshold = entry.compact_threshold or config.default_compact_threshold
if threshold is None: continue
# 上一轮 usage 里有真实 total_tokens 就用它,否则用 tiktoken 估
token_count = previous_usage.total_tokens or self._count_tokens(input, model=model)
if token_count > threshold:
compacted = await self.compact_openai_response(model=model, input=input)
return list(compacted.output) # 用摘要替换原输入
优先信任 provider 回报的 total_tokens(openai_responses.py:1510-1511)——那是最准的;拿不到才自己估。
6.3 compact_openai_response:摘要怎么生成
compact_openai_response(openai_responses.py:1241)的核心动作:把整段历史转成 chat 消息 → 末尾追加一条"请你摘要"的用户消息(summarization_prompt)→ 用 summarization_model(没配就用当轮模型)跑一次非流式 chat completion → 拿回 摘要文本。
产出的"压缩结果"结构很讲究(openai_responses.py:1324-1353):
输出 = [ 所有原始 user 消息(逐字保留) ..., 一个 OpenAIResponseCompaction 项 ]
└ encrypted_content = summary_prefix + 摘要正文
即:用户说过的话一字不改地留着(对齐 OpenAI 行为),把 assistant/工具那些冗长的中间过程压成一段摘要,并加个 summary_prefix 前缀把它"框"成给下一个 LLM 的交接说明。压缩结果会被存成一个正常的 OpenAIResponseObject,于是它的 id 能当 previous_response_id 继续用(openai_responses.py:1378-1397)。
6.4 token 计数:一条 5 级降级链
自己估 token 时,得先决定用哪个 tiktoken 编码。_resolve_encoding(openai_responses.py:1406)是一条5 级降级链,从"最精确/最该硬失败"到"最兜底":
① 请求级覆盖 extra_body["tokenizer_encoding"] → 无效就【硬报错】(用户明确指定了,不能悄悄忽略)
② 管理员默认 config.tokenizer_encoding → 启动时已校验,直接用
③ tiktoken 按模型名自动解析 → 解析不了就往下(软失败)
④ 模型家族前缀映射 model_tokenizer_mappings → 如 "llama*"→cl100k_base(软失败)
⑤ 都不行 → 返回 None,回落到"字符数 ÷ 4"的粗估
_count_tokens(openai_responses.py:1444)据此二选一:拿到编码就精确 encode(_count_with_encoding),拿不到就 _estimate_tokens_by_chars(每 4 字符约 1 token,openai_responses.py:1481)。无论哪条路,都先用 _extract_text_segments(openai_responses.py:1454)把各种输入项(消息、compaction、工具参数/输出)里的文本抠出来再算。
设计取舍很清楚:用户显式指定的编码错了要立刻报错(第①级 raise),但系统自己推断失败时绝不阻断请求,一路软降级到字符估算。CompactionConfig(config.py:35)在启动时就用 field_validator 校验 tokenizer_encoding 和 model_tokenizer_mappings 里的编码名都合法(config.py:82-106),把"配错编码"的问题挡在启动期而非请求期。
7. 巧妙之处(可借鉴)
- 执行永不抛异常,只返回
(error, result)。_execute_tool(tool_executor.py:473-476)把所有异常收进二元组,让"工具失败"降级成一条能喂回模型的消息,而不是掀翻响应流——这是 agentic 系统容错的教科书写法。 - 同一次执行产出两条消息。 output(给客户端看的结构化结果)和 input(喂回模型的文本)分离(
tool_executor.py:518),职责清晰,也让"给人看"和"给模型看"能各自演化。 - 会话按
(endpoint, headers)缓存 + 双检锁。MCPSessionManager.get_session(mcp.py:120)用极少的代码解决了 #4452 的重复连接问题,并发安全还不牺牲快路径。 - 协议探测结果进 TTL 缓存。
protocol_cache(mcp.py:70)+TTLDict,命中协议记 1 小时,既省探测又能自动失效——用monotonic()防时钟回拨是加分项。 - 后台任务携带请求上下文。
_BackgroundWorkItem打包capture_request_context()(openai_responses.py:908),worker 里activate_request_context还原,让异步任务不丢租户/认证身份。 - token 计数分"硬失败/软失败"两档。 用户指定错了立刻报,系统推断失败一路软降级(
openai_responses.py:1406),把严格性用在该严格的地方。
8. 边界与局限(诚实说)
- 后台队列有硬上限。 队列满 100 就直接拒绝新的后台请求(
openai_responses.py:940-943),单个后台响应超 300 秒判失败(openai_responses.py:192)。这是有意的背压,但意味着高并发长任务场景需要外部排队/扩容配合。 - MCP 会话只在一次请求内复用。
MCPSessionManager的生命周期是单次流式响应(openai_responses.py:1143),跨请求不复用连接——换来的是干净的生命周期和隔离,代价是每个请求首个 MCP 调用仍要建连。 - file search 的
filename可能退化成document_id。_build_result_messages里 file search 结果的filename直接填了doc_id(tool_executor.py:593),真实文件名要靠citation_files另走一路,代码里能看出这块对旧数据的兼容还不够统一。 - 字符估算 token 是粗糙兜底。 回落到"字符 ÷ 4"(
openai_responses.py:1483)对中文/代码等场景误差不小,只适合当"实在没编码时别崩"的下限。 - 压缩会丢失中间过程细节。 摘要天然有损:assistant 的推理、工具的完整输出被压成一段(
openai_responses.py:1324-1353),只有 user 原文逐字保留。对需要精确回溯中间步骤的任务要谨慎开自动压缩。
9. 代码地图(导航索引)
用符号名 grep 定位比行号更抗漂移。以下均相对克隆根 aiRef/repos/ogx/。
| 主题 | 文件路径 | 符号名 |
|---|---|---|
| 工具执行入口(五步流水线) | src/ogx/providers/inline/responses/builtin/responses/tool_executor.py | ToolExecutor.execute_tool_call |
| 工具分派表(MCP/RAG/web/函数) | 同上 | ToolExecutor._execute_tool |
| 内置 RAG:多库并行检索 + 组装 | 同上 | ToolExecutor._execute_file_search_via_vector_store |
| 流式进度事件 | 同上 | ToolExecutor._emit_progress_events |
| 流式完成/失败事件 | 同上 | ToolExecutor._emit_completion_events |
| 拼装 output/input 双消息 | 同上 | ToolExecutor._build_result_messages |
| MCP 工具发现 + 三张映射表 | src/ogx/providers/inline/responses/builtin/responses/streaming.py | StreamingResponseOrchestrator._process_mcp_tool |
| MCP 清单缓存复用 | 同上 | _reuse_mcp_list_tools / _add_mcp_list_tools |
| 批准门判定 | 同上 | _approval_required |
| 发批准请求 | 同上 | _add_mcp_approval_request |
| 工具调用分类 + 批准流程 | 同上 | _separate_tool_calls |
| MCP 会话复用(双检锁缓存) | src/ogx/providers/utils/tools/mcp.py | MCPSessionManager.get_session |
| MCP 建连 + 协议探测/降级 | 同上 | MCPSessionManager._create_session |
| 列/调 MCP 工具 | 同上 | list_mcp_tools / invoke_mcp_tool |
| 带过期的协议缓存 | src/ogx/providers/utils/tools/ttl_dict.py | TTLDict |
| 后台响应入队 | src/ogx/providers/inline/responses/builtin/responses/openai_responses.py | OpenAIResponsesImpl._create_background_response |
| 后台 worker(三种终态) | 同上 | _background_worker |
| 后台处理循环 + 协作取消 | 同上 | _run_background_response_loop |
| 后台队列项(带请求上下文) | 同上 | _BackgroundWorkItem |
| 手动压缩 + 摘要生成 | 同上 | compact_openai_response |
| 自动压缩阈值判定 | 同上 | _maybe_auto_compact |
| token 计数 | 同上 | _count_tokens |
| tiktoken 编码 5 级降级链 | 同上 | _resolve_encoding |
| 压缩/token 配置(启动期校验) | src/ogx/providers/inline/responses/builtin/config.py | CompactionConfig |
相邻章节: 主循环骨架见 04 Responses agentic 循环;多租户/请求上下文传播见 06 存储与租户隔离;工具调用如何落到某个推理后端见 03 路由与 OpenAI 适配。