Python 异步编程完整指南：从迭代协议到 Coroutine、Task 与线程池

写在前面：这篇文章要解决什么问题

很多 Python 异步教程从 async def 和 await 开始，却没有先解释这些概念之间的关系。结果往往是：代码可以照着写，一旦遇到顺序执行、阻塞、取消或线程池，心智模型就开始混乱。

例如：

async def main() -> None:
    await fetch_a()
    await fetch_b()

读完本文后，你应该能够准确回答：

• 调用 async def 函数后，函数体为什么没有立即执行？
• await 后面究竟可以放什么对象？
• 为什么连续两次 await 通常是顺序执行？
• 为什么事件循环不能跳过第一行，直接运行同一协程的下一行？
• Coroutine、Task 和 Future 分别解决什么问题？
• async for、async with 背后依赖什么协议？
• 为什么异步程序仍然需要线程池或进程池？
• FastAPI、LLM 流式输出和 Agent 工具调用中应该怎样组织并发？

本文面向 Python 3.11+，其中 TaskGroup 和 asyncio.timeout() 从 Python 3.11 开始提供；文中的技术细节依据 Python 3.14.6 官方文档核对。¹

先记住六句话

async def main() -> None:
    await fetch_a()
    await fetch_b()

1. 调用普通函数会同步执行函数体；调用协程函数只会创建协程对象。
2. await 等待的是 awaitable；常见 awaitable 包括 Coroutine、Task 和 Future。
3. await 只会暂停当前 Task，不会跳到当前协程尚未执行的下一行。
4. 连续直接 await 通常表示顺序控制流；并发需要先创建多个 Task，或交给 TaskGroup / gather()。
5. async def 不会把同步阻塞代码自动变成非阻塞代码。
6. asyncio 主要擅长 I/O 并发；CPU 密集型工作通常需要进程、多解释器、专用 Worker，或能真正并行执行的运行环境。

这六句话就是整篇文章的主线。

一、先建立正确的并发心智模型

1. 事件循环、Task 与协作式调度

典型的 asyncio 程序可以先简化为：

一个事件循环线程
+
多个可以暂停和恢复的 Task
+
I/O 完成通知、定时器与回调

事件循环通常在一个线程中运行，并在这个线程里执行回调和 Task。同一时刻，一个事件循环线程只会执行一个 Task；当前 Task 运行到能够挂起的 await 时，事件循环才有机会运行其他已就绪的 Task。这个模型叫作协作式调度。^{2 1}

可以把它想象成只有一名厨师的厨房：

• 切菜、翻锅等操作需要厨师亲自执行；
• 汤放到炉子上等待烧开时，厨师可以去处理另一道菜；
• 但如果厨师开始做一项持续五分钟、完全不让手的工作，其他菜就只能等。

这个比喻的边界也很重要：

事件循环不会阅读源码并推测“下一行也许可以先执行”。它只能调度已经存在的 Task、回调和就绪事件。

2. 并发不等于并行

asyncio 的核心价值通常是：当一个 Task 在等待网络、数据库或定时器时，让其他 Task 使用事件循环线程。

它并不意味着两个 Python 函数一定在两个 CPU 核心上同时执行。

3. 阻塞与挂起不是一回事

import time

time.sleep(5)

这会阻塞当前线程。假如当前线程就是事件循环线程，其他 Task 也无法继续运行。

import asyncio

await asyncio.sleep(5)

这会挂起当前 Task。事件循环线程仍然可以运行其他 Task。Python 官方文档明确说明，asyncio.sleep() 总会挂起当前 Task，让其他 Task 获得运行机会。¹

因此，异步编程最重要的判断不是“代码里有没有 async”，而是：

当前执行路径在等待期间，是否把事件循环线程还给了调度器？

二、先理解 Python 的四种函数形态

def / async def 与“函数体中是否出现 yield”是两个独立维度。

决定函数类别的是 def、async def 与 yield，而不是有没有写 return。Python 语言参考规定：普通函数体中出现 yield，它就成为生成器函数；async def 中出现 yield，它就成为异步生成器函数。^{3 4}

1. 普通函数

def add(a: int, b: int) -> int:
    return a + b


result = add(1, 2)

调用 add() 时，函数体会在当前调用栈中同步执行。没有显式 return 时，普通函数返回 None。

2. 生成器函数

def countdown(n: int):
    while n > 0:
        yield n
        n -= 1

调用生成器函数不会一次执行完整函数体：

