Agent 降级策略：当 LLM 不可用时的容灾方案

┌──────────────────────────────────────────────────┐
│              LLM 故障分类                         │
├──────────────┬───────────┬───────────────────────┤
│   故障类型    │ 持续时间  │      影响范围          │
├──────────────┼───────────┼───────────────────────┤
│ API 完全宕机  │ 5-60min   │ 所有请求失败           │
│ 限流(429)    │ 1-10min   │ 超额请求失败           │
│ 超时         │ 30-120s   │ 单请求受影响           │
│ 模型退版     │ 永久      │ 特定版本不可用         │
│ 区域故障     │ 5-30min   │ 特定区域不可用         │
│ 质量退化     │ 未知      │ 输出质量下降           │
└──────────────┴───────────┴───────────────────────┘

@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: 优雅错误页面                  │     │
│  │         引导用户重试或联系人工              │     │
│  └────────────────────────────────────────────┘     │

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)

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

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

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)

□ 多 LLM Provider 热备，自动切换 < 5s
□ 降级模型已配置，功能限制明确
□ 语义缓存层已部署，命中率 > 30%
□ 高频问题预计算，离线生成
□ 规则引擎已覆盖核心业务意图
□ 降级通知 UX 已设计，用户可感知
□ 健康检查每 30s 执行，自动恢复
□ 降级演练每月执行，结果归档
□ 人工客服兜底渠道畅通
□ 降级期间数据不丢失（请求排队）