引言 2026年3月15日，OpenAI API 全球性宕机 47 分钟。那些没有降级方案的 Agent 应用全部显示"服务不可用"，而准备好容灾方案的产品几乎无感知地度过了这次故障。LLM 是 Agent 的核心，但它不是 100% 可靠的基础设施。本文将系统化构建 Agent 的多级降级体系。
一、故障分类与影响分析 1.1 LLM 服务故障类型 ┌──────────────────────────────────────────────────┐ │ LLM 故障分类 │ ├──────────────┬───────────┬───────────────────────┤ │ 故障类型 │ 持续时间 │ 影响范围 │ ├──────────────┼───────────┼───────────────────────┤ │ API 完全宕机 │ 5-60min │ 所有请求失败 │ │ 限流(429) │ 1-10min │ 超额请求失败 │ │ 超时 │ 30-120s │ 单请求受影响 │ │ 模型退版 │ 永久 │ 特定版本不可用 │ │ 区域故障 │ 5-30min │ 特定区域不可用 │ │ 质量退化 │ 未知 │ 输出质量下降 │ └──────────────┴───────────┴───────────────────────┘ 1.2 故障影响评估 @dataclass class FaultImpact: fault_type: str user_visible: bool # 用户是否感知 data_loss: bool # 是否丢数据 recovery_time: str # 恢复时间 business_impact: str # 低/中/高/严重 LLM_FAULT_IMPACTS = [ FaultImpact("API宕机", True, False, "5-60min", "严重"), FaultImpact("限流", True, False, "1-10min", "中"), FaultImpact("超时", True, False, "30-120s", "低"), FaultImpact("质量退化", False, False, "未知", "中"), ] 二、多级降级架构 ┌──────────────────────────────────────────────────────┐ │ 请求入口 │ │ ┌──────┴──────┐ │ │ │ 请求路由 │ │ │ └──────┬──────┘ │ │ │ │ │ ┌─────────────────────▼──────────────────────┐ │ │ │ Level 0: 主模型 (GPT-5) │ │ │ │ 延迟 < 2s, 成功率 99.5% │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 1: 备用模型 (Claude Opus) │ │ │ │ 延迟 < 3s, 成功率 99.5% │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 2: 降级模型 (GPT-5-mini) │ │ │ │ 延迟 < 1s, 功能受限 │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 3: 缓存/预计算结果 │ │ │ │ 精度下降，但可用 │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 4: 规则引擎/模板回复 │ │ │ │ 基本功能，无AI能力 │ │ │ └────────────────┬───────────────────────────┘ │ │ 失败 ↓ │ │ ┌─────────────────▼──────────────────────────┐ │ │ │ Level 5: 优雅错误页面 │ │ │ │ 引导用户重试或联系人工 │ │ │ └────────────────────────────────────────────┘ │ 三、各级降级实现 Level 0-1：模型切换 from enum import Enum import asyncio class ModelTier(Enum): PRIMARY = "gpt-5" FALLBACK = "claude-opus-4-2026" DEGRADED = "gpt-5-mini" EMERGENCY = "gpt-4o-mini" class LLMFailoverChain: """LLM 多级故障转移链""" def __init__(self): self.chain = [ {"model": "gpt-5", "timeout": 30, "retries": 2}, {"model": "claude-opus-4-2026", "timeout": 45, "retries": 1}, {"model": "gpt-5-mini", "timeout": 15, "retries": 2}, {"model": "gpt-4o-mini", "timeout": 10, "retries": 3}, ] self.circuit_breakers = {m["model"]: CircuitBreaker() for m in self.chain} self.health_checker = HealthChecker() async def invoke(self, messages: list, **kwargs) -> str: errors = [] for tier in self.chain: model = tier["model"] cb = self.circuit_breakers[model] # 检查熔断器状态 if not cb.can_call(): errors.append(f"{model}: circuit breaker open") continue # 检查健康状态 if not await self.health_checker.is_healthy(model): errors.append(f"{model}: unhealthy") continue try: response = await self._call_with_retry( model, messages, timeout=tier["timeout"], retries=tier["retries"], **kwargs ) cb.record_success() # 如果降级了，通知调用方 if model != self.chain[0]["model"]: logger.warning(f"Degraded to {model}") return response except (TimeoutError, RateLimitError) as e: cb.record_failure() errors.append(f"{model}: {e}") continue except Exception as e: cb.record_failure() errors.append(f"{model}: {e}") continue # 所有模型都失败，进入更深层的降级 raise AllModelsFailedError(errors) Level 2：功能降级 class DegradedMode: """降级模式：功能减少但核心可用""" DEGRADATION_RULES = { "gpt-5-mini": { "disable_tools": False, # 仍可用工具 "max_iterations": 3, # 减少迭代次数 "disable_multi_step": True, # 禁用多步推理 "simplified_prompt": True, # 简化 System Prompt "max_context_messages": 5, # 减少上下文 }, "gpt-4o-mini": { "disable_tools": True, # 禁用工具 "max_iterations": 1, # 单轮回复 "disable_multi_step": True, "simplified_prompt": True, "max_context_messages": 3, } } def apply(self, request: dict, model: str) -> dict: rules = self.DEGRADATION_RULES.get(model, {}) degraded_request = request.copy() if rules.get("simplified_prompt"): degraded_request["system"] = self._simplify_prompt(request.get("system", "")) if rules.get("disable_tools"): degraded_request.pop("tools", None) if rules.get("max_context_messages"): messages = degraded_request.get("messages", []) degraded_request["messages"] = messages[-rules["max_context_messages"]:] degraded_request["max_tokens"] = min( degraded_request.get("max_tokens", 4096), 1024 # 降级时限制输出长度 ) return degraded_request Level 3：缓存降级 class CacheFallback: """缓存降级：使用历史相似查询的结果""" async def try_cached_response(self, query: str) -> CachedResponse | None: # 1. 精确匹配 exact = await self.cache.get(query) if exact: return CachedResponse( content=exact, source="exact_cache", warning="This response is from cache due to AI service degradation." ) # 2. 语义匹配 similar = await self.semantic_cache.search(query, threshold=0.85) if similar: return CachedResponse( content=similar.response, source="semantic_cache", warning="This response is based on a similar historical query." ) # 3. 预计算回答（高频问题） precomputed = await self.precomputed_db.get(query) if precomputed: return CachedResponse( content=precomputed, source="precomputed", warning=None # 预计算结果质量有保证 ) return None Level 4：规则引擎降级 class RuleEngineFallback: """规则引擎：LLM 不可用时的最终保底""" def __init__(self): self.rules = self._load_rules() def _load_rules(self) -> list[Rule]: """加载预定义规则""" return [ # 意图匹配规则 Rule( pattern=r"(.*)价格(.*)", action=lambda m: f"我们的产品价格请参考：{self._get_pricing_table()}", intent="pricing_inquiry" ), Rule( pattern=r"(.*)退款(.*)", action=lambda m: f"退款申请已记录。您的退款将在3-5个工作日内处理。" f"工单号：{self._generate_ticket()}", intent="refund_request" ), Rule( pattern=r"(hello|hi|你好|嗨)", action=lambda m: "您好！我是智能助手。" "AI服务暂时受限，但我可以处理基础请求。请说明您需要什么帮助。", intent="greeting" ), ] def handle(self, user_input: str) -> str: for rule in self.rules: match = re.match(rule.pattern, user_input, re.IGNORECASE) if match: return rule.action(match) # 兜底回复 return ( "AI 服务暂时不可用，我无法处理您的复杂请求。\n" "您可以：\n" "1. 稍后重试\n" "2. 联系人工客服：400-xxx-xxxx\n" "3. 访问帮助中心：https://help.example.com" ) 四、降级决策器 class DegradationOrchestrator: """降级编排器：协调各级降级策略""" def __init__(self): self.llm_chain = LLMFailoverChain() self.cache = CacheFallback() self.rules = RuleEngineFallback() self.degraded_mode = DegradedMode() self.health_monitor = HealthMonitor() async def handle_request( self, messages: list, tools: list | None = None, user_id: str = "", priority: str = "normal" ) -> Response: # 检查系统健康状态 health = await self.health_monitor.get_status() # 根据健康状态选择策略 if health.llm_available: try: return await self.llm_chain.invoke(messages, tools=tools) except AllModelsFailedError: health.llm_available = False if health.cache_available: cached = await self.cache.try_cached_response( messages[-1]["content"] ) if cached: return Response( content=cached.content, degraded=True, source=cached.source, warning=cached.warning ) # 规则引擎兜底 rule_response = self.rules.handle(messages[-1]["content"]) return Response( content=rule_response, degraded=True, source="rule_engine", warning="AI 服务暂时不可用，正在使用规则引擎处理。" ) async def health_check_loop(self): """持续健康检查，自动恢复""" while True: await asyncio.sleep(30) # 每30秒检查一次 if not self.health_monitor.llm_available: # 尝试恢复 test_result = await self._test_llm_health() if test_result: self.health_monitor.llm_available = True logger.info("LLM service recovered!") await self._notify_recovery() 五、降级通知与用户体验 class DegradedUX: """降级时的用户体验设计""" MESSAGES = { "model_switch": { "inline": "（响应速度可能略有不同）", "banner": "⚠️ AI 服务正在切换到备用节点，响应可能略有延迟。" }, "cache_fallback": { "inline": "（以下为缓存回复）", "banner": "⚠️ AI 服务暂时繁忙，正在使用历史缓存回复您。" }, "rule_engine": { "inline": "（基础模式）", "banner": "⚠️ AI 服务暂时不可用，正在使用基础模式处理您的请求。" }, "full_failure": { "banner": "❌ AI 服务暂时不可用。您可以通过以下方式获取帮助：\n" "📞 客服热线：400-xxx-xxxx\n" "💬 在线客服：点击右下角\n" "📧 邮件：support@example.com\n" "我们正在紧急修复中，请稍后重试。" } } def render(self, response: Response) -> dict: if not response.degraded: return {"content": response.content, "banner": None} ux = self.MESSAGES.get(response.source, {}) return { "content": response.content, "banner": ux.get("banner"), "inline_note": ux.get("inline"), "retry_available": True, } 六、降级演练 class DegradationDrill: """降级演练：定期验证降级策略有效性""" async def run_drill(self): """模拟 LLM 故障，验证降级链""" results = [] # 测试 1: 主模型超时 with mock_llm_timeout(ModelTier.PRIMARY): result = await self.orchestrator.handle_request( messages=[{"role": "user", "content": "What is 2+2?"}] ) results.append(DrillResult( test="primary_timeout", passed=result.degraded == True, response_source=result.source )) # 测试 2: 所有模型宕机 with mock_all_llm_down(): result = await self.orchestrator.handle_request( messages=[{"role": "user", "content": "你好"}] ) results.append(DrillResult( test="all_llm_down", passed=result.source == "rule_engine", response_source=result.source )) # 测试 3: 缓存命中 with mock_all_llm_down(): # 先正常请求一次建立缓存 await self.orchestrator.handle_request( messages=[{"role": "user", "content": "What is your return policy?"}] ) # 再次请求相同问题 result = await self.orchestrator.handle_request( messages=[{"role": "user", "content": "What is your return policy?"}] ) results.append(DrillResult( test="cache_hit", passed=result.source in ("exact_cache", "semantic_cache"), response_source=result.source )) return DrillReport(results=results) 七、降级 Checklist □ 多 LLM Provider 热备，自动切换 < 5s □ 降级模型已配置，功能限制明确 □ 语义缓存层已部署，命中率 > 30% □ 高频问题预计算，离线生成 □ 规则引擎已覆盖核心业务意图 □ 降级通知 UX 已设计，用户可感知 □ 健康检查每 30s 执行，自动恢复 □ 降级演练每月执行，结果归档 □ 人工客服兜底渠道畅通 □ 降级期间数据不丢失（请求排队） 结语 降级策略的本质是"接受不完美，但永不不可用"。在 LLM 依赖度越来越高的今天，没有降级方案的 Agent 系统就像没有安全网的走钢丝。投资降级不是悲观——它是工程成熟度的标志。最好的降级是用户从未注意到的降级。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
...
