第3章异步编程¶

学习目标¶

理解 asyncio 事件循环与协程的执行原理
熟练使用 async/await 编写异步函数
掌握 gather、wait、TaskGroup 三种并发模式的差异与选型
识别并规避阻塞调用、异常吞噬等常见异步陷阱
运用信号量、超时、队列优化异步程序性能

问题导入¶

AI Agent 的典型工作负载是"I/O 密集"而非"CPU 密集"：一次对话可能要并发调用多个外部工具（搜索、数据库、知识库）、流式拉取 LLM 输出、读写缓存。如果用同步代码串行处理这些 I/O，CPU 大量时间花在"等待网络响应"上，单机吞吐被严重浪费。线程池虽能缓解，但线程切换开销与 GIL 限制使其在高并发下力不从心。

Python 的 asyncio 提供了基于单线程事件循环的协程模型：当某个协程在等待 I/O 时，它主动 await 让出控制权，事件循环转去执行其他就绪协程，I/O 完成后再回来继续。这种"协作式多任务"以极低的内存开销（每个协程仅 KB 级）支撑上万并发连接，是构建高并发 AI Agent 服务的核心技术。本章将系统讲解 asyncio 的原理、并发模式、性能优化与陷阱规避。

核心概念¶

事件循环与协程¶

事件循环（Event Loop）是 asyncio 的核心，它维护一个就绪任务队列与一个定时器堆，不断轮询：执行就绪回调、检查定时器、等待 I/O 就绪。协程（coroutine）是用 async def 定义的函数，调用后返回一个协程对象，需被事件循环调度才能真正执行。await 暂停当前协程并交还控制权给事件循环，等待一个"可等待对象"（awaitable：协程、Task、Future）完成。

async/await 语义¶

async def 声明协程函数；await expr 只能在协程内使用，表示"等待 expr 完成，期间让出 CPU"；async with 用于异步上下文管理器（如异步客户端）；async for 用于异步迭代（如流式响应）。注意 await 不会阻塞整个线程，只"暂停"当前协程。

并发模式¶

asyncio.gather(*aws)：并发运行多个可等待对象，按输入顺序返回结果列表；默认任一异常会向上抛出，但其它任务仍会继续运行。适合"全部完成、收集结果"。
asyncio.wait(aws)：返回 (done, pending) 两个集合，可通过 return_when=FIRST_COMPLETED 实现"先到先处理"。适合"任意一个完成即可"。
asyncio.TaskGroup（3.11+）：用 async with 创建任务组，组内任一任务异常会取消其余任务并抛出 ExceptionGroup，是官方推荐的现代并发原语，更安全、更易排查。

性能优化¶

信号量（Semaphore）：限制并发数，避免打垮下游服务或触发限流。
超时（timeout / wait_for）：防止某个慢请求拖垮整体；3.11+ 推荐用 async with asyncio.timeout(t)。
连接复用：复用 httpx.AsyncClient / 数据库连接池，避免每次请求都建立新连接。
批处理：将多个独立请求合并为一次批量调用，减少往返。

常见陷阱¶

阻塞调用：在协程中调用 time.sleep、requests.get、同步文件 I/O 会阻塞整个事件循环，所有协程"卡死"。必须用 asyncio.sleep、httpx.AsyncClient、aiofiles 等异步版本，或用 asyncio.to_thread 把阻塞调用丢到线程池。
异常吞噬：gather 默认 return_exceptions=False，但若忘记 await 某个 Task，其异常可能被静默丢弃；TaskGroup 能更好暴露异常。
创建未调度的 Task：asyncio.create_task(coro) 立即调度，但若不持有引用，可能被垃圾回收取消；应保存引用。
协程未 await：直接调用 async def 函数而不 await，只产生一个未运行的协程对象，Python 会发出 RuntimeWarning。

原理图¶

sequenceDiagram
    participant Main as 主协程
    participant Loop as 事件循环
    participant T1 as Task A (HTTP)
    participant T2 as Task B (DB)
    participant T3 as Task C (LLM)
    Main->>Loop: asyncio.gather(A, B, C)
    Loop->>T1: 调度 A 开始
    Loop->>T2: 调度 B 开始
    Loop->>T3: 调度 C 开始
    T1-->>Loop: await (等待网络) 让出
    T2-->>Loop: await (等待查询) 让出
    T3-->>Loop: await (等待 LLM) 让出
    Note over Loop: 空闲时 select/poll 监听 I/O
    T2-->>Loop: I/O 就绪，恢复 B
    Loop-->>T2: 继续执行 B
    T2-->>Main: 返回结果
    T1-->>Loop: I/O 就绪，恢复 A
    T1-->>Main: 返回结果
    T3-->>Loop: 流式 token 就绪
    T3-->>Main: 返回最终结果
    Main->>Main: gather 返回 [A, B, C]

