ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

asyncio.run() cannot be called from a running event loop:在 async 函数里该怎么改

asyncio.run() cannot be called from a running event loop:在 async 函数里该怎么改

把一个本来能在命令行运行的异步示例搬进 Web 框架的async处理函数或支持顶层await的交互环境时,常见报错是:

RuntimeError: asyncio.run() cannot be called from a running event loop

这不是“异步函数不能调用异步函数”,而是当前线程已经有正在运行的事件循环,代码又试图用asyncio.run()启动一个新的顶层循环。Python 3.14 官方文档明确规定:同一线程中已有运行中的事件循环时,不能调用asyncio.run()。排查的关键是找到谁拥有当前事件循环,再决定该用await还是由程序入口调用asyncio.run()。

适用范围与前提

下面的代码在 Windows、Python 3.14.7 上实际运行;只用标准库,不依赖 AI 模型、Web 服务或 Jupyter。它模拟 AI 应用中一个异步获取答案的函数,返回值ok是本地固定值,不是模型响应。本文只解释此处的asyncio.run()嵌套错误;其他事件循环报错仍需按堆栈和运行环境分别判断。

1. 最小复现与修复

保存为2026-11-03-asyncio-nested-run.py,运行python -X utf8 2026-11-03-asyncio-nested-run.py:

"""Reproduce and fix a nested asyncio.run() error on Python 3.14.""" import asyncio async def get_answer(): await asyncio.sleep(0) return "ok" async def main(): coro = get_answer() try: asyncio.run(coro) # Wrong: main() already runs in an event loop. except RuntimeError as exc: print(f"reproduced: {type(exc).__name__}: {exc}") finally: coro.close() # The rejected coroutine was never awaited. print(f"fixed: {await get_answer()}") if __name__ == "__main__": asyncio.run(main())

我在本机得到的实际输出:

reproduced: RuntimeError: asyncio.run() cannot be called from a running event loop fixed: ok

脚本退出码为 0,是因为示例有意捕获了第一个异常,然后继续运行正确写法。coro.close()只是清理这次被拒绝执行的协程对象,避免另一个“coroutine was never awaited”警告;不要把它当作业务代码的修复方式。若删掉except,异常会向外传播,进程会失败。

2. 原因:程序入口和已有事件循环的职责不同

asyncio.run(main())适合放在普通 Python 脚本的顶层入口。它负责创建、运行和关闭事件循环。进入main()后,事件循环已经在当前线程运行;此时再次调用asyncio.run(get_answer())就触发上述限制。Python 文档也建议把asyncio.run()作为程序主入口,理想情况下只调用一次。

在已经运行的async def中,直接await get_answer()。如果需要并发启动并稍后等待,可以用asyncio.create_task(),但仍应持有任务引用并处理结果或取消。简单顺序调用只用await即可,不需要额外建任务。

如果你在 Web 框架的异步路由中遇到此错误,通常应删除路由内部的asyncio.run()并await目标协程。若你在普通同步脚本入口调用异步函数,则保留顶层的asyncio.run(main())。这两种位置不能机械地互换。

3. Jupyter / IPython 场景要单独看

IPython 文档说明它支持顶层await,但这个功能具有实验性,Notebook 还涉及 IPykernel 版本与运行方式差异。因此,在支持该功能的单元格中,通常写await get_answer(),而不是再调用asyncio.run(get_answer())。普通.py文件的模块顶层不能直接照抄这种写法;仍需使用async def main()加顶层asyncio.run(main())。

本文没有在 Notebook 中实测,以上范围依据 IPython 官方文档;如果你使用特定 Jupyter/IPykernel 版本,请先确认该环境的顶层await行为。

4. 验证自己的项目时怎么查

  1. 从异常堆栈定位自己代码里的asyncio.run(...),看它是否位于async def、异步回调或交互环境单元格中。
  2. 若调用者已经是async def,先改为await target(),重新运行并检查返回值与异常传播。本文的fixed: ok就是这个验证。
  3. 若调用者是同步入口,确认它没有被已有事件循环调用,再将asyncio.run(main())留在入口处。
  4. 若修复后还有coroutine was never awaited,检查是否只调用了target()却忘记await或创建任务;这与嵌套asyncio.run()是两个问题。

常见误区

  • 把run_until_complete()换上去试。它同样试图控制事件循环,无法解决“当前循环已经在运行”的根因;先辨认调用位置。
  • 把警告当成主错误。嵌套调用先抛RuntimeError,传入的协程未执行,随后可能出现未等待警告。本文用close()只为让复现输出聚焦。
  • 在所有地方都写await。await只能在允许的异步语法环境中使用;普通脚本入口仍需由运行器启动协程。
  • 看到ok就认为真实 AI 请求成功。本例只有asyncio.sleep(0)和固定字符串,没有网络请求,也不验证模型 SDK。

一手资料

  • Python 3.14:asyncio runners
  • Python 3.14:coroutines and tasks
  • Python 3.14:event loop
  • IPython:Autoawait
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进