LLM 后端底座设计：从单点推理到弹性服务，大模型应用的生产级架构

一、LLM 推理服务的工程瓶颈：模型不是产品，服务才是

大模型应用的交付瓶颈往往不在模型本身，而在推理服务的基础设施。一个在离线评测中表现优秀的模型，部署为在线服务后可能面临三个致命问题：推理延迟波动大（P99 延迟是均值的 5 倍）、并发能力受限（GPU 显存决定了最大批处理大小）、成本不可控（按请求计费 vs 按实例计费的选择困境）。

更深层的问题是推理服务的弹性。白天高峰期需要 10 个 GPU 实例，夜间低谷期只需 2 个。但 GPU 实例的启动时间长达数分钟，无法像 CPU 实例那样秒级扩容。冷启动期间的服务降级不可避免。如何在成本和可用性之间找到平衡，是 LLM 后端架构的核心挑战。

二、LLM 推理服务的弹性架构

flowchart TD
    A[API 请求] --> B[请求路由层]
    B --> B1[模型版本路由: v1/v2/canary]
    B --> B2[优先级队列: 实时/批量]
    B --> B3[负载均衡: GPU 利用率感知]
    B1 --> C[推理引擎层]
    B2 --> C
    C --> C1[连续批处理 Continuous Batching]
    C --> C2[KV Cache 管理]
    C --> C3[推测解码 Speculative Decoding]
    C1 --> D[弹性伸缩层]
    D --> D1[预测性扩容: 基于队列深度]
    D --> D2[缩容保护: 最小实例数]
    D --> D3[模型预热: 预加载权重]

2.1 连续批处理与请求调度

# inference_scheduler.py — LLM 推理调度器
# 设计意图：实现连续批处理（Continuous Batching），
# 动态组批提升 GPU 利用率，同时保证请求级延迟 SLO

import time
import asyncio
from dataclasses import dataclass, field
from typing import Optional
from collections import deque

@dataclass
class InferenceRequest:
    request_id: str
    prompt: str
    max_tokens: int
    priority: int             # 0=实时, 1=批量
    created_at: float = field(default_factory=time.time)
    future: asyncio.Future = field(default=None)

@dataclass
class BatchConfig:
    max_batch_size: int = 32      # 最大批大小
    max_waiting_ms: int = 50      # 最大等待时间
    max_seq_length: int = 4096    # 最大序列长度

class InferenceScheduler:
    def __init__(self, config: BatchConfig):
        self.config = config
        self.realtime_queue: deque[InferenceRequest] = deque()
        self.batch_queue: deque[InferenceRequest] = deque()
        self.active_batch: list[InferenceRequest] = []
        self.running = False

    async def submit(self, request: InferenceRequest) -> str:
        """提交推理请求"""
        request.future = asyncio.get_event_loop().create_future()

        if request.priority == 0:
            self.realtime_queue.append(request)
        else:
            self.batch_queue.append(request)

        # 等待推理结果
        result = await request.future
        return result

    async def run(self, inference_engine):
        """调度主循环"""
        self.running = True

        while self.running:
            # 尝试组建批处理
            batch = self._form_batch()

            if not batch:
                # 无请求，短暂休眠
                await asyncio.sleep(0.01)
                continue

            # 执行推理
            try:
                results = await inference_engine.infer(batch)
                # 将结果分发到对应的 Future
                for req, result in zip(batch, results):
                    if not req.future.done():
                        req.future.set_result(result)
            except Exception as e:
                for req in batch:
                    if not req.future.done():
                        req.future.set_exception(e)

    def _form_batch(self) -> list[InferenceRequest]:
        """组建推理批次"""
        batch: list[InferenceRequest] = []

        # 优先填充实时请求
        while self.realtime_queue and len(batch) < self.config.max_batch_size:
            req = self.realtime_queue.popleft()
            batch.append(req)

        # 用批量请求填充剩余槽位
        while self.batch_queue and len(batch) < self.config.max_batch_size:
            req = self.batch_queue.popleft()
            batch.append(req)

        return batch

    def get_queue_depth(self) -> dict[str, int]:
        """获取队列深度，用于弹性伸缩决策"""
        return {
            "realtime": len(self.realtime_queue),
            "batch": len(self.batch_queue),
            "total": len(self.realtime_queue) + len(self.batch_queue),
        }

