云原生 AI 模型版本管理：从模型注册到灰度发布的工程实践

一、模型版本的"混乱地带"：10 个模型文件，没人知道哪个在线上

AI 模型从训练到上线涉及多个版本迭代，但版本管理远比软件发布复杂。某推荐系统团队同时维护 10 个模型文件（A/B 测试 3 个、灰度 2 个、生产 1 个、回滚备份 4 个），但没有统一的版本注册中心。线上出问题时，排查 30 分钟才确认当前运行的是哪个版本。更严重的是，模型文件名包含时间戳和超参数摘要，人工无法快速判断版本间的差异和演进关系。

云原生环境下的模型版本管理需要解决三个核心问题：版本注册与元数据管理、灰度发布与流量控制、回滚与版本追溯。这不是简单的文件存储问题，而是模型全生命周期的治理问题。

二、云原生模型版本管理的架构

flowchart TB
    subgraph 注册层["模型注册中心"]
        direction TB
        R1[模型仓库<br/>MLflow / DVC<br/>版本号 + 元数据]
        R2[模型签名<br/>输入/输出 Schema<br/>兼容性校验]
        R3[模型指标<br/>精度/延迟/吞吐<br/>质量门禁]
    end

    subgraph 部署层["灰度发布引擎"]
        direction TB
        D1[Canary 发布<br/>5% → 20% → 100%<br/>自动指标对比]
        D2[流量镜像<br/>影子流量验证<br/>不影响线上]
        D3[A/B 测试<br/>用户分桶<br/>统计显著性]
    end

    subgraph 运维层["版本运维"]
        direction TB
        O1[一键回滚<br/>秒级切换<br/>版本追溯]
        O2[模型监控<br/>数据漂移检测<br/>性能退化告警]
        O3[生命周期管理<br/>过期版本清理<br/>存储成本控制]
    end

    R1 --> D1
    R2 --> D1
    R3 --> D1
    D1 --> O1
    D2 --> O2
    D3 --> O2
    O1 --> O3

    style 注册层 fill:#eef,stroke:#333
    style 部署层 fill:#fee,stroke:#333
    style 运维层 fill:#efe,stroke:#333

三、云原生模型版本管理的代码实现

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from enum import Enum
from datetime import datetime
import hashlib


class ModelStage(Enum):
    DEVELOPMENT = "development"
    STAGING = "staging"
    PRODUCTION = "production"
    ARCHIVED = "archived"


class DeploymentStrategy(Enum):
    ROLLOUT = "rollout"           # 全量发布
    CANARY = "canary"             # 金丝雀发布
    BLUE_GREEN = "blue_green"     # 蓝绿部署
    A_B_TEST = "a_b_test"         # A/B 测试


@dataclass
class ModelSignature:
    """模型签名：定义输入输出 Schema"""
    input_schema: Dict            # {"features": [{"name": "x", "type": "float"}]}
    output_schema: Dict           # {"prediction": [{"name": "y", "type": "float"}]}
    version: str = "1.0"

    def is_compatible(self, other: "ModelSignature") -> bool:
        """检查与另一个签名是否兼容"""
        # 输入字段必须包含旧版本的所有字段
        old_inputs = {f["name"] for f in self.input_schema.get("features", [])}
        new_inputs = {f["name"] for f in other.input_schema.get("features", [])}
        if not old_inputs.issubset(new_inputs):
            return False
        return True


@dataclass
class ModelMetrics:
    """模型质量指标"""
    accuracy: Optional[float] = None
    f1_score: Optional[float] = None
    latency_p99_ms: Optional[float] = None
    throughput_qps: Optional[float] = None
    custom_metrics: Dict[str, float] = field(default_factory=dict)


@dataclass
class ModelVersion:
    """模型版本"""
    model_name: str
    version: str                  # 语义化版本号
    stage: ModelStage = ModelStage.DEVELOPMENT
    artifact_path: str = ""       # 模型文件存储路径
    signature: Optional[ModelSignature] = None
    metrics: Optional[ModelMetrics] = None
    training_config: Dict = field(default_factory=dict)
    created_at: datetime = field(default_factory=datetime.now)
    created_by: str = ""
    description: str = ""
    checksum: str = ""


