Python asyncio事件循环嵌套陷阱：pydantic-ai run_sync报错排查与解决

1. 这个报错为什么总在“跑完之后”才出现

第一次遇到RuntimeError: asyncio.run() cannot be called from a running event loop的时候，我盯着屏幕愣了好几秒。代码明明已经跑完了，结果在收尾阶段给我来这么一下，那种感觉就像你吃完饭准备结账，服务员告诉你厨房着火了。更让人抓狂的是，这个报错往往出现在你调用run_until_complete或者run_sync之后，程序看起来一切正常，日志也打完了，突然就给你抛一个异常出来。

这个现象在pydantic-ai这类框架里特别常见。你写了一个 agent，调用了run_sync，数据也拿到了，结果在某个回调或者清理阶段，RuntimeError就冒出来了。很多人第一反应是“我代码写错了”，但实际上，问题往往出在事件循环的生命周期管理上。

先把这个问题的本质说清楚：Python 的asyncio事件循环是一个“单例”性质的东西。一个线程里只能有一个正在运行的事件循环。当你调用asyncio.run()的时候，它会创建一个新的事件循环，跑完协程之后关闭它。但如果你已经在一个事件循环里了，再调用asyncio.run()，它就会直接抛RuntimeError。而run_until_complete和run_sync这类方法，本质上也是在操作事件循环，只是它们的行为更“隐蔽”一些。

那为什么报错总在“跑完之后”？因为很多框架在内部做了异步到同步的桥接。比如pydantic-ai的run_sync，它内部会检查当前线程有没有正在运行的事件循环。如果没有，它就自己创建一个，跑完关闭；如果有，它就用run_until_complete往现有循环里塞任务。问题就出在这个“检查”和“使用”之间的时间窗口上。如果你的代码在run_sync返回之后，又触发了某个异步回调，而这个回调试图再次操作事件循环，就会撞上已经关闭或者正在关闭的循环，于是RuntimeError就来了。

还有一个更隐蔽的场景：你在 Jupyter Notebook 或者某些 IDE 的交互环境里跑代码。这些环境本身就在一个事件循环里运行。你调用run_sync，它检测到有循环，就用run_until_complete把任务塞进去。任务跑完了，你的代码继续往下走，但那个循环还在。等到你下一次再调用run_sync，或者某个后台任务试图清理资源的时候，就会发现循环的状态不对劲了。

所以这个报错的核心矛盾是：同步代码和异步代码在同一个线程里争夺事件循环的控制权。谁先拿到，谁后释放，释放的时候有没有清理干净，都会影响最终结果。而“跑完之后”才报错，恰恰说明问题不在业务逻辑本身，而在资源清理和状态恢复的阶段。

2. 从 asyncio 到 pydantic-ai：事件循环的嵌套陷阱

2.1 asyncio.run 和 run_until_complete 的本质区别

很多人把asyncio.run()和loop.run_until_complete()当成差不多的东西，觉得都是“跑一个协程”。但它们的语义完全不同。

asyncio.run()是一个入口函数。它的设计意图是“从同步世界进入异步世界，跑完再回来”。它做了几件事：创建一个新的事件循环，把它设为当前线程的循环，跑传入的协程，取消所有剩余任务，关闭循环，最后把线程的循环引用清掉。整个过程是封闭的，像一个一次性容器。

loop.run_until_complete()是一个循环方法。它假设循环已经存在，并且可能已经在运行或者即将运行。它只是把协程包装成任务，塞进循环里，然后一直跑到这个任务完成。它不会关闭循环，也不会清理其他任务。

这两个东西混用的后果就是：你用asyncio.run()创建了一个循环，跑完了，循环关闭了。但某个地方还持有这个循环的引用，或者某个后台任务还在往这个已经关闭的循环里塞东西。等到下一次事件循环被创建或者被引用的时候，状态就乱了。

在pydantic-ai里，run_sync的实现大致是这样的逻辑：

def run_sync(self, *args, **kwargs): try: loop = asyncio.get_running_loop() except RuntimeError: loop = None if loop and loop.is_running(): # 已经有循环在跑，用 run_until_complete return loop.run_until_complete(self.run(*args, **kwargs)) else: # 没有循环，用 asyncio.run return asyncio.run(self.run(*args, **kwargs))

