ML 流水线设计与自动化

from enum import Enum

class MLOpsLevel(Enum):
    LEVEL_0 = "手动流程"
    LEVEL_1 = "ML 流水线自动化"
    LEVEL_2 = "CI/CD 流水线自动化"
    LEVEL_3 = "持续训练"


def assess_mlops_level(
    has_pipeline: bool,
    has_cicd: bool,
    has_continuous_training: bool
) -> MLOpsLevel:
    """评估当前 ML 项目的 MLOps 成熟度等级"""
    if has_continuous_training and has_cicd and has_pipeline:
        return MLOpsLevel.LEVEL_3
    elif has_cicd and has_pipeline:
        return MLOpsLevel.LEVEL_2
    elif has_pipeline:
        return MLOpsLevel.LEVEL_1
    else:
        return MLOpsLevel.LEVEL_0

level = assess_mlops_level(
    has_pipeline=True,
    has_cicd=True,
    has_continuous_training=False
)
print(f"Current level: {level.value}")

# mlops-maturity-checklist.yaml
level_0:
  - 手动数据探索
  - Jupyter Notebook 训练
  - 手工导出模型文件
level_1:
  - 自动化数据预处理
  - 可复现的训练脚本
  - 实验参数记录
level_2:
  - Git 触发训练流水线
  - 自动化模型评估
  - 一键部署到生产
level_3:
  - 数据漂移自动检测
  - 定时或事件触发重训练
  - 模型自动回滚机制

# 初始化 DVC 并连接远程存储
dvc init
dvc remote add -d myremote s3://my-bucket/dvc-store

# 追踪大型数据集
dvc add data/raw/training_data.parquet
dvc add data/processed/features.pkl

# 提交元数据到 Git（不是数据本身）
git add data/raw/training_data.parquet.dvc
git add data/processed/features.pkl.dvc
git commit -m "Track training data v2.1"

# 推送数据到远程存储
dvc push

# 在另一台机器上拉取数据
dvc pull

# dvc.yaml - 定义数据处理流水线
stages:
  data_ingestion:
    cmd: python scripts/ingest.py --config configs/ingest.yaml
    deps:
      - scripts/ingest.py
      - configs/ingest.yaml
    outs:
      - data/raw/dataset.csv

  feature_engineering:
    cmd: python scripts/features.py
    deps:
      - scripts/features.py
      - data/raw/dataset.csv
    outs:
      - data/processed/features.pkl

  train_model:
    cmd: python scripts/train.py --epochs 50
    deps:
      - scripts/train.py
      - data/processed/features.pkl
    outs:
      - models/model_v1.pkl
    metrics:
      - metrics/eval.json:

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

mlflow.set_tracking_uri("http://localhost:5000")
mlflow.set_experiment("customer_churn_prediction")

with mlflow.start_run(run_name="rf_baseline") as run:
    # 记录参数
    params = {"n_estimators": 100, "max_depth": 10, "random_state": 42}
    mlflow.log_params(params)

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

    # 记录指标
    accuracy = accuracy_score(y_test, model.predict(X_test))
    mlflow.log_metric("accuracy", accuracy)
    mlflow.log_metric("f1_score", f1_score(y_test, model.predict(X_test), average="weighted"))

    # 保存模型 artifact
    mlflow.sklearn.log_model(model, "model")
    print(f"Run ID: {run.info.run_id}")

import wandb
from transformers import Trainer, TrainingArguments

# 初始化 W&B 实验
wandb.init(project="sentiment-finetune", config={
    "learning_rate": 2e-5,
    "num_train_epochs": 3,
    "batch_size": 32,
    "model_name": "bert-base-chinese"
})

training_args = TrainingArguments(
    output_dir="./results",
    num_train_epochs=3,
    per_device_train_batch_size=32,
    report_to="wandb"  # 自动上报到 W&B
)

trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=val_dataset
)
trainer.train()

# 记录自定义指标
wandb.log({"final_loss": trainer.state.log_history[-1]["loss"]})
wandb.finish()

import mlflow
from mlflow.tracking import MlflowClient

client = MlflowClient()

# 将实验运行的模型注册到注册表
result = mlflow.register_model(
    model_uri="runs:/<run_id>/model",
    name="churn_prediction_model"
)
print(f"Registered version: {result.version}")

# 为模型版本添加描述
client.update_model_version(
    name="churn_prediction_model",
    version=result.version,
    description="Random Forest baseline - accuracy 0.89"
)

# 推进到 Staging 阶段
client.transition_model_version_stage(
    name="churn_prediction_model",
    version=result.version,
    stage="Staging"
)

import mlflow.pyfunc

# 从注册表加载生产阶段的最佳模型
model_name = "churn_prediction_model"
model_uri = f"models:/{model_name}/Production"

loaded_model = mlflow.pyfunc.load_model(model_uri)

# 获取模型签名以验证输入格式
model_info = mlflow.models.get_model_info(model_uri)
print(f"Model signature: {model_info.signature}")

# 批量预测
import pandas as pd
test_data = pd.DataFrame({
    "age": [35, 42, 28],
    "tenure": [24, 6, 12],
    "monthly_charges": [85.0, 95.5, 70.0]
})

predictions = loaded_model.predict(test_data)
print(f"Predictions: {predictions}")

