필사 모드: MLOps 与模型生命周期管理:MLflow、DVC 与 LLMOps 完全指南

# Kubeflow Pipeline 示例 - Level 1 CT 流水线
import kfp
from kfp import dsl

@dsl.component
def data_validation_op(data_path: str) -> bool:
    import great_expectations as ge
    ds = ge.read_csv(data_path)
    results = ds.expect_column_values_to_not_be_null("target")
    return results["success"]

@dsl.component
def train_model_op(data_path: str, model_output: str):
    import mlflow
    import sklearn
    # 模型训练逻辑
    pass

@dsl.pipeline(name="CT Pipeline")
def ct_pipeline(data_path: str):
    validation = data_validation_op(data_path=data_path)
    with dsl.Condition(validation.output == True):
        train_model_op(data_path=data_path, model_output="/models/")

源代码变更或数据触发
        ↓
CI 流水线(测试、构建)
        ↓
CD 流水线(部署流水线)
        ↓
CT 流水线(自动重训练)
        ↓
模型评估 → 通过/不通过
        ↓
注册进模型注册表
        ↓
Staging → Production 晋升
        ↓
监控与告警

import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score, f1_score
import numpy as np

# 配置 MLflow Tracking 服务器
mlflow.set_tracking_uri("http://mlflow-server:5000")
mlflow.set_experiment("fraud-detection-v2")

with mlflow.start_run(run_name="rf-baseline") as run:
    # 记录超参数
    params = {
        "n_estimators": 100,
        "max_depth": 10,
        "min_samples_split": 5,
        "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_score": f1_score(y_test, y_pred, average="weighted"),
    }
    mlflow.log_metrics(metrics)

    # 模型自动记录(使用 sklearn autolog 时)
    # mlflow.sklearn.autolog()

    # 保存模型
    mlflow.sklearn.log_model(
        sk_model=model,
        artifact_path="model",
        registered_model_name="fraud-detection",
        input_example=X_test[:5],
        signature=mlflow.models.infer_signature(X_train, y_pred)
    )

    # 自定义产出物
    import matplotlib.pyplot as plt
    from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay
    cm = confusion_matrix(y_test, y_pred)
    disp = ConfusionMatrixDisplay(confusion_matrix=cm)
    disp.plot()
    plt.savefig("confusion_matrix.png")
    mlflow.log_artifact("confusion_matrix.png")

    print(f"Run ID: {run.info.run_id}")
    print(f"Accuracy: {metrics['accuracy']:.4f}")

import mlflow

# 自动检测框架并记录
mlflow.autolog()

# PyTorch 专用 autolog
mlflow.pytorch.autolog(
    log_every_n_epoch=1,
    log_models=True,
    disable=False,
    exclusive=False,
    log_datasets=True
)

# XGBoost 专用 autolog
mlflow.xgboost.autolog(
    log_input_examples=True,
    log_model_signatures=True,
    log_models=True,
    log_datasets=True
)

# MLproject 文件
name: fraud-detection

conda_env: conda.yaml

entry_points:
  main:
    parameters:
      n_estimators: { type: int, default: 100 }
      max_depth: { type: int, default: 10 }
      data_path: { type: str, default: 'data/train.csv' }
    command: 'python train.py --n_estimators {n_estimators} --max_depth {max_depth} --data_path {data_path}'
  evaluate:
    parameters:
      model_uri: { type: str }
      test_data: { type: str }
    command: 'python evaluate.py --model_uri {model_uri} --test_data {test_data}'

import wandb
from wandb.integration.keras import WandbCallback

# 初始化 W&B
run = wandb.init(
    project="image-classification",
    config={
        "learning_rate": 0.001,
        "epochs": 50,
        "batch_size": 32,
        "architecture": "ResNet50"
    }
)

# 用 W&B Sweep 做超参数优化
sweep_config = {
    "method": "bayes",
    "metric": {"name": "val_accuracy", "goal": "maximize"},
    "parameters": {
        "learning_rate": {"min": 1e-5, "max": 1e-2},
        "batch_size": {"values": [16, 32, 64]},
        "dropout": {"min": 0.1, "max": 0.5}
    }
}
sweep_id = wandb.sweep(sweep_config, project="image-classification")
wandb.agent(sweep_id, function=train_fn, count=50)