class ModelRegistry:
    """
    模型注册中心：管理模型版本的全生命周期
    """

    def __init__(self):
        self._models: Dict[str, List[ModelVersion]] = {}

    def register(self, model: ModelVersion) -> str:
        """注册模型版本"""
        if model.model_name not in self._models:
            self._models[model.model_name] = []

        # 检查版本号唯一性
        existing_versions = [
            v.version for v in self._models[model.model_name]
        ]
        if model.version in existing_versions:
            raise ValueError(
                f"版本 {model.version} 已存在: {model.model_name}"
            )

        # 计算校验和
        model.checksum = self._compute_checksum(model)

        self._models[model.model_name].append(model)
        return f"{model.model_name}/{model.version}"

    def promote(self, model_name: str, version: str,
                target_stage: ModelStage) -> bool:
        """提升模型阶段（如 staging → production）"""
        model = self._find_version(model_name, version)
        if not model:
            return False

        # 阶段提升规则校验
        if target_stage == ModelStage.PRODUCTION:
            if model.stage != ModelStage.STAGING:
                return False
            if not model.metrics or not model.signature:
                return False
            # 质量门禁：精度必须高于阈值
            if model.metrics.accuracy and model.metrics.accuracy < 0.85:
                return False

        # 将当前 production 版本降级为 archived
        if target_stage == ModelStage.PRODUCTION:
            for v in self._models.get(model_name, []):
                if v.stage == ModelStage.PRODUCTION:
                    v.stage = ModelStage.ARCHIVED

        model.stage = target_stage
        return True

    def get_production_version(self, model_name: str) -> Optional[ModelVersion]:
        """获取当前生产版本"""
        for v in self._models.get(model_name, []):
            if v.stage == ModelStage.PRODUCTION:
                return v
        return None

    def get_version_history(self, model_name: str) -> List[Dict]:
        """获取版本演进历史"""
        versions = self._models.get(model_name, [])
        return [
            {
                "version": v.version,
                "stage": v.stage.value,
                "accuracy": v.metrics.accuracy if v.metrics else None,
                "created_at": v.created_at.isoformat(),
            }
            for v in sorted(versions, key=lambda v: v.created_at)
        ]

    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

    @staticmethod
    def _compute_checksum(model: ModelVersion) -> str:
        """计算模型校验和"""
        content = f"{model.model_name}:{model.version}:{model.artifact_path}"
        return hashlib.md5(content.encode()).hexdigest()[:12]


# ============ 灰度发布引擎 ============

@dataclass
class CanaryConfig:
    """金丝雀发布配置"""
    model_name: str
    new_version: str
    initial_percentage: float = 5.0     # 初始流量比例
    step_percentage: float = 15.0       # 每步增加比例
    max_percentage: float = 100.0
    evaluation_interval_sec: int = 300   # 评估间隔（秒）
    rollback_threshold: float = 0.05    # 指标退化阈值


class CanaryDeployer:
    """
    金丝雀发布引擎：逐步放量并自动评估
    """

    def __init__(self, registry: ModelRegistry):
        self._registry = registry
        self._deployments: Dict[str, Dict] = {}

    def start_canary(self, config: CanaryConfig) -> Dict:
        """启动金丝雀发布"""
        new_model = self._registry._find_version(
            config.model_name, config.new_version
        )
        if not new_model:
            return {"status": "error", "message": "版本不存在"}

        current_prod = self._registry.get_production_version(config.model_name)
        if not current_prod:
            return {"status": "error", "message": "无当前生产版本"}

        # 签名兼容性检查
        if (current_prod.signature and new_model.signature and
                not current_prod.signature.is_compatible(new_model.signature)):
            return {"status": "error", "message": "签名不兼容"}

        deployment = {
            "model_name": config.model_name,
            "old_version": current_prod.version,
            "new_version": config.new_version,
            "current_percentage": config.initial_percentage,
            "step_percentage": config.step_percentage,
            "status": "running",
            "started_at": datetime.now().isoformat(),
            "metrics_history": [],
        }
        self._deployments[config.model_name] = deployment

        return {
            "status": "started",
            "canary_percentage": config.initial_percentage,
            "old_version": current_prod.version,
            "new_version": config.new_version,
        }

    def evaluate_and_advance(self, model_name: str,
                              current_metrics: ModelMetrics) -> Dict:
        """评估当前指标并决定是否推进"""
        deployment = self._deployments.get(model_name)
        if not deployment or deployment["status"] != "running":
            return {"status": "not_running"}

        # 获取基线指标（旧版本）
        old_model = self._registry._find_version(
            model_name, deployment["old_version"]
        )
        baseline_accuracy = (
            old_model.metrics.accuracy if old_model and old_model.metrics else None
        )

        # 指标对比
        should_rollback = False
        if baseline_accuracy and current_metrics.accuracy:
            degradation = baseline_accuracy - current_metrics.accuracy
            if degradation > 0.05:  # 退化超过 5%
                should_rollback = True

        if should_rollback:
            deployment["status"] = "rolled_back"
            return {
                "status": "rolled_back",
                "reason": f"精度退化 {degradation:.2%}",
                "current_percentage": deployment["current_percentage"],
            }

        # 推进到下一步
        new_percentage = min(
            deployment["current_percentage"] + deployment["step_percentage"],
            100.0
        )
        deployment["current_percentage"] = new_percentage

        if new_percentage >= 100.0:
            # 全量发布完成，提升新版本为 production
            self._registry.promote(
                model_name, deployment["new_version"],
                ModelStage.PRODUCTION
            )
            deployment["status"] = "completed"
            return {
                "status": "completed",
                "new_version": deployment["new_version"],
            }

        return {
            "status": "advancing",
            "new_percentage": new_percentage,
        }

    def rollback(self, model_name: str) -> Dict:
        """紧急回滚"""
        deployment = self._deployments.get(model_name)
        if not deployment:
            return {"status": "not_found"}

        deployment["status"] = "rolled_back"
        deployment["current_percentage"] = 0

        return {
            "status": "rolled_back",
            "active_version": deployment["old_version"],
        }


