Agent 工具调用的超时熔断单次卡住不能阻塞整个会话一、Agent 调用搜索 APIAPI 挂了 30 秒没响应整个会话就卡死在这了Agent 工具调用的可靠性不是工具能正常工作——那是最理想情况——而是在工具不工作了的情况下 Agent 如何处理。最常见的问题Agent 调用了某个外部 API搜索、计算、数据库查询这个 API 因为网络抖动或后端故障一直没返回。Agent 的 naive 实现是同步等待返回于是整个会话阻塞 30 秒不动——用户看着空白页面以为系统崩了。实际需要的是一个多层防护体系超时超过 N 秒就不等了、熔断连续失败 N 次就暂时跳过这个工具、降级用缓存结果或跳过这个工具步骤继续推进。这三个机制不是加一个就行——它们的组合才是生产级可靠性。二、底层机制与原理剖析三层防护机制超时Timeout最基础的保护。每个工具调用都必须设定一个最大等待时间。时间的设定取决于工具的 SLA——搜索 API 如果 P99 是 2 秒超时设 5 秒足够。数据库查询如果超 10 秒说明索引有问题设 15 秒留了点 buffer。熔断Circuit Breaker当工具连续失败达到阈值时暂时跳过这个工具——不让它继续拖慢整个会话。熔断器有三种状态关闭正常调用、打开快速失败、不实际调用、半开允许一次探测调用来判断是否恢复。关键是冷却时间——熔断后不能永远跳过这个工具过一段时间如 30 秒要尝试探测。降级Fallback工具不可用时的 Plan B。搜索 API 熔断 → 使用上次缓存的搜索结果可能有 5 分钟延迟但比无结果好。计算 API 超时 → 让 Agent 基于已有信息给一个精度较低的估计值。降级策略在 Agent 设计阶段就要想好不能等故障发生了再临时想。三、生产级代码实现 Agent 工具调用超时熔断器 三层防护超时 → 熔断 → 降级 import asyncio import time import logging import functools from typing import Dict, Any, Callable, Optional from dataclasses import dataclass from enum import Enum logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class CircuitState(Enum): CLOSED closed # 正常调用 OPEN open # 熔断——直接返回降级结果 HALF_OPEN half_open # 半开——允许探测请求 dataclass class CircuitBreakerConfig: 熔断器配置 failure_threshold: int 5 # 连续失败 N 次 → 熔断 success_threshold: int 2 # 半开状态连续成功 N 次 → 恢复 timeout_seconds: float 10.0 # 工具调用超时时间 cooldown_seconds: float 30.0 # 熔断冷却时间 half_open_max_requests: int 1 # 半开状态允许的探测请求数 class ToolCircuitBreaker: 工具调用熔断器 每个工具实例一个熔断器——不同工具的故障独立 def __init__(self, tool_name: str, config: Optional[CircuitBreakerConfig] None): self.tool_name tool_name self.config config or CircuitBreakerConfig() self.state CircuitState.CLOSED self.failure_count 0 self.success_count 0 self.last_failure_time: float 0 self.last_state_change: float time.time() # 降级缓存工具成功调用时缓存结果 self.fallback_cache: Optional[Dict[str, Any]] None self.cache_timestamp: float 0 # 统计 self.total_calls 0 self.total_failures 0 self.total_timeouts 0 async def call(self, tool_fn: Callable, *args, **kwargs) - Dict[str, Any]: 调用工具——带超时、熔断、降级保护 返回格式{success: bool, data: ..., error: ..., from_cache: bool} self.total_calls 1 # 状态检查 if self.state CircuitState.OPEN: if time.time() - self.last_state_change self.config.cooldown_seconds: # 冷却时间到 → 半开 self._transition_to(CircuitState.HALF_OPEN) else: # 仍在熔断 → 直接返回降级 return self._get_fallback() if (self.state CircuitState.HALF_OPEN and self.success_count self.config.half_open_max_requests): # 半开状态下已达到探测请求上限——拒绝额外请求 return self._get_fallback() # 执行调用带超时 try: result await asyncio.wait_for( self._execute_safely(tool_fn, *args, **kwargs), timeoutself.config.timeout_seconds, ) # 调用成功 self._on_success(result) return {success: True, data: result, from_cache: False} except asyncio.TimeoutError: self.total_timeouts 1 self._on_failure() logger.warning(Tool %s: timeout after %.1fs, self.tool_name, self.config.timeout_seconds) return self._get_fallback(errortimeout) except Exception as e: self._on_failure() logger.error(Tool %s: execution failed: %s, self.tool_name, e) return self._get_fallback(errorstr(e)) async def _execute_safely(self, fn: Callable, *args, **kwargs): 安全执行工具函数 if asyncio.iscoroutinefunction(fn): return await fn(*args, **kwargs) else: # 同步函数——在 executor 中执行避免阻塞事件循环 loop asyncio.get_event_loop() return await loop.run_in_executor( None, functools.partial(fn, *args, **kwargs) ) def _on_success(self, result: Any): 调用成功时的处理 # 更新降级缓存 self.fallback_cache {data: result, timestamp: time.time()} self.cache_timestamp time.time() # 恢复状态 if self.state CircuitState.HALF_OPEN: self.success_count 1 if self.success_count self.config.success_threshold: self._transition_to(CircuitState.CLOSED) logger.info(Tool %s: circuit closed (recovered), self.tool_name) elif self.state CircuitState.CLOSED: # 重置失败计数 self.failure_count 0 self.success_count 1 def _on_failure(self): 调用失败时的处理 self.total_failures 1 self.failure_count 1 self.last_failure_time time.time() self.success_count 0 if (self.state CircuitState.CLOSED and self.failure_count self.config.failure_threshold): self._transition_to(CircuitState.OPEN) logger.warning(Tool %s: circuit OPEN after %d failures, self.tool_name, self.failure_count) elif (self.state CircuitState.HALF_OPEN): # 探测请求也失败了 → 回到熔断状态 self._transition_to(CircuitState.OPEN) logger.warning(Tool %s: probe failed, circuit OPEN again, self.tool_name) def _get_fallback(self, error: Optional[str] None) - Dict[str, Any]: 返回降级结果 降级策略 1. 有缓存 → 返回缓存标记 from_cacheTrue 2. 无缓存 → 返回空结果 错误信息 if self.fallback_cache: # 检查缓存是否过期超过 5 分钟不用 if time.time() - self.cache_timestamp 300: return { success: True, data: self.fallback_cache[data], from_cache: True, cache_age_seconds: int(time.time() - self.cache_timestamp), } return { success: False, data: None, error: error or fTool {self.tool_name} is unavailable, from_cache: False, } def _transition_to(self, new_state: CircuitState): 状态转换 old_state self.state self.state new_state self.last_state_change time.time() if new_state CircuitState.CLOSED: self.failure_count 0 self.success_count 0 logger.info(Tool %s: %s → %s, self.tool_name, old_state.value, new_state.value) def get_stats(self) - dict: 获取统计信息 return { tool: self.tool_name, state: self.state.value, total_calls: self.total_calls, total_failures: self.total_failures, total_timeouts: self.total_timeouts, failure_rate: ( self.total_failures / self.total_calls if self.total_calls 0 else 0 ), has_fallback_cache: self.fallback_cache is not None, } # --------------------------------------------------------------------------- # Agent 侧使用示例 # --------------------------------------------------------------------------- class AgentToolManager: Agent 工具管理器——管理所有工具的熔断器 def __init__(self): self.breakers: Dict[str, ToolCircuitBreaker] {} def get_breaker(self, tool_name: str) - ToolCircuitBreaker: 获取或创建指定工具的熔断器 if tool_name not in self.breakers: self.breakers[tool_name] ToolCircuitBreaker( tool_name, CircuitBreakerConfig( failure_threshold3, # 连续 3 次失败 → 熔断 timeout_seconds8.0, # 8 秒超时 cooldown_seconds30.0, # 30 秒冷却 ), ) return self.breakers[tool_name] async def call_tool(self, tool_name: str, tool_fn: Callable, *args, **kwargs) - Dict[str, Any]: Agent 工具调用的统一入口 breaker self.get_breaker(tool_name) return await breaker.call(tool_fn, *args, **kwargs) def get_all_stats(self) - list: 获取所有工具的统计信息 return [b.get_stats() for b in self.breakers.values()] # --------------------------------------------------------------------------- # 模拟使用 # --------------------------------------------------------------------------- async def main(): manager AgentToolManager() # 模拟工具函数 async def search_api(query: str) - dict: # 模拟前 3 次成功之后失败 await asyncio.sleep(0.5) return {results: [fSearch results for: {query}]} async def failing_api(query: str) - dict: # 模拟总是失败 raise ConnectionError(API unavailable) # 正常调用 for i in range(3): result await manager.call_tool(search, search_api, fquery_{i}) print(fSearch #{i}: success{result[success]}, cached{result.get(from_cache, False)}) # 故障调用——触发熔断 for i in range(5): result await manager.call_tool(failing, failing_api, test) print(fFailing #{i}: success{result[success]}, error{result.get(error, N/A)}) # 查看统计 print(\n 工具统计 ) for stat in manager.get_all_stats(): print(f{stat[tool]}: state{stat[state]}, ffailure_rate{stat[failure_rate]:.1%}) if __name__ __main__: asyncio.run(main())四、边界分析与架构权衡超时时间的设定太短1-2 秒→ 正常慢请求如大数据集搜索被误判为超时降级太激进太长30 秒→ 用户等待太久失去了超时保护的意义建议基于工具的 P99 延迟来设。如果搜索 API 的 P99 是 3 秒超时设为 6 秒2x buffer熔断恢复的探测定时冷却时间太短5 秒→ 工具还在故障中就被探测反复开闭flapping冷却时间太长5 分钟→ 工具已经恢复了但 Agent 还在用降级结果推荐 30-60 秒作为起步值观察实际故障恢复的平均时间后调整降级的精度损失用缓存结果降级 → 可能用 5 分钟前的旧数据取决于缓存过期时间完全跳过这一步 → Agent 继续推理但缺少关键信息最终回答可能是我不确定Agent 需要在回答中告知用户由于 XX 工具暂时不可用以下分析基于上次缓存数据五、总结Agent 工具调用的可靠性保护链超时按工具的 P99 设 2x buffer→ 熔断连续 N 次失败后跳过该工具→ 降级用缓存结果或跳过步骤。三者不是简单的加一层就安全了——是协同运作的。超时是第一道防线保护 Agent 不被单次慢请求卡死熔断是第二道防止重复失败的工具浪费整个会话时间降级是最后的兜底用户永远能得到一个回复即使不完美。关键设置超时基于工具 P99 来定熔断阈值和冷却时间通过观察实际故障模式调整降级策略在设计阶段就想好。