# .github/workflows/ml-pipeline.yaml
name: ML CI/CD Pipeline

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test-code:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run unit tests
        run: pytest tests/ -v

  validate-data:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Validate data schema
        run: python scripts/validate_data.py

  train-and-evaluate:
    needs: [test-code, validate-data]
    runs-on: gpu-runner
    steps:
      - uses: actions/checkout@v4
      - name: Train model
        run: python scripts/train.py --config configs/prod.yaml
      - name: Evaluate model
        run: python scripts/evaluate.py --threshold 0.85

  deploy:
    needs: [train-and-evaluate]
    if: github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    steps:
      - name: Deploy to production
        run: python scripts/deploy.py --env production

import pytest
import mlflow
import numpy as np

def test_model_performance_threshold():
    """测试模型性能是否达到最低要求"""
    model = mlflow.sklearn.load_model("models:/churn_model/Production")
    predictions = model.predict(X_test)
    accuracy = accuracy_score(y_test, predictions)
    assert accuracy >= 0.85, f"Accuracy {accuracy} below threshold 0.85"

def test_no_data_leakage():
    """确保训练集和测试集没有重叠"""
    train_ids = set(X_train.index)
    test_ids = set(X_test.index)
    overlap = train_ids & test_ids
    assert len(overlap) == 0, f"Found {len(overlap)} overlapping samples"

def test_feature_distribution():
    """检查特征分布是否合理"""
    assert X_train["age"].between(0, 120).all()
    assert X_train["monthly_charges"].min() >= 0
    assert not X_train.isnull().any().any(), "Found null values in features"

from evidently.report import Report
from evidently.metric_preset import DataDriftPreset
import pandas as pd

# 计算数据漂移报告
reference_data = pd.read_csv("data/reference_monthly.csv")
current_data = pd.read_csv("data/current_monthly.csv")

drift_report = Report(metrics=[DataDriftPreset()])
drift_report.run(reference_data=reference_data,
                 current_data=current_data)

# 提取漂移检测结果
report_dict = drift_report.as_dict()
drifted_features = []
for feature, result in report_dict["metrics"][0]["result"]["drift_by_columns"].items():
    if result["drift_detected"]:
        drifted_features.append(feature)

if len(drifted_features) > 3:
    print(f"DRIFT ALERT: {len(drifted_features)} features drifted")
    # 触发持续训练
    trigger_continuous_training()

import schedule
import time
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)

def continuous_training_pipeline():
    """持续训练主函数"""
    logging.info(f"Starting CT pipeline at {datetime.now()}")

    # 1. 检查是否需要训练
    if not should_trigger_training():
        logging.info("No training trigger needed")
        return

    # 2. 拉取最新数据
    data = fetch_latest_training_data()

    # 3. 训练新模型
    new_model = train_model(data)

    # 4. 与生产模型对比
    champion = load_production_model()
    if evaluate_and_compare(new_model, champion):
        # 5. 注册并部署
        register_and_deploy(new_model)
        logging.info("New model deployed successfully")
    else:
        logging.warning("New model did not outperform champion")

# 每天凌晨 2 点执行
schedule.every().day.at("02:00").do(continuous_training_pipeline)

while True:
    schedule.run_pending()
    time.sleep(60)

import kfp
from kfp import dsl
from kfp.dsl import component

@component(
    packages_to_install=["pandas", "scikit-learn"],
    base_image="python:3.11"
)
def preprocess_data(input_path: str, output_path: dsl.OutputPath("Dataset")):
    import pandas as pd
    from sklearn.model_selection import train_test_split

    df = pd.read_csv(input_path)
    df = df.dropna()
    train_df, test_df = train_test_split(df, test_size=0.2)
    train_df.to_csv(f"{output_path}_train.csv", index=False)
    test_df.to_csv(f"{output_path}_test.csv", index=False)

@component(
    packages_to_install=["scikit-learn", "joblib"],
    base_image="python:3.11"
)
def train_model(
    train_path: dsl.InputPath("Dataset"),
    model_path: dsl.OutputPath("Model"),
    n_estimators: int = 100
):
    import pandas as pd
    from sklearn.ensemble import RandomForestClassifier
    import joblib

    train_df = pd.read_csv(train_path)
    X = train_df.drop("target", axis=1)
    y = train_df["target"]
    model = RandomForestClassifier(n_estimators=n_estimators)
    model.fit(X, y)
    joblib.dump(model, model_path)

@dsl.pipeline(name="ml-training-pipeline", description="End-to-end ML pipeline")
def ml_pipeline(data_path: str, n_estimators: int = 100):
    preprocess_task = preprocess_data(input_path=data_path)
    train_model(
        train_path=preprocess_task.outputs["output_path"],
        model_path="model.pkl",
        n_estimators=n_estimators
    )

import kfp

# 连接到 Kubeflow Pipelines 服务器
client = kfp.Client(host="http://kubeflow.example.com")

# 编译流水线定义
from kfp import compiler
compiler.Compiler().compile(
    pipeline_func=ml_pipeline,
    package_path="ml_pipeline.yaml"
)

# 创建实验并启动运行
experiment = client.create_experiment(name="production-runs")

run = client.run_pipeline(
    experiment_id=experiment.id,
    job_name="churn-model-v2",
    pipeline_package_path="ml_pipeline.yaml",
    params={
        "data_path": "gs://my-bucket/data/churn_data.csv",
        "n_estimators": 200
    }
)
print(f"Run ID: {run.run_id}")

# 查看运行状态
run_detail = client.get_run(run.run_id)
print(f"Status: {run_detail.state}")