企业官网建设流程全解析

做Agent工具调用这一层的时候，我遇到过一个特别典型的Case：用户在界面上点了“停止”，后端却把这次中断当成了“超时”，自动把同一个任务重试了三遍，最后还把“重试成功”的结果返回给用户。用户当时就很纳闷——我明明已经取消了，你为什么还要替我执行完？

这个问题的本质在于：Agent工具调用链路里，“超时”和“用户取消”在表象上几乎一模一样，都是任务没有正常返回。但它们背后的信号来源完全不同，业务上需要的应对方式也完全相反。超时要重试、要降级，取消要立刻终止、要回滚。如果不在信号层把这两个语义分清楚，上层策略写得再漂亮，落到执行器里都会变成一锅粥。这篇文章聊的就是怎么在信号层把这两类中断区分开，并且给出一套可以直接抄进项目里的Python asyncio实现。

1. 为什么“超时”和“用户取消”总被当成一回事

1.1 工具调用链路里的两类中断

Agent的工具调用，本质上是一个有明确边界的异步任务：发起调用、等待外部系统响应、返回结果给推理循环。这个过程里，最常见的非正常结束只有两种。

一种是超时：给工具定了一个时间预算，比如30秒，超过预算仍然没有返回，系统主动放弃。另一种是取消：用户在界面上点了停止按钮，或者上游系统下发了一个revoke指令，要求立即中断当前任务。

从表面看，它们确实太像了。如果你的代码是“不管三七二十一，await一个任务，出了问题就catch”，那这两个信号都会被同一个异常处理器接住，然后走同一段逻辑。在单次调用里这个麻烦还不明显，但一旦放进Agent的多工具循环里，问题就会发酵。

我踩过这样一个坑：某个Agent流程要顺序执行“检索资料 → 调用大模型 → 写数据库”。用户在中途反悔，点了取消。结果因为“取消被当成超时”，系统自动重试了第一环节的检索，前两次都超时了，第三次终于成功，然后就继续往下跑了——用户根本不想让它继续，它却替用户把整套活全干完了。

1.2 混淆之后会引发哪几类故障

把超时当取消，或者把取消当超时，实际会引发三类故障。

第一类是资源浪费。超时的正确姿势是“重试”或“降级”，取消的正确姿势是“立即终止”。一旦两者反转，用户的取消意图会被当成超时，触发自动重试，相当于你按了电梯的紧急停止，电梯反而继续往下开，还开了好几层门。对Agent框架来说，这意味着算力、token配额、外部API调用额度全部白费。

第二类是副作用泄漏。工具调用经常是带副作用的：发HTTP请求、写数据库、调外部Webhook、扣费。如果用户取消了但系统仍然把它当成超时来重试，就可能出现重复扣款、重复发送消息等脏效应。反过来，一个真正超时的任务被当成用户取消，直接终止，则可能让一个已经写了一半的数据库事务停下来，留下脏状态。

第三类是监控与排障失真。当系统把取消和超时都记成同一种错误码时，SRE和算法工程师都看不懂线上到底发生了什么。容量规划、超时时间调整、重试策略优化，全都失去依据。我曾在一个项目里见过告警面板把“用户取消”全部计入“上游超时”，结果值班同学连续调了三轮超时阈值，问题却一点没改善。

这也正是“信号层区分”的必要性：在执行器（Executor）和业务逻辑之间加一层信号识别，让每一种中断都带着自己的身份标签向上传播。

2. 信号层设计：先给每种中断发一张“身份证明”

2.1 在哪里区分最合适

先想清楚一件事：区分不应该发生在业务层，而应该发生在信号层。

业务层关心的是“我下一步该干什么”——重试、回滚、还是终止。信号层才关心“刚才发生了什么”——是超时了、是取消了、还是工具自身报错。如果让业务层自己去猜，判断逻辑就会散落到每个工具的异常处理点里，最后一定维护不住。

一个合适的边界是：在执行器内部做信号分类，对外只暴露三类结果：正常结果、超时信号、取消信号。上层Agent循环只需要针对这三种结果写自己的策略。

