《AI Agent 开发入门》14.2 超时、取消与结构化并发

14.2 超时、取消与结构化并发

14.1 让 Agent 学会了「失败就重试」，但重试本身会耗时——一个 5 秒超时的工具重试三次就是 15 秒。如果这一轮里还有另外九个工具在等它，整轮就被这一个慢调用拖住了。可靠性不只是「失败了怎么办」，更是「拖太久怎么办」。

本节把 minagent 推进到 minagent/resilience/concurrency.py：用 asyncio.timeout 给单次调用设墙钟上限，用 asyncio.TaskGroup 做结构化并发，让超时与取消能可靠地传播、不留孤儿任务。

14.2.1 超时：把「等下去」变成「止损」

一个没有超时的 Agent 会陷入最难排查的故障：它没报错，它就是不动。请求发出去了，await 挂在那里，既不返回也不抛异常，CPU 和内存都正常——从监控上看一切健康，从用户视角看它已经死了。这种「静默挂起」比崩溃危险得多，因为它不会触发任何告警。

超时的本质是给每个等待加一个墙钟上限：超过这个时间，不管对方在干什么，我们单方面认定它不会好了，主动放弃。这里有三个必须想清楚的点：

第三点尤其重要：超时设太短会误杀正常请求，设太长等于没设。工程上的经验值是该接口 p99 延迟的 2–3 倍——留给长尾足够的空间，又能在真正卡死时及时止损。模型生成尤其难估：一次带长推理的响应可能几秒也可能几十秒，所以给模型的超时通常要比给普通工具的宽松，或者干脆拆成「首字节超时」与「总时长超时」两个阈值。

14.2.2 asyncio.timeout：一个上下文管理器搞定超时

Python 3.11 起，标准库给了 asyncio.timeout——它比早年的 wait_for 更清晰，因为它是一个上下文管理器，作用域内的所有 await 共享同一个截止时间：

import asyncio

async def run_with_timeout(coro, seconds: float):
    async with asyncio.timeout(seconds):
        return await coro

三行，但语义完整：进入 async with 时启动一个定时器，作用域内任何一个 await 阻塞超过 seconds，asyncio.timeout 就会向当前任务注入一个 CancelledError；如果这个取消一路传播到 async with 边界都没被「消化」，它在退出时被转换成 TimeoutError 抛出。

在 3.11 之前，大家用 asyncio.wait_for(coro, timeout)。它能用，但语义不如 asyncio.timeout 清晰：wait_for 是「函数」，超时时直接抛 TimeoutError，作用域内的取消与超时的区分不够明确，嵌套组合也差。asyncio.timeout 是「上下文管理器」，可以自然嵌套、可以把一段超时预算分摊给作用域内的多个 await，这正是 Agent 一轮多工具场景需要的。新代码一律用 asyncio.timeout。

这个「取消转异常」的设计很讲究：作用域内部看到的是取消（这样才能清理资源），作用域外部看到的是超时（这样才能区别于用户主动取消）。所以调用方写 except TimeoutError 就能捕获超时，而作用域内部的 finally 会正常执行。用一个快工具和一个慢工具验证：

正常返回: fast ok
超时: slow 未在 0.2s 内完成，已取消

fast 在 0.5s 预算内正常返回；slow 要 5s，超过 0.2s 预算，TimeoutError 被抛出，调用方拿到的是一个明确的「超时」信号，而不是永远悬着的 await。

14.2.3 结构化并发：TaskGroup 保证「不落单」

光有超时还不够。Agent 一轮里往往并发跑多个工具（第 4.3 节讲过），如果只是「每个工具各自超时」，会出现一个尴尬的中间态：慢的那个被超时取消了，快的几个还在后台跑——它们是孤儿，没人等它们，出错了也没人知道。

asyncio.TaskGroup 解决的就是这个。它把一组任务收进一个作用域，保证要么全部成功、要么一起收场：

async with asyncio.TaskGroup() as tg:
    tg.create_task(slow_tool("a", 0.1, log))
    tg.create_task(slow_tool("b", 5.0, log))
    tg.create_task(slow_tool("c", 0.2, log))

TaskGroup 的规则很简单：作用域退出前，它保证组内所有任务都已结束。如果其中一个任务抛了异常（或整个作用域被外部取消，比如被 asyncio.timeout 取消），TaskGroup 会取消其余所有还在跑的任务，然后等它们全部退出作用域才继续。这带来两个硬保证：不会有孤儿任务，每个任务的 finally 都会被执行。

对比一下三兄弟的分工，就明白为什么卷一选 TaskGroup：

gather 也能并发，但它不保证「一个失败就取消其余」——默认行为是等其他任务跑完，孤儿风险就落到了调用方头上。TaskGroup 把「同生共死」写进了语言机制，是「结构化并发」这个理念在 Python 里的落地。

