LLM 服务高可用架构：从单点部署到多活容灾，大模型推理服务的稳定性保障

一、LLM 服务的可用性挑战：推理延迟高与资源消耗大的双重约束

大模型推理服务与传统 Web 服务有本质区别。单次推理请求耗时数百毫秒到数秒，远超普通 API 的毫秒级响应。模型权重占用数十到数百 GB 显存，单张 GPU 卡无法承载大参数模型。推理过程中 GPU 利用率接近 100%，无法像 CPU 服务那样通过超卖提升利用率。

这些特性使得 LLM 服务的高可用设计面临独特挑战：单节点故障影响面大（一个 GPU 节点下线可能导致整个模型实例不可用），故障恢复慢（模型加载需要数十秒到数分钟），流量调度复杂（请求需要路由到有足够显存的节点）。传统的高可用方案（多副本 + 负载均衡）在 LLM 场景下需要重新设计。

二、LLM 服务高可用架构设计

LLM 服务的高可用需要在四个层面实现：推理引擎层的冗余部署、流量调度层的智能路由、数据层的模型缓存与预加载、容灾层的多活与降级。

flowchart TD
    A[客户端请求] --> B[流量调度层]
    B --> B1[负载均衡: 基于队列深度路由]
    B --> B2[健康检查: 推理就绪状态]
    B --> B3[灰度切换: 模型版本管理]

    B1 --> C[推理引擎集群]
    B2 --> C
    B3 --> C

    C --> C1[推理实例 A: GPU 节点 1]
    C --> C2[推理实例 B: GPU 节点 2]
    C --> C3[推理实例 C: GPU 节点 3]

    C1 --> D[模型存储层]
    C2 --> D
    C3 --> D

    D --> D1[共享存储: S3/NFS]
    D --> D2[本地缓存: SSD 热模型]
    D --> D3[预加载队列: 按需预热]

    C1 --> E[容灾层]
    C2 --> E
    C3 --> E

    E --> E1[多活部署: 跨可用区]
    E --> E2[降级策略: 小模型兜底]
    E --> E3[请求排队: 背压控制]

    style B fill:#e1f5fe
    style D fill:#e8f5e9
    style E fill:#fff3e0

2.1 基于队列深度的智能路由

# inference_router.py — 推理服务智能路由
# 设计意图：根据各推理实例的实时负载（队列深度、GPU 利用率）
# 动态路由请求，避免将请求发送到过载的实例

import time
import hashlib
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class InferenceInstance:
    """推理服务实例"""
    instance_id: str
    endpoint: str
    model_name: str
    gpu_utilization: float = 0.0
    queue_depth: int = 0
    avg_latency_ms: float = 0.0
    is_healthy: bool = True
    last_health_check: float = 0.0
    weight: int = 100  # 路由权重

@dataclass
class RouteDecision:
    """路由决策结果"""
    instance: InferenceInstance
    reason: str
    estimated_wait_ms: float

class InferenceRouter:
    def __init__(self, max_queue_depth: int = 20):
        self.instances: dict[str, InferenceInstance] = {}
        self.max_queue_depth = max_queue_depth

    def register_instance(self, instance: InferenceInstance):
        self.instances[instance.instance_id] = instance

    def remove_instance(self, instance_id: str):
        self.instances.pop(instance_id, None)

    def update_metrics(self, instance_id: str, metrics: dict):
        """更新实例的实时指标"""
        instance = self.instances.get(instance_id)
        if not instance:
            return

        instance.gpu_utilization = metrics.get('gpu_utilization', 0.0)
        instance.queue_depth = metrics.get('queue_depth', 0)
        instance.avg_latency_ms = metrics.get('avg_latency_ms', 0.0)
        instance.is_healthy = metrics.get('is_healthy', True)
        instance.last_health_check = time.time()

    def route(self, model_name: str, request_id: str = "") -> Optional[RouteDecision]:
        """为请求选择最优推理实例"""
        # 筛选健康且匹配模型的实例
        candidates = [
            inst for inst in self.instances.values()
            if inst.is_healthy
            and inst.model_name == model_name
            and inst.queue_depth < self.max_queue_depth
        ]

        if not candidates:
            return None

        # 策略 1：优先选择队列最浅的实例
        candidates.sort(key=lambda x: x.queue_depth)

        best = candidates[0]
        estimated_wait = best.queue_depth * best.avg_latency_ms

        # 如果所有实例队列都很深，返回最浅的那个（背压控制）
        if best.queue_depth >= self.max_queue_depth * 0.8:
            return RouteDecision(
                instance=best,
                reason=f"所有实例负载较高，选择队列最浅的 {best.instance_id}",
                estimated_wait_ms=estimated_wait,
            )

        return RouteDecision(
            instance=best,
            reason=f"选择队列深度最低的实例 {best.instance_id}",
            estimated_wait_ms=estimated_wait,
        )

    def health_check(self, timeout_seconds: float = 30.0):
        """检查实例健康状态，标记长时间无心跳的实例为不健康"""
        now = time.time()
        for instance in self.instances.values():
            if now - instance.last_health_check > timeout_seconds:
                instance.is_healthy = False