这个逻辑看起来没问题，但它忽略了一种情况：循环存在但没有在运行。比如在某些测试框架或者交互环境里，循环被创建了，但处于暂停状态。这时候get_running_loop()会抛异常，代码走asyncio.run()分支，创建一个新循环。但旧循环还在，而且可能持有一些资源。等新循环跑完关闭，旧循环被重新激活的时候，就会出问题。

2.2 pydantic-ai 的 run_sync 到底做了什么

pydantic-ai是一个用于构建 AI agent 的框架，它的核心抽象是Agent。你定义一个 agent，给它一个模型，然后调用run_sync来同步执行。这个run_sync内部会把同步调用转换成异步执行，因为底层的模型调用、工具调用都是异步的。

问题在于，pydantic-ai的 agent 在执行过程中可能会触发多个异步操作：调用模型 API、执行工具函数、处理流式响应、运行回调。这些操作都在同一个事件循环里。当run_sync返回的时候，它只保证主协程完成了，但不保证所有后台任务都清理干净了。

我遇到过一种情况：agent 里定义了一个工具函数，这个工具函数内部又调用了另一个run_sync。这就形成了嵌套。外层run_sync创建了一个循环，内层run_sync检测到有循环在跑，就用run_until_complete塞任务。内层任务跑完了，但外层循环还在。等外层任务也跑完，asyncio.run()开始清理循环的时候，发现内层可能还留了一些未完成的任务或者回调，于是RuntimeError就出现了。

这种嵌套在pydantic-ai里特别容易发生，因为 agent 的工具调用是异步的，但很多工具函数是同步写的。框架为了兼容同步工具，会在内部做桥接。桥接的次数多了，事件循环的状态就变得难以追踪。

2.3 为什么“跑完之后”才报错

“跑完之后”报错，说明错误发生在清理阶段而不是执行阶段。Python 的asyncio.run()在关闭循环之前，会做几件事：

1. 取消所有剩余的 Task
2. 等待所有被取消的 Task 完成
3. 关闭异步生成器
4. 关闭事件循环
5. 关闭默认的 executor

如果在这个过程中，有任何代码试图往循环里塞新任务，或者引用已经关闭的循环，就会抛RuntimeError。而“跑完之后”这个时间点，恰恰是清理逻辑集中执行的时候。

一个典型的场景是：你在 agent 里用了asyncio.create_task()创建了一个后台任务，但没有 await 它。主协程跑完了，asyncio.run()开始清理，发现这个后台任务还在跑，就取消它。如果这个后台任务在取消的时候又触发了新的异步操作，就会撞上正在关闭的循环。

另一个场景是：你用了asyncio.get_event_loop()获取了循环引用，把它存到了一个全局变量或者闭包里。主协程跑完，循环关闭了，但你的引用还在。后续代码再试图用这个引用做事情，就会报错。

所以“跑完之后”报错，本质上是一个生命周期管理问题。你的代码在时间维度上越界了：它在循环已经结束之后，还在试图操作循环。

3. 实操排查：定位 RuntimeError 的真正来源

3.1 用最小复现锁定问题边界

遇到这种报错，第一步不是改代码，而是复现。但复现不是把整个项目跑一遍，而是把问题剥离出来，写一个最小的测试用例。

我的做法是：新建一个空文件，只保留最核心的调用链。比如：

from pydantic_ai import Agent agent = Agent('openai:gpt-4') result = agent.run_sync('Hello') print(result.data)

如果这个最小用例不报错，就逐步往里加东西：加一个工具函数，加一个回调，加一个嵌套调用。每加一个，跑一次。直到报错出现，你就找到了触发条件。

如果最小用例本身就报错，那问题就在框架层面，可能是版本兼容性问题。这时候可以试试降级或者升级pydantic-ai，看看报错是否消失。

我试过的一个案例是：在 FastAPI 的路由函数里调用agent.run_sync()。FastAPI 本身是异步框架，路由函数如果是async def，它就在一个事件循环里跑。这时候run_sync检测到有循环，就用run_until_complete。但 FastAPI 的循环是长期运行的，run_until_complete塞进去的任务跑完之后，可能会留下一些未清理的回调。等请求结束，FastAPI 试图清理这个请求的上下文时，就撞上了RuntimeError。

