Skip to content
←

필사 모드: MLflow 完全指南：从实验追踪到 Model Registry、生产部署

中文
0%
정확도 0%
💡 왼쪽 원문을 읽으면서 오른쪽에 따라 써보세요. Tab 키로 힌트를 받을 수 있습니다.

MLflow 是什么？

MLflow 是一个用于管理 ML 生命周期的开源平台，由四个核心组件构成：

  • MLflow Tracking：记录实验参数、指标和产物
  • MLflow Projects：将可复现的 ML 代码打包
  • MLflow Models：将各种框架的模型打包为统一格式
  • MLflow Model Registry：模型版本管理与部署工作流

安装与服务器配置

基础安装

# pip 安装
pip install mlflow

# 支持额外框架
pip install mlflow[extras]  # sklearn、tensorflow、pytorch 等

# 启动服务器（本地）
mlflow server --host 0.0.0.0 --port 5000

# 使用 PostgreSQL + S3 后端的生产服务器
mlflow server \
  --backend-store-uri postgresql://mlflow:password@localhost:5432/mlflow \
  --default-artifact-root s3://mlflow-artifacts/ \
  --host 0.0.0.0 --port 5000

用 Docker Compose 部署

# docker-compose.yml
services:
  mlflow:
    image: ghcr.io/mlflow/mlflow:v2.18.0
    ports:
      - '5000:5000'
    environment:
      - MLFLOW_BACKEND_STORE_URI=postgresql://mlflow:password@postgres:5432/mlflow
      - MLFLOW_DEFAULT_ARTIFACT_ROOT=s3://mlflow-artifacts/
      - AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
      - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
    command: >
      mlflow server
      --backend-store-uri postgresql://mlflow:password@postgres:5432/mlflow
      --default-artifact-root s3://mlflow-artifacts/
      --host 0.0.0.0 --port 5000
    depends_on:
      - postgres

  postgres:
    image: postgres:16
    environment:
      POSTGRES_USER: mlflow
      POSTGRES_PASSWORD: password
      POSTGRES_DB: mlflow
    volumes:
      - pgdata:/var/lib/postgresql/data

volumes:
  pgdata:

实验追踪(Tracking)

基本用法

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, f1_score, precision_score

# 配置追踪服务器
mlflow.set_tracking_uri("http://localhost:5000")

# 创建/设置实验
mlflow.set_experiment("iris-classification")

# 准备数据
X, y = load_iris(return_X_y=True)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 运行实验
with mlflow.start_run(run_name="rf-baseline"):
    # 记录参数
    params = {
        "n_estimators": 100,
        "max_depth": 5,
        "min_samples_split": 2,
        "random_state": 42
    }
    mlflow.log_params(params)

    # 训练模型
    model = RandomForestClassifier(**params)
    model.fit(X_train, y_train)

    # 预测与指标
    y_pred = model.predict(X_test)
    metrics = {
        "accuracy": accuracy_score(y_test, y_pred),
        "f1_macro": f1_score(y_test, y_pred, average="macro"),
        "precision_macro": precision_score(y_test, y_pred, average="macro")
    }
    mlflow.log_metrics(metrics)

    # 标签
    mlflow.set_tag("model_type", "random_forest")
    mlflow.set_tag("dataset", "iris")

    # 保存模型
    mlflow.sklearn.log_model(
        model,
        artifact_path="model",
        registered_model_name="iris-classifier"
    )

    # 自定义产物(图表、报告等)
    import matplotlib.pyplot as plt
    from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay

    cm = confusion_matrix(y_test, y_pred)
    fig, ax = plt.subplots()
    ConfusionMatrixDisplay(cm).plot(ax=ax)
    fig.savefig("confusion_matrix.png")
    mlflow.log_artifact("confusion_matrix.png")

    print(f"Run ID: {mlflow.active_run().info.run_id}")
    print(f"Metrics: {metrics}")

超参数调优追踪

import optuna
import mlflow