2.2 模型预加载与缓存管理

# model_cache_manager.py — 模型缓存与预加载管理
# 设计意图：将模型权重缓存到推理节点的本地 SSD，
# 冷启动时从本地加载而非远程存储，将加载时间从分钟级降到秒级

import os
import time
import shutil
from pathlib import Path
from typing import Optional

@dataclass
class ModelCacheEntry:
    model_name: str
    model_path: str           # 本地缓存路径
    model_size_bytes: int
    last_access_time: float
    access_count: int
    is_loaded: bool           # 是否已加载到 GPU

class ModelCacheManager:
    def __init__(
        self,
        cache_dir: str = "/mnt/ssd/model-cache",
        max_cache_size_bytes: int = 500 * 1024 * 1024 * 1024,  # 500GB
        remote_storage: str = "s3://models",
    ):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self.max_cache_size_bytes = max_cache_size_bytes
        self.remote_storage = remote_storage
        self.cache_entries: dict[str, ModelCacheEntry] = {}
        self.current_cache_size = 0

        # 启动时扫描已有缓存
        self._scan_existing_cache()

    def get_model(self, model_name: str) -> Optional[str]:
        """获取模型路径，如果本地没有则从远程下载"""
        entry = self.cache_entries.get(model_name)

        if entry and entry.is_loaded:
            entry.last_access_time = time.time()
            entry.access_count += 1
            return entry.model_path

        # 本地有缓存但未加载到 GPU
        if entry and os.path.exists(entry.model_path):
            entry.last_access_time = time.time()
            entry.access_count += 1
            return entry.model_path

        # 本地无缓存，从远程下载
        return self._download_model(model_name)

    def preload_model(self, model_name: str) -> bool:
        """预加载模型到本地缓存（不加载到 GPU）"""
        entry = self.cache_entries.get(model_name)
        if entry and os.path.exists(entry.model_path):
            return True  # 已有缓存

        result = self._download_model(model_name)
        return result is not None

    def _download_model(self, model_name: str) -> Optional[str]:
        """从远程存储下载模型到本地缓存"""
        # 检查缓存空间是否足够
        model_size = self._estimate_model_size(model_name)
        if self.current_cache_size + model_size > self.max_cache_size_bytes:
            # 缓存空间不足，淘汰最久未访问的模型
            self._evict_lru(model_size)

        local_path = self.cache_dir / model_name
        try:
            # 实际实现使用 s3cmd 或 boto3 下载
            # download_from_s3(f"{self.remote_storage}/{model_name}", local_path)
            print(f"下载模型 {model_name} 到 {local_path}")

            entry = ModelCacheEntry(
                model_name=model_name,
                model_path=str(local_path),
                model_size_bytes=model_size,
                last_access_time=time.time(),
                access_count=0,
                is_loaded=False,
            )
            self.cache_entries[model_name] = entry
            self.current_cache_size += model_size
            return str(local_path)

        except Exception as e:
            print(f"模型下载失败: {model_name}, 错误: {e}")
            return None

    def _evict_lru(self, required_space: int):
        """淘汰最久未访问的模型缓存"""
        sorted_entries = sorted(
            self.cache_entries.values(),
            key=lambda e: e.last_access_time,
        )

        freed = 0
        for entry in sorted_entries:
            if freed >= required_space:
                break

            # 不淘汰正在使用的模型
            if entry.is_loaded:
                continue

            shutil.rmtree(entry.model_path, ignore_errors=True)
            self.current_cache_size -= entry.model_size_bytes
            freed += entry.model_size_bytes
            del self.cache_entries[entry.model_name]

    def _estimate_model_size(self, model_name: str) -> int:
        """估算模型文件大小"""
        # 简化：基于模型名称中的参数量估算
        if "70b" in model_name.lower():
            return 140 * 1024 * 1024 * 1024  # 140GB
        elif "7b" in model_name.lower():
            return 14 * 1024 * 1024 * 1024   # 14GB
        return 10 * 1024 * 1024 * 1024       # 默认 10GB

    def _scan_existing_cache(self):
        """扫描已有的本地缓存"""
        if not self.cache_dir.exists():
            return

        for model_dir in self.cache_dir.iterdir():
            if model_dir.is_dir():
                size = sum(
                    f.stat().st_size for f in model_dir.rglob('*') if f.is_file()
                )
                self.cache_entries[model_dir.name] = ModelCacheEntry(
                    model_name=model_dir.name,
                    model_path=str(model_dir),
                    model_size_bytes=size,
                    last_access_time=time.time(),
                    access_count=0,
                    is_loaded=False,
                )
                self.current_cache_size += size

