Claude Code ML Pipeline全链路实战

在动手编码之前，先用Claude Code规划整个Pipeline架构。Claude Code擅长生成Mermaid流程图和架构设计：

项目	要求
Claude Code	已安装并配置（`npm install -g @anthropic-ai/claude-code`）
Python	3.10+
关键依赖	`mlflow>=2.18`, `scikit-learn`, `pandas`, `great_expectations`, `dvc`
操作系统	macOS / Linux / WSL2
推荐	使用虚拟环境（`uv venv` 或 `conda`）

graph TD
    A[原始数据 CSV/Parquet] --> B[数据校验<br/>Great Expectations]
    B -->|通过| C[特征工程<br/>Pandas/Sklearn Pipeline]
    B -->|失败| B1[告警 & 日志]
    C --> D[数据版本控制<br/>DVC]
    D --> E[模型训练<br/>Sklearn/XGBoost]
    E --> F[实验追踪<br/>MLflow]
    F --> G[模型校验<br/>阈值判断]
    G -->|Pass| H[模型注册<br/>MLflow Registry]
    G -->|Fail| E
    H --> I[模型部署<br/>FastAPI/Docker]
    I --> J[监控 & 告警<br/>Prometheus/Grafana]

    style A fill:#e1f5fe
    style F fill:#fff3e0
    style H fill:#e8f5e9
    style I fill:#fce4ec

核心设计原则： 1. 数据优先校验——任何训练开始前先跑数据质量检查 2. 全链路可追溯——MLflow记录每次实验的参数、指标、产物 3. 显式阈值门控——模型必须通过评估门槛才能进入注册阶段 4. 训练与推理分离——训练脚本和推理服务各自独立部署

三、Step 1：数据校验——Pipeline的第一道关卡

3.1 创建数据校验模块

请创建 src/data_validation.py，实现以下功能：
1. 使用 Great Expectations 对CSV数据进行校验
2. 检查：label列无空值、数值特征范围在0-1之间、行数>100
3. 校验失败时抛出明确异常
4. 返回校验通过的DataFrame

# src/data_validation.py
import pandas as pd
import great_expectations as gx
from typing import Dict, Any


def validate_training_data(filepath: str) -> pd.DataFrame:
    """对训练数据执行Schema和分布校验。校验失败直接中断。"""
    df = pd.read_csv(filepath)

    context = gx.get_context()
    data_source = context.data_sources.add_pandas("training_data")
    data_asset = data_source.add_dataframe_asset(name="raw_data")
    batch_definition = data_asset.add_batch_definition_whole_dataframe("batch")

    # 定义校验规则
    suite = context.suites.add(gx.ExpectationSuite(name="training_suite"))

    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToNotBeNull(column="label")
    )
    suite.add_expectation(
        gx.expectations.ExpectTableRowCountToBeGreaterThanOrEqual(min_value=100)
    )
    suite.add_expectation(
        gx.expectations.ExpectColumnValuesToBeBetween(
            column="feature_1", min_value=0, max_value=1
        )
    )

    batch = batch_definition.get_batch(batch_parameters={"dataframe": df})
    results = batch.validate(suite)

    if not results.success:
        failed = [r for r in results.results if not r.success]
        msg = "; ".join(
            f"{r.expectation_config.type}: {r.result}" for r in failed
        )
        raise ValueError(f"数据校验失败 — {msg}")

    print(f"[OK] 数据校验通过: {len(df)} 行, {len(df.columns)} 列")
    return df

3.2 常见问题与解决方案

四、Step 2：特征工程——构建可复用的转换Pipeline

4.1 用Claude Code生成特征Pipeline

Claude Code的核心优势在于理解业务语义并生成符合sklearn Pipeline规范的代码：

问题	原因	解决
`gx.get_context()` 报错	Great Expectations版本不兼容	`uv pip install "great_expectations>=0.18"`
校验超时	数据量过大	对大数据集先 `df.sample()` 做采样校验
列名不匹配	CSV列名含空格	`df.columns = df.columns.str.strip()` 预处理

