12. MLOps 与实验管理

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score

# 设置追踪 URI
mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("customer-churn-prediction")

with mlflow.start_run(run_name="rf_baseline"):
    # 记录参数
    mlflow.log_param("n_estimators", 200)
    mlflow.log_param("max_depth", 10)
    mlflow.log_param("random_state", 42)
    
    # 训练
    model = RandomForestClassifier(n_estimators=200, max_depth=10, random_state=42)
    model.fit(X_train, y_train)
    
    # 记录指标
    train_acc = accuracy_score(y_train, model.predict(X_train))
    test_acc = accuracy_score(y_test, model.predict(X_test))
    mlflow.log_metric("train_accuracy", train_acc)
    mlflow.log_metric("test_accuracy", test_acc)
    mlflow.log_metric("overfitting", train_acc - test_acc)
    
    # 记录模型
    mlflow.sklearn.log_model(model, "model")
    
    # 记录工件
    import matplotlib.pyplot as plt
    plt.figure(figsize=(10, 6))
    pd.Series(model.feature_importances_, index=X.columns).sort_values().plot(kind='barh')
    plt.savefig("feature_importance.png")
    mlflow.log_artifact("feature_importance.png")

from sklearn.model_selection import GridSearchCV
import numpy as np

mlflow.set_experiment("hyperparameter-tuning")

with mlflow.start_run(run_name="xgb_grid_search"):
    param_grid = {
        'max_depth': [3, 6, 9],
        'learning_rate': [0.01, 0.1, 0.3],
        'n_estimators': [100, 200]
    }
    
    for i, params in enumerate(ParameterGrid(param_grid)):
        with mlflow.start_run(nested=True, run_name=f"run_{i}"):
            mlflow.log_params(params)
            
            model = XGBClassifier(**params, random_state=42)
            model.fit(X_train, y_train)
            
            train_f1 = f1_score(y_train, model.predict(X_train), average='macro')
            val_f1 = f1_score(y_val, model.predict(X_val), average='macro')
            
            mlflow.log_metric("train_f1", train_f1)
            mlflow.log_metric("val_f1", val_f1)
            mlflow.log_metric("gap", train_f1 - val_f1)
            
            mlflow.sklearn.log_model(model, "model")

# 为支持的框架自动记录参数、指标、模型
mlflow.sklearn.autolog()
mlflow.tensorflow.autolog()
mlflow.pytorch.autolog()
mlflow.xgboost.autolog()

# 只需正常训练，MLflow 自动追踪
model = XGBClassifier(n_estimators=100)
model.fit(X_train, y_train, eval_set=[(X_val, y_val)])
# 参数、指标、模型自动记录

# 启动 Tracking Server
mlflow server \
  --backend-store-uri postgresql://user:pass@localhost/mlflow \
  --default-artifact-root s3://mlflow-artifacts \
  --host 0.0.0.0 \
  --port 5000

# 或使用本地文件系统（开发环境）
mlflow server --backend-store-uri sqlite:///mlflow.db \
  --default-artifact-root ./mlruns \
  --host 0.0.0.0

# 初始化
git init
dvc init
git commit -m "Initialize DVC"

# 添加数据到版本控制
dvc add data/train.csv
# 生成 train.csv.dvc 和 .gitignore，将大文件存入本地缓存

git add data/train.csv.dvc data/.gitignore
git commit -m "Add training data"

# 配置远程存储
dvc remote add -d myremote s3://mybucket/dvcstore
dvc push  # 上传到远程

# 团队协作
git pull
dvc pull  # 拉取数据

# dvc.yaml
stages:
  prepare:
    cmd: python src/prepare.py data/raw data/prepared
    deps:
      - src/prepare.py
      - data/raw
    outs:
      - data/prepared
  
  train:
    cmd: python src/train.py data/prepared model.pkl
    deps:
      - src/train.py
      - data/prepared
    params:
      - train.epochs
      - train.lr
    outs:
      - model.pkl
    metrics:
      - metrics.json:
          cache: false
  
  evaluate:
    cmd: python src/evaluate.py model.pkl data/prepared
    deps:
      - src/evaluate.py
      - model.pkl
      - data/prepared
    metrics:
      - evaluation/metrics.json:
          cache: false
    plots:
      - evaluation/plots:
          cache: false

# 运行流水线
dvc repro

# 对比实验
dvc metrics diff  # 显示指标变化

# params.yaml
prepare:
  split: 0.2
  seed: 42

train:
  epochs: 50
  lr: 0.001
  batch_size: 32
  dropout: 0.3

evaluate:
  threshold: 0.5

# 实验矩阵：批量运行不同参数组合
for lr in 0.001 0.01 0.1; do
  dvc exp run --set-param train.lr=$lr
done

# 查看实验结果
dvc exp show

import mlflow
from mlflow.tracking import MlflowClient

client = MlflowClient()

# 注册模型
result = mlflow.register_model(
    "runs:/some_run_id/model",
    "churn-prediction-model"
)

# 模型版本管理
client.transition_model_version_stage(
    name="churn-prediction-model",
    version=1,
    stage="Staging"
)

client.transition_model_version_stage(
    name="churn-prediction-model",
    version=2,
    stage="Production"
)

# 加载生产环境模型
model = mlflow.pyfunc.load_model("models:/churn-prediction-model/Production")
predictions = model.predict(X_test)

# .github/workflows/ml-pipeline.yml
name: ML Pipeline

on:
  push:
    paths:
      - 'src/**'
      - 'data/**'
      - 'params.yaml'

