《AI Agent 工程实战》8.3 取消、超时与泄漏：慢工具注入验证止损

8.3 取消、超时与泄漏：慢工具注入验证止损

8.2 给并发装了闸门，但闸门只能管「同时几个」，管不了「一个卡多久」。一个检索工具如果因为下游网络抖动挂在那里 30 秒，它占着的那一个并发槽就被锁死了 30 秒——并发上限是 10 的话，十次这样的抖动就能让整个服务停摆。并发越大，越要保证单个慢调用不会拖住整轮。

本节把 ResearchCopilot 推进到「可止损」：用 asyncio.timeout 给每个工具调用设墙钟上限，演示用户断开时取消如何一路传播到最内层，剖析两类任务泄漏，真跑注入一个睡 10 秒的慢工具，验证 1 秒超时后整个 run 及时止损并返回明确错误。

8.3.1 三个必须防的故障

并发的 run 里，有三类故障最难查，因为它们在监控上往往「一切正常」：

这三类故障的共同点是不会触发告警：CPU 正常、错误率为零、请求还在「处理中」。用户看到的是「转圈不动」，运维看到的是「指标健康」。本节的三个演示，就是要把它们从「看不见」变成「有证据」。

8.3.2 给每个工具调用加超时

超时的原语，卷一 14.2 已经定下：用 asyncio.timeout 这个上下文管理器，而不是老的 asyncio.wait_for。作用域内任何 await 超过预算，就注入一个 CancelledError，一路传到边界若没被消化，就转成 TimeoutError 抛出。给单个工具调用套上它：

async def run_tool(name, delay, budget, log):
    try:
        async with asyncio.timeout(budget):      # 这个工具的墙钟上限
            return await slow_tool(name, delay, log)
    except TimeoutError:
        log.append(f"{name}: 超时({budget}s) → 降级")
        return {"error": "timeout"}              # 降级成工具错误回灌

关键设计是超时降级成 {"error": "timeout"}，而不是让异常炸穿整轮。这样模型下一轮能看到「这个工具超时了」，像处理普通工具错误一样决定是重试还是换策略。这正是卷一的约定：工具执行出错包成 JSON 回灌，不抛异常——超时是工具错误的一种。

slow_tool 在 finally 里记账，好让我们看清「启动 / 完成 / 被取消」各几次：

async def slow_tool(name, delay, log):
    log.append(f"{name}: 启动")
    try:
        await asyncio.sleep(delay)
        log.append(f"{name}: 完成")
        return f"{name} ok"
    except asyncio.CancelledError:
        log.append(f"{name}: 被取消")
        raise                                     # 清理可以，吞掉不行
    finally:
        pass                                      # 真实实现：在此释放连接 / 回滚事务

8.3.3 真跑：注入 10 秒慢工具，1 秒超时

把预算设成 1 秒，慢工具设成 10 秒——模拟一个下游卡死的检索：

import asyncio, time

async def main():
    log = []
    t0 = time.perf_counter()
    r = await run_tool("slow_search", 10.0, 1.0, log)
    dt = time.perf_counter() - t0
    print(f"[1] 耗时 {dt:.2f}s 结果 {r}")
    print(f"    轨迹 {log}")

真跑（Python 3.14.6，time.perf_counter 计时）：

[1] 耗时 1.00s 结果 {'error': 'timeout'}
    轨迹 ['slow_search: 启动', 'slow_search: 被取消', 'slow_search: 超时(1.0s) → 降级']

逐项读这份证据链：

• 耗时 1.00s：慢工具要睡 10 秒，但整个调用在 1.00 秒就返回了——超时确实触发了，没有傻等 10 秒。
• 结果 {'error': 'timeout'}：调用方拿到的是一个明确的错误值，而不是一个悬着的 await 或一个炸穿的异常。run 可以继续。
• 轨迹三步：启动 → 被取消 → 超时降级。顺序说明：工具先启动（并发是真的），超时后它被取消（CancelledError 传到了最内层），边界处转成 TimeoutError 被降级。清理在取消之后、降级之前执行，没有跳过。

