# Claude Code ML Pipeline全链路实战：从数据准备到模型交付
> **学习目标**：掌握使用Claude Code构建端到端机器学习Pipeline的完整工作流——涵盖数据校验、特征工程、模型训练、实验追踪（MLflow）、模型注册与部署。本文提供一套可复现的实战框架，将传统"手工脚本"式ML开发升级为工程化、可观测、可回滚的生产级Pipeline。
>
> **适用人群**：有Python基础的数据科学家、ML工程师，以及希望将AI编程工具引入ML工程化流程的技术管理者。
---
## 一、前置条件
| 项目 | 要求 |
|------|------|
| 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`） |
### 初始化项目结构
```bash
# 创建项目目录
mkdir ml-pipeline-demo && cd ml-pipeline-demo
git init
# Claude Code初始化CLAUDE.md
claude "/init"
```
在 `CLAUDE.md` 中追加ML专属指令：
```markdown
## ML Pipeline Rules
- 所有训练脚本必须包含MLflow实验追踪
- 特征工程代码必须包含数据校验（Great Expectations）
- 模型评估必须有明确的pass/fail阈值
- 禁止在代码中硬编码凭证或路径
- 所有随机种子必须固定（SEED=42）
- 训练与推理代码严格分离
```
### 安装依赖
```bash
uv venv && source .venv/bin/activate
uv pip install mlflow scikit-learn pandas numpy great_expectations dvc
```
---
## 二、ML Pipeline全流程架构
在动手编码之前，先用Claude Code规划整个Pipeline架构。Claude Code擅长生成Mermaid流程图和架构设计：
```mermaid
graph TD
A[原始数据 CSV/Parquet] --> B[数据校验
Great Expectations]
B -->|通过| C[特征工程
Pandas/Sklearn Pipeline]
B -->|失败| B1[告警 & 日志]
C --> D[数据版本控制
DVC]
D --> E[模型训练
Sklearn/XGBoost]
E --> F[实验追踪
MLflow]
F --> G[模型校验
阈值判断]
G -->|Pass| H[模型注册
MLflow Registry]
G -->|Fail| E
H --> I[模型部署
FastAPI/Docker]
I --> J[监控 & 告警
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 创建数据校验模块
向Claude Code发出指令：
```
请创建 src/data_validation.py，实现以下功能：
1. 使用 Great Expectations 对CSV数据进行校验
2. 检查：label列无空值、数值特征范围在0-1之间、行数>100
3. 校验失败时抛出明确异常
4. 返回校验通过的DataFrame
```
Claude Code生成代码：
```python
# 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 常见问题与解决方案
| 问题 | 原因 | 解决 |
|------|------|------|
| `gx.get_context()` 报错 | Great Expectations版本不兼容 | `uv pip install "great_expectations>=0.18"` |
| 校验超时 | 数据量过大 | 对大数据集先 `df.sample()` 做采样校验 |
| 列名不匹配 | CSV列名含空格 | `df.columns = df.columns.str.strip()` 预处理 |
---
## 四、Step 2：特征工程——构建可复用的转换Pipeline
### 4.1 用Claude Code生成特征Pipeline
Claude Code的核心优势在于理解业务语义并生成符合sklearn Pipeline规范的代码：
```
请创建 src/feature_pipeline.py，要求：
1. 使用 sklearn Pipeline + ColumnTransformer
2. 数值特征：StandardScaler
3. 类别特征：OneHotEncoder(handle_unknown='ignore')
4. 输出是一个可序列化的Pipeline对象
5. 包含 get_feature_names() 方法以便后续追踪
```
生成代码：
```python
# 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而不是手写函数
向Claude Code询问工程化决策：
```
对比 sklearn Pipeline 和手写转换函数在ML生产环境中的优劣
```
Claude Code会指出关键差异：
| 维度 | sklearn Pipeline | 手写函数 |
|------|------------------|----------|
| **数据泄露** | `fit`只在训练集上执行，`transform`应用到测试集 | 容易在全局`fit`导致泄露 |
| **序列化** | `joblib.dump` 一行搞定 | 需要手动管理每个转换器 |
| **可复现** | Pipeline对象包含完整转换链 | 函数调用顺序依赖人工记忆 |
| **MLflow集成** | Pipeline可直接log为artifact | 需要手动打包 |
---
## 五、Step 3：模型训练与MLflow实验追踪
### 5.1 配置MLflow
```bash
# 启动MLflow Tracking Server
mlflow server --host 0.0.0.0 --port 5000 --backend-store-uri sqlite:///mlflow.db
```
在Claude Code中创建训练脚本：
```
请创建 src/train.py，要求：
1. 使用MLflow追踪所有实验
2. 支持多种模型（RandomForest, GradientBoosting, LogisticRegression）
3. 自动记录参数、指标、模型
4. 包含模型评估阈值判断（accuracy > 0.85 才注册）
5. 注册通过阈值的模型到MLflow Model Registry
```
生成代码：
```python
# 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 Code协作时的关键原则：
```markdown
## 追踪规则（写入CLAUDE.md）
- **Always log**: 参数、指标、模型artifact、输入样例
- **Never**: 在没有实验追踪的情况下启动训练
- **Threshold**: 必须定义显式的通过阈值
- **Reproducibility**: 固定所有随机种子，pin所有依赖版本
```
向Claude Code提出MLflow集成问题：
```
我想在不修改训练代码的情况下追踪Claude Code对话本身。
有没有办法用MLflow追踪Claude Code的agent调用？
```
这是2026年MLflow 3.x的新能力——MLflow Tracing可以自动追踪Claude Code对话和Agent调用链：
```python
# 通过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运行报告
```
```python
# 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
```bash
# 准备示例数据（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加载模型
```python
# 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 容器化部署
Claude Code可以直接生成 `Dockerfile` 和 `docker-compose.yml`：
```
请为这个ML推理服务生成生产级Dockerfile：
1. 多阶段构建，最终镜像<500MB
2. 非root用户运行
3. 健康检查
4. MLflow连接通过环境变量配置
```
```dockerfile
# 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 模型版本切换——零停机
```bash
# 将新版本模型推向Production
mlflow models transition-stage \
--model-uri "models:/ml-pipeline-demo-random_forest/2" \
--stage "Production"
# 旧版本自动归档到Archived
# 推理服务下次加载时自动使用新版本
```
---
## 八、实战案例：客户流失预测Pipeline
以下是一个完整的、可运行的端到端项目，用Claude Code从零构建。
### 8.1 项目初始化
```bash
mkdir customer-churn-pipeline && cd customer-churn-pipeline
git init
claude "/init"
```
向Claude Code提供完整需求：
```
我要构建一个客户流失预测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的工作过程
Claude Code会按以下顺序自主完成任务：
```mermaid
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生成）
```python
# 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 运行测试
```bash
# Claude Code自动运行测试
claude "运行所有测试并修复失败的用例"
# 等价于手动执行：
pytest tests/ -v --tb=short
```
---
## 九、Claude Code在ML开发中的五大高级技巧
### 技巧1：用CLAUDE.md预置ML规范
在项目根目录的 `CLAUDE.md` 中写入ML Pipeline规范，Claude Code会在每次生成代码时自动遵循：
```markdown
## 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，实现不离开终端即可管理实验：
```json
// .claude/settings.json
{
"mcpServers": {
"mlflow": {
"command": "uvx",
"args": ["mlflow", "mcp"],
"env": {
"MLFLOW_TRACKING_URI": "http://localhost:5000"
}
}
}
}
```
集成后可以直接在Claude Code中：
```
查询最近的5个实验运行，按accuracy降序排列
将最佳模型推向Production阶段
对比 random_forest 和 gradient_boosting 两个实验的F1差异
```
### 技巧4：DVC数据版本控制
用DVC跟踪数据变更，确保训练数据可追溯：
```bash
# 初始化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"
```
```yaml
# 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 Code Hooks，在数据变更时自动触发Pipeline：
```json
// .claude/settings.json
{
"hooks": {
"PostToolUse": [
{
"matcher": "Write|Edit",
"command": "if echo '$TOOL_INPUT' | grep -q 'data/'; then echo '⚠️ 数据文件已变更，建议重新运行Pipeline'; fi"
}
]
}
}
```
---
## 十、ML Pipeline可观测性——不只是训练
### 10.1 监控架构
```mermaid
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 关键监控指标
向Claude Code询问：
```
请为这个ML推理服务实现Prometheus指标采集，追踪以下指标：
1. 请求延迟分布（直方图）
2. 预测类别分布
3. 模型置信度分布
4. 请求QPS
```
Claude Code生成：
```python
# 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",
)
```
---
## 十一、完整工作流总结
```mermaid
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中的定位
| 阶段 | Claude Code的作用 | 时间节省 |
|------|-------------------|----------|
| 数据探索 | 并行Subagent分析数据分布 | ~60% |
| 代码生成 | 生成sklearn Pipeline、训练脚本、测试 | ~70% |
| 实验管理 | 通过MLflow MCP管理实验 | ~40% |
| 调试 | 快速定位数据泄露、维度错误 | ~50% |
| 文档 | 自动生成模型卡片、API文档 | ~80% |
| 部署 | 生成Dockerfile、docker-compose | ~65% |
---
## 十二、常见踩坑与排错指南
### 问题1：MLflow无法连接Tracking Server
```bash
# 检查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
请帮我修复这个序列化问题
```
解决方案：将lambda替换为sklearn内置的FunctionTransformer或自定义类。
### 问题3：特征维度训练/推理不一致
这是ML工程中最常见的问题。Claude Code的 `CLAUDE.md` 规则可以有效预防：
```markdown
## 防御规则
- 特征工程Pipeline必须序列化并随模型一起打包
- 推理服务启动时加载同一个preprocessor.joblib
- 新增特征必须同步更新preprocessor和模型版本号
```
---
## 十三、延伸阅读
- [MLflow官方文档 — Experiment Tracking](https://mlflow.org/docs/latest/tracking.html) — MLflow实验追踪完整参考
- [Great Expectations数据校验指南](https://docs.greatexpectations.io/docs/guides/validation/) — 数据质量门控最佳实践
- [Anthropic — Claude Code Best Practices](https://docs.anthropic.com/en/docs/claude-code/best-practices) — Claude Code官方最佳实践
- [MLflow + Claude Code集成博客](https://mlflow.org/blog/mlflow-claude-code) — MLflow官方关于Claude Code集成的教程
- [sklearn Pipeline用户指南](https://scikit-learn.org/stable/modules/compose.html) — Pipeline和ColumnTransformer深入指南
---
## 十四、训练营适用性评估
| 维度 | 评估 |
|------|------|
| **适合训练营类型** | ML工程化训练营、MLOps入门营、AI开发者实战营 |
| **教学时长** | 4-6小时（含动手实验） |
| **难度等级** | 中级（需要Python和基础ML知识） |
| **前置课程** | Claude Code入门 + Python ML基础 |
| **实验环境** | 每人一台云主机（4C8G），预装Claude Code和MLflow |
| **考核方式** | 提交完整的Pipeline代码 + MLflow实验截图 + 推理服务演示 |
---
## 参考来源
1. [ML Pipeline Skill for Claude Code](https://jeffallan.github.io/claude-skills/skills/data-ml/ml-pipeline/) — 开源ML Pipeline Skill的完整SKILL.md定义，包含代码模板和约束规则
2. [MLflow + Claude Code集成指南](https://mlflow.org/blog/mlflow-claude-code) — MLflow官方博客，详解如何用MLflow追踪Claude Code对话和Agent调用
3. [Getting Started with Claude Code for Data Scientists](https://www.dataquest.io/blog/getting-started-with-claude-code-for-data-scientists/) — DataQuest数据科学工作流教程
4. [Claude Code Best Practices](https://github.com/awattar/claude-code-best-practices) — 社区维护的Claude Code最佳实践仓库，含自定义命令和Agent模板
5. [Azure Databricks — Tracing Claude Code with MLflow](https://learn.microsoft.com/en-us/azure/databricks/mlflow3/genai/tracing/integrations/claude-code) — 微软官方文档，MLflow Tracing自动追踪Claude Code Agent
> [!tip] 51学通信知识星球
> **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。
> **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。
> **当前规模**：5200 篇主题 / 990 个回答，198 元/年。
> **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。
> **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。