三、容灾与降级策略

3.1 多活部署与故障切换

# failover_controller.py — 多活容灾控制器
# 设计意图：跨可用区部署推理服务，单可用区故障时自动切换流量，
# 切换前验证目标可用区的服务就绪状态

import time
from enum import Enum
from typing import Optional

class ZoneStatus(Enum):
    ACTIVE = "active"       # 正常服务
    DEGRADED = "degraded"   # 降级服务
    FAILOVER = "failover"   # 故障切换中
    OFFLINE = "offline"     # 离线

@dataclass
class AvailabilityZone:
    zone_id: str
    endpoint: str
    status: ZoneStatus = ZoneStatus.ACTIVE
    inference_instances: list[InferenceInstance] = field(default_factory=list)
    healthy_instance_count: int = 0
    last_check_time: float = 0.0

class FailoverController:
    def __init__(self, health_check_interval: float = 10.0):
        self.zones: dict[str, AvailabilityZone] = {}
        self.health_check_interval = health_check_interval
        self.primary_zone: Optional[str] = None

    def add_zone(self, zone: AvailabilityZone, is_primary: bool = False):
        self.zones[zone.zone_id] = zone
        if is_primary:
            self.primary_zone = zone.zone_id

    def get_active_zone(self) -> Optional[AvailabilityZone]:
        """获取当前活跃的可用区"""
        # 优先返回主可用区
        if self.primary_zone:
            primary = self.zones.get(self.primary_zone)
            if primary and primary.status == ZoneStatus.ACTIVE:
                return primary

        # 主可用区不可用，选择健康的备可用区
        for zone in self.zones.values():
            if zone.status == ZoneStatus.ACTIVE and zone.healthy_instance_count > 0:
                return zone

        return None

    def handle_zone_failure(self, failed_zone_id: str) -> Optional[str]:
        """处理可用区故障，切换流量到备用区"""
        failed_zone = self.zones.get(failed_zone_id)
        if not failed_zone:
            return None

        failed_zone.status = ZoneStatus.OFFLINE

        # 选择目标可用区
        target_zone = None
        for zone in self.zones.values():
            if zone.zone_id != failed_zone_id and zone.status == ZoneStatus.ACTIVE:
                if zone.healthy_instance_count > 0:
                    target_zone = zone
                    break

        if not target_zone:
            # 所有可用区都不可用，触发降级
            return self._activate_degradation()

        # 验证目标可用区就绪
        if self._verify_zone_readiness(target_zone):
            target_zone.status = ZoneStatus.ACTIVE
            return target_zone.zone_id

        return None

    def _verify_zone_readiness(self, zone: AvailabilityZone) -> bool:
        """验证目标可用区是否就绪"""
        healthy = sum(
            1 for inst in zone.inference_instances if inst.is_healthy
        )
        return healthy >= len(zone.inference_instances) * 0.5

    def _activate_degradation(self) -> str:
        """激活降级策略：使用小模型兜底"""
        return "degradation:switch-to-small-model"

四、边界分析与架构权衡

跨可用区延迟：多活部署意味着推理请求可能跨可用区路由，网络延迟从毫秒级增加到十毫秒级。对于延迟敏感的场景，需要优先在本可用区内路由，仅在本地不可用时才跨区。

模型缓存的存储成本：每个可用区都需要缓存模型权重，存储成本翻倍。对于 70B 模型，三个可用区的缓存需要 420GB SSD 空间。需要根据模型使用频率选择性缓存，低频模型仅在主可用区缓存。

降级策略的用户体验：从大模型降级到小模型时，输出质量会明显下降。必须在响应中标注降级状态，让用户知道当前使用的是降级服务。同时，降级不应静默进行，需要通知运维团队尽快恢复。

故障切换的数据一致性：推理服务本身无状态，故障切换不涉及数据一致性问题。但如果推理服务依赖外部状态（如对话历史缓存），切换后需要确保状态可用。建议将对话历史存储在独立的状态服务中，与推理服务解耦。

五、总结

LLM 服务的高可用设计需要针对推理延迟高、资源消耗大的特点进行专门优化。核心策略包括：基于队列深度的智能路由避免过载，模型本地缓存加速冷启动，跨可用区多活部署防止单点故障，小模型降级保障基本可用。落地建议：推理实例部署健康检查和心跳机制，路由层实时感知实例负载；模型权重缓存到本地 SSD，将冷启动时间从分钟级降到秒级；跨可用区部署至少两个推理集群，主集群故障时自动切换；降级策略需要明确告知用户，避免输出质量下降导致信任损失。