# 初始化 DVC
git init
dvc init

# 配置远程存储(S3)
dvc remote add -d myremote s3://my-bucket/dvc-store
dvc remote modify myremote region us-east-1

# 添加数据
dvc add data/train.csv
git add data/train.csv.dvc .gitignore
git commit -m "Add training data v1"
dvc push

# 在其他环境中拉取数据
git pull
dvc pull

# dvc.yaml
stages:
  prepare:
    cmd: python src/prepare.py --input data/raw.csv --output data/processed/
    deps:
      - src/prepare.py
      - data/raw.csv
    outs:
      - data/processed/train.csv
      - data/processed/test.csv
    params:
      - prepare:
          - split_ratio
          - random_seed

  featurize:
    cmd: python src/featurize.py
    deps:
      - src/featurize.py
      - data/processed/train.csv
    outs:
      - data/features/train_features.pkl
    params:
      - featurize:
          - max_features
          - ngrams

  train:
    cmd: python src/train.py
    deps:
      - src/train.py
      - data/features/train_features.pkl
    outs:
      - models/model.pkl
    metrics:
      - reports/metrics.json:
          cache: false
    params:
      - train:
          - n_estimators
          - max_depth
          - random_seed

  evaluate:
    cmd: python src/evaluate.py
    deps:
      - src/evaluate.py
      - models/model.pkl
      - data/processed/test.csv
    metrics:
      - reports/eval_metrics.json:
          cache: false
    plots:
      - reports/plots/confusion_matrix.csv:
          cache: false

# params.yaml
prepare:
  split_ratio: 0.8
  random_seed: 42
featurize:
  max_features: 1000
  ngrams: 2
train:
  n_estimators: 100
  max_depth: 10
  random_seed: 42

# 运行流水线
dvc repro

# 创建实验分支
dvc exp run --set-param train.n_estimators=200 --name exp-200-trees

# 比较实验
dvc exp show

# 输出指标表
dvc metrics show
dvc metrics diff

# feature_repo/feature_store.yaml
project: fraud_detection
registry: data/registry.db
provider: local
online_store:
  type: redis
  connection_string: "localhost:6379"
offline_store:
  type: bigquery
  dataset: feast_dev

# feature_repo/features.py
from datetime import timedelta
from feast import Entity, Feature, FeatureView, FileSource, ValueType
from feast.types import Float32, Int64, String

# 定义实体
user = Entity(
    name="user_id",
    value_type=ValueType.INT64,
    description="用户 ID"
)

# 定义数据源
user_stats_source = FileSource(
    path="data/user_stats.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created"
)

# 定义特征视图
user_stats_fv = FeatureView(
    name="user_stats",
    entities=["user_id"],
    ttl=timedelta(days=7),
    features=[
        Feature(name="transaction_count_7d", dtype=Float32),
        Feature(name="avg_transaction_amount", dtype=Float32),
        Feature(name="days_since_last_login", dtype=Int64),
        Feature(name="account_age_days", dtype=Int64),
    ],
    online=True,
    source=user_stats_source,
    tags={"team": "fraud", "version": "v2"},
)

# 使用特征存储 - 训练
from feast import FeatureStore
import pandas as pd

store = FeatureStore(repo_path="feature_repo/")

# 获取训练数据(离线)
entity_df = pd.DataFrame({
    "user_id": [1001, 1002, 1003],
    "event_timestamp": pd.to_datetime(["2026-03-01", "2026-03-01", "2026-03-01"])
})

training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "user_stats:transaction_count_7d",
        "user_stats:avg_transaction_amount",
        "user_stats:days_since_last_login",
    ]
).to_df()

# 在线服务 - 实时特征查询
feature_vector = store.get_online_features(
    features=[
        "user_stats:transaction_count_7d",
        "user_stats:avg_transaction_amount",
    ],
    entity_rows=[{"user_id": 1001}]
).to_dict()

from evidently import ColumnMapping
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

# 生成特征漂移报告
report = Report(metrics=[DataDriftPreset()])
report.run(
    reference_data=reference_features,
    current_data=current_features,
    column_mapping=ColumnMapping(target="label")
)
report.save_html("feature_drift_report.html")

