9.4 降级与容错
在上一节中，我们讨论了缓存策略如何让一部分请求绕过 LLM 直接返回。但缓存只能解决"重复请求"的问题，无法应对一个更严重的场景：LLM API 真的挂了怎么办？
2024~2025 年间，主流 LLM 提供商（OpenAI、Anthropic、Google）都发生过多次服务中断。当你的 Agent 系统深度依赖这些外部服务时，一个上游故障就可能让你的整个产品瘫痪。本节就来回答这个核心问题：如何在 LLM 服务不可用时，仍然向用户提供可接受的服务。
9.4.1 Agent 系统的脆弱性来源
Agent 系统对外部 LLM 服务有深度依赖，这带来了独特的脆弱性。
生活类比：假设你的餐厅只有一家供应商提供食材。平时没问题，但这家供应商一旦停工，你的餐厅就只能关门。聪明的做法是准备 2~3 家备用供应商，哪怕主供应商断供，也不至于开不了门。
脆弱性一：LLM API 不可用
云服务商的 API 可能因为各种原因不可用：区域性故障、账号欠费、配额用尽、服务维护。主流 LLM 提供商的公开事故报告显示，SLA 通常为 99.5%~99.9%，这意味着每年可能有 4~44 小时的服务不可用时间。
脆弱性二：响应质量退化
即使 API 可用，也可能出现响应质量退化：输出截断、格式错误、幻觉增加。这通常发生在模型版本更新或负载高峰期间。
脆弱性三：级联故障
一个 Agent 操作可能涉及 5~10 次 LLM 调用。如果其中一次调用失败，整个操作链可能中断，导致用户看到的是"系统错误"而非部分结果。
脆弱性四：成本失控
在降级场景下，如果不加控制，重试可能导致成本急剧增加——特别是当系统反复重试一个注定失败的请求时。
9.4.2 模型切换：主 → 备用 → 规则兜底
模型切换是最常用、最有效的降级策略。核心思想是：不要让用户感知到失败，而是静默切换到备选方案。
┌──────────────┐
│ 用户请求 │
└──────┬───────┘
│
┌──────▼───────┐
┌────│ 主模型 │────┐
│ │ (GPT-4o) │ │ 成功
│ └──────────────┘ │
│ 失败/超时 │
│ ┌──────────────┐ │
├────│ 备用模型 │────┤
│ │ (Claude) │ │ 成功
│ └──────────────┘ │
│ 失败/超时 │
│ ┌──────────────┐ │
├────│ 经济模型 │────┤
│ │ (DeepSeek) │ │ 成功
│ └──────────────┘ │
│ 失败/超时 │
│ ┌──────────────┐ │
└────│ 规则兜底 │◄───┘
│ (模板回复) │
└──────────────┘生活类比：就像你去餐厅点菜，服务员说"不好意思，这道菜今天卖完了，要不要试试我们另一道招牌菜？"——你不会空着肚子离开，而是得到了一个替代选择。
多级模型切换实现
python
import asyncio # 导入异步框架
import openai # 导入 OpenAI SDK
import anthropic # 导入 Anthropic SDK
from enum import Enum # 导入枚举
from typing import Optional, List, Dict, Any # 导入类型注解
from dataclasses import dataclass # 导入数据类
import time # 导入时间模块
import random # 导入随机数模块
class ModelTier(Enum):
PREMIUM = "premium" # 主模型：GPT-4o
BACKUP = "backup" # 备用模型：Claude
ECONOMY = "economy" # 经济模型：DeepSeek
FALLBACK = "fallback" # 规则兜底
@dataclass
class ModelConfig:
"""模型配置"""
provider: str # 提供商
model: str # 模型名称
timeout: float # 超时时间（秒）
max_retries: int # 最大重试次数
priority: int # 优先级（1 最高）
class ModelRouter:
"""多级模型路由器"""
def __init__(self):
# 配置模型层级（按优先级排序）
self.models = [
ModelConfig(
provider="openai", # OpenAI
model="gpt-4o", # GPT-4o
timeout=30.0, # 30 秒超时
max_retries=2, # 最多重试 2 次
priority=1 # 最高优先级
),
ModelConfig(
provider="anthropic", # Anthropic
model="claude-3-5-sonnet-20241022",
timeout=25.0,
max_retries=1,
priority=2 # 第二优先级
),
ModelConfig(
provider="deepseek", # DeepSeek
model="deepseek-chat",
timeout=20.0,
max_retries=1,
priority=3 # 第三优先级
),
]
# 初始化各提供商客户端
self.clients = {
"openai": openai.AsyncOpenAI(), # OpenAI 异步客户端
"anthropic": anthropic.AsyncAnthropic(), # Anthropic 异步客户端
"deepseek": openai.AsyncOpenAI( # DeepSeek 兼容 OpenAI 接口
api_key="your-deepseek-key",
base_url="https://api.deepseek.com"
),
}
# 模型健康状态（初始全部健康）
self.health_status: Dict[str, bool] = {
config.model: True for config in self.models
}
# 统计信息
self.stats: Dict[str, Dict[str, int]] = {
config.model: {"success": 0, "failure": 0, "fallback_triggers": 0}
for config in self.models
}
async def route(self, messages: List[dict], **kwargs) -> dict:
"""按优先级依次尝试模型"""
errors = [] # 收集错误信息
for config in sorted(self.models, key=lambda x: x.priority):
# 跳过不健康的模型
if not self.health_status[config.model]:
self.stats[config.model]["fallback_triggers"] += 1
continue # 跳过，尝试下一个
try:
result = await self._call_with_timeout(
config, messages, **kwargs
)
self.stats[config.model]["success"] += 1 # 记录成功
return {
"content": result, # 返回内容
"model_used": config.model, # 使用的模型
"tier": config.provider, # 提供商
"fallback_occurred": config.priority > 1 # 是否发生了降级
}
except asyncio.TimeoutError: # 超时
self.stats[config.model]["failure"] += 1
errors.append(f"{config.model}: timeout")
self.health_status[config.model] = False # 标记不健康
except Exception as e: # 其他异常
self.stats[config.model]["failure"] += 1
errors.append(f"{config.model}: {str(e)}")
self.health_status[config.model] = False
# 所有模型都失败，使用规则兜底
return self._rule_fallback(messages, errors)
async def _call_with_timeout(self, config: ModelConfig,
messages: List[dict], **kwargs) -> str:
"""带超时的模型调用"""
try:
result = await asyncio.wait_for( # 设置超时
self._call_model(config, messages, **kwargs),
timeout=config.timeout # 超时阈值
)
return result
except asyncio.TimeoutError:
raise # 超时直接抛出
except Exception as e:
# 如果是 Rate Limit，标记为不健康
if "rate_limit" in str(e).lower() or "429" in str(e):
self.health_status[config.model] = False
raise
async def _call_model(self, config: ModelConfig,
messages: List[dict], **kwargs) -> str:
"""调用具体的模型"""
client = self.clients[config.provider] # 获取客户端
for attempt in range(config.max_retries + 1): # 重试循环
try:
response = await client.chat.completions.create(
model=config.model, # 指定模型
messages=messages, # 消息列表
temperature=kwargs.get("temperature", 0.7), # 温度
max_tokens=kwargs.get("max_tokens", 2000), # 最大输出
)
return response.choices[0].message.content # 返回内容
except Exception as e:
if attempt == config.max_retries: # 最后一次重试
raise
wait = (2 ** attempt) + random.uniform(0, 1) # 指数退避+抖动
await asyncio.sleep(wait) # 等待后重试
def _rule_fallback(self, messages: List[dict], errors: List[str]) -> dict:
"""规则兜底：返回预设的模板回复"""
last_message = messages[-1]["content"] if messages else ""
# 简单的关键词匹配规则
fallback_responses = {
"天气": "抱歉，当前无法获取天气信息，请您稍后再试。",
"计算": "抱歉，计算服务暂时不可用，您可以使用其他工具。",
"翻译": "抱歉，翻译服务暂时不可用，请稍后再试。",
"搜索": "抱歉，搜索服务暂时不可用。建议您直接在搜索引擎中查询。",
"代码": "抱歉，代码生成服务暂时不可用，您可以参考相关文档。",
}
for keyword, response in fallback_responses.items():
if keyword in last_message: # 关键词匹配
return {
"content": response,
"model_used": "rule_fallback",
"tier": "fallback",
"fallback_occurred": True,
"errors": errors
}
# 默认兜底回复
return {
"content": "非常抱歉，我们的 AI 服务暂时遇到了技术问题，正在紧急修复中。请稍后再试，或联系客服获取帮助。",
"model_used": "rule_fallback",
"tier": "fallback",
"fallback_occurred": True,
"errors": errors
}
async def health_check(self):
"""定期健康检查：尝试恢复不健康的模型"""
for config in self.models:
if not self.health_status[config.model]: # 只检查不健康的
try:
# 发送一个简单的健康检查请求
await asyncio.wait_for(
self._call_model(
config,
[{"role": "user", "content": "ping"}] # 最小请求
),
timeout=10.0 # 10 秒超时
)
self.health_status[config.model] = True # 恢复健康
print(f"✅ {config.model} 已恢复")
except Exception:
pass # 仍未恢复，保持不健康状态与 FastAPI 集成
python
from fastapi import FastAPI, HTTPException # 导入 FastAPI
from pydantic import BaseModel # 导入数据模型
import asyncio # 导入异步框架
app = FastAPI() # 创建应用
router = ModelRouter() # 创建模型路由器
class ChatRequest(BaseModel): # 请求模型
message: str
history: list = []
@app.post("/api/chat")
async def chat(request: ChatRequest):
"""带自动降级的聊天接口"""
messages = request.history + [ # 拼接历史和当前消息
{"role": "user", "content": request.message}
]
result = await router.route(messages) # 路由到模型
# 记录降级事件
if result.get("fallback_occurred"):
print(f"⚠️ 触发降级: 使用 {result['model_used']}")
return {
"reply": result["content"], # 回复内容
"model": result["model_used"], # 使用的模型
"degraded": result.get("fallback_occurred", False) # 是否降级
}
@app.get("/api/health/models")
async def models_health():
"""查看模型健康状态"""
return {
"health": router.health_status, # 各模型健康状态
"stats": router.stats # 统计信息
}
# 后台健康检查任务
async def background_health_check():
"""每 30 秒执行一次健康检查"""
while True:
await asyncio.sleep(30)
await router.health_check()9.4.3 熔断器模式（Circuit Breaker）
熔断器是防止级联故障的关键模式。当 LLM 服务连续失败时，熔断器"断开"电路，直接返回错误或降级响应，而不是继续向已经故障的服务发送请求。
生活类比：熔断器就像家里的保险丝（空气开关）。当电流过载时，保险丝自动跳闸，保护电器不被烧毁。等故障排除后，手动推上去恢复供电。熔断器的三状态设计就是在模拟这个过程。
熔断器的三种状态
┌──────────┐
│ CLOSED │ 正常状态，请求正常通过
└─────┬────┘
│ 失败次数达到阈值
▼
┌──────────┐
│ OPEN │ 熔断状态，直接拒绝请求
└─────┬────┘
│ 等待超时时间后
▼
┌──────────┐
│ HALF_OPEN│ 半开状态，允许少量请求探测
└─────┬────┘
成功 │ 失败
┌───┴───┐
▼ ▼
CLOSED OPEN- CLOSED（关闭）：正常状态，所有请求正常通过。每次失败增加计数。
- OPEN（打开）：连续失败达到阈值后进入。直接拒绝请求，不调用下游服务。
- HALF_OPEN（半开）：OPEN 状态等待恢复超时后进入。允许少量探测请求，成功则恢复 CLOSED，失败则回到 OPEN。
实现
python
import asyncio # 导入异步框架
import time # 导入时间模块
from enum import Enum # 导入枚举
from typing import Callable, Any # 导入类型注解
class CircuitState(Enum):
CLOSED = "closed" # 正常
OPEN = "open" # 熔断
HALF_OPEN = "half_open" # 半开
class CircuitBreaker:
"""LLM 服务熔断器"""
def __init__(
self,
failure_threshold: int = 5, # 连续失败多少次后熔断
recovery_timeout: float = 60.0, # 熔断后多久进入半开状态
half_open_max_requests: int = 3, # 半开状态允许的探测请求数
success_threshold: int = 2, # 半开状态多少次成功后关闭
):
self.failure_threshold = failure_threshold # 失败阈值
self.recovery_timeout = recovery_timeout # 恢复超时
self.half_open_max_requests = half_open_max_requests # 半开最大请求数
self.success_threshold = success_threshold # 成功恢复阈值
self.state = CircuitState.CLOSED # 初始状态：关闭
self.failure_count = 0 # 连续失败计数
self.success_count = 0 # 半开状态成功计数
self.last_failure_time = 0 # 最后失败时间
self.half_open_requests = 0 # 半开已发请求数
self._lock = asyncio.Lock() # 异步锁
# 统计信息
self.total_successes = 0
self.total_failures = 0
self.total_rejected = 0 # 被熔断器拒绝的请求数
async def call(self, func: Callable, *args, **kwargs) -> Any:
"""通过熔断器调用函数"""
async with self._lock: # 加锁
# 检查是否应该转换状态
if self.state == CircuitState.OPEN:
# 检查是否已过恢复超时
if time.time() - self.last_failure_time >= self.recovery_timeout:
self.state = CircuitState.HALF_OPEN # 转为半开
self.half_open_requests = 0
self.success_count = 0
print("🟡 熔断器进入半开状态")
else:
# 仍在熔断中，直接拒绝
self.total_rejected += 1
raise CircuitBreakerOpenError(
f"熔断器已打开，将在 {self.recovery_timeout - (time.time() - self.last_failure_time):.0f}s 后重试"
)
# 半开状态下限制请求数
if self.state == CircuitState.HALF_OPEN:
if self.half_open_requests >= self.half_open_max_requests:
self.total_rejected += 1
raise CircuitBreakerOpenError("半开状态请求数已达上限")
self.half_open_requests += 1 # 增加探测计数
# 执行实际调用（不加锁，避免阻塞其他请求）
try:
result = await func(*args, **kwargs) # 调用目标函数
await self._on_success() # 成功回调
return result
except Exception as e:
await self._on_failure() # 失败回调
raise
async def _on_success(self):
"""成功时的处理"""
async with self._lock:
self.total_successes += 1
self.failure_count = 0 # 重置连续失败计数
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.success_threshold:
self.state = CircuitState.CLOSED # 恢复正常
print("🟢 熔断器已关闭（恢复正常）")
async def _on_failure(self):
"""失败时的处理"""
async with self._lock:
self.total_failures += 1
self.failure_count += 1
self.last_failure_time = time.time()
if self.state == CircuitState.CLOSED and self.failure_count >= self.failure_threshold:
self.state = CircuitState.OPEN # 触发熔断
print(f"🔴 熔断器已打开（连续失败 {self.failure_count} 次）")
elif self.state == CircuitState.HALF_OPEN:
self.state = CircuitState.OPEN # 半开失败，重新熔断
print("🔴 半开状态失败，重新打开熔断器")
class CircuitBreakerOpenError(Exception):
"""熔断器打开异常"""
pass
# ============ 使用示例 ============
circuit_breaker = CircuitBreaker(
failure_threshold=3, # 连续失败 3 次触发熔断
recovery_timeout=30.0, # 30 秒后进入半开
half_open_max_requests=2, # 半开允许 2 个探测请求
success_threshold=2 # 连续成功 2 次恢复
)
async def call_llm(messages: list) -> str:
"""通过熔断器调用 LLM"""
async def _call():
client = openai.AsyncOpenAI() # 创建客户端
response = await client.chat.completions.create(
model="gpt-4o", # 指定模型
messages=messages
)
return response.choices[0].message.content # 返回内容
try:
return await circuit_breaker.call(_call) # 通过熔断器调用
except CircuitBreakerOpenError as e:
# 熔断器打开，直接降级
return f"[降级响应] {str(e)}"9.4.4 超时与重试策略
超时配置
Agent 系统中的超时配置需要分层设置——不同层级的操作需要不同的超时阈值。
python
# 超时配置层级
TIMEOUTS = {
"llm_api": 30.0, # 单次 LLM API 调用：30 秒
"tool_call": 10.0, # 单次工具调用（如搜索、数据库查询）：10 秒
"agent_operation": 120.0, # 整个 Agent 操作（包括多步推理）：2 分钟
"http_request": 180.0, # 整个 HTTP 请求（从用户到响应）：3 分钟
}生活类比：就像装修工期管理。刷一面墙（工具调用）最多 1 天，一个房间（LLM 调用）最多 3 天，整套房子（Agent 操作）最多 2 个月。不同层级有不同的时间预算。
智能重试
python
import asyncio # 导入异步框架
import random # 导入随机数
from typing import Type, Tuple # 导入类型注解
class RetryPolicy:
"""可配置的重试策略"""
def __init__(
self,
max_retries: int = 3, # 最大重试次数
base_delay: float = 1.0, # 基础延迟
max_delay: float = 60.0, # 最大延迟
backoff_factor: float = 2.0, # 退避因子
jitter: bool = True, # 是否添加抖动
retryable_exceptions: Tuple[Type[Exception], ...] = (
asyncio.TimeoutError, # 可重试：超时
ConnectionError, # 可重试：连接错误
)
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.backoff_factor = backoff_factor
self.jitter = jitter
self.retryable_exceptions = retryable_exceptions
async def execute(self, func: Callable, *args, **kwargs) -> Any:
"""执行带重试的函数"""
last_exception = None
for attempt in range(self.max_retries + 1): # 重试循环
try:
return await func(*args, **kwargs) # 尝试执行
except self.retryable_exceptions as e:
last_exception = e
if attempt == self.max_retries: # 最后一次
raise # 不再重试
# 计算退避时间：base * factor^attempt
delay = min(
self.base_delay * (self.backoff_factor ** attempt),
self.max_delay
)
if self.jitter: # 添加随机抖动
delay = delay * (0.5 + random.random()) # 0.5x ~ 1.5x
print(f"重试 {attempt + 1}/{self.max_retries}，等待 {delay:.1f}s...")
await asyncio.sleep(delay) # 等待
except Exception as e:
# 不可重试的异常，直接抛出
raise
raise last_exception # 不应到达此行
# 使用示例
retry_policy = RetryPolicy(
max_retries=3,
base_delay=1.0,
max_delay=30.0,
retryable_exceptions=(asyncio.TimeoutError, ConnectionError, RateLimitError)
)
async def resilient_llm_call(messages):
return await retry_policy.execute(_call_llm_api, messages)9.4.5 完整的降级决策树
将前面所有组件整合成一个完整的降级管理器：
python
class DegradationManager:
"""降级管理器：整合所有降级策略"""
def __init__(self):
self.circuit_breaker = CircuitBreaker(failure_threshold=3) # 熔断器
self.model_router = ModelRouter() # 模型路由器
self.retry_policy = RetryPolicy(max_retries=2) # 重试策略
self.cache = ResponseCache() # 缓存
async def process(self, messages: list, user_id: str = None) -> dict:
"""完整的降级处理流程"""
# 第 1 层：尝试语义缓存
try:
cached = await self.cache.get(messages, "gpt-4o")
if cached:
return {"content": cached, "source": "cache"} # 缓存命中
except Exception:
pass # 缓存不可用时跳过，不阻塞主流程
# 第 2 层：通过熔断器调用主模型
try:
result = await self.circuit_breaker.call(
self._call_primary_model, messages
)
await self.cache.set(messages, "gpt-4o", result) # 缓存成功结果
return {"content": result, "source": "primary"}
except CircuitBreakerOpenError:
pass # 熔断器打开，进入第 3 层
# 第 3 层：模型路由切换（尝试备用模型）
try:
result = await self.model_router.route(messages)
return {"content": result["content"], "source": result["tier"]}
except Exception:
pass # 所有模型都失败，进入第 4 层
# 第 4 层：返回缓存的过期数据（如果有的话）
# 第 5 层：规则兜底
return {
"content": "服务暂时不可用，请稍后再试。",
"source": "rule_fallback"
}常见误区
"重试次数越多越安全"——过多的重试不仅增加延迟，还可能引发"重试风暴"。当大量请求同时重试时，会加重下游服务的负担。建议最多重试 2~3 次，且必须加抖动。
"熔断器打开后应该立即恢复"——熔断器的恢复需要一个"试探"过程（半开状态）。如果一恢复就放全量请求，很可能再次失败。半开状态的限流探测是必要的。
"降级就是返回错误页面"——好的降级是返回"可接受但不完美"的结果，而不是让用户看到"系统错误"。比如用便宜模型替代贵模型，用缓存替代实时调用，用模板回复替代 AI 生成。
"超时设置越长越安全"——过长的超时会导致请求堆积，最终压垮整个系统。快速失败（Fail Fast）比长时间等待更好——宁可返回错误，也不要让用户无止境等待。
"规则兜底没用"——很多人觉得模板回复太"蠢"，用户会不满意。但实际上，"服务暂时不可用，以下是您可能需要的链接"远比"系统错误"或无限等待好得多。用户能接受故障，但不能接受被忽视。
"只需要被动检测故障"——被动检测（熔断器）是基础，但主动检测（定期健康检查、订阅服务商状态页）能更早发现问题、更快恢复。
本节小结
| 要点 | 说明 |
|---|---|
| 降级层级 | 主模型 → 备用模型 → 经济模型 → 缓存 → 规则兜底 |
| 熔断器 | 三种状态（CLOSED/OPEN/HALF_OPEN），防止级联故障 |
| 重试策略 | 指数退避 + 随机抖动，区分可重试和不可重试的错误 |
| 超时配置 | 分层设置（API 30s / 工具 10s / Agent 120s / HTTP 180s） |
| 健康检查 | 被动（熔断器）+ 主动（定期探测），加速故障恢复 |
| 可观测性 | 记录每次降级事件，便于事后分析和优化 |
降级与容错让系统在风雨中站稳，但还有一件事我们一直在回避——钱。每次 LLM 调用都在消耗真金白银，如果不管不顾，一个月的账单可能让你心惊肉跳。下一节就来谈成本控制：如何用更少的钱做同样的事。
下一节：9.5 成本控制——Token 监控、模型分层、预算告警，管好每一分钱。