LLM 服务容灾设计：多模型/多区域/降级策略

单点风险分析

LLM 服务链路中的每个环节都可能成为单点故障：

故障点	影响	概率	恢复时间
LLM 提供商 API 宕机	完全不可用	中	5-60min
LLM 提供商限流	部分请求失败	高	1-5min
模型版本下线	特定模型不可用	低	需代码变更
网络抖动	延迟升高/超时	高	秒级
自建 GPU 节点故障	推理能力下降	中	10-30min
Embedding 服务故障	RAG 检索失效	低	5-15min

核心原则：任何单一 LLM 提供商都不应成为系统的单点依赖。

多模型冗余

模型池设计

from dataclasses import dataclass, field
from enum import Enum
import asyncio
from collections import defaultdict

class ModelStatus(Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"   # 可用但质量/延迟有问题
    UNHEALTHY = "unhealthy"

@dataclass
class ModelEndpoint:
    name: str
    provider: str           # openai / anthropic / local
    priority: int           # 1=首选, 2=备选...
    status: ModelStatus = ModelStatus.HEALTHY
    consecutive_errors: int = 0
    last_error_time: float = 0

class ModelPool:
    def __init__(self):
        self.endpoints = [
            ModelEndpoint("gpt-4o", "openai", priority=1),
            ModelEndpoint("claude-sonnet", "anthropic", priority=2),
            ModelEndpoint("llama-70b-local", "local", priority=3),
        ]
        self.circuit_breakers = defaultdict(CircuitBreaker)

    def get_available(self, exclude_degraded=False):
        """按优先级返回可用端点"""
        endpoints = sorted(self.endpoints, key=lambda e: e.priority)
        result = []
        for ep in endpoints:
            if ep.status == ModelStatus.UNHEALTHY:
                continue
            if exclude_degraded and ep.status == ModelStatus.DEGRADED:
                continue
            result.append(ep)
        return result

故障检测与自动切换

class FailoverHandler:
    def __init__(self, pool: ModelPool, max_retries=3):
        self.pool = pool
        self.max_retries = max_retries

    async def complete_with_failover(self, messages, **kwargs):
        endpoints = self.pool.get_available()
        last_error = None

        for ep in endpoints:
            try:
                response = await self._call_with_timeout(
                    ep, messages, timeout=30, **kwargs
                )
                # 成功调用，重置错误计数
                ep.consecutive_errors = 0
                ep.status = ModelStatus.HEALTHY
                return response, ep.name

            except asyncio.TimeoutError:
                self._record_error(ep, "timeout")
                last_error = "timeout"
            except RateLimitError:
                self._record_error(ep, "rate_limit")
                last_error = "rate_limit"
            except Exception as e:
                self._record_error(ep, str(e))
                last_error = str(e)

        # 所有端点都失败
        raise AllEndpointsFailedError(f"Last error: {last_error}")

    def _record_error(self, endpoint: ModelEndpoint, reason: str):
        endpoint.consecutive_errors += 1
        endpoint.last_error_time = time.time()

        if endpoint.consecutive_errors >= 5:
            endpoint.status = ModelStatus.UNHEALTHY
            logger.warning(f"Endpoint {endpoint.name} marked unhealthy: {reason}")
        elif endpoint.consecutive_errors >= 2:
            endpoint.status = ModelStatus.DEGRADED
            logger.info(f"Endpoint {endpoint.name} degraded: {reason}")

    async def _call_with_timeout(self, endpoint, messages, timeout, **kwargs):
        return await asyncio.wait_for(
            self._dispatch(endpoint, messages, **kwargs),
            timeout=timeout
        )

优雅降级

当所有高端模型不可用时，系统应自动降级到更小/更便宜的模型，而非直接报错。

class GracefulDegradation:
    def __init__(self, failover: FailoverHandler):
        self.failover = failover

    async def complete(self, messages, **kwargs):
        try:
            # 1. 尝试正常调用（全模型池）
            return await self.failover.complete_with_failover(
                messages, **kwargs
            )
        except AllEndpointsFailedError:
            # 2. 降级策略
            return await self._degrade(messages, **kwargs)

    async def _degrade(self, messages, **kwargs):
        # 策略1: 切换到更小模型
        try:
            response = await self._call_small_model(messages, **kwargs)
            return response, "degraded:small_model"
        except Exception:
            pass

        # 策略2: 去掉 RAG context，减少 token
        try:
            stripped = self._strip_context(messages)
            response = await self._call_small_model(stripped, **kwargs)
            return response, "degraded:no_context"
        except Exception:
            pass

        # 策略3: 返回缓存结果（如果有）
        cached = await self._get_cached_response(messages)
        if cached:
            return cached, "degraded:cached"

        # 策略4: 静态兜底
        return self._static_fallback(messages), "degraded:static"

    def _strip_context(self, messages):
        """移除 system 中的长上下文，只保留核心指令"""
        return [
            {**msg, "content": msg["content"][:500] if len(msg["content"]) > 500
             else msg["content"]}
            for msg in messages
        ]

    def _static_fallback(self, messages):
        """静态兜底回答"""
        return (
            "抱歉，AI 服务暂时不可用。请稍后重试或联系人工客服。"
            "I'm sorry, the AI service is temporarily unavailable."
        )

降级策略对比

策略	响应质量	实现复杂度	用户体验
小模型替代	80%	低	好
去上下文	60%	低	中
缓存返回	90%（旧）	中	中
静态兜底	0%	低	差
排队等待	100%	高	差（长等待）

熔断器

import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"      # 正常
    OPEN = "open"          # 熔断
    HALF_OPEN = "half_open"  # 半开（试探）

class CircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60,
                 half_open_max=3):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max = half_open_max
        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.last_failure_time = 0
        self.half_open_calls = 0

    def can_call(self) -> bool:
        if self.state == CircuitState.CLOSED:
            return True
        if self.state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitState.HALF_OPEN
                self.half_open_calls = 0
                return True
            return False
        if self.state == CircuitState.HALF_OPEN:
            return self.half_open_calls < self.half_open_max

    def record_success(self):
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.CLOSED
        self.failure_count = 0

    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.state == CircuitState.HALF_OPEN:
            self.state = CircuitState.OPEN
        elif self.failure_count >= self.failure_threshold:
            self.state = CircuitState.OPEN