generator = countdown(3)

此时得到的是生成器对象。第一次调用 next(generator) 时，函数才开始运行到第一个 yield：

next(generator)  # 3
next(generator)  # 2
next(generator)  # 1

生成器会保存局部变量、指令位置和异常处理状态，下次恢复时从上一个 yield 后继续执行。³

3. 协程函数

async def fetch_data() -> str:
    return "done"

调用：

coroutine = fetch_data()

得到的是协程对象，函数体还没有被调度执行。仅仅创建协程对象不会让它自动运行；要真正推进它，需要 await、创建 Task，或把它交给其他能够运行 awaitable 的 API。¹

4. 异步生成器函数

async def stream_messages():
    await asyncio.sleep(1)
    yield "第一条"

    await asyncio.sleep(1)
    yield "第二条"

调用后得到异步生成器对象：

stream = stream_messages()

使用 async for 消费：

async for message in stream:
    print(message)

异步生成器的核心价值是：

它可以逐项产出结果，而且在准备下一项的过程中可以执行异步等待。

三、从 Iterable、Iterator 到 Generator

理解普通迭代协议，是理解 async for 的最好入口。

1. Iterable：可迭代对象

Iterable 表示“可以通过 iter(obj) 得到迭代器的对象”。

常见 Iterable 包括 list、tuple、dict、set、str 和 range。

例如：

numbers = [10, 20, 30]
iterator = iter(numbers)

很多容器每次调用 iter() 都会返回一个新的迭代器，因此可以多次独立遍历。不过，“可重复遍历”不是 Iterable 协议本身的硬性保证。

2. Iterator：维护遍历位置的对象

Iterator 需要支持：

__iter__()
__next__()

其中：

• __iter__() 返回迭代器本身；
• __next__() 返回下一项；
• 没有更多元素时抛出 StopIteration。

numbers = [10, 20, 30]

it1 = iter(numbers)
it2 = iter(numbers)

next(it1)  # 10
next(it1)  # 20
next(it2)  # 10

it1 和 it2 保存各自的遍历位置。

Iterator 通常是一次性的：

it = iter([1, 2, 3])

list(it)  # [1, 2, 3]
list(it)  # []

第一次遍历已经把游标推进到末尾。

3. Generator 是一种 Iterator

生成器对象实现了迭代器协议，因此关系可以记成：

Generator 是一种 Iterator
Iterator 本身也是 Iterable

生成器的主要价值是：

• 延迟计算；
• 逐项产出；
• 降低峰值内存；
• 自然表达数据流。

4. for 循环的概念展开

for item in source:
    consume(item)

概念上接近：

iterator = iter(source)

while True:
    try:
        item = next(iterator)
    except StopIteration:
        break

    consume(item)

for 本质上是同步迭代协议的语法糖。

四、Coroutine、Task、Future 与 Awaitable

1. 协程函数与协程对象

async def load_user() -> dict[str, str]:
    return {"name": "Andy"}

这里的 load_user 是协程函数。

coroutine = load_user()

这里的 coroutine 是协程对象。

可以通过 inspect 检查：

import inspect

inspect.iscoroutinefunction(load_user)  # True
inspect.iscoroutine(coroutine)          # True
inspect.isawaitable(coroutine)          # True

如果创建了协程对象，却既不等待也不调度：

load_user()

程序通常会在对象被回收时发出：

RuntimeWarning: coroutine was never awaited

Python 官方文档明确说明：仅仅调用协程函数不会调度它执行。^{1 2}

2. Awaitable：可以放在 await 后面的对象

Awaitable 可以理解为：

能够被 await，并最终产生结果、异常或取消状态的对象。

result = await awaitable

在 asyncio 中，最常见的三类 awaitable 是：

Coroutine
Task
Future

Python 数据模型还允许对象通过 __await__() 协议自定义 awaitable；该方法必须返回一个迭代器。⁴

3. 三者分别解决什么问题

可以先用这组近似理解：

Coroutine：一段可暂停和恢复的异步执行
Task：把 Coroutine 包装成事件循环可以独立调度的工作
Future：保存“未来结果”的低层状态容器

Coroutine

协程对象描述执行逻辑，但它本身不是一个独立调度单位。

result = await load_user()

直接等待协程时，当前 Task 会进入并推进这个协程，直到它完成，或等待一个尚未完成的对象。

Task

Task 会把协程交给事件循环调度：

task = asyncio.create_task(load_user())

Task 负责：