请创建 src/feature_pipeline.py，要求：
1. 使用 sklearn Pipeline + ColumnTransformer
2. 数值特征：StandardScaler
3. 类别特征：OneHotEncoder(handle_unknown='ignore')
4. 输出是一个可序列化的Pipeline对象
5. 包含 get_feature_names() 方法以便后续追踪

# src/feature_pipeline.py
import pandas as pd
import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.model_selection import train_test_split
import joblib
import os


SEED = 42


def build_feature_pipeline(
    numeric_features: list[str],
    categorical_features: list[str],
) -> ColumnTransformer:
    """构建特征转换Pipeline，确保可序列化和可复现。"""

    numeric_transformer = Pipeline(steps=[
        ("imputer", SimpleImputer(strategy="median")),
        ("scaler", StandardScaler()),
    ])

    categorical_transformer = Pipeline(steps=[
        ("imputer", SimpleImputer(strategy="most_frequent")),
        ("encoder", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
    ])

    preprocessor = ColumnTransformer(
        transformers=[
            ("num", numeric_transformer, numeric_features),
            ("cat", categorical_transformer, categorical_features),
        ],
        verbose_feature_names_out=False,
    )
    return preprocessor


def prepare_data(
    df: pd.DataFrame,
    target_col: str = "label",
    test_size: float = 0.2,
) -> dict:
    """执行特征工程并返回切分后的数据。"""
    np.random.seed(SEED)

    feature_cols = [c for c in df.columns if c != target_col]
    numeric_features = df[feature_cols].select_dtypes(include=[np.number]).columns.tolist()
    categorical_features = df[feature_cols].select_dtypes(exclude=[np.number]).columns.tolist()

    preprocessor = build_feature_pipeline(numeric_features, categorical_features)

    X = df[feature_cols]
    y = df[target_col]

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=test_size, random_state=SEED, stratify=y
    )

    X_train_processed = preprocessor.fit_transform(X_train)
    X_test_processed = preprocessor.transform(X_test)

    # 保存Pipeline以便推理时复用
    os.makedirs("artifacts", exist_ok=True)
    joblib.dump(preprocessor, "artifacts/preprocessor.joblib")

    feature_names = preprocessor.get_feature_names_out()
    print(f"[OK] 特征工程完成: {len(feature_names)} 个特征")
    print(f"  训练集: {X_train_processed.shape}, 测试集: {X_test_processed.shape}")

    return {
        "X_train": X_train_processed,
        "X_test": X_test_processed,
        "y_train": y_train,
        "y_test": y_test,
        "feature_names": feature_names,
        "preprocessor": preprocessor,
    }

4.2 为什么用sklearn Pipeline而不是手写函数

对比 sklearn Pipeline 和手写转换函数在ML生产环境中的优劣

五、Step 3：模型训练与MLflow实验追踪

5.1 配置MLflow

维度	sklearn Pipeline	手写函数
数据泄露	`fit`只在训练集上执行，`transform`应用到测试集	容易在全局`fit`导致泄露
序列化	`joblib.dump` 一行搞定	需要手动管理每个转换器
可复现	Pipeline对象包含完整转换链	函数调用顺序依赖人工记忆
MLflow集成	Pipeline可直接log为artifact	需要手动打包

# 启动MLflow Tracking Server
mlflow server --host 0.0.0.0 --port 5000 --backend-store-uri sqlite:///mlflow.db

请创建 src/train.py，要求：
1. 使用MLflow追踪所有实验
2. 支持多种模型（RandomForest, GradientBoosting, LogisticRegression）
3. 自动记录参数、指标、模型
4. 包含模型评估阈值判断（accuracy > 0.85 才注册）
5. 注册通过阈值的模型到MLflow Model Registry

# src/train.py
import mlflow
import mlflow.sklearn
import numpy as np
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, f1_score, precision_score, recall_score
import warnings

warnings.filterwarnings("ignore")

SEED = 42
EXPERIMENT_NAME = "ml-pipeline-demo"
ACCURACY_THRESHOLD = 0.85