多区域部署

                    ┌──────────────┐
                    │   CDN/WAF    │
                    │  (Cloudflare) │
                    └──────┬───────┘
                           │
                    ┌──────▼───────┐
                    │ Global LB    │
                    │ (DNS路由)     │
                    └──┬────┬──┬───┘
            ┌─────────┘    │  └─────────┐
     ┌──────▼──────┐ ┌────▼─────┐ ┌────▼──────┐
     │  Region US  │ │Region EU │ │ Region AP │
     │  OpenAI     │ │Anthropic │ │  本地模型  │
     │  GPT-4o     │ │ Claude   │ │  Llama    │
     │  Redis (US) │ │Redis(EU) │ │ Redis(AP) │
     └─────────────┘ └──────────┘ └───────────┘

class RegionRouter:
    def __init__(self):
        self.regions = {
            "us": {"primary": "openai", "fallback": "local"},
            "eu": {"primary": "anthropic", "fallback": "openai"},
            "ap": {"primary": "local", "fallback": "openai"},
        }

    def route(self, user_region):
        config = self.regions.get(user_region, self.regions["us"])
        return config["primary"], config["fallback"]

    async def complete(self, messages, user_region="us"):
        primary, fallback = self.route(user_region)
        try:
            return await self._call(primary, messages)
        except Exception:
            logger.warning(f"Primary {primary} failed, using {fallback}")
            return await self._call(fallback, messages)

限流降级

当系统过载时，主动拒绝低优先级请求以保护核心功能：

class PriorityRateLimiter:
    def __init__(self, capacity=100):
        self.capacity = capacity
        self.current_load = 0
        self.queues = {
            "critical": [],   # 付费用户、生产调用
            "normal": [],     # 普通用户
            "low": [],        # 批量任务、预计算
        }

    async def acquire(self, priority="normal"):
        if self.current_load >= self.capacity:
            if priority == "critical":
                # 高优先级挤占低优先级
                if self.queues["low"]:
                    self.queues["low"].pop(0)
                    self.current_load -= 1
                elif self.queues["normal"]:
                    self.queues["normal"].pop(0)
                    self.current_load -= 1
                else:
                    raise RateLimitError("System at capacity")
            elif priority == "normal":
                if self.queues["low"]:
                    self.queues["low"].pop(0)
                    self.current_load -= 1
                else:
                    raise RateLimitError("System at capacity")
            else:
                raise RateLimitError("Low priority requests rejected")

        self.current_load += 1
        self.queues[priority].append(time.time())

容灾演练

定期演练是验证容灾能力的关键：

class ChaosTest:
    """模拟故障的混沌测试"""

    async def test_provider_down(self):
        """模拟 OpenAI 完全宕机"""
        self.pool.mark_unhealthy("gpt-4o")
        response, model = await self.handler.complete_with_failover(
            test_messages
        )
        assert model != "gpt-4o", "Should failover to backup"
        assert response is not None
        self.pool.mark_healthy("gpt-4o")

    async def test_all_external_down(self):
        """模拟所有外部 API 不可用"""
        self.pool.mark_unhealthy("gpt-4o")
        self.pool.mark_unhealthy("claude-sonnet")
        response, level = await self.degradation.complete(
            test_messages
        )
        assert "degraded" in level
        assert response is not None

总结

LLM 服务容灾的核心是"不把鸡蛋放在一个篮子里"。多模型冗余消除单提供商依赖；熔断器防止级联故障；优雅降级确保极端情况下仍能返回可用结果；多区域部署降低地域性故障影响。容灾不是一次性的架构设计，而是需要定期演练的持续能力。记住：你的系统在最坏情况下的表现，才是真正的可用性。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。

单点风险分析#

多模型冗余#

模型池设计#

故障检测与自动切换#

优雅降级#

降级策略对比#

熔断器#

多区域部署#

限流降级#

容灾演练#

总结#

加入讨论#

📌 相关推荐

单点风险分析

多模型冗余

模型池设计

故障检测与自动切换

优雅降级

降级策略对比

熔断器

多区域部署

限流降级

容灾演练

总结

加入讨论