企业官网建设流程全解析

openai-agents-python 人工介入（HITL）完整指南：暂停、审批、恢复与状态持久化

【免费下载链接】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-loop，HITL）机制展开，系统讲解如何让智能体在敏感工具调用前暂停执行、通过interruptions呈现待审批项、使用RunState序列化暂停状态并在决策后恢复运行。读完本文，你将掌握needs_approval/require_approval的声明方式、手动审批与程序化审批两条路径、自定义拒绝消息、流式与会话场景下的 HITL，以及面向生产环境的长期审批持久化方案。

什么是人工介入（HITL）

在真实业务中，智能体的部分工具调用是高风险的——例如取消订单、发送邮件、执行 Shell 命令、修改文件或调用 MCP 服务器上的敏感接口。人工介入（HITL）流程允许你在这类调用真正执行前暂停整个运行，等待一个人批准或拒绝后再继续。

核心设计有四点：

• 工具声明审批需求：工具通过needs_approval声明自己在什么情况下需要审批；
• 运行结果呈现中断项：运行暂停后，RunResult.interruptions（或RunResultStreaming.interruptions）中会包含ToolApprovalItem条目，携带agent.name、tool_name、arguments等详细信息；
• RunState支持持久化：暂停的运行可以序列化保存，跨进程、跨时间恢复；
• 恢复运行：做出批准或拒绝决策后，用Runner.run(agent, state)或Runner.run_streamed(agent, state)从暂停处继续。

值得强调的是：审批覆盖面是整个运行（run-wide），并不限于当前的顶层智能体。无论工具属于当前智能体、通过任务转移（handoff）到达的智能体，还是嵌套的Agent.as_tool()执行，都走同一套中断流程。在嵌套Agent.as_tool()的场景下，中断仍然呈现在外层运行中，因此你需要在外层RunState上批准或拒绝，然后恢复原始的顶层运行。

使用Agent.as_tool()时，审批可能发生在两个不同层级：智能体工具本身可以通过Agent.as_tool(..., needs_approval=...)要求审批；嵌套智能体内部的工具也可能在嵌套运行开始后发起自己的审批请求。两种情况都通过相同的外层运行中断流程处理。

标记需要审批的工具

静态标记：始终需要审批

将needs_approval=True传入工具装饰器，即可让该工具在每次被调用时都触发审批：

from agents import Agent from agents.decorators import tool @tool(needs_approval=True) async def cancel_order(order_id: int) -> str: return f"Cancelled order {order_id}"

动态规则：按调用内容决定

needs_approval也可以是一个异步函数，针对每次调用分别做出决策。该可调用对象接收三个参数：运行上下文、解析后的工具参数、工具调用 ID，返回bool表示本次调用是否需要审批：

async def requires_review(_ctx, params, _call_id) -> bool: return "refund" in params.get("subject", "").lower() @tool(needs_approval=requires_review) async def send_email(subject: str, body: str) -> str: return f"Sent '{subject}'" agent = Agent( name="Support agent", instructions="Handle tickets and ask for approval when needed.", tools=[cancel_order, send_email], )

从源码实现看，needs_approval的类型签名是bool | Callable[[RunContextWrapper[Any], dict[str, Any], str], Awaitable[bool]]（见 src/agents/tool.py），既接受布尔值，也接受返回协程的异步函数。

失败关闭（fail-closed）策略

当 SDK 无法安全地检查参数时，可调用审批规则会采取失败关闭策略：如果参数是格式错误的 JSON、是有效 JSON 但不是对象（例如null或列表），或包含NaN、Infinity、-Infinity等非标准常量，则不会调用该可调用对象，该调用直接进入手动审批。这是安全默认行为——宁可多暂停一次，也不要在参数不可信时跳过审批。Runner 和 Realtime 工具调用的行为相同。

哪些工具支持审批

needs_approval在以下工具类型上可用：

需要说明的限制：托管 Shell 环境不支持needs_approval或on_approval（从 src/agents/tool.py 可以看到，ShellTool在托管环境下会强制把这些字段重置为False）。

审批流程如何工作

五个关键步骤

1. 评估审批规则：当模型发出工具调用时，运行器会评估其审批规则（needs_approval、require_approval或托管 MCP 的对应规则）；
2. 检查已存储决策：如果该工具调用的审批决策已存储在RunContextWrapper中，运行器将直接继续执行，不再提示。单次调用的审批作用域限定于特定调用 ID；传入always_approve=True或always_reject=True，可在本次运行的剩余期间，为后续对同一工具标识的调用保留相同决策（sticky decision）；
3. 暂停运行：如果审批规则要求审批且尚未存储该调用的决策，执行暂停，RunResult.interruptions（或RunResultStreaming.interruptions）包含ToolApprovalItem条目。这也包括任务转移之后或嵌套Agent.as_tool()执行内部发起的审批；
4. 批准或拒绝：用result.to_state()将结果转换为RunState，调用state.approve(...)或state.reject(...)，然后用Runner.run(agent, state)或Runner.run_streamed(agent, state)恢复运行——注意agent必须是该次运行的原始顶层智能体；
5. 继续执行：恢复后的运行从暂停处继续，并在需要新审批时重新进入此流程。