MODEL_REGISTRY = {
    "random_forest": {
        "class": RandomForestClassifier,
        "params": {"n_estimators": 200, "max_depth": 10, "random_state": SEED},
    },
    "gradient_boosting": {
        "class": GradientBoostingClassifier,
        "params": {"n_estimators": 150, "learning_rate": 0.1, "max_depth": 5, "random_state": SEED},
    },
    "logistic_regression": {
        "class": LogisticRegression,
        "params": {"max_iter": 1000, "random_state": SEED},
    },
}


def train_and_track(
    X_train: np.ndarray,
    X_test: np.ndarray,
    y_train: np.ndarray,
    y_test: np.ndarray,
    feature_names: list,
) -> dict:
    """训练多个模型，用MLflow追踪所有实验，返回最佳结果。"""
    mlflow.set_experiment(EXPERIMENT_NAME)
    results = {}

    for model_name, config in MODEL_REGISTRY.items():
        with mlflow.start_run(run_name=model_name) as run:
            # 记录参数
            mlflow.log_params(config["params"])
            mlflow.log_param("model_type", model_name)
            mlflow.log_param("seed", SEED)
            mlflow.log_param("train_samples", len(X_train))
            mlflow.log_param("n_features", len(feature_names))

            # 训练
            model = config["class"](**config["params"])
            model.fit(X_train, y_train)

            # 评估
            y_pred = model.predict(X_test)
            metrics = {
                "accuracy": accuracy_score(y_test, y_pred),
                "f1": f1_score(y_test, y_pred, average="weighted"),
                "precision": precision_score(y_test, y_pred, average="weighted"),
                "recall": recall_score(y_test, y_pred, average="weighted"),
            }
            mlflow.log_metrics(metrics)

            # 记录模型
            mlflow.sklearn.log_model(
                model,
                artifact_path="model",
                input_example=X_train[:3],
            )

            # 阈值门控：只有达标的模型才能注册
            if metrics["accuracy"] >= ACCURACY_THRESHOLD:
                model_uri = f"runs:/{run.info.run_id}/model"
                mlflow.register_model(
                    model_uri=model_uri,
                    name=f"{EXPERIMENT_NAME}-{model_name}",
                )
                status = "REGISTERED"
            else:
                status = f"SKIPPED (acc={metrics['accuracy']:.4f} < {ACCURACY_THRESHOLD})"

            results[model_name] = {
                "run_id": run.info.run_id,
                "metrics": metrics,
                "status": status,
            }
            print(f"  [{model_name}] acc={metrics['accuracy']:.4f}, f1={metrics['f1']:.4f} → {status}")

    return results

5.2 MLflow追踪的黄金法则

## 追踪规则（写入CLAUDE.md）
- **Always log**: 参数、指标、模型artifact、输入样例
- **Never**: 在没有实验追踪的情况下启动训练
- **Threshold**: 必须定义显式的通过阈值
- **Reproducibility**: 固定所有随机种子，pin所有依赖版本

我想在不修改训练代码的情况下追踪Claude Code对话本身。
有没有办法用MLflow追踪Claude Code的agent调用？

这是2026年MLflow 3.x的新能力——MLflow Tracing可以自动追踪Claude Code对话和Agent调用链：

# 通过MLflow MCP Server集成Claude Code
# 在 .claude/settings.json 中添加MCP配置
{
  "mcpServers": {
    "mlflow": {
      "command": "uvx",
      "args": ["mlflow", "mcp"]
    }
  }
}

六、Step 4：Pipeline编排——串联所有环节

6.1 创建主Pipeline入口

请创建 src/pipeline.py，将数据校验、特征工程、训练串联起来。
要求：
1. 接受配置参数（数据路径、模型类型、阈值等）
2. 每个环节有明确的状态输出
3. 任何环节失败立即中断并报错
4. 最终输出Pipeline运行报告

# src/pipeline.py
import json
import time
from datetime import datetime
from data_validation import validate_training_data
from feature_pipeline import prepare_data
from train import train_and_track