推进协程
→ 在 await 处挂起
→ 等待依赖完成
→ 恢复协程
→ 保存返回值、异常或取消状态

Task 本身也是 awaitable：

user = await task

它继承了大部分 Future 接口，例如 done()、result()、exception() 和 add_done_callback()，但不能由调用者手动执行 set_result() 或 set_exception()；Task 的结果必须来自内部协程。¹

Future

Future 通常不负责执行函数，它主要表示：

某个结果以后会准备好，但现在还没有完成。

PENDING
→ FINISHED（结果或异常）
→ CANCELLED

Future 是连接底层回调式代码与上层 async / await 的桥梁。普通业务代码通常不需要直接创建或暴露裸 Future。⁵

4. 一个非常实用的区别：协程对象通常只能等待一次

同一个原生协程对象不能重复 await：

coroutine = load_user()

user1 = await coroutine
user2 = await coroutine  # RuntimeError

如果一个结果需要被多个位置等待，应先创建 Task：

task = asyncio.create_task(load_user())

user1 = await task
user2 = await task

asyncio.Future 可以被多次等待，并返回同一个完成结果；Task 继承了这种 Future-like 行为。^{4 5}

五、await 到底做了什么

1. 第一步：先计算右侧表达式

result = await operation()

执行顺序首先是：

awaitable = operation()

然后 Python 才尝试等待这个返回对象。

因此，await 不是“把右侧函数自动扔到后台”。右侧表达式仍然会按照普通表达式求值规则执行。

2. 第二步：根据 awaitable 的状态决定是否挂起

对象尚未完成

当前 Task 执行 await
→ 当前 Task 挂起
→ 事件循环运行其他已就绪 Task、回调或 I/O
→ awaitable 完成
→ 当前 Task 恢复
→ await 表达式产生结果或抛出异常

对象已经完成

await 可能立即取得结果并继续执行，不一定发生可观察到的任务切换。

因此，准确说法是：

await 提供挂起当前 Task 的能力，但不是每个 await 都必然切换到另一个 Task。

3. 为什么 await time.sleep(5) 既阻塞又报错

错误代码：

import time


async def example() -> str:
    await time.sleep(5)
    return "done"

真实执行顺序是：

1. 同步调用 time.sleep(5)
2. 当前线程被阻塞 5 秒
3. time.sleep() 返回 None
4. Python 尝试执行 await None
5. 抛出 TypeError

报错类似：

TypeError: object NoneType can't be used in 'await' expression

正确写法：

import asyncio


async def example() -> str:
    await asyncio.sleep(5)
    return "done"

两者的本质区别是：

time.sleep(5)
= 阻塞当前线程

await asyncio.sleep(5)
= 挂起当前 Task

4. async def 内部仍然可以阻塞

下面的函数虽然使用了 async def，但其中没有任何能够让出事件循环的操作：

async def calculate() -> int:
    total = 0
    for number in range(50_000_000):
        total += number * number
    return total

一旦这个协程开始运行，它会持续占用事件循环线程，直到循环结束。

所以：

async def 只声明了协程函数，不保证函数体天然非阻塞。

六、为什么连续 await 是顺序执行

async def main() -> None:
    result_a = await fetch_a()
    result_b = await fetch_b()

时间线是：

main Task
  |
  |-- 调用 fetch_a()，得到协程对象
  |-- 进入并推进 fetch_a
  |-- 等待 A，main Task 挂起
  |
  |   事件循环可以运行其他已经存在的 Task
  |   但 fetch_b() 尚未被调用
  |
  |-- A 完成，main Task 恢复
  |-- 执行下一行，调用 fetch_b()
  |-- 等待 B

关键点是：

fetch_b() 属于同一个 main Task 的后续控制流。第一行没有完成时，第二行还没有执行，B 的协程对象和 Task 都不存在。

事件循环可以切换到其他 已经调度 的 Task，却不能跳到当前 Task 尚未执行的下一条语句。

七、怎样真正并发运行多个协程

1. create_task()：先创建独立 Task

task_a = asyncio.create_task(fetch_a())
task_b = asyncio.create_task(fetch_b())

result_a = await task_a
result_b = await task_b

前两行已经创建两个独立 Task：

Task A
Task B

当 main Task 等待 task_a 时，事件循环仍然可以推进 task_b。

即使代码先写：

await task_a

也不代表 Task B 没有运行。它已经被调度，能够在 A 等待 I/O 时继续推进。