jobs:
  prepare:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: iterative/setup-dvc@v1
      
      - name: Pull DVC data
        run: |
          dvc remote modify myremote access_key_id ${{ secrets.AWS_ACCESS_KEY_ID }}
          dvc remote modify myremote secret_access_key ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          dvc pull
      
      - name: Prepare data
        run: dvc repro prepare
      
      - uses: actions/upload-artifact@v4
        with:
          name: prepared-data
          path: data/prepared

  train:
    needs: prepare
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Download prepared data
        uses: actions/download-artifact@v4
        with:
          name: prepared-data
          path: data/prepared
      
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.10'
      
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Train model
        run: |
          dvc repro train
          echo "METRICS=$(cat metrics.json)" >> $GITHUB_ENV
      
      - name: Upload model
        uses: actions/upload-artifact@v4
        with:
          name: trained-model
          path: model.pkl

  evaluate:
    needs: train
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Download model
        uses: actions/download-artifact@v4
        with:
          name: trained-model
      
      - name: Evaluate
        run: dvc repro evaluate
      
      - name: Check metrics threshold
        run: |
          ACCURACY=$(jq '.accuracy' evaluation/metrics.json)
          if (( $(echo "$ACCURACY < 0.85" | bc -l) )); then
            echo "Accuracy $ACCURACY below threshold 0.85"
            exit 1
          fi

  deploy:
    needs: evaluate
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
      - name: Deploy to production
        run: |
          mlflow models serve -m models:/churn-prediction-model/Production -p 5001

# tests/test_model.py
import pytest
import numpy as np
from model import MyModel

class TestModel:
    def test_prediction_shape(self):
        model = MyModel.load('model.pkl')
        X = np.random.randn(10, 20)
        preds = model.predict(X)
        assert preds.shape == (10,)
    
    def test_prediction_range(self):
        model = MyModel.load('model.pkl')
        X = np.random.randn(10, 20)
        preds = model.predict(X)
        assert np.all((preds >= 0) & (preds <= 1))
    
    def test_invariance(self):
        """输入微小变化，输出不应剧变"""
        model = MyModel.load('model.pkl')
        X = np.random.randn(1, 20)
        X_perturbed = X + np.random.normal(0, 0.01, X.shape)
        
        pred1 = model.predict(X)
        pred2 = model.predict(X_perturbed)
        assert np.abs(pred1 - pred2) < 0.1
    
    def test_drift_detection(self):
        """检测数据漂移"""
        from scipy import stats
        
        # 当前批次 vs 训练数据
        _, p_value = stats.ks_2samp(current_batch, training_data)
        assert p_value > 0.05, "Data drift detected!"

from prometheus_client import Counter, Histogram, Gauge, start_http_server

# 定义指标
inference_counter = Counter('model_inferences_total', 'Total inferences')
latency_histogram = Histogram('model_latency_seconds', 'Inference latency')
prediction_drift = Gauge('model_prediction_drift', 'Prediction distribution drift')

class MonitoredModel:
    def __init__(self, model):
        self.model = model
        self.baseline_preds = []
    
    @latency_histogram.time()
    def predict(self, X):
        inference_counter.inc()
        return self.model.predict(X)
    
    def check_drift(self, recent_preds, threshold=0.05):
        from scipy import stats
        _, p_value = stats.ks_2samp(recent_preds, self.baseline_preds)
        prediction_drift.set(1 - p_value)
        return p_value < threshold

# 启动指标服务
start_http_server(8000)

# alert_rules.yml
groups:
  - name: ml-model
    rules:
      - alert: HighPredictionLatency
        expr: histogram_quantile(0.99, rate(model_latency_seconds_bucket[5m])) > 0.5
        for: 5m
        annotations:
          summary: "模型推理延迟过高"
      
      - alert: ModelDriftDetected
        expr: model_prediction_drift > 0.1
        for: 10m
        annotations:
          summary: "模型预测分布漂移，需重新训练"
      
      - alert: LowPredictionConfidence
        expr: avg(model_prediction_entropy) > 2.0
        for: 15m
        annotations:
          summary: "模型预测不确定性增加"

# pipeline.yaml
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
  name: ml-pipeline
spec:
  entrypoint: training-pipeline
  templates:
    - name: training-pipeline
      steps:
        - - name: data-preparation
            template: prepare-data
        - - name: model-training
            template: train-model
            arguments:
              parameters:
                - name: prepared-data
                  value: "{{steps.data-preparation.outputs.artifacts.data}}"
        - - name: model-evaluation
            template: evaluate-model
        - - name: model-deployment
            template: deploy-model
            when: "{{steps.model-evaluation.outputs.parameters.accuracy}} > 0.85"

# inference-service.yaml
apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: sklearn-iris
spec:
  predictor:
    sklearn:
      storageUri: "s3://models/sklearn/iris"
      resources:
        limits:
          cpu: "1"
          memory: 2Gi
    minReplicas: 1
    maxReplicas: 10
    scaleTarget: 1  # 并发请求数
    canaryTrafficPercent: 10  # 金丝雀发布

# Feast 特征服务
from feast import FeatureStore
import pandas as pd

store = FeatureStore(repo_path=".")

# 离线特征（训练）
training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_stats:avg_purchase_amount",
        "user_stats:total_orders",
        "item_stats:ctr_7d"
    ]
).to_df()

# 在线特征（推理）
online_features = store.get_online_features(
    features=["user_stats:avg_purchase_amount"],
    entity_rows=[{"user_id": 123}]
).to_dict()