AI 推理服务的架构设计：从模型加载到弹性调度的全栈方案

一、推理服务上线后的真实挑战：延迟、吞吐与资源的三角博弈

将一个大语言模型从实验环境推向生产，远非"加载模型、暴露接口"这么简单。线上推理服务面临三重压力：首 Token 延迟（TTFT）要压到 200ms 以内，并发吞吐要支撑业务峰值，GPU 资源成本还要控制在预算之内。

某内容平台上线 RAG 服务后，白天流量低谷时单卡 A100 利用率不足 15%，夜间批量推理任务却排队等待 GPU 释放。流量突增时，请求排队导致 P99 延迟飙到 8 秒，用户直接放弃等待。这种"闲时浪费、忙时排队"的困境，本质上是架构设计没有将推理服务当作一等公民来对待。

AI 推理服务的架构设计，需要在延迟、吞吐、成本三者之间找到平衡点。这不是简单的参数调优，而是从请求调度、模型管理、资源编排三个维度进行系统性设计。

二、推理服务架构的核心机制：请求调度与模型生命周期

推理服务的架构核心，在于如何将用户请求高效地路由到合适的模型实例上，同时管理模型的生命周期以适配不同的负载模式。下图展示了一个生产级推理服务的架构全貌：

flowchart TB
    A[API Gateway] --> B[请求调度器]
    B --> C{路由策略}
    C -->|优先级路由| D[高优实例池]
    C -->|批量路由| E[批量推理池]
    C -->|冷启动路由| F[模型加载器]

    D --> G[vLLM / Triton 推理引擎]
    E --> G
    F --> H[模型仓库]
    H -->|动态加载| G

    G --> I[KV Cache 管理器]
    I -->|PagedAttention| J[GPU 显存池]

    B --> K[指标采集]
    K --> L[自动伸缩控制器]
    L -->|扩缩副本| M[K8s Deployment]

    style A fill:#f9f,stroke:#333
    style J fill:#bbf,stroke:#333
    style L fill:#bfb,stroke:#333

关键设计点有三个：一是请求调度器根据请求类型（在线/离线）和优先级做差异化路由；二是 KV Cache 管理器通过 PagedAttention 机制复用 GPU 显存，减少重复计算；三是自动伸缩控制器基于实时指标动态调整推理副本数。

三、生产级推理服务的代码实现

3.1 请求调度器——优先级感知的请求路由

"""
基于优先级的推理请求调度器
支持在线推理（低延迟）和批量推理（高吞吐）的差异化调度
"""
import asyncio
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional


class RequestPriority(Enum):
    REALTIME = 1    # 在线推理，延迟敏感
    BATCH = 2       # 批量推理，吞吐优先
    BACKGROUND = 3  # 后台任务，资源空闲时执行


@dataclass(order=True)
class InferenceRequest:
    priority: int = field(compare=True)
    submit_time: float = field(compare=True)
    request_id: str = field(compare=False)
    model_name: str = field(compare=False)
    prompt: str = field(compare=False)
    max_tokens: int = field(compare=False, default=512)
    callback: Optional[callable] = field(compare=False, default=None)


class PriorityScheduler:
    """优先级调度器：在线请求优先，批量请求填充空闲槽位"""

    def __init__(self, max_concurrent: int = 8):
        self.max_concurrent = max_concurrent
        self.current_running = 0
        self.pending_queue: list[InferenceRequest] = []
        self._lock = asyncio.Lock()
        self._condition = asyncio.Condition(self._lock)

    async def submit(self, request: InferenceRequest) -> str:
        """提交推理请求到调度队列"""
        async with self._condition:
            self.pending_queue.append(request)
            # 按优先级排序，同优先级按提交时间排序
            self.pending_queue.sort()
            self._condition.notify()
        return request.request_id

    async def dispatch(self, engine_pool: dict) -> Optional[InferenceRequest]:
        """从队列中取出请求并分配到推理引擎"""
        async with self._condition:
            while not self.pending_queue and self.current_running >= self.max_concurrent:
                await self._condition.wait()

            if not self.pending_queue:
                return None

            request = self.pending_queue.pop(0)
            self.current_running += 1
            return request

    async def complete(self, request_id: str):
        """请求完成，释放并发槽位"""
        async with self._condition:
            self.current_running -= 1
            self._condition.notify_all()

3.2 模型生命周期管理——动态加载与卸载

"""
模型生命周期管理器
支持多模型共存、LRU 淘汰、显存水位控制
"""
import threading
from collections import OrderedDict
from typing import Optional