# ============ Kubernetes 部署配置生成 ============

class K8sModelDeployer:
    """
    K8s 部署配置生成器：将模型版本映射为 K8s 资源
    """

    def generate_canary_manifest(self, model_name: str,
                                  deployment: Dict) -> Dict:
        """生成金丝雀发布的 K8s Manifest"""
        return {
            "apiVersion": "apps/v1",
            "kind": "Deployment",
            "metadata": {
                "name": f"{model_name}-canary",
                "labels": {
                    "app": model_name,
                    "track": "canary",
                    "version": deployment["new_version"],
                },
            },
            "spec": {
                "replicas": self._calculate_replicas(
                    deployment["current_percentage"]
                ),
                "selector": {
                    "matchLabels": {"app": model_name, "track": "canary"}
                },
                "template": {
                    "metadata": {
                        "labels": {
                            "app": model_name,
                            "track": "canary",
                            "version": deployment["new_version"],
                        }
                    },
                    "spec": {
                        "containers": [{
                            "name": "model-server",
                            "image": f"registry/{model_name}:{deployment['new_version']}",
                            "resources": {
                                "requests": {"nvidia.com/gpu": "1"},
                                "limits": {"nvidia.com/gpu": "1"},
                            },
                            "env": [{
                                "name": "MODEL_VERSION",
                                "value": deployment["new_version"],
                            }],
                        }]
                    },
                },
            },
        }

    @staticmethod
    def _calculate_replicas(percentage: float,
                             total_replicas: int = 10) -> int:
        """根据流量比例计算副本数"""
        return max(1, int(total_replicas * percentage / 100))

    def generate_istio_virtual_service(self, model_name: str,
                                        deployment: Dict) -> Dict:
        """生成 Istio VirtualService 用于流量切分"""
        canary_weight = int(deployment["current_percentage"])
        stable_weight = 100 - canary_weight

        return {
            "apiVersion": "networking.istio.io/v1beta1",
            "kind": "VirtualService",
            "metadata": {"name": f"{model_name}-vs"},
            "spec": {
                "hosts": [model_name],
                "http": [{
                    "route": [
                        {
                            "destination": {
                                "host": model_name,
                                "subset": "stable",
                            },
                            "weight": stable_weight,
                        },
                        {
                            "destination": {
                                "host": model_name,
                                "subset": "canary",
                            },
                            "weight": canary_weight,
                        },
                    ],
                }],
            },
        }

四、云原生模型版本管理的 Trade-offs

签名兼容性与模型演进的矛盾。严格的签名兼容性检查可以防止不兼容模型上线，但也限制了模型的演进自由度。添加新输入特征是向后兼容的，但删除或重命名特征则不兼容。建议采用"只增不删"的演进策略，废弃字段通过标记而非删除来处理。

金丝雀评估的统计显著性。5% 流量下的指标波动可能由随机性引起，而非模型质量差异。需要足够的评估时间（通常至少 30 分钟）和样本量才能做出可靠判断。紧急修复场景下，团队可能跳过充分评估直接全量发布，增加了风险。

存储成本与版本保留策略。每个模型版本可能占用数 GB 存储空间。保留所有历史版本的成本随时间线性增长。建议按阶段设置保留策略：development 保留最近 5 个，staging 保留最近 3 个，archived 保留最近 10 个。

多集群同步的一致性。跨集群部署时，模型版本需要同步到多个 K8s 集群。对象存储（如 S3/MinIO）作为模型仓库可以解决跨集群访问问题，但缓存一致性需要额外处理——集群 A 更新模型后，集群 B 的本地缓存可能仍是旧版本。

五、总结

云原生 AI 模型版本管理覆盖从注册到灰度发布再到回滚的全生命周期。模型注册中心管理版本号、签名和指标，灰度发布引擎通过逐步放量 + 自动评估降低上线风险，K8s 部署配置生成器将版本映射为容器和流量资源。关键权衡在于签名兼容性与模型演进、金丝雀评估的统计显著性、存储成本与版本保留策略，以及多集群同步的一致性。模型版本管理的核心目标是让线上运行的模型版本可追溯、可回滚、可审计。