def run_pipeline(config: dict) -> dict:
    """执行完整的ML Pipeline。"""
    start = time.time()
    report = {
        "pipeline_id": datetime.now().strftime("%Y%m%d_%H%M%S"),
        "config": config,
        "stages": {},
    }

    # Stage 1: 数据校验
    print("=" * 50)
    print("Stage 1: 数据校验")
    print("=" * 50)
    try:
        df = validate_training_data(config["data_path"])
        report["stages"]["validation"] = {"status": "PASS", "rows": len(df)}
    except Exception as e:
        report["stages"]["validation"] = {"status": "FAIL", "error": str(e)}
        _save_report(report, start)
        raise

    # Stage 2: 特征工程
    print("\nStage 2: 特征工程")
    print("=" * 50)
    try:
        data = prepare_data(df, target_col=config["target_col"])
        report["stages"]["feature_engineering"] = {
            "status": "PASS",
            "n_features": len(data["feature_names"]),
        }
    except Exception as e:
        report["stages"]["feature_engineering"] = {"status": "FAIL", "error": str(e)}
        _save_report(report, start)
        raise

    # Stage 3: 模型训练与追踪
    print("\nStage 3: 模型训练")
    print("=" * 50)
    try:
        results = train_and_track(
            data["X_train"], data["X_test"],
            data["y_train"], data["y_test"],
            data["feature_names"],
        )
        report["stages"]["training"] = {
            "status": "PASS",
            "results": {k: {"accuracy": v["metrics"]["accuracy"], "status": v["status"]}
                        for k, v in results.items()},
        }
    except Exception as e:
        report["stages"]["training"] = {"status": "FAIL", "error": str(e)}
        _save_report(report, start)
        raise

    _save_report(report, start)
    return report


def _save_report(report: dict, start_time: float):
    report["duration_seconds"] = round(time.time() - start_time, 2)
    with open("pipeline_report.json", "w") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)
    print(f"\n{'=' * 50}")
    print(f"Pipeline Report: {report['pipeline_id']}")
    print(f"Duration: {report['duration_seconds']}s")
    for stage, info in report["stages"].items():
        print(f"  {stage}: {info['status']}")

6.2 运行Pipeline

# 准备示例数据（Claude Code可以帮你生成）
claude "生成一份模拟的 iris_classification.csv，包含4个数值特征和1个label列，200行数据，保存到 data/"

# 执行Pipeline
python -c "
from src.pipeline import run_pipeline
config = {
    'data_path': 'data/iris_classification.csv',
    'target_col': 'label',
    'accuracy_threshold': 0.85,
}
report = run_pipeline(config)
"

==================================================
Stage 1: 数据校验
==================================================
[OK] 数据校验通过: 200 行, 5 列

Stage 2: 特征工程
==================================================
[OK] 特征工程完成: 4 个特征
  训练集: (160, 4), 测试集: (40, 4)

Stage 3: 模型训练
==================================================
  [random_forest] acc=0.9250, f1=0.9234 → REGISTERED
  [gradient_boosting] acc=0.9000, f1=0.8987 → REGISTERED
  [logistic_regression] acc=0.8750, f1=0.8732 → REGISTERED

==================================================
Pipeline Report: 20260521_143022
Duration: 3.47s
  validation: PASS
  feature_engineering: PASS
  training: PASS

七、Step 5：模型部署——从注册到推理服务

7.1 从MLflow Registry加载模型

# src/serve.py
import mlflow
import joblib
import numpy as np
from fastapi import FastAPI
from pydantic import BaseModel


MODEL_NAME = "ml-pipeline-demo-random_forest"
MODEL_STAGE = "Production"

app = FastAPI(title="ML Prediction Service")


class PredictionRequest(BaseModel):
    features: list[list[float]]


class PredictionResponse(BaseModel):
    predictions: list[int]
    model_version: str
    confidence: list[float]


