MLOps 工具链

MLflow 实验追踪 / 模型注册、DVC 数据版本管理、Weights & Biases、模型 CI/CD

MLOps 概述

MLOps（Machine Learning DevOps）是将 DevOps 理念应用于机器学习系统的工程实践，旨在标准化和自动化 ML 生命周期的各个环节。与传统的 DevOps 相比，MLOps 不仅要管理代码，还需管理数据、模型和实验，因此复杂度更高。

MLOps 生命周期包含以下阶段：

1. 数据阶段：数据采集、清洗、标注、版本管理
2. 实验阶段：特征工程、模型选择、超参数调优
3. 训练阶段：分布式训练、资源调度、训练监控
4. 部署阶段：模型打包、A/B 测试、灰度发布
5. 监控阶段：模型性能衰减检测、数据漂移、反馈闭环

MLOps 与传统 DevOps 的关键区别：

维度	DevOps	MLOps
版本管理对象	源代码	代码 + 数据 + 模型 + 实验配置
测试验证	单元测试、集成测试	数据验证、模型评估、公平性测试
部署产物	二进制/容器	模型 artifact + 推理服务
监控重点	服务可用性、响应延迟	模型准确率、数据分布漂移
回滚策略	代码版本回退	模型版本回退 + 数据版本回退

---

MLflow 实验追踪与模型注册

MLflow 是一个开源 ML 生命周期管理平台，其核心组件包括 Tracking 和 Model Registry。

MLflow Tracking

MLflow Tracking 提供实验和运行管理功能：

• Experiment（实验）：一组相关运行的集合，可类比为一个项目的研究过程
• Run（运行）：单次模型训练的执行记录，包含参数、指标、标签和产物
• autolog 自动日志：通过 mlflow.autolog() 自动捕获主流框架（PyTorch、TensorFlow、Scikit-learn）的训练参数和指标
• artifact 存储：模型权重、图表、日志文件等产物的集中管理

import mlflow
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, precision_score

# 设置实验
mlflow.set_experiment("iris_classification")

with mlflow.start_run(run_name="rf_v1"):
    # 自动日志（支持 sklearn / pytorch / tensorflow 等）
    mlflow.autolog()

    # 参数记录
    mlflow.log_param("n_estimators", 100)
    mlflow.log_param("max_depth", 10)

    # 训练模型
    model = RandomForestClassifier(n_estimators=100, max_depth=10)
    model.fit(X_train, y_train)
    y_pred = model.predict(X_test)

    # 指标记录
    mlflow.log_metric("accuracy", accuracy_score(y_test, y_pred))
    mlflow.log_metric("precision", precision_score(y_test, y_pred, average="macro"))

    # 注册模型到 Model Registry
    mlflow.sklearn.log_model(
        model,
        "model",
        registered_model_name="IrisClassifier"
    )

MLflow Model Registry

Model Registry 提供模型版本管理和生命周期控制：

• 版本管理：每次注册自动递增版本号，支持版本间对比
• 阶段转换：支持 Staging（预发布）→ Production（生产）→ Archived（归档）的阶段流转
• 模型别名：为特定版本赋予别名（如 "champion"、"challenger"），便于推理服务引用

from mlflow.tracking import MlflowClient

client = MlflowClient()

# 将模型版本 2 转为 Staging 阶段
client.transition_model_version_stage(
    name="IrisClassifier",
    version=2,
    stage="Staging"
)

# 设置模型别名
client.set_registered_model_alias(
    name="IrisClassifier",
    alias="champion",
    version=3
)

---

DVC 数据版本管理

DVC（Data Version Control）是一个面向 ML 项目的开源数据版本管理工具，与 Git 协同工作。

数据版本化

DVC 的核心工作流如下：

1. dvc init：初始化 DVC 环境，生成 .dvc/ 目录
2. dvc add data/raw：追踪数据目录，生成 .dvc 元文件（存入 Git），实际数据加入 .dvc/cache
3. dvc push：将缓存数据推送到远程存储

# 初始化 DVC
dvc init

# 添加数据文件，生成 data.dvc 元文件
dvc add data/dataset.csv

# 配置远程存储（以本地目录为例）
dvc remote add -d myremote /mnt/dvc-storage

# 推送数据到远程
dvc push

# 切换版本时，先 git checkout 切换 .dvc 文件，再执行
dvc pull

远程存储配置

DVC 支持多种远程存储后端：

# AWS S3
dvc remote add -d s3remote s3://my-bucket/dvc-store

# Google Cloud Storage
dvc remote add -d gcsremote gs://my-bucket/dvc-store

# 本地目录
dvc remote add -d localremote /mnt/shared/dvc-store

数据管道

DVC 通过 dvc.yaml 定义有向无环图（DAG）形式的 pipeline，支持缓存和增量执行——仅当依赖发生变化时才重新运行对应阶段。

# dvc.yaml
stages:
  preprocess:
    cmd: python src/preprocess.py --input data/raw --output data/processed
    deps:
      - data/raw
      - src/preprocess.py
    outs:
      - data/processed
    params:
      - preprocess.min_count
      - preprocess.max_len

  train:
    cmd: python src/train.py --data data/processed --model models/model.pkl
    deps:
      - data/processed
      - src/train.py
    outs:
      - models/model.pkl
    params:
      - train.learning_rate
      - train.epochs

  evaluate:
    cmd: python src/evaluate.py --model models/model.pkl --data data/processed
    deps:
      - models/model.pkl
      - src/evaluate.py
    metrics:
      - metrics/eval.json:
          cache: false