# 查看漂移结果
results = report.as_dict()
drifted_features = [
    col for col, info in results["metrics"][0]["result"]["drift_by_columns"].items()
    if info["drift_detected"]
]
print(f"检测到漂移的特征: {drifted_features}")

import mlflow
from mlflow.tracking import MlflowClient

client = MlflowClient()

# 注册新模型
model_uri = f"runs:/{run_id}/model"
model_version = mlflow.register_model(
    model_uri=model_uri,
    name="fraud-detection"
)

# 添加模型说明
client.update_registered_model(
    name="fraud-detection",
    description="支付欺诈检测模型 - 基于 RandomForest"
)

client.update_model_version(
    name="fraud-detection",
    version=model_version.version,
    description=f"Accuracy: 0.956, F1: 0.943 on test set"
)

# 切换到 Staging
client.transition_model_version_stage(
    name="fraud-detection",
    version=model_version.version,
    stage="Staging",
    archive_existing_versions=False
)

# 加载并校验 Staging 模型
staging_model = mlflow.pyfunc.load_model(
    model_uri="models:/fraud-detection/Staging"
)
staging_preds = staging_model.predict(X_val)
staging_accuracy = accuracy_score(y_val, staging_preds)

# 校验通过后晋升到 Production
if staging_accuracy > 0.95:
    client.transition_model_version_stage(
        name="fraud-detection",
        version=model_version.version,
        stage="Production",
        archive_existing_versions=True  # 归档之前的 Production 版本
    )
    print(f"模型 v{model_version.version} 已晋升到 Production")

from huggingface_hub import HfApi, Repository
from transformers import AutoModelForSequenceClassification, AutoTokenizer

api = HfApi()

# 上传模型
api.upload_folder(
    folder_path="./fine-tuned-model",
    repo_id="myorg/sentiment-classifier-v2",
    repo_type="model",
)

# 更新模型卡片
api.upload_file(
    path_or_fileobj="README.md",
    path_in_repo="README.md",
    repo_id="myorg/sentiment-classifier-v2",
)

# 为特定版本打标签
api.create_tag(
    repo_id="myorg/sentiment-classifier-v2",
    tag="v2.1.0",
    tag_message="Improved accuracy on edge cases"
)

# .github/workflows/ml-cicd.yml
name: ML CI/CD Pipeline

on:
  push:
    branches: [main, develop]
    paths:
      - 'src/**'
      - 'params.yaml'
      - 'dvc.yaml'
  schedule:
    - cron: '0 2 * * 1' # 每周一凌晨 2 点自动重训练

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'
      - name: Install dependencies
        run: pip install -r requirements.txt
      - name: Run unit tests
        run: pytest tests/ -v --cov=src
      - name: Data validation
        run: python src/validate_data.py

  train-and-evaluate:
    needs: test
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Configure DVC remote
        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 }}
      - name: Pull data
        run: dvc pull
      - name: Run DVC pipeline
        run: dvc repro
      - name: Log metrics to MLflow
        run: python src/log_results.py
        env:
          MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}
      - name: Check model performance gate
        run: |
          python src/check_performance_gate.py \
            --min-accuracy 0.95 \
            --min-f1 0.93
      - name: Push results
        run: |
          dvc push
          git add reports/metrics.json dvc.lock
          git commit -m "chore: update metrics [skip ci]"
          git push

  deploy-staging:
    needs: train-and-evaluate
    runs-on: ubuntu-latest
    if: github.ref == 'refs/heads/main'
    steps:
      - name: Promote model to Staging
        run: python src/promote_model.py --stage Staging
      - name: Run integration tests
        run: pytest tests/integration/ -v
      - name: Deploy to staging endpoint
        run: kubectl apply -f k8s/staging/

  deploy-production:
    needs: deploy-staging
    runs-on: ubuntu-latest
    environment: production
    steps:
      - name: Promote model to Production
        run: python src/promote_model.py --stage Production
      - name: Blue/Green deployment
        run: ./scripts/blue_green_deploy.sh
      - name: Smoke tests
        run: pytest tests/smoke/ -v