2.2 KV Cache 管理与上下文复用

# kv_cache_manager.py — KV Cache 生命周期管理
# 设计意图：管理推理引擎的 KV Cache，支持上下文复用，
# 减少重复计算，提升多轮对话场景的推理效率

import hashlib
from dataclasses import dataclass
from typing import Optional

@dataclass
class CacheEntry:
    cache_id: str
    prompt_hash: str
    token_count: int
    gpu_memory_mb: float
    created_at: float
    last_accessed_at: float
    access_count: int

class KVCacheManager:
    def __init__(self, max_memory_mb: float = 8192):
        self.max_memory_mb = max_memory_mb
        self.used_memory_mb = 0.0
        self.cache: dict[str, CacheEntry] = {}

    def get(self, prompt: str) -> Optional[str]:
        """查找已有的 KV Cache"""
        prompt_hash = self._hash_prompt(prompt)
        entry = self.cache.get(prompt_hash)

        if entry:
            entry.last_accessed_at = time.time()
            entry.access_count += 1
            return entry.cache_id

        return None

    def put(self, prompt: str, cache_id: str,
            token_count: int, gpu_memory_mb: float) -> bool:
        """存储 KV Cache"""
        # 检查是否有足够空间
        if self.used_memory_mb + gpu_memory_mb > self.max_memory_mb:
            # 驱逐最少使用的缓存
            self._evict(gpu_memory_mb)

        prompt_hash = self._hash_prompt(prompt)
        entry = CacheEntry(
            cache_id=cache_id,
            prompt_hash=prompt_hash,
            token_count=token_count,
            gpu_memory_mb=gpu_memory_mb,
            created_at=time.time(),
            last_accessed_at=time.time(),
            access_count=1,
        )

        self.cache[prompt_hash] = entry
        self.used_memory_mb += gpu_memory_mb
        return True

    def _evict(self, required_mb: float) -> None:
        """驱逐缓存释放空间"""
        # 按 LRU 策略排序
        sorted_entries = sorted(
            self.cache.values(),
            key=lambda e: e.last_accessed_at
        )

        freed = 0.0
        for entry in sorted_entries:
            if freed >= required_mb:
                break
            del self.cache[entry.prompt_hash]
            self.used_memory_mb -= entry.gpu_memory_mb
            freed += entry.gpu_memory_mb

    def _hash_prompt(self, prompt: str) -> str:
        """计算 prompt 的哈希值，用于缓存查找"""
        return hashlib.sha256(prompt.encode()).hexdigest()[:16]

    def get_stats(self) -> dict:
        return {
            "total_entries": len(self.cache),
            "used_memory_mb": self.used_memory_mb,
            "max_memory_mb": self.max_memory_mb,
            "utilization": self.used_memory_mb / self.max_memory_mb,
        }

三、弹性伸缩与成本控制

3.1 预测性扩容

# elastic_scaler.py — GPU 实例弹性伸缩器
# 设计意图：基于队列深度和推理延迟预测扩容需求，
# 提前预热实例避免冷启动导致的服务降级

import time
from dataclasses import dataclass
from typing import Optional

@dataclass
class ScaleConfig:
    min_instances: int = 2
    max_instances: int = 20
    target_latency_ms: float = 500.0    # 目标 P95 延迟
    scale_up_threshold: float = 0.7     # GPU 利用率 > 70% 扩容
    scale_down_threshold: float = 0.3   # GPU 利用率 < 30% 缩容
    cooldown_seconds: int = 300         # 扩缩容冷却期
    warmup_seconds: int = 120           # 实例预热时间

