实验管理与可复现：MLflow、版本化与模型注册表

代码 git commit sha   数据 数据集哈希/快照 ID   环境 依赖版本
超参 学习率/批大小/结构   随机种子 seed   产物 权重/日志/图表/指标

Experiment（实验）：一个项目/一个任务方向
  └── Run（运行）  ：一次具体的训练
        ├── Params   ：输入参数（超参、配置）
        ├── Metrics  ：输出指标（可随步数记录）
        ├── Artifacts：产物文件（模型、图表、日志）
        └── Tags     ：元信息（git sha、作者、数据版本）

import mlflow
# 本地默认存到 ./mlruns；生产建议用数据库 + 对象存储
mlflow.set_tracking_uri("http://127.0.0.1:5000")
mlflow.set_experiment("fraud-detection")
# 启动服务：mlflow server --host 0.0.0.0 --port 5000

import mlflow
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import f1_score
with mlflow.start_run(run_name="rf-baseline"):
    params = {"n_estimators": 200, "max_depth": 8, "random_state": 42}
    mlflow.log_params(params)
    model = RandomForestClassifier(**params)
    model.fit(X_train, y_train)
    mlflow.log_metric("f1_macro", f1_score(y_test, model.predict(X_test)))
    mlflow.sklearn.log_model(model, "model")
    mlflow.set_tag("git_sha", git_sha)
    print("run id:", mlflow.active_run().info.run_id)

mlflow.sklearn.autolog()          # 自动记录超参、指标、模型
mlflow.pytorch.autolog()          # PyTorch 训练循环自动记录
mlflow.transformers.autolog()     # HuggingFace 微调自动记录

for epoch in range(epochs):
    train_loss = train_one_epoch(model, loader)
    val_loss, val_acc = evaluate(model, val_loader)
    mlflow.log_metrics({"train_loss": train_loss,
                        "val_loss": val_loss,
                        "val_acc": val_acc}, step=epoch)

import matplotlib.pyplot as plt
fig, ax = plt.subplots(); ax.plot(history["val_loss"])
mlflow.log_figure(fig, "curves/val_loss.png")   # 直接 log figure 对象
mlflow.log_artifact("val_loss.png")             # 单个文件
mlflow.log_artifacts("reports/", "reports")     # 整个目录

# 用父 run 包住多个子 run，适合「一次调参扫一批」
with mlflow.start_run(run_name="hpo-sweep") as parent:
    for lr in [1e-3, 1e-4, 1e-5]:
        with mlflow.start_run(run_name=f"lr-{lr}", nested=True):
            mlflow.log_param("lr", lr); mlflow.log_metric("f1", train_and_eval(lr))
# 用 Pandas 拉回结果做分析
df = mlflow.search_runs(experiment_names=["fraud-detection"])
cols = ["params.n_estimators", "params.max_depth",
        "metrics.f1_macro", "tags.git_sha"]
print(df[cols].sort_values("metrics.f1_macro", ascending=False).head())

代码   → git（commit sha 记进 run 的 tag）
数据   → 快照 / 哈希 / DVC
环境   → 依赖锁文件 / 容器镜像

import subprocess
def git_sha():
    return subprocess.check_output(["git", "rev-parse", "HEAD"]).decode().strip()
mlflow.set_tag("git_sha", git_sha())
mlflow.set_tag("git_dirty", bool(subprocess.check_output(
    ["git", "status", "--porcelain"]).strip()))

import hashlib
def data_fingerprint(path):
    """对数据内容做哈希，作为数据版本号"""
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 20), b""):
            h.update(chunk)
    return h.hexdigest()[:16]
mlflow.set_tag("data_version", data_fingerprint("data/train.parquet"))

dvc init
dvc remote add -d storage s3://my-bucket/dvc
dvc add data/train.parquet
git add data/train.parquet.dvc data/.gitignore && git commit -m "track data v3"
git checkout <sha> && dvc checkout     # 复现历史某次实验

mlflow.log_artifact("requirements.txt")
mlflow.set_tag("python_version", sys.version.split()[0])
mlflow.set_tag("image", "registry.example.com/train:cuda12.1-v3")

import mlflow
with mlflow.start_run() as run:
    mlflow.sklearn.log_model(model, "model", registered_model_name="fraud-detector")
client = mlflow.MlflowClient()
client.transition_model_version_stage(
    name="fraud-detector", version="3", stage="Staging")
# 验证通过后转生产，旧的自动归档
client.transition_model_version_stage(
    name="fraud-detector", version="3", stage="Production",
    archive_existing_versions=True)

client.set_model_version_tag("fraud-detector", "3", "validation_f1", "0.87")
client.update_model_version(name="fraud-detector", version="3",
    description="改用 LightGBM + 新特征 v4，AUC 提升 1.2 个点")
# 线上服务加载 Production 阶段，而不是硬编码某个版本号
model = mlflow.pyfunc.load_model("models:/fraud-detector/Production")
old = mlflow.pyfunc.load_model("models:/fraud-detector/2")   # 回滚验证

训练完成 → 自动注册为 Staging → 跑验证集与偏置检查
  → 通过则转 Production → 触发部署流水线 → 监控告警
  → 指标劣化则回退到上一 Production 版本

runs = df[["params.max_depth", "params.n_estimators", "metrics.f1_macro"]].dropna()
runs["params.max_depth"] = runs["params.max_depth"].astype(int)
print(runs.pivot_table(index="params.max_depth",
                       columns="params.n_estimators",
                       values="metrics.f1_macro", aggfunc="max").round(4))

best = df.sort_values("metrics.f1_macro", ascending=False).iloc[0]
print("最佳 run:", best["run_id"])
print("参数:", {k: v for k, v in best.items() if k.startswith("params.")})

mlflow server \
  --backend-store-uri postgresql://user:pwd@db:5432/mlflow \
  --default-artifact-root s3://my-bucket/mlflow \
  --host 0.0.0.0 --port 5000

实验名 项目-任务（fraud-detection）  run 名 模型-关键变量（lgbm-lr1e-3）
标签 owner / team / data_version / git_sha / stage

mlflow.search_runs(filter_string="tags.owner = 'alice' and metrics.f1_macro > 0.8")

git checkout <sha>          # 1. 切到记录的代码版本
dvc checkout                # 2. 恢复数据版本
mlflow artifacts download -r <run_id> -a config.yaml   # 3. 拉取超参配置
docker run --gpus all train:cuda12.1-v3 python train.py --config config.yaml

1. 每次训练必须开 run，禁止裸跑
2. 必记 git_sha、data_version、seed、owner
3. 模型产物只存到注册表候选，其余 run 不存
4. 上线只从 Production 阶段拉取，结论必附 run_id

阶段一：开 run + 记参数指标（当天可落地）
阶段二：加 git_sha / data_version / seed 标签
阶段三：上集中式 server + 对象存储
阶段四：引入模型注册表与阶段流转
阶段五：接入 CI/CD，自动验证与回滚