1. 为什么非要在信号层较这个真先讲一个我上周实际遇到的场景。早上起来打开后台面板看到一排工具调用失败的日志清一色写着 timeout exceeded。我下意识地认为这是某个第三方 API 又抽风了于是顺手给相关链路加了重试策略。结果一个小时之后线上报来一个新问题用户在界面里点了“停止生成”Agent 却还在没完没了地调用工具甚至把重试当做新的请求发了出去。那一刻我才意识到——超时和用户取消这两种完全不同的信号在当初落代码的时候被我混在了一个异常分支里。这件事在 Agent 开发里非常典型。尤其随着多步骤 Agent、人工审批环节、异步任务队列越来越普遍“这次工具调用到底是怎么没的”已经不是一个日志归类问题而是一个真实的业务正确性问题。你想想看一个 agent 在工作流里调用了支付接口、调用了数据库写入、调用了外部通知服务如果代码层面分不清“它是因为等太久被强制中断”还是“用户主动让整个任务停下来”后续的动作就全乱了。前者通常应该触发重试、熔断、降级后者则应该立刻停止一切副作用、清理资源、不回滚任何已经完成的步骤。很多人第一反应是这有什么难的超时抛 TimeoutError取消抛 CancelledError 不就行了。理论上是这样但真正落到 Agent 工具调用这个具体场景事情会变得非常拧巴。因为 Agent 的工具调用不会是简单的一层 try except它会涉及异步任务、线程池、进程信号、底层 HTTP 连接、外部子进程每一层都有自己的“超时”和“取消”语义。如果不在信号层把这些状态钉死、建模清楚迟早会在某个环节出现“张冠李戴”。这篇文章就是把我在 Agent 工具调用这一块关于超时和用户取消的区分思路完整摊开。我会从信号层的概念讲起然后给出一套三层划分的代码方案包括异常类型、取消令牌、进程信号三个维度的配合最后把我在实际项目中踩过的坑和排查经验整理成速查表。无论你是在用 Python 写单机 Agent还是在做多 Agent 编排平台这套区分策略都能直接落地。2. 先说清楚什么是“信号层”2.1 把异常和任务状态当成一条信号链路“信号层”这个词听起来有点玄乎其实用大白话讲就是程序在“事情已经没法正常完成”时用来传递“为什么停下来”的那一套机制。在 Python 里面它主要由三类东西组成异常类型、任务状态、进程信号。这三类东西背后都有明确的语义如果设计得当你在任何一个环节拿到一个状态都能反推出这个 Agent 工具调用到底经历了什么。我用一个生活化的类比来解释。你去餐厅点了个外卖最后没吃到饭原因无非两种外卖超时没送到或者你自己取消了订单。虽然结果都是“没吃到”但处理逻辑完全不一样。超时平台会给你赔券、商家会重新做取消订单直接终止钱原路退回。如果你在系统里把这两种状态都记成“未送达”那之后的赔付、退款、商家结算就全乱了。信号层要做的就是给“未送达”这个结果打上精确的标签。放到 Agent 工具调用的语境里这个标签体系通常需要覆盖三个层面。最外层是 Python 的 BaseException 体系比如 asyncio.TimeoutError、asyncio.CancelledError这属于异常信号第二层是任务对象的状态比如 concurrent.futures.Future 的 CANCELLED、CANCELLED_AND_NOTIFIED、RUNNING、FINISHED这属于任务生命周期信号第三层是操作系统级的进程信号比如用户按 CtrlC 触发的 SIGINT或者 kill 命令发来的 SIGTERM这属于外部干预信号。一个设计良好的 Agent 工具调用封装必须把这些信号统一映射到自己的领域模型里而不是直接原样抛给上层业务代码。2.2 超时和用户取消在信号层的本质差异超时和用户取消在信号层面最核心的差异在于超时是一个“单向被动”的事件取消是一个“双向交互”的事件。超时的本质是调用方等不起了。你设置了一个 deadline到了时间还没拿到结果于是主动放弃等待。这个时候底层那个真正在工作的任务——比如一个 HTTP 请求、一个数据库查询——其实可能还在跑只是你的代码不再等它了。这一点非常重要。asyncio.wait_for 超时之后内部任务会被取消但如果你用的是线程池或外部子进程任务不一定真的停了。这也是很多人踩坑的地方你以为超时之后资源就释放了实际上底层线程还在那里执行。用户取消的本质是任务的上游叫停了。用户点击“停止”前端发来一个 cancel 请求这时候发生的是“协作式取消”代码要在某个安全的暂停点主动响应取消指令。在 Python 的 asyncio 生态里协作式取消由 CancelledError 来传达任务收到取消请求后会在下一个 await 点抛出 CancelledError让你有机会清理资源、记录状态然后再把取消信号继续向上传播。还有一个容易被忽略的差异是取消的“传播路径”。超时通常只终止“当前这一层”的等待它和任务树里的其他兄弟节点没关系而用户取消往往要沿着任务树级联传播父任务取消了所有子任务、孙任务都要跟着取消。如果只是把超时和取消混在一个 except Exception 里处理这把级联关系也会被处理得稀碎。后面我将要展示的方案核心就是让“超时”和“取消”走两条完全独立的代码路径互不干扰。3. 一套可落地的三层信号区分方案3.1 第一层用“异常类型”给超时下定义在 Python asyncio 生态里超时的标准做法是用 asyncio.wait_for 或者 Python 3.11 引入的 asyncio.timeout() 上下文管理器。超时到达时协程内部会先取消当前任务然后抛出 TimeoutError。注意这里有个容易混淆的点asyncio.wait_for 内部取消任务时任务本身会经历一次 CancelledError 的抛出和吞没但最终交给调用方的是 TimeoutError。所以第一层的核心原则是所有“等待超时”的异常统一收敛为 TimeoutError绝不允许内部的 CancelledError 直接冒泡到业务层。下面是一段封装示例我在实际项目里基本就是按这个骨架写的import asyncio import logging class ToolInvokeError(Exception): 工具调用通用异常基类 class ToolTimeoutError(ToolInvokeError): 工具调用超时 async def call_tool_with_timeout(tool, *args, timeout: float 10.0): try: async with asyncio.timeout(timeout): return await tool.invoke(*args) except TimeoutError as e: # 这里一定是真的超时 logging.warning(ftool {tool.name} invoke timeout after {timeout}s) raise ToolTimeoutError(f{tool.name} timeout) from e except asyncio.CancelledError: # 这个分支要非常谨慎外层取消信号不该被 TimeoutError 包装 # 直接向上传播让上层决定是转为取消还是重新调度 logging.info(ftool {tool.name} cancelled by outer task) raise这里有一个关键决策except asyncio.CancelledError 分支不能和 TimeoutError 合并处理更不能把 CancelledError 改写为普通 Exception。一旦你把它包成别的异常就要额外引入一层机制来判断“这个普通异常到底是不是取消”反而弄巧成拙。在非异步场景比如你想用 ThreadPoolExecutor 执行一个会阻塞的 SDK 调用超时的捕获逻辑变成这样import concurrent.futures import time def call_tool_in_thread(tool, *args, timeout: float 5.0): with concurrent.futures.ThreadPoolExecutor(max_workers2) as pool: future pool.submit(tool.invoke, *args) try: return future.result(timeouttimeout) except concurrent.futures.TimeoutError: # 注意future 还在跑这里需要额外决定是否真正取消线程 # Python 线程无法强行 kill只能在业务代码里埋取消点 logging.warning(ftool {tool.name} thread timed out after {timeout}s) raise ToolTimeoutError(f{tool.name} timeout)这段代码里有个值得注意的点future.result(timeouttimeout) 抛出 concurrent.futures.TimeoutError 之后线程池里的任务其实还在继续运行。所以你只是在“信号的接收层”判断了超时并没有真正终止执行。这也是为什么很多 Agent 框架宁可把工具调用全部放到异步协程里也不愿意用线程池——协程的超时是协作式取消起码给了业务代码一个清理的机会。3.2 第二层用“取消令牌”给用户取消下定义用户取消和超时不同它不是“等不起”而是“不想等了”。如果只有一个简单的一层调用你可以只依赖 CancelledError 来传递取消。但 Agent 工具调用往往会有多级封装协调器、工具执行器、HTTP 传输层每一层都可能捕获异常、做日志、做埋点。如果取消信号只表现为一个单纯的 CancelledError上层在捕获它时很难立刻知道“这是用户主动取消”还是“某个内部环节的取消”。我建议引入一个显式的取消令牌CancellationToken让用户取消这个语义不依赖异常类型来传达。令牌的核心是一个状态标记加一组回调谁持有令牌谁就可以主动发起取消并且这个取消意图可以跨协程、跨线程传递。下面是一个简易实现import asyncio import threading from enum import Enum class CancelState(Enum): NORMAL normal CANCELLED_BY_USER cancelled_by_user CANCELLED_BY_TIMEOUT cancelled_by_timeout class CancellationToken: def __init__(self): self._state CancelState.NORMAL self._lock threading.Lock() self._waiters [] def cancel_by_user(self): 用户主动取消 with self._lock: if self._state is not CancelState.NORMAL: return self._state CancelState.CANCELLED_BY_USER waiters list(self._waiters) self._waiters.clear() for waiter in waiters: waiter.set() def cancel_by_timeout(self): 超时取消和用户取消区分开 with self._lock: if self._state is not CancelState.NORMAL: return self._state CancelState.CANCELLED_BY_TIMEOUT waiters list(self._waiters) self._waiters.clear() for waiter in waiters: waiter.set() property def is_cancelled(self): return self._state is not CancelState.NORMAL property def state(self): return self._state async def wait_for_cancel(self): loop asyncio.get_running_loop() event asyncio.Event() self._waiters.append(event) try: await event.wait() finally: self._waiters.remove(event)这个令牌的妙处在于它可以同时挂在多个工具调用上。当用户在 Agent 界面点击“停止”协调器只需要调用一次 token.cancel_by_user()所有监听这个令牌的工具调用都会醒来并发现“这是用户取消”而不是去猜测异常类型。这样在日志系统里你可以非常明确地记录 cancel_sourceuser之后的项目复盘、成本计量、重试策略都有了可靠依据。有了取消令牌之后工具调用的主流程改成这样async def call_tool_with_cancellation(tool, token: CancellationToken, timeout: float): # 组合使用外层超时内层监听取消令牌 try: async with asyncio.timeout(timeout): cancel_task asyncio.create_task(token.wait_for_cancel()) invoke_task asyncio.create_task(tool.invoke()) done, pending await asyncio.wait( {cancel_task, invoke_task}, return_whenasyncio.FIRST_COMPLETED, ) if cancel_task in done and token.is_cancelled: # 用户取消优先占位 for p in pending: p.cancel() raise UserCancelError(cancelled by user) return invoke_task.result() except TimeoutError: token.cancel_by_timeout() raise ToolTimeoutError(timeout)这里的关键是 asyncio.wait(FIRST_COMPLETED) 让你同时监听两个信号取消令牌和工具调用本身。哪个先到就知道这轮调用是怎么终结的。用户取消和超时走了完全不一样的分支不会出现“日志里写着超时其实是用户点了停止”的混乱。3.3 第三层用进程信号兜住最外层的被动取消前面两层解决的是程序内部的“超时”和“取消”。但 Agent 服务运行在真实环境里还有一个来源不可忽略用户直接按下 CtrlC或者运维通过 kill 命令终止进程。这一层如果拦截不好整个 Agent 会“死得不明不白”工具调用挂在半空中资源没人清理。Python 的 signal 模块可以处理 SIGINT 和 SIGTERM。单独用 signal.signal 注册处理器很容易踩坑因为处理器只在主线程执行而且会被 asyncio 事件循环打断。在 asyncio 应用中正确做法是使用 loop.add_signal_handler把系统信号转成回调和事件再转发给取消令牌import asyncio import signal async def main(): token CancellationToken() loop asyncio.get_running_loop() def on_sigint(): # 用户按下了 CtrlC等价于用户取消 print(receive SIGINT, treating as user cancel) token.cancel_by_user() def on_sigterm(): # 进程被 kill可以记为外部终止也可以归为用户取消 print(receive SIGTERM, treating as user cancel) token.cancel_by_user() loop.add_signal_handler(signal.SIGINT, on_sigint) loop.add_signal_handler(signal.SIGTERM, on_sigterm) # 后面正常启动 Agent 任务 await run_agent(token)这样一来从最外层的 CtrlC到中间的 Agent 协调器取消请求再到工具调用的 CancelledError整条链路都是同一个语义用户取消。你不再需要事后靠日志里那点蛛丝马迹去猜“到底是网线断了还是用户点了停止”。需要额外提醒一句loop.add_signal_handler 在 Windows 上并不完全可靠尤其是 SIGTERM。如果你要跨平台支持进程信号建议在 Windows 下退回到 signal.signal并把回调封装成线程安全的投递方式。我自己的经验是在 Windows 上尽量以“事件循环收到退出消息”作为进程终止信号而不是依赖 SIGTERM。4. 完整实操跑通一次“被取消”的工具调用4.1 搭建最小实验环境理论说再多不如跑一遍。我用一个极简的模拟场景来做完整演示有一个假想的“慢速工具”比如一个 sleep 10 秒才返回结果的函数Agent 调用它同时在 3 秒的时候模拟用户点击“停止”。我们观察信号层的日志输出看能不能在日志里准确分辨超时和用户取消。环境准备就一个 Python 3.11不需要任何第三方库标准库里的 asyncio、concurrent.futures、signal、dataclasses 足够。我建议你新建一个文件 demo_agent_cancel.py把下面的代码按段贴进去边跑边看输出。4.2 完整代码与调用时序先定义领域异常和一个 PretendToolimport asyncio import logging import time from dataclasses import dataclass, field from typing import Any logging.basicConfig(levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s) class ToolError(Exception): pass class ToolTimeoutError(ToolError): pass class UserCancelError(ToolError): pass class PretendTool: 假装是一个很慢的外部工具 def __init__(self, name: str, delay: float): self.name name self.delay delay async def invoke(self, payload: Any): logging.info(ftool {self.name} started, delay{self.delay}s) await asyncio.sleep(self.delay) return fresult from {self.name}接着是实现取消令牌的状态存储。为了便于观察我用一个事件日志列表记录每一次信号流转dataclass class CallOutcome: tool_name: str signal: str # timeout | user_cancel | success | tool_error detail: str events: list field(default_factorylist) elapsed: float 0.0然后是核心的调用封装。这一步我要把前面讲的两层方案合并到一个函数里因为它反映的是真实场景下的组合使用async def call_tool(tool: PretendTool, timeout: float, cancel_event: asyncio.Event, outcome: CallOutcome): start time.monotonic() outcome.tool_name tool.name try: async with asyncio.timeout(timeout): invoke_task asyncio.create_task(tool.invoke(None)) cancel_task asyncio.create_task(cancel_event.wait()) done, pending await asyncio.wait( {invoke_task, cancel_task}, return_whenasyncio.FIRST_COMPLETED, ) if cancel_task in done and cancel_event.is_set(): # 用户取消立刻取消没有完成的 invoke_task for p in pending: p.cancel() outcome.signal user_cancel outcome.detail cancel_event triggered by user outcome.events.append(signalcancel_event fired first) raise UserCancelError(user cancelled) result invoke_task.result() outcome.signal success outcome.detail str(result) outcome.events.append(signalinvoke_task completed) return result except TimeoutError: # 超时需要在超时分支里把可能存在的 cancel_task 取消掉 for p in asyncio.all_tasks(): if p is not asyncio.current_task() and p.get_name() invoke_task: p.cancel() outcome.signal timeout outcome.events.append(signalasyncio.timeout fired first) raise ToolTimeoutError(f{tool.name} timeout) except asyncio.CancelledError: # 外层用户取消信号比如进程信号或上层协调器取消 outcome.signal user_cancel outcome.events.append(signalouter CancelledError propagated) raise finally: outcome.elapsed time.monotonic() - start上面代码里有一个需要特别说明的点超时分支里清理 pending 任务的方式我用了 asyncio.all_tasks() 去筛这个写法在生产代码里不推荐因为很容易误杀其他无关任务。更稳的做法是保存 invoke_task 和 cancel_task 的引用在超时分支里直接对引用执行 cancel。下面这段是更干净的写法async def call_tool_v2(tool: PretendTool, timeout: float, cancel_event: asyncio.Event, outcome: CallOutcome): start time.monotonic() outcome.tool_name tool.name invoke_task asyncio.create_task(tool.invoke(None)) cancel_task asyncio.create_task(cancel_event.wait()) try: async with asyncio.timeout(timeout): done, pending await asyncio.wait( {invoke_task, cancel_task}, return_whenasyncio.FIRST_COMPLETED, ) if cancel_task in done and cancel_event.is_set(): for p in pending: p.cancel() outcome.signal user_cancel outcome.events.append(signalcancel_event fired first) raise UserCancelError(user cancelled) result invoke_task.result() outcome.signal success outcome.events.append(signalinvoke_task completed) return result except TimeoutError: cancel_task.cancel() invoke_task.cancel() outcome.signal timeout outcome.events.append(signalasyncio.timeout fired first) raise ToolTimeoutError(f{tool.name} timeout) finally: outcome.elapsed time.monotonic() - start两种写法我都在项目里见过如果你要复用我的代码用 v2 就好。模拟用户取消的主程序我用一个 0.5 秒后 set 的 asyncio.Event 来模拟用户点击“停止”async def main(): tool PretendTool(slow-api, delay10.0) outcome CallOutcome(tool_nametool.name, signal, detail) cancel_event asyncio.Event() # 延迟 1.5 秒模拟用户取消 async def simulate_user_cancel(): await asyncio.sleep(1.5) logging.info(simulate user cancel now) cancel_event.set() simulator asyncio.create_task(simulate_user_cancel()) try: await call_tool_v2(tool, timeout3.0, cancel_eventcancel_event, outcomeoutcome) except UserCancelError: logging.info(catched UserCancelError, no retry) except ToolTimeoutError: logging.info(catched ToolTimeoutError, should retry) await simulator print(\n--- outcome detail ---) print(ftool: {outcome.tool_name}) print(fsignal: {outcome.signal}) print(fdetail: {outcome.detail}) print(fevents: {outcome.events}) print(felapsed: {outcome.elapsed:.2f}s) if __name__ __main__: asyncio.run(main())运行之后会看到这样的关键日志2025-01-10 12:00:01 [INFO] tool slow-api started, delay10.0s 2025-01-10 12:00:02 [INFO] simulate user cancel now --- outcome detail --- tool: slow-api signal: user_cancel detail: cancel_event triggered by user events: [signalcancel_event fired first] elapsed: 1.50s注意 outcome.signal 是 user_cancel不是 timeout尽管调用方 timeout 设置的是 3 秒。因为取消事件在 1.5 秒先到了。这正是信号层分离的直观体现。如果你把 simulate_user_cancel 里的 sleep 时间改成 5 秒那么 3 秒超时会先触发最终 signal 会变成 timeout。同一个工具、同一个封装不需要改任何逻辑输出的信号就能准确反映“到底是超时还是用户取消”。4.3 输出结果怎么解读实际跑通之后你会发现最大的收益在日志和后续动作。当 outcome.signal 是 user_cancel 时Agent 协调器的处理策略是直接终止当前任务不重试不补偿顶多记一条“用户主动取消”。当 outcome.signal 是 timeout 时协调器会根据工具属性决定重试次数、是否降级或者把这次调用标记为失败。我甚至会把 outcome 对象序列化成 JSON写进结构化日志之后做计量分析时直接按 signal 字段分桶统计一目了然。这套封装的价值还体现在可测试性上。你可以不依赖真实网络单纯用 event.set() 来模拟用户取消用 sleep 来控制超时时长把 Agent 的“超时/取消”行为作为一等公民进行单测。如果你只是想凭异常类型猜状态单测永远覆盖不了“用户取消先到”这条路径。5. 踩坑实录与排查速查表5.1 wait_for 会把内部取消伪装成超时的坑这是我在代码审查里见过最多的隐蔽问题。假设你在一个已有代码里用了 asyncio.wait_for(do_something(), timeout10)然后 do_something 内部因为某些原因抛出了 CancelledError——可能是它的上层任务被 cancel也可能它自己内部调用了 cancel。wait_for 的默认行为是超时到达时它先取消内部任务再抛 TimeoutError但如果内部任务已经因为外部取消而处于取消中wait_for 可能直接把内部任务的 CancelledError 转成 TimeoutError 抛给调用方。结果就是你以为日志里那个 timeout 就是真的超时其实可能是“任务被外部取消”的误报。要规避这个坑最简单的手段是不要在同一个函数里既做超时控制又做取消控制而是像我前面那样把取消事件和调用任务交给 asyncio.wait 同时监听谁先完成谁说话。还有一个很现实的场景Agent 框架通常会为每个工具调用设置默认超时比如 30 秒。用户点了停止后框架会一层层 cancel 任务。如果工具调用内部用了 wait_for用户取消产生的 CancelledError 在到达业务层时已经被包装成了 TimeoutError。上层协调器一看是 timeout就触发重试于是出现了“用户取消后还在被重试”的诡异行为。破解方法就是先让取消信号在 async with asyncio.timeout: 内自由传播不要着急捕获。5.2 线程池 Future.cancel() 只对未启动任务有效如果你没法把所有工具调用都改造成协程线程池方案的坑也需要提前知道。concurrent.futures.Future 的 cancel() 方法有一个硬性限制只有任务尚未开始运行时cancel() 才能返回 True 并真正取消。一旦任务已经在线程里跑起来了cancel() 返回 False底层函数照跑。实际上有很多 Agent 工具是同步阻塞型 SDK比如某些云厂商的 Python SDK你只能在业务代码里埋“取消检查点”。我通常的做法是给这类工具传入一个 dict 作为取消状态容器工具内部在每个循环迭代里检查这个 dict 里的 is_cancelled 字段。虽然做不到立即杀死线程但至少能在下一个检查点优雅退出。下面是一个很简单的骨架def long_running_tool_sync(config, cancel_slot: dict): for i in range(1000): if cancel_slot.get(is_cancelled): return cancelled time.sleep(0.01) return done def call_in_thread_checked(tool_fn, cancel_slot: dict, timeout: float): with ThreadPoolExecutor(max_workers1) as pool: future pool.submit(tool_fn, cancel_slot) try: return future.result(timeouttimeout) except concurrent.futures.TimeoutError: cancel_slot[is_cancelled] True raise ToolTimeoutError(timeout)在这个设计里超时发生时会主动往取消槽里写标记线程任务在下一个迭代点自行退出。注意它并不能保证立刻释放资源但至少不会出现“超时之后线程还在执行危险操作”的情况。对于内部是长连接的工具你还得在 finally 里做连接关闭。5.3 常见信号混乱场景排查速查表下面是我在 Agent 工具调用这一路踩出来的高频问题速查表遇到类似日志可以直接对照现象可能原因排查方向日志显示 timeout但用户明明点了停止wait_for 把 CancelledError 包装成 TimeoutError改用 asyncio.wait 双监听取消信号走独立路径用户取消之后工具还在重试业务层把取消当成了超时触发了重试策略检查异常类型分支确认 UserCancelError 不在重试范围内进程 CtrlC 后被 asyncio 吞掉任务挂死signal.signal 处理函数与事件循环不兼容改用 loop.add_signal_handler线程池超时后底层任务依然在写库future.result(timeout) 只是不再等待不是终止线程引入取消槽在工具内部做协作式退出多级任务取消时子任务只取消了一层CancelledError 被某层 except Exception 捕获吞掉全局搜索 except Exception确认 CancelledError 单独处理日志里出现“cancelled_by_timeout”这种模糊状态取消令牌初始化时未区分状态来源使用枚举区分 CANCELLED_BY_USER 和 CANCELLED_BY_TIMEOUT排查的时候我个人的习惯是先看最外层事件来源进程信号、上层协调器、还是本层计时器。来源定了再往底层看异常类型。如果日志系统里能带上类似 outcome.events 这样的信号序列定位问题的速度会快一个数量级。最后再分享一个小技巧。如果你用 Python 3.11 以上版本可以在自定义的取消令牌里顺便记录 cancel 的调用栈快照这样以后线上出现“用户取消”疑惑时你能直接知道是哪个模块、哪一行发起的取消。这个信息在排查多 Agent 编排平台的偶发挂起问题时价值巨大。我在一次复盘里就是靠取消栈快照发现了某个后台任务误调用了 token.cancel_by_user()源头是一个 API 路径写错导致的误触。信号层设计的投入往往会在这种最不起眼的时刻回本。
1. 为什么非要在信号层较这个真先讲一个我上周实际遇到的场景。早上起来打开后台面板看到一排工具调用失败的日志清一色写着 timeout exceeded。我下意识地认为这是某个第三方 API 又抽风了于是顺手给相关链路加了重试策略。结果一个小时之后线上报来一个新问题用户在界面里点了“停止生成”Agent 却还在没完没了地调用工具甚至把重试当做新的请求发了出去。那一刻我才意识到——超时和用户取消这两种完全不同的信号在当初落代码的时候被我混在了一个异常分支里。这件事在 Agent 开发里非常典型。尤其随着多步骤 Agent、人工审批环节、异步任务队列越来越普遍“这次工具调用到底是怎么没的”已经不是一个日志归类问题而是一个真实的业务正确性问题。你想想看一个 agent 在工作流里调用了支付接口、调用了数据库写入、调用了外部通知服务如果代码层面分不清“它是因为等太久被强制中断”还是“用户主动让整个任务停下来”后续的动作就全乱了。前者通常应该触发重试、熔断、降级后者则应该立刻停止一切副作用、清理资源、不回滚任何已经完成的步骤。很多人第一反应是这有什么难的超时抛 TimeoutError取消抛 CancelledError 不就行了。理论上是这样但真正落到 Agent 工具调用这个具体场景事情会变得非常拧巴。因为 Agent 的工具调用不会是简单的一层 try except它会涉及异步任务、线程池、进程信号、底层 HTTP 连接、外部子进程每一层都有自己的“超时”和“取消”语义。如果不在信号层把这些状态钉死、建模清楚迟早会在某个环节出现“张冠李戴”。这篇文章就是把我在 Agent 工具调用这一块关于超时和用户取消的区分思路完整摊开。我会从信号层的概念讲起然后给出一套三层划分的代码方案包括异常类型、取消令牌、进程信号三个维度的配合最后把我在实际项目中踩过的坑和排查经验整理成速查表。无论你是在用 Python 写单机 Agent还是在做多 Agent 编排平台这套区分策略都能直接落地。2. 先说清楚什么是“信号层”2.1 把异常和任务状态当成一条信号链路“信号层”这个词听起来有点玄乎其实用大白话讲就是程序在“事情已经没法正常完成”时用来传递“为什么停下来”的那一套机制。在 Python 里面它主要由三类东西组成异常类型、任务状态、进程信号。这三类东西背后都有明确的语义如果设计得当你在任何一个环节拿到一个状态都能反推出这个 Agent 工具调用到底经历了什么。我用一个生活化的类比来解释。你去餐厅点了个外卖最后没吃到饭原因无非两种外卖超时没送到或者你自己取消了订单。虽然结果都是“没吃到”但处理逻辑完全不一样。超时平台会给你赔券、商家会重新做取消订单直接终止钱原路退回。如果你在系统里把这两种状态都记成“未送达”那之后的赔付、退款、商家结算就全乱了。信号层要做的就是给“未送达”这个结果打上精确的标签。放到 Agent 工具调用的语境里这个标签体系通常需要覆盖三个层面。最外层是 Python 的 BaseException 体系比如 asyncio.TimeoutError、asyncio.CancelledError这属于异常信号第二层是任务对象的状态比如 concurrent.futures.Future 的 CANCELLED、CANCELLED_AND_NOTIFIED、RUNNING、FINISHED这属于任务生命周期信号第三层是操作系统级的进程信号比如用户按 CtrlC 触发的 SIGINT或者 kill 命令发来的 SIGTERM这属于外部干预信号。一个设计良好的 Agent 工具调用封装必须把这些信号统一映射到自己的领域模型里而不是直接原样抛给上层业务代码。2.2 超时和用户取消在信号层的本质差异超时和用户取消在信号层面最核心的差异在于超时是一个“单向被动”的事件取消是一个“双向交互”的事件。超时的本质是调用方等不起了。你设置了一个 deadline到了时间还没拿到结果于是主动放弃等待。这个时候底层那个真正在工作的任务——比如一个 HTTP 请求、一个数据库查询——其实可能还在跑只是你的代码不再等它了。这一点非常重要。asyncio.wait_for 超时之后内部任务会被取消但如果你用的是线程池或外部子进程任务不一定真的停了。这也是很多人踩坑的地方你以为超时之后资源就释放了实际上底层线程还在那里执行。用户取消的本质是任务的上游叫停了。用户点击“停止”前端发来一个 cancel 请求这时候发生的是“协作式取消”代码要在某个安全的暂停点主动响应取消指令。在 Python 的 asyncio 生态里协作式取消由 CancelledError 来传达任务收到取消请求后会在下一个 await 点抛出 CancelledError让你有机会清理资源、记录状态然后再把取消信号继续向上传播。还有一个容易被忽略的差异是取消的“传播路径”。超时通常只终止“当前这一层”的等待它和任务树里的其他兄弟节点没关系而用户取消往往要沿着任务树级联传播父任务取消了所有子任务、孙任务都要跟着取消。如果只是把超时和取消混在一个 except Exception 里处理这把级联关系也会被处理得稀碎。后面我将要展示的方案核心就是让“超时”和“取消”走两条完全独立的代码路径互不干扰。3. 一套可落地的三层信号区分方案3.1 第一层用“异常类型”给超时下定义在 Python asyncio 生态里超时的标准做法是用 asyncio.wait_for 或者 Python 3.11 引入的 asyncio.timeout() 上下文管理器。超时到达时协程内部会先取消当前任务然后抛出 TimeoutError。注意这里有个容易混淆的点asyncio.wait_for 内部取消任务时任务本身会经历一次 CancelledError 的抛出和吞没但最终交给调用方的是 TimeoutError。所以第一层的核心原则是所有“等待超时”的异常统一收敛为 TimeoutError绝不允许内部的 CancelledError 直接冒泡到业务层。下面是一段封装示例我在实际项目里基本就是按这个骨架写的import asyncio import logging class ToolInvokeError(Exception): 工具调用通用异常基类 class ToolTimeoutError(ToolInvokeError): 工具调用超时 async def call_tool_with_timeout(tool, *args, timeout: float 10.0): try: async with asyncio.timeout(timeout): return await tool.invoke(*args) except TimeoutError as e: # 这里一定是真的超时 logging.warning(ftool {tool.name} invoke timeout after {timeout}s) raise ToolTimeoutError(f{tool.name} timeout) from e except asyncio.CancelledError: # 这个分支要非常谨慎外层取消信号不该被 TimeoutError 包装 # 直接向上传播让上层决定是转为取消还是重新调度 logging.info(ftool {tool.name} cancelled by outer task) raise这里有一个关键决策except asyncio.CancelledError 分支不能和 TimeoutError 合并处理更不能把 CancelledError 改写为普通 Exception。一旦你把它包成别的异常就要额外引入一层机制来判断“这个普通异常到底是不是取消”反而弄巧成拙。在非异步场景比如你想用 ThreadPoolExecutor 执行一个会阻塞的 SDK 调用超时的捕获逻辑变成这样import concurrent.futures import time def call_tool_in_thread(tool, *args, timeout: float 5.0): with concurrent.futures.ThreadPoolExecutor(max_workers2) as pool: future pool.submit(tool.invoke, *args) try: return future.result(timeouttimeout) except concurrent.futures.TimeoutError: # 注意future 还在跑这里需要额外决定是否真正取消线程 # Python 线程无法强行 kill只能在业务代码里埋取消点 logging.warning(ftool {tool.name} thread timed out after {timeout}s) raise ToolTimeoutError(f{tool.name} timeout)这段代码里有个值得注意的点future.result(timeouttimeout) 抛出 concurrent.futures.TimeoutError 之后线程池里的任务其实还在继续运行。所以你只是在“信号的接收层”判断了超时并没有真正终止执行。这也是为什么很多 Agent 框架宁可把工具调用全部放到异步协程里也不愿意用线程池——协程的超时是协作式取消起码给了业务代码一个清理的机会。3.2 第二层用“取消令牌”给用户取消下定义用户取消和超时不同它不是“等不起”而是“不想等了”。如果只有一个简单的一层调用你可以只依赖 CancelledError 来传递取消。但 Agent 工具调用往往会有多级封装协调器、工具执行器、HTTP 传输层每一层都可能捕获异常、做日志、做埋点。如果取消信号只表现为一个单纯的 CancelledError上层在捕获它时很难立刻知道“这是用户主动取消”还是“某个内部环节的取消”。我建议引入一个显式的取消令牌CancellationToken让用户取消这个语义不依赖异常类型来传达。令牌的核心是一个状态标记加一组回调谁持有令牌谁就可以主动发起取消并且这个取消意图可以跨协程、跨线程传递。下面是一个简易实现import asyncio import threading from enum import Enum class CancelState(Enum): NORMAL normal CANCELLED_BY_USER cancelled_by_user CANCELLED_BY_TIMEOUT cancelled_by_timeout class CancellationToken: def __init__(self): self._state CancelState.NORMAL self._lock threading.Lock() self._waiters [] def cancel_by_user(self): 用户主动取消 with self._lock: if self._state is not CancelState.NORMAL: return self._state CancelState.CANCELLED_BY_USER waiters list(self._waiters) self._waiters.clear() for waiter in waiters: waiter.set() def cancel_by_timeout(self): 超时取消和用户取消区分开 with self._lock: if self._state is not CancelState.NORMAL: return self._state CancelState.CANCELLED_BY_TIMEOUT waiters list(self._waiters) self._waiters.clear() for waiter in waiters: waiter.set() property def is_cancelled(self): return self._state is not CancelState.NORMAL property def state(self): return self._state async def wait_for_cancel(self): loop asyncio.get_running_loop() event asyncio.Event() self._waiters.append(event) try: await event.wait() finally: self._waiters.remove(event)这个令牌的妙处在于它可以同时挂在多个工具调用上。当用户在 Agent 界面点击“停止”协调器只需要调用一次 token.cancel_by_user()所有监听这个令牌的工具调用都会醒来并发现“这是用户取消”而不是去猜测异常类型。这样在日志系统里你可以非常明确地记录 cancel_sourceuser之后的项目复盘、成本计量、重试策略都有了可靠依据。有了取消令牌之后工具调用的主流程改成这样async def call_tool_with_cancellation(tool, token: CancellationToken, timeout: float): # 组合使用外层超时内层监听取消令牌 try: async with asyncio.timeout(timeout): cancel_task asyncio.create_task(token.wait_for_cancel()) invoke_task asyncio.create_task(tool.invoke()) done, pending await asyncio.wait( {cancel_task, invoke_task}, return_whenasyncio.FIRST_COMPLETED, ) if cancel_task in done and token.is_cancelled: # 用户取消优先占位 for p in pending: p.cancel() raise UserCancelError(cancelled by user) return invoke_task.result() except TimeoutError: token.cancel_by_timeout() raise ToolTimeoutError(timeout)这里的关键是 asyncio.wait(FIRST_COMPLETED) 让你同时监听两个信号取消令牌和工具调用本身。哪个先到就知道这轮调用是怎么终结的。用户取消和超时走了完全不一样的分支不会出现“日志里写着超时其实是用户点了停止”的混乱。3.3 第三层用进程信号兜住最外层的被动取消前面两层解决的是程序内部的“超时”和“取消”。但 Agent 服务运行在真实环境里还有一个来源不可忽略用户直接按下 CtrlC或者运维通过 kill 命令终止进程。这一层如果拦截不好整个 Agent 会“死得不明不白”工具调用挂在半空中资源没人清理。Python 的 signal 模块可以处理 SIGINT 和 SIGTERM。单独用 signal.signal 注册处理器很容易踩坑因为处理器只在主线程执行而且会被 asyncio 事件循环打断。在 asyncio 应用中正确做法是使用 loop.add_signal_handler把系统信号转成回调和事件再转发给取消令牌import asyncio import signal async def main(): token CancellationToken() loop asyncio.get_running_loop() def on_sigint(): # 用户按下了 CtrlC等价于用户取消 print(receive SIGINT, treating as user cancel) token.cancel_by_user() def on_sigterm(): # 进程被 kill可以记为外部终止也可以归为用户取消 print(receive SIGTERM, treating as user cancel) token.cancel_by_user() loop.add_signal_handler(signal.SIGINT, on_sigint) loop.add_signal_handler(signal.SIGTERM, on_sigterm) # 后面正常启动 Agent 任务 await run_agent(token)这样一来从最外层的 CtrlC到中间的 Agent 协调器取消请求再到工具调用的 CancelledError整条链路都是同一个语义用户取消。你不再需要事后靠日志里那点蛛丝马迹去猜“到底是网线断了还是用户点了停止”。需要额外提醒一句loop.add_signal_handler 在 Windows 上并不完全可靠尤其是 SIGTERM。如果你要跨平台支持进程信号建议在 Windows 下退回到 signal.signal并把回调封装成线程安全的投递方式。我自己的经验是在 Windows 上尽量以“事件循环收到退出消息”作为进程终止信号而不是依赖 SIGTERM。4. 完整实操跑通一次“被取消”的工具调用4.1 搭建最小实验环境理论说再多不如跑一遍。我用一个极简的模拟场景来做完整演示有一个假想的“慢速工具”比如一个 sleep 10 秒才返回结果的函数Agent 调用它同时在 3 秒的时候模拟用户点击“停止”。我们观察信号层的日志输出看能不能在日志里准确分辨超时和用户取消。环境准备就一个 Python 3.11不需要任何第三方库标准库里的 asyncio、concurrent.futures、signal、dataclasses 足够。我建议你新建一个文件 demo_agent_cancel.py把下面的代码按段贴进去边跑边看输出。4.2 完整代码与调用时序先定义领域异常和一个 PretendToolimport asyncio import logging import time from dataclasses import dataclass, field from typing import Any logging.basicConfig(levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s) class ToolError(Exception): pass class ToolTimeoutError(ToolError): pass class UserCancelError(ToolError): pass class PretendTool: 假装是一个很慢的外部工具 def __init__(self, name: str, delay: float): self.name name self.delay delay async def invoke(self, payload: Any): logging.info(ftool {self.name} started, delay{self.delay}s) await asyncio.sleep(self.delay) return fresult from {self.name}接着是实现取消令牌的状态存储。为了便于观察我用一个事件日志列表记录每一次信号流转dataclass class CallOutcome: tool_name: str signal: str # timeout | user_cancel | success | tool_error detail: str events: list field(default_factorylist) elapsed: float 0.0然后是核心的调用封装。这一步我要把前面讲的两层方案合并到一个函数里因为它反映的是真实场景下的组合使用async def call_tool(tool: PretendTool, timeout: float, cancel_event: asyncio.Event, outcome: CallOutcome): start time.monotonic() outcome.tool_name tool.name try: async with asyncio.timeout(timeout): invoke_task asyncio.create_task(tool.invoke(None)) cancel_task asyncio.create_task(cancel_event.wait()) done, pending await asyncio.wait( {invoke_task, cancel_task}, return_whenasyncio.FIRST_COMPLETED, ) if cancel_task in done and cancel_event.is_set(): # 用户取消立刻取消没有完成的 invoke_task for p in pending: p.cancel() outcome.signal user_cancel outcome.detail cancel_event triggered by user outcome.events.append(signalcancel_event fired first) raise UserCancelError(user cancelled) result invoke_task.result() outcome.signal success outcome.detail str(result) outcome.events.append(signalinvoke_task completed) return result except TimeoutError: # 超时需要在超时分支里把可能存在的 cancel_task 取消掉 for p in asyncio.all_tasks(): if p is not asyncio.current_task() and p.get_name() invoke_task: p.cancel() outcome.signal timeout outcome.events.append(signalasyncio.timeout fired first) raise ToolTimeoutError(f{tool.name} timeout) except asyncio.CancelledError: # 外层用户取消信号比如进程信号或上层协调器取消 outcome.signal user_cancel outcome.events.append(signalouter CancelledError propagated) raise finally: outcome.elapsed time.monotonic() - start上面代码里有一个需要特别说明的点超时分支里清理 pending 任务的方式我用了 asyncio.all_tasks() 去筛这个写法在生产代码里不推荐因为很容易误杀其他无关任务。更稳的做法是保存 invoke_task 和 cancel_task 的引用在超时分支里直接对引用执行 cancel。下面这段是更干净的写法async def call_tool_v2(tool: PretendTool, timeout: float, cancel_event: asyncio.Event, outcome: CallOutcome): start time.monotonic() outcome.tool_name tool.name invoke_task asyncio.create_task(tool.invoke(None)) cancel_task asyncio.create_task(cancel_event.wait()) try: async with asyncio.timeout(timeout): done, pending await asyncio.wait( {invoke_task, cancel_task}, return_whenasyncio.FIRST_COMPLETED, ) if cancel_task in done and cancel_event.is_set(): for p in pending: p.cancel() outcome.signal user_cancel outcome.events.append(signalcancel_event fired first) raise UserCancelError(user cancelled) result invoke_task.result() outcome.signal success outcome.events.append(signalinvoke_task completed) return result except TimeoutError: cancel_task.cancel() invoke_task.cancel() outcome.signal timeout outcome.events.append(signalasyncio.timeout fired first) raise ToolTimeoutError(f{tool.name} timeout) finally: outcome.elapsed time.monotonic() - start两种写法我都在项目里见过如果你要复用我的代码用 v2 就好。模拟用户取消的主程序我用一个 0.5 秒后 set 的 asyncio.Event 来模拟用户点击“停止”async def main(): tool PretendTool(slow-api, delay10.0) outcome CallOutcome(tool_nametool.name, signal, detail) cancel_event asyncio.Event() # 延迟 1.5 秒模拟用户取消 async def simulate_user_cancel(): await asyncio.sleep(1.5) logging.info(simulate user cancel now) cancel_event.set() simulator asyncio.create_task(simulate_user_cancel()) try: await call_tool_v2(tool, timeout3.0, cancel_eventcancel_event, outcomeoutcome) except UserCancelError: logging.info(catched UserCancelError, no retry) except ToolTimeoutError: logging.info(catched ToolTimeoutError, should retry) await simulator print(\n--- outcome detail ---) print(ftool: {outcome.tool_name}) print(fsignal: {outcome.signal}) print(fdetail: {outcome.detail}) print(fevents: {outcome.events}) print(felapsed: {outcome.elapsed:.2f}s) if __name__ __main__: asyncio.run(main())运行之后会看到这样的关键日志2025-01-10 12:00:01 [INFO] tool slow-api started, delay10.0s 2025-01-10 12:00:02 [INFO] simulate user cancel now --- outcome detail --- tool: slow-api signal: user_cancel detail: cancel_event triggered by user events: [signalcancel_event fired first] elapsed: 1.50s注意 outcome.signal 是 user_cancel不是 timeout尽管调用方 timeout 设置的是 3 秒。因为取消事件在 1.5 秒先到了。这正是信号层分离的直观体现。如果你把 simulate_user_cancel 里的 sleep 时间改成 5 秒那么 3 秒超时会先触发最终 signal 会变成 timeout。同一个工具、同一个封装不需要改任何逻辑输出的信号就能准确反映“到底是超时还是用户取消”。4.3 输出结果怎么解读实际跑通之后你会发现最大的收益在日志和后续动作。当 outcome.signal 是 user_cancel 时Agent 协调器的处理策略是直接终止当前任务不重试不补偿顶多记一条“用户主动取消”。当 outcome.signal 是 timeout 时协调器会根据工具属性决定重试次数、是否降级或者把这次调用标记为失败。我甚至会把 outcome 对象序列化成 JSON写进结构化日志之后做计量分析时直接按 signal 字段分桶统计一目了然。这套封装的价值还体现在可测试性上。你可以不依赖真实网络单纯用 event.set() 来模拟用户取消用 sleep 来控制超时时长把 Agent 的“超时/取消”行为作为一等公民进行单测。如果你只是想凭异常类型猜状态单测永远覆盖不了“用户取消先到”这条路径。5. 踩坑实录与排查速查表5.1 wait_for 会把内部取消伪装成超时的坑这是我在代码审查里见过最多的隐蔽问题。假设你在一个已有代码里用了 asyncio.wait_for(do_something(), timeout10)然后 do_something 内部因为某些原因抛出了 CancelledError——可能是它的上层任务被 cancel也可能它自己内部调用了 cancel。wait_for 的默认行为是超时到达时它先取消内部任务再抛 TimeoutError但如果内部任务已经因为外部取消而处于取消中wait_for 可能直接把内部任务的 CancelledError 转成 TimeoutError 抛给调用方。结果就是你以为日志里那个 timeout 就是真的超时其实可能是“任务被外部取消”的误报。要规避这个坑最简单的手段是不要在同一个函数里既做超时控制又做取消控制而是像我前面那样把取消事件和调用任务交给 asyncio.wait 同时监听谁先完成谁说话。还有一个很现实的场景Agent 框架通常会为每个工具调用设置默认超时比如 30 秒。用户点了停止后框架会一层层 cancel 任务。如果工具调用内部用了 wait_for用户取消产生的 CancelledError 在到达业务层时已经被包装成了 TimeoutError。上层协调器一看是 timeout就触发重试于是出现了“用户取消后还在被重试”的诡异行为。破解方法就是先让取消信号在 async with asyncio.timeout: 内自由传播不要着急捕获。5.2 线程池 Future.cancel() 只对未启动任务有效如果你没法把所有工具调用都改造成协程线程池方案的坑也需要提前知道。concurrent.futures.Future 的 cancel() 方法有一个硬性限制只有任务尚未开始运行时cancel() 才能返回 True 并真正取消。一旦任务已经在线程里跑起来了cancel() 返回 False底层函数照跑。实际上有很多 Agent 工具是同步阻塞型 SDK比如某些云厂商的 Python SDK你只能在业务代码里埋“取消检查点”。我通常的做法是给这类工具传入一个 dict 作为取消状态容器工具内部在每个循环迭代里检查这个 dict 里的 is_cancelled 字段。虽然做不到立即杀死线程但至少能在下一个检查点优雅退出。下面是一个很简单的骨架def long_running_tool_sync(config, cancel_slot: dict): for i in range(1000): if cancel_slot.get(is_cancelled): return cancelled time.sleep(0.01) return done def call_in_thread_checked(tool_fn, cancel_slot: dict, timeout: float): with ThreadPoolExecutor(max_workers1) as pool: future pool.submit(tool_fn, cancel_slot) try: return future.result(timeouttimeout) except concurrent.futures.TimeoutError: cancel_slot[is_cancelled] True raise ToolTimeoutError(timeout)在这个设计里超时发生时会主动往取消槽里写标记线程任务在下一个迭代点自行退出。注意它并不能保证立刻释放资源但至少不会出现“超时之后线程还在执行危险操作”的情况。对于内部是长连接的工具你还得在 finally 里做连接关闭。5.3 常见信号混乱场景排查速查表下面是我在 Agent 工具调用这一路踩出来的高频问题速查表遇到类似日志可以直接对照现象可能原因排查方向日志显示 timeout但用户明明点了停止wait_for 把 CancelledError 包装成 TimeoutError改用 asyncio.wait 双监听取消信号走独立路径用户取消之后工具还在重试业务层把取消当成了超时触发了重试策略检查异常类型分支确认 UserCancelError 不在重试范围内进程 CtrlC 后被 asyncio 吞掉任务挂死signal.signal 处理函数与事件循环不兼容改用 loop.add_signal_handler线程池超时后底层任务依然在写库future.result(timeout) 只是不再等待不是终止线程引入取消槽在工具内部做协作式退出多级任务取消时子任务只取消了一层CancelledError 被某层 except Exception 捕获吞掉全局搜索 except Exception确认 CancelledError 单独处理日志里出现“cancelled_by_timeout”这种模糊状态取消令牌初始化时未区分状态来源使用枚举区分 CANCELLED_BY_USER 和 CANCELLED_BY_TIMEOUT排查的时候我个人的习惯是先看最外层事件来源进程信号、上层协调器、还是本层计时器。来源定了再往底层看异常类型。如果日志系统里能带上类似 outcome.events 这样的信号序列定位问题的速度会快一个数量级。最后再分享一个小技巧。如果你用 Python 3.11 以上版本可以在自定义的取消令牌里顺便记录 cancel 的调用栈快照这样以后线上出现“用户取消”疑惑时你能直接知道是哪个模块、哪一行发起的取消。这个信息在排查多 Agent 编排平台的偶发挂起问题时价值巨大。我在一次复盘里就是靠取消栈快照发现了某个后台任务误调用了 token.cancel_by_user()源头是一个 API 路径写错导致的误触。信号层设计的投入往往会在这种最不起眼的时刻回本。
知识建站
内容驱动
阅读友好版式
SEO 精细化
线索转化