如何在既能作为脚本执行、又能在Jupyter笔记本中运行的代码里使用asyncio？

编程语言 2026-07-10

我经常把Jupyter笔记本做成纯.py文件，并以 # %% 单元格分隔符来区分，如下所示：

# %%
import polars as pl
import numpy as np
# %%
df = pl.read_parquet('my_cool.parquet')
...

然后我可以通过两种方式运行：要么把它作为笔记本在 vscode-jupyter 中运行，要么在命令行把它当作脚本运行。

然而，当我在笔记本中尝试使用 asyncio 时就会出问题。Jupyter期望你在顶层写入 await（结合 autoawait 功能），但要作为脚本运行则需要在顶层使用 asyncio.run：

# %%
# this works in Jupyter, but in a script you get:
# SyntaxError: 'await' outside function
foo = await my_async_func()
# %%
# this works in a script, but in Jupyter you get:
# RuntimeError: asyncio.run() cannot be called from a running event loop
bar = asyncio.run(my_async_func)

怎么写出在两种场景下都能工作的代码呢？我尝试过的一些方法：

nest-asyncio：与我正在使用的其他一些库不兼容
在任何导入之前使用 get_ipython().autoawait = False：无法修复Jupyter
使用shebang #!/usr/bin/env -S uv run ipython：无法修复脚本

解决方案

你可以用一个辅助函数来检测事件循环是否已经在运行（就像在Jupyter中那样）。如果是，就让协程在一个独立的线程中运行，使用它自己的事件循环。而在普通脚本中，这一步将只是调用 asyncio.run()。

# %%
import asyncio
import concurrent.futures

def run(coro):
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.run(coro)
    else:
        # Jupyter: loop already running, use a separate thread
        with concurrent.futures.ThreadPoolExecutor(1) as pool:
            return pool.submit(asyncio.run, coro).result()

Then replace await/asyncio.run() with run() everywhere:

# %%
foo = run(my_async_func())

关于它实际作用的更详细说明：

在普通脚本中：get_running_loop() 会引发 RuntimeError，并回退到 asyncio.run()
在vscode-jupyter中：get_running_loop() 成功，并在一个拥有自己全新事件循环的线程中启动，从而避免“不能在正在运行的事件循环中调用”的错误

在通过Jupyter运行时，协程会在工作线程中执行。如果你的异步代码依赖于与主线程共享的线程本地状态，那么这些状态不会传递过来。对于IO密集型的工作（网络、数据库、文件）来说，这并不是问题。

我已经在我的机器上从两个场景都做过测试，但也有可能在你那里有某些原因导致它不起作用，所以如果确实不行，请随时留言说明。

Python