MLOps 流水线：模型训练、评估与发布的 CI/CD 实践

传统交付：提交代码 → 测试 → 构建 → 部署 → 监控（确定性系统）
ML 交付：  数据变更 → 训练 → 评估 → 注册 → 发布 → 监控 → 再训练
每个环节都有不确定因素：数据分布、模型指标、线上行为都会变

0 层 Notebook 手工训练 / 1 层训练脚本进 CI/CD / 2 层模型注册+自动发布 / 3 层监控+自动回滚
多数团队从第 1 层起步：训练/评估脚本进 Git 统一流水线触发，收益最大

dvc init && dvc add data/train.parquet data/test.parquet
dvc remote add -d storage s3://ml-data-bucket/dvc
dvc push   # Git 只提交几十字节的 .dvc 指针
# 线上跑偏：git checkout <旧commit> && dvc checkout 拉回数据快照

from feast import FeatureView, Field
from feast.types import Float32
view = FeatureView(name="user_features", entities=["user_id"],
    schema=[Field(name="ctr_7d", dtype=Float32)],
    online=True, source=parquet_source)

# Kubeflow Pipelines（概念）
@dsl.pipeline(name="train-pipeline")
def pipe(data_uri, epochs):
    prep = data_prep_op(data_uri)
    train = train_op(prep, epochs)     # GPU 训练
    eval_ = evaluate_op(train)         # 离线评估
    register_op(eval_)                 # 写回 MLflow Registry

整体指标：AUC / F1（全局）；分组指标：按地域/设备/年龄切片防"偏科"
鲁棒指标：新样本与对抗样本稳定性；业务指标：转化率、召回率（离线近似）

rules:
  auc: { min: 0.86 }
  group_auc_android: { min: 0.85 }   # 安卓端偏科 → 拦截
  drift_score: { max: 0.15 }

联内评估：训练完在流水线内评估（训练=构建，评估=测试）
持续评估：对已注册模型定时拉生产数据跑离线指标，发现劣化提前预警

mlflow models register -m runs:/<run_id>/model -n recommendation_model
# 阶段流转：None → Staging → Production → Archived，评估门禁通过后自动置 Staging

# MLflow run 固化血缘标签
mlflow.start_run(run_name="daily_retrain_20260930")
mlflow.log_params({"data_version": "v42", "feast_view": "user_features@2026-09-29"})

不可变：模型一经发布不可覆盖，只能新增版本
命名：<业务>-<日期>-<run_id前8位>，如 reco-20260930-a3f2c9d1
保留：生产版本 + 最近 N 个历史版本可回滚，其余归档清缓存

from evidently.report import Report
from evidently.metric_preset import DataDriftPreset
Report(metrics=[DataDriftPreset()]).run(
    reference_data=train_sample, current_data=production_sample)
# 输出每个特征的 PSI/drifted 标记，作为 CI 断言

用历史流量回放：过去 N 天线上请求喂给新模型打分，对比新旧模型预测分布与命中指标。
不切流量即可评估新模型线上表现

# Istio A/B 切分 90/10
http:
  - match: [{ headers: { ab-group: { exact: "B" } } }]
    route: [{ destination: { host: reco-svc, subset: v2 } }]
  - route:
      - destination: { host: reco-svc, subset: v1, weight: 90 }
      - destination: { host: reco-svc, subset: v2, weight: 10 }

任一触发即回滚：线上 AUC/点击率环比下降超阈值 / P99 超 SLO 3 个周期 /
  预测失败率 >1%。回滚 = 把流量切回上一稳定模型版本

train:
  stage: train
  image: pytorch/pytorch:2.3-cuda12.1-cudnn8
  tags: [gpu-runner]              # 专用 GPU Runner 池
  script:
    - dvc pull && python train.py --epochs 20
    - python evaluate.py
  artifacts: { paths: [metrics.json] }

缓存：特征结果与预训练 checkpoint 存对象存储
并发：同机型训练全局最多 N 个并发
计量：按部门/项目打标签，FinOps 分摊 GPU 账单（nvidia-smi + gpu_utilization 看板）

系统层：延迟/吞吐/错误率/GPU 利用率；模型层：预测分布、空预测率、置信度均值
业务层：点击率、转化率；漂移层：特征漂移、标签漂移

sum(rate(reco_click_total[1h])) / sum(rate(reco_impression_total[1h]))  # 点击率
histogram_quantile(0.99,
  sum(rate(reco_inference_duration_seconds_bucket[5m])) by (le))        # P99 延迟
rate(reco_prediction_errors_total[5m]) > 0.01                           # 错误率超阈值

回滚不是"删掉新模型"而是"指回上一稳定版本"：
  拉 Registry 上一 Production 版本 → 更新服务镜像 tag（不可变）→
  切流量 + 观察 → 复盘（数据/特征/模型问题）再决定修复重训

某电商推荐团队（概念案例）：每天凌晨 2 点重训，先手工跑脚本再手动部署，返工率高
改造：Airflow 定时触发 Kubeflow → MLflow 自动注册 → CI 门禁（AUC/分组/漂移）
  → 金丝雀 10% 观察 1 小时 → 全量
一次：监控发现召回率环比降 8%，自动回滚到上一版模型
效果：发布从"周"缩短到"日"，回滚从小时级降到分钟级

□ 数据用 DVC 版本化，代码/数据/模型三者对齐
□ 特征收敛到 Feature Store，消除训练-服务偏差
□ 评估门禁覆盖整体 + 分组 + 漂移，阈值可配
□ 模型进 Registry，阶段流转走审批，不可变
□ 血缘链完整：数据→特征→run→模型→部署 revision
□ 发布走金丝雀/A-B，配置自动回滚触发器
□ GPU 分队列调度，训练独占卡，缓存中间产物
□ 线上监控分层（系统/模型/业务/漂移）
□ 每次发布记录版本指纹，可随时追溯与回滚