class ElasticScaler:
    def __init__(self, config: ScaleConfig):
        self.config = config
        self.current_instances = config.min_instances
        self.last_scale_time = 0.0
        self.instance_pool: dict[str, dict] = {}  # 实例ID → 状态

    def evaluate(
        self,
        queue_depth: int,
        gpu_utilization: float,
        p95_latency_ms: float
    ) -> Optional[str]:
        """评估是否需要扩缩容"""
        now = time.time()

        # 冷却期内不调整
        if now - self.last_scale_time < self.config.cooldown_seconds:
            return None

        # 扩容条件：队列积压 或 GPU 利用率过高 或 延迟超标
        if (queue_depth > self.current_instances * 10 or
            gpu_utilization > self.config.scale_up_threshold or
            p95_latency_ms > self.config.target_latency_ms * 1.5):

            if self.current_instances < self.config.max_instances:
                return self._scale_up(queue_depth)

        # 缩容条件：GPU 利用率低 且 队列无积压 且 延迟正常
        if (gpu_utilization < self.config.scale_down_threshold and
            queue_depth == 0 and
            p95_latency_ms < self.config.target_latency_ms):

            if self.current_instances > self.config.min_instances:
                return self._scale_down()

        return None

    def _scale_up(self, queue_depth: int) -> str:
        """执行扩容"""
        # 基于队列深度计算需要的实例数
        target = min(
            self.config.max_instances,
            max(self.current_instances + 1,
                (queue_depth // 10) + self.config.min_instances)
        )

        added = target - self.current_instances
        self.current_instances = target
        self.last_scale_time = time.time()

        # 启动新实例（预热期间不接收流量）
        for i in range(added):
            instance_id = f"gpu-{int(time.time())}-{i}"
            self.instance_pool[instance_id] = {
                "status": "warming",
                "started_at": time.time(),
                "ready_at": time.time() + self.config.warmup_seconds,
            }

        return f"扩容 {added} 个实例，目标 {target} 个"

    def _scale_down(self) -> str:
        """执行缩容"""
        target = max(self.config.min_instances, self.current_instances - 1)
        removed = self.current_instances - target
        self.current_instances = target
        self.last_scale_time = time.time()

        return f"缩容 {removed} 个实例，目标 {target} 个"

四、边界分析与架构权衡

连续批处理的延迟长尾：连续批处理提升了吞吐量，但长序列请求会拖慢整个批次的完成时间。一个 max_tokens=2048 的请求和 10 个 max_tokens=128 的请求同批处理时，短请求必须等待长请求完成。解决方案是实现迭代级调度——每个解码步骤后检查是否有请求完成，将完成的请求从批次中移除并加入新请求。

KV Cache 的显存竞争：KV Cache 占用的显存与序列长度成正比。长上下文请求的 Cache 可能占用数 GB 显存，挤压可用于批处理的显存空间。需要为 Cache 设置显存上限，超限时拒绝新请求或驱逐旧 Cache。但驱逐正在使用的 Cache 会导致重新计算，反而增加延迟。

弹性伸缩的冷启动成本：GPU 实例的启动和模型权重加载需要 1-3 分钟。在此期间，新实例无法服务请求。预测性扩容可以提前启动，但预测不准时会导致资源浪费。Spot 实例可以降低成本，但可能被随时回收。

多模型版本的路由复杂度：同时服务多个模型版本（如 v1 稳定版、v2 灰度版）时，每个版本需要独立的 GPU 实例池。版本间的流量比例调整需要精细的权重控制，且切换过程中可能出现请求路由不一致。

五、总结

LLM 后端底座的核心挑战在于将离线推理能力转化为在线服务能力。通过连续批处理提升 GPU 利用率，KV Cache 管理减少重复计算，弹性伸缩应对流量波动，三层机制协同构建生产级推理服务。关键实践包括：优先级队列分离实时和批量请求，迭代级调度消除长尾延迟，LRU 策略管理 KV Cache 显存，预测性扩容减少冷启动影响。但批处理延迟长尾、Cache 显存竞争、冷启动成本和多版本路由是需要权衡的边界条件。落地建议：从固定实例数开始验证推理引擎；逐步引入连续批处理和 Cache 管理；弹性伸缩基于队列深度和延迟双指标决策；多版本灰度发布从 5% 流量开始。

补充落地建议：围绕“LLM 后端底座设计：从单点推理到弹性服务，大模型应用的生产级架构”继续推进时，应把验证标准写成可执行清单，而不是停留在经验判断。性能类方案要给出基准数据，架构类方案要给出故障隔离方式，AI 类方案要给出输出质量和人工兜底策略。每一次迭代都应回答三个问题：收益是否可量化，失败是否可回滚，维护成本是否被团队接受。

如果短期资源有限，可以先保留最关键的观测指标，包括处理耗时、失败率、资源占用和人工介入次数。等这些指标稳定后，再扩展自动化能力。这样的节奏更慢，但风险更低，也更符合生产级技术文章强调的工程可验证性。