👁 92 次阅读

多智能体HITL审批循环：pydantic-ai如何正确实现人工审批

本文讨论的问题源自 pydantic-ai 项目 issue #3274：[Human in the Loop Approval for Multi Agent Systems](https://github.com/pydantic/pydantic-ai/issues/3274)。该 issue 当前状态为 open，标签为 feature，已有 42 条评论，属于尚未合入主干的功能请求讨论。提问者正在构建多智能体系统，需要一个 coordinator 将任务委派给 specialist（例如文件管理专家），当 specialist 需要执行敏感操作（如删除文件）时，先经过人工审批再执行。

现象描述

提问者的当前实现流程如下： 1. Coordinator 将任务委派给 specialist 2. Specialist 识别到敏感操作，返回 DeferredToolRequest（延迟工具请求） 3. Coordinator 捕获到该请求后抛出 ApprovalRequired 异常 4. 系统暂停，等待人工审批 5. 审批通过后，构造 DeferredToolResults 传回给 coordinator 6. Coordinator 收到结果后，再次调用 specialist，形成循环核心问题：DeferredToolResults 传回 coordinator 后，coordinator 将其视为”specialist 尚未完成”的状态，于是再次触发 delegate_to_specialist 调用，而不是直接完成已批准的动作。

背景原因

循环产生的根本原因在于 coordinator 的决策逻辑与审批流程的语义不匹配：

Coordinator 的 prompt 通常引导其将任务委派给 specialist，并将 specialist 的返回值作为”任务完成”的标志
DeferredToolResults 在 coordinator 看来等同于”specialist 返回了一个中间结果”，而非”审批已完成，直接执行”
pydantic-ai 中 ResultRequest 表示模型需要更多信息才能完成响应，coordinator 收到此类信号后倾向于继续调用工具（包括再次委派）
审批结果缺少”终止委派链”的语义标记，coordinator 无法区分”需要继续委派”与”审批已完成，应直接执行工具”

在 pydantic-ai 的 tool_call_handler 机制中，工具调用的生命周期由 ModelRequest 和 ResultRequest 驱动。当 specialist 返回 DeferredToolRequest 时，coordinator 的工具调用上下文被中断；但 DeferredToolResults 的注入方式并未改变 coordinator 的委派倾向。

排查与量化方法

排查此类循环问题时，可从以下维度入手：

排查维度	检查内容	预期表现
Coordinator prompt	是否包含”遇到 DeferredToolResults 时直接执行”的指令	审批后不再委派
DeferredToolResults 结构	是否包含工具执行结果而非委派请求	coordinator 识别为完成态
Approval ID 匹配	审批 ID 与 DeferredToolRequest ID 是否一致	无 ID 不匹配导致的重试
重试机制	是否因超时或异常触发请求重放	无重复委派记录
Specialist 返回值	specialist 是否在收到审批结果后仍返回 DeferredToolRequest	specialist 应执行工具
量化方法：在 coordinator 的 ModelRequest 日志中统计 delegate_to_specialist 的调用次数。若同一审批事件触发超过 2 次委派调用，即可确认循环问题存在。
完整解决方案
当前 pydantic-ai 官方尚未提供统一的 multi-agent HITL API（issue #3274 仍处于 open 状态）。以下方案基于 pydantic-ai 现有概念（DeferredToolRequest、DeferredToolResults、tool_call_handler、ModelRequest、ResultRequest）设计，属于社区讨论中的可行模式。
方案一：审批结果作为 coordinator 直接工具参数
核心思路：coordinator 在收到审批结果后，不重新委派给 specialist，而是将审批结果作为参数直接调用敏感工具。
# 伪代码示意 - 基于 pydantic-ai 概念 async def coordinator_tool_call_handler(model_request: ModelRequest): # 检查是否为审批结果回传 if model_request.tool_call_results: approval_result = model_request.tool_call_results.get("approval") if approval_result and approval_result.approved: # 直接执行已批准的工具，不再委派给 specialist tool_result = await execute_sensitive_tool( tool_name=approval_result.tool_name, params=approval_result.params, approval_id=approval_result.approval_id ) return ResultRequest( content=f"已执行已批准操作: {tool_result}" ) # 正常委派流程 return delegate_to_specialist(task=model_request)
方案二：审批状态注入 specialist context
核心思路：将审批状态通过 context 变量注入 specialist，specialist 根据审批状态决定是否执行工具或返回 DeferredToolRequest。
# 伪代码示意 - 基于 pydantic-ai 概念 # specialist 内部逻辑 async def specialist_agent(task: str, context: dict): if task.requires_sensitive_action: approval_status = context.get("approval_status") if approval_status and approval_status.approved: # 审批已通过，直接执行 result = await execute_sensitive_tool( tool_name=task.tool_name, params=task.params ) return result else: # 未审批，返回延迟请求 return DeferredToolRequest( tool_name=task.tool_name, params=task.params, reason="需要人工审批" )
方案三：coordinator 层拦截委派调用
核心思路：在 coordinator 的 tool_call_handler 中拦截 delegate_to_specialist 调用，当检测到审批结果时直接消费，不传递至 specialist。
# 伪代码示意 - 基于 pydantic-ai 概念 async def coordinator_tool_call_handler(model_request: ModelRequest): tool_call = model_request.tool_call if tool_call.name == "delegate_to_specialist": # 检查是否有待处理的审批结果 pending_approval = get_pending_approval(tool_call.params) if pending_approval and pending_approval.approved: # 拦截委派，直接执行审批通过的工具 result = await execute_tool_directly( tool_name=pending_approval.tool_name, params=pending_approval.params ) return ResultRequest(content=f"审批通过，已执行: {result}") # 无审批结果，正常委派 return delegate_to_specialist(**tool_call.params) return handle_other_tools(model_request)
方案对比
方案	实现复杂度	侵入性	适用场景	循环风险
方案一：审批结果作为工具参数	中	低	coordinator 可直接调用工具	低
方案二：审批状态注入 context	低	中	specialist 逻辑需改造	低
方案三：coordinator 层拦截	高	高	需要精细控制委派链	最低

