AI 工程化演进：从单模型部署到平台化服务

AI 工程化演进：从单模型部署到平台化服务

一、单模型部署为什么撑不起规模化推理需求

早期的 AI 接入方式很直接：选一个模型，写一个推理脚本，部署到一台 GPU 服务器，请求通过 HTTP 到达服务返回结果。这个流程在验证阶段够用，但进入规模化阶段后问题暴露：新模型上线需要手动配置服务器，模型切换需要重建服务，GPU 资源分配靠人工协调，监控告警只覆盖单点。

基础设施不需要漂亮话，手动部署就是不可重复的部署，靠人协调资源就是不可靠的协调。规模化推理需要的是平台化能力：模型一键上线、资源自动分配、服务自治管理、可观测性全面覆盖。从单模型到平台的演进，不是"多做几个模型就行"，而是工程体系从手工作坊到工业化的质变。

二、从单模型到平台的演进路径

AI 接理的工程化演进分为四个阶段，每个阶段解决一类核心问题。

flowchart TB subgraph S1["阶段一: 单模型部署"] direction LR S1M["单个模型脚本"] S1D["单台 GPU 服务器"] S1H["HTTP 直接调用"] end subgraph S2["阶段二: 多模型服务化"] direction LR S2G["推理网关: 路由分发"] S2C["容器化: Docker + K8s"] S2R["资源池化: GPU 共享调度"] end subgraph S3["阶段三: 平台化自治"] direction LR S3A["自动伸缩: HPA + GPU 调度"] S3O["可观测性: 全链路追踪"] S3P["流量管理: 灰度 + 回放"] end subgraph S4["阶段四: 智能化运营"] direction LR S4I["智能调度: 成本最优"] S4S["自治运维: 自愈 + 预警"] S4E["效率度量: 成本/质量指标"] end S1 -->|"模型数量增长"| S2 -->|"运维复杂度攀升"| S3 -->|"运营成本优化"| S4 style S1 fill:#fce4ec style S2 fill:#fff3e0 style S3 fill:#e3f2fd style S4 fill:#e8f5e9

各阶段的关键能力差距：

三、平台化服务的架构实现

平台化的核心是四个子系统：模型注册中心、推理调度器、可观测性平台、自治运维引擎。

模型注册中心——管理模型版本和部署配置：

# model_registry.py — 模型注册中心 import json import logging from datetime import datetime from typing import List, Optional from dataclasses import dataclass, field logging.basicConfig(level=logging.INFO) logger = logging.getLogger("model-registry") @dataclass class ModelVersion: """模型版本记录""" model_name: str version: str framework: str # "pytorch", "onnx", "triton" image_tag: str # 推理镜像版本 weights_path: str # 模型权重存储路径 config_path: str # 推理配置路径 gpu_type: str # "nvidia-a100", "nvidia-l40" gpu_count: int # 单 Pod GPU 数 min_replicas: int # 最小副本数 max_replicas: int # 最大副本数 status: str = "registered" # registered / staging / production / retired registered_at: str = field(default_factory=lambda: datetime.now().isoformat()) health_check_path: str = "/health" @dataclass class DeploymentRecord: """部署记录""" model_name: str version: str namespace: str replicas: int deployed_at: str deploy_type: str # "fresh", "upgrade", "rollback" previous_version: Optional[str] = None status: str = "deploying" # deploying / running / failed class ModelRegistry: """模型注册中心，管理模型生命周期""" def __init__(self, storage_path: str = "/data/registry"): self.storage_path = storage_path self.models: dict = {} # model_name -> ModelVersion 列表 self.deployments: List[DeploymentRecord] = [] self._load_registry() def register_model(self, version: ModelVersion) -> bool: """注册新模型版本""" key = version.model_name if key not in self.models: self.models[key] = [] # 检查版本是否已存在 for v in self.models[key]: if v.version == version.version: logger.warning(f"模型版本已存在: {key}@{version.version}") return False self.models[key].append(version) self._save_registry() logger.info(f"模型注册成功: {key}@{version.version}, framework={version.framework}") return True def promote_to_staging(self, model_name: str, version: str) -> bool: """将模型版本推进到预发环境""" mv = self._find_version(model_name, version) if mv is None: logger.error(f"版本不存在: {model_name}@{version}") return False if mv.status != "registered": logger.warning(f"版本状态不允许推进: {mv.status}") return False mv.status = "staging" self._save_registry() logger.info(f"模型推进到预发: {model_name}@{version}") return True def promote_to_production(self, model_name: str, version: str) -> bool: """将模型版本推进到生产环境（灰度发布）""" mv = self._find_version(model_name, version) if mv is None: return False # 查找当前生产版本 current_prod = self._find_production_version(model_name) if current_prod: logger.info(f"替换当前生产版本: {current_prod.version} -> {version}") mv.status = "production" if current_prod: current_prod.status = "retired" # 记录部署历史 self.deployments.append(DeploymentRecord( model_name=model_name, version=version, namespace="ai-inference", replicas=mv.min_replicas, deployed_at=datetime.now().isoformat(), deploy_type="upgrade", previous_version=current_prod.version if current_prod else None, )) self._save_registry() logger.info(f"模型上线: {model_name}@{version}") return True def rollback(self, model_name: str) -> bool: """回滚到上一个生产版本""" current_prod = self._find_production_version(model_name) if current_prod is None: logger.error(f"无可回滚的生产版本: {model_name}") return False # 找到上一个版本 retired_versions = [ v for v in self.models.get(model_name, []) if v.status == "retired" ] if not retired_versions: logger.error(f"无可用的回滚版本") return False # 回滚到最新的 retired 版本 rollback_version = retired_versions[-1] current_prod.status = "retired" rollback_version.status = "production" self.deployments.append(DeploymentRecord( model_name=model_name, version=rollback_version.version, namespace="ai-inference", replicas=rollback_version.min_replicas, deployed_at=datetime.now().isoformat(), deploy_type="rollback", previous_version=current_prod.version, )) self._save_registry() logger.info(f"回滚完成: {model_name} {current_prod.version} -> {rollback_version.version}") return True def _find_version(self, model_name: str, version: str) -> Optional[ModelVersion]: """查找指定版本""" for v in self.models.get(model_name, []): if v.version == version: return v return None def _find_production_version(self, model_name: str) -> Optional[ModelVersion]: """查找当前生产版本""" for v in self.models.get(model_name, []): if v.status == "production": return v return None def _load_registry(self): """从存储加载注册数据""" filepath = f"{self.storage_path}/registry.json" try: with open(filepath, "r", encoding="utf-8") as f: data = json.load(f) # 反序列化模型版本列表 for name, versions in data.get("models", {}).items(): self.models[name] = [ ModelVersion(**v) for v in versions ] logger.info(f"加载注册数据: {len(self.models)} 个模型") except (FileNotFoundError, json.JSONDecodeError): logger.info("注册数据不存在，初始化空注册中心") def _save_registry(self): """持久化注册数据""" filepath = f"{self.storage_path}/registry.json" data = { "models": { name: [v.__dict__ for v in versions] for name, versions in self.models.items() }, "deployments": [d.__dict__ for d in self.deployments], } try: with open(filepath, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) except IOError as e: logger.error(f"注册数据保存失败: {e}")