从源码看，approve/reject最终落到RunContextWrapper的approve_tool/reject_tool上（见 src/agents/run_state.py），并且会优先把审批路由到嵌套智能体运行的状态上（_find_nested_approval_state），这正对应了"嵌套审批呈现在外层"的文档约定。

持久决策（sticky decisions）

使用always_approve=True或always_reject=True创建的持久决策会存储在运行状态中，因此当你之后恢复同一个已暂停的运行时，这些决策在state.to_string()/RunState.from_string(...)与state.to_json()/RunState.from_json(...)之后仍然有效。

对于来自HostedMCPTool的审批请求，SDK 使用server_label与工具名称的组合来标识持久工具决策。在一个托管 MCP 服务器上对lookup_account做出的始终批准决策，不会批准另一个服务器上同名的工具。并且，只有当托管 MCP 审批请求同时包含两个非空标识字段时，SDK 才会持久保存始终批准或始终拒绝的决策。

可以部分解决审批

你不必在同一轮处理中解决所有待处理审批。interruptions可以同时包含常规函数工具、托管 MCP 审批以及嵌套的Agent.as_tool()审批。如果你仅批准或拒绝部分项目后重新运行，已解决的调用可以继续，而未解决的调用仍会保留在interruptions中，并再次暂停运行。

自定义拒绝消息

默认情况下，被拒绝的工具调用会将 SDK 的标准拒绝文本返回运行中。你可以在两个层级自定义该消息：

• 全运行范围的后备设置：设置RunConfig.tool_error_formatter，控制整个运行中审批遭拒时默认向模型显示的消息；
• 单次调用覆盖：向state.reject(...)传入rejection_message=...，让某个特定的被拒绝工具调用呈现不同消息。

如果两者都已提供，单次调用的rejection_message优先于全运行范围的格式化器。

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_formatter=format_rejection) # Later, while resolving a specific interruption: state.reject( interruption, rejection_message="Publish action was canceled because the reviewer denied approval.", )

格式化器收到的ToolErrorFormatterArgs数据类包含kind（如"approval_rejected"、"tool_not_found"）、tool_type、tool_name、call_id、default_message和run_context等字段（见 src/agents/run_config.py），返回str | None，返回None时回退到 SDK 默认消息。

完整展示两层结合的代码示例见 examples/agent_patterns/human_in_the_loop_custom_rejection.py。

自动审批决策

手动interruptions是最通用的模式，但并非唯一方式：

• 本地ShellTool和ApplyPatchTool可以使用on_approval，在代码中立即批准或拒绝；
• HostedMCPTool可以结合使用tool_config={"require_approval": "always"}与on_approval_request，做出同类程序化决策；
• 普通function_tool工具和Agent.as_tool()使用本页介绍的手动中断流程。

当这些回调返回决策时，运行会继续，而无需暂停等待人工响应。适合那些"可以用代码判断"的场景，例如：允许执行白名单命令、拒绝所有写操作等。对于 Realtime 和语音会话 API，请参阅 Realtime 指南 中的审批流程。

流式传输与会话

同一中断流程也适用于流式运行。流式运行暂停后，应持续消费RunResultStreaming.stream_events()直到迭代器结束；然后检查RunResultStreaming.interruptions、解决其中的中断项，并在希望恢复后的输出继续流式传输时，使用Runner.run_streamed(...)恢复。有关此模式的流式版本，请参阅 流式传输。

如果你还使用了会话，请在从RunState恢复时继续传入同一个会话实例，或者传入针对相同会话 ID 和后端存储配置的另一个会话对象。恢复后的轮次随后会追加到同一份已存储的对话历史中。有关会话生命周期的详细信息，请参阅 会话。

完整示例：暂停、批准与恢复