图：事件循环在多个协程间切换，I/O 等待期间不占用 CPU，实现高并发

示例代码¶

基本异步函数与并发¶

import asyncio
import time

async def fetch(name: str, delay: float) -> str:
    print(f"[{name}] 开始，预计 {delay}s")
    await asyncio.sleep(delay)  # 模拟 I/O 等待
    print(f"[{name}] 完成")
    return f"{name}:done"

async def main() -> None:
    start = time.perf_counter()
    # 串行：总耗时约 1+2+3=6s
    # 并发：总耗时约 max(1,2,3)=3s
    results = await asyncio.gather(
        fetch("A", 1.0),
        fetch("B", 2.0),
        fetch("C", 3.0),
    )
    print(f"结果: {results}")
    print(f"总耗时: {time.perf_counter() - start:.2f}s")

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

运行结果：

[A] 开始，预计 1.0s
[B] 开始，预计 2.0s
[C] 开始，预计 3.0s
[A] 完成
[B] 完成
[C] 完成
结果: ['A:done', 'B:done', 'C:done']
总耗时: 3.00s

TaskGroup 与异常组（3.11+）¶

import asyncio

async def task_ok(n: int) -> int:
    await asyncio.sleep(0.1)
    return n * 2

async def task_fail() -> None:
    await asyncio.sleep(0.2)
    raise RuntimeError("故意失败")

async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            t1 = tg.create_task(task_ok(5))
            t2 = tg.create_task(task_ok(10))
            t3 = tg.create_task(task_fail())
        # 若全部成功才会执行到这里
        print(t1.result(), t2.result())
    except* RuntimeError as eg:
        # except* 处理异常组
        print(f"捕获到异常: {eg.exceptions}")

if __name__ == "__main__":
    asyncio.run(main())
# 输出: 捕获到异常: (RuntimeError('故意失败'),)

信号量限流 + 超时¶

import asyncio
from typing import Sequence

async def call_llm(prompt: str, sem: asyncio.Semaphore) -> str:
    async with sem:  # 限制并发数
        async with asyncio.timeout(2.0):  # 单次超时
            await asyncio.sleep(0.5)  # 模拟 LLM 调用
            return f"resp<{prompt}>"

async def batch_call(prompts: Sequence[str], max_concurrency: int = 3) -> list[str]:
    sem = asyncio.Semaphore(max_concurrency)
    return await asyncio.gather(*(call_llm(p, sem) for p in prompts))

async def main() -> None:
    prompts = [f"q{i}" for i in range(10)]
    results = await batch_call(prompts, max_concurrency=3)
    print(results)

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

异步 HTTP 请求（httpx）¶

import asyncio
import httpx

async def fetch_all(urls: list[str]) -> list[str]:
    async with httpx.AsyncClient(timeout=10.0) as client:
        tasks = [client.get(url) for url in urls]
        responses = await asyncio.gather(*tasks)
        return [r.text[:50] for r in responses]

async def main() -> None:
    urls = ["https://httpbin.org/delay/1"] * 3
    texts = await fetch_all(urls)
    for i, t in enumerate(texts):
        print(f"{i}: {t[:30]}...")

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

阻塞调用的正确处理¶

import asyncio
import time

def blocking_heavy(n: int) -> int:
    # 同步 CPU 密集或阻塞库，会卡住事件循环
    time.sleep(1)
    return n * n

async def main() -> None:
    # 错误：直接调用会阻塞事件循环
    # result = blocking_heavy(10)

    # 正确：丢到线程池执行
    result = await asyncio.to_thread(blocking_heavy, 10)
    print(f"结果: {result}")

asyncio.run(main())

企业案例¶

案例 1：高并发 LLM 网关¶

某 AI 平台用 FastAPI 对外提供 LLM 代理服务，单实例需支撑上千 QPS。同步 requests 版本单机仅 50 QPS（线程开销大）；改为 httpx 异步 + asyncio 后单机达 800 QPS，平均延迟下降 60%。关键优化：复用全局 AsyncClient 连接池、用 Semaphore 限制对上游 OpenAI 的并发避免触发限流、用 asyncio.timeout 防止慢请求拖垮队列。

案例 2：Agent 多工具并发编排¶

