大模型 API 容错设计重试、降级、熔断一个都不能少一、ChatGPT API 超时了然后我们整个 Agent 就卡住了大模型 API 最常见的故障模式超时30s模型推理卡住API 不返回也不报错限流429 Too Many Requests你的 QPS 超了配额服务不可用503上游 API 故障内容过滤content_filter被安全策略拦截返回空每种故障需要不同的处理策略。但很多 Agent 实现只做了最粗糙的try/except print(exc)这在生产环境就是定时炸弹。容错设计的核心原则即使 API 不可用也要优雅降级不能让用户干等或看到白屏。flowchart TD A[LLM API 调用] -- B{响应?} B --|成功| C[正常返回] B --|超时| D{重试策略} B --|429 限流| E{退避策略} B --|5xx 错误| F{熔断检查} B --|content_filter| G{降级策略} D -- D1[指数退避重试] D1 -- D2{重试次数 3?} D2 --|是| A D2 --|否| H[返回缓存/降级响应] E -- E1[读取 Retry-After 头] E1 -- E2[等待后重试] E2 -- A F -- F1{熔断器开启?} F1 --|否| D1 F1 --|是| I[快速失败br/返回降级响应] G -- G1[日志记录] G1 -- J[返回友好提示br/内容暂不可用]二、三种核心容错模式模式一指数退避重试Exponential Backoff Retry不是固定间隔重试。第一次重试等 1s第二次等 2s第三次等 4s——给上游喘息空间。配合 jitter抖动避免雷群效应——多个客户端同时重试同时打崩上游。模式二熔断器Circuit Breaker当错误率超过阈值时跳闸后续请求快速失败而不实际调用 API。一段时间后进入半开状态——放一个请求过去试探成功了关闭熔断器失败了继续熔断。模式三降级策略Fallback当所有重试和熔断都失效时Agent 应该给用户一个有意义的响应而非报错。降级策略包括返回缓存中的上一次成功响应告知用户正在重试请稍候切换到备用模型GPT-4 挂了降级到 4o-mini三、生产级容错实现import asyncio import random import time from dataclasses import dataclass, field from enum import Enum from typing import Optional, Callable, Awaitable from functools import wraps class CircuitState(Enum): CLOSED closed # 正常 OPEN open # 熔断 HALF_OPEN half_open # 半开试探 dataclass class RetryConfig: max_retries: int 3 base_delay: float 1.0 # 基础延迟秒 max_delay: float 30.0 # 最大延迟 backoff_factor: float 2.0 # 指数因子 jitter: bool True # 是否添加随机抖动 retryable_exceptions: tuple (TimeoutError, ConnectionError) dataclass class CircuitBreakerConfig: failure_threshold: int 5 # 连续失败 N 次后熔断 recovery_timeout: float 30.0 # 熔断后 N 秒尝试恢复 half_open_max_requests: int 3 # 半开状态最多试探请求数 class CircuitBreaker: 熔断器实现。 状态转换 CLOSED → OPEN连续失败 threshold OPEN → HALF_OPEN等待 recovery_timeout 后 HALF_OPEN → CLOSED试探成功 HALF_OPEN → OPEN试探失败 def __init__(self, config: CircuitBreakerConfig): self._config config self._state CircuitState.CLOSED self._failure_count 0 self._last_failure_time: float 0 self._half_open_requests 0 self._lock asyncio.Lock() async def call(self, func: Callable[[], Awaitable]): 执行调用自动处理熔断逻辑。 async with self._lock: if self._state CircuitState.OPEN: if time.time() - self._last_failure_time self._config.recovery_timeout: self._state CircuitState.HALF_OPEN self._half_open_requests 0 else: raise CircuitBreakerOpenError(Circuit breaker is open) if self._state CircuitState.HALF_OPEN: if self._half_open_requests self._config.half_open_max_requests: raise CircuitBreakerOpenError(Half-open max requests reached) self._half_open_requests 1 try: result await func() async with self._lock: if self._state CircuitState.HALF_OPEN: self._state CircuitState.CLOSED self._failure_count 0 return result except Exception as e: async with self._lock: self._failure_count 1 self._last_failure_time time.time() if self._failure_count self._config.failure_threshold: self._state CircuitState.OPEN raise e class CircuitBreakerOpenError(Exception): 熔断器开启异常——调用方应使用降级策略。 pass class LLMClient: 带全量容错的大模型 API 客户端。 三层防护 1. 重试指数退避 jitter处理瞬时故障 2. 熔断连续失败后快速失败防止雪崩 3. 降级切换到备用模型或返回缓存 def __init__( self, retry_config: Optional[RetryConfig] None, cb_config: Optional[CircuitBreakerConfig] None, fallback_model: str gpt-4o-mini, ): self._retry retry_config or RetryConfig() self._breaker CircuitBreaker(cb_config or CircuitBreakerConfig()) self._fallback_model fallback_model self._response_cache: dict[str, dict] {} async def chat_completion( self, model: str, messages: list[dict], temperature: float 0.7, ) - dict: 带容错的 chat completion 调用。 async def _call(): return await self._raw_api_call(model, messages, temperature) try: # 熔断器包裹调用 return await self._breaker.call(lambda: self._retry_call(_call)) except (CircuitBreakerOpenError, MaxRetriesExceededError): # 主模型不可用尝试降级 return await self._fallback_call(messages, temperature) async def _retry_call(self, func: Callable[[], Awaitable]): 指数退避重试逻辑。 last_exception None for attempt in range(self._retry.max_retries 1): try: return await func() except self._retry.retryable_exceptions as e: last_exception e if attempt self._retry.max_retries: break # 计算退避延迟 delay min( self._retry.base_delay * (self._retry.backoff_factor ** attempt), self._retry.max_delay, ) if self._retry.jitter: # 添加 ±25% 随机抖动避免雷群效应 delay * random.uniform(0.75, 1.25) await asyncio.sleep(delay) except Exception as e: # 非可重试异常——直接向上抛给熔断器 raise e raise MaxRetriesExceededError( fMax retries ({self._retry.max_retries}) exceeded ) from last_exception async def _fallback_call( self, messages: list[dict], temperature: float ) - dict: 降级策略切换到 fallback 模型。 try: return await self._raw_api_call( self._fallback_model, messages, temperature ) except Exception: # fallback 也失败了——返回缓存或兜底消息 cache_key self._build_cache_key(messages) if cache_key in self._response_cache: return self._response_cache[cache_key] return { choices: [{ message: { role: assistant, content: 服务暂时不可用正在重试。请稍后再问。, } }] } async def _raw_api_call( self, model: str, messages: list[dict], temperature: float ) - dict: 底层 API 调用——这里用伪代码表示结构。 # import openai # client openai.AsyncOpenAI(timeout30) # response await client.chat.completions.create( # modelmodel, # messagesmessages, # temperaturetemperature, # ) # return response.model_dump() # 模拟实现 await asyncio.sleep(0.1) return { choices: [{message: {role: assistant, content: 你好}}], } def _build_cache_key(self, messages: list[dict]) - str: import hashlib import json raw json.dumps(messages, sort_keysTrue) return hashlib.sha256(raw.encode()).hexdigest() class MaxRetriesExceededError(Exception): 重试耗尽异常。 pass四、容错策略的代价重试放大问题重试请求会放大上游负载。如果原始 QPS 是 1003 次重试意味着上游实际承载 300 QPS。在故障场景下这会让上游雪上加霜。必须配合 jitter 限流。降级的信息丢失fallback 模型如 GPT-4o-mini的推理能力显著低于 GPT-4。复杂任务降级后质量可能大幅下降——用户能感受到。因此降级响应应该附带提示。缓存的过期风险过期的缓存响应可能导致用户看到过时信息。TLL 设置需要根据内容类型调整事实性问题短 TTL5min通用问题长 TTL1h。五、总结大模型 API 容错是 Agent 系统的必选项而非可选项。三层防护各司其职重试处理瞬时故障、熔断防止雪崩扩散、降级保障用户体验不断点。每层的参数都需要根据实际场景调优——重试次数太多反而加重故障熔断阈值太低导致误判降级策略太弱让用户丧失信任。容错不是静态配置是持续调优的过程。