Python ML Pipeline 版本管理：模型、数据和代码的联合版本控制

一、模型跑出的结果和上周不一样了，但没人记得改了什么

ML 团队调试了两天，发现一个线上模型的推理结果悄悄变了。问题是：

• 没人改代码（Git 提交记录是干净的）
• 模型文件看起来没变
• 输入数据格式和上周一致

最终定位：数据管线中的一个上游表在两周前做了字段迁移，新字段的空值处理逻辑和旧逻辑不同。但这些变更没有被记录下来，也没有触发任何告警。

这就是 ML Pipeline 版本管理的核心难题：模型效果取决于代码 + 数据 + 模型权重三个变量的联合状态，任何一个变了都会影响结果。只管理代码版本是不够的。

二、联合版本管理架构

该架构主要围绕三个核心层级构建：可版本化资产、Pipeline 版本快照以及可追溯性机制。

在资产层，所有影响模型效果的变量均需纳入版本控制，具体包括代码（Git Commit Hash）、数据（Schema + 切分 + Hash）、模型（权重文件 + 超参数）以及环境（Docker Image + 依赖版本）。

在快照层，系统将这些资产的哈希值组合生成唯一的 Pipeline 快照，并统一存储于版本注册中心（如 MLflow 或 DVC）。

在追溯层，基于注册中心实现完整的可追溯性：支持查询特定结果的来源（代码、数据、模型版本），支持复现历史结果，以及支持对比不同版本间的结果差异。

三、Python 实现联合版本管理

版本快照管理器

import hashlib
import json
import os
import time
from typing import Dict, Any, Optional, List

from dataclasses import dataclass, field, asdict
from datetime import datetime
import subprocess

@dataclass
class PipelineSnapshot:
"""ML Pipeline 版本快照——记录所有输入资产的哈希"""
snapshot_id: str
pipeline_name: str
timestamp: str

# 代码版本
git_commit: str           # Git commit hash
git_branch: str           # Git 分支名
entry_file: str           # 入口脚本路径

# 数据版本
data_hash: str            # 数据集的 MD5/SHA256
data_schema_hash: str     # 数据 Schema 的哈希
data_split: str           # 数据切分方式（如 "80/20"）

# 模型版本
model_hash: str           # 模型文件哈希
model_type: str           # 模型类型 (e.g., "lightgbm", "torch")
hyperparams: Dict[str, Any]  # 超参数

# 环境版本
docker_image: str         # Docker 镜像标签
python_version: str       # Python 版本
dependency_hashes: Dict[str, str]  # 关键依赖的版本 + 哈希

# 结果
metrics: Dict[str, float]  # 评估指标
artifacts: List[str]       # 产出的文件路径

class PipelineVersionManager:
"""ML Pipeline 版本管理器"""

def __init__(self, registry_path: str = "./pipeline_registry.json"):
    self.registry_path = registry_path
    self._registry: List[PipelineSnapshot] = self._load_registry()

def take_snapshot(
    self,
    pipeline_name: str,
    data_path: str,
    model_path: str,
    hyperparams: Dict[str, Any],
    metrics: Dict[str, float],
    artifacts: List[str],
) -> PipelineSnapshot:
    """
    创建 Pipeline 快照
    ——在每次训练/推理完成后调用
    """
    snapshot = PipelineSnapshot(
        snapshot_id=self._generate_snapshot_id(),
        pipeline_name=pipeline_name,
        timestamp=datetime.now().isoformat(),
        
        # 代码版本：从 Git 获取
        git_commit=self._get_git_commit(),
        git_branch=self._get_git_branch(),
        entry_file=__file__,
        
        # 数据版本：计算数据文件 + Schema 的哈希
        data_hash=self._compute_file_hash(data_path),
        data_schema_hash=self._compute_schema_hash(data_path),
        data_split="80/20",
        
        # 模型版本：计算模型文件的哈希
        model_hash=self._compute_file_hash(model_path),
        model_type=self._detect_model_type(model_path),
        hyperparams=hyperparams,
        
        # 环境版本
        docker_image=os.environ.get("DOCKER_IMAGE", "unknown"),
        python_version=self._get_python_version(),
        dependency_hashes=self._get_dependency_hashes(),
        
        # 结果
        metrics=metrics,
        artifacts=artifacts,
    )
    
    # 保存到注册中心
    self._registry.append(snapshot)
    self._save_registry()
    
    print(f"快照已创建: {snapshot.snapshot_id}")
    return snapshot