TaskGroup 的「失败即取消」还有一个额外价值：如果某个工具因为不可重试的错误（比如权限 403）而抛出异常，TaskGroup 会立刻取消其余工具——不用再等那些注定没用的调用跑完。这和 14.1 的「快速失败」理念一致：确定的错误要尽早止损，而不是让整轮陪跑。

14.2.4 真跑：慢工具注入，验证止损与取消传播

把超时和 TaskGroup 叠起来，就是 Agent 一轮并发的正确写法——超时管「等多久」，TaskGroup 管「一起收场」：

async def main() -> None:
    log = {"started": 0, "ok": 0, "cancelled": 0, "finally": 0}
    try:
        async with asyncio.timeout(0.3):
            async with asyncio.TaskGroup() as tg:
                tg.create_task(slow_tool("a", 0.1, log))
                tg.create_task(slow_tool("b", 5.0, log))   # 远超预算
                tg.create_task(slow_tool("c", 0.2, log))
    except TimeoutError:
        print("超时触发：整组取消")
    print("统计:", log)
    leaked = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    print("未完成的残留任务数:", len(leaked))

slow_tool 用 asyncio.sleep 模拟一个耗时工具，并在 finally 里记账，这样我们能数清楚「有几个任务真正跑完、几个被取消、几个执行了清理」：

async def slow_tool(name: str, delay: float, log: dict) -> str:
    log["started"] += 1
    try:
        await asyncio.sleep(delay)
        log["ok"] += 1
        return f"{name} ok"
    except asyncio.CancelledError:
        log["cancelled"] += 1
        raise
    finally:
        log["finally"] += 1

三个工具延迟分别是 0.1s、5.0s、0.2s，超时预算 0.3s。真跑（Python 3.14.6）：

  a: 启动 (delay=0.1s)
  b: 启动 (delay=5.0s)
  c: 启动 (delay=0.2s)
  b: 被取消
超时触发：整组取消
统计: {'started': 3, 'ok': 2, 'cancelled': 1, 'finally': 3}
未完成的残留任务数: 0

逐行读这份证据：三个任务全部启动（started=3），说明并发是真的；a 和 c 在 0.3s 预算内跑完（ok=2），b 因为要 5s 被超时取消（cancelled=1）；关键是 finally=3——三个任务无论成功还是被取消，清理逻辑都执行了，没有任何一个被「跳过清理」地硬杀；最后 未完成的残留任务数: 0，证明 TaskGroup 退出时确实收干净了。

这就是「超时能止损」的完整证据链：超时触发 → 慢任务被取消 → 所有任务 finally 都执行 → 无残留。四步缺一不可。

这里还有个反直觉的点：finally 之所以能跑，是因为 Python 的取消是协作式的——CancelledError 在下一个 await 点被抛出，而不是像杀进程一样立刻终止。这意味着如果任务卡在一个不响应取消的同步阻塞调用里（比如 time.sleep 或一个卡死的 socket 读），取消根本传不进去。所以异步工具应尽量用 await asyncio.sleep / 异步 I/O；同步阻塞的活儿要用 asyncio.to_thread 丢到线程池，让事件循环保持可取消。这是「结构化并发」能生效的前提。

14.2.5 反例：裸 create_task 会留下孤儿

为了看清 TaskGroup 到底帮我们挡住了什么，把上面的写法换成「裸 create_task」，其余不变：

async def leaky():
    log = {"started": 0, "ok": 0, "cancelled": 0, "finally": 0}
    orphan = asyncio.create_task(slow_tool("orphan", 5.0, log))  # 裸 task
    try:
        async with asyncio.timeout(0.2):
            await slow_tool("main", 1.0, log)
    except TimeoutError:
        print("超时触发（裸 task 版）")
    print("孤儿任务仍在运行:", not orphan.done())
    orphan.cancel()            # 必须手动收拾，否则泄漏
    await asyncio.sleep(0)

真跑输出：

超时触发（裸 task 版）
统计: {'started': 2, 'ok': 0, 'cancelled': 1, 'finally': 1}
孤儿任务仍在运行: True
手动取消后 finally 总数: 2

asyncio.timeout 只取消了它作用域内正在 await 的那个任务（main），orphan 是在作用域外面用 create_task 起的，超时对它毫无影响——孤儿任务仍在运行: True。此刻 finally 只记了 1 次，说明 orphan 的清理还没跑；必须手动 orphan.cancel() 再让出一次事件循环，它才收尾（finally 变成 2）。

这个对比点出了 TaskGroup 的核心价值：它把「谁负责取消、谁负责等待、谁负责清理」变成结构上不可能搞错的事——任务的生命周期被绑在 async with 的作用域上，作用域结束必然全部结束。裸 create_task 把这份责任还给了程序员，而人会忘。