打个比方：工具调用就像一条热线电话座席，“超时”是排队等了很久之后系统主动挂断，“取消”是用户自己挂了电话。你必须在电话交换机那一层就识别出是哪一种结束，而不是等后台客服去猜。交换机就是信号层，客服就是业务层。

2.2 自定义信号类型

为了在代码里把这些信号变成“一等公民”，第一步是定义各自的类型。这里建议用两个独立的异常类：

from dataclasses import dataclass @dataclass class TimeoutSignal(Exception): """工具调用超时信号""" tool_name: str timeout: float @dataclass class CancelSignal(Exception): """用户主动取消信号""" job_id: str reason: str = "user_cancelled"

有人会问：为什么不用Python自带的TimeoutError和asyncio.CancelledError？因为自带的类型语义太薄，而且在上层框架流转时很容易被吞掉。自定义异常的意义在于：可以携带业务上下文（哪个工具、超时时长、哪个任务被取消），不会被误认为是普通的编程异常，后续接监控系统时可以直接拿异常类名做错误码映射。

2.3 为什么不能只靠异常类型做判断

可能有同行觉得：根本不用这么麻烦，我在await外层写两个except，一个捕获CancelledError，一个捕获TimeoutError，不就能分开了？

这个想法在纯单任务场景下确实能用，但放到真实的Agent工具调用链路里就会四处漏风。具体原因我会在第4节详细展开，这里先剧透最重要的一点：asyncio.wait_for在发生超时时，会先从内部发起对任务的取消，然后再向上抛TimeoutError。这个过程中，任务的CancelledError已经发出来过了一轮，凡是中间挂着清理逻辑的协程，都会被额外触发取消流程，很容易和外部取消混在一起。

换句话说，异常类型是“最终结果”，而你真正需要区分的是“谁先发出中断信号”。信号层要做的，是把“谁先发出中断”这件事显式化，而不是靠异常类型的差异去猜。

3. Python asyncio 三路等待：一个可复现的具体实现

3.1 核心思路：让时钟、取消、结果三方赛跑

先给出方案的核心思路，理解了它，后面的代码就是水到渠成。

把一次工具调用看成一场“三方赛跑”：

• 选手一：工具调用任务本身。
• 选手二：用户取消事件。
• 选手三：挂在旁边的超时闹钟。

我们用asyncio.wait同时等待两个任务——工具任务和取消事件任务，并设置一个timeout参数当作闹钟。这三件事谁先发生，谁就先从wait里返回。换句话说：

• 工具任务先完成 → 正常返回。
• 取消事件先触发 → 判定为用户取消。
• wait的超时先到 → 判定为工具超时。

这个方案不依赖异常类型的语义猜测，也不依赖“内部先cancel再抛错”的隐晦行为，把三方竞赛的结果直接摆在明面上。

这里有一点需要强调：asyncio.wait的timeout参数本身不会自动取消任何任务，它只是到了时间就唤醒wait返回(done, pending)两个集合。这正是我们想要的，因为“谁触发中断”的信号源是干净的。

3.2 完整代码实现

下面给出一个可以直接抄进项目里的实现：

import asyncio from typing import Any, Callable, Sequence async def execute_tool_call( func: Callable[..., Any], args: Sequence[Any], *, tool_name: str, timeout: float, cancel_event: asyncio.Event, ) -> Any: """ 在信号层区分 正常完成 / 超时 / 用户取消。 - 正常完成：直接返回工具调用的结果。 - 超时：抛出 TimeoutSignal。 - 用户取消：抛出 CancelSignal。 """ tool_task = asyncio.create_task(func(*args)) cancel_task = asyncio.create_task(cancel_event.wait()) try: done, pending = await asyncio.wait( {tool_task, cancel_task}, timeout=timeout, return_when=asyncio.FIRST_COMPLETED, ) # 工具先完成，直接返回结果 if tool_task in done: return await tool_task # 剩下两种情况都需要取消还没结束的任务 for task in pending: task.cancel() # 取消事件先触发 -> 用户取消 if cancel_task in done: if not tool_task.done(): tool_task.cancel() raise CancelSignal(job_id="agent-main", reason="user_cancelled") # 走到这里说明是超时 if not tool_task.done(): tool_task.cancel() raise TimeoutSignal(tool_name=tool_name, timeout=timeout) finally: await asyncio.gather( *( task for task in (tool_task, cancel_task) if not task.done() ), return_exceptions=True, )

