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-ai

Pydantic 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+）：直接使用顶层await

Jupyter/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 环境。优先推荐方案一（顶层await），nest-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 run

3.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 provider'sapi_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 中断时主动取消并清理自己的任务与连接，避免任务泄漏。这一点在文档中虽未展开，但有助于理解"为什么同步方法在某些编排环境中不可用"。

七、总结：排查路径速查

若以上问题未覆盖你的场景，可查阅 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