class ModelLifecycleManager:
    """管理 GPU 上的模型加载与卸载，基于 LRU 策略释放显存"""

    def __init__(self, max_vram_gb: float = 80.0):
        self.max_vram_gb = max_vram_gb
        self.used_vram_gb = 0.0
        # 有序字典实现 LRU：最近访问的模型移到末尾
        self.loaded_models: OrderedDict[str, dict] = OrderedDict()
        self._lock = threading.RLock()
        # 模型元数据注册表（模型名 → 显存占用、加载函数）
        self._registry: dict[str, dict] = {}

    def register_model(self, name: str, vram_gb: float, loader: callable):
        """注册模型元信息"""
        self._registry[name] = {
            "vram_gb": vram_gb,
            "loader": loader,
        }

    def get_model(self, name: str) -> Optional[object]:
        """获取模型实例，不存在则加载，显存不足则 LRU 淘汰"""
        with self._lock:
            # 命中已加载模型，移到 LRU 末尾
            if name in self.loaded_models:
                self.loaded_models.move_to_end(name)
                return self.loaded_models[name]["instance"]

            # 模型未注册
            if name not in self._registry:
                raise ValueError(f"模型 {name} 未注册")

            model_meta = self._registry[name]
            required_vram = model_meta["vram_gb"]

            # 显存不足时，按 LRU 策略淘汰旧模型
            while self.used_vram_gb + required_vram > self.max_vram_gb:
                if not self.loaded_models:
                    raise RuntimeError(
                        f"显存不足：需要 {required_vram}GB，"
                        f"总量 {self.max_vram_gb}GB"
                    )
                # 淘汰最久未使用的模型
                evict_name, evict_meta = self.loaded_models.popitem(last=False)
                self.used_vram_gb -= evict_meta["vram_gb"]
                # 执行模型卸载回调
                evict_meta.get("unloader", lambda: None)()

            # 加载新模型
            instance = model_meta["loader"]()
            self.loaded_models[name] = {
                "instance": instance,
                "vram_gb": required_vram,
            }
            self.used_vram_gb += required_vram
            return instance

3.3 自动伸缩控制器——基于指标弹性调整副本

"""
基于 Prometheus 指标的推理服务自动伸缩控制器
核心指标：平均队列深度 + P99 延迟
"""
import time
import logging
import requests


class InferenceAutoscaler:
    """根据实时指标动态调整推理服务副本数"""

    def __init__(
        self,
        prometheus_url: str,
        deployment_name: str,
        namespace: str = "ai-inference",
        min_replicas: int = 1,
        max_replicas: int = 10,
        target_queue_depth: float = 5.0,
        target_p99_ms: float = 2000.0,
        cooldown_seconds: int = 60,
    ):
        self.prometheus_url = prometheus_url
        self.deployment_name = deployment_name
        self.namespace = namespace
        self.min_replicas = min_replicas
        self.max_replicas = max_replicas
        self.target_queue_depth = target_queue_depth
        self.target_p99_ms = target_p99_ms
        self.cooldown_seconds = cooldown_seconds
        self.last_scale_time = 0

    def _query_metric(self, query: str) -> float:
        """从 Prometheus 查询指标值"""
        try:
            resp = requests.get(
                f"{self.prometheus_url}/api/v1/query",
                params={"query": query},
                timeout=5,
            )
            result = resp.json()
            if result["status"] == "success" and result["data"]["result"]:
                return float(result["data"]["result"][0]["value"][1])
        except Exception as e:
            logging.warning(f"指标查询失败: {e}")
        return 0.0

    def compute_desired_replicas(self, current_replicas: int) -> int:
        """计算目标副本数"""
        now = time.time()
        # 冷却期内不调整
        if now - self.last_scale_time < self.cooldown_seconds:
            return current_replicas

        queue_depth = self._query_metric(
            f'inference_queue_depth{{deployment="{self.deployment_name}"}}'
        )
        p99_latency = self._query_metric(
            f'histogram_quantile(0.99, '
            f'rate(inference_request_duration_bucket{{deployment="{self.deployment_name}"}}[1m]))'
        ) * 1000  # 转换为毫秒

        # 基于队列深度的扩缩策略
        queue_ratio = queue_depth / self.target_queue_depth
        latency_ratio = p99_latency / self.target_p99_ms if p99_latency > 0 else 1.0

        # 取两个指标中更大的压力比
        pressure_ratio = max(queue_ratio, latency_ratio)
        desired = max(self.min_replicas,
                      min(self.max_replicas,
                          int(current_replicas * pressure_ratio)))

        if desired != current_replicas:
            self.last_scale_time = now
            logging.info(
                f"伸缩决策: {current_replicas} → {desired}, "
                f"队列深度={queue_depth:.1f}, P99={p99_latency:.0f}ms"
            )
        return desired

四、架构设计的隐性代价与适用边界

推理服务架构的每一层设计都有代价。

优先级调度的代价是批量任务的饥饿风险。当在线请求持续涌入时，低优先级的批量任务可能长时间得不到执行。生产中需要引入"防饥饿"机制：为低优先级任务预留最低执行槽位，或设置最长等待时间超时后自动提升优先级。

动态模型加载的代价是冷启动延迟。一个大模型从磁盘加载到 GPU 显存通常需要 10-30 秒。对于延迟敏感的在线服务，冷启动不可接受。解决方案是模型预热：在流量高峰前预加载，或保持至少一个实例常驻热备。

自动伸缩的代价是资源浪费窗口。从触发扩容到新副本就绪，通常需要 30-60 秒（含模型加载时间）。在这段时间内，现有实例可能已经被压垮。解决方案是预测性扩容：基于历史流量模式提前扩容，而非等到指标超标才响应。

适用边界：这套架构适用于在线推理为主、批量推理为辅的混合场景。如果业务以离线批量推理为主（如大规模 Embedding 生成），则应采用 Job 队列模式而非常驻服务模式，资源利用率更高。

五、总结

AI 推理服务的架构设计，核心是在延迟、吞吐、成本之间找到业务最优解。优先级调度解决请求差异化问题，模型生命周期管理解决多模型共存问题，自动伸缩解决资源利用率问题。落地时需要关注三个关键点：冷启动延迟必须通过预热或常驻热备来消除；防饥饿机制是优先级调度的必要补充；自动伸缩应结合预测性扩容，而非纯被动响应。架构没有银弹，只有持续迭代和精细化运营。