OpenAI Agents SDK 工具执行生命周期深度解析:规划、审批、守卫、并发与取消的完整链路
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
工具调用(Function Tool / Tool Calling)是 Agent 与外部世界交互的核心通道,也是多 Agent 工作流中最容易出错、最难调试的环节:同一个call_id被模型重复使用、工具在审批等待期间被禁用、多个并行工具中一个失败导致其他成功结果被吞掉、父任务取消后遗留的后台任务泄漏事件循环……这些问题一旦出现,往往表现为"偶发性"的行为错乱,极难复现和定位。
本文基于 openai-agents-python 仓库的.agents/references/tool-execution-lifecycle.md(工具执行生命周期参考文档),结合 src/agents/run_internal/tool_planning.py、src/agents/run_internal/tool_execution.py、src/agents/tool.py 等源码,系统拆解一个函数工具从"被模型发现"到"输出进入运行状态"的完整生命周期:规划(Planning)→ 审批(Approval)→ 守卫(Guardrails)→ 并发执行(Concurrency)→ 取消/失败(Cancellation & Failure)→ 钩子与追踪(Hooks & Tracing)→ 工具选择复位(Tool Choice Reset)。读完本文,你将能够理解该框架在工具执行边界上的全部设计决策,并掌握在二次开发、插件扩展或排查线上问题时应该遵守的契约约束。
一、总览:工具执行的生命周期全景
在 openai-agents-python 中,一次完整的函数工具执行可以被划分为六个阶段,每一阶段都有独立的职责与归属模块:
- 发现(Discovery):
process_model_response()解析模型输出,识别出可执行的工作(函数调用、handoff、computer action、shell call、apply_patch、custom tool 等)。 - 规划(Planning):src/agents/run_internal/tool_planning.py 中的
ToolExecutionPlan决定"哪些工作现在可以运行",并对新鲜轮次(fresh turn)与恢复轮次(resume turn)分别构建不同的执行计划。 - 审批分区(Approval Partitioning):按
call_id和工具身份解析审批状态,将工具运行划分为"已批准(approved)"、"待中断(pending interruption)"与"被拒绝(rejected)"三组。 - 执行(Invocation):src/agents/run_internal/tool_execution.py 中的
_FunctionToolBatchExecutor以有界并发方式实际运行工具处理器,并在此边界上执行输入守卫、输出守卫、钩子与追踪 span。 - 输出收束(Output Settlement):将各工具结果按模型原始顺序组装为
ToolCallOutputItem,提交为可接受的运行状态(accepted run state)或持久化会话历史。 - 复位与清理(Reset & Cleanup):
AgentToolUseTracker按 Agent 身份记录工具使用情况,在reset_tool_choice=True时复位下一轮的工具选择;同时保证每个启动的任务与每次运行持有的资源都到达确定性的终态。
参考文档特别强调了一条贯穿始终的原则:"先规划,后副作用"(Plan Before Side Effects)。发现(discovery)与规划(planning)必须分离,规划与调用(invocation)也必须分离;发现阶段找到的"可执行工作"绝不等于"现在就应该执行的工作"。这一分离使得审批、守卫、去重、并发控制全部可以插在中间层,而不需要侵入工具处理器本身。
二、规划层:ToolExecutionPlan与新鲜/恢复轮次的差异化处理
2.1 规划层的数据结构
ToolExecutionPlan定义于 src/agents/run_internal/tool_planning.py,它表示"单个回合中需要执行的工具工作",字段包括:
| 字段 | 类型 | 说明 |
|---|---|---|
function_runs | list[ToolRunFunction] | 常规函数工具调用 |
computer_actions | list[ToolRunComputerAction] | 计算机操作(截图、点击等) |
custom_tool_calls | list[ToolRunCustom] | 自定义工具调用 |
shell_calls | list[ToolRunShellCall] | shell 工具调用 |
apply_patch_calls | list[ToolRunApplyPatchCall] | apply_patch 编辑操作 |
local_shell_calls | list[ToolRunLocalShellCall] | 本地 shell 调用 |
pending_interruptions | list[ToolApprovalItem] | 待人工审批的中断项 |
approved_mcp_responses | list[RunItem] | 已批准(或自动批准)的 MCP 审批响应 |
mcp_requests_with_callback | list[ToolRunMCPApprovalRequest] | 走回调的托管 MCP 审批请求 |
2.2 新鲜轮次 vs 恢复轮次的规划差异
参考文档明确指出:"Fresh and resumed turns need different plans"(新鲜轮次与恢复轮次需要不同的计划)。源码中对应两个构建函数:
_build_plan_for_fresh_turn()(tool_planning.py):从processed_response直接搬运全部工具运行与审批请求。_build_plan_for_resume_turn()(tool_planning.py):不重新发现工具,而是接收上一轮中断时保留下来的function_runs、computer_actions、shell_calls等参数,只执行"尚未解决或新批准"的工作;且恢复轮次的local_shell_calls固定为空([])。
这一差异的工程意义在于:恢复一个被中断的回合,绝不能重新发现并重新运行已经完成的调用。如果模型在第一次调用后收到审批中断,恢复时框架必须从RunState中还原之前的调用,而不是再次向模型要一次工具调用参数——否则可能造成同一个副作用被执行两次。
2.3 执行计划的分发
_execute_tool_plan()(tool_planning.py)是规划层的"总调度器"。它区分两种模式:
- 并行模式(
parallel=True,默认):通过gather_with_cancel同时启动execute_function_tool_calls、execute_computer_actions、execute_custom_tool_calls、execute_shell_calls、execute_apply_patch_calls、execute_local_shell_calls六个执行器,并传入一个共享的sibling_category_failure事件——当某一类工具(如 shell)失败时,其余类别可以感知并进入对应的收束路径。 - 串行模式(
parallel=False):按类别依次执行,便于在特定配置下获得确定性顺序。
gather_with_cancel定义于 src/agents/util/_asyncio_tasks.py,是框架内部对asyncio.gather的封装,增加了取消传播与失败仲裁能力。
三、审批分区:needs_approval的权威性与"不重复判定"约束
3.1 审批状态是权威的
参考文档的核心约束之一是:"Approval state is authoritative once resolved. Do not call a dynamicneeds_approvalchecker again for a call whose status is already approved or rejected."(审批状态一旦确定即为权威;对于已经批准或拒绝的调用,不得再次调用动态的needs_approval检查器。)
源码中,_collect_runs_by_approval()(tool_planning.py)精确实现了这一语义:
- 首先通过
context_wrapper.get_approval_status(tool_name, call_id, ...)查询持久化的审批状态; - 仅当状态为
None(从未判定)时,才调用needs_approval_checker(run); - 再次查询状态后,按
False(拒绝,构造 rejection item)→True(批准,加入 approved runs)→needs_approval=False(无需审批,直接放行)→ 否则(加入pending_interruptions)四条分支处理。
同样的逻辑在恢复路径_select_function_tool_runs_for_resume()(tool_planning.py)中再次出现:if approval_status is None:才调用needs_approval_checker,然后立刻重查状态。这意味着needs_approval是一个"一次性的、可变的"判定,而审批结论是"跨RunState恢复仍然有效"的持久化事实。
3.2 审批与拒绝消息的粘附性
"Persisted approval decisions and rejection messages must remain attached to the same tool identity and call ID acrossRunStateresume."(持久化的审批决策与拒绝消息必须始终粘附在同一个工具身份与 call ID 上。)
在_collect_runs_by_approval中,每个工具运行都被包装为ToolApprovalItem,其中携带了完整的身份信息:
tool_name:工具名;tool_namespace:通过get_tool_call_namespace()解析的命名空间(来自调用载荷);tool_origin:通过get_function_tool_origin()解析的工具来源;tool_lookup_key:通过get_function_tool_lookup_key_for_tool()解析的规范查找键。
get_approval_status正是以tool_name + call_id(并辅以 lookup key / namespace)为键查询状态的,从而保证审批结论在跨轮次恢复时不会错配到其他工具。
3.3 调用身份去重:重复的工具定义不是重复的调用
参考文档要求:"Deduplicate by invocation identity before execution while preserving model order... A repeated tool definition is not a repeated call, and a repeated call ID must not execute twice."(执行前按调用身份去重,同时保留模型顺序……重复的工具定义不是重复的调用,重复的 call ID 绝不能执行两次。)
_dedupe_processed_response_invocations()(tool_planning.py)实现了完整的去重矩阵:
- 历史完成调用索引:遍历
existing_items,用tool_output_identity构建"已完成输出键"集合,并用tool_invocation_identity构建"历史调用身份 → (类型, call_id, 指纹)"映射。 - 同响应内去重:
current_response_invocations记录本次响应中已出现的 call ID;若同一响应内模型复用了 call ID 但身份不同,抛出ModelBehaviorError("Model reused a tool call ID for a different invocation in one response.")。 - 跨响应历史去重:若某 call ID 在历史中已存在且身份不一致,抛出
ModelBehaviorError("Model reused a completed tool call ID for a different invocation.");若身份一致且filter_completed=True,则该调用被跳过(skipped_raw_item_ids.add(...))。 - 已执行未提交检测:
binding_status[2] and not binding_status[1]时抛出ModelBehaviorError("A tool call already executed, but its output was not committed."),防止"执行了但输出丢失"的状态被静默重试。
值得注意的是,_tool_call_identity()(tool_planning.py)在 call ID 缺失时回退到(call_id, name, hashable_args)三元组,其中参数通过_hashable_identity_value()稳定序列化(dict/list 用json.dumps(sort_keys=True)),从而在非 OpenAI 格式或不含 ID 的载荷下依然可以完成身份识别。
四、调用边界:to_thread、超时、失败转换与嵌套执行
4.1 同步函数通过asyncio.to_thread执行
参考文档:"Decorated synchronous Python functions run throughasyncio.to_thread()so they do not block the event loop. Async function tools run in the event loop and are the only decorated handlers that support SDK timeouts."(被装饰的同步 Python 函数通过asyncio.to_thread()运行,以免阻塞事件循环。异步函数工具在事件循环中运行,是唯一支持 SDK 超时的装饰处理器。)
在 src/agents/tool.py 中可以看到精确实现:
if not is_sync_function_tool: if schema.takes_context: result = await the_func(ctx, *args, **kwargs_dict) else: result = await the_func(*args, **kwargs_dict) else: if schema.takes_context: result = await asyncio.to_thread(the_func, ctx, *args, **kwargs_dict) else: result = await asyncio.to_thread(the_func, *args, **kwargs_dict)这一设计直接决定了超时能力的边界:asyncio.to_thread中的线程任务无法被事件循环的asyncio.wait_for可靠取消(线程不会被中断),因此只有 async 工具处理器才能使用 SDK 的超时机制。若你需要在同步工具上实施超时,必须自行在处理器内部实现(例如内部再次提交到线程池并等待)。
4.2 超时与普通异常是两套独立策略
参考文档:"timeout_behaviorandtimeout_error_functionown timeout conversion;failure_error_function=Nonemeans ordinary exceptions propagate instead of becoming model-visible output."(timeout_behavior与timeout_error_function负责超时转换;failure_error_function=None意味着普通异常直接传播,而不会变成模型可见的输出。)
在 src/agents/tool.py 附近的超时分支中,timeout_behavior支持两种取值:
"error_as_result"(默认):超时被转换为模型可见的错误文本;若提供了timeout_error_function,则用它格式化超时消息;"raise_exception":超时作为异常直接抛出。
而failure_error_function是独立的一层:它仅用于"把工具执行失败转换为模型可见输出"的场景(配合@function_tool装饰器的failure_error_function参数,见 tool.py 与set_function_tool_failure_error_function/resolve_function_tool_failure_error_function)。若其值为None,普通异常将向上传播到运行层,而不是伪装成工具输出回传给模型——这保证了"错误不会被静默吞掉"。
4.3 每次调用恰好一个 span 与一对钩子
参考文档:"Tool start/end hooks and function spans surround the actual invocation once per call, including failure and cancellation paths. Do not emit a successful end state before output guardrails complete."(工具开始/结束钩子与函数 span 每次调用恰好包裹一次真实调用,包括失败与取消路径;在输出守卫完成之前,不得发出成功的结束状态。)
在_FunctionToolBatchExecutor._run_single_tool()(tool_execution.py)中,每次工具调用都在一个with function_span(trace_tool_name) as span_fn:块内执行;异常路径通过_error_tracing.attach_error_to_current_span(...)给 span 打上SpanError,其中trace_include_sensitive_data配置决定输入输出是否写入 span(默认脱敏)。随后在_execute_single_tool_body()中,hooks.on_tool_start(运行级钩子)与agent_hooks.on_tool_start(Agent 级钩子)通过gather_with_cancel并行触发——注意这里并不存在对称的on_tool_end钩子,钩子的收尾职责由 span 上下文管理器与结果提交共同完成。
4.4 嵌套Agent.as_tool()拥有独立的嵌套运行状态
"嵌套Agent.as_tool()执行拥有一个嵌套运行循环与嵌套的可恢复状态。嵌套状态必须按父RunState与调用身份进行缓存,而不是仅按可复用的 Agent 或工具对象缓存。"源码中通过get_agent_tool_state_scope(context_wrapper)生成tool_state_scope_id,并配合peek_agent_tool_run_result/consume_agent_tool_run_result(来自 src/agents/agent_tool_state.py)在嵌套中断后恢复内部 Agent 的运行。这一点对于"工具内部再跑一个 Agent"的场景(如research_bot示例中的子 Agent)至关重要:多个并行的嵌套运行共享同一个外层 Agent 对象,但状态必须按调用隔离。
4.5 每运行资源(run-scoped resources)的初始化与释放
"Per-run resources such as resolvedComputerimplementations must be initialized and disposed by the run that acquired them."(每次运行持有的资源——例如已解析的Computer实现——必须由获取它的那次运行负责初始化与释放。)
initialize_computer_tools()(tool_execution.py)演示了该模式的实现:对每个ComputerTool,调用resolve_computer(tool=tool, run_context=context_wrapper)解析出Computer实例,并通过dataclasses.replace(tool, computer=resolved_computer)生成运行局部的工具副本,而不是修改 Agent 上的共享工具对象。这样每个并发运行都持有独立的计算机实现,不会相互污染。
五、并发与失败语义:有界并发、顺序保持与确定性仲裁
5.1 SDK 侧并发 ≠ 提供方并发
参考文档给出了一个关键澄清:"SDK-side function-tool concurrency is independent of provider-side parallel tool-call generation. The provider controls how many calls appear in one response;RunConfig.tool_execution.max_function_tool_concurrencycontrols how many local function handlers run at once."(SDK 侧的函数工具并发独立于提供方侧的并行工具调用生成。提供方决定一次响应中出现多少个调用;RunConfig.tool_execution.max_function_tool_concurrency决定本地函数处理器同时运行多少个。)
两个配置项分别在两层:
- 提供方层:模型请求中的
parallel_tool_calls等参数,决定模型一次性发出多少工具调用; - SDK 层:src/agents/run_config.py 中的
ToolExecutionConfig.max_function_tool_concurrency(默认None,即启动本轮全部调用)限制本地处理器同时执行的个数;设为1即完全串行。__post_init__校验该值必须>= 1。
SDK 层的并发由_FunctionToolBatchExecutor._fill_tool_task_slots()(tool_execution.py)实现:它是一个"插槽"调度器——当有任务完成时(_drain_pending_tasks中asyncio.wait(FIRST_COMPLETED)返回),再从未处理的pending_tool_runs中按顺序取出新任务填充空位,形成有界并发的工作队列。
5.2 输出顺序保持模型顺序
"Preserve model order in emitted outputs even when handlers complete out of order."(即使处理器乱序完成,发出的输出也必须保持模型顺序。)
_FunctionToolTaskState中保存了order(枚举顺序);_record_completed_function_tool_tasks()(tool_execution.py)在记录结果时按order排序(sorted(completed_tasks, key=lambda task: task_states[task].order));最终_build_function_tool_results()会按工具运行顺序重新组装FunctionToolResult列表,确保喂给模型的输出与模型发出调用的顺序一致——这直接避免了"模型按 A、B、C 顺序调用,结果却按 C、A、B 顺序返回"造成的推理错乱。
5.3 兄弟隔离:一个失败不吞掉成功的输出
"Isolate sibling results. A cancelled or failed call must not discard outputs already produced by successful siblings."(隔离兄弟结果。被取消或失败的调用不得丢弃已经由成功兄弟产生的输出。)
该语义由_raise_failure_after_draining_siblings()(tool_execution.py)实现,其流程是:
- 将待处理任务分为"可取消任务"与"post-invoke 阶段任务"两类(
_partition_pending_tasks); - 取消可取消的兄弟任务,并在有界窗口内**排干(drain)**它们——即等待它们完成必要的自我推进步骤,避免强行取消留下脏状态(
_drain_cancelled_function_tool_tasks,超时_FUNCTION_TOOL_CANCELLED_DRAIN_SECONDS = 0.25秒,立即推进步数上限_FUNCTION_TOOL_CANCELLED_IMMEDIATE_STEP_LIMIT = 64); - 等待 post-invoke 任务短暂收尾(
_wait_pending_function_tool_tasks_for_timeout,_FUNCTION_TOOL_POST_INVOKE_WAIT_SECONDS = 0.1秒); - 通过
_merge_late_function_tool_failure把"迟到的失败"合并进主失败,但不掩盖根因; - 对仍然存活的取消任务与 post-invoke 任务附加
add_done_callback报告器(_attach_function_tool_task_result_callbacks),把迟到的异常交给事件循环异常处理器,而不是无限等待。
重要语义:isolate_parallel_failures的取值决定失败是否隔离。在_execute_tool_plan中,当函数工具多于一个、或并行执行且存在其他类别工具时,isolate_function_tool_failures=True——即兄弟失败不再中断已成功的调用,而是走上述排干路径。这保证了多工具并行场景下部分成功的输出能够保留。
5.4 工具本地取消 vs 父运行取消
"Distinguish cancellation of one tool handler from cancellation of the parent run. Tool-local cancellation can follow the configured tool failure policy; parent cancellation must propagate promptly."(区分单个工具处理器的取消与父运行的取消。工具本地取消可以遵循配置的工具失败策略;父运行取消必须立即传播。)
在execute()的except asyncio.CancelledError分支(tool_execution.py)中:
- 若取消来自兄弟类别失败(
sibling_category_failure.is_set()),走_drain_pending_tasks_for_sibling_category_failure()——温和排干; - 否则视为父取消,走
_cancel_pending_tasks_for_parent_cancellation()——立即取消所有剩余任务并附加报告回调,绝不让工具输出冒充父取消的结果。
而在单任务内部(_invoke_tool_and_run_post_invoke,tool_execution.py),捕获到asyncio.CancelledError后,若该任务不在teardown_cancelled_tasks集合中(说明是工具自身被取消而非父取消的清理阶段),会尝试调用maybe_invoke_function_tool_failure_error_function将取消转换为模型可见的失败输出;转换失败才重新抛出。
5.5 确定性失败仲裁
"Select and raise failures deterministically when several tasks fail, while still observing secondary failures."(当多个任务失败时,确定性地选择并抛出失败,同时仍然观察次要失败。)
_get_function_tool_failure_priority()(tool_execution.py)定义了失败优先级:
asyncio.CancelledError→ 优先级 0(最低);- 普通
Exception→ 优先级 1; - 其他
BaseException(如KeyboardInterrupt、SystemExit)→ 优先级 2(最高)。
_select_function_tool_failure()(tool_execution.py)在优先级相同时按order打破平局——取调用顺序更靠前的失败。由于order来自模型调用顺序而非任务完成顺序,因此无论任务集合的迭代顺序如何、任务是否被急于执行,最终抛出的失败都是确定性的。
六、守卫(Guardrails)的执行时机与顺序
6.1 输入守卫的"两次运行"设计
参考文档的核心点:"Pre-approval input guardrails are an early rejection optimization. They may run before an approval interruption, but input guardrails must run again immediately before invocation because state, policy, or arguments may have changed while approval was pending."(审批前输入守卫是一种早期拒绝优化。它们可以在审批中断之前运行,但输入守卫必须在调用之前再次运行,因为审批等待期间状态、策略或参数可能已经改变。)
这对应ToolExecutionConfig.pre_approval_tool_input_guardrails(默认False,见 run_config.py)开关。当开启时,_maybe_execute_tool_approval()(tool_execution.py)会在审批状态为None且需要审批时,先跑一次输入守卫做早期拒绝;但无论是否提前跑过,_execute_single_tool_body()在真实调用前都会再次执行_execute_tool_input_guardrails()——这就是"Rechecking guardrails does not mean rechecking approval"(重查守卫不意味着重查审批)的含义。
6.2 输入/输出守卫的完成顺序
- 工具输入守卫(
_execute_tool_input_guardrails,tool_execution.py):在本地副作用之前完成。守卫输出支持三种行为:raise_exception→ 抛ToolInputGuardrailTripwireTriggered;reject_content→ 返回拒绝消息,由调用方转为FunctionToolResult(拒绝项);allow→ 放行。
- 工具输出守卫(
_execute_tool_output_guardrails,tool_execution.py):在输出成为可接受的运行状态、模型输入或持久化会话历史之前完成;拒绝时以is_rejection=True标记替换输出为拒绝消息。
FunctionTool通过tool_input_guardrails与tool_output_guardrails属性挂载守卫,测试覆盖见 tests/test_tool_guardrails.py(输入/输出、同步/异步、三种行为、装饰器形式、tripwire 保留已完成回合结果等)。
6.3 守卫管道的适用范围边界
"Tool guardrail pipeline applies toFunctionToolinvocation. Handoffs, hosted tools, built-in provider tools, and nestedAgent.as_tool()runs have separate execution boundaries unless they explicitly opt into equivalent checks."(工具守卫管道仅适用于FunctionTool调用。Handoff、托管工具、内置提供方工具与嵌套的Agent.as_tool()运行拥有独立的执行边界,除非它们显式选择加入等效检查。)
这意味着:如果你的安全策略依赖工具守卫,必须明确它只保护本地的FunctionTool调用;handoff 与嵌套 Agent 的安全边界需要通过各自机制(如 handoff 过滤、子 Agent 自己的守卫)另行配置。
七、MCP 审批请求的特殊生命周期
参考文档没有展开 MCP 细节,但规划层对 MCP 审批请求有完整的处理路径(这是理解"托管工具"执行边界的重要补充):
_partition_mcp_approval_requests()(tool_planning.py)将 MCP 审批请求分为两类:带on_approval_request回调的(回调处理)与需要人工审批的(manual bucket)。_preflight_mcp_approval_requests()(tool_planning.py)对同 ID 的兄弟请求做"变化检测":若同一 call ID 被用于不同调用,抛出ModelBehaviorError;完全重复的请求则合并。execute_mcp_approval_requests()(tool_planning.py)执行回调:先查询持久化审批状态,未判定时标记_mark_tool_invocation_executed防止重复回调,然后调用on_approval_request,最后将结果写入approve/reject状态并构造MCPApprovalResponseItem。若回调已运行但响应未提交,会抛出ModelBehaviorError("A Hosted MCP approval callback already ran, but its response was not committed.")。
八、工具选择复位(Tool Choice Reset)与AgentToolUseTracker
参考文档:"AgentToolUseTrackerrecords tool use per agent identity. Whenreset_tool_choice=True, reset the effective next-turn tool choice after that agent uses a tool sorequiredor a named choice cannot force an accidental loop."(AgentToolUseTracker按 Agent 身份记录工具使用。当reset_tool_choice=True时,在该 Agent 使用工具后复位下一轮的有效工具选择,避免required或具名选择强制造成意外循环。)
maybe_reset_tool_choice()(tool_execution.py)实现如下:
def maybe_reset_tool_choice(agent, tool_use_tracker, model_settings): if agent.reset_tool_choice is True and tool_use_tracker.has_used_tools(agent): return dataclasses.replace(model_settings, tool_choice=None) return model_settings即:仅当agent.reset_tool_choice is True且该 Agent 已使用过工具时,才将下一轮的tool_choice复位为None;否则原样返回。注意它通过dataclasses.replace生成新的ModelSettings,不会修改 Agent 声明的配置——"do not mutate the agent's declared settings across independent runs"。
AgentToolUseTracker(src/agents/run_internal/tool_use_tracker.py)按 Agent 身份(通过_build_agent_identity_keys_by_id等工具处理重复 Agent 名)记录HandoffCallItem、ToolCallItem、ToolCallOutputItem等工具事件,并支持serialize_tool_use_tracker/hydrate_tool_use_tracker序列化——这正是参考文档要求的"跨中断与沙箱恢复时持久化并恢复工具使用追踪器,包括存在重复 Agent 名的图"。
测试见 tests/test_tool_choice_reset.py:覆盖tool_choice="required"单工具/多工具、具名工具foo_bar、多轮运行不陷入无限循环等场景(maybe_reset_tool_choice仅在 has_used_tools 且 reset 开关开启时复位)。
九、恢复路径(Resume)的契约:已批准工具的再执行
execute_approved_tools()(tool_execution.py)是 HITL(Human-in-the-loop)恢复路径的核心:当审批中断被用户批准后,它根据中断项集合重新构造ToolRunFunction并执行。其健壮性校验包括:
- 缺少工具名 → 追加错误输出;
- 缺少 call ID → 追加错误输出(并尽力解析工具来源用于追踪);
- 审批状态为
False→ 追加拒绝消息(resolve_approval_rejection_message或默认REJECTION_MESSAGE); - 审批状态不为
True→ "Tool approval status unclear"; - 工具未找到、非函数工具、raw_item 类型非法 → 分别追加错误输出。
最终构造出的tool_runs再次交给execute_function_tool_calls()走完整执行管道(含守卫、钩子、span)。这一路径印证了"审批一旦批准,执行仍要走完整生命周期"的设计——批准只解决"能不能执行"的问题,不跳过执行期校验。
十、审查清单:修改工具执行行为前必须验证的五个维度
参考文档最后给出了变更审查清单(Review Checklist),任何涉及工具规划、审批、守卫、并发、取消、超时、钩子、错误转换或恢复执行的行为变更,都应逐项验证:
- 路径追踪:分别追踪新鲜执行、审批中断、审批拒绝、序列化恢复四条路径;
- 顺序验证:验证守卫、审批、钩子、追踪 span、调用、输出、持久化之间的先后顺序;
- 并发与取消路径测试:覆盖顺序执行、有界并发、兄弟失败、工具本地取消、父取消五类路径;
- 失败转换测试:覆盖默认、自定义、禁用三种失败转换,以及适用的超时行为;
- 终态确定性:确认每个启动的任务与每次运行持有的资源都到达确定性终态(无泄漏、无悬挂任务)。
对应源码中的测试锚点包括 tests/test_agent_runner.py、tests/test_agent_runner_streamed.py、tests/test_function_tool.py、tests/test_tool_guardrails.py、tests/test_tool_choice_reset.py、tests/test_tool_use_tracker.py 与 tests/test_run_state.py。
十一、实践建议汇总
基于以上生命周期分析,针对二次开发与问题排查给出以下可操作建议:
- 编写新的工具执行路径时:始终把"发现(discovery)"与"规划(planning)"分离;不要让模型输出直接驱动副作用,先经过
ToolExecutionPlan的审批分区与去重。 - 使用审批(HITL)时:不要在
needs_approval回调中依赖"审批前的状态快照"做最终决策;审批恢复后输入守卫会重新执行,参数与策略可能已变化,应把输入守卫当作最终防线。 - 使用并发工具时:
RunConfig(tool_execution=ToolExecutionConfig(max_function_tool_concurrency=N))控制 SDK 侧并发;若要强顺序副作用,设置为1。注意pre_approval_tool_input_guardrails=True只是早期拒绝优化,不减少审批后的最终校验。 - 为同步工具实现超时:SDK 超时仅适用于 async 工具;同步工具请自行在处理器内部实现超时控制。
- 避免 call ID 复用:自定义模型或提供方接入时,务必保证每个工具调用使用唯一 call ID,否则将触发
ModelBehaviorError去重保护。 - 多 Agent 图 + 工具选择:启用
reset_tool_choice=True防止required循环;跨会话恢复时依赖 tracker 的序列化能力保持工具选择行为一致。
参考资料(仓库内)
- 生命周期参考:.agents/references/tool-execution-lifecycle.md
- 规划实现:src/agents/run_internal/tool_planning.py
- 执行实现:src/agents/run_internal/tool_execution.py
- 执行配置:src/agents/run_config.py
- 工具定义与调用封装:src/agents/tool.py
- 工具使用追踪:src/agents/run_internal/tool_use_tracker.py
- 官方文档:
docs/running_agents.md、docs/tools.md、docs/guardrails.md、docs/human_in_the_loop.md - 测试锚点:tests/test_agent_runner.py、tests/test_function_tool.py、tests/test_tool_guardrails.py、tests/test_tool_choice_reset.py、tests/test_tool_use_tracker.py、tests/test_run_state.py
【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考