2. 可运行的时间对比示例

import asyncio
import time


async def fetch(name: str, delay: float) -> str:
    print(f"{name} started")
    await asyncio.sleep(delay)
    print(f"{name} finished")
    return name


async def sequential() -> None:
    started = time.perf_counter()

    await fetch("A", 2)
    await fetch("B", 2)

    print(f"sequential: {time.perf_counter() - started:.1f}s")


async def concurrent() -> None:
    started = time.perf_counter()

    task_a = asyncio.create_task(fetch("A", 2))
    task_b = asyncio.create_task(fetch("B", 2))

    await task_a
    await task_b

    print(f"concurrent: {time.perf_counter() - started:.1f}s")


async def main() -> None:
    await sequential()  # 约 4 秒
    await concurrent()  # 约 2 秒


asyncio.run(main())

这仍然通常是单线程 I/O 并发，不意味着两个协程在两个 CPU 核心上并行计算。

3. TaskGroup：相关并发任务的首选

Python 3.11+ 提供 asyncio.TaskGroup：

import asyncio


async def main() -> None:
    async with asyncio.TaskGroup() as group:
        task_a = group.create_task(fetch_a())
        task_b = group.create_task(fetch_b())

    result_a = task_a.result()
    result_b = task_b.result()

退出 async with 时，TaskGroup 会等待组内任务完成。

如果某个子任务以 CancelledError 以外的异常失败，TaskGroup 会取消剩余任务，等待它们结束，再以 ExceptionGroup 或 BaseExceptionGroup 汇总异常。¹

它的优势在于：

• 子任务生命周期被限制在清晰的代码块中；
• 相关任务失败时能够协同取消；
• 不容易遗漏 Task 或未处理异常；
• 更接近“创建了多少并发工作，就在这里负责收尾多少工作”的结构化并发原则。

4. gather()：简单收集一组结果

results = await asyncio.gather(
    fetch_a(),
    fetch_b(),
)

gather() 会把传入的协程自动调度为 Task，并按输入顺序返回结果。¹

它适合：

• 简单并发运行一组 awaitable；
• 需要按输入顺序收集结果；
• 明确理解并控制异常策略。

默认情况下，如果一个子任务抛出异常，异常会传播给等待 gather() 的调用者，但其他子任务不会因此自动全部取消，而会继续运行。相关任务需要“同生共死”时，通常优先使用 TaskGroup。¹

5. 后台 Task 必须被管理

task = asyncio.create_task(do_work())

不要把它理解为“创建后就可以完全不管”。官方文档提醒，事件循环只保存 Task 的弱引用；可靠的后台任务需要保存强引用，并处理完成状态和异常。¹

background_tasks: set[asyncio.Task[object]] = set()


task = asyncio.create_task(do_work(), name="background:do_work")
background_tasks.add(task)
task.add_done_callback(background_tasks.discard)

如果任务与当前业务操作相关，TaskGroup 通常比“裸后台任务”更安全。

八、Future：为什么它存在

Future 是底层异步基础设施中的结果占位对象。

1. 用 Future 桥接回调

import asyncio


async def main() -> None:
    loop = asyncio.get_running_loop()
    future: asyncio.Future[str] = loop.create_future()

    loop.call_later(1, future.set_result, "数据准备好了")

    result = await future
    print(result)


asyncio.run(main())

执行过程：

创建 Future，状态为 PENDING
→ main 等待 Future 并挂起
→ 定时回调调用 set_result(...)
→ Future 变为 FINISHED
→ main 恢复并取得结果

Future 也可以保存异常：

future.set_exception(RuntimeError("操作失败"))

之后 await future 会在等待位置重新抛出这个异常。

2. 为什么业务代码通常不直接创建 Future

在 HTTP 客户端、数据库驱动和网络框架中，Future 常用于把底层回调或 I/O 通知转换成可以 await 的接口。

上层业务通常只需要：

response = await http_client.get(url)

而不需要知道内部创建了多少 Future。

官方建议：普通用户 API 不要暴露裸 Future；确实需要创建时，应使用 loop.create_future()，让不同事件循环实现有机会提供自己的优化实现。⁵

3. 两种 Future 不要混淆

asyncio.Future

属于事件循环体系，可以直接 await：

result = await asyncio_future

concurrent.futures.Future

通常来自线程池、进程池或多解释器执行器：

future = executor.submit(blocking_function)

