大模型 API 容错设计：重试、降级、熔断一个都不能少

大模型 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( f"Max 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(timeout=30) # response = await client.chat.completions.create( # model=model, # messages=messages, # temperature=temperature, # ) # 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_keys=True) return hashlib.sha256(raw.encode()).hexdigest() class MaxRetriesExceededError(Exception): """重试耗尽异常。""" pass

四、容错策略的代价

重试放大问题

重试请求会放大上游负载。如果原始 QPS 是 100，3 次重试意味着上游实际承载 300 QPS。在故障场景下这会让上游雪上加霜。必须配合 jitter + 限流。

降级的信息丢失

fallback 模型（如 GPT-4o-mini）的推理能力显著低于 GPT-4。复杂任务降级后质量可能大幅下降——用户能感受到。因此降级响应应该附带提示。

缓存的过期风险

过期的缓存响应可能导致用户看到过时信息。TLL 设置需要根据内容类型调整：事实性问题短 TTL（5min），通用问题长 TTL（1h）。

五、总结

大模型 API 容错是 Agent 系统的必选项而非可选项。三层防护各司其职：重试处理瞬时故障、熔断防止雪崩扩散、降级保障用户体验不断点。每层的参数都需要根据实际场景调优——重试次数太多反而加重故障，熔断阈值太低导致误判，降级策略太弱让用户丧失信任。容错不是静态配置，是持续调优的过程。