openai-agents-python 人工介入HITL工具审批实战interruptions 中断、RunState 持久化与审批/拒绝流程【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python在openai-agents-python中Human-in-the-LoopHITL流程允许你在模型发起敏感工具调用时暂停 Agent 执行等待人工批准或拒绝后再继续。本文以仓库文档 docs/ko/human_in_the_loop.md 为主体结合src/agents下的源码与examples/中的示例脚本完整讲清三件事如何用needs_approval声明哪些工具需要审批、如何通过RunResult.interruptionsRunState完成暂停—持久化—审批—恢复的闭环以及如何用回调实现不暂停的编程式审批。读完本文你可以为任意 Agent 应用落地一套可跨进程、可跨机器等待人工决策的审批基础设施。1. HITL 的总体模型审批作用于整个运行而非单个 AgentHITL 的核心由三个概念组成工具声明审批需求工具通过needs_approval或 MCP 场景的require_approval声明何时需要人工批准结果暴露中断interruptions运行结果RunResult/RunResultStreaming把待审批的工具调用以interruptions列表暴露出来RunState支持暂停与恢复RunState可把暂停中的运行序列化在决策做出后可能是几秒后也可能是几天后反序列化并继续。需要特别注意的一点是该审批界面是整个运行级别的而不局限于当前顶层 Agent。无论工具属于当前 Agent、属于通过 handoff 到达的另一个 Agent、还是嵌套的Agent.as_tool()执行都适用同一套模式。对于嵌套的Agent.as_tool()场景中断依然暴露在外层运行上因此你在外层RunState上调用approve/reject然后恢复的是原来的顶层运行。源码中这一外层转发到嵌套状态的路由逻辑可以直接看到RunState.approve 会先通过_find_nested_approval_state判断该审批项是否属于某个嵌套的 agent-as-tool 运行是则转发给嵌套的RunState否则写入当前运行的_context.approve_tool。此外Agent.as_tool()的审批可能发生在两个层级作为工具的 Agent 本身通过Agent.as_tool(..., needs_approval...)要求审批委托前先审嵌套运行启动后嵌套 Agent 内部的工具再各自发起审批。两者都经由同一个外层运行的中断流程处理。本文聚焦于通过interruptions的手动审批流程如果你的应用可以在代码里直接做决定部分工具类型还支持编程式审批回调让运行不暂停继续见第 5 节。2. 标记需要审批的工具needs_approval2.1 两种声明方式布尔值或按调用判定的函数将needs_approval设为True表示总是要求审批提供一个可调用对象函数则可以按每次调用动态决定。该可调用对象接收三个参数运行上下文run context、解析后的工具参数字典、以及工具调用 IDfrom agents import Agent from agents.decorators import tool tool(needs_approvalTrue) async def cancel_order(order_id: int) - str: return fCancelled order {order_id} async def requires_review(_ctx, params, _call_id) - bool: return refund in params.get(subject, ).lower() tool(needs_approvalrequires_review) async def send_email(subject: str, body: str) - str: return fSent {subject} agent Agent( nameSupport agent, instructionsHandle tickets and ask for approval when needed., tools[cancel_order, send_email], )2.2 fail-closed无法安全解析参数时一律转人工文档特别强调当 SDK 无法安全地检查参数时可调用审批规则默认要求审批fail closed。具体触发条件包括参数是非法 JSON参数是合法 JSON 但不是对象例如null或列表参数包含NaN、Infinity、-Infinity这类非标准 JSON 常量。此时可调用对象根本不会被调用该次调用直接要求手动审批。这一行为在 Runner 与 Realtime 工具调用中一致。从源码看这一保证由 src/agents/util/_approvals.py 的parse_function_tool_arguments实现它使用json.loads并传入parse_constant_reject_nonstandard_json_constant拒绝非常量任何ValueError或解析结果不是 dict都会返回None表示审批策略无法检查参数随后 evaluate_needs_approval_setting 统一处理bool与可调用含 awaitable两种形态非法类型会抛出UserError。这个模块的注释也说明它被刻意放在util下以便run_internal与realtime两个包共用同一套审批求值逻辑。2.3 各工具类型的审批支持矩阵文档列出的needs_approval/require_approval支持范围如下工具/服务器类型审批机制备注function_tool含tool装饰器needs_approval走手动 interruption 流程Agent.as_toolneeds_approval中断暴露在外层运行ShellTool/ApplyPatchTool本地needs_approval 可选on_approval回调回调可跳过中断直接批准/拒绝托管hostedShell 环境不支持源码在构造时强制将两者清零见下本地 MCP 服务器MCPServerStdio/MCPServerSse/MCPServerStreamableHttprequire_approval对 MCP 工具调用做门禁托管 MCPHostedMCPTooltool_config{require_approval: always} 可选on_approval_requestnever用于可信服务器关于托管 Shell 环境不支持审批这一点src/agents/tool.py 中有硬性约束当ShellTool使用托管环境且设置了needs_approval或on_approval时会抛出UserErrorShellTool with hosted environment does not support needs_approval or on_approval.并在初始化时把两者重置为False/None。3. 审批流程的完整工作机制文档给出的标准流程分为五步模型发出工具调用后Runner 评估其审批规则needs_approval、require_approval或托管 MCP 的等价物如果该工具调用的审批决定已经存储在RunContextWrapper中Runner 直接继续、不再询问。按调用的审批只作用于特定调用 ID若希望同一运行内后续对该同一工具身份的调用都沿用此决定需传always_approveTrue或always_rejectTrue若规则要求审批且没有已存决定则执行暂停RunResult.interruptions或RunResultStreaming.interruptions中包含ToolApprovalItem条目携带agent.name、tool_name、arguments等细节。handoff 之后或嵌套Agent.as_tool()执行中产生的审批也包含在内用result.to_state()把结果转为RunState调用state.approve(...)或state.reject(...)然后用Runner.run(agent, state)或Runner.run_streamed(agent, state)恢复其中agent必须是该运行的原始顶层 Agent恢复后的运行从中断点继续如果又出现新的审批需求则重新进入此流程。3.1 ToolApprovalItem中断条目的数据结构ToolApprovalItem定义在 src/agents/items.py。它的raw_item字段是一个联合类型ToolApprovalRawItem可承载函数工具调用、自定义工具调用、Shell 工具调用、MCP 调用、MCP 审批请求、本地 Shell 调用乃至普通 dict——这也是为什么一个interruptions列表可以同时混合常规函数工具、托管 MCP 审批和嵌套Agent.as_tool()审批。tool_name未显式给出时会在__post_init__中从raw_item回退提取并顺带推导tool_namespace与tool_lookup_key规范化的函数工具查找键。3.2 粘性决定sticky decisions与跨进程恢复用always_approveTrue/always_rejectTrue创建的持续性决定会存入运行状态因此当你稍后恢复同一个暂停中的运行时它们能经受住state.to_string()/RunState.from_string(...)以及state.to_json()/RunState.from_json(...)的往返。源码侧的佐证是 RunState._serialize_approvals序列化时会把上下文中每个工具的审批记录approved/rejected列表、rejection_messages、sticky_rejection_message、sticky_scope写入 JSON 友好的映射。针对HostedMCPTool的审批请求SDK 用server_label 工具名的组合来识别一个粘性工具决定——即对服务器 A 上lookup_account的 always-approve不会批准服务器 B 上同名工具。并且只有当托管 MCP 审批请求同时包含两个非空身份字段时SDK 才会持久化 always-approve / always-reject 决定。3.3 可以分批处理待审批项你不必在一轮内处理完所有待审批项。interruptions里可能是常规函数工具、托管 MCP 审批与嵌套Agent.as_tool()审批的混合体只批准/拒绝其中一部分后重新运行已解决的调用会继续推进未解决的留在interruptions中使运行再次暂停——这正适合多个敏感操作排队人工逐条过的运维场景。4. 自定义拒绝消息被拒绝的工具调用默认会把 SDK 的标准拒绝文本送回运行内模型能看到。文档支持两个层级的自定义运行级兜底设置RunConfig.tool_error_formatter控制整个运行内审批被拒时模型看到的默认消息按调用覆盖在state.reject(...)上传rejection_message...让某一次被拒的工具调用返回不同消息。两者同时提供时按调用的rejection_message优先于运行级 formatter。RunState.reject 的签名即reject(approval_item, always_rejectFalse, *, rejection_messageNone)其 docstring 明确写道提供rejection_message时该文本会在运行恢复时原样发给模型否则回退到运行级 tool error formatter 或 SDK 默认消息。from agents import RunConfig, ToolErrorFormatterArgs def format_rejection(args: ToolErrorFormatterArgs[None]) - str | None: if args.kind ! approval_rejected: return None return Publish action was canceled because approval was rejected. run_config RunConfig(tool_error_formatterformat_rejection) # Later, while resolving a specific interruption: state.reject( interruption, rejection_messagePublish action was canceled because the reviewer denied approval., )两层结合使用的完整示例见 examples/agent_patterns/human_in_the_loop_custom_rejection.py。5. 编程式自动审批on_approval 与 on_approval_request手动interruptions是最通用的模式但不是唯一模式。文档列出的自动决策能力本地ShellTool与ApplyPatchTool可通过on_approval在代码中立即批准或拒绝不暴露中断、不等待人HostedMCPTool可把tool_config{require_approval: always}与on_approval_request一起使用实现同类的编程式决定普通function_tool与Agent.as_tool()则只能走本文的手动中断流程。当这些回调返回决定时运行不再暂停等待人工响应。ShellTool的needs_approvalbool | ShellApprovalFunction与on_approval字段定义见 src/agents/tool.py。相关示例自动决策参考 examples/tools/shell.py手动中断处理参考 examples/tools/shell_human_in_the_loop.py。Realtime 与语音会话 API 的审批流程则参考 Realtime 指南。6. 流式运行与 Session 的兼容同一套中断流程在流式运行中同样可用流式运行暂停后请继续消费RunResultStreaming.stream_events()直到迭代器结束然后检查RunResultStreaming.interruptions、处理审批若希望恢复后的输出继续保持流式就用Runner.run_streamed(...)恢复。流式版本的完整模式可参考 Streaming 文档。如果同时使用 Session从RunState恢复时要继续传入同一个 Session 实例或传入配置为相同 session ID 与相同后端存储的另一个 Session 对象这样恢复的轮次会追加到同一份已保存的对话历史中。Session 生命周期细节见 Sessions 文档。7. 完整示例暂停、持久化、审批、恢复下面的示例展示了暂停—落盘—重载—收决策—恢复的完整闭环暂停时工具需要审批、状态写入磁盘、之后重新加载并恢复import asyncio import json from pathlib import Path from agents import Agent, Runner, RunState from agents.decorators import tool async def needs_oakland_approval(_ctx, params, _call_id) - bool: return Oakland in params.get(city, ) tool(needs_approvalneeds_oakland_approval) async def get_temperature(city: str) - str: return fThe temperature in {city} is 20° Celsius agent Agent( nameWeather assistant, instructionsAnswer weather questions with the provided tools., tools[get_temperature], ) STATE_PATH Path(.cache/hitl_state.json) def prompt_approval(tool_name: str, arguments: str | None) - bool: answer input(fApprove {tool_name} with {arguments}? [y/N]: ).strip().lower() return answer in {y, yes} async def main() - None: result await Runner.run(agent, What is the temperature in Oakland?) while result.interruptions: # Persist the paused state. state result.to_state() STATE_PATH.parent.mkdir(parentsTrue, exist_okTrue) STATE_PATH.write_text(state.to_string()) # Load the state later (could be a different process). stored json.loads(STATE_PATH.read_text()) state await RunState.from_json(agent, stored) for interruption in result.interruptions: approved await asyncio.get_running_loop().run_in_executor( None, prompt_approval, interruption.name or unknown_tool, interruption.arguments ) if approved: state.approve(interruption, always_approveFalse) else: state.reject(interruption) result await Runner.run(agent, state) print(result.final_output) if __name__ __main__: asyncio.run(main())实现要点while result.interruptions循环是关键恢复后模型可能继续发起新的敏感调用产生新的中断所以必须循环直到interruptions为空。仓库内 examples/agent_patterns/human_in_the_loop.py 采用同样的while has_interruptions结构每次循环result.to_state()、逐个state.approve(interruption)/state.reject(...)然后用恢复后的结果重新判断是否还有中断示例中prompt_approval是同步函数用input()通过run_in_executor(...)放入线程池执行如果你的审批来源本身是异步的HTTP 请求、异步数据库查询直接写成async def并await即可状态经state.to_string()落盘、再由RunState.from_json(agent, ...)重建因此收集决策这一步可以在另一个进程甚至另一台机器上完成这正是第 8 节长时审批的基础。如果运行可能暂停等待审批又需要流式输出调用Runner.run_streamed把result.stream_events()消费到结束然后执行与上面完全相同的result.to_state()与恢复步骤。8. 长时审批RunState 序列化选项与版本管理RunState被设计为可持久durable。用state.to_json()或state.to_string()把待办工作存入数据库或消息队列之后用RunState.from_json(...)或RunState.from_string(...)重建。有用的序列化选项选项作用context_serializer自定义非映射non-mapping上下文对象的序列化方式context_deserializer用RunState.from_json(...)/from_string(...)加载状态时重建非映射上下文对象strict_contextTrue上下文不是映射且未提供context_serializer时序列化失败不是映射且未提供context_deserializer时反序列化失败context_override加载状态时替换已序列化的上下文适合不想还原原上下文对象的场景注意它不会把该上下文从已序列化载荷中删除include_tracing_api_keyTrue需要恢复后的工作用相同凭据继续导出 trace 时把 tracing API key 写入序列化的 trace 载荷序列化后的运行状态包含你的应用上下文以及 SDK 管理的运行时元数据审批记录、用量usage、序列化的tool_input、嵌套 agent-as-tool 的恢复信息、trace 元数据、服务器管理的会话设置等。若计划存储或传输序列化状态请把RunContextWrapper.context当作持久化数据处理——除非你明确希望它们随状态一起流动否则不要把秘密信息放进去。待审批任务的版本管理如果审批可能长时间挂起请把 Agent 定义或 SDK 的版本标记与序列化状态一起存储。反序列化时再按版本路由到匹配的代码路径避免模型、提示词或工具定义变更后引发的不兼容。9. 仓库中的完整模式索引文档末尾汇总了仓库内各场景对应的可运行示例可作为进一步深入学习的入口场景示例/文档基础暂停—审批—恢复examples/agent_patterns/human_in_the_loop.py流式审批先抽干stream_events()再Runner.run_streamed(agent, state)恢复examples/agent_patterns/human_in_the_loop_stream.py运行级tool_error_formatter 按调用rejection_message组合examples/agent_patterns/human_in_the_loop_custom_rejection.pyAgent.as_tool(..., needs_approval...)嵌套中断仍暴露在外层运行恢复的是原顶层 Agent见本文第 1 节本地 Shell / apply_patchstate.approve(interruption, always_approveTrue)或state.reject(..., always_rejectTrue)缓存决定自动决策用on_approval手动决策处理中断examples/tools/shell.py、examples/tools/shell_human_in_the_loop.py本地 MCP 服务器门禁MCPServerStdio/MCPServerSse/MCPServerStreamableHttp的require_approvalexamples/mcp/get_all_mcp_tools_example/main.py、examples/mcp/tool_filter_example/main.py托管 MCPHostedMCPTool设tool_config{require_approval: always}强制 HITL可选on_approval_request自动批准/拒绝可信服务器用neverexamples/hosted_mcp/human_in_the_loop.py、examples/hosted_mcp/on_approval.py、examples/hosted_mcp/simple.pySession 与记忆把 session 传入Runner.run让审批与对话历史跨多轮保留SQLite / OpenAI Conversations 变体examples/memory/memory_session_hitl_example.py、examples/memory/openai_session_hitl_example.pyRealtimeRealtimeSession通过 WebSocket 的approve_tool_call/reject_tool_call消息审批或拒绝工具调用examples/realtime/app/server.py服务端处理器、Realtime 指南10. 小结把 HITL 落到生产的检查清单基于文档与源码落地一套生产级 HITL 审批时建议核对审批声明优先用可调用needs_approval做细粒度判断并理解其fail-closed语义参数无法安全解析时强制转人工审批循环写成while result.interruptions:支持部分处理与多轮再暂停需要跨调用沿用决定时用always_approve/always_reject并注意其粘性决定会随RunState序列化往返保留拒绝消息按需分层定制运行级RunConfig.tool_error_formatter兜底 按调用rejection_message覆盖能代码化决策的工具Shell / apply_patch / 托管 MCP改用on_approval/on_approval_request减少不必要的暂停状态落库时启用context_serializer/strict_context等选项并把context视为持久化数据避免泄露秘密审批可能长期挂起时随状态存储 Agent/SDK 版本标记反序列化时按版本路由规避工具与提示词变更带来的不兼容。【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考