MLflow 与血缘

这一课你将能做到

1. 启动集中式 MLflow Server（元数据 + artifact 分离），把模型注册进 Model Registry
2. 用别名（alias）实现模型晋升 challenger → champion，并让服务按别名动态加载
3. 打通「模型版本 → run → 数据版本/代码版本」的血缘链，做到任何线上模型可回溯

为什么要学

上一课的 Tracking 解决了「实验记下来」，但还有三个生产问题没解决：哪个 run 有资格上线（需要注册与门禁）、线上跑的是哪一版（需要版本指针）、出事后怎么查（需要血缘）。MLflow Model Registry 就是模型界的「发布仓库」：run 是草稿，Model Version 是待审品，alias（champion/challenger）是线上指针，门禁是审核规则。

人话：把「把最新模型文件 scp 到服务器覆盖旧的」这种裸奔操作，升级成「注册 → 评估 → 晋升 → 回滚都动指针」的正规军流程。本站判题超时模型将获得一个真正的生产身份：timeout_classifier@champion。

先建立图像

训练 run (src/train.py)
   │ mlflow.register_model
   ▼
Model Registry: timeout_classifier
├─ Version 5  alias: challenger   ← 新模型先挂挑战者
├─ Version 4  alias: champion     ← 线上正在服务
└─ Version 3  (无 alias，可随时指回)

线上服务启动时读 models:/timeout_classifier@champion
晋升 = 把 champion 从 v4 挪到 v5（一次 API 调用），服务重启即生效
回滚 = 把 champion 指回 v4，秒级完成，不需要重新部署

血缘：Version 5 ──▶ run abc123 ──▶ params/tags ──▶ git_sha + 数据文件哈希

跟着做

步骤 1：启动集中式 MLflow Server

mlflow server \
  --backend-store-uri sqlite:///mlflow.db \
  --artifacts-destination ./mlartifacts \
  --host 127.0.0.1 --port 5000

元数据（实验/run/注册信息）进 SQLite，模型文件进本地目录。生产把两者换成 PostgreSQL + S3/OSS 即可，客户端代码不变。

步骤 2：训练时直连 Server 并注册

给上一课的 src/train.py 增加两处：脚本开头设置远端 URI，run 内注册模型。

# src/train.py 顶部新增
import os
mlflow.set_tracking_uri(os.getenv("MLFLOW_TRACKING_URI", "http://127.0.0.1:5000"))

# main() 里 log_model 之后新增
mlflow.register_model(
    model_uri=f"runs:/{run.info.run_id}/model",
    name="timeout_classifier",
    tags={"git_sha": git_sha()},
)

export MLFLOW_TRACKING_URI=http://127.0.0.1:5000
python src/train.py
# UI 的 Models 页面出现 timeout_classifier 的最新 Version

步骤 3：用别名管理晋升与回滚

新版 MLflow 推荐 alias 替代旧的 stage（staging/production）机制——后者已标记弃用，但面试仍会问到，概念一致。

# src/promote.py
import sys
from mlflow.tracking import MlflowClient

MODEL = "timeout_classifier"
client = MlflowClient()

def promote(version: int):
    v = client.get_model_version(MODEL, str(version))
    assert v.run_id, "必须来自真实 run，拒绝裸文件注册"
    client.set_registered_model_alias(MODEL, "challenger", version)
    print(f"v{version} → challenger")

def gate_and_release(version: int, min_auc: float = 0.80):
    run = client.get_run(client.get_model_version(MODEL, str(version)).run_id)
    auc = run.data.metrics["roc_auc"]
    if auc < min_auc:
        sys.exit(f"门禁不通过：auc={auc:.3f} < {min_auc}")   # 指标门禁
    client.set_registered_model_alias(MODEL, "champion", version)
    client.update_model_version(
        name=MODEL, version=str(version),
        description=f"released, auc={auc:.3f}")            # 留痕
    print(f"v{version} → champion")

if __name__ == "__main__":
    cmd, version = sys.argv[1], int(sys.argv[2])
    if cmd == "stage":
        promote(version)
    elif cmd == "release":
        gate_and_release(version)
    else:
        sys.exit("用法: python src/promote.py stage|release <version>")