这就是「止损」的完整证据：超时触发 → 慢工具被取消 → 返回明确错误 → 整轮不被拖住。四步缺一不可。

8.3.4 取消传播：用户断开时会发生什么

超时是「时间到了自动取消」，还有一种是「用户主动取消」。在 ResearchCopilot 里，这发生在用户关掉浏览器、断开 SSE 连接时——服务端收到断开信号，应该把对应的 run 也取消掉，而不是让它继续烧算力和 token。

取消在 asyncio 里是协作式传播的：CancelledError 从一个 await 点传到下一个，穿过每一层 async with，直到最内层的 asyncio.sleep 才真正唤醒协程、让它有机会跑清理。用两个都睡 5 秒的工具验证：

async def run(log):
    async with asyncio.TaskGroup() as tg:
        tg.create_task(slow_tool("tool_a", 5.0, log))
        tg.create_task(slow_tool("tool_b", 5.0, log))

async def main():
    log = []
    task = asyncio.create_task(run(log))
    await asyncio.sleep(0.2)          # 让工具跑起来
    task.cancel()                      # 模拟用户断开
    try:
        await task
    except asyncio.CancelledError:
        print("[2] run 被取消（用户断开）")
    print(f"    轨迹 {log}")

真跑输出：

[2] run 被取消（用户断开）
    轨迹 ['tool_a: 启动', 'tool_b: 启动', 'tool_a: 被取消', 'tool_b: 被取消']

读轨迹：两个工具都启动了（并发是真的），取消到来后两个都被取消（CancelledError 从最外层的 task.cancel() 一路传进了两个子任务）。这里 TaskGroup 帮了大忙——取消一个 TaskGroup，它会取消组内所有任务再等它们退出，不会只取消一个、留下另一个当孤儿。如果这里是裸的两个 create_task，task.cancel() 只会取消 run 自己，两个子任务是「孤儿」，得手动一个个取消。

对服务端来说，这个机制要和 SSE 断开钩子接起来：客户端断开 → 找到对应的 run task → task.cancel() → 一路取消所有在跑的工具 → 释放并发槽与连接。取消是省钱也是省资源：一个被放弃的 run 继续跑完，白白烧 token 和连接。

8.3.5 任务泄漏：两种典型的漏

泄漏指的是「任务起了却没被正确收场」。有两种最常见的形态，先用 TaskGroup 的失败行为看清第一种：

async def boom():
    await asyncio.sleep(0.1)
    raise RuntimeError("tool 403")

async def run3(log):
    async with asyncio.TaskGroup() as tg:
        tg.create_task(slow_tool("ok_tool", 5.0, log))   # 还在跑
        tg.create_task(boom())                            # 0.1s 后失败

async def main():
    log = []
    try:
        await run3(log)
    except ExceptionGroup as eg:
        print(f"[3] 组内异常 {type(eg.exceptions[0]).__name__}: {eg.exceptions[0]}")
    print(f"    轨迹 {log}")

真跑输出：

[3] 组内异常 RuntimeError: tool 403
    轨迹 ['ok_tool: 启动', 'ok_tool: 被取消']

ok_tool 已经启动，但 boom 抛异常后它被立即取消——它没有继续傻等 5 秒。这是 TaskGroup 的「失败即取消」：一个任务失败，组会取消其余所有任务，不留下还在后台跑的任务。所以用 TaskGroup 起任务，本身就是一种防泄漏——没有孤儿。

第二种泄漏更隐蔽：裸 create_task 却忘了 await。看这段：

async def bg():
    await asyncio.sleep(30)          # 一个长跑的后台任务

async def main():
    asyncio.create_task(bg())         # 没有引用、没有 await —— 泄漏
    pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]
    print(f"[4] 未 await 的游离任务数 = {len(pending)}，仍存活 = {not pending[0].done()}")

真跑输出：

[4] 未 await 的游离任务数 = 1，仍存活 = True

这个 bg 任务没人引用、没人 await、没人取消——它会一直跑到 30 秒结束（或被事件循环关闭时强杀）。如果这是「每来一个请求就 create_task 一个后台任务」，而请求量上千，就会有上千个游离任务同时存活，内存和连接缓慢上涨，且没有任何监控能看到它们。

