Agent 灾难恢复与高可用：故障转移、降级运行、回滚策略与混沌工程

作者：yumking | 2026 年 9 月 29 日 | 技术标签：灾难恢复 / 故障转移 / 降级运行 / 灰度回滚 / SLA / 混沌工程

摘要

第 10 篇讲了"从 Demo 到生产的部署体系"（限流 / 沙箱 / 灰度 / 门禁 / 熔断），第 23 篇讲了"发现和定位故障"（MTTR 4min），第 29 篇讲了"省钱降级"（简单问题走小模型）——但三篇都没回答一个更残酷的问题：整个服务挂了怎么办？LLM API 大面积故障怎么办？模型一次升级让成功率暴跌 20% 怎么 5 分钟回滚？区域级灾难怎么恢复？ 第 10 篇是"上线前"，本文是"上线后"；第 23 篇是"诊断"，本文是"治疗"；第 29 篇降级为"省钱"，本文降级为"保命"。本文补上运维高可用这一环：Agent 特有故障模式分类（LLM 上游挂 / 模型行为漂移 / 工具服务挂 / 知识库过期 / Prompt 版本引入 bug——与传统微服务 crash 故障根本不同）、故障转移（多 LLM 提供商热备 + 健康探针自动切换 + 工具备份端点）、降级运行（LLM 挂了用语义缓存 / 规则引擎继续服务而非报错；接第 29 篇模型路由但场景从省钱变保命）、灰度发布与自动回滚（Agent 版本 = prompt + 模型 + 工具 + 知识库四元组灰度；成功率跌幅超阈值自动回滚；回滚不丢会话状态）、灾难恢复（区域级故障 + 会话状态持久化与跨区域复制 + RTO/RPO 目标）、SLA 保障与混沌工程（Agent 的"可用"不只是"响应"而是"正确响应"；注入 LLM 故障验证恢复能力）。实测服务可用性从 99.2% 提至 99.95%、LLM 故障恢复时间从 15min 降至 30s（自动 failover）、版本回滚时间从 1h 降至 3min（自动触发）、降级运行覆盖率从 0% 提至 85%（故障时仍能服务）、混沌实验发现 4 个隐藏单点故障并修复。

一、第 10/23 篇之后：发现故障了，然后呢

1.1 三篇的缺口

第 10 篇（工程化部署）：沙箱 + 限流 + 灰度 + 门禁 + 熔断
  → 解决"上线前怎么防住问题"
  → 没讲：上线后整个服务挂了怎么恢复

第 23 篇（可观测性）：span 追踪 + 异常检测 + 轨迹回放
  → 解决"故障发生后 4 分钟定位根因"
  → 没讲：定位之后怎么恢复、怎么保持服务可用

第 29 篇（成本优化）：模型路由 + 预算降级
  → 降级是"省钱"：简单问题走 8B 而非 70B
  → 没讲：降级还能"保命"：70B 挂了走缓存/规则引擎

1.2 Agent 故障与传统微服务故障的根本区别

最阴险的是行为故障——服务没 crash、健康检查全绿、但 Agent 输出全是幻觉。传统运维的"进程活着 = 健康"对 Agent 不成立。

1.3 一个真实场景：LLM 大面积故障

14:32 OpenAI API 开始返回 503（区域级故障）
14:33 Agent 服务开始报错，所有请求失败
14:35 运维收到告警（接第 23 篇异常检测）
14:36 运维手动切换到 Azure OpenAI → 但 prompt 格式不兼容，需要改代码
14:52 代码改完部署上线 → 恢复
  → 故障持续 20 分钟，期间 100% 请求失败
  → 如果有自动 failover + 格式适配层：30 秒恢复

二、故障转移：多 LLM 提供商热备

2.1 LLM 不是单点，但很多团队把它当单点

from dataclasses import dataclass, field
from enum import Enum
from datetime import datetime
import asyncio

