Agent 级联失败防护一个工具挂了不该拖垮整个会话一、用户问天气Agent 调天气 API 失败然后整个会话就废了Agent 的典型调用链路是用户消息 → LLM 推理 → Function Calling → 执行工具 → 返回结果 → LLM 推理 → 下一个工具…如果一个工具调用失败超时、返回 500、返回格式错误没有防护的 Agent 就会进入两种失败模式模式 A无限重试。LLM 看到工具返回了 error它重新构造调用参数再试一次。重试又失败再试。三轮下来 Token 烧了一堆用户等着Agent 在傻循环。模式 B错误传播。工具 A 失败后返回的错误信息被直接喂给 LLM 作为下一轮推理的上下文。LLM 拿到错误信息后开始分析为什么失败——它开始编造原因、建议排查步骤、甚至自己构造一个假的工具返回结果。这比模式 A 更危险因为用户可能被错误引导。级联失败的本质是Agent 的工具调用链没有隔离墙。工具 A 的失败可以通过上下文污染传播到工具 B 的决策。时间久了整个会话的推理质量就被失败的上下文带偏了。二、底层机制与原理剖析级联失败防护需要在 Agent 的 Function Calling 回路中插入三个拦截点flowchart TD A[LLM 决定调用工具] -- B{调用前检查} B --|最近3次调用全失败?| C[终止工具链路] B --|通过| D[执行工具调用] D -- E{工具返回结果} E --|成功| F[标准化结果格式] E --|超时| G[注入降级响应] E --|权限错误| H[触发权限异常处理] E --|未知错误| I[注入兜底错误消息] F -- J[LLM 下一轮推理] G -- J H -- K[中断会话不注入错误到上下文] I -- J J -- L{错误检测} L --|连续3轮有错误?| K L --|正常| M[继续推理] C -- N[返回用户功能暂时不可用] K -- N三个拦截点对应三种防护策略隔离墙工具返回的错误信息在喂给 LLM 之前先经过一个净化层。净化层做的事情是脱敏去掉内部堆栈、IP 地址、简化把 500 字错误堆栈压缩成一句话服务暂时不可用、分类区分可重试错误和不可重试错误。熔断器和微服务里的熔断器逻辑一致。统计最近 N 次工具调用的成功率低于阈值时直接跳过工具调用告诉 LLM此工具当前不可用请换一种方式回答用户。关键是不要让 LLM 看到工具不可用这个原始错误。降级响应为每个工具预定义降级输出——一个标准化的占位响应。比如天气工具不可用时返回{status: unavailable, suggestion: 请稍后重试或查询离线缓存}。这个占位响应是可控的、干净的不携带任何可能误导 LLM 的错误信息。三、生产级代码实现 Agent 工具调用级联失败防护 三层防护隔离墙 → 熔断器 → 降级响应 设计理念工具失败不应展开为对话内容 from dataclasses import dataclass, field from typing import Dict, List, Optional, Any, Callable from enum import Enum import time import threading import json class ErrorCategory(Enum): 工具错误分类 RETRYABLE retryable # 可重试超时、限流 NON_RETRYABLE non_retry # 不可重试权限、参数错误 SERVICE_DOWN service_down # 服务宕机连接拒绝、DNS 解析失败 dataclass class ToolErrorPolicy: 单个工具的容错策略 tool_name: str # 降级响应工具不可用时返回这个给 LLM fallback_response: Dict[str, Any] # 同一会话中该工具的最大连续失败次数 max_consecutive_failures: int 3 # 失败后的冷却时间秒 cooldown_seconds: float 30.0 # 错误分类映射 error_category_map: Dict[str, ErrorCategory] field(default_factorydict) dataclass class CircuitBreaker: 工具级熔断器 设计决策每个工具独立的熔断器 避免一个工具的故障影响其他工具。 tool_name: str failure_threshold: int 5 # 连续失败几次后熔断 recovery_timeout: float 60.0 # 熔断后多久尝试恢复 state: str CLOSED # CLOSED / OPEN / HALF_OPEN failure_count: int 0 last_failure_time: float 0.0 last_state_change: float field(default_factorytime.time) _lock: threading.Lock field(default_factorythreading.Lock) def before_call(self) - bool: 调用前检查是否允许请求通过 with self._lock: if self.state CLOSED: return True if self.state OPEN: if time.time() - self.last_state_change self.recovery_timeout: self.state HALF_OPEN self.last_state_change time.time() return True return False # HALF_OPEN允许探测请求通过 return True def on_success(self): with self._lock: self.failure_count 0 if self.state HALF_OPEN: self.state CLOSED self.last_state_change time.time() def on_failure(self): with self._lock: self.failure_count 1 self.last_failure_time time.time() if self.failure_count self.failure_threshold: self.state OPEN self.last_state_change time.time() class ErrorSanitizer: 错误净化层 核心职责把工具返回的原始错误转换成一个干净的、不会误导 LLM的描述。 为什么需要这层LLM 看到堆栈信息和内部错误详情后 会把这些信息当作上下文的一部分在后续推理中产生不可预知的行为。 净化后的消息只传递发生了什么和接下来怎么做。 MAX_ERROR_LENGTH 200 # 错误描述最大字符数防止错误信息塞满上下文窗口 def sanitize(self, tool_name: str, error: Exception) - str: 将原始异常转化为安全输出 error_type type(error).__name__ error_msg str(error) # 分类并生成标准化消息 if isinstance(error, TimeoutError) or timeout in error_msg.lower(): category ErrorCategory.RETRYABLE sanitized f工具 [{tool_name}] 调用超时。 elif connection refused in error_msg.lower() or dns in error_msg.lower(): category ErrorCategory.SERVICE_DOWN sanitized f工具 [{tool_name}] 服务不可用。 elif permission in error_msg.lower() or unauthorized in error_msg.lower(): category ErrorCategory.NON_RETRYABLE sanitized f工具 [{tool_name}] 权限不足。 else: category ErrorCategory.NON_RETRYABLE # 截断错误消息去除换行和多余空白 short_msg .join(error_msg.split())[:self.MAX_ERROR_LENGTH] sanitized f工具 [{tool_name}] 调用失败{short_msg} return sanitized class CascadeFailureGuard: 级联失败防护主控类 def __init__(self): self.circuit_breakers: Dict[str, CircuitBreaker] {} self.tool_policies: Dict[str, ToolErrorPolicy] {} self.sanitizer ErrorSanitizer() # 全局失败计数器 self._global_failure_count 0 self._max_global_failures 10 self._lock threading.Lock() def register_tool(self, policy: ToolErrorPolicy): 注册工具及其容错策略 self.tool_policies[policy.tool_name] policy self.circuit_breakers[policy.tool_name] CircuitBreaker( tool_namepolicy.tool_name, failure_thresholdpolicy.max_consecutive_failures, recovery_timeoutpolicy.cooldown_seconds, ) def guard_call( self, tool_name: str, call_fn: Callable, *args, **kwargs ) - Dict[str, Any]: 受保护的工具调用 流程 1. 检查熔断器 - 如果熔断直接返回降级响应 2. 执行工具调用 - 捕获异常 3. 异常被净化 - 返回标准化的错误描述不泄露堆栈 breaker self.circuit_breakers.get(tool_name) policy self.tool_policies.get(tool_name) # 检查全局失败上限 if self._global_failure_count self._max_global_failures: return { success: False, error: 会话中累积过多失败建议重启会话, action: restart_session, } # 检查熔断器 if breaker and not breaker.before_call(): return { success: False, error: f工具 [{tool_name}] 暂时不可用熔断保护中, fallback: policy.fallback_response if policy else {}, action: use_fallback, } try: result call_fn(*args, **kwargs) # 成功重置对应熔断器 if breaker: breaker.on_success() return {success: True, data: result} except Exception as e: # 全局计数 with self._lock: self._global_failure_count 1 # 熔断器记录失败 if breaker: breaker.on_failure() # 错误净化核心不让原始错误进入 LLM 上下文 sanitized_msg self.sanitizer.sanitize(tool_name, e) # 判断是否返回降级响应 if policy: return { success: False, error: sanitized_msg, fallback: policy.fallback_response, action: use_fallback, } return { success: False, error: sanitized_msg, action: skip_tool, } def build_llm_context(self, tool_results: List[Dict]) - str: 为 LLM 构造下一轮的上下文消息 关键设计失败的工具调用不展开描述。 只有简洁地告诉 LLM此工具不可用继续。 防止 LLM 花 tokens 分析错误原因。 messages [] for result in tool_results: if result[success]: messages.append(f工具 [{result.get(tool, )}] 返回{json.dumps(result.get(data, {}), ensure_asciiFalse)}) else: action result.get(action, skip_tool) if action use_fallback: # 用降级数据替代错误LLM 看到的是正常的降级数据 fallback result.get(fallback, {}) messages.append(f工具返回了降级数据{json.dumps(fallback, ensure_asciiFalse)}) elif action restart_session: messages.append(系统提示建议结束当前会话并重新开始。) else: messages.append(f工具暂时不可用请继续。) return \n.join(messages) def reset_session(self): 重置会话级计数器 with self._lock: self._global_failure_count 0 for breaker in self.circuit_breakers.values(): breaker.state CLOSED breaker.failure_count 0 # 使用示例 # 定义工具降级响应模板 WEATHER_FALLBACK { status: unavailable, message: 天气数据暂不可用, suggestion: 请稍后重试, } gard CascadeFailureGuard() gard.register_tool(ToolErrorPolicy( tool_nameget_weather, fallback_responseWEATHER_FALLBACK, max_consecutive_failures3, cooldown_seconds30, )) gard.register_tool(ToolErrorPolicy( tool_namesearch_knowledge_base, fallback_response{matches: [], fallback: True}, max_consecutive_failures5, cooldown_seconds10, ))四、边界分析与架构权衡级联失败防护的缺点过于激进的错误净化可能丢失有用的错误信息。比如 API 返回了参数 {city} 拼写错误建议使用 {corrected}——这是一个有用的纠错信息。如果净化层粗暴地把所有错误消息都截断为调用失败LLM 就没法帮用户修正参数。熔断器的阈值设置是一个需要持续校准的参数。太保守如失败 1 次就熔断会导致抖动——网络波动一下就把工具封了 60 秒。太激进如失败 10 次才熔断会导致降级响应迟迟不触发。适用边界最适合调用外部 API 较多的 Agent——天气、地图、支付、数据库查询等。每个外部依赖都是不可靠的。也适合有多工具编排的 Agent工具数量越多独立熔断的价值越大。禁用场景不适合只有少量确定性工具调用的 Agent。如果工具失败的概率极低如本地文件操作不需要全套防护。也不适合需要 LLM 理解错误详情来做决策的场景如代码调试 Agent此时过度净化反而降低准确度。五、总结Agent 的工具调用链路需要和微服务一样做故障隔离。三级防护各司其职熔断器做快速失败、净化层做信息脱敏、降级响应做业务兜底。核心原则不让工具调用的失败信息污染 LLM 的上下文窗口。不要让一个工具的超时演变为整个会话的崩溃。用可控的错误信息替代不可控的异常堆栈。