企业官网建设流程全解析

Agent Tool State

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python

::: agents.agent_tool_state

它并不是一篇手写教程，而是由 [docs/scripts/generate_ref_files.py](https://link.gitcode.com/i/d7134c1d9bb54d203477a6153f315422) 自动生成的 mkdocstrings 引用占位文件：脚本扫描 `src/agents/` 下所有非下划线开头的 `.py` 文件，为每个模块生成一个 `docs/ref/<module>.md` 参考页，内容为标题加 `::: <模块全名>` 指令，由 mkdocstrings 在文档构建时提取模块 docstring 与签名渲染成 API 参考。因此，该文档的"真正主体"是 [src/agents/agent_tool_state.py](https://link.gitcode.com/i/576ccf070553b78e33ef8108182f1db8) 这个内部模块。本文即以该模块源码为核心展开。 值得说明的是：该模块的所有函数均以下划线或"内部使用"语义存在（模块本身属于 `agents` 包的内部实现），并非面向最终用户的公共 API。理解它的价值在于：当你使用 `Agent.as_tool()`、`agents-as-tools` 模式、嵌套 Agent 的 human-in-the-loop 审批与 `Runner.run()` 恢复（resume）机制时，能清楚知道状态是如何被记录、查找、隔离与释放的。 ## 二、模块职责与核心数据结构 从源码结构看，该模块要解决的核心问题是：**当外层 Agent 调用一个"Agent 工具"时，内层 Agent 运行产生的 `RunResult`（或流式结果、恢复检查点）需要被暂存起来，供后续查询（如读取嵌套中断、判断审批状态）与消费，并在运行结束后释放。** ### 2.1 三种签名类型 模块顶部定义了三个类型别名（[src/agents/agent_tool_state.py](https://link.gitcode.com/i/576ccf070553b78e33ef8108182f1db8#L14-L16)）： ```python ToolCallSignature = tuple[str, str, str, str, str | None, str | None] ScopedToolCallSignature = tuple[str | None, ToolCallSignature] ScopedToolCallObject = tuple[str | None, int]

• ToolCallSignature：由工具调用的call_id、name、arguments、type、id、status六个字段构成的稳定签名（见_tool_call_signature），用于跨实例的兜底查找。
• ScopedToolCallSignature：在签名基础上加上scope_id前缀，保证不同作用域下恢复的状态互不冲突。
• ScopedToolCallObject：(scope_id, 对象 id)二元组，作为模块级缓存字典的键，其中对象 id 是id(tool_call)。

2.2 恢复检查点_AgentToolResumeCheckpoint

@dataclass class _AgentToolResumeCheckpoint: state: Any # 存活的嵌套 RunState approval_identities: frozenset[tuple[str, str, str, str]]

它用于在"已批准的恢复（approved resume）"进行期间保存存活的嵌套RunState快照（to_state()取回），approval_identities记录该检查点已接受的审批项身份集合，供agent_tool_resume_checkpoint_owns_approval判断某个审批项是否已被该检查点接管。

2.3 四张模块级临时映射表

模块用四张进程内字典实现"按工具调用对象存结果、按签名兜底查、按弱引用清残留"：

这些字典的注释明确写道：它们是"同一运行内把工具调用对象与嵌套 Agent 结果关联起来的临时映射"，"按对象身份存储、按稳定签名索引以避免 call ID 冲突"，并且通过弱引用把缓存生命周期绑定到工具调用对象上以防泄漏。

三、核心 API 逐一解析

3.1 作用域读写：get_agent_tool_state_scope/set_agent_tool_state_scope

_AGENT_TOOL_STATE_SCOPE_ATTR = "_agent_tool_state_scope_id" def get_agent_tool_state_scope(context: Any) -> str | None: scope_id = getattr(context, _AGENT_TOOL_STATE_SCOPE_ATTR, None) return scope_id if isinstance(scope_id, str) else None def set_agent_tool_state_scope(context: Any, scope_id: str | None) -> None: if context is None: return if scope_id is None: try: delattr(context, _AGENT_TOOL_STATE_SCOPE_ATTR) except Exception: return return try: setattr(context, _AGENT_TOOL_STATE_SCOPE_ATTR, scope_id) except Exception: return

作用域 ID 被挂在上下文包装器（RunContextWrapper或ToolContext）的私有属性_agent_tool_state_scope_id上。两处容错设计值得注意：

• context is None时直接返回，读写都不报错；
• setattr/delattr用try/except包裹，即使上下文是只读对象（如纯object()）也能静默容忍。这一行为被 tests/test_agent_tool_state.py 中的test_agent_tool_state_scope_helpers_tolerate_missing_or_readonly_contexts明确验证。

作用域 ID 的来源之一在 src/agents/run_context.py：复制上下文时用uuid4().hex生成新的 scope id，从而让每次独立恢复的运行拥有专属隔离域。

3.2 结果记录：record_agent_tool_run_result

def record_agent_tool_run_result(tool_call, run_result, *, scope_id=None): tool_call_obj_id = id(tool_call) scoped_object = (scope_id, tool_call_obj_id) _agent_tool_run_results_by_obj[scoped_object] = run_result _index_agent_tool_run_result(tool_call, scoped_object, scope_id=scope_id) _register_tool_call_ref(tool_call, tool_call_obj_id)

记录时同时做三件事：按(scope_id, id(tool_call))存入主字典；建立签名索引（_index_agent_tool_run_result内部用_scoped_tool_call_signature生成带 scope 的签名并加入集合）；注册工具调用对象的弱引用（_register_tool_call_ref）。

3.3 恢复检查点记录：record_agent_tool_resume_state

def record_agent_tool_resume_state(tool_call, state, *, scope_id=None, approval_items=None): resolved_approval_items = approval_items if resolved_approval_items is None: get_interruptions = getattr(state, "get_interruptions", None) interruptions = get_interruptions() if callable(get_interruptions) else [] resolved_approval_items = interruptions if isinstance(interruptions, list) else [] approval_identities = frozenset( identity for item in resolved_approval_items if (identity := tool_invocation_identity_and_scope( item.raw_item, tool_lookup_key=getattr(item, "tool_lookup_key", None), tool_name=getattr(item, "tool_name", None), )) is not None ) record_agent_tool_run_result(tool_call, _AgentToolResumeCheckpoint(state, approval_identities), scope_id=scope_id)

它在已批准的恢复进行期间，保存"存活的嵌套RunState检查点"：若未显式传入approval_items，会从state.get_interruptions()提取中断列表；随后借助tool_invocation_identity_and_scope（定义于 src/agents/_tool_invocation.py）为每个审批项计算身份，构建approval_identities冻结集合，最终以_AgentToolResumeCheckpoint形式写入缓存。

3.4 查询与消费：peek/consume/drop

三个函数结构高度对称，均遵循"先按对象身份直查，查不到再按签名兜底"的两段式逻辑：

• peek_agent_tool_run_result：返回缓存结果但不删除；
• consume_agent_tool_run_result：返回并删除缓存结果；
• drop_agent_tool_run_result：只删除，不返回值。

签名兜底有一个关键安全约束：若同一签名命中多个候选对象（len(candidate_ids) != 1），一律返回None或直接放弃，绝不猜测。这由 tests/test_agent_tool_state.py 的test_agent_tool_run_result_returns_none_for_ambiguous_signature_matches验证：对两个签名相同（call-1、相同参数）的不同调用对象分别记录两个结果后，用第三个同签名实例去 peek/consume 均返回None，而原始两个对象仍能各自取出自己的结果。

3.5 检查点辅助查询

• get_agent_tool_resume_state(run_result)：若缓存结果是_AgentToolResumeCheckpoint，返回其中存活的嵌套RunState，否则返回None；
• agent_tool_resume_checkpoint_owns_approval(run_result, approval_item)：判断进行中的嵌套恢复是否已接受某个审批项——仅当结果是检查点、且该审批项的身份出现在approval_identities中时为True。

四、作用域（Scope）隔离机制：为什么需要 scope_id

多层嵌套、多次恢复的场景下，同一个工具调用可能以不同身份被反复执行（例如外层 Agent 多次调用同一 Agent 工具，或一次运行被保存后从不同会话恢复）。如果不加隔离，缓存会互相污染。模块给出的解法是：

1. 作用域 ID 随上下文传递：RunState在构造时从上下文读取 scope id（src/agents/run_state.py），并在复制时同步拷贝（src/agents/run_state.py）；RunState还拥有自己的_agent_tool_state_scope_id字段（src/agents/run_state.py），随状态序列化。
2. 作用域 ID 随恢复上下文重挂：src/agents/run_internal/agent_runner_helpers.py 在恢复（resume）时把run_state._agent_tool_state_scope_id重新写回上下文包装器。
3. 作用域 ID 在运行结束时清空：src/agents/run.py 与 src/agents/run.py 在运行收尾处调用set_agent_tool_state_scope(context_wrapper, None)。
4. 缓存键带 scope：所有记录/查询/消费操作都以(scope_id, ...)为键，保证"同一工具调用、不同 scope"互不干扰。

这一隔离语义被 tests/test_agent_tool_state.py 的test_agent_tool_run_result_keeps_same_call_isolated_by_scope直接验证：同一个tool_call在scope-1与scope-2下分别记录两个结果，peek(scope-1)拿到第一个、peek(scope-2)拿到第二个，且消费scope-1的结果不影响scope-2。

五、生命周期与内存安全：弱引用 + GC 回调

缓存不能无限增长。模块通过_register_tool_call_ref把缓存生命周期绑定到工具调用对象：

def _register_tool_call_ref(tool_call, tool_call_obj_id): def _on_tool_call_gc(_ref): run_results = _agent_tool_run_results_by_obj if isinstance(run_results, dict): scoped_objects = [key for key in run_results if key[1] == tool_call_obj_id] for scoped_object in scoped_objects: run_results.pop(scoped_object, None) _drop_agent_tool_run_result(scoped_object) _agent_tool_call_refs_by_obj[tool_call_obj_id] = weakref.ref(tool_call, _on_tool_call_gc)

当ResponseFunctionToolCall对象被垃圾回收时，_on_tool_call_gc回调会清理主字典中所有指向该对象 id 的条目，并级联清理签名索引。_drop_agent_tool_run_result在删除时还会维护"对象 → 签名"反查表与弱引用表，避免悬挂引用。对应测试是test_agent_tool_run_result_is_dropped_when_tool_call_is_collected（tests/test_agent_tool_state.py）：del tool_call; gc.collect()后，四张表均不再残留该对象 id。此外，_drop_agent_tool_run_result对"全局表在解释器关闭时被清成None"的情况做了防御（test_drop_agent_tool_run_result_handles_cleared_globals），保证收尾阶段不抛异常。

六、在Agent.as_tool()调用链中的实际运作

该模块最核心的消费方是 src/agents/agent.py 中Agent.as_tool()返回的工具实现_run_agent_impl。关键调用序列如下（行号对应 src/agents/agent.py）：

1. 读取作用域（src/agents/agent.py）：tool_state_scope_id = get_agent_tool_state_scope(context)；
2. 构建嵌套上下文并传播作用域（src/agents/agent.py）：如果是ToolContext，新建一个全新的ToolContext（避免与父运行共享审批状态），随后set_agent_tool_state_scope(nested_context, tool_state_scope_id)把作用域写入嵌套上下文；
3. 查询已有结果（src/agents/agent.py）：peek_agent_tool_run_result(context.tool_call, scope_id=tool_state_scope_id)，检查该工具调用是否已有待处理的嵌套结果；
4. 处理恢复检查点（src/agents/agent.py）：get_agent_tool_resume_state取回存活状态直接复用；若嵌套结果存在中断，则用内部_nested_approvals_status汇总审批状态——pending时直接返回原结果不再重复运行，approved/rejected时调用record_agent_tool_resume_state建立检查点（src/agents/agent.py）；
5. 执行嵌套运行（src/agents/agent.py）：Runner.run_streamed（有on_stream时）或Runner.run，恢复模式下以resume_state作为输入；
6. 记录嵌套结果（src/agents/agent.py）：嵌套运行存在中断且位于ToolContext内时，record_agent_tool_run_result(context.tool_call, run_result, scope_id=tool_state_scope_id)把结果按工具调用身份暂存，供外层后续查询中断状态。

除Agent.as_tool()外，该模块还被多处内部代码引用：工具执行器读取作用域（src/agents/run_internal/tool_execution.py）、轮次解析读取作用域（src/agents/run_internal/turn_resolution.py）、项目清理由此丢弃结果（src/agents/run_internal/items.py），以及在复制运行结果时同步拷贝待处理的嵌套 Agent 工具状态（src/agents/result.py）。

七、实战场景与使用建议

7.1 典型场景：多智能体编排中的翻译/路由 Agent

examples/agent_patterns/agents_as_tools.py 展示了标准的 agents-as-tools 用法：编排 Agent 持有三个翻译子 Agent 的as_tool()工具，外层运行时会按上文机制逐个触发嵌套运行并暂存结果：

orchestrator_agent = Agent( name="orchestrator_agent", instructions=( "You are a translation agent. You use the tools given to you to translate." "If asked for multiple translations, you call the relevant tools in order." "You never translate on your own, you always use the provided tools." ), tools=[ spanish_agent.as_tool(tool_name="translate_to_spanish", ...), french_agent.as_tool(tool_name="translate_to_french", ...), italian_agent.as_tool(tool_name="translate_to_italian", ...), ], )

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python