在2026年的今天基于大语言模型LLM的Agent系统已经深度嵌入到企业级工作流、自动化运维、金融交易和智能客服等关键业务场景中。这些Agent不再是简单的“聊天机器人”而是能够调用数十种外部工具和API来完成复杂任务的自主实体——从查询数据库、发送邮件到控制工业设备、执行代码部署。然而一个残酷的现实是外部API永远不会100%可靠。根据Cloudflare 2026年Q1的互联网服务可用性报告主流第三方API的平均年度宕机时间累计达4.2小时月度故障率高达13.7%。当Agent所依赖的天气API、支付网关或知识库服务突然不可用时整个自动化链条可能瞬间崩溃。这引出了一个核心问题当外部API宕机时Agent如何优雅处理而不是粗暴地抛出异常或返回无意义的错误信息本文将深入探讨工具调用的异常重试与降级策略从理论基础到实战代码构建一个健壮的Agent容错体系。全文超过五千字包含完整可运行的Python实现覆盖指数退避重试、熔断器模式、缓存降级、语义回退和用户友好交互等全方位策略。目录第一章问题域分析——Agent工具调用的脆弱性1.1 Agent工具调用的典型架构1.2 故障类型枚举1.3 为什么不能简单重试第二章核心策略框架——三层防御体系2.1 第一层智能重试——不只是“再试一次”指数退避与抖动Exponential Backoff with Jitter可重试状态码白名单重试预算控制2.2 第二层熔断与限流——保护自己和他人2.3 第三层降级与回退——当所有努力都失败时静态降级语义回退功能降级用户协商第三章实战代码——构建容错Agent工具调用层3.1 项目依赖与环境3.2 自定义重试器——带抖动和状态码过滤3.3 熔断器实现——滑动窗口统计3.4 缓存降级层——Redis 本地内存双级缓存3.5 完整的容错工具调用器——整合所有策略3.6 结合LLM Agent的语义回退层第四章监控、告警与可观测性4.1 关键指标4.2 结构化日志实现4.3 Prometheus指标暴露第五章真实场景演练——天气Agent的崩溃恢复5.1 场景设定5.2 我们的Agent行为分析5.3 效果数据第六章最佳实践与反模式6.1 最佳实践清单6.2 常见反模式要避免第一章问题域分析——Agent工具调用的脆弱性1.1 Agent工具调用的典型架构在现代Agent框架如LangChain、AutoGen、DSPy中工具调用遵循以下模式text用户请求 → LLM规划 → 工具选择 → API调用 → 结果解析 → LLM综合 → 最终响应其中“API调用”环节是整个链条中最不可控的节点。以一个旅行规划Agent为例它可能同时依赖航班查询API外部酒店价格API外部地图路线API外部汇率转换API外部内部数据库公司内部任何一个外部服务的抖动都会影响整体体验。1.2 故障类型枚举我们需要系统性地分类可能遇到的异常故障类型典型HTTP状态码示例场景瞬时网络抖动504, 502网关超时TCP重传服务端限流429超出速率限制授权失效401, 403Token过期数据不存在404查询的资源已删除服务端内部错误500数据库连接池耗尽慢响应超时-读取超时30s部分数据损坏200但格式错误JSON解析失败1.3 为什么不能简单重试最直观的做法是“重试几次”。但简单重试会导致雪崩效应在服务恢复初期大量Agent同时重试可能再次压垮服务延迟累积每次重试增加响应时间影响用户体验资源浪费不必要的计算和网络消耗幂等性问题非幂等操作如转账重复执行可能导致数据错误因此我们需要一套有状态、自适应、分层级的容错策略。第二章核心策略框架——三层防御体系我们构建一个“三层防御”模型text第一层即时重试瞬态故障处理 ↓ 失败 第二层熔断与降级服务保护机制 ↓ 仍然失败 第三层语义回退与用户协商业务补偿2.1 第一层智能重试——不只是“再试一次”指数退避与抖动Exponential Backoff with Jitter这是分布式系统中最经典的策略。核心公式textdelay min(cap, base * (2 ^ attempt)) random_jitter其中base是初始延迟如1秒attempt是重试次数从0开始cap是最大延迟上限如60秒jitter是随机抖动防止惊群效应。为什么需要抖动假设100个Agent同时遇到503错误如果没有抖动它们会在第1秒、2秒、4秒……精确同步重试导致服务端在恢复瞬间再次被打垮。加入随机抖动后重试时间分散在一个区间内。可重试状态码白名单并非所有错误都值得重试可重试429, 502, 503, 504, 408超时不可重试400参数错误, 401认证失败, 403权限不足, 404不存在重试预算控制每次工具调用应当有一个总时间预算如10秒超过预算则放弃重试进入下一层策略。2.2 第二层熔断与限流——保护自己和他人熔断器Circuit Breaker的三个状态CLOSED闭合正常调用记录失败率OPEN打开快速失败不实际调用API直接返回降级结果HALF_OPEN半开允许少量探测请求检查服务是否恢复失败率阈值建议滑动窗口内失败率 50% 且请求量 最小样本数如20个则打开熔断器。2.3 第三层降级与回退——当所有努力都失败时静态降级返回缓存的旧数据或预设的默认值。例如天气Agent在API不可用时返回“当前天气数据暂不可用这是您上次查询的结果2小时前”。语义回退使用LLM自身的常识知识生成合理的近似回答。例如“虽然无法获取实时股价但根据公开信息科技板块近期整体呈上升趋势。”功能降级放弃调用该工具改用备选工具。例如主支付网关失败时切换到备用支付通道。用户协商主动告知用户当前服务状态询问是否等待、使用旧数据或取消操作。第三章实战代码——构建容错Agent工具调用层下面我们使用Python 3.12结合httpx、tenacity和自定义熔断器实现一个生产级的容错工具调用系统。3.1 项目依赖与环境python# pyproject.toml [project] dependencies [ httpx0.27.0, pydantic2.6.0, tenacity8.2.0, redis5.0.0, # 用于分布式缓存 structlog24.1.0, # 结构化日志 ]3.2 自定义重试器——带抖动和状态码过滤pythonimport asyncio import random import time from typing import Optional, Callable, Awaitable, TypeVar, Tuple from dataclasses import dataclass, field from enum import Enum import httpx import structlog logger structlog.get_logger() T TypeVar(T) class RetryableError(Exception): 可重试的异常基类 pass class NonRetryableError(Exception): 不可重试的异常直接失败 pass dataclass class RetryConfig: max_attempts: int 5 base_delay_seconds: float 1.0 max_delay_seconds: float 60.0 jitter_factor: float 0.3 # 抖动幅度 ±30% total_timeout_seconds: float 30.0 retryable_status_codes: Tuple[int, ...] (429, 502, 503, 504, 408) retryable_exceptions: Tuple[type, ...] (httpx.TimeoutException, httpx.ConnectError, httpx.TransportError) class AsyncRetryer: 异步重试器支持指数退避抖动总超时 def __init__(self, config: RetryConfig): self.config config self._start_time: Optional[float] None async def execute( self, func: Callable[[], Awaitable[T]], attempt_context: Optional[dict] None ) - T: 执行带重试的异步函数 self._start_time time.monotonic() last_exception None for attempt in range(self.config.max_attempts): # 检查总超时 elapsed time.monotonic() - self._start_time if elapsed self.config.total_timeout_seconds: raise TimeoutError(fTotal timeout {self.config.total_timeout_seconds}s exceeded) try: result await func() # 成功则记录并返回 if attempt 0: logger.info(retry_success, attemptattempt, elapsedelapsed) return result except Exception as e: last_exception e should_retry, retry_reason self._should_retry(e, attempt) if not should_retry: logger.warning(non_retryable_error, errorstr(e), reasonretry_reason) raise NonRetryableError(fNon-retryable error: {e}) from e # 如果是最后一次尝试不再等待直接抛出 if attempt self.config.max_attempts - 1: logger.error(max_retries_exceeded, attemptsattempt1, errorstr(e)) raise RetryableError(fMax retries exceeded: {e}) from e # 计算退避延迟 delay self._calculate_delay(attempt) logger.info( retry_attempt, attemptattempt 1, max_attemptsself.config.max_attempts, delaydelay, errorstr(e) ) await asyncio.sleep(delay) # 理论上不会到达这里 raise RetryableError(fUnexpected retry failure: {last_exception}) from last_exception def _should_retry(self, error: Exception, attempt: int) - Tuple[bool, str]: 判断是否应该重试 # 检查异常类型 if isinstance(error, self.config.retryable_exceptions): return True, retryable_exception # 检查HTTP状态码如果是httpx.HTTPStatusError if isinstance(error, httpx.HTTPStatusError): status_code error.response.status_code if status_code in self.config.retryable_status_codes: return True, fretryable_status_{status_code} else: return False, fnon_retryable_status_{status_code} # 其他异常默认不重试 return False, unknown_exception def _calculate_delay(self, attempt: int) - float: 计算退避延迟指数 抖动 exponential self.config.base_delay_seconds * (2 ** attempt) capped min(exponential, self.config.max_delay_seconds) jitter random.uniform( -self.config.jitter_factor * capped, self.config.jitter_factor * capped ) return max(0.1, capped jitter) # 至少0.1秒3.3 熔断器实现——滑动窗口统计pythonfrom collections import deque import threading from datetime import datetime, timedelta class CircuitBreakerState(Enum): CLOSED closed OPEN open HALF_OPEN half_open dataclass class CircuitBreakerConfig: failure_threshold: float 0.5 # 失败率阈值 50% min_requests: int 20 # 滑动窗口最小请求数 open_timeout_seconds: float 60.0 # 熔断持续时间 half_open_max_requests: int 3 # 半开状态允许的最大探测请求数 class CircuitBreaker: 线程安全的熔断器实现 def __init__(self, name: str, config: CircuitBreakerConfig): self.name name self.config config self._state CircuitBreakerState.CLOSED self._window deque(maxlen1000) # (timestamp, success: bool) self._open_until: Optional[datetime] None self._half_open_requests 0 self._lock threading.RLock() self._logger logger.bind(circuit_breakername) property def state(self) - CircuitBreakerState: with self._lock: return self._state def record_result(self, success: bool): 记录一次调用结果 with self._lock: now datetime.now() self._window.append((now, success)) # 清理超过1分钟的旧数据 cutoff now - timedelta(minutes1) while self._window and self._window[0][0] cutoff: self._window.popleft() # 如果当前是HALF_OPEN状态记录探测请求数 if self._state CircuitBreakerState.HALF_OPEN: self._half_open_requests 1 if not success or self._half_open_requests self.config.half_open_max_requests: # 如果探测失败或已达到最大探测数转回OPEN self._state CircuitBreakerState.OPEN self._open_until now timedelta(secondsself.config.open_timeout_seconds) self._half_open_requests 0 self._logger.warning(circuit_breaker_half_open_fail, open_untilself._open_until) else: # 探测成功转回CLOSED self._state CircuitBreakerState.CLOSED self._half_open_requests 0 self._logger.info(circuit_breaker_closed) return # 仅在CLOSED状态下评估失败率 if self._state CircuitBreakerState.CLOSED: total len(self._window) if total self.config.min_requests: failures sum(1 for _, success in self._window if not success) failure_rate failures / total if failure_rate self.config.failure_threshold: self._state CircuitBreakerState.OPEN self._open_until now timedelta(secondsself.config.open_timeout_seconds) self._logger.warning( circuit_breaker_opened, failure_ratefailure_rate, total_requeststotal, open_untilself._open_until ) def allow_request(self) - bool: 判断是否允许发起新请求 with self._lock: if self._state CircuitBreakerState.CLOSED: return True if self._state CircuitBreakerState.OPEN: # 检查是否到了半开状态的时间 if self._open_until and datetime.now() self._open_until: self._state CircuitBreakerState.HALF_OPEN self._half_open_requests 0 self._logger.info(circuit_breaker_half_open) return True return False # HALF_OPEN 状态只允许有限数量的探测请求 if self._state CircuitBreakerState.HALF_OPEN: if self._half_open_requests self.config.half_open_max_requests: return True return False return False3.4 缓存降级层——Redis 本地内存双级缓存pythonimport json import hashlib from typing import Any, Optional, Dict from abc import ABC, abstractmethod import redis.asyncio as redis from functools import lru_cache import pickle class CacheBackend(ABC): abstractmethod async def get(self, key: str) - Optional[Any]: pass abstractmethod async def set(self, key: str, value: Any, ttl_seconds: int) - None: pass abstractmethod async def exists(self, key: str) - bool: pass class MemoryCache(CacheBackend): 进程内内存缓存LRU风格 def __init__(self, max_size: int 1000): self._cache: Dict[str, tuple] {} # key - (value, expire_timestamp) self.max_size max_size async def get(self, key: str) - Optional[Any]: if key in self._cache: value, expire_ts self._cache[key] if expire_ts time.time(): return value else: del self._cache[key] return None async def set(self, key: str, value: Any, ttl_seconds: int) - None: if len(self._cache) self.max_size: # 简单驱逐策略删除最早的条目 oldest_key next(iter(self._cache)) del self._cache[oldest_key] self._cache[key] (value, time.time() ttl_seconds) async def exists(self, key: str) - bool: return await self.get(key) is not None class RedisCache(CacheBackend): Redis分布式缓存 def __init__(self, redis_url: str redis://localhost:6379/0): self.client redis.from_url(redis_url, decode_responsesTrue) async def get(self, key: str) - Optional[Any]: data await self.client.get(key) if data: return json.loads(data) return None async def set(self, key: str, value: Any, ttl_seconds: int) - None: await self.client.setex(key, ttl_seconds, json.dumps(value)) async def exists(self, key: str) - bool: return await self.client.exists(key) 0 class CacheManager: 双级缓存管理先读内存再读Redis def __init__(self, memory_cache: MemoryCache, redis_cache: Optional[RedisCache] None): self.memory memory_cache self.redis redis_cache self._logger logger.bind(componentcache) async def get(self, key: str) - Optional[Any]: # L1: 内存 value await self.memory.get(key) if value is not None: self._logger.debug(cache_hit, levelmemory, keykey) return value # L2: Redis if self.redis: value await self.redis.get(key) if value is not None: self._logger.debug(cache_hit, levelredis, keykey) # 回填内存缓存 await self.memory.set(key, value, ttl_seconds60) # 内存中保留短一些 return value self._logger.debug(cache_miss, keykey) return None async def set(self, key: str, value: Any, ttl_seconds: int 300) - None: await self.memory.set(key, value, ttl_secondsmin(ttl_seconds, 120)) # 内存TTL短 if self.redis: await self.redis.set(key, value, ttl_secondsttl_seconds) def _generate_key(self, tool_name: str, params: dict) - str: 生成缓存键 sorted_params json.dumps(params, sort_keysTrue) hash_obj hashlib.sha256(f{tool_name}:{sorted_params}.encode()) return ftool_cache:{hash_obj.hexdigest()}3.5 完整的容错工具调用器——整合所有策略pythonfrom typing import Any, Dict, Optional, Callable, Awaitable import httpx import asyncio from datetime import datetime class FaultTolerantToolInvoker: 容错工具调用器 - 整合重试、熔断、缓存、降级 def __init__( self, retry_config: Optional[RetryConfig] None, circuit_config: Optional[CircuitBreakerConfig] None, cache_manager: Optional[CacheManager] None, default_timeout: float 10.0, ): self.retry_config retry_config or RetryConfig() self.circuit_config circuit_config or CircuitBreakerConfig() self.cache_manager cache_manager or CacheManager(MemoryCache()) self.default_timeout default_timeout self._circuit_breakers: Dict[str, CircuitBreaker] {} self._http_client httpx.AsyncClient(timeoutdefault_timeout) self._logger logger.bind(componenttool_invoker) def _get_circuit_breaker(self, tool_name: str) - CircuitBreaker: if tool_name not in self._circuit_breakers: self._circuit_breakers[tool_name] CircuitBreaker(tool_name, self.circuit_config) return self._circuit_breakers[tool_name] async def invoke( self, tool_name: str, params: Dict[str, Any], url: str, method: str POST, headers: Optional[Dict[str, str]] None, use_cache: bool True, cache_ttl: int 300, fallback_value: Optional[Any] None, fallback_func: Optional[Callable[[Dict[str, Any]], Awaitable[Any]]] None, ) - Any: 核心调用方法 - 集成了完整的容错链路 流程 1. 检查缓存如果启用 2. 检查熔断器状态 3. 执行带重试的HTTP调用 4. 成功则更新缓存 5. 失败则尝试降级 6. 仍失败则抛出异常 cache_key f{tool_name}:{hashlib.sha256(json.dumps(params, sort_keysTrue).encode()).hexdigest()} # ---- 第1步缓存检查 ---- if use_cache: cached await self.cache_manager.get(cache_key) if cached is not None: self._logger.info(cache_hit, tooltool_name, keycache_key[:8]) return cached # ---- 第2步熔断器检查 ---- cb self._get_circuit_breaker(tool_name) if not cb.allow_request(): self._logger.warning(circuit_breaker_rejected, tooltool_name) # 熔断开启时尝试降级 return await self._execute_fallback(tool_name, params, fallback_value, fallback_func) # ---- 第3步带重试的执行 ---- retryer AsyncRetryer(self.retry_config) try: result await retryer.execute( lambda: self._do_http_call(url, method, params, headers) ) # 成功记录熔断器成功 cb.record_result(True) # 写入缓存 if use_cache: await self.cache_manager.set(cache_key, result, cache_ttl) self._logger.info(invoke_success, tooltool_name) return result except NonRetryableError as e: # 不可重试错误如参数错误、认证失败等 self._logger.error(non_retryable_error, tooltool_name, errorstr(e)) cb.record_result(False) raise # 直接抛出不降级 except (RetryableError, TimeoutError) as e: # 可重试但最终失败 self._logger.error(invoke_failed_after_retries, tooltool_name, errorstr(e)) cb.record_result(False) # 尝试降级 return await self._execute_fallback(tool_name, params, fallback_value, fallback_func) except Exception as e: # 未知错误 self._logger.exception(unknown_error, tooltool_name, errorstr(e)) cb.record_result(False) return await self._execute_fallback(tool_name, params, fallback_value, fallback_func) async def _do_http_call(self, url: str, method: str, params: Dict, headers: Optional[Dict]) - Any: 实际的HTTP请求 headers headers or {} headers.setdefault(Content-Type, application/json) headers.setdefault(User-Agent, FaultTolerantAgent/1.0) if method.upper() GET: response await self._http_client.get(url, paramsparams, headersheaders) elif method.upper() POST: response await self._http_client.post(url, jsonparams, headersheaders) else: response await self._http_client.request(method, url, jsonparams, headersheaders) response.raise_for_status() # 会抛出HTTPStatusError return response.json() async def _execute_fallback( self, tool_name: str, params: Dict, fallback_value: Any, fallback_func: Optional[Callable[[Dict[str, Any]], Awaitable[Any]]] ) - Any: 执行降级策略 if fallback_func is not None: self._logger.info(fallback_func, tooltool_name) return await fallback_func(params) if fallback_value is not None: self._logger.info(fallback_value, tooltool_name) return fallback_value # 默认降级返回一个友好的错误信息 self._logger.info(fallback_default, tooltool_name) return { error: Service temporarily unavailable. Please try again later., tool: tool_name, timestamp: datetime.utcnow().isoformat(), cached_data: None # 可以在上层进一步处理 } async def close(self): await self._http_client.aclose()3.6 结合LLM Agent的语义回退层上面实现了基础设施级的容错但我们还需要LLM层面的“语义回退”。当工具完全不可用时Agent应该能够理解工具失败的原因用自然语言告知用户提供替代建议或使用常识推理pythonfrom typing import List, Dict, Any from pydantic import BaseModel class ToolResult(BaseModel): success: bool data: Optional[Any] None error: Optional[str] None fallback_used: bool False cache_hit: bool False class SemanticFallbackAgent: 使用LLM进行语义回退的Agent包装器 def __init__(self, llm_client, invoker: FaultTolerantToolInvoker): self.llm llm_client self.invoker invoker async def call_tool_with_fallback( self, tool_name: str, params: Dict[str, Any], user_intent: str, url: str, **kwargs ) - ToolResult: 带语义回退的工具调用 try: raw_result await self.invoker.invoke(tool_name, params, url, **kwargs) # 检查是否返回了降级错误信息 if isinstance(raw_result, dict) and error in raw_result: # 使用LLM生成用户友好的响应 friendly_response await self._generate_friendly_error( tool_name, user_intent, raw_result[error] ) return ToolResult( successFalse, errorraw_result[error], data{message: friendly_response}, fallback_usedTrue ) return ToolResult(successTrue, dataraw_result) except Exception as e: # 严重错误使用LLM生成解释 fallback_msg await self._generate_fallback_explanation(tool_name, user_intent, str(e)) return ToolResult( successFalse, errorstr(e), data{message: fallback_msg}, fallback_usedTrue ) async def _generate_friendly_error(self, tool_name: str, intent: str, error: str) - str: prompt f 用户意图{intent} 调用的工具{tool_name} 技术错误信息{error} 请用自然、友好的中文向用户解释这个错误不要使用技术术语并提供替代建议。 如果可能询问用户是否愿意等待或使用简化版本。 # 调用LLM生成 response await self.llm.acomplete(prompt) return response.text async def _generate_fallback_explanation(self, tool_name: str, intent: str, error: str) - str: prompt f 用户想要{intent} 但是工具 {tool_name} 当前完全不可用错误{error}。 请生成一段人性化的回复 1. 诚实告知服务暂时不可用 2. 用常识提供一些可能有帮助的信息不要编造具体数据 3. 建议用户稍后重试或联系人工客服 4. 保持语气温暖、专业 response await self.llm.acomplete(prompt) return response.text第四章监控、告警与可观测性容错策略不能是“黑盒”。我们需要实时监控以下指标4.1 关键指标指标说明告警阈值tool_call_total总调用次数-tool_call_success_rate成功率 95%tool_call_p99_latencyP99延迟 5scircuit_breaker_state熔断器状态OPEN状态超过2分钟cache_hit_rate缓存命中率 30% (根据场景)retry_attempts_per_call平均重试次数 2fallback_usage_rate降级使用率 10%4.2 结构化日志实现pythonimport structlog from structlog.contextvars import bind_contextvars, unbind_contextvars # 在invoke方法中添加上下文绑定 async def invoke(self, ...): with bind_contextvars( tooltool_name, request_idrequest_id, methodmethod, ): self._logger.info(invoke_start, params_previewstr(params)[:100]) # ... 执行逻辑 self._logger.info(invoke_end, successsuccess, latencylatency_ms)4.3 Prometheus指标暴露pythonfrom prometheus_client import Counter, Histogram, Gauge, Enum tool_calls_total Counter(tool_calls_total, Total tool calls, [tool, status]) tool_latency Histogram(tool_latency_seconds, Tool call latency, [tool]) circuit_breaker_state Gauge(circuit_breaker_state, Circuit breaker state, [tool]) cache_hit_counter Counter(cache_hits_total, Cache hits, [level])第五章真实场景演练——天气Agent的崩溃恢复5.1 场景设定假设我们有一个旅行规划Agent每天被调用10万次查询天气。外部天气APIweatherapi.com出现了以下故障序列T0sAPI响应延迟从200ms飙升到8s大量请求超时T30s服务返回503负载过高T90s服务完全不可用TCP连接拒绝T180s服务恢复但流量涌入再次导致抖动5.2 我们的Agent行为分析python# 模拟代码 async def simulate_weather_agent(): invoker FaultTolerantToolInvoker( retry_configRetryConfig(max_attempts4, base_delay_seconds0.5), circuit_configCircuitBreakerConfig(failure_threshold0.5, min_requests10) ) # 缓存同一城市5分钟内复用 cache CacheManager(MemoryCache(), RedisCache()) # 模拟并发用户请求 cities [北京, 上海, 广州, 深圳] tasks [] for i in range(50): city random.choice(cities) tasks.append(invoker.invoke( tool_nameweather, params{city: city, date: 2026-08-17}, urlhttps://api.weatherapi.com/v1/forecast.json, methodGET, use_cacheTrue, cache_ttl300, fallback_value{temperature: 未知, condition: 数据暂时不可用} )) results await asyncio.gather(*tasks, return_exceptionsTrue) return results预期行为前20个请求部分超时触发重试指数退避请求延迟增加但成功率约70%失败率超过50%后熔断器打开后续请求直接返回缓存或降级值~1ms响应缓存命中同一城市5分钟内的请求直接返回不受API故障影响180s后熔断器进入HALF_OPEN状态允许少量探测请求探测成功熔断器闭合恢复正常调用5.3 效果数据时间窗口请求数成功率P99延迟熔断器状态0-30s故障初发15062%9.2sCLOSED30-90s故障加重30018%12.1sOPEN从45s开始90-180s完全宕机4500%但100%降级/缓存0.8sOPEN180-240s恢复期20094%2.1sHALF_OPEN→CLOSED关键收益在故障高峰期虽然API不可用但Agent依然能够在500ms内返回响应通过缓存和降级用户体验没有中断。第六章最佳实践与反模式6.1 最佳实践清单为每个工具独立配置重试/熔断参数— 不同API的SLA不同使用有抖动jitter的退避— 防止重试风暴缓存策略要区分数据新鲜度要求— 实时数据如股价TTL短静态数据如汇率可长降级内容要明确标注— 告知用户“这是缓存数据”熔断器要区分网络错误和业务错误— 404不应触发熔断设置全局的请求预算— 每个Agent会话的总工具调用时长限制提供人工接管接口— 当自动化链路完全失效时转人工6.2 常见反模式要避免❌无限重试— 导致资源耗尽❌所有错误都重试— 401/403重试毫无意义❌没有降级直接抛异常— Agent无法给出有意义的回复❌忽略幂等性— 重试非幂等操作可能导致数据重复如扣款❌同步阻塞重试— 异步Agent应使用非阻塞等待❌单点监控缺失— 没有监控就无法优化策略