这个函数在使用上很简单。外部维护一个全局的或任务粒度的cancel_event，需要取消时调用cancel_event.set()。每次工具调用都带着这个事件进来。

代码里有几个细节值得单独说明。

第一，为什么要先判断tool_task in done。因为即使取消事件恰好和工具完成同时发生，只要工具已跑出结果，返回结果总是更符合直觉和业务一致性。用户取消指令虽然到了，但活已经干完，直接拿结果并没有额外副作用。

第二，为什么要把pending里的任务取消掉。当超时发生时，pending里通常同时有tool_task和cancel_task，手动取消是为了防止子任务悬挂在后台。finally里的gather是兜底清理，确保万无一失。

第三，为什么不用asyncio.shield。shield会把外层取消挡住，让表达式继续等待内层任务。它在某些场景里有用，但用它来区分超时与取消时，反而会让“到底是谁取消的”变得更模糊。这个坑在第4节展开。

3.3 三个演示case

为了确认这个方案真的能区分三条路径，我写了几个最小演示。

async def demo_normal(): async def tool(): await asyncio.sleep(0.1) return "done" cancel_event = asyncio.Event() try: result = await execute_tool_call( tool, [], tool_name="demo", timeout=2, cancel_event=cancel_event ) print("normal ->", result) finally: cancel_event.set() async def demo_timeout(): async def slow_tool(): await asyncio.sleep(10) return "too slow" cancel_event = asyncio.Event() try: await execute_tool_call( slow_tool, [], tool_name="demo", timeout=0.2, cancel_event=cancel_event ) except TimeoutSignal as e: print("timeout ->", e.tool_name, e.timeout) finally: cancel_event.set() async def demo_cancel(): async def slow_tool(): await asyncio.sleep(10) return "too slow" cancel_event = asyncio.Event() async def cancel_later(): await asyncio.sleep(0.2) cancel_event.set() canceller = asyncio.create_task(cancel_later()) try: await execute_tool_call( slow_tool, [], tool_name="demo", timeout=10, cancel_event=cancel_event ) except CancelSignal as e: print("cancel ->", e.reason) finally: canceller.cancel()

把这三个demo放进同一个事件循环里跑，终端里会看到：

normal -> done timeout -> demo 0.2 cancel -> user_cancelled

三条路径各自独立、互不干扰，信号在源头就被分清楚了。

4. asyncio 取消语义的三个深坑

这一节希望大家都读仔细。上一节的方案能成立，某种程度上就是因为绕开了下面这三个坑。如果要在现有代码里改造成这个方案，或者要排查线上诡异的中断问题，这些坑迟早会碰到。

4.1 CancelledError 不是 Exception

Python 3.8起，asyncio.CancelledError从Exception挪到了BaseException体系下。这意味着你在工具函数体里写except Exception是接不住它的。

很多Agent框架的工具注册表里，都喜欢在外面包一层“所有异常都转成ToolError”的处理器。如果这层处理器只捕获Exception，那么CancelledError会完全无视这层处理，一路穿透到事件循环里，任务被静默杀死。你的“统一异常处理”在用户取消面前是失灵的。

反过来，如果有人写了非常宽泛的except BaseException，那也要小心，这会把取消信号和真正的致命错误混在一起，同样危险。

我的建议是：在工具实现里不要大范围捕获BaseException；如果必须有CancelledError相关的兜底逻辑，应该用独立的except asyncio.CancelledError块，并且在处理完后重新抛出，因为取消信号是需要继续向上传递的。

4.2 wait_for 会偷偷取消你的任务

asyncio.wait_for是很多人选择的最短路径，它确实好用：传入超时时间，超时了就抛asyncio.TimeoutError。但它的实现有一个副作用——超时发生时，它首先会向内部的任务发起cancel()，再抛出超时异常。