解决方案是把路由函数改成同步的def，让 FastAPI 在线程池里跑它。这样run_sync就会走asyncio.run()分支，创建一个独立的新循环，跑完就关，不影响 FastAPI 的主循环。

3.2 用日志和 traceback 定位清理阶段

RuntimeError的 traceback 有时候会指向框架内部，看起来跟你写的代码没关系。这时候不要慌，仔细看 traceback 的最后几行。通常它会告诉你是在哪个__del__、哪个close()、哪个cancel()里出的错。

我习惯在关键位置加日志：

import asyncio import logging logging.basicConfig(level=logging.DEBUG) loop = asyncio.get_event_loop() print(f"Loop running: {loop.is_running()}") print(f"Loop closed: {loop.is_closed()}")

在run_sync调用前后各打一次，看看循环的状态变化。如果调用前循环是 running，调用后变成 closed，那说明run_sync内部把循环关掉了。如果调用前是 closed，调用后还是 closed，但报错说“cannot be called from a running event loop”，那就说明有另一个循环在别的地方被创建了。

还有一种情况是：报错信息里出现了cudagetdevicecount这样的字眼。这通常跟 GPU 相关，比如 PyTorch 或者 CUDA 的初始化。有些 AI 框架在导入的时候会初始化 CUDA，而 CUDA 的初始化可能会创建自己的线程和事件循环。如果你的代码在 CUDA 初始化完成之前就跑了asyncio.run()，等 CUDA 初始化的时候，它可能会试图往一个已经关闭的循环里塞东西。

这种问题的排查方法是：把 CUDA 相关的导入和初始化放到最前面，确保它们在事件循环创建之前完成。或者用torch.cuda.is_available()先探测一下，避免在异步上下文里做重量级的初始化。

3.3 常见触发场景速查表

这张表是我踩了无数次坑之后总结出来的。每次遇到新的RuntimeError，先对照这张表，看看能不能对上号。大部分情况下，问题都出在“循环的创建和销毁不在同一个作用域里”。

4. 解决方案：从临时补丁到架构调整

4.1 临时方案：nest_asyncio 和 loop 复用

如果你只是想在 Jupyter Notebook 里快速跑通代码，nest_asyncio是最省事的方案。它通过猴子补丁的方式，让事件循环支持嵌套运行。

import nest_asyncio nest_asyncio.apply()

打上这个补丁之后，asyncio.run()在已有循环的情况下也能跑，它会复用现有的循环而不是创建新的。但要注意，nest_asyncio是一个全局补丁，它会影响整个进程的事件循环行为。在生产环境里用要谨慎，因为它可能掩盖一些深层次的问题。

另一个临时方案是手动管理循环：

import asyncio loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete(agent.run('Hello')) finally: loop.close()

这个方案的思路是：不依赖run_sync的内部逻辑，自己创建循环、自己跑、自己关。这样你对循环的生命周期有完全的控制权。缺点是代码量多了，而且如果 agent 内部还有嵌套的异步调用，仍然可能出问题。

我个人的经验是：nest_asyncio适合原型验证和调试，不适合生产。生产环境要么全同步，要么全异步，不要混着来。

4.2 推荐方案：统一异步接口，避免 run_sync

最彻底的解决方案是不要用run_sync。pydantic-ai提供了异步的run方法，直接用await调用：

async def main(): result = await agent.run('Hello') print(result.data) asyncio.run(main())

这样整个调用链都是异步的，事件循环只有一个，生命周期清晰。没有桥接，没有嵌套，没有清理阶段的意外。

如果你用的是 FastAPI，路由函数直接写成async def：

@app.post("/chat") async def chat(message: str): result = await agent.run(message) return {"reply": result.data}

FastAPI 自己管理事件循环，你的代码只需要 await。这样既高效又安全。

如果你必须从同步代码里调用异步 agent，比如在一个 Django 视图或者一个 CLI 脚本里，可以用asyncio.run()包一层：

def sync_chat(message: str): return asyncio.run(agent.run(message))