# src/check_retrain_trigger.py
import mlflow
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

def should_retrain(
    current_data,
    reference_data,
    performance_threshold=0.92,
    drift_threshold=0.3
) -> tuple[bool, str]:
    """判断是否需要重训练"""

    # 1. 基于性能的触发条件
    current_metrics = get_current_metrics()
    if current_metrics["accuracy"] < performance_threshold:
        return True, f"性能下降: accuracy={current_metrics['accuracy']:.3f}"

    # 2. 数据漂移触发条件
    report = Report(metrics=[DataDriftPreset()])
    report.run(reference_data=reference_data, current_data=current_data)
    results = report.as_dict()
    drift_share = results["metrics"][0]["result"]["share_of_drifted_columns"]

    if drift_share > drift_threshold:
        return True, f"数据漂移: 检测到 {drift_share:.1%} 的特征发生漂移"

    return False, "无需重训练"

def get_current_metrics():
    client = mlflow.tracking.MlflowClient()
    prod_model = client.get_latest_versions("fraud-detection", stages=["Production"])[0]
    run = client.get_run(prod_model.run_id)
    return {
        "accuracy": float(run.data.metrics.get("accuracy", 0)),
        "f1_score": float(run.data.metrics.get("f1_score", 0))
    }

import pandas as pd
from evidently.report import Report
from evidently.test_suite import TestSuite
from evidently import ColumnMapping
from evidently.metric_preset import (
    DataDriftPreset,
    DataQualityPreset,
    TargetDriftPreset,
    ClassificationPreset
)
from evidently.tests import (
    TestNumberOfDriftedColumns,
    TestShareOfDriftedColumns,
    TestColumnDrift
)

# 配置列映射
column_mapping = ColumnMapping(
    target="fraud_label",
    prediction="fraud_score",
    numerical_features=["amount", "transaction_count_7d", "avg_amount"],
    categorical_features=["merchant_category", "payment_method"]
)

# 综合漂移报告
report = Report(metrics=[
    DataDriftPreset(),
    DataQualityPreset(),
    TargetDriftPreset(),
    ClassificationPreset()
])
report.run(
    reference_data=reference_df,
    current_data=production_df,
    column_mapping=column_mapping
)
report.save_html("monitoring/report.html")

# 告警测试套件
test_suite = TestSuite(tests=[
    TestNumberOfDriftedColumns(lt=3),
    TestShareOfDriftedColumns(lt=0.3),
    TestColumnDrift(column_name="amount"),
    TestColumnDrift(column_name="transaction_count_7d"),
])
test_suite.run(
    reference_data=reference_df,
    current_data=production_df
)

# 测试失败时发送告警
results = test_suite.as_dict()
failed_tests = [t for t in results["tests"] if t["status"] == "FAIL"]
if failed_tests:
    send_alert(f"监控告警: {len(failed_tests)} 个测试失败")

# src/monitoring/metrics.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# 定义指标
prediction_counter = Counter(
    "model_predictions_total",
    "预测总数",
    ["model_version", "result"]
)
prediction_latency = Histogram(
    "model_prediction_latency_seconds",
    "预测延迟",
    buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0]
)
model_accuracy = Gauge(
    "model_accuracy_current",
    "当前模型准确率"
)
drift_score = Gauge(
    "feature_drift_score",
    "特征漂移分数",
    ["feature_name"]
)

# 在服务代码中更新指标
def predict_with_monitoring(features, model_version="v2.1"):
    start_time = time.time()
    prediction = model.predict(features)
    latency = time.time() - start_time

    prediction_latency.observe(latency)
    prediction_counter.labels(
        model_version=model_version,
        result="fraud" if prediction[0] == 1 else "normal"
    ).inc()

    return prediction

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langsmith import Client
import os

# LangSmith 配置
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "production-chatbot"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key"

# 定义 LangChain 链(自动记录到 LangSmith 的追踪中)
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
prompt = ChatPromptTemplate.from_template(
    "你是一名友善的客户服务代表。\n\n问题: {question}\n\n回答:"
)
chain = prompt | llm

# 执行 - 自动追踪
response = chain.invoke({"question": "退款政策是怎样的?"})

