Pydantic AI 故障排查完全指南：事件循环、同步嵌套运行与 API Key 常见错误修复

Pydantic AI 故障排查完全指南事件循环、同步嵌套运行与 API Key 常见错误修复【免费下载链接】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 是一款类型安全的 Python AI Agent 框架涵盖 Agent、实时语音、图像生成与 Embedding 等能力。本文基于仓库中 docs/troubleshooting.md 整理聚焦开发者最容易踩坑的几类运行期错误——Jupyter 事件循环冲突、Event loop is closed、同步工具内嵌套运行死锁、API Key 配置缺失等逐一给出可复现的修复方案并结合框架源码pydantic_ai_slim/pydantic_ai/_utils.py、pydantic_graph/pydantic_graph/_utils.py解释错误背后的底层机制。读完本文你将能快速定位并修复 Pydantic AI 运行中的高频异常并理解同步/异步入口在事件循环层面的设计约束。一、Jupyter Notebook 中的RuntimeError: This event loop is already running在 Jupyter、Google Colab 或 Marimo 等交互式环境中运行 Agent最常见的报错就是RuntimeError: This event loop is already running。这是因为这些环境自身已经启动并持有一个事件循环而旧的同步调用方式尝试在已有循环之上再驱动一个循环从而发生冲突。1.1 现代环境Jupyter/IPython 7.0直接使用顶层awaitJupyter/IPython 7.0 及以上版本原生支持单元内的顶层await此时无需任何额外配置直接使用异步的Agent.run()即可from pydantic_ai import Agent agent Agent(openai:gpt-5.2) result await agent.run(Who let the dogs out?)1.2 旧版环境或特定集成nest-asyncio如果仍遇到事件循环冲突可以使用nest-asyncio修补嵌套循环限制然后调用同步的run_sync()import nest_asyncio from pydantic_ai import Agent nest_asyncio.apply() agent Agent(openai:gpt-5.2) result agent.run_sync(Who let the dogs out?)注意这一限制同样适用于 Google Colab 和 Marimo 环境。优先推荐方案一顶层awaitnest-asyncio仅在无法使用顶层await的旧环境中作为兜底。二、RuntimeError: Event loop is closed的成因与正确修复2.1 错误机制同步方法如何复用事件循环从源码看同步入口并非每次创建新循环而是复用当前线程的事件循环。见 pydantic_ai_slim/pydantic_ai/_utils.py#L1079-L1088 中的get_event_loop()def get_event_loop() - asyncio.AbstractEventLoop: try: event_loop asyncio.get_event_loop() except RuntimeError: event_loop None if event_loop is None or event_loop.is_closed(): event_loop asyncio.new_event_loop() asyncio.set_event_loop(event_loop) return event_loop逻辑很清晰如果当前线程没有事件循环或者已有的循环已被关闭is_closed()则创建一个全新的事件循环并设为当前线程的默认循环。因此Agent.run_sync()这类同步方法在正常情况下可以自愈——即使别处代码关闭了循环它也会安装一个新循环继续工作。2.2 真正的根因模型/Provider 连接池绑定已死循环如果Event loop is closed是从httpx2或旧版httpx在模型请求期间抛出的说明该 Agent 在它的事件循环被关闭之前就已经被使用过Provider 的 HTTP 连接池中仍持有绑定到已关闭循环上的连接。正确做法是连同模型和 Provider 一起重建 Agent或给 Provider 传入一个新的http_client不要复用已有的Model实例——复用会继续保留已失效的连接池同时避免在别处关闭其他代码仍在使用的循环。三、UserError: 同步工具内不能调用run_sync()/run_stream_sync()这是 Pydantic AI 中最具框架特色的一个防御性报错完整信息为UserError:Agent.run_sync()andAgent.run_stream_sync()cannot be used inside a synchronous tool, output function, or other function called during an agent run3.1 为什么会报错死锁风险该错误表示某个同步 tool、output function 或其他在 Agent 运行期间被调用的函数试图用run_sync()或run_stream_sync()启动一个嵌套运行。问题在于同步运行方法只能在普通应用代码中一次运行之外使用。当它在一次运行内部被调用时父运行正等待你的函数返回而嵌套的同步运行又会阻塞该函数二者互相等待即形成死锁。Pydantic AI 选择快速失败并给出指引而不是让程序挂死。3.2 源码级防御机制框架在 pydantic_ai_slim/pydantic_ai/_utils.py#L82-L97 实现了check_no_nested_sync_run()def check_no_nested_sync_run() - None: Reject sync agent entry points inside sync callbacks dispatched by Pydantic AI. ... if _in_sync_callback.get(): raise 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. )其核心是一个ContextVar_in_sync_callback同步工具、输出函数等回调经由run_in_executor分发时会标记回调所在上下文的该标志。无论回调运行在工作线程上还是在disable_threads()如 emscripten、Temporal 沙箱环境下内联执行只要标志为真就会立即抛出UserError。同步入口AbstractAgent.run_sync在真正执行前第一件事就是调用_utils.check_no_nested_sync_run()。3.3 正确修复改用async def并await把委派函数改为async def并在内部await内层运行参见 Agent delegation。父 Agent 仍然可以从普通同步应用代码中用run_sync()启动如果委派函数自身还需要执行阻塞性工作只把阻塞部分丢给asyncio.to_thread()即可。3.4 测试用例佐证仓库测试 tests/test_nested_sync_agent.py 完整覆盖了这一行为test_run_sync_from_sync_tool_is_rejected同步工具内调用run_sync()抛出UserError且守卫只作用于回调上下文——随后在普通应用代码中调用inner_agent.run_sync(hello)依然正常第 10-29 行test_run_stream_sync_from_sync_output_function_is_rejected同步输出函数内调用run_stream_sync()同样被拒第 32-45 行test_run_sync_from_sync_tool_is_rejected_when_threads_disabled在disable_threads()模拟 emscripten/Temporal 内联执行下规则依旧生效第 48-63 行test_async_tool_can_delegate_with_await用async def工具await内层运行的官方推荐模式则被允许第 66 行起。此外tests/durable_exec/test_dbos.py#L4659 与 tests/durable_exec/test_prefect.py#L4720 也验证了该守卫在 DBOS、Prefect 等持久化执行集成中保持一致行为。四、API Key 配置UserError: Set the [PROVIDER]_API_KEY environment variable ...4.1 报错含义UserError: Set the[PROVIDER]_API_KEYenvironment variable or pass it via the providersapi_key...argument当模型请求缺少对应 Provider 的密钥时触发。[PROVIDER]是占位符例如 OpenAI 对应OPENAI_API_KEY、Anthropic 对应ANTHROPIC_API_KEY。4.2 两种配置方式环境变量按 Models 页面说明设置对应[PROVIDER]_API_KEY环境变量代码传参通过 Provider 的api_key...参数显式传入。4.3 无密钥体验内置test模型如果只是想在不配置任何 API Key 的情况下体验 Pydantic AI可直接使用内置测试模型Agent(test)。该模型无需网络请求即可返回预定义的模拟响应非常适合单元测试与快速验证流程。五、监控 HTTPX 请求自定义客户端与 Logfire 集成5.1 自定义httpx2/httpx客户端可以在模型中注入自定义的httpx2或旧版httpx客户端从而在运行时访问具体的请求、响应与请求头便于调试、抓包与指标采集。5.2 结合 Logfire 的 HTTPX 集成对于生产级可观测性推荐使用logfire的 HTTPX integration 自动监控 HTTP 请求链路将模型调用、工具执行与底层 HTTP 请求统一呈现在追踪视图中快速定位超时、重试与状态码异常。六、补充同步入口在特殊事件循环下的行为除了文档列举的常见场景从源码还可以看到框架对无法由调用方驱动的事件循环如 Temporal workflow 的沙箱事件循环做了专门处理。在 pydantic_graph/pydantic_graph/_utils.py#L88-L121 中run_until_complete()会检测循环是否实现标准的run_until_complete若不支持则抛出UnsupportedEventLoopError在 Pydantic AI 层被包装为UserError提示应改用await agent.run()而非agent.run_sync()。同时该函数还会在被 Ctrl-C 中断时主动取消并清理自己的任务与连接避免任务泄漏。这一点在文档中虽未展开但有助于理解为什么同步方法在某些编排环境中不可用。七、总结排查路径速查报错根因首选修复RuntimeError: This event loop is already runningJupyter 等环境已有活动循环用顶层await agent.run()旧环境用nest_asyncio.apply()run_sync()RuntimeError: Event loop is closed来自 httpx连接池绑定已关闭的循环连同模型与 Provider 重建 Agent或传入新的http_clientUserError: cannot be used inside a synchronous tool同步回调内嵌套同步运行会死锁委派函数改为async def并await内层运行UserError: Set the [PROVIDER]_API_KEY ...缺少 API Key设置环境变量或传入api_key...体验用Agent(test)若以上问题未覆盖你的场景可查阅 docs/help.md 寻求社区支持。总体而言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-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考