下面的代码片段采用与 JavaScript HITL 指南相同的流程：在工具需要审批时暂停，将状态持久化到磁盘，重新加载状态，并在收集决策后恢复运行。

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_approval=needs_oakland_approval) async def get_temperature(city: str) -> str: return f"The temperature in {city} is 20° Celsius" agent = Agent( name="Weather assistant", instructions="Answer 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(f"Approve {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(parents=True, exist_ok=True) 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_approve=False) else: state.reject(interruption) result = await Runner.run(agent, state) print(result.final_output) if __name__ == "__main__": asyncio.run(main())

几点实战提示：

• 示例中prompt_approval是同步函数（因为它使用input()），通过run_in_executor(...)执行以免阻塞事件循环。如果你的审批来源本身是异步的（例如 HTTP 请求或异步数据库查询），则可以改用async def函数并直接await它；
• 若要在可能因审批而暂停的运行中使用流式传输，调用Runner.run_streamed，消费result.stream_events()直至完成，然后执行与上述相同的result.to_state()和恢复步骤；
• 循环条件while result.interruptions会持续处理直到所有审批都被解决——这也印证了前文"可以部分解决审批"的语义。

仓库模式与代码示例

仓库提供了覆盖各场景的 HITL 参考实现，可以直接对照阅读：

• 流式审批：examples/agent_patterns/human_in_the_loop_stream.py 展示了如何完整消费stream_events()，然后批准待处理的工具调用，最后使用Runner.run_streamed(agent, state)恢复运行。示例中的_needs_temperature_approval演示了动态审批（仅对 Oakland 城市要求审批）；
• 自定义拒绝文本：examples/agent_patterns/human_in_the_loop_custom_rejection.py 展示了在审批被拒绝时，如何将运行级tool_error_formatter与单次调用的rejection_message覆盖设置结合使用；
• 智能体工具审批：当委托给智能体的任务需要审核时，Agent.as_tool(..., needs_approval=...)会应用相同的中断流程。嵌套中断仍会呈现在外层运行中，因此应恢复原始顶层智能体，而不是嵌套智能体；
• 本地 Shell 和 apply_patch 工具：ShellTool和ApplyPatchTool也支持needs_approval。使用state.approve(interruption, always_approve=True)或state.reject(..., always_reject=True)，可在本次运行的剩余期间为该工具的后续调用缓存决策。对于自动决策，提供on_approval（参阅 examples/tools/shell.py）；对于手动决策，处理中断项（参阅 examples/tools/shell_human_in_the_loop.py）。托管的 Shell 环境不支持needs_approval或on_approval，详见 工具指南；
• 本地 MCP 服务器：在MCPServerStdio/MCPServerSse/MCPServerStreamableHttp上使用require_approval控制 MCP 工具调用（参阅 examples/mcp/get_all_mcp_tools_example/main.py 和 examples/mcp/tool_filter_example/main.py）；
• 托管 MCP 服务器：在HostedMCPTool上设置tool_config={"require_approval": "always"}以强制执行 HITL，并可选择提供on_approval_request来自动批准或拒绝（参阅 examples/hosted_mcp/human_in_the_loop.py 和 examples/hosted_mcp/on_approval.py）。对于受信任的服务器，请使用"never"（examples/hosted_mcp/simple.py）；
• 会话与记忆：向Runner.run传入会话，使审批和对话历史能够跨多个轮次保留。SQLite 和 OpenAI Conversations 会话变体位于 examples/memory/memory_session_hitl_example.py 和 examples/memory/openai_session_hitl_example.py；
• 实时智能体：实时演示提供了 WebSocket 消息，可通过RealtimeSession上的approve_tool_call/reject_tool_call批准或拒绝工具调用（有关服务器端处理程序，请参阅 examples/realtime/app/server.py；有关 API 接口，请参阅 Realtime 指南）。

长期审批：RunState持久化

RunState专为持久化而设计。使用state.to_json()或state.to_string()将待处理工作存储在数据库或队列中，之后再使用RunState.from_json(...)或RunState.from_string(...)重新创建它。这允许审批跨进程、跨机器甚至跨数小时/数天完成。

实用的序列化选项：

从源码看，to_json()序列化的内容（见 src/agents/run_state.py）包括：usage（用量）、approvals（审批记录，含 approved/rejected 列表与 sticky 消息）、tool_invocations（工具调用台账）、context及其元数据等。

安全提醒：已序列化的运行状态包含应用上下文，以及 SDK 管理的运行时元数据（审批、用量、序列化的tool_input、嵌套的智能体工具恢复信息、追踪元数据和服务器管理的对话设置）。如果你计划存储或传输已序列化的状态，请将RunContextWrapper.context视为持久化数据；除非你明确希望密钥随状态一起传递，否则避免将密钥放入其中。

待处理任务的版本管理

如果审批可能长时间处于待处理状态（例如人工审批需要数小时），请将智能体定义或 SDK 的版本标记与已序列化状态一同存储。这样，你就可以将反序列化操作路由到匹配的代码路径，避免模型、提示词或工具定义发生变化时出现不兼容问题。实践中可以约定：序列化状态时同时写入agents.__version__（见 src/agents/version.py）与自定义的智能体 schema 版本号，恢复时先校验版本再选择对应的反序列化逻辑。

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