也就是说，当超时发生时，你的任务其实已经被“取消”过了一轮。任务内部的finally、清理逻辑、资源释放都会被触发。如果此时还有别的外部取消信号同时到达，就很难分清到底哪个信号才是真正的元凶。

再叠加一个更隐蔽的问题：wait_for还有一个取消竞争的边界场景。外部任务被取消时，wait_for会把内部的取消往上传递，这种情况下你收到的是CancelledError而不是TimeoutError。于是代码里的except asyncio.TimeoutError完全没机会执行，你以为的超时处理逻辑全部跳过。

所以在信号层做区分时，我建议放弃wait_for，改用第3节的三路等待方案。核心区别是：三路等待把“超时”当成一个外部事件来处理，而不是让超时先伪装成取消再去触发异常。

4.3 shield 别乱用

asyncio.shield是用来给任务加保护罩的：外部取消不会中断内部任务。这听起来很适合“工具调用不该被轻易取消”的场景，但实际用起来，它会把信号来源搅浑。

比如你写await asyncio.shield(tool_task)，外面用户取消时，这个await表达式会抛出CancelledError，但底层tool_task还在继续跑。如果你不做任何后续处理，这个工具调用就会悬挂在后台，完全脱离你的控制，甚至可能在你已经判定“取消”之后继续产生副作用。

盾牌的意义是“让任务继续完成”，比如你想保证某个写库操作一定要落地，不想被取消打断。但作为信号层区分手段，盾牌是反模式。正确做法永远是用显式的事件和控制权来管理任务生命周期，而不是把取消“藏起来”。

5. 落到Agent框架：执行器加信号消费策略

5.1 AgentRunner 执行器骨架

前两节解决了“如何区分信号”的核心问题，接下来要解决的是“信号区分完之后怎么办”。Agent的工具循环里，每一轮都可能是工具调用、LLM推理、记忆更新的组合。上面实现的execute_tool_call基本可以无缝接入一个执行器。

下面是一个简化版AgentRunner骨架：

from dataclasses import dataclass from typing import Any, Callable, Dict @dataclass class ToolResult: ok: bool data: Any = None error: str = "" retryable: bool = False class AgentCancelled(Exception): """Agent执行被用户取消""" class AgentRunner: def __init__(self): self.cancel_event = asyncio.Event() self.tools: Dict[str, Callable] = {} self.max_retry = 2 self.timeout_budget: Dict[str, float] = { "search": 5, "llm": 60, "db_query": 15, "file_write": 30, } def request_cancel(self): self.cancel_event.set() async def _invoke_tool(self, name: str, args: tuple): if name not in self.tools: return ToolResult(ok=False, error=f"unknown tool: {name}") func = self.tools[name] timeout = self.timeout_budget.get(name, 10) try: data = await execute_tool_call( func, args, tool_name=name, timeout=timeout, cancel_event=self.cancel_event, ) return ToolResult(ok=True, data=data) except TimeoutSignal: return ToolResult( ok=False, error=f"timeout after {timeout}s", retryable=True, ) except CancelSignal: raise AgentCancelled() async def run(self, plan): results = [] try: for step in plan: if step.kind == "tool": result = await self._invoke_tool(step.name, step.args) # 超时分支：按需重试 if not result.ok and result.retryable: for attempt in range(1, self.max_retry + 1): data = await self._retry_tool(step, attempt) if data is not None: result = ToolResult(ok=True, data=data) break results.append(result) elif step.kind == "reason": # 调LLM等推理逻辑 ... except AgentCancelled: await self._rollback(results) return {"status": "cancelled", "partial": results} return {"status": "finished", "results": results}

可以看到，执行器把三个信号分得清清楚楚：

• 超时信号返回一个retryable=True的失败结果，交给Agent循环决定是否重试。
• 取消信号直接抛AgentCancelled，中断整个执行循环。
• 工具自身异常则作为普通ToolResult处理，由Agent根据上下文决定是否更换策略。

5.2 超时分支：重试与降级

超时分支的策略，本质上属于“系统繁忙”的一类。常见做法有三种。