def diff(self, snapshot_id_a: str, snapshot_id_b: str) -> Dict[str, Any]:
    """
    对比两个版本快照的差异
    ——用于追溯模型行为变化的原因
    """
    a = self._find_snapshot(snapshot_id_a)
    b = self._find_snapshot(snapshot_id_b)
    
    if not a or not b:
        return {"error": "快照不存在"}
    
    diffs = {}
    
    # 代码变化
    if a.git_commit != b.git_commit:
        diffs["code"] = {
            "before": a.git_commit[:8],
            "after": b.git_commit[:8],
            "branch": a.git_branch,
        }
    
    # 数据变化
    if a.data_hash != b.data_hash:
        diffs["data"] = {
            "before_hash": a.data_hash[:16],
            "after_hash": b.data_hash[:16],
        }
    
    # 模型变化
    if a.model_hash != b.model_hash:
        diffs["model"] = {
            "before_type": a.model_type,
            "after_type": b.model_type,
            "hyperparams_changed": a.hyperparams != b.hyperparams,
        }
    
    # 指标变化
    if a.metrics != b.metrics:
        metric_diffs = {}
        for key in set(list(a.metrics.keys()) + list(b.metrics.keys())):
            before_val = a.metrics.get(key, 0)
            after_val = b.metrics.get(key, 0)
            if before_val != after_val:
                metric_diffs[key] = {
                    "before": before_val,
                    "after": after_val,
                    "delta": after_val - before_val,
                }
        diffs["metrics"] = metric_diffs
    
    return diffs

def reproduce_check(self, snapshot_id: str) -> bool:
    """
    复现性检查：验证当前环境能否复现历史结果
    ——检查代码、数据、模型的哈希是否一致
    """
    snapshot = self._find_snapshot(snapshot_id)
    if not snapshot:
        return False
    
    current_commit = self._get_git_commit()
    if current_commit != snapshot.git_commit:
        print(f"代码版本不一致: 当前 {current_commit[:8]} != 快照 {snapshot.git_commit[:8]}")
        return False
    
    # 检查依赖版本
    current_deps = self._get_dependency_hashes()
    for lib, ver in snapshot.dependency_hashes.items():
        if current_deps.get(lib) != ver:
            print(f"依赖版本不一致: {lib}")
            return False
    
    print(f"快照 {snapshot_id} 可复现")
    return True

def _generate_snapshot_id(self) -> str:
    """生成快照 ID: pipeline_name + 时间戳 + 随机后缀"""
    import uuid
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    return f"snapshot_{ts}_{uuid.uuid4().hex[:8]}"

def _get_git_commit(self) -> str:
    """获取当前 Git commit hash"""
    try:
        result = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            capture_output=True, text=True, timeout=5,
        )
        return result.stdout.strip()
    except Exception:
        return "unknown"

def _get_git_branch(self) -> str:
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--abbrev-ref", "HEAD"],
            capture_output=True, text=True, timeout=5,
        )
        return result.stdout.strip()
    except Exception:
        return "unknown"

def _compute_file_hash(self, file_path: str, algorithm: str = "md5") -> str:
    """计算文件的哈希值"""
    if not os.path.exists(file_path):
        return f"file_not_found:{file_path}"
    
    hasher = hashlib.new(algorithm)
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            hasher.update(chunk)
    return hasher.hexdigest()

def _compute_schema_hash(self, data_path: str) -> str:
    """
    计算数据 Schema 的哈希
    ——通过对列名 + 数据类型排序后哈希实现
    """
    import pandas as pd
    
    try:
        df = pd.read_csv(data_path, nrows=0)  # 只读表头
        schema = {
            col: str(df[col].dtype) 
            for col in sorted(df.columns)
        }
        schema_str = json.dumps(schema, sort_keys=True)
        return hashlib.md5(schema_str.encode()).hexdigest()
    except Exception:
        return "schema_unknown"