# 启动时加载模型和预处理器
model = mlflow.sklearn.load_model(f"models:/{MODEL_NAME}/{MODEL_STAGE}")
preprocessor = joblib.load("artifacts/preprocessor.joblib")


@app.post("/predict", response_model=PredictionResponse)
def predict(request: PredictionRequest):
    X = np.array(request.features)
    predictions = model.predict(X)
    probabilities = model.predict_proba(X).max(axis=1)

    return PredictionResponse(
        predictions=predictions.tolist(),
        model_version=mlflow.get_latest_versions(MODEL_NAME, stages=[MODEL_STAGE])[0].version,
        confidence=probabilities.tolist(),
    )


@app.get("/health")
def health():
    return {"status": "healthy", "model": MODEL_NAME}

7.2 容器化部署

请为这个ML推理服务生成生产级Dockerfile：
1. 多阶段构建，最终镜像<500MB
2. 非root用户运行
3. 健康检查
4. MLflow连接通过环境变量配置

# Dockerfile
FROM python:3.10-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

FROM python:3.10-slim
WORKDIR /app

# 安全：非root用户
RUN useradd -m -u 1000 mluser
COPY --from=builder /usr/local/lib/python3.10/site-packages /usr/local/lib/python3.10/site-packages
COPY --from=builder /usr/local/bin /usr/local/bin
COPY src/ src/
COPY artifacts/ artifacts/

USER mluser
EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "src.serve:app", "--host", "0.0.0.0", "--port", "8000"]

7.3 模型版本切换——零停机

# 将新版本模型推向Production
mlflow models transition-stage \
    --model-uri "models:/ml-pipeline-demo-random_forest/2" \
    --stage "Production"

# 旧版本自动归档到Archived
# 推理服务下次加载时自动使用新版本

八、实战案例：客户流失预测Pipeline

以下是一个完整的、可运行的端到端项目，用Claude Code从零构建。

8.1 项目初始化

mkdir customer-churn-pipeline && cd customer-churn-pipeline
git init
claude "/init"

我要构建一个客户流失预测Pipeline。请帮我完成以下工作：

1. 生成模拟数据 data/churn_data.csv（1000行），包含字段：
   - customer_id, tenure, monthly_charges, total_charges,
   - contract_type (Month-to-month/One year/Two year),
   - internet_service (DSL/Fiber optic/No),
   - churn (0/1)

2. 创建完整的Pipeline代码：
   - src/data_validation.py — 数据校验
   - src/feature_pipeline.py — 特征工程
   - src/train.py — 训练+MLflow追踪
   - src/pipeline.py — Pipeline编排
   - src/serve.py — FastAPI推理服务

3. 创建 tests/ 目录，包含所有模块的单元测试

4. 创建 Dockerfile 和 docker-compose.yml

5. 确保所有代码符合 CLAUDE.md 中的ML规则

8.2 Claude Code的工作过程

sequenceDiagram
    participant U as 开发者
    participant C as Claude Code
    participant F as 文件系统
    participant M as MLflow Server

    U->>C: 提交需求（Prompt）
    C->>F: 创建目录结构
    C->>F: 生成 data/churn_data.csv
    C->>F: 创建 src/data_validation.py
    C->>F: 创建 src/feature_pipeline.py
    C->>F: 创建 src/train.py
    C->>F: 创建 src/pipeline.py
    C->>F: 创建 src/serve.py
    C->>F: 创建 tests/
    C->>F: 创建 Dockerfile
    Note over C,F: 全部代码遵循CLAUDE.md规则

    U->>C: 运行Pipeline
    C->>F: python src/pipeline.py
    F->>M: log experiments
    M-->>C: 返回run_id
    C->>U: 展示Pipeline报告

8.3 数据校验脚本（Claude Code生成）

# tests/test_data_validation.py
import pytest
import pandas as pd
from src.data_validation import validate_training_data


def test_valid_data(tmp_path):
    """正常数据应通过校验。"""
    df = pd.DataFrame({
        "tenure": [1, 2, 3] * 40,
        "monthly_charges": [0.5, 0.3, 0.8] * 40,
        "churn": [0, 1, 0] * 40,
    })
    path = tmp_path / "test.csv"
    df.to_csv(path, index=False)
    result = validate_training_data(str(path))
    assert len(result) == 120