推理调度器——根据模型配置生成部署资源：

# inference_scheduler.py — 推理调度器 import json import logging from model_registry import ModelRegistry, ModelVersion logging.basicConfig(level=logging.INFO) logger = logging.getLogger("inference-scheduler") class InferenceScheduler: """推理调度器：根据模型注册信息生成 K8s 部署配置""" def __init__(self, registry: ModelRegistry): self.registry = registry def schedule_deployment(self, model_name: str, version: str) -> dict: """生成推理服务的 Deployment 配置""" mv = self.registry._find_version(model_name, version) if mv is None: logger.error(f"调度失败: 模型版本不存在 {model_name}@{version}") return {} # 根据模型配置生成 Deployment YAML deployment = { "apiVersion": "apps/v1", "kind": "Deployment", "metadata": { "name": f"{model_name}-inference", "namespace": "ai-inference", "labels": { "app": f"{model_name}-inference", "model": model_name, "version": version, }, }, "spec": { "replicas": mv.min_replicas, "selector": { "matchLabels": { "app": f"{model_name}-inference", "model": model_name, }, }, "template": { "metadata": { "labels": { "app": f"{model_name}-inference", "model": model_name, "version": version, }, }, "spec": { "affinity": self._generate_affinity(mv), "containers": [{ "name": "inference-server", "image": f"registry.internal.com/ai-inference/{model_name}:{mv.image_tag}", "ports": [{"containerPort": 8000}], "env": [ {"name": "MODEL_DIR", "value": f"/opt/models/{version}"}, {"name": "MODEL_NAME", "value": model_name}, ], "resources": self._generate_resources(mv), "livenessProbe": { "httpGet": {"path": mv.health_check_path, "port": 8000}, "initialDelaySeconds": 120, "periodSeconds": 30, }, }], }, }, }, } logger.info(f"调度配置生成: {model_name}@{version}") return deployment def _generate_affinity(self, mv: ModelVersion) -> dict: """生成 Affinity 配置""" return { "nodeAffinity": { "requiredDuringSchedulingIgnoredDuringExecution": { "nodeSelectorTerms": [{ "matchExpressions": [{ "key": "gpu-type", "operator": "In", "values": [mv.gpu_type], }], }], }, }, } def _generate_resources(self, mv: ModelVersion) -> dict: """生成资源请求配置""" return { "limits": { "nvidia.com/gpu": mv.gpu_count, "cpu": "4", "memory": "16Gi", }, "requests": { "nvidia.com/gpu": mv.gpu_count, "cpu": "4", "memory": "16Gi", }, } def generate_hpa(self, model_name: str, version: str) -> dict: """生成 HPA 自动伸缩配置""" mv = self.registry._find_version(model_name, version) if mv is None: return {} return { "apiVersion": "autoscaling/v2", "kind": "HorizontalPodAutoscaler", "metadata": { "name": f"{model_name}-hpa", "namespace": "ai-inference", }, "spec": { "scaleTargetRef": { "apiVersion": "apps/v1", "kind": "Deployment", "name": f"{model_name}-inference", }, "minReplicas": mv.min_replicas, "maxReplicas": mv.max_replicas, "metrics": [{ "type": "Resource", "resource": { "name": "nvidia.com/gpu", "target": { "type": "Utilization", "averageUtilization": 70, }, }, }], }, }