def objective(trial):
    params = {
        "n_estimators": trial.suggest_int("n_estimators", 50, 500),
        "max_depth": trial.suggest_int("max_depth", 2, 20),
        "min_samples_split": trial.suggest_int("min_samples_split", 2, 10),
        "min_samples_leaf": trial.suggest_int("min_samples_leaf", 1, 5),
    }

    with mlflow.start_run(nested=True, run_name=f"trial-{trial.number}"):
        mlflow.log_params(params)

        model = RandomForestClassifier(**params, random_state=42)
        model.fit(X_train, y_train)
        y_pred = model.predict(X_test)

        accuracy = accuracy_score(y_test, y_pred)
        mlflow.log_metric("accuracy", accuracy)

        return accuracy

# 运行 Optuna Study
with mlflow.start_run(run_name="hyperparameter-tuning"):
    study = optuna.create_study(direction="maximize")
    study.optimize(objective, n_trials=50)

    # 记录最优结果
    mlflow.log_params(study.best_params)
    mlflow.log_metric("best_accuracy", study.best_value)
    mlflow.set_tag("best_trial", study.best_trial.number)

PyTorch 模型追踪

import torch
import torch.nn as nn
import mlflow.pytorch

class SimpleNet(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim):
        super().__init__()
        self.fc1 = nn.Linear(input_dim, hidden_dim)
        self.relu = nn.ReLU()
        self.fc2 = nn.Linear(hidden_dim, output_dim)

    def forward(self, x):
        return self.fc2(self.relu(self.fc1(x)))

with mlflow.start_run(run_name="pytorch-model"):
    model = SimpleNet(4, 32, 3)
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
    criterion = nn.CrossEntropyLoss()

    mlflow.log_params({
        "hidden_dim": 32,
        "learning_rate": 0.001,
        "optimizer": "Adam",
        "epochs": 100
    })

    for epoch in range(100):
        # 训练逻辑...
        loss = criterion(model(X_tensor), y_tensor)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()

        # 按 epoch 记录指标
        mlflow.log_metric("train_loss", loss.item(), step=epoch)

    # 保存 PyTorch 模型
    mlflow.pytorch.log_model(model, "model")

Model Registry

模型注册与版本管理

from mlflow import MlflowClient

client = MlflowClient()

# 注册模型(在 log_model 中使用 registered_model_name 时会自动注册)
# 或手动注册：
result = client.create_registered_model(
    name="iris-classifier",
    description="鸢尾花分类模型"
)

# 将特定 run 的模型注册为一个版本
model_version = client.create_model_version(
    name="iris-classifier",
    source=f"runs:/{run_id}/model",
    run_id=run_id,
    description="RandomForest baseline v1"
)

print(f"Model Version: {model_version.version}")

用 Alias 进行部署管理

# MLflow 2.x 使用 Alias(Stage 已弃用)
client = MlflowClient()

# 设置生产环境 alias
client.set_registered_model_alias(
    name="iris-classifier",
    alias="champion",
    version=3
)

# 设置挑战者模型
client.set_registered_model_alias(
    name="iris-classifier",
    alias="challenger",
    version=5
)

# 通过 Alias 加载模型
champion_model = mlflow.pyfunc.load_model("models:/iris-classifier@champion")
challenger_model = mlflow.pyfunc.load_model("models:/iris-classifier@challenger")

# A/B 测试
champion_pred = champion_model.predict(X_test)
challenger_pred = challenger_model.predict(X_test)

print(f"Champion accuracy: {accuracy_score(y_test, champion_pred)}")
print(f"Challenger accuracy: {accuracy_score(y_test, challenger_pred)}")

模型标签的使用

# 为模型版本添加标签
client.set_model_version_tag(
    name="iris-classifier",
    version=3,
    key="validation_status",
    value="approved"
)

client.set_model_version_tag(
    name="iris-classifier",
    version=3,
    key="approved_by",
    value="data-science-lead"
)

# 按标签搜索模型
from mlflow import search_model_versions

approved_versions = search_model_versions(
    "name='iris-classifier' AND tag.validation_status='approved'"
)

模型服务(Serving)

MLflow 内置服务

# 本地 REST API 服务
mlflow models serve \
  -m "models:/iris-classifier@champion" \
  --port 8080 \
  --no-conda

# 测试请求
curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{"inputs": [[5.1, 3.5, 1.4, 0.2]]}'