def test_missing_label_raises(tmp_path):
    """label列含空值时应抛出异常。"""
    df = pd.DataFrame({
        "tenure": [1, 2, None],
        "monthly_charges": [0.5, None, 0.8],
        "churn": [0, None, 0],
    })
    path = tmp_path / "bad.csv"
    df.to_csv(path, index=False)
    with pytest.raises(ValueError, match="数据校验失败"):
        validate_training_data(str(path))

8.4 运行测试

# Claude Code自动运行测试
claude "运行所有测试并修复失败的用例"
# 等价于手动执行：
pytest tests/ -v --tb=short

九、Claude Code在ML开发中的五大高级技巧

技巧1：用CLAUDE.md预置ML规范

在项目根目录的 CLAUDE.md 中写入ML Pipeline规范，Claude Code会在每次生成代码时自动遵循：

## ML Pipeline Conventions
- 数据路径：所有数据文件放在 data/ 目录
- 模型存储：所有artifact放在 artifacts/ 目录
- 实验名称：{project_name}-{model_type} 格式
- 随机种子：统一使用 SEED = 42
- 评估指标：必须包含 accuracy, f1, precision, recall
- 模型注册阈值：accuracy >= 0.85

技巧2：利用Subagent并行探索数据

当面对陌生的数据集时，同时启动多个Subagent分析不同维度：

请用3个subagent并行分析 data/churn_data.csv：
1. Agent A：分析数值特征的分布和异常值
2. Agent B：分析类别特征的基数和分布
3. Agent C：分析目标变量churn的类别平衡情况

最后汇总三个Agent的分析结果，给出特征工程建议。

技巧3：MLflow MCP Server集成

将MLflow作为MCP Server接入Claude Code，实现不离开终端即可管理实验：

// .claude/settings.json
{
  "mcpServers": {
    "mlflow": {
      "command": "uvx",
      "args": ["mlflow", "mcp"],
      "env": {
        "MLFLOW_TRACKING_URI": "http://localhost:5000"
      }
    }
  }
}

查询最近的5个实验运行，按accuracy降序排列
将最佳模型推向Production阶段
对比 random_forest 和 gradient_boosting 两个实验的F1差异

技巧4：DVC数据版本控制

# 初始化DVC
dvc init

# 跟踪训练数据
dvc add data/churn_data.csv
git add data/churn_data.csv.dvc data/.gitignore
git commit -m "track training data with DVC"

# Claude Code可以帮助管理DVC pipeline
claude "创建 dvc.yaml 定义完整的训练pipeline stages"

# dvc.yaml
stages:
  validate:
    cmd: python src/data_validation.py
    deps:
      - data/churn_data.csv
    outs:
      - artifacts/validated_data.csv

  train:
    cmd: python src/train.py
    deps:
      - artifacts/validated_data.csv
      - src/train.py
    outs:
      - artifacts/model.joblib
    metrics:
      - metrics.json:
          cache: false

技巧5：利用Hooks自动触发训练

// .claude/settings.json
{
  "hooks": {
    "PostToolUse": [
      {
        "matcher": "Write|Edit",
        "command": "if echo '$TOOL_INPUT' | grep -q 'data/'; then echo '⚠️ 数据文件已变更，建议重新运行Pipeline'; fi"
      }
    ]
  }
}

十、ML Pipeline可观测性——不只是训练

10.1 监控架构

graph LR
    A[推理请求] --> B[FastAPI服务]
    B --> C[模型预测]
    C --> D[日志记录]
    D --> E[Prometheus采集]
    E --> F[Grafana面板]

    G[MLflow] --> H[实验对比]
    G --> I[模型版本追踪]

    F --> J[告警规则]
    J -->|精度下降| K[自动重训触发]
    K --> G

    style A fill:#e3f2fd
    style F fill:#fff8e1
    style K fill:#fce4ec