一个 Agent 在回答用户问题时需要并发调用 5 个工具：搜索引擎、向量库、SQL、天气 API、计算器。串行调用总耗时约 4.5s；用 asyncio.gather 并发后降至 1.2s（最慢工具耗时）。若某工具失败，Agent 仍可用其余结果生成回答，通过 return_exceptions=True 收集异常而非整体失败。

案例 3：流式响应的背压控制¶

一个 SSE 服务向客户端推送 LLM token，若客户端慢于生产端，缓冲区会无限增长导致 OOM。团队用 asyncio.Queue(maxsize=100) 实现背压：队列满时生产协程 await queue.put() 自动阻塞，实现生产消费速率匹配，内存占用稳定。

常见问题¶

Q1：asyncio 是多线程吗？为什么能高并发？ 不是。asyncio 默认单线程单事件循环，通过协程在 I/O 等待时主动让出、事件循环调度其他就绪协程实现"并发"。它不增加线程切换开销，仅靠协作式调度，故内存占用低、能支撑高并发连接。

Q2：什么时候不该用 asyncio？ CPU 密集型任务（如大量数值计算、模型推理）不适合，因为单线程 GIL 无法利用多核；应改用 multiprocessing 或 ProcessPoolExecutor，或用原生扩展（Cython、numba）。混合场景可用 asyncio + run_in_executor 调度计算到进程池。

Q3：gather 和 TaskGroup 怎么选？ 新代码优先 TaskGroup（3.11+）：它保证任一任务异常会取消其他任务，避免资源泄漏与异常吞噬，错误信息以 ExceptionGroup 形式清晰呈现。gather 在需要 return_exceptions=True 收集所有结果（含异常）时仍方便。

Q4：协程里能调用普通同步函数吗？ 能调用不阻塞的纯计算函数；但调用阻塞 I/O（time.sleep、requests、同步文件读写）会卡住整个事件循环。需用异步等价物或 asyncio.to_thread 包裹。可在 CI 中用 asyncio.get_event_loop().set_debug(True) 检测阻塞调用（会打印耗时超过阈值的警告）。

面试题¶

题1：解释 asyncio.run() 做了什么，为什么不能重复调用？ 参考答案：asyncio.run(coro) 创建一个新事件循环，运行主协程至完成，最后关闭循环并清理。它设计为程序的入口，重复调用会因已有运行中的循环而报错；且每次创建/销毁循环有开销。应在单次 run 内组织整个异步程序结构。

题2：await asyncio.sleep(0) 有什么用？ 参考答案：它让出当前协程的执行权给事件循环，允许其他就绪协程运行，随后立即被重新调度。常用于协程合作式让步，避免长循环饿死其他任务，或在测试中手动推进事件循环。

题3：如何让一个异步函数同时支持超时与重试？ 参考答案：组合 asyncio.timeout(t)（或 wait_for）与重试循环。每次尝试包在超时上下文中，捕获 TimeoutError 后退避重试，达到上限抛出。注意超时取消的任务需确保资源被清理（用 async with 管理客户端）。

题4：asyncio.Queue 和 queue.Queue 的区别？ 参考答案：asyncio.Queue 用于协程间通信，put/get 是协程方法，队列空/满时挂起当前协程而非阻塞线程；queue.Queue 用于线程间通信，操作会阻塞线程。在 asyncio 程序中应使用前者，否则会阻塞事件循环。

练习题¶

上机：编写异步函数 fetch_user_and_posts(user_id)，并发拉取用户信息与该用户文章列表（用 httpx 访问 https://jsonplaceholder.typicode.com），返回组合结果。测量并发与串行的耗时差异。
思考：若 asyncio.gather 中 3 个任务有 1 个抛异常，另外 2 个会怎样？如何让全部任务都跑完并收集各自结果或异常？请写出代码并说明 return_exceptions 的行为。
上机：用 asyncio.Semaphore(2) 与 asyncio.timeout(1.0) 实现一个并发受限且带超时的批量 URL 抓取函数，对 8 个 URL（每个模拟 0.5s 延迟）总耗时应在 2s 左右。

小结¶

asyncio 通过事件循环与协程，以单线程低开销实现高并发，是 AI Agent 这类 I/O 密集场景的利器。掌握 async/await 语义、gather/TaskGroup 并发模式、信号量与超时优化，并警惕阻塞调用与异常吞噬，才能写出真正高性能且可靠的异步程序。后续章节的 FastAPI、httpx 都建立在这些概念之上。延伸阅读：asyncio 文档、PEP 654 异常组。

第3章 异步编程¶