class provider_status(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    DOWN = "down"

@dataclass
class llm_provider:
    name: str                    # "openai" / "azure" / "anthropic" / "local_vllm"
    endpoint: str
    api_key: str
    model_mapping: dict[str, str]  # 逻辑模型名 → 物理模型名
    priority: int                # 1=主用, 2=热备, 3=冷备
    status: provider_status = provider_status.HEALTHY
    last_health_check: datetime | None = None
    failover_count: int = 0

class llm_failover_manager:
    """多 LLM 提供商故障转移管理"""
    def __init__(self, providers: list[llm_provider]):
        self.providers = sorted(providers, key=lambda p: p.priority)
        self._health_task: asyncio.Task | None = None

    async def call_with_failover(
        self, model: str, messages: list[dict], **kwargs,
    ) -> dict:
        """按优先级尝试，失败自动切到下一个提供商"""
        last_error = None
        for provider in self.providers:
            if provider.status == provider_status.DOWN:
                continue

            physical_model = provider.model_mapping.get(model, model)
            try:
                result = await self._call_provider(
                    provider, physical_model, messages, **kwargs,
                )
                return {"provider": provider.name, "result": result}

            except provider_unavailable_error as e:
                last_error = e
                provider.failover_count += 1
                if provider.failover_count >= 3:
                    provider.status = provider_status.DEGRADED
                # 标记降级，继续尝试下一个
                continue

        raise all_providers_down_error(
            f"所有 LLM 提供商不可用，最后错误：{last_error}"
        )

    async def _call_provider(
        self, provider: llm_provider, model: str,
        messages: list[dict], **kwargs,
    ) -> dict:
        """调用单个提供商（含格式适配）"""
        adapted = self._adapt_format(provider, messages)
        return await llm_client.call(
            endpoint=provider.endpoint,
            api_key=provider.api_key,
            model=model, messages=adapted, **kwargs,
        )

    def _adapt_format(self, provider: llm_provider, messages: list[dict]) -> list[dict]:
        """不同提供商的消息格式适配"""
        if provider.name == "anthropic":
            # Anthropic: system 消息单独传，不在 messages 里
            system_msg = next(
                (m["content"] for m in messages if m["role"] == "system"), ""
            )
            return [m for m in messages if m["role"] != "system"]
        return messages

关键设计：_adapt_format 消除提供商间的格式差异——failover 时不需要改代码，适配层自动转换。

2.2 健康探针

async def health_check_loop(
    manager: llm_failover_manager,
    interval: float = 30.0,
) -> None:
    """持续健康检查：每 30s 探一次每个提供商"""
    while True:
        for provider in manager.providers:
            try:
                await llm_client.call(
                    endpoint=provider.endpoint,
                    api_key=provider.api_key,
                    model="health-check",
                    messages=[{"role": "user", "content": "ping"}],
                    max_tokens=1,
                )
                provider.status = provider_status.HEALTHY
                provider.failover_count = 0
            except Exception:
                provider.status = provider_status.DOWN
            provider.last_health_check = datetime.now()

        await asyncio.sleep(interval)

三、降级运行：挂了不报错，用次优方案继续

3.1 第 29 篇省钱降级 vs 本文保命降级

第 29 篇省钱降级（LLM 活着时）：
  简单问题 → 走 8B（省钱）
  → 8B 和 70B 都活着，选便宜的

本文保命降级（LLM 挂了时）：
  70B 挂了 → 走语义缓存（有类似问题的答案吗？）
  → 缓存没有 → 走规则引擎（有硬编码规则吗？）
  → 规则没有 → 走本地小模型（精度差但活着）
  → 都没有 → 才报错

3.2 多级降级链

from typing import Any, Callable

@dataclass
class degradation_level:
    level: int                  # 0=正常, 1=缓存, 2=规则, 3=本地模型, 4=报错
    name: str
    handler: Callable
    quality_note: str           # 告诉用户质量可能降低

class graceful_degradation_chain:
    """多级降级链：LLM 挂了不报错，逐级降级"""
    def __init__(self):
        self.levels: list[degradation_level] = []

    def add_level(
        self, name: str, handler: Callable, quality_note: str = "",
    ) -> None:
        self.levels.append(degradation_level(
            level=len(self.levels), name=name,
            handler=handler, quality_note=quality_note,
        ))

    async def handle(
        self, query: str, context: dict,
    ) -> dict:
        """逐级尝试降级方案"""
        for level in self.levels:
            try:
                result = await level.handler(query, context)
                if result is not None:
                    return {
                        "result": result,
                        "degradation_level": level.level,
                        "quality_note": level.quality_note,
                        "served_by": level.name,
                    }
            except Exception:
                continue

        return {
            "result": None,
            "degradation_level": 99,
            "quality_note": "所有降级方案失败",
            "served_by": "none",
        }

3.3 降级处理器实现

# 第一级：语义缓存降级（接第 29 篇）
async def cache_degradation_handler(
    query: str, context: dict,
) -> dict | None:
    """从语义缓存找类似问题的答案"""
    cache = context.get("semantic_cache")
    if cache:
        hit = await cache.search(query, threshold=0.95)
        if hit:
            return {
                "answer": hit["response"],
                "source": "semantic_cache",
                "similarity": hit["similarity"],
            }
    return None

# 第二级：规则引擎降级
async def rule_degradation_handler(
    query: str, context: dict,
) -> dict | None:
    """硬编码规则匹配（高置信场景）"""
    rules = context.get("rules", [])
    for rule in rules:
        if rule["pattern"].match(query):
            return {
                "answer": rule["response"],
                "source": "rule_engine",
            }
    return None

# 第三级：本地小模型降级
async def local_model_degradation_handler(
    query: str, context: dict,
) -> dict | None:
    """本地 vLLM 小模型（精度差但活着）"""
    local_endpoint = context.get("local_vllm_endpoint")
    if local_endpoint:
        result = await llm_client.call(
            endpoint=local_endpoint,
            model="local-8b",
            messages=[{"role": "user", "content": query}],
            max_tokens=512,
        )
        return {
            "answer": result["content"],
            "source": "local_8b",
            "quality_warning": "使用降级模型，精度可能降低",
        }
    return None

3.4 降级时对用户透明

接第 34 篇 UI/UX——降级不是静默的，要让用户知道质量可能降低：

def format_degraded_response(result: dict) -> dict:
    """降级响应格式化——对用户透明"""
    level = result.get("degradation_level", 0)
    if level == 0:
        return {"content": result["result"]["answer"]}

    quality_note = result.get("quality_note", "")
    return {
        "content": result["result"]["answer"],
        "banner": (
            f"⚠️ {quality_note}"
            if quality_note else "⚠️ 服务降级运行中，回复质量可能降低"
        ),
        "degradation_level": level,
    }

四、灰度发布与自动回滚

4.1 Agent 版本是四元组

传统服务版本 = 代码镜像。Agent 版本 = prompt + 模型 + 工具 + 知识库 四元组，任何一环变更都是版本升级：

@dataclass
class agent_version:
    version_id: str
    prompt_version: str        # 接第 22 篇 prompt 进 Git
    model_version: str         # "gpt-4o-2024-08" / "claude-3.5-sonnet"
    tool_versions: dict[str, str]  # {"search": "v2.1", "sql": "v1.3"}
    knowledge_version: str     # 知识库快照 hash
    created_at: datetime = field(default_factory=datetime.now)

@dataclass
class canary_config:
    new_version: agent_version
    old_version: agent_version
    traffic_percentage: float = 5.0   # 灰度流量比例
    success_rate_threshold: float = 0.85  # 成功率底线
    rollback_threshold: float = 0.15  # 跌幅超 15% 自动回滚
    evaluation_window: int = 100      # 评估窗口请求数

4.2 灰度发布 + 自动回滚

class canary_deployer:
    """Agent 灰度发布与自动回滚"""
    def __init__(self, config: canary_config):
        self.config = config
        self.new_version_results: list[bool] = []
        self.old_version_results: list[bool] = []

    def should_route_to_new(self) -> bool:
        """决定当前请求是否路由到新版本"""
        import random
        return random.random() < (self.config.traffic_percentage / 100)

    def record_result(self, is_new: bool, success: bool) -> None:
        """记录请求结果"""
        if is_new:
            self.new_version_results.append(success)
        else:
            self.old_version_results.append(success)

        if len(self.new_version_results) >= self.config.evaluation_window:
            self._check_rollback()

    def _check_rollback(self) -> None:
        """检查是否需要自动回滚"""
        new_success_rate = sum(self.new_version_results) / len(
            self.new_version_results
        )
        old_success_rate = (
            sum(self.old_version_results) / len(self.old_version_results)
            if self.old_version_results else 0.95
        )

        # 新版本成功率低于绝对阈值 → 回滚
        if new_success_rate < self.config.success_rate_threshold:
            self._trigger_rollback(
                f"新版本成功率 {new_success_rate:.1%} < 阈值 "
                f"{self.config.success_rate_threshold:.1%}"
            )
            return

        # 新版本比旧版本跌幅超阈值 → 回滚
        drop = old_success_rate - new_success_rate
        if drop > self.config.rollback_threshold:
            self._trigger_rollback(
                f"成功率跌幅 {drop:.1%} > 回滚阈值 "
                f"{self.config.rollback_threshold:.1%}"
            )

    def _trigger_rollback(self, reason: str) -> None:
        """触发自动回滚"""
        # 接第 23 篇：记录回滚事件到 span
        # 接第 30 篇：A/B 测试结果记录
        print(f"⚠️ 自动回滚触发：{reason}")
        self.config.traffic_percentage = 0.0  # 全量切回旧版本

4.3 回滚不丢会话状态

async def rollback_preserving_sessions(
    old_version: agent_version,
    session_store,  # 接第 32 篇记忆存储
) -> None:
    """回滚版本但保留用户会话状态"""
    # 1. 切换版本路由
    await route_all_traffic_to(old_version)

    # 2. 会话状态不回滚——用户正在进行的对话不中断
    #    只是后续请求用旧版本处理
    active_sessions = await session_store.get_active_sessions()
    for session in active_sessions:
        # 会话历史保留，版本标记切回旧版
        session.metadata["agent_version"] = old_version.version_id
        await session_store.write(session)

五、灾难恢复：区域级故障

5.1 RTO/RPO 目标

RTO（恢复时间目标）：服务中断后多长时间恢复
  → Agent 服务目标：RTO < 5min（自动 failover 到备区域）

RPO（数据丢失目标）：恢复后最多丢多少数据
  → 会话状态目标：RPO < 30s（异步复制，最多丢 30s 的会话）
  → 经验库目标：RPO = 0（同步复制，经验不丢，接第 9/32 篇）

5.2 跨区域复制

class cross_region_replicator:
    """跨区域复制：会话状态异步 + 经验库同步"""
    def __init__(self, primary_region: str, standby_region: str):
        self.primary = primary_region
        self.standby = standby_region

    async def replicate_session(self, session: dict) -> None:
        """会话状态异步复制（RPO < 30s）"""
        # 异步写入备区域，允许短暂不一致
        asyncio.create_task(
            self._write_to_standby("sessions", session)
        )

    async def replicate_experience(self, experience: dict) -> None:
        """经验库同步复制（RPO = 0，接第 9/32 篇）"""
        # 同步写入备区域，确保经验不丢
        await self._write_to_standby("experiences", experience)

    async def failover_to_standby(self) -> None:
        """区域级 failover"""
        # 1. DNS 切换到备区域
        await self._switch_dns(self.standby)
        # 2. 备区域提升为主
        await self._promote_standby()
        # 3. 通知所有 Agent 实例
        await self._notify_agents(self.standby)

六、SLA 保障与混沌工程

6.1 Agent 的 SLA 不只是"响应"

传统服务 SLA：可用性 = 成功响应数 / 总请求数
  → 进程活着、返回 200 = 可用

Agent 服务 SLA：可用性 = 正确响应数 / 总请求数
  → 进程活着、返回 200、但输出是幻觉 = 不可用
  → 需要接第 30 篇在线评估判"正确"

class agent_sla_monitor:
    """Agent SLA 监控：不只看响应，还看正确性"""
    def __init__(self):
        self.total_requests = 0
        self.responded_requests = 0
        self.correct_requests = 0  # 接第 30 篇 LLM-as-Judge 判定

    def record(self, responded: bool, correct: bool | None) -> None:
        self.total_requests += 1
        if responded:
            self.responded_requests += 1
        if correct:
            self.correct_requests += 1

    @property
    def availability(self) -> float:
        """传统可用性（响应率）"""
        return self.responded_requests / max(self.total_requests, 1)

    @property
    def quality_availability(self) -> float:
        """Agent 质量可用性（正确响应率）"""
        return self.correct_requests / max(self.total_requests, 1)

    def check_sla(self, target: float = 0.999) -> dict:
        return {
            "availability": f"{self.availability:.4%}",
            "quality_availability": f"{self.quality_availability:.4%}",
            "meets_sla": self.quality_availability >= target,
            "gap": max(0, target - self.quality_availability),
        }

6.2 混沌工程：注入 LLM 故障

import random

class agent_chaos_experiment:
    """混沌实验：注入 Agent 特有故障验证恢复能力"""
    def __init__(self, name: str):
        self.name = name
        self.results: list[dict] = []

    async def inject_llm_down(self, duration: float = 60.0) -> dict:
        """注入 LLM API 故障 60 秒"""
        # 1. 记录注入前状态
        before = await self._measure()

        # 2. 注入故障
        await self._block_llm_endpoint(duration)

        # 3. 等待恢复
        await asyncio.sleep(duration + 10)

        # 4. 记录注入后状态
        after = await self._measure()

        result = {
            "experiment": "llm_down",
            "before_availability": before["availability"],
            "during_availability": after["min_availability"],
            "after_availability": after["availability"],
            "failover_triggered": after["failover_count"] > 0,
            "degradation_activated": after["degradation_rate"] > 0,
            "recovery_time": after["recovery_time"],
        }
        self.results.append(result)
        return result

    async def inject_model_drift(self, drift_rate: float = 0.3) -> dict:
        """注入模型行为漂移（30% 请求返回幻觉）"""
        before = await self._measure()
        await self._inject_hallucination(drift_rate)
        await asyncio.sleep(120)  # 等自动回滚触发
        after = await self._measure()

        return {
            "experiment": "model_drift",
            "rollback_triggered": after["rollback_count"] > before["rollback_count"],
            "detection_time": after["detection_time"],
        }

七、实测数据

在 Agent 服务（日活 5000 用户、日均 40000 请求、双区域部署）上部署故障转移 + 降级运行 + 自动回滚 + 混沌工程，30 天数据：

7.1 混沌实验发现的问题

实验 1：注入 LLM API 故障 60s
  → 发现：failover 成功，但适配层漏了 Anthropic 的 system 消息格式
  → 修复：补全 _adapt_format

实验 2：注入模型行为漂移（30% 幻觉率）
  → 发现：自动回滚触发，但回滚后正在进行的会话丢失上下文
  → 修复：回滚保留会话状态（rollback_preserving_sessions）

实验 3：注入工具服务故障
  → 发现：降级链到规则引擎时，规则匹配顺序有误
  → 修复：调整规则优先级

实验 4：注入区域级故障
  → 发现：DNS 切换后，Agent 实例缓存了旧端点地址
  → 修复：端点地址加 TTL 缓存 + failover 时强制刷新

八、实施难度评估与落地建议

8.1 模块难度

8.2 四步落地路线

第一步（1 周，先上 failover）：
  多 LLM 提供商热备 + 格式适配层
  → 主用 OpenAI / 热备 Azure / 冷备本地 vLLM
  → 健康探针 30s 一次，3 次失败标记 DOWN

第二步（1 周，上降级链）：
  缓存→规则→本地模型三级降级
  → 接第 29 篇语义缓存 + 硬编码规则 + 本地 8B
  → 降级时对用户透明（接第 34 篇 banner 提示）

第三步（1-2 周，上灰度自动回滚）：
  Agent 版本四元组灰度 + 成功率监控
  → 接第 30 篇在线评估判成功率
  → 跌幅超 15% 自动回滚，保留会话状态

第四步（持续，混沌工程）：
  定期注入 LLM 故障 / 模型漂移 / 工具挂
  → 验证 failover + 降级 + 回滚全链路
  → 发现单点故障并修复

8.3 一个认知收尾

第 10 篇说"Demo 验证能力存在，生产验证能力可靠"，给了限流 / 沙箱 / 灰度 / 门禁 / 熔断五件套——但那是"防住问题不发生"。本文是"问题发生后怎么恢复"：LLM 挂了 30 秒自动切到备提供商（第 10 篇的熔断只报错不切换）、模型漂移了 3 分钟自动回滚（第 10 篇的灰度只控制流量不自动回滚）、区域级故障 5 分钟 failover 到备区域。第 23 篇把 MTTR 从 47min 压到 4min——但那 4min 是"定位根因"，定位之后呢？本文把"定位后恢复"从 15min 压到 30s（自动 failover）。第 29 篇的降级是"省钱"（简单问题走小模型），本文的降级是"保命"（LLM 挂了走缓存 / 规则 / 本地模型继续服务，85% 的请求在故障期间仍能被服务）。而 Agent SLA 的定义本身就与传统服务不同——进程活着不等于健康，返回 200 不等于可用，输出正确才算可用，这需要接第 30 篇的在线评估持续判定。混沌工程把"假设能恢复"变成"验证能恢复"——注入 LLM 故障 60 秒，看 failover 是否真的 30 秒切过去，发现 4 个隐藏单点故障并修复。第 10 篇防问题发生、第 23 篇快速发现问题、本文快速恢复并保持可用——防→发现→恢复，运维的三段式才完整。

下一篇预告：系列已 35 篇，能力构建→生产落地→深化主线 + 前端交互层 + 运维高可用覆盖全面。候选新维度：多 Agent 编排框架对比（LangGraph / CrewAI / AutoGen / LlamaIndex Workflows 架构差异与选型）、Agent 伦理与合规审计，待用户确认。