`Agent Tool State` 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 恢复检查点_AgentToolResumeCheckpointdataclass 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_tool_run_results_by_objScopedToolCallObject嵌套RunResult/流式结果/检查点主存储按对象身份直查_agent_tool_run_results_by_signatureScopedToolCallSignatureset[ScopedToolCallObject]签名兜底索引避免 call ID 冲突_agent_tool_run_result_signature_by_objScopedToolCallObjectScopedToolCallSignature反查对象 → 签名_agent_tool_call_refs_by_objint对象 idweakref.ref(ResponseFunctionToolCall)弱引用挂钩工具调用被 GC 时清理缓存这些字典的注释明确写道它们是同一运行内把工具调用对象与嵌套 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_resultdef record_agent_tool_run_result(tool_call, run_result, *, scope_idNone): 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_idscope_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_statedef record_agent_tool_resume_state(tool_call, state, *, scope_idNone, approval_itemsNone): 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_keygetattr(item, tool_lookup_key, None), tool_namegetattr(item, tool_name, None), )) is not None ) record_agent_tool_run_result(tool_call, _AgentToolResumeCheckpoint(state, approval_identities), scope_idscope_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否则返回Noneagent_tool_resume_checkpoint_owns_approval(run_result, approval_item)判断进行中的嵌套恢复是否已接受某个审批项——仅当结果是检查点、且该审批项的身份出现在approval_identities中时为True。四、作用域Scope隔离机制为什么需要 scope_id多层嵌套、多次恢复的场景下同一个工具调用可能以不同身份被反复执行例如外层 Agent 多次调用同一 Agent 工具或一次运行被保存后从不同会话恢复。如果不加隔离缓存会互相污染。模块给出的解法是作用域 ID 随上下文传递RunState在构造时从上下文读取 scope idsrc/agents/run_state.py并在复制时同步拷贝src/agents/run_state.pyRunState还拥有自己的_agent_tool_state_scope_id字段src/agents/run_state.py随状态序列化。作用域 ID 随恢复上下文重挂src/agents/run_internal/agent_runner_helpers.py 在恢复resume时把run_state._agent_tool_state_scope_id重新写回上下文包装器。作用域 ID 在运行结束时清空src/agents/run.py 与 src/agents/run.py 在运行收尾处调用set_agent_tool_state_scope(context_wrapper, None)。缓存键带 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_collectedtests/test_agent_tool_state.pydel 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读取作用域src/agents/agent.pytool_state_scope_id get_agent_tool_state_scope(context)构建嵌套上下文并传播作用域src/agents/agent.py如果是ToolContext新建一个全新的ToolContext避免与父运行共享审批状态随后set_agent_tool_state_scope(nested_context, tool_state_scope_id)把作用域写入嵌套上下文查询已有结果src/agents/agent.pypeek_agent_tool_run_result(context.tool_call, scope_idtool_state_scope_id)检查该工具调用是否已有待处理的嵌套结果处理恢复检查点src/agents/agent.pyget_agent_tool_resume_state取回存活状态直接复用若嵌套结果存在中断则用内部_nested_approvals_status汇总审批状态——pending时直接返回原结果不再重复运行approved/rejected时调用record_agent_tool_resume_state建立检查点src/agents/agent.py执行嵌套运行src/agents/agent.pyRunner.run_streamed有on_stream时或Runner.run恢复模式下以resume_state作为输入记录嵌套结果src/agents/agent.py嵌套运行存在中断且位于ToolContext内时record_agent_tool_run_result(context.tool_call, run_result, scope_idtool_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 典型场景多智能体编排中的翻译/路由 Agentexamples/agent_patterns/agents_as_tools.py 展示了标准的 agents-as-tools 用法编排 Agent 持有三个翻译子 Agent 的as_tool()工具外层运行时会按上文机制逐个触发嵌套运行并暂存结果orchestrator_agent Agent( nameorchestrator_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_nametranslate_to_spanish, ...), french_agent.as_tool(tool_nametranslate_to_french, ...), italian_agent.as_tool(tool_nametranslate_to_italian, ...), ], )【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考