但要注意，这个sync_chat函数不能在已经有事件循环的上下文里调用。如果调用方本身就在异步环境里，那就应该把整个调用链改成异步。

4.3 架构层面：把异步边界画清楚

从架构角度看，RuntimeError频发的根本原因是异步边界不清晰。一个项目里，哪些部分是同步的，哪些部分是异步的，应该在设计阶段就定好。

我的建议是：

• 入口层保持同步：CLI、定时任务、消息队列消费者，这些入口用同步代码写，内部用asyncio.run()调用异步核心。
• 核心层保持异步：agent 的执行、模型调用、工具调用，全部用async def定义，用await调用。
• 桥接层只做转换：如果必须从同步调异步，写一个薄薄的桥接函数，只负责创建循环和跑协程，不掺业务逻辑。
• 禁止嵌套桥接：不要在异步函数里调用同步桥接函数，也不要在同步桥接函数里再调用另一个同步桥接函数。

这样画清楚之后，事件循环的创建和销毁就集中在入口层，核心层不关心循环的存在，只关心 await。清理阶段的问题自然就少了。

还有一个细节：pydantic-ai的 agent 在执行过程中会创建一些内部任务，比如流式响应的处理、工具调用的调度。这些任务在run返回之前应该都已经完成了。如果你在run返回之后还持有 agent 的引用，并且试图用它做事情，可能会触发一些延迟的清理逻辑。所以用完 agent 之后，及时释放引用，不要长期持有。

5. 避坑心得与常见问题实录

5.1 那些年我踩过的循环坑

第一个坑：在__del__里调用异步清理。我写过一个类，在__del__方法里试图关闭一个异步连接。结果__del__被调用的时候，事件循环可能已经关了，或者正在关。RuntimeError直接糊脸。后来改成显式的close()方法，让调用方负责清理，问题就没了。

第二个坑：用asyncio.get_event_loop()获取循环。这个函数在 Python 3.10 之后行为变了。如果没有正在运行的循环，它会抛DeprecationWarning或者RuntimeError。正确的做法是用asyncio.get_running_loop()，它只在有运行中的循环时返回，否则抛异常，语义更清晰。

第三个坑：在信号处理函数里操作循环。我写过一个脚本，用signal.signal注册了一个 SIGINT 处理器，在处理器里试图取消所有任务并关闭循环。结果信号来的时候，循环可能正在跑某个协程，取消操作和关闭操作撞在一起，直接死锁。后来改成在信号处理器里只设置一个标志位，让主循环自己检查标志位并优雅退出。

第四个坑：多线程里共享事件循环。asyncio的事件循环是线程绑定的。你在主线程创建的循环，不能在子线程里直接用。如果要在子线程里跑异步代码，必须用asyncio.run_coroutine_threadsafe()把协程提交到主线程的循环里。我试过在子线程里直接asyncio.run()，结果主线程的循环和子线程的循环互相干扰，报错信息完全看不懂。

第五个坑：测试框架里的事件循环。pytest默认不支持异步测试，需要装pytest-asyncio。但即使装了，如果测试用例里混用了同步和异步的 fixture，事件循环的创建和销毁时机也会变得很微妙。我的做法是：异步测试用例全部用@pytest.mark.asyncio标记，同步测试用例不碰任何异步代码，两边完全隔离。

5.2 常见问题速查

问题一：为什么第一次跑没事，第二次跑就报错？

因为第一次跑的时候，asyncio.run()创建了一个新循环，跑完关闭了。但某个地方（比如全局变量、类属性、闭包）还留着这个循环的引用。第二次跑的时候，代码试图复用这个已经关闭的循环，就报错了。解决方法是找到那个引用，把它清掉，或者改成每次重新创建循环。

问题二：为什么在本地跑没事，部署到服务器就报错？

本地环境和服务器环境的差异可能在于：Python 版本、依赖库版本、是否有 GPU、是否用了某种 ASGI 服务器。比如本地用 Flask 开发服务器，它是同步的；服务器用 Uvicorn，它是异步的。同步环境里run_sync走asyncio.run()分支，异步环境里走run_until_complete分支，行为完全不同。排查的时候要确认部署环境的异步栈是什么。

问题三：报错信息里有 cudagetdevicecount，跟事件循环有关系吗？