它不能直接作为原生 asyncio Future 使用。需要桥接时，可通过 asyncio.wrap_future() 包装；而 loop.run_in_executor() 已经会返回可以由 asyncio 等待的对象。^{5 6}

九、取消与超时：异步程序必须设计失败路径

真实系统不仅会成功，还会遇到：

客户端断开连接
上游请求超时
服务正在关闭
并发子任务失败
调用方主动取消

1. cancel() 是取消请求，不是强制终止

task.cancel()

它会请求 Task 在下一个合适的机会向内部协程抛出 asyncio.CancelledError。协程可以执行清理逻辑，也理论上可以压制取消，因此 Task.cancel() 不保证 Task 最终一定进入 cancelled 状态。完全吞掉取消通常是不正确的做法。¹

async def worker() -> None:
    resource = await acquire_resource()

    try:
        await do_work(resource)
    finally:
        await release_resource(resource)

如果显式捕获 CancelledError，通常应在清理后重新抛出：

async def worker() -> None:
    try:
        await do_work()
    except asyncio.CancelledError:
        await cleanup()
        raise

CancelledError 直接继承自 BaseException，而不是普通 Exception。TaskGroup 和 asyncio.timeout() 都依赖取消机制；随意吞掉它可能破坏结构化并发的语义。¹

2. 使用 asyncio.timeout()

async def call_model() -> str:
    try:
        async with asyncio.timeout(10):
            return await llm_request()
    except TimeoutError:
        return "request timed out"

超时上下文会取消当前 Task，并把内部的 CancelledError 转换成 TimeoutError。因此 TimeoutError 应在 async with 代码块外捕获。¹

十、async for 与异步迭代

普通 for 每次同步调用 next()。

如果“获取下一项”本身可能需要等待，例如：

• 等待下一个 LLM Token；
• 等待 WebSocket 消息；
• 等待下一页 API 数据；
• 等待数据库游标返回下一行；

就需要异步迭代。

1. AsyncIterable 与 AsyncIterator

异步迭代协议包括：

__aiter__()
__anext__()

规则是：

• __aiter__() 返回异步迭代器；
• __anext__() 返回 awaitable；
• 遍历结束时抛出 StopAsyncIteration。⁴

2. async for 的概念展开

async for item in source:
    consume(item)

概念上接近：

iterator = source.__aiter__()

while True:
    try:
        item = await iterator.__anext__()
    except StopAsyncIteration:
        break

    consume(item)

日常代码也可以使用内置 anext()：

item = await anext(iterator)

与普通 for 的核心区别是：

获取下一项的过程可以异步等待

3. 同步生成器与异步生成器

异步生成器存在的主要原因不是“让 yield 自动并发”，而是：

在产生下一项之前，可以先等待异步 I/O，并在等待期间让其他 Task 运行。

async def stream_llm_tokens():
    while True:
        token = await receive_next_token()
        if token is None:
            return

        yield token

如果异步生成器内部没有任何异步等待：

async def example():
    yield "A"
    yield "B"

它依然必须用 async for 消费，但通常没有必要写成异步生成器。

4. LLM 流式输出示例

async def stream_answer(prompt: str):
    async for event in llm_client.stream(prompt):
        if event.type == "token":
            yield event.text

调用者：

async for token in stream_answer("Explain RAG"):
    print(token, end="")

这就是：

async def + yield
= 异步生成器

5. 提前退出时显式关闭异步生成器

如果异步生成器持有连接、游标或其他资源，并且消费方可能提前 break，可以使用 contextlib.aclosing()：

import contextlib


async with contextlib.aclosing(stream_answer("hello")) as stream:
    async for token in stream:
        if should_stop(token):
            break

显式关闭能让异步生成器的 finally 清理逻辑在可预测的任务上下文中执行。Python 3.14 的 asyncio 开发文档专门列出了异步生成器的显式关闭最佳实践。^{2 7}

十一、async with 与异步资源管理

普通上下文管理器：

with resource:
    ...

依赖：

__enter__()
__exit__()

异步上下文管理器：

async with resource:
    ...

依赖：

__aenter__()
__aexit__()

后两个方法必须返回 awaitable。⁴

概念上可以理解为：

manager = create_manager()
resource = await manager.__aenter__()

try:
    await use(resource)
finally:
    await manager.__aexit__(...)

典型场景包括：

• 异步 HTTP Client；
• 数据库事务；
• WebSocket；
• 异步锁；
• 超时作用域；
• TaskGroup；
• Trace 或会话生命周期。