def _get_python_version(self) -> str:
    import sys
    return sys.version.split()[0]

def _get_dependency_hashes(self) -> Dict[str, str]:
    """获取关键依赖的版本"""
    import importlib.metadata
    
    key_libs = [
        "numpy", "pandas", "scikit-learn", "torch",
        "lightgbm", "xgboost", "transformers",
    ]
    
    hashes = {}
    for lib in key_libs:
        try:
            version = importlib.metadata.version(lib)
            hashes[lib] = version
        except importlib.metadata.PackageNotFoundError:
            hashes[lib] = "not_installed"
    
    return hashes

def _detect_model_type(self, model_path: str) -> str:
    """检测模型类型"""
    if model_path.endswith(".pkl") or model_path.endswith(".joblib"):
        return "pickle/sklearn"
    elif model_path.endswith(".pt") or model_path.endswith(".pth"):
        return "pytorch"
    elif model_path.endswith(".h5"):
        return "keras/tf"
    elif model_path.endswith(".txt") or model_path.endswith(".model"):
        return "lightgbm/xgboost"
    return "unknown"

def _find_snapshot(self, snapshot_id: str) -> Optional[PipelineSnapshot]:
    for s in self._registry:
        if s.snapshot_id == snapshot_id:
            return s
    return None

def _load_registry(self) -> List[PipelineSnapshot]:
    if os.path.exists(self.registry_path):
        with open(self.registry_path, "r") as f:
            data = json.load(f)
            return [PipelineSnapshot(**item) for item in data]
    return []

def _save_registry(self):
    with open(self.registry_path, "w") as f:
        json.dump(
            [asdict(s) for s in self._registry],
            f, indent=2, ensure_ascii=False,
        )

### 集成到 ML Pipeline 中

```python
# 在训练脚本中使用版本管理
def train_with_versioning():
    manager = PipelineVersionManager()
    
    # 1. 数据预处理
    data_path = "data/processed/train_v3.csv"
    preprocess(data_path)
    
    # 2. 训练模型
    model, hyperparams = train_model(data_path)
    model_path = "models/classifier_v3.pkl"
    save_model(model, model_path)
    
    # 3. 评估
    metrics = evaluate(model, data_path)
    
    # 4. 创建版本快照
    snapshot = manager.take_snapshot(
        pipeline_name="user_churn_predictor",
        data_path=data_path,
        model_path=model_path,
        hyperparams=hyperparams,
        metrics=metrics,
        artifacts=[model_path, "reports/feature_importance.png"],
    )
    
    # 5. 与上一个版本对比
    prev_snapshots = manager._registry
    if len(prev_snapshots) >= 2:
        diff = manager.diff(
            prev_snapshots[-2].snapshot_id,
            snapshot.snapshot_id,
        )
        print(f"版本对比: {json.dumps(diff, indent=2)}")

四、边界分析与 Trade-offs

数据哈希的粒度：

• 对数据文件做全量哈希成本高（大文件 <1GB 约 1-3 秒）
• 对于 TB 级数据，建议用采样哈希（取前 N 行 + 随机采样 N 行）
• Schema 哈希可以快速检测字段结构变化

快照膨胀：每次训练产生一个快照，一个月可能有上百个。需要设置淘汰策略（保留最近 N 个 + 最佳 N 个）。

Docker 镜像管理：生产环境建议锁定镜像 tag，避免 latest 标签导致的环境漂移。

与 MLflow/DVC 的关系：本文是实现原理，生产环境可使用 MLflow Tracking + DVC 的组合方案。

五、总结

ML Pipeline 的版本管理需要做到"给定一个快照 ID，可以追溯到完整的运行环境"：

1. 代码：Git Commit Hash
2. 数据：文件哈希 + Schema 哈希
3. 模型：权重文件哈希 + 超参数
4. 环境：Docker 镜像 + 依赖版本
5. 结果：评估指标 + 产出文件

这五个维度的联合哈希构成了一个不可伪造的"实验指纹"。当模型行为发生变化时，通过 diff 功能可以快速定位是哪个环节变了。