两种泄漏的防御手段正好相反：

8.3.6 把慢工具注入变成常驻演练

本节的慢工具是一次性的手工注入。在真实服务里，它应该变成可开关的故障注入：给工具注册表加一个「毒性开关」，按工具名注入固定延迟或异常，用来在测试和预发环境验证止损链路是通的。

class FaultyTool:
    """包装一个工具，按配置注入延迟 / 异常，用于故障演练。"""
    def __init__(self, name, inner, delay=0.0, raise_exc=None):
        self.name, self._inner = name, inner
        self._delay, self._raise = delay, raise_exc

    async def __call__(self, **kwargs):
        if self._delay:
            await asyncio.sleep(self._delay)     # 注入「慢」
        if self._raise:
            raise self._raise                     # 注入「错」
        return await self._inner(**kwargs)

这样测试里能一行切出「检索工具慢 10 秒」「Postgres 工具抛超时」等场景，断言 run 能在预算内止损。它比手改 asyncio.sleep 的值可靠——因为它是配置驱动的，能在 CI 里按矩阵跑各种组合（第 18.2 节的故障演练会复用这套注入）。止损能力不演练就等于没有：线上第一次遇到慢工具时才发现超时没配、取消没接，代价就是一次事故。

8.3.7 超时值怎么定

超时设太短会误杀正常请求，设太长等于没设。它不该拍脑袋，而该按这个调用的正常耗时分布来定：

几条经验：

• 按 p99 的 2–3 倍定，不是按平均值——平均值太乐观，会误杀正常的长尾请求。
• 分层设超时：单工具一个阈值、整轮 run 一个更大的阈值。单工具管「一个别太久」，整轮管「一轮别太久」，两级都要有。
• 区分只读与写入：只读超时可以安全重试，写入超时不能盲目重试（可能已经执行），要配幂等键——这是卷一的结论，卷二原样沿用。
• 超时是配置项：和并发上限一样，放第 2 章的分层配置里，压测出真实分布后再调。

8.3.8 小结与验收

本节给并发的 run 装上了「止损 + 取消 + 防漏」三件套：

• 超时止损：10 秒的慢工具在 1 秒超时后返回 {"error": "timeout"}，整轮不被拖住（实测 1.00s）。
• 取消传播：用户断开 → 取消 run → 两个在跑的工具都被取消，TaskGroup 保证不留孤儿。
• 防泄漏：TaskGroup 的失败即取消挡住了「兄弟任务泄漏」，裸 create_task 无引用则是要主动防的坑。

验收清单：

• 每个工具调用都有独立的超时，超时降级成 {"error": "timeout"} 回灌，不炸整轮
• 整轮 run 有一个更大的兜底超时
• 用户断开的钩子会取消对应的 run task
• 中间层没有 except BaseException / 裸 except: 吞掉取消
• 所有后台任务要么用 TaskGroup，要么保留引用并登记清理
• 超时值是配置项，按 p99 的 2–3 倍定

真实环境 / 真实模型下会怎样：本节的慢工具是 asyncio.sleep，取消它不花任何代价。真实工具（一次 HTTP 写、一次数据库写入）被取消时，底层连接可能已经发出、服务器可能已经执行——取消只保证「本地不再等」，不保证「远端没发生」。所以真实系统里超时后重试必须配幂等键。另一个真实差异是取消的传播路径更长：从 SSE 断开信号到实际取消 run，中间可能隔着 Redis 队列、worker 进程、HTTP 连接池，取消信号要能跨进程传（通常靠「在共享存储里标记 run 已取消，worker 轮询发现后自杀」）。最后，真实模型 API 的耗时主要是「生成 token」，超时要包在整个流式读取外面，而不是只包住建连——否则一个「开始很快、中途卡住」的响应会绕过超时，成为最难查的那种静默挂起。

阅读导航：上一节：8.2 背压与资源池：信号量限并发 · 下一节：9.1 事件协议：token/tool/step 统一 schema，TS 与 Python 共享类型 。