Pydantic AI 延迟工具实战：审批工具、外部执行与两种解析路径

Pydantic AI 延迟工具实战审批工具、外部执行与两种解析路径【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-aiPydantic AI 的延迟工具Deferred Tools机制面向如下场景模型要调用的工具不能、也不应在当前 agent 运行中同步执行——它需要人工审批、依赖前端或外部服务提供结果或由后台 worker 完成耗时任务。本文以 docs/deferred-tools.md 为核心结合仓库源码实现完整讲解如何声明需审批工具与外部工具、如何使用内联 handlerHandleDeferredToolCallscapability与结束运行→新运行带回结果两条解析路径以及如何在流式事件流中观察这些延迟调用帮助读者直接落地 human-in-the-loop 审批与外部任务交接。为什么需要延迟工具在 agent 系统中工具调用往往不是可以立即安全执行的。根据 docs/deferred-tools.md 的界定至少有以下几类场景工具不应或不能在同一 agent run、同一 Python 进程中同步执行需要用户先批准如删除文件、转账等敏感操作模型不能直接执行结果依赖外部方结果需要由上游服务、前端或用户提供结果生成耗时过长不值得让 agent 进程一直挂着等待。为支持这些用例Pydantic AI 提供延迟工具概念分为两种需要审批的工具require approval外部执行的工具executed externally关键设计在于当模型调用延迟工具时run 不会崩溃或挂死而是被暂停由调用方选择两条解析路径之一。两种解析路径内联 handler 与两段式 run路径一内联解析inline使用带 handler 的HandleDeferredToolCallscapability 解析部分或全部待处理调用。agent run 在单次调用内继续执行无需结束再重启。适合解析器审批闸门、外部服务客户端与 agent 同进程的场景。路径二结束运行由新运行带回结果stop-the-worldrun 以DeferredToolRequests输出对象结束其中携带延迟调用的信息调用方收集审批/结果后发起一次新的 agent run同时传入原 run 的 message history 和一个DeferredToolResults对象。这次 follow-up 是独立的 agent run拥有自己的run_id不要复用被暂停 run 的 run_id通过conversation_id保持暂停/恢复的关联。适合解析器位于 agent 进程之外的场景——例如 UI 适配器把待处理调用呈现给用户、拿到回复后再启动 follow-up run。两条路径可以组合handler 可以解析一部分调用让其余的作为DeferredToolRequests输出冒泡交由外层调用方处理。stop-the-world 流程的前提DeferredToolRequests必须在Agent的output_type中以便正确推断 agent run 输出的可能类型。如果你的 agent 也可能在没有延迟工具的上下文中使用、又不想到处处理这个类型可以改为在调用 [agent.run()]、[agent.run_sync()]、[agent.run_stream()] 或 [agent.iter()] 时传output_type参数。注意出于类型推断原因运行时的output_type会覆盖构造时声明的类型因此需显式把原始输出类型一并包含进去。这一行为有源码佐证在工具执行管线 _tool_execution.py 中若出现延迟调用但DeferredToolRequests不在输出类型里框架会直接报错提示将其加入输出类型。从源码结构看这两个数据类的设计意图也很清晰pydantic_ai_slim/pydantic_ai/_deferred.pyDeferredToolRequests持有三个字段calls待外部执行的工具调用list[ToolCallPart]、approvals待审批的工具调用list[ToolCallPart]和metadata按tool_call_id键控的dict[str, dict[str, Any]]见 L37-L42DeferredToolResults的calls字段映射 tool call ID 到任意值 /ToolReturn/ 异常approvals字段映射 tool call ID 到布尔 /ToolApproved/ToolDenied见 L166-L173其to_tool_call_results()方法在转换为管线内部格式时会把True/False归一化为ToolApproved/ToolDenied并把外部调用的普通值包装为ToolReturnL181-L205。用 handler 解析延迟调用推荐的延迟工具调用处理方式是注册一个HandleDeferredToolCallscapability其 handler 接收DeferredToolRequests返回解析了部分或全部请求的DeferredToolResults。工具执行管线会内联应用这些结果agent run 在一次调用中继续执行如同延迟工具正常返回了一样。配置了 handler 后DeferredToolRequests就不再需要声明为输出类型——除非你还希望未解析的调用冒泡给调用方见下文。从实现看HandleDeferredToolCalls是一个包装 handler 函数sync 或 async 均可内部用inspect.isawaitable判断是否需要 await的轻量 dataclass见 L51-L75handler 返回包含部分/全部结果的DeferredToolResults时对应调用被内联解析handler返回None或在结果中省略某些调用时下一个HandleDeferredToolCalls或任何覆盖handle_deferred_tool_callshook 的 capability获得处理机会最终仍未解析的调用作为DeferredToolRequests输出冒泡。源码侧DeferredToolRequests.remaining()方法L88-L99负责计算结果与待处理请求的差集全部解析则返回None允许冒泡的前提是把DeferredToolRequests加入 agent 的output_type——从而可以把内联处理与 stop-the-world 流程组合使用。build_results()是便捷构造器它验证每个 tool call ID 都对应正确类别的待处理请求否则抛ValueError见 L70-L80并支持approve_allTrue自动批准未显式列出的审批请求填入默认ToolApproved()。from pydantic_ai import ( Agent, ApprovalRequired, CallDeferred, DeferredToolRequests, DeferredToolResults, RunContext, ToolDenied, ) from pydantic_ai.capabilities import HandleDeferredToolCalls async def handle_deferred( ctx: RunContext, requests: DeferredToolRequests ) - DeferredToolResults: approvals: dict[str, bool | ToolDenied] {} for call in requests.approvals: if call.tool_name delete_file: approvals[call.tool_call_id] ToolDenied(Deleting files is not allowed) else: approvals[call.tool_call_id] True calls {call.tool_call_id: f(external result for {call.tool_name}) for call in requests.calls} return requests.build_results(approvalsapprovals, callscalls) agent Agent( openai:gpt-5.2, capabilities[HandleDeferredToolCalls(handlerhandle_deferred)], ) agent.tool_plain(requires_approvalTrue) def delete_file(path: str) - str: return fFile {path!r} deleted # (1)! agent.tool def update_file(ctx: RunContext, path: str, content: str) - str: if path .env and not ctx.tool_call_approved: raise ApprovalRequired return fFile {path!r} updated: {content!r} agent.tool_plain async def send_to_worker(task: str) - str: raise CallDeferred # (2)!这里永远不会执行到——handler 拒绝了这个调用模型看到的是拒绝消息。handler 为该外部调用提供结果因此工具函数体只负责发出延迟信号。如果你正在构建自定义 capability且需要自行解析审批或外部调用例如暴露延迟工具的沙箱请在你的 capability 上直接覆盖handle_deferred_tool_callshook而不是再注册一个HandleDeferredToolCalls。同一个 hook 也可以通过Hookscapability 使用——见 Hooks。下文将分别介绍 handler 可解析的两类延迟工具以及每一类的 stop-the-world 替代流程。多个 capability 如何组合含WrapperCapability与capabilities[...]列表见 Capabilities。Human-in-the-Loop 工具审批声明需审批的工具如果工具函数总是需要审批可以向agent.tool装饰器、agent.tool_plain装饰器、Tool类、FunctionToolset.tool装饰器或FunctionToolset.add_function()方法传入requires_approvalTrue。进入函数后你可以假定该工具调用已经获得批准。如果审批与否取决于工具调用的参数或 agent run context依赖项、消息历史等请在工具函数内抛出ApprovalRequired。RunContext.tool_call_approved属性在该调用已获批准时为True。也可以在工具的args_validator中抛出它——args_validator在工具函数之前运行让你在请人审批之前先拒绝无效参数。如需对某个 toolset如 MCP server提供的工具调用要求审批见ApprovalRequiredToolset文档。实时会话注意在 realtime session 中审批必须内联解析通常由HandleDeferredToolCallshandler 完成capability hook 也可以没有任何解析器的调用会被每次拒绝。!!! warning 审批不是针对不可信客户端的授权边界 当你通过 UI 适配器对外提供服务时审批决定由客户端随请求一起提交适配器没有服务端记录来核对它发出过哪些工具调用。能触达该端点的客户端可以批准它自己发起的工具调用。Human-in-the-loop 审批防的是模型未经人工签核自行行动它不能替代对适配器端点做认证、并在敏感操作的工具函数内部强制鉴权——工具函数无论调用如何进入历史都会执行。这一约束适用于任何接受客户端message_history的端点不只是适配器——见 Trust boundary for client-supplied history 与 Trust model for client-submitted messages。审批流程细节stop-the-world 流程当模型调用了需要审批的工具agent run 会以DeferredToolRequests输出对象结束其approvals列表持有ToolCallPart包含工具名、已校验的参数和唯一的 tool call ID。收集到用户的批准/拒绝后构造一个DeferredToolResults其approvals字典把每个 tool call ID 映射为一个布尔值True/False一个ToolApproved对象可选override_args批准时可用它替换原始参数见 L103-L109或一个ToolDenied对象可选自定义message供模型看到见 L112-L121。还可以在DeferredToolResults上提供metadata字典每个 tool call ID 映射到一个元数据字典可在工具的RunContext.tool_call_metadata属性中取到。然后把这个DeferredToolResults对象与原 run 的 message history 一起作为deferred_tool_results传给 agent 的任一运行方法。下面是一个完整示例所有文件删除、以及对受保护文件的更新都要求审批该示例完整、可直接运行from pydantic_ai import ( Agent, ApprovalRequired, DeferredToolRequests, DeferredToolResults, RunContext, ToolDenied, ) agent Agent(openai:gpt-5.2, output_type[str, DeferredToolRequests]) PROTECTED_FILES {.env} agent.tool def update_file(ctx: RunContext, path: str, content: str) - str: if path in PROTECTED_FILES and not ctx.tool_call_approved: raise ApprovalRequired(metadata{reason: protected}) # (1)! return fFile {path!r} updated: {content!r} agent.tool_plain(requires_approvalTrue) def delete_file(path: str) - str: return fFile {path!r} deleted result agent.run_sync(Delete __init__.py, write Hello, world! to README.md, and clear .env) messages result.all_messages() assert isinstance(result.output, DeferredToolRequests) requests result.output print(requests) DeferredToolRequests( calls[], approvals[ ToolCallPart( tool_nameupdate_file, args{path: .env, content: }, tool_call_idupdate_file_dotenv, ), ToolCallPart( tool_namedelete_file, args{path: __init__.py}, tool_call_iddelete_file, ), ], metadata{update_file_dotenv: {reason: protected}}, ) results DeferredToolResults() for call in requests.approvals: result False if call.tool_name update_file: # Approve all updates result True elif call.tool_name delete_file: # deny all deletes result ToolDenied(Deleting files is not allowed) results.approvals[call.tool_call_id] result result agent.run_sync( Now create a backup of README.md, # (2)! message_historymessages, deferred_tool_resultsresults, ) print(result.output) Heres what Ive done: - Attempted to delete __init__.py, but deletion is not allowed. - Updated README.md with: Hello, world! - Cleared .env (set to empty). - Created a backup at README.md.bak containing: Hello, world! If you want a different backup name or format (e.g., timestamped like README_2025-11-24.bak), let me know. 可选的metadata参数可向延迟工具调用附加任意上下文按tool_call_id键控可从DeferredToolRequests.metadata访问。第二次 agent run 从第一次 run 中断处继续提供审批结果并可选携带新的user_prompt给模型补充指令。从该示例的消息历史可以看到完整的暂停—恢复过程形态第一个ModelRequest携带用户指令ModelResponse发出三个ToolCallPartfollow-up run 先追加一条ModelRequest其中包含两个审批相关调用的结果——被拒调用是带outcomedenied的ToolReturnPart批准的.env更新是普通ToolReturnPart——之后模型继续执行创建README.md.bak备份并输出总结。!!! note 工具结果顺序 工具结果遵循模型发出对应工具调用的顺序。在上面消息历史中delete_file的被拒结果出现在update_file的.env结果之前因为模型先发出的是delete_file。这是 v2 的有意行为变更结果不再按工具类别分组你看到的顺序即模型发出调用的顺序。外部工具执行如果一个工具调用的结果无法在发起调用的同一个 agent run 内生成该工具即为外部工具。典型例子由 web/app 前端实现的客户端工具以及交给后台 worker 或外部服务执行、而不是让 agent 进程干等的耗时任务。用 CallDeferred 做条件延迟如果是否外部执行取决于调用参数、agent run context如依赖项、消息历史或任务预计耗时可以定义工具函数并条件性地抛出CallDeferred异常。抛出前工具函数通常先调度一个后台任务并传递RunContext.tool_call_id以便之后把结果匹配回该延迟调用。从源码看CallDeferred与ApprovalRequired都是携带可选metadata字典的轻量异常exceptions.pymetadata 会以tool_call_id为键出现在DeferredToolRequests.metadata中——外部执行示例正是靠它带出task_id。工具执行管线在 _tool_execution.py 中显式捕获这两个异常并转入延迟分支。与审批类似工具的args_validator也可以抛出CallDeferred这样只有参数合法的调用才会被交接出去。用 ExternalToolset 处理基于 schema 的外部工具如果一个工具总是外部执行且其定义连同参数的 JSON schema 一起提供给你的代码可以使用ExternalToolset。如果外部工具事先已知、但你手头没有参数的 JSON schema也可以定义一个签名合适的工具函数其唯一动作就是抛出CallDeferred。外部执行流程细节当模型调用外部工具时agent run 以DeferredToolRequests输出对象结束其calls列表持有ToolCallPart包含工具名、已校验的参数和唯一 tool call ID。当工具调用结果就绪后构造DeferredToolResults其calls字典把每个 tool call ID 映射为任意将返回给模型的值内部自动包为ToolReturn一个ToolReturn对象或调用失败时的异常ModelRetry让模型重试或ToolFailed把失败作为失败结果报告给模型不消耗该工具的 retry 预算由其决定如何继续。随后把这个DeferredToolResults对象与原 run 的 message history 一起作为deferred_tool_results传给运行方法。下面是一个完整示例把耗时任务移到后台任务完成后把结果返回给模型运行前请确保导入asyncio并加上asyncio.run(main())无需其他改动import asyncio from dataclasses import dataclass from typing import Any from pydantic_ai import ( Agent, CallDeferred, DeferredToolRequests, DeferredToolResults, ModelRetry, RunContext, ) dataclass class TaskResult: task_id: str result: Any async def calculate_answer_task(task_id: str, question: str) - TaskResult: await asyncio.sleep(1) return TaskResult(task_idtask_id, result42) agent Agent(openai:gpt-5.2, output_type[str, DeferredToolRequests]) tasks: list[asyncio.Task[TaskResult]] [] agent.tool async def calculate_answer(ctx: RunContext, question: str) - str: task_id ftask_{len(tasks)} # (1)! task asyncio.create_task(calculate_answer_task(task_id, question)) tasks.append(task) raise CallDeferred(metadata{task_id: task_id}) # (2)! async def main(): result await agent.run(Calculate the answer to the ultimate question of life, the universe, and everything) messages result.all_messages() assert isinstance(result.output, DeferredToolRequests) requests result.output print(requests) DeferredToolRequests( calls[ ToolCallPart( tool_namecalculate_answer, args{ question: the ultimate question of life, the universe, and everything }, tool_call_idpyd_ai_tool_call_id, ) ], approvals[], metadata{pyd_ai_tool_call_id: {task_id: task_0}}, ) done, _ await asyncio.wait(tasks) # (3)! task_results [task.result() for task in done] task_results_by_task_id {result.task_id: result.result for result in task_results} results DeferredToolResults() for call in requests.calls: try: task_id requests.metadata[call.tool_call_id][task_id] result task_results_by_task_id[task_id] except KeyError: result ModelRetry(No result for this tool call was found.) results.calls[call.tool_call_id] result result await agent.run(message_historymessages, deferred_tool_resultsresults) print(result.output) # The answer to the ultimate question of life, the universe, and everything is 42.生成一个可独立于 tool call ID 追踪的任务 ID。可选的metadata参数传递task_id便于之后与结果匹配按tool_call_id键控可从DeferredToolRequests.metadata访问。实际场景中这一步通常发生在一个独立进程里——它轮询任务状态或在所有挂起任务完成时被通知。在事件流中观察延迟工具调用与其他工具调用一样延迟工具调用会向事件流发出FunctionToolCallEvent——但仅凭这个事件流消费者无法得知该调用正挂起等待交互也无法得知期望什么类型的交互。另有两种AgentStreamEvent携带这些上下文DeferredToolRequestsEvent——每批延迟调用发出一次携带DeferredToolRequests。它在任何HandleDeferredToolCallshandler 运行之前发出因此消费方可以例如在 handler 等待期间通知前端需要输入。若没有任何 handler 解析全部请求run 会以待处理请求作为其DeferredToolRequests输出结束。DeferredToolResultsEvent——handler 解析部分请求时发出携带DeferredToolResults。被解析的调用随后走常规管线执行各自发出FunctionToolResultEvent。当结果改为通过deferred_tool_results提供给新 run 时不会发出该事件——那种情况下调用方本来就已知结果。这让解析与呈现解耦handler 可以只包含纯解析逻辑例如在 durable execution 工作流中等待信号而流消费者负责与前端的所有通信无需自行维护哪些工具是交互式的映射。在管线源码中DeferredToolRequestsEvent的发出发生在延迟调用被批次化时见 _tool_execution.py L1036。继续上面的 handler 示例from pydantic_ai import DeferredToolRequestsEvent, DeferredToolResultsEvent from deferred_tool_handler import agent async def main(): async with agent.run_stream_events( Delete __init__.py, write Hello, world! to README.md, and clear .env ) as events: async for event in events: if isinstance(event, DeferredToolRequestsEvent): print(fApprovals needed: {[call.tool_name for call in event.requests.approvals]}) # Approvals needed: [update_file, delete_file] elif isinstance(event, DeferredToolResultsEvent): print(fResolved: {list(event.results.approvals)}) # Resolved: [update_file_dotenv, delete_file]延伸阅读Function Tools — 工具基础概念与注册Advanced Tool Features — 自定义 schema、动态工具与执行细节含args_validator、ToolReturn、ModelRetry、ToolFailedToolsets — 工具集合管理包括外部工具用的ExternalToolset与审批用的ApprovalRequiredToolsetMessage History — 延迟工具的消息历史含run_id/conversation_id的用法Realtime tools — 实时语音会话中的审批与延迟工具Capabilities — 多个 capability 的组合方式含WrapperCapability与capabilities[...]列表关键类型与入口速查条目说明源码位置DeferredToolRequests暂停输出calls/approvals/metadatapydantic_ai_slim/pydantic_ai/_deferred.pyDeferredToolResults恢复输入approvals接受 bool /ToolApproved/ToolDeniedcalls接受任意值 /ToolReturn/ 异常pydantic_ai_slim/pydantic_ai/_deferred.pybuild_results()便捷构造器校验 ID 归属支持approve_allTruepydantic_ai_slim/pydantic_ai/_deferred.pyremaining()计算未解析请求支撑 handler 链式处理与冒泡pydantic_ai_slim/pydantic_ai/_deferred.pyHandleDeferredToolCalls内联解析 capabilityhandler 支持 sync/async、可返回None跳过pydantic_ai_slim/pydantic_ai/capabilities/deferred_tool_handler.pyCallDeferred/ApprovalRequired标记延迟/需审批调用的异常均可携带可选metadatapydantic_ai_slim/pydantic_ai/exceptions.py管线延迟分支捕获两类异常、发出DeferredToolRequestsEvent输出类型缺失时报错pydantic_ai_slim/pydantic_ai/_tool_execution.py【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考