async with 的价值不仅是语法简洁，更重要的是把“异步获取资源”和“异步释放资源”绑定到确定的生命周期中。

十二、为什么异步程序仍然需要线程池

1. async 不能改变同步库的性质

假设某个库只有同步接口：

response = requests.get(url, timeout=10)

即使把它写进 async def：

async def handler():
    response = requests.get(url, timeout=10)
    return response.text

它仍然会阻塞事件循环线程。

FastAPI、Starlette 或 asyncio 不会因为外层使用了 async def，就自动识别并迁移函数体内的任意同步调用。

2. 使用 asyncio.to_thread() 桥接同步阻塞 I/O

import asyncio
import requests


async def fetch_sync_api(url: str) -> str:
    response = await asyncio.to_thread(
        requests.get,
        url,
        timeout=10,
    )
    response.raise_for_status()
    return response.text

执行关系是：

requests.get(...)
→ 工作线程执行

事件循环线程
→ 等待期间继续运行其他 Task

asyncio.to_thread() 返回一个协程，因此仅仅调用它并不会自动开始工作；需要 await 它，或把返回的协程调度为 Task。它还会把当前 contextvars.Context 复制到工作线程。¹

线程池并没有消除阻塞，只是把阻塞移出了事件循环线程。

3. 取消等待不等于停止线程

如果等待 to_thread() 的 Task 被取消，已经开始运行的同步函数通常不能被 Python 强行终止。等待者可以停止等待结果，但工作线程中的函数可能继续执行到自然结束。concurrent.futures.Future.cancel() 对已经运行的调用也无法成功取消。⁶

因此，线程桥接不等于：

原生可取消的异步 I/O

对于长时间运行的同步函数，应考虑：

• 给底层 I/O 设置超时；
• 让函数主动检查取消信号；
• 使用可终止的独立进程或 Worker；
• 优先替换为原生异步客户端。

十三、CPU 密集型任务应该怎样处理

asyncio.to_thread() 主要用于避免同步 I/O 阻塞事件循环，而不是默认的 CPU 加速方案。

在常见的 GIL-enabled CPython 中，纯 Python CPU 密集代码放进多个线程，通常无法获得理想的多核并行。官方文档也把 to_thread() 的主要用途定位为 I/O-bound 函数；能释放 GIL 的扩展模块则可能例外。¹

常见选择包括：

ProcessPoolExecutor
InterpreterPoolExecutor（Python 3.14+）
专门的任务队列或 Worker 服务
能释放 GIL 的本地扩展
GPU / 外部计算服务

1. 进程池示例

import asyncio
from concurrent.futures import ProcessPoolExecutor


def cpu_heavy(value: int) -> int:
    return sum(number * number for number in range(value))


async def main() -> None:
    loop = asyncio.get_running_loop()

    with ProcessPoolExecutor() as executor:
        result = await loop.run_in_executor(
            executor,
            cpu_heavy,
            10_000_000,
        )

    print(result)


if __name__ == "__main__":
    asyncio.run(main())

Python 3.14 的 InterpreterPoolExecutor 让每个工作线程运行在独立解释器中，每个解释器有自己的 GIL，因此可以获得真正的多核并行，但代价是解释器隔离、对象共享受限和序列化成本。⁶

2. 关于 free-threaded CPython

从 Python 3.13 开始，CPython 提供可选的 free-threaded 构建，可以在禁用 GIL 时让线程并行使用多个 CPU 核心。它是单独安装或构建的运行模式，并且部分第三方扩展可能尚未兼容，甚至会重新启用 GIL。⁸

因此，在通用教程和默认部署假设下，仍然应把下面这条作为主要工程判断：

不要把 asyncio.to_thread() 当成纯 Python CPU 密集任务的通用多核加速器。

十四、FastAPI / Starlette 中的线程池边界

1. def endpoint 与 async def endpoint

FastAPI 可以混合使用两种路径函数：

@app.get("/async")
async def async_endpoint():
    response = await async_client.get(...)
    return response.json()

@app.get("/sync")
def sync_endpoint():
    response = sync_client.get(...)
    return response.json()

FastAPI 官方建议：第三方库提供可 await 的异步 API 时使用 async def；只有同步阻塞接口时，可以使用普通 def 路径函数。普通 def 路径函数和同步依赖会在线程池中运行。⁹

需要特别注意：

@app.get("/wrong")
async def wrong_endpoint():
    response = sync_client.get(...)
    return response.json()

