为什么 Pydantic AI 报 UserError 不允许在同步工具里调用 run_sync，该如何改造

为什么 Pydantic AI 报 UserError 不允许在同步工具里调用 run_sync该如何改造【免费下载链接】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在 Pydantic AI 里做多 agent 协作一个 agent 通过工具调用另一个 agent时如果你的工具是同步函数def并在其中调用sub_agent.run_sync(...)去跑另一个 agent运行会直接抛出一个UserError类似这样Agent.run_sync() and Agent.run_stream_sync() cannot be used inside a synchronous tool, output function, or other function called during an agent run, as they can deadlock the run. Make the function async def and use await agent.run(...) or async with agent.run_stream(...) instead.这个错误只发生在一条特定路径上父 agent 的运行进行中某个同步工具、输出函数或其他被调用的同步函数试图用Agent.run_sync()/Agent.run_stream_sync()发起一次嵌套运行。本文的改造目标是把发起委托的函数改成async def、用await跑内层 agent同时保留父 agent 用run_sync()从普通同步代码启动的方式。依据来自 Troubleshooting 文档 和 Agent delegation 文档。先确认触发条件报错的抛出点在 check_no_nested_sync_runPydantic AI 分发同步工具、同步输出函数等回调时会标记当前上下文如果在这个上下文里再调用同步的 agent 入口方法run_sync()、run_stream_sync()就抛出上面这个UserError。所以满足以下全部条件才会遇到它当前处于某次 agent 运行内部工具、输出函数等被框架调用的同步函数该函数调用了run_sync()或run_stream_sync()来启动另一 agent或又一次运行。反过来在普通应用代码不在任何一次运行内部里用run_sync()启动父 agent 是正常用法不受此限制。为什么框架要拒绝这个调用run_sync 的文档串说明run_sync()内部用loop.run_until_complete(...)驱动事件循环因此不能用在已有活跃事件循环的 async 代码里。而 check_no_nested_sync_run 的说明给出了嵌套场景下具体会出什么问题同步回调要么在 worker 线程上执行此时嵌套的同步运行会启动第二个事件循环可能与绑定在父运行循环上的异步资源相互等待而死锁要么在禁用线程池时内联执行此时它要驱动的循环已经在运行同样会失败。Troubleshooting 文档对此的表述是父运行还在等你的函数返回嵌套的同步运行又会把它阻塞住存在死锁风险所以框架选择提前报错并给出改造指引而不是让你实际遇到死锁。另外一个容易混淆的点dependencies 文档强调用run还是run_sync与工具本身是同步还是异步无关——run_sync只是run的包装agent 总是在异步上下文里运行。因此把工具改成async def并不需要你把整个应用改成异步。改造步骤改造只有一处核心动作把发起委托的函数改为async def并用await调用内层 agent 的run()对应流式场景则用async with agent.run_stream(...)。按 Agent delegation 文档的完整可运行示例改造后的结构如下from pydantic_ai import Agent, RunContext, UsageLimits joke_selection_agent Agent( openai:gpt-5.2, namejoke_selection_agent, instructions( Use the joke_factory to generate some jokes, then choose the best. You must return just a single joke. ), ) joke_generation_agent Agent( google:gemini-3-flash-preview, namejoke_generation_agent, output_typelist[str] ) joke_selection_agent.tool async def joke_factory(ctx: RunContext, count: int) - list[str]: r await joke_generation_agent.run( fPlease generate {count} jokes., usagectx.usage, ) return r.output result joke_selection_agent.run_sync( Tell me a joke., usage_limitsUsageLimits(request_limit5, total_tokens_limit500), ) print(result.output) print(result.usage)这个示例文档标注为完整、可原样运行_This example is complete, it can be run as is_涉及的关键点joke_factory是父 agent 的工具但它是async def内部用await joke_generation_agent.run(...)而不是run_sync()。文档明确说这是必需写法required, not stylisticrun_sync()/run_stream_sync()在工具、输出函数或其他运行期间被调用的函数里都会抛UserError。父 agent 仍然从普通同步代码用run_sync()启动这一点不受影响——只有委托函数需要是async def。通过usagectx.usage把父运行的 usage 传给委托运行的usage参数这样最终result.usage会包含两个 agent 的用量如果不需要合并统计可以去掉该参数但按文档的说法这通常是期望行为youll generally want to passctx.usage。如果父、委托 agent 需要共享依赖给两个 agent 设置相同的deps_type并在工具里传depsctx.deps带依赖的完整变体见 agent delegation and dependencies 一节运行该变体需要import asyncio并调用asyncio.run(main())。如果委托函数里还有阻塞式工作如同步 I/O保持async def不变只把那部分推入asyncio.to_thread()其余部分继续await。准备条件方面示例使用了openai:gpt-5.2和google:gemini-3-flash-preview两个模型需要按 Models 文档配置对应 provider 的 API key设置[PROVIDER]_API_KEY环境变量或通过 provider 的api_key参数传入不想配置 key 的话文档建议使用内置的test模型。如何验证改造生效按文档给出的判断标准不再抛出UserError运行完成print(result.output)能打印父 agent 的最终输出。文档示例的输出是仅作示例实际内容取决于模型# Did you hear about the toothpaste scandal? They called it Colgate.usage 合并正确传入usagectx.usage后result.usage包含父 agent 与委托 agent 两边的请求。上面的简单示例对应的文档示例值为RunUsage( costDecimal(0.00051200), input_tokens165, output_tokens24, requests3, tool_calls1, )带依赖的变体委托 agent 内部还有一个工具则显示 4 个请求——父 agent 2 个、委托 agent 2 个。这里的数值同样是文档示例不要当作固定预期。可选两个 agent 都传了name如果接入了 Logfire父、委托两个 agent 的 run span 会分别带各自的名字可以用来确认委托确实发生、并区分两者见 Tracing Agent Delegation。容易混淆的相邻问题Temporal workflow 里报UserError是另一条错误在 Temporal workflow 内调用run_sync()会抛出agent.run_sync()cannot be used inside a Temporal workflow. Useawait agent.run()instead.因为run_sync()会自己驱动事件循环而 Temporal 的 workflow 事件循环不允许workflow 之外则照常可用。见 Temporal 文档。RuntimeError: Event loop is closed看起来与事件循环相关但那是 agent 的事件循环被其他代码关闭、provider 连接池还绑着死循环上的连接导致的解法是重建 agent连同 model/provider与本文的嵌套同步运行问题不同。这两类都在 Troubleshooting 文档中单独列了一节。改完之后如果你后续遇到更复杂的多 agent 控制流多个 agent 依次交接、或需要图式状态机可以继续看 multi-agent-applications 文档中的 Programmatic agent hand-off 和 Pydantic Graphs 章节。【免费下载链接】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），仅供参考