用 FastAPI 自定义服务

from fastapi import FastAPI
import mlflow.pyfunc
import numpy as np

app = FastAPI()

# 加载模型(服务器启动时执行一次)
model = mlflow.pyfunc.load_model("models:/iris-classifier@champion")

@app.post("/predict")
async def predict(features: list[list[float]]):
    predictions = model.predict(np.array(features))
    return {
        "predictions": predictions.tolist(),
        "model_version": "champion"
    }

@app.get("/health")
async def health():
    return {"status": "healthy", "model": "iris-classifier@champion"}

实验比较与分析

在 MLflow UI 中比较

# 搜索实验(CLI)
mlflow runs list --experiment-id 1

# 按指标搜索
mlflow runs list \
  --experiment-id 1 \
  --filter "metrics.accuracy > 0.95" \
  --order-by "metrics.accuracy DESC"

用 Python API 分析

import mlflow
import pandas as pd

# 查询实验中的全部 run
runs = mlflow.search_runs(
    experiment_ids=["1"],
    filter_string="metrics.accuracy > 0.9",
    order_by=["metrics.accuracy DESC"],
    max_results=10
)

# 以 DataFrame 形式分析
print(runs[["run_id", "params.n_estimators", "params.max_depth", "metrics.accuracy"]])

# 查找最优 run
best_run = runs.iloc[0]
print(f"Best run: {best_run.run_id}, Accuracy: {best_run['metrics.accuracy']}")

生产环境检查清单

□ 将后端存储设置为 PostgreSQL/MySQL
□ 将产物存储设置为 S3/GCS/MinIO
□ 配置认证/授权(OIDC、Basic Auth)
□ 设置自动实验记录(autolog)
□ 制定 Model Registry 的 alias 规范
□ 在 CI/CD 中自动化模型验证
□ 配置模型服务的健康检查
□ 制定实验清理策略(归档旧的 run)
📝 确认测验(6 题)

Q1. MLflow 的四个核心组件是什么？

Tracking、Projects、Models、Model Registry

Q2. mlflow.log_params 与 mlflow.log_metrics 的区别是什么？

log_params 记录训练超参数(字符串)，log_metrics 记录性能指标(数值)。指标可以通过 step 参数按 epoch 进行追踪。

Q3. MLflow 2.x 中用于模型部署管理的概念是什么？

Alias(例如 @champion、@challenger)。Stage 已被弃用。

Q4. nested=True 参数在什么时候使用？

用于像超参数调优这样，需要在父 run 内部记录多个子 run 的场景。

Q5. 为什么用 S3 作为产物存储？

把模型文件、图表等大体积产物保存在可扩展的对象存储中，便于团队间共享与版本管理。

Q6. mlflow.autolog() 的优点和缺点是什么？

优点：无需修改代码即可自动记录参数/指标/模型。缺点：可能会记录过多不必要的信息，自定义指标仍需单独记录。

测验

Q1：《MLflow 完全指南：从实验追踪到 Model Registry、生产部署》一文主要讨论的内容是什么？

通过动手实践，使用 MLflow 完成 ML 实验管理的完整工作流程。用 Tracking 记录实验，用 Model Registry 进行版本管理，直至完成生产部署。

Q2：MLflow 是什么？ MLflow 是一个用于管理 ML 生命周期的开源平台，由四个核心组件构成：MLflow Tracking：记录实验参数、 指标和产物 MLflow Projects：将可复现的 ML 代码打包 MLflow Models：将各种框架的模型打包为统一格式 MLflow Model Registry：模型版本管理与部署工作流

Q3：安装与服务器配置的关键步骤是什么？ 基础安装 用 Docker Compose 部署

Q4：实验追踪(Tracking)有哪些要点？ 基本用法 超参数调优追踪 PyTorch 模型追踪

Q5：Model Registry 是如何工作的？ 模型注册与版本管理 用 Alias 进行部署管理 模型标签的使用

현재 단락 (1/234)

MLflow 是一个用于管理 ML 生命周期的开源平台，由四个核心组件构成：

작성 글자: 0자원문 글자: 7,918자작성 단락: 0/234