同步调用位于 async def 内部时，不会被 FastAPI 自动迁移到线程池，仍然会直接阻塞事件循环线程。

2. Starlette 使用 AnyIO 工作线程

Starlette 会在多个场景使用线程池，例如：

• def 定义的同步 endpoint；
• 同步 BackgroundTask；
• 文件响应；
• 文件上传；
• 部分内部同步操作。

本文核对的 Starlette 1.3.1 通过 anyio.to_thread.run_sync() 执行这些同步代码。默认线程容量限制为 40 个 token，并且会与 FastAPI 的同步依赖等场景共享。盲目提高容量可能增加内存占用和上下文切换成本。^{10 11}

3. 同步 Iterable 怎样桥接成 AsyncIterator

Starlette 的 StreamingResponse 可以接收异步生成器，也可以接收普通生成器或迭代器。¹²

当输入是同步 Iterable 时，框架需要在线程池中执行每一次 next()，避免同步迭代阻塞事件循环。概念上类似：

async def iterate_in_threadpool(iterable):
    iterator = iter(iterable)

    while True:
        try:
            item = await run_next_in_worker_thread(iterator)
        except EndOfIterator:
            break

        yield item

这段代码是为了说明协议桥接，不是 Starlette 源码的逐字复制。在本文核对的 Starlette 1.3.1 实现中，工作线程会把 StopIteration 转换为自定义异常，再由异步生成器一侧结束循环，因为同步迭代终止信号不能简单地原样跨越这个 await / 线程边界。¹³

十五、AI 与 Web 后端中的典型模式

1. 并发调用互不依赖的工具

import asyncio


async def collect_context(query: str) -> tuple[dict, dict]:
    async with asyncio.TaskGroup() as group:
        web_task = group.create_task(
            search_web(query),
            name="tool:web-search",
        )
        db_task = group.create_task(
            search_database(query),
            name="tool:database-search",
        )

    return web_task.result(), db_task.result()

前提是两个调用互不依赖，可以同时启动。

2. 有数据依赖的 Agent 步骤必须顺序执行

plan = await create_plan(user_message)
tool_result = await execute_tool(plan)
answer = await generate_answer(tool_result)

这里不能为了追求“并发”而把三步全部创建为 Task，因为：

execute_tool 依赖 plan
generate_answer 依赖 tool_result

并发不是目标本身。正确目标是：

只并发那些没有依赖关系、可以安全同时开始的工作。

3. 流式返回 LLM Token

async def stream_response(prompt: str):
    async for event in llm.stream(prompt):
        if event.type == "token":
            yield event.text

异步生成器使服务在等待下一批网络数据时，不必占住事件循环线程。

4. 同步文档解析器接入异步服务

async def parse_document(path: str) -> list[str]:
    return await asyncio.to_thread(sync_parser.parse, path)

如果解析主要是阻塞文件 I/O，这种桥接通常合理；如果解析包含大量纯 Python CPU 计算，应考虑进程池、InterpreterPoolExecutor 或独立 Worker。

5. 限制并发量

不要无上限地为成千上万个外部请求创建 Task。

import asyncio


limit = asyncio.Semaphore(10)


async def limited_call(item: str) -> str:
    async with limit:
        async with asyncio.timeout(20):
            return await call_external_api(item)

并发限制可以降低：

• LLM 或第三方 API 限流风险；
• 数据库连接耗尽风险；
• 本机文件描述符与内存压力；
• 下游服务被瞬时流量压垮的概率。

在生产系统中，通常还需要配合队列容量、重试策略、指数退避和整体截止时间。

十六、调试异步程序

1. 开启 asyncio Debug Mode

asyncio.run(main(), debug=True)

也可以设置：

PYTHONASYNCIODEBUG=1

Debug Mode 可以帮助发现：

• 从错误线程调用非线程安全 API；
• 过慢的回调；
• 未等待的协程；
• 配合警告配置发现未关闭的传输、事件循环等资源。²

2. 检查运行时对象类型

import inspect

inspect.iscoroutinefunction(func)
inspect.iscoroutine(obj)
inspect.isawaitable(obj)
inspect.isasyncgenfunction(func)
inspect.isasyncgen(obj)

这些方法适合学习、框架开发和排查第三方 SDK 返回类型。普通业务代码应优先依赖类型标注和官方文档，而不是到处动态探测。

3. 给 Task 命名

task = asyncio.create_task(
    call_external_api(),
    name="llm:generate-answer",
)

