把一个本来能在命令行运行的异步示例搬进 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. 验证自己的项目时怎么查
- 从异常堆栈定位自己代码里的
asyncio.run(...),看它是否位于async def、异步回调或交互环境单元格中。 - 若调用者已经是
async def,先改为await target(),重新运行并检查返回值与异常传播。本文的fixed: ok就是这个验证。 - 若调用者是同步入口,确认它没有被已有事件循环调用,再将
asyncio.run(main())留在入口处。 - 若修复后还有
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