# 用 LangSmith 客户端做评估
langsmith_client = Client()

# 创建数据集
dataset = langsmith_client.create_dataset(
    dataset_name="customer-service-eval",
    description="客户服务聊天机器人评估数据集"
)

# 添加评估样例
langsmith_client.create_examples(
    inputs=[{"question": "请告诉我退款政策"}],
    outputs=[{"answer": "购买后 30 天内可以退款。"}],
    dataset_id=dataset.id
)

# 执行自动评估
from langsmith.evaluation import evaluate, LangChainStringEvaluator

evaluators = [
    LangChainStringEvaluator("cot_qa"),
    LangChainStringEvaluator("labeled_criteria", config={
        "criteria": "correctness"
    })
]
results = evaluate(
    lambda x: chain.invoke(x),
    data=dataset.name,
    evaluators=evaluators,
    experiment_prefix="gpt4o-baseline"
)

# prompt_registry.py
import mlflow
from dataclasses import dataclass
from typing import Optional

@dataclass
class PromptVersion:
    template: str
    version: str
    description: str
    metrics: Optional[dict] = None

class PromptRegistry:
    def __init__(self, mlflow_uri: str):
        mlflow.set_tracking_uri(mlflow_uri)
        self.experiment_name = "prompt-versions"
        mlflow.set_experiment(self.experiment_name)

    def register_prompt(self, prompt: PromptVersion) -> str:
        with mlflow.start_run(run_name=f"prompt-{prompt.version}") as run:
            mlflow.log_param("version", prompt.version)
            mlflow.log_param("description", prompt.description)
            mlflow.log_text(prompt.template, "prompt_template.txt")
            if prompt.metrics:
                mlflow.log_metrics(prompt.metrics)
            return run.info.run_id

    def get_prompt(self, version: str) -> str:
        client = mlflow.tracking.MlflowClient()
        runs = client.search_runs(
            experiment_ids=[mlflow.get_experiment_by_name(self.experiment_name).experiment_id],
            filter_string=f"params.version = '{version}'"
        )
        if not runs:
            raise ValueError(f"未找到提示词版本 {version}")
        artifact_uri = runs[0].info.artifact_uri
        return mlflow.artifacts.load_text(f"{artifact_uri}/prompt_template.txt")

# 使用示例
registry = PromptRegistry("http://mlflow-server:5000")
registry.register_prompt(PromptVersion(
    template="你是一名 {role}。{context}\n\n问题: {question}\n回答:",
    version="v1.2.0",
    description="加入上下文的提示词改进版",
    metrics={"accuracy": 0.87, "hallucination_rate": 0.03}
))

# fine_tuning_pipeline.py
from transformers import (
    AutoModelForCausalLM,
    AutoTokenizer,
    TrainingArguments,
    Trainer
)
from peft import LoraConfig, get_peft_model, TaskType
import mlflow

def fine_tune_with_lora(
    base_model: str,
    dataset_path: str,
    output_dir: str,
    lora_r: int = 16,
    lora_alpha: int = 32
):
    mlflow.set_experiment("llm-fine-tuning")

    with mlflow.start_run():
        # LoRA 配置
        lora_config = LoraConfig(
            task_type=TaskType.CAUSAL_LM,
            r=lora_r,
            lora_alpha=lora_alpha,
            target_modules=["q_proj", "v_proj"],
            lora_dropout=0.05,
            bias="none"
        )
        mlflow.log_params({
            "base_model": base_model,
            "lora_r": lora_r,
            "lora_alpha": lora_alpha
        })

        # 准备模型
        model = AutoModelForCausalLM.from_pretrained(base_model)
        model = get_peft_model(model, lora_config)
        model.print_trainable_parameters()

        # 训练配置
        training_args = TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=3,
            per_device_train_batch_size=4,
            gradient_accumulation_steps=4,
            learning_rate=2e-4,
            fp16=True,
            report_to="mlflow"
        )

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

        # 保存并注册模型
        model.save_pretrained(output_dir)
        mlflow.transformers.log_model(
            transformers_model={"model": model, "tokenizer": tokenizer},
            artifact_path="fine-tuned-model",
            registered_model_name="customer-service-llm"
        )