10.2 关键监控指标

请为这个ML推理服务实现Prometheus指标采集，追踪以下指标：
1. 请求延迟分布（直方图）
2. 预测类别分布
3. 模型置信度分布
4. 请求QPS

# src/metrics.py
from prometheus_client import Histogram, Counter, Gauge, generate_latest
from fastapi import Response


PREDICTION_LATENCY = Histogram(
    "ml_prediction_latency_seconds",
    "Prediction latency in seconds",
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0],
)

PREDICTION_CLASSES = Counter(
    "ml_prediction_classes_total",
    "Predicted class distribution",
    ["class_label"],
)

CONFIDENCE_SCORE = Histogram(
    "ml_confidence_score",
    "Model confidence distribution",
    buckets=[0.5, 0.6, 0.7, 0.8, 0.9, 0.95, 1.0],
)

REQUEST_COUNT = Counter(
    "ml_requests_total",
    "Total prediction requests",
)


async def metrics_endpoint():
    return Response(
        content=generate_latest(),
        media_type="text/plain",
    )

十一、完整工作流总结

graph TD
    subgraph "开发阶段"
        A1[CLAUDE.md 规范定义] --> A2[Claude Code 生成代码]
        A2 --> A3[数据校验模块]
        A2 --> A4[特征工程Pipeline]
        A2 --> A5[训练脚本]
    end

    subgraph "训练阶段"
        B1[DVC 版本化数据] --> B2[运行Pipeline]
        B2 --> B3[MLflow 追踪实验]
        B3 --> B4[阈值门控判断]
        B4 -->|Pass| B5[模型注册]
        B4 -->|Fail| B6[调参重训]
    end

    subgraph "部署阶段"
        C1[MLflow Registry] --> C2[容器化构建]
        C2 --> C3[FastAPI 推理服务]
        C3 --> C4[Prometheus 监控]
    end

    A5 --> B1
    B5 --> C1

    style A1 fill:#e8eaf6
    style B3 fill:#fff3e0
    style C3 fill:#e8f5e9

Claude Code在ML Pipeline中的定位

十二、常见踩坑与排错指南

问题1：MLflow无法连接Tracking Server

阶段	Claude Code的作用	时间节省
数据探索	并行Subagent分析数据分布	~60%
代码生成	生成sklearn Pipeline、训练脚本、测试	~70%
实验管理	通过MLflow MCP管理实验	~40%
调试	快速定位数据泄露、维度错误	~50%
文档	自动生成模型卡片、API文档	~80%
部署	生成Dockerfile、docker-compose	~65%

# 检查Server状态
curl http://localhost:5000/health

# 如果是远程Server，确认环境变量
export MLFLOW_TRACKING_URI=http://your-server:5000

# Claude Code可以帮助排查
claude "我的MLflow连接超时，请检查 src/train.py 中的连接配置"

问题2：sklearn Pipeline序列化失败

常见原因：Pipeline中包含lambda函数或不可pickle的对象。Claude Code可以帮助诊断：

运行 joblib.dump(preprocessor, "test.joblib") 时报错
_pickle.PicklingError: Can't pickle <function ...>
请帮我修复这个序列化问题

解决方案：将lambda替换为sklearn内置的FunctionTransformer或自定义类。

问题3：特征维度训练/推理不一致

这是ML工程中最常见的问题。Claude Code的 CLAUDE.md 规则可以有效预防：

维度	评估
适合训练营类型	ML工程化训练营、MLOps入门营、AI开发者实战营
教学时长	4-6小时（含动手实验）
难度等级	中级（需要Python和基础ML知识）
前置课程	Claude Code入门 + Python ML基础
实验环境	每人一台云主机（4C8G），预装Claude Code和MLflow
考核方式	提交完整的Pipeline代码 + MLflow实验截图 + 推理服务演示

Claude Code ML Pipeline全链路实战：从数据准备到模型交付

一、前置条件

初始化项目结构

安装依赖

二、ML Pipeline全流程架构