经验法则是：只要你在 async with 里 create_task，就用 TaskGroup；只有当你明确需要「任务活得比作用域长」时，才用裸 create_task——而 Agent 场景几乎从不需要后者。把默认选择设成 TaskGroup，能消灭一整类难查的「后台任务偷偷报错」问题。

14.2.6 接进 minagent 的循环

把这两块接进 Agent 主循环，改动集中在「一轮工具调用」这一处：给整轮加一个 asyncio.timeout，用 TaskGroup 起所有工具任务，把每个工具的结果（或超时错误）按 tool_call_id 收集回消息。工具是同步的，用 asyncio.to_thread 丢进线程池再 await：

async def run_tools(calls, budget, exec_tool):
    results = {}
    async def one(tc):
        results[tc.id] = await asyncio.to_thread(exec_tool, tc)
    try:
        async with asyncio.timeout(budget):
            async with asyncio.TaskGroup() as tg:
                for tc in calls:
                    tg.create_task(one(tc))
    except TimeoutError:
        for tc in calls:
            results.setdefault(tc.id, '{"error": "timeout"}')
    return results

注意 setdefault：预算耗尽时，只给那些还没来得及产出结果的调用补超时错误，已经拿到结果的保留原样。跑三个工具，延迟 0.1s、0.2s、5.0s，预算 0.5s：

t1: {"tool": "t1", "ok": true}
t2: {"tool": "t2", "ok": true}
t3: {"error": "timeout"}

t1、t2 正常返回，t3 超时被降级成 {"error": "timeout"}。一个工具超时不该让整轮崩溃——超时应该被降级成一条工具结果回灌给模型，让模型像处理普通工具错误一样决定要不要换策略。这正是 14.1 的「工具错误回灌」和本节的「超时」汇合的地方：超时是工具错误的一种。

注意 run_tools 收集的是每个工具的结果，而不是「整轮成功 / 失败」——即使有工具超时，其他工具的结果依然保留。这样模型下一轮能看到「t1、t2 成功，t3 超时」，从而决定是重试 t3 还是换一个工具。把部分失败表达成「部分成功」，而不是「整轮失败」，是 Agent 容错的关键设计。

14.2.7 取消会一路传到最内层

还有一个细节值得强调：取消不是「停下当前这一行」，而是一个会一路传播的信号。当 asyncio.timeout 触发，CancelledError 从最外层的 await 往里传，穿过每一层 async with、每一个 await，直到最内层的 asyncio.sleep——那里才真正把协程唤醒、让它有机会跑 finally。这条路径决定了清理能否发生：

这里最容易犯的错是在中间层 except Exception 把取消「吃掉」。因为 CancelledError 在 3.8 之后继承自 BaseException 而非 Exception，裸的 except Exception 不会误捕它——但如果你写了 except BaseException 或 except:，取消就会被静默吞掉，超时机制随之失效。清理可以，吞掉不行：finally 里该做的做完，然后让 CancelledError 继续往上抛。

14.2.8 超时之后：降级还是重试

超时不是终点，而是一个需要决策的岔路口。同一个超时，处理方式取决于这个工具是只读还是写入（正好接上下一节）：

预算怎么定？它应该是这一轮所有工具超时的上界。如果单工具超时是 5s、并发 10 个，理想情况整轮不超过 5s；但保险起见给一轮设 8–10s 的硬上限，防止某个工具「内部重试」把时间拖长。单工具超时管「一个工具别太久」，整轮预算管「一轮别太久」，两级都要有。

这张表把本节的「超时」和下一节的「幂等」连了起来：超时本身不可怕，可怕的是「超时后不知道该不该重试」。判据还是那条——幂等就能重试，不幂等就不能。

至此 14.2 给 minagent 装上了第二层韧性：超时止损 + 结构化取消。它保证 Agent 不会被单个慢调用拖死，也不会在超时后留下跑在后台的孤儿。但取消有个副作用值得警惕——一个工具可能「执行到一半」就被取消了，而它已经产生的副作用不会自己撤销。这正是下一节「幂等」要处理的问题。

真实模型下会怎样：本节的慢工具是 asyncio.sleep 假装的，取消它不花任何代价。真实工具（一次 HTTP 写请求、一次数据库写入）在被取消时，底层连接可能已经发出、服务器可能已经执行——取消只保证「本地不再等」，不保证「远端没发生」。这是分布式系统的固有难题：客户端超时永远无法确定服务端到底做没做。所以真实系统里，超时要配合下一节的幂等键一起用：超时后重试是安全的，前提是重试是幂等的。另外，真实模型 API 的长耗时主要是「生成 token」，asyncio.timeout 要包在整个流式读取外面，而不是只包住建立连接那一步，否则一个开始很快、中途卡住的响应会绕过超时。

阅读导航：上一节：14.1 分类错误与重试 · 下一节：14.3 幂等与副作用 。