工具执行、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:72(ToolExecutor.execute_tool_call),它是个异步生成器——一边执行一边 yield 进度事件,最后 yield 带结果的 ToolExecutionResult。

---

3. ToolExecutor:工具调用怎么被真正执行

3.1 五步骨架

execute_tool_call(tool_executor.py:72)本身很短,像一条流水线,把活分给四个私有方法:

# 示意,非源码。重点看"一次调用被切成五步"
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:316)。于是"工具挂了"不会掀翻整个响应流,只会变成一条 status="failed" 的输出消息。容错的根就扎在这里。

3.2 _execute_tool:一张分派表

_execute_tool(tool_executor.py:316)按工具名把调用路由到不同后端。看清这张表,就看清了 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:328),因为 MCP 工具名是运行时动态注册的,不能和内置名撞车。web_search 分支还会把 response 级配置(allowed_domains、user_location、search_context_size)从 ctx.response_tools 抠出来塞进 kwargs(tool_executor.py:369-383),这样调用方在工具定义上写的过滤条件才真正生效。

每条真执行路径外面都裹了一层 OpenTelemetry span(如 tracer.start_as_current_span("invoke_mcp_tool", ...),tool_executor.py:341),用于分布式追踪工具耗时。

3.3 内置 RAG:file search 怎么接向量库

这是本章工程含量最高的一块。file_search / knowledge_search 不是简单转发,而是在服务端跑一遍完整的检索 + 组装 RAG 上下文。核心在 _execute_file_search_via_vector_store(tool_executor.py:131)。

它要解决的小问题: 模型说"帮我搜 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:156-158),不拖垮其它库。多库检索用 asyncio.gather 一把并发(tool_executor.py:162)。
• 检索走的是向量库的 search API,不是底层 chunk 接口:self.vector_io_api.openai_search_vector_store(...)(tool_executor.py:144),这样才能支持 filters 和 ranking_options。检索模式取自配置 chunk_retrieval_params.default_search_mode。
• RAG 提示词全是模板化的:header/footer/context/annotation 四类模板都来自 VectorStoresConfig,注释开关关掉时回落到默认模板(tool_executor.py:178-187)。这让平台方能不改代码就定制"检索结果长什么样、怎么要求模型 cite"。
• 兼容旧数据:file_id 可能在 result_item.file_id,也可能藏在 attributes["document_id"] 里,代码两处都兜(tool_executor.py:196-198、227-232)。

产出的 metadata 里带了 document_ids / chunks / scores / attributes / citation_files(tool_executor.py:241-247),后面 _build_result_messages 会用它拼出结构化的 file search 结果给客户端。

3.4 流式进度事件:让客户端看见"工具在跑"

工具执行往往要几百毫秒到几秒,期间得给客户端发"心跳"。_emit_progress_events(tool_executor.py:250)和 _emit_completion_events(tool_executor.py:411)按工具类型发不同的语义事件:

工具类型	开始阶段发的事件	结束阶段发的事件
MCP 工具	McpCallInProgress	McpCallCompleted / McpCallFailed
web_search	WebSearchCallInProgress → ...Searching	WebSearchCallCompleted
file_search / knowledge_search	FileSearchCallInProgress → ...Searching	FileSearchCallCompleted

注意 web/file search 在"开始"时会连发两个事件(InProgress 后紧跟 Searching,tool_executor.py:293-314),对应 UI 上"已发起 → 正在搜索"两个状态。每发一个事件 sequence_number += 1——这个全局递增序号是流式协议的排序锚点,04 已详述。

3.5 _build_result_messages:一次执行,两条消息

执行完得到 result,_build_result_messages(tool_executor.py:451)把它拼成两条用途不同的消息:

                       ┌─ 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:481-488);file search 的会把 metadata 里的 document_ids/chunks/scores 展开成一条条 FileSearchToolCallResults(tool_executor.py:516-531)。
• input_message 是给下一轮模型看的"工具返回了啥"。文本直接放,图片转成 data:image;base64,... 的 image part(tool_executor.py:549-554);执行失败时回落成一句 str(error_exc) 或 "Tool execution failed"(tool_executor.py:564-566)——模型于是能"看见"工具报错并自我纠正。这是 agentic 容错的最后一环。

---

4. MCP 集成:调用别人家的工具

MCP(Model Context Protocol,模型上下文协议——让模型能连到外部"工具服务器"的开放协议)是 OGX 工具能力的外部扩展口。它的复杂度分两层:编排器侧管"发现工具、缓存清单、人工批准",会话层管"连上服务器、复用连接"。

4.1 编排器侧:发现工具与缓存清单

当请求里带了一个 type: "mcp" 的工具配置,编排器走 _process_mcp_tool(streaming.py:1616)。它做四件事:

_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:1688,Duplicate tool name),避免两个服务器的同名工具互相覆盖。

缓存复用:多轮对话里,上一轮已经列过的 MCP 工具清单不必重列。_process_tools(streaming.py:1717)会先遍历 previous_tool_listings,对每个走 _reuse_mcp_list_tools(streaming.py:1808)——直接拿旧清单重建工具、重发 list_tools 事件,跳过一次真实的 tools/list 网络往返。

4.2 人工批准:把危险工具挡在门外

有些 MCP 工具(比如"删库""发邮件")不能让模型自作主张调用。OGX 用一道批准门:_approval_required(streaming.py:1733)按服务器配置的 require_approval 判定:

require_approval 配置	判定
"always"	需要批准
"never"	不需要
ApprovalFilter(always=[...], never=[...])	按工具名分别命中;都没命中则默认需要

判定发生在编排器分类工具调用时(_separate_tool_calls,streaming.py:825)。流程是这样的:

模型调用一个需批准的 MCP 工具
        │
  查 ctx.approval_response(找客户端此前是否回过批准)
        ├ 已批准(approve=True) → 放进 non_function_tool_calls,正常执行
        ├ 已拒绝(approve=False) → 丢弃,标记 has_deferred_or_denied
        └ 还没回 → 加进 approvals 列表,发一个 MCP 批准请求给客户端,本轮到此为止

"还没批准"时,_add_mcp_approval_request(streaming.py:1748)生成一个 OpenAIResponseMCPApprovalRequest(带唯一 approval_... id),作为输出项发给客户端并结束本轮。客户端下次带着批准结果再发请求,循环才继续。

一个易漏的细节:当这一轮里有工具被挡下(has_deferred_or_denied),编排器会改写本轮的 assistant 消息——只保留真正执行了的工具调用,或者整条消息弹掉(streaming.py:846-853)。这样喂回模型的历史才自洽,不会出现"消息里挂着一个从没执行的工具调用"。

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)做三件事:

1. 懒启动 worker:_ensure_workers_started(openai_responses.py:158)在当前请求的事件循环里补齐 worker。注释解释了为什么不在 initialize 里启动——provider 初始化用的是临时事件循环,init 一结束就销毁,任务会被连带取消(openai_responses.py:150-155)。
2. 先存一个 queued 响应再入队,保证客户端立刻能查到这个 id。
3. 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:406-409)把所有异常收进二元组,让"工具失败"降级成一条能喂回模型的消息,而不是掀翻响应流——这是 agentic 系统容错的教科书写法。
• 同一次执行产出两条消息。 output(给客户端看的结构化结果)和 input(喂回模型的文本)分离(tool_executor.py:451),职责清晰,也让"给人看"和"给模型看"能各自演化。
• 会话按 (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:526),真实文件名要靠 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 适配。