自治运维引擎——健康检查与自动恢复：

# autonomous_ops.py — 自治运维引擎 import logging import time from typing import Dict, List logging.basicConfig(level=logging.INFO) logger = logging.getLogger("autonomous-ops") class AutonomousOpsEngine: """自治运维引擎：检测异常并触发恢复动作""" def __init__(self, check_interval: int = 30): self.check_interval = check_interval self.health_status: Dict[str, str] = {} self.recovery_rules: List[dict] = [] def register_recovery_rule(self, rule: dict): """注册恢复规则""" self.recovery_rules.append(rule) logger.info(f"注册恢复规则: {rule['name']}") def check_and_recover(self, model_name: str, deployment_name: str): """检查服务健康状态，触发恢复""" # 模拟健康检查（实际调用 kubectl 或 API Server） health = self._check_health(deployment_name) self.health_status[model_name] = health if health != "healthy": logger.warning(f"服务异常: {model_name} status={health}") # 匹配恢复规则 for rule in self.recovery_rules: if rule["condition"](health): action = rule["action"] logger.info(f"触发恢复动作: {rule['name']} -> {action}") self._execute_recovery(model_name, action) def _check_health(self, deployment_name: str) -> str: """检查 Deployment 健康状态""" # 实际实现：调用 kubectl rollout status 或 API Server # 此处返回模拟状态 return "healthy" def _execute_recovery(self, model_name: str, action: str): """执行恢复动作""" actions = { "restart_pod": f"kubectl rollout restart deployment {model_name}-inference -n ai-inference", "scale_up": f"kubectl scale deployment {model_name}-inference --replicas=4 -n ai-inference", "rollback": f"kubectl rollout undo deployment {model_name}-inference -n ai-inference", } cmd = actions.get(action, "") if cmd: logger.info(f"执行恢复命令: {cmd}") # 实际实现：subprocess.run(cmd, ...) else: logger.error(f"未知恢复动作: {action}") # 注册恢复规则 engine = AutonomousOpsEngine() engine.register_recovery_rule({ "name": "Pod 全部不可用", "condition": lambda status: status == "unhealthy", "action": "rollback", }) engine.register_recovery_rule({ "name": "Pod 部分不可用", "condition": lambda status: status == "degraded", "action": "scale_up", })

四、平台化演进的节奏与取舍

阶段一到阶段二的跳跃。单模型部署到多模型服务化，核心变化是引入推理网关和容器化。网关解决多模型路由问题，容器化解决部署标准化问题。这个跳跃的技术成本不高——一个 FastAPI 网关加几个 Dockerfile——但组织成本显著：团队需要统一镜像构建规范、配置管理流程、部署审批机制。过早跳跃会导致规范执行不到位，反而增加混乱。

阶段二到阶段三的关键门槛。服务化到平台化的核心变化是引入自动伸缩和全链路可观测性。自动伸缩依赖准确的 GPU 利用率指标，全链路追踪依赖 Service Mesh 或自定义中间件。这两个能力的实施前提是：指标采集稳定运行、告警规则验证充分、伸缩策略经过压力测试。没有这些前提，自动伸缩可能误触发，可观测性数据不可信。

阶段三到阶段四的距离。平台化到智能化运营是长期演进。智能调度需要大量运行数据训练成本模型，自愈引擎需要积累足够的故障模式。不急于跳到阶段四，先在阶段三积累数据和经验。成本优化的第一步是度量——没有准确的成本数据，优化无从谈起。

各阶段的停顿时机。不是所有团队都需要走到阶段四。如果模型数量不超过 10 个、日推理量不超过百万级，阶段二足够支撑。强行推进到更高阶段，投入产出比可能倒挂。演进节奏根据业务规模和团队容量决定，而不是技术理想。

五、总结

AI 推理的工程化演进从单模型部署到平台化服务，分四个阶段：单模型部署解决验证问题，多模型服务化解决规模化路由和容器化，平台化自治解决自动伸缩和全链路可观测性，智能化运营解决成本优化和自愈闭环。每个阶段解决一类核心问题，跳跃的前提是前一阶段的基础能力稳定运行。模型注册中心管理版本生命周期，推理调度器根据配置生成部署资源，自治运维引擎检测异常并触发恢复。演进节奏根据业务规模决定，不是所有团队都需要走到最远阶段。平台化的本质是让部署、运维、扩展从手动操作变成可重复的工程流程，让基础设施具备自治能力。