一是重试。适合幂等、且只是偶发慢的工具。注意重试次数要有上限，每次重试最好配合退避（backoff），不要做无脑死循环。

二是降级。比如搜索工具超时，可以用本地缓存或一个已有的快照临时顶替。Agent会让模型知道“当前结果可能不是最新”，然后继续后续步骤。

三是直接跳过。某些工具的结果对整体任务不是必须的，超时后可以让Agent继续，只是缺失这个结果。

关键是：超时不该中断整个Agent任务，取消必须中断整个Agent任务。这一对语义差异，是整个信号层方案最核心的产出。

5.3 取消分支：回滚与资源释放

取消发生后，第一步永远是资源清理：把还没结束的工具任务全部取消，关闭HTTP连接、数据库会话、文件句柄。这里最容易漏掉的一个点是：很多工具函数内部通过finally来释放资源，而execute_tool_call在抛出CancelSignal之前已经执行了tool_task.cancel()，所以finally会正常触发。如果你用的是别的方案，记得要显式调用task.cancel()，否则工具内部挂在后台的连接会一直占着资源。

第二步是业务回滚。如果工具调用已经产生了副作用，比如调用了外部API，一般不存在一个万能的回滚方法，只能针对具体工具提供补偿逻辑。我的经验是在工具注册表里增加一个可选的rollback字段，取消时由执行器统一调用。

第三步是通知上游。Agent执行的取消，通常意味着整个用户会话要进入“已取消”状态。这时候应该把部分结果、取消原因、可续接的断点信息返回给上层，这样用户在UI上能看到“任务已取消，已完成2/5个步骤”之类的明确状态。

6. 常见问题与排查技巧实录

6.1 高频问题排查

把这些年在实际项目里遇到的典型问题整理成一张速查表：

6.2 测试与监控建议

信号层方案是否可靠，必须靠自动化测试兜底。我习惯给每个工具调用备三套测试用例：

• 正常路径：mock一个快速返回的异步函数，断言返回结果。
• 超时路径：mock一个一直sleep不返回的函数，超时时间设为0.01秒，断言抛出TimeoutSignal。
• 取消路径：mock一个sleep函数，在0.01秒后触发cancel_event.set()，断言抛出CancelSignal。

测试代码按第3节的三个demo扩写即可，不需要额外依赖。关键是三条case必须分别断言独立的异常类型，这样才能防止后续有人把实现改成wait_for之后，回归测试把问题暴露出来。

监控方面，建议在执行器里对每个信号打点：

# 伪代码，接入Prometheus等监控 metrics.tool_call_total.labels( tool=tool_name, result="ok" | "timeout" | "cancelled" | "error", ).inc()

把timeout和cancelled分开打点之后，线上告警和容量评估会立刻清晰不少。我见过一个Agent服务把取消全记成了超时，导致告警平台每五分钟响一次，值班同学误以为是第三方API不稳定，反复调超时时间，实际上问题全在用户端取消的语义没有剥离开。

7. 个人经验：信号层方案帮我避开的线上事故

做了这么多年的异步任务调度，一个深切的体会是：Agent工具调用这一层，最容易出问题的不是工具写得好不好，而是中断之后如何体面收场。超时和取消在表象上太像了，像到很多框架干脆用同一个错误码敷衍过去——可它们对用户来说是完全相反的体验。用户取消时，系统的任务是“停下来”；系统超时时，用户期待的是“再试一次”。

我在信号层做区分的方法，本质上只是把一个常识问题技术化了：先弄清楚谁先发出了信号，再决定怎么应对。如果要用一句话总结这套方案的价值，那就是“把要不要继续执行的决策权，从异常机制手里拿回来，交还给Agent的业务逻辑”。

最后再分享一个小技巧：如果你接手的老代码已经四处是asyncio.wait_for，一时半会儿改不动，可以先在工具函数的最外层加一个兜底except asyncio.CancelledError，在里面判断asyncio.current_task().cancelling()，配合日志把取消来源打出来。短期内能缓解排查困境，但长期来看，还是建议往三路等待这个方向收敛。信号层干净了，上层策略怎么做都是顺的。