有关系。CUDA 的初始化是同步的，而且可能会创建自己的线程。如果 CUDA 初始化发生在事件循环创建之后，它可能会试图往循环里注册一些回调。等循环关闭的时候，这些回调就成了孤儿，触发RuntimeError。解决方法是在创建事件循环之前完成 CUDA 初始化，比如在模块导入的时候就调用torch.cuda.is_available()。

问题四：用了 nest_asyncio 之后不报错了，但性能变差了，为什么？

nest_asyncio通过复用现有循环来避免创建新循环，但它的实现方式是在循环的run_until_complete里做递归调用。这会导致循环的调用栈变深，而且一些清理逻辑被跳过。短期用没问题，长期用会积累未清理的任务和回调，最终导致内存泄漏或者性能下降。生产环境不建议用。

问题五：怎么判断一个函数是不是在事件循环里跑的？

import asyncio def is_in_event_loop(): try: asyncio.get_running_loop() return True except RuntimeError: return False

这个函数返回 True 就说明当前线程有正在运行的事件循环。你可以用它来做条件分支：在循环里就用 await，不在循环里就用 asyncio.run。

5.3 一个真实的排查案例

最后分享一个我最近遇到的案例。项目是一个 FastAPI 服务，内部用pydantic-ai做 agent 调度。测试环境一切正常，生产环境偶发RuntimeError，报错信息指向run_until_complete。

排查过程：

1. 先看日志，发现报错都发生在请求量比较大的时候。请求量小的时候不报错。
2. 怀疑是并发问题。FastAPI 的异步路由是并发执行的，多个请求可能同时操作同一个事件循环。
3. 检查代码，发现 agent 是全局单例，多个请求共享同一个 agent 实例。
4. pydantic-ai的 agent 内部有一些状态，比如消息历史、工具调用记录。多个请求同时操作这些状态，可能导致内部任务调度混乱。
5. 解决方案：每个请求创建一个新的 agent 实例，或者用contextvars隔离状态。改完之后，报错消失。

这个案例的教训是：异步框架里的共享状态要特别小心。同步代码里，多个线程共享一个对象，你会想到加锁。异步代码里，多个协程共享一个对象，虽然没有线程切换，但 await 点就是切换点。在 await 前后，对象的状态可能已经被其他协程改了。所以要么用不可变对象，要么用contextvars做隔离，要么每个协程用独立的实例。

还有一个细节：pydantic-ai的 agent 在run的时候会创建一个内部的RunContext，这个上下文在run结束后应该被释放。如果 agent 是全局的，RunContext可能被多个请求复用，导致清理逻辑错乱。所以即使性能上有点损失，每个请求新建 agent 也是值得的。

6. 写在最后：一些个人体会

RuntimeError这类报错，最让人头疼的地方在于它的非确定性。同样的代码，有时候跑得好好的，有时候就崩了。这种不确定性会让人怀疑人生。但只要你理解了事件循环的生命周期，理解了同步和异步的边界，大部分问题都能定位到具体的代码行。

我的经验是：不要跟事件循环较劲。如果你发现自己在写复杂的循环管理代码，那大概率是架构出了问题。好的异步代码应该是简单的：入口处asyncio.run()，内部全是await，没有嵌套，没有桥接，没有全局循环引用。

另外，pydantic-ai这类框架在快速迭代，版本之间的行为可能有差异。遇到RuntimeError的时候，先查一下框架的 issue 列表，看看是不是已知问题。如果是，升级或者降级往往比改代码更有效。

最后再分享一个小技巧：在开发阶段，把PYTHONASYNCIODEBUG=1环境变量打开。它会让 asyncio 输出更详细的调试信息，包括未等待的协程、未关闭的循环、慢回调等。这些信息对定位RuntimeError非常有帮助。生产环境记得关掉，因为它会影响性能。

这个内容后续还可以这样扩展：如果你用的是其他异步框架，比如httpx、aiohttp、asyncpg，它们的事件循环管理策略各有不同。可以针对每个框架单独写一篇排查指南。另外，Python 3.12 对 asyncio 做了一些改进，比如更好的任务组管理和异常处理，值得关注一下升级后的行为变化。