# 运行 pipeline（自动检测变更，仅执行需要更新的阶段）
dvc repro

# 查看 pipeline DAG
dvc dag

---

Weights & Biases（WandB）

Weights & Biases 是一个面向 AI 研究者的实验追踪和协作平台，以其实时仪表盘和超参数搜索功能著称。

MLflow vs W&B 对比

功能维度	MLflow	Weights & Biases
部署方式	自托管 / Databricks	SaaS 为主，支持私有部署
实时仪表盘	需刷新页面	浏览器实时更新
超参数搜索	需结合 Optuna 等	内置 Sweep 引擎
报告与协作	基础功能	富文本报告、团队协作
数据版本管理	不支持	支持 Artifact
开源协议	Apache 2.0	部分开源

W&B 实验追踪

import wandb
import torch
import torch.nn as nn

# 初始化 W&B 运行
wandb.init(
    project="iris-classification",
    config={
        "learning_rate": 0.001,
        "epochs": 50,
        "batch_size": 32,
        "optimizer": "adam"
    }
)

# 自动记录 PyTorch 训练
wandb.watch(model, log_freq=100)

for epoch in range(config.epochs):
    train_loss = train_one_epoch(model, dataloader)
    val_acc = evaluate(model, val_dataloader)

    # 记录指标（实时更新到 W&B 仪表盘）
    wandb.log({
        "epoch": epoch,
        "train_loss": train_loss,
        "val_accuracy": val_acc,
        "learning_rate": current_lr
    })

# 保存模型 artifact
artifact = wandb.Artifact("iris-model", type="model")
artifact.add_file("model.pth")
wandb.log_artifact(artifact)

wandb.finish()

W&B Sweep 超参数搜索

import wandb

sweep_config = {
    "method": "bayes",
    "metric": {"name": "val_accuracy", "goal": "maximize"},
    "parameters": {
        "learning_rate": {"min": 1e-4, "max": 1e-2},
        "batch_size": {"values": [16, 32, 64]},
        "dropout": {"min": 0.1, "max": 0.5}
    }
}

sweep_id = wandb.sweep(sweep_config, project="iris-hpo")
wandb.agent(sweep_id, function=train_with_config, count=20)

---

模型 CI/CD

将 ML 工作流集成到 CI/CD 系统中，实现训练自动化和部署自动化。

模型训练 CI

训练 CI pipeline 在代码或数据变更时自动触发：

# .github/workflows/model-train.yml
name: Model Training Pipeline

on:
  push:
    branches: [main]
    paths:
      - "src/**"
      - "data/**"
      - "dvc.yaml"

jobs:
  train:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Pull DVC data
        run: |
          pip install dvc
          dvc pull -r s3remote

      - name: Data validation
        run: python src/validate_data.py

      - name: Train model
        run: python src/train_pipeline.py

      - name: Evaluate model
        run: python src/evaluate.py --report report.json

      - name: Register model (if accuracy > threshold)
        run: |
          accuracy=$(python -c "import json; print(json.load(open('report.json'))['accuracy'])")
          if (( $(echo "$accuracy > 0.95" | bc -l) )); then
            python src/register_model.py
          fi

模型部署 CD

模型注册后自动触发部署到生产环境：

# .github/workflows/model-deploy.yml
name: Model Deployment Pipeline

on:
  workflow_run:
    workflows: ["Model Training Pipeline"]
    types: [completed]

jobs:
  deploy:
    runs-on: ubuntu-latest
    if: ${{ github.event.workflow_run.conclusion == 'success' }}
    steps:
      - uses: actions/checkout@v3

      - name: Download model artifact
        run: python src/download_model.py --version latest

      - name: Build Docker image
        run: |
          docker build -t inference-service:latest -f deploy/Dockerfile .
          docker tag inference-service:latest registry.example.com/inference-service:${{ github.sha }}

      - name: Push to registry
        run: |
          docker push registry.example.com/inference-service:${{ github.sha }}

      - name: Deploy to Kubernetes
        run: |
          kubectl set image deployment/inference-service \
            inference-service=registry.example.com/inference-service:${{ github.sha }} \
            -n production

完整的 MLOps 工具链功能对比

工具	核心能力	部署方式	开源	适用场景
MLflow	实验追踪、模型注册、部署	自托管 / 托管	是	通用 ML 生命周期管理
DVC	数据版本管理、Pipeline	自托管	是	数据密集型 ML 项目
W&B	实验追踪、超参搜索、协作	SaaS / 私有部署	部分开源	研究与快速迭代
GitHub Actions / GitLab CI	自动化流水线	托管 / 自托管	是	CI/CD 编排

推荐组合方案：使用 MLflow 管理实验和模型生命周期，DVC 管理大数据集和训练 pipeline，W&B 作为研究阶段的实验探索平台，GitHub Actions 或 GitLab CI 串联整个 MLOps 流水线，实现从数据准备到模型部署的全流程自动化。