Agent 工具调用的超时熔断：单次卡住不能阻塞整个会话
一、Agent 调用搜索 API，API 挂了 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(level=logging.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), timeout=self.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(error="timeout") except Exception as e: self._on_failure() logger.error("Tool %s: execution failed: %s", self.tool_name, e) return self._get_fallback(error=str(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_cache=True） 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 f"Tool {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_threshold=3, # 连续 3 次失败 → 熔断 timeout_seconds=8.0, # 8 秒超时 cooldown_seconds=30.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": [f"Search 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, f"query_{i}") print(f"Search #{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(f"Failing #{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']}, " f"failure_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 来定，熔断阈值和冷却时间通过观察实际故障模式调整，降级策略在设计阶段就想好。