常见错误排查

错误一：DeferredToolResults 结构与工具返回值混淆 DeferredToolResults 是审批结果的容器，不应与工具的实际返回值混用。若 coordinator 将 DeferredToolResults 当作工具返回值处理，会触发再次委派。确保 DeferredToolResults 仅用于传递审批状态，工具执行结果通过 ResultRequest 返回。 错误二：coordinator prompt 引导其再次委派 coordinator 的 system prompt 若包含”遇到中间结果时继续委派”的指令，会在收到审批结果后再次调用 specialist。需要在 prompt 中明确区分”中间结果”与”审批完成结果”，或添加”审批完成后直接执行工具”的指令。 错误三：审批结果被 specialist 忽略 当 coordinator 再次委派给 specialist 时，specialist 可能未检查审批状态，直接返回新的 DeferredToolRequest，形成无限循环。specialist 必须在执行敏感操作前检查审批状态。 错误四：approval ID 不匹配 审批 ID 与 DeferredToolRequest 中的 ID 不一致时，审批结果无法关联到原始请求，系统可能触发重试或创建新的审批请求。确保审批 ID 在 DeferredToolRequest 和 DeferredToolResults 之间保持一致。 错误五：重试机制重放请求 网络超时或异常触发的重试可能重放已审批的请求，导致工具被执行多次。在 coordinator 的 tool_call_handler 中添加幂等性检查，确保同一 approval ID 只执行一次。

实践建议

1. 在 coordinator 的 tool_call_handler 中明确处理审批结果分支，避免将其混入正常委派流程 2. specialist 在执行敏感操作前必须检查审批状态，未审批时返回 DeferredToolRequest，已审批时直接执行 3. 审批 ID 应在 DeferredToolRequest 中生成，并在 DeferredToolResults 中回传，确保关联一致性 4. 由于 pydantic-ai 尚未提供统一的 multi-agent HITL API，建议在 coordinator 层封装审批逻辑，降低对 specialist 的侵入 5. 在 ModelRequest 日志中记录委派调用次数，监控循环问题

FAQ

Q: pydantic-ai 是否已提供 multi-agent HITL 的官方 API？ A: 截至 issue #3274 讨论时，pydantic-ai 尚未合入统一的 multi-agent HITL API。该 issue 标签为 feature，状态为 open，属于功能请求讨论阶段。当前实现需要基于 DeferredToolRequest、DeferredToolResults、tool_call_handler 等现有概念自行组合。 Q: DeferredToolRequest 和 DeferredToolResults 的语义区别是什么？ A: DeferredToolRequest 表示工具调用需要延迟执行（如等待人工审批），由 specialist 返回；DeferredToolResults 表示延迟调用的结果（如审批通过/拒绝），由审批系统构造后传回。两者构成一对请求-响应关系。 Q: 为什么 coordinator 收到 DeferredToolResults 后会再次委派？ A: coordinator 的决策逻辑通常将 specialist 的返回值视为”任务进行中”，DeferredToolResults 未包含”任务已完成”的语义标记，导致 coordinator 倾向于继续委派。需在 tool_call_handler 中显式处理审批结果分支。 Q: 方案一和方案二哪个更适合生产环境？ A: 方案二（审批状态注入 context）侵入性较低，适合 specialist 逻辑可改造的场景；方案一（审批结果作为工具参数）侵入性更低，适合 coordinator 可直接调用工具的场景。具体选择取决于系统架构。 Q: 如何验证循环问题已修复？ A: 在 coordinator 的 ModelRequest 日志中统计 delegate_to_specialist 的调用次数。修复后，同一审批事件应仅触发一次委派调用。若仍出现多次调用，检查 coordinator prompt 和 tool_call_handler 的审批结果分支逻辑。