MLflow Model Registry 2026 生产级部署教程：从 Postgres 后端到 CI/CD 流水线，避坑指南与成本优化

# 创建数据库和用户
sudo -u postgres psql

CREATE DATABASE mlflow_registry;
CREATE USER mlflow_user WITH PASSWORD 'your_strong_password_2026';
GRANT ALL PRIVILEGES ON DATABASE mlflow_registry TO mlflow_user;
\c mlflow_registry
GRANT ALL ON SCHEMA public TO mlflow_user;

mlflow server \
    --backend-store-uri postgresql://mlflow_user:your_strong_password_2026@your-rds-endpoint:5432/mlflow_registry \
    --default-artifact-root s3://your-mlflow-artifacts-bucket/models \
    --host 0.0.0.0 \
    --port 5000 \
    --workers 4 \
    --gunicorn-opts "--timeout 120 --keep-alive 5"

import mlflow
from mlflow.tracking import MlflowClient

mlflow.set_tracking_uri("http://your-mlflow-server:5000")
mlflow.set_experiment("fraud_detection_model_v3")

with mlflow.start_run() as run:
    # 训练代码...
    model = train_model(X_train, y_train)
    
    # 记录指标
    mlflow.log_metric("auc", 0.892)
    mlflow.log_metric("f1", 0.874)
    
    # 记录模型并注册到 Registry
    mlflow.sklearn.log_model(
        sk_model=model,
        artifact_path="model",
        registered_model_name="fraud_detection_xgboost"
    )
    
    # 设置别名（2026 年新特性，比 Stage 更灵活）
    client = MlflowClient()
    client.set_registered_model_alias(
        name="fraud_detection_xgboost",
        version=1,
        alias="champion"
    )

import mlflow.pyfunc

# 方式一：通过别名加载（推荐）
model = mlflow.pyfunc.load_model(
    model_uri="models:/fraud_detection_xgboost@champion"
)

# 方式二：通过版本号加载
model = mlflow.pyfunc.load_model(
    model_uri="models:/fraud_detection_xgboost/3"
)

# 方式三：通过 Stage 加载（老派）
model = mlflow.pyfunc.load_model(
    model_uri="models:/fraud_detection_xgboost/Production"
)

name: MLflow Model Promotion

on:
  workflow_dispatch:
    inputs:
      model_name:
        description: 'Model Registry Name'
        required: true
      model_version:
        description: 'Version to promote'
        required: true
      target_stage:
        description: 'Target stage (Staging/Production)'
        required: true
        default: 'Staging'

jobs:
  promote-model:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Promote model version
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}
        run: |
          pip install mlflow boto3
          
          # 验证模型指标是否达标
          python scripts/validate_model_metrics.py \
            --model-name ${{ github.event.inputs.model_name }} \
            --version ${{ github.event.inputs.model_version }}
          
          # 执行阶段转换
          mlflow models transition-model-stage \
            --model-uri "models:/${{ github.event.inputs.model_name }}/${{ github.event.inputs.model_version }}" \
            --stage ${{ github.event.inputs.target_stage }} \
            --archive-existing-versions

# scripts/auto_rollback.py
import mlflow
from mlflow.tracking import MlflowClient
import time

def monitor_and_rollback(model_name, version, threshold_metric="auc", threshold=0.85):
    client = MlflowClient()
    
    # 启动监控
    time.sleep(300)  # 给模型 5 分钟热身
    
    # 检查线上指标
    current_metric = get_online_metrics(model_name, version)  # 自定义函数
    
    if current_metric[threshold_metric] < threshold:
        print(f"⚠️ 模型 {model_name} v{version} 指标不达标，触发回滚")
        
        # 找到上一个生产版本
        previous_versions = client.search_model_versions(
            f"name='{model_name}' and stage='Archived'"
        )
        
        if previous_versions:
            best_previous = max(previous_versions, key=lambda v: v.version)
            client.transition_model_version_stage(
                name=model_name,
                version=best_previous.version,
                stage="Production"
            )
            print(f"✅ 已回滚到 v{best_previous.version}")

# 细粒度权限控制
from mlflow.tracking import MlflowClient

client = MlflowClient()

# 限制模型注册权限
client.create_registered_model(
    name="production_model",
    tags={
        "allowed_teams": "ml-core,data-science",
        "requires_approval": "true"
    }
)

# 审计日志
client.create_model_version(
    name="production_model",
    source="s3://artifacts/run-12345/model",
    run_id="12345",
    tags={
        "approved_by": "alice@company.com",
        "approval_date": "2026-07-05",
        "compliance_check": "passed"
    }
)

# 离线镜像 MLflow
docker pull mlflow/mlflow:2.15.0
docker save mlflow/mlflow:2.15.0 | gzip > mlflow-offline.tar.gz

# 传到内网仓库
docker tag mlflow/mlflow:2.15.0 your-private-registry/mlflow:2.15.0
docker push your-private-registry/mlflow:2.15.0

# 自动清理旧版本
def cleanup_old_versions(model_name, max_versions=10):
    client = MlflowClient()
    versions = client.search_model_versions(f"name='{model_name}'")
    
    # 按版本号排序，保留最新的 N 个
    sorted_versions = sorted(versions, key=lambda v: v.version, reverse=True)
    
    for version in sorted_versions[max_versions:]:
        if version.stage == "Archived":
            client.transition_model_version_stage(
                name=model_name,
                version=version.version,
                stage="Archived"
            )
            print(f"🗑️ 清理旧版本: {model_name} v{version.version}")