有意义的 Task 名称会出现在 repr() 和部分调试信息中，能明显降低排查并发问题的成本。

4. 两个常见警告

coroutine was never awaited

原因：

fetch_data()

修复：

result = await fetch_data()

或：

task = asyncio.create_task(fetch_data())
result = await task

Task exception was never retrieved

常见原因是创建了后台 Task，却没有等待它，也没有读取异常。

优先使用 TaskGroup；确实需要后台任务时，应保存引用并建立明确的异常处理策略。

十七、最常见的错误认知

误区一：async def 会自动把代码变成非阻塞

错误。同步网络请求、time.sleep() 和纯 Python 长循环仍然会阻塞事件循环。

误区二：写了 await 就一定不会阻塞

错误。await sync_function() 会先同步执行 sync_function()，再尝试等待其返回值。

误区三：每个 await 都必然切换 Task

不准确。awaitable 已经完成时，当前 Task 可能直接继续。

误区四：连续 await 会自动并发

错误。

await a()
await b()

B 要等 A 完成后才会被调用。

误区五：事件循环会跳到当前协程的下一行

错误。当前 Task 在 await 处挂起后，它自己的后续代码也处于挂起状态。

误区六：Task 就是线程

错误。Task 是事件循环中的协程调度对象，通常仍在同一个事件循环线程中执行。

误区七：Future 与 Task 完全相同

错误。Future 主要表示未来结果；Task 还负责运行一个协程。

误区八：取消 Task 等于强制停止所有底层工作

错误。取消是协作式请求；线程中已经开始的同步函数通常仍会继续运行。

误区九：线程池能自动加速所有 CPU 计算

错误。在常见 GIL-enabled CPython 中，纯 Python CPU 密集任务通常更适合进程池、多解释器或专用计算服务。

误区十：并发越大，吞吐量一定越高

错误。过量并发可能造成限流、连接耗尽、内存增长、调度开销和下游雪崩。

十八、完整速查表

十九、工程实践清单

1. 调用协程函数只创建协程对象，不会自动调度执行。
2. 不要创建一个协程对象后既不 await 也不调度。
3. await 后面的表达式必须最终产生 awaitable。
4. 连续直接 await 表示顺序控制流。
5. 相关并发任务优先使用 TaskGroup。
6. 有数据依赖的步骤必须保持顺序。
7. 不要在事件循环线程中执行同步阻塞 I/O。
8. 同步阻塞 I/O 可以用 to_thread() 临时桥接。
9. 不要把取消等待误认为停止工作线程。
10. 默认环境下，CPU 密集任务优先考虑进程池、多解释器或专用 Worker。
11. Task 不是线程，asyncio 并发不等于 CPU 并行。
12. 使用 Semaphore、队列容量和连接池限制外部并发。
13. 通过 try/finally 释放异步资源。
14. 不要随意吞掉 CancelledError。
15. 提前退出异步生成器时考虑显式关闭。
16. 为外部 I/O 设置超时，并设计取消与重试策略。
17. 开发阶段开启 asyncio Debug Mode。
18. 对后台 Task 保存引用并处理异常。

结语

Python 异步体系虽然术语很多，但可以沿着一条主线理解：

Iterable / Iterator
解决“怎样逐项取得同步数据”

Generator
解决“怎样暂停函数并逐项产出同步数据”

Coroutine
描述“一段可以暂停和恢复的异步执行”

Awaitable
定义“什么对象可以被 await”

Task
把 Coroutine 变成事件循环可以独立调度的工作

Future
表示“未来会完成的结果”

Event Loop
调度已经存在的 Task、回调、定时器与 I/O 事件

async for
异步地逐项取得数据

async with
异步地获取和释放资源

Thread / Process / Interpreter Pool
把不适合直接运行在事件循环线程中的工作隔离出去

最后回到最容易混淆的代码：

await fetch_a()
await fetch_b()

它为什么顺序执行？

因为第一行控制着当前 Task 的执行位置。第一行没有完成时，第二行尚未执行，fetch_b() 还没有被调用，也没有独立的 Task 可供事件循环调度。

而：

task_a = asyncio.create_task(fetch_a())
task_b = asyncio.create_task(fetch_b())

已经显式创建两个独立 Task。事件循环才有机会在 A 与 B 各自等待 I/O 时交替推进它们。

理解这条边界后，Coroutine、Task、Future、事件循环与并发之间的关系就真正串起来了。

参考资料