机器学习流水线:从 notebook 到可复现的产物

机器学习流水线:从 notebook 到可复现的产物

本节摘要:模型不是产品,流水线(Pipeline)才是。流水线是从原始数据到部署预测的一切步骤,每一步都必须可复现。你有一个 notebook:加载数据、用中位数填缺失、缩放特征、训练模型、打印准确率。一个月后有人重训,得到不同结果——因为中位数是在含测试数据的全集上算的(数据泄漏);因为缩放参数没存,推理时用了不同统计量;因为特征工程代码在训练和服务两处被复制粘贴,两份拷贝已经发散;因为某个类别列在生产里出现了编码器从未见过的新值。这些不是假想,是 ML 系统在生产里翻车最常见的原因。流水线把每个变换步骤打包成一个有序、可复现的单一对象,一举解决这些问题:变换只在训练数据上拟合(无泄漏)、推理时套用同样的变换、整个对象可序列化部署、交叉验证按折应用流水线。本节将从零实现一个自定义变换器和流水线类,讲透 ColumnTransformer 如何对数值和类别列分别预处理,并引入 MLflow/DVC 做实验追踪与数据版本管理。

学习目标

阅读完本节,你应当能够:

1. 从零搭建一条机器学习流水线,把填补、缩放、编码、模型训练链接成一个可复现对象。
2. 识别数据泄漏场景,解释流水线如何通过只在训练数据上拟合变换器来防止泄漏。
3. 构造一个 ColumnTransformer,对数值和类别特征应用不同预处理。
4. 实现流水线序列化,演示同一已拟合流水线在训练与生产里给出一致结果。

一、问题与直觉

你有一个 notebook,加载数据、用中位数填缺失、缩放特征、训练模型、打印准确率。它能跑。你把它上线了。

一个月后,有人重训模型,得到不同结果。中位数是在含测试数据的全集上算的(数据泄漏)。缩放参数没存,推理时用了不同统计量。特征工程代码在训练和服务之间被复制粘贴,两份拷贝已经发散。某个类别列在生产里出现了编码器从未见过的新值。

这些不是假想,是 ML 系统在生产里翻车最常见的原因。流水线通过把每个变换步骤打包成一个有序、可复现的单一对象,把它们全部解决。

流水线是什么

流水线是一连串有序的数据变换,后跟一个模型。每一步把上一步的输出当输入。整条流水线在训练数据上一次性拟合好。推理时,同一个已拟合的流水线变换新数据并产出预测。

流水线保证:

• 变换只在训练数据上拟合(无泄漏)。
• 推理时套用同样的变换。
• 整个对象可序列化为单一产物部署。
• 交叉验证按折应用流水线,防止微妙泄漏。

数据泄漏:沉默的杀手

数据泄漏(Data Leakage)发生在测试集或未来数据的信息污染了训练。流水线防止最常见的几种形式。

有泄漏(错误):

X = df.drop("target", axis=1)
y = df["target"]

scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

X_train, X_test = X_scaled[:800], X_scaled[800:]
y_train, y_test = y[:800], y[800:]
​

scaler 看到了测试数据。均值和标准差里混入了测试样本。这会虚高准确率估计。

正确:

X_train, X_test = X[:800], X[800:]

scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)
​

有了流水线,你不必操心这个——流水线自动处理。

sklearn Pipeline

sklearn 的 Pipeline 把变换器和一个估计器链接起来。它暴露 .fit()、.predict()、.score(),按顺序应用所有步骤。

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression

pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("model", LogisticRegression()),
])

pipe.fit(X_train, y_train)
predictions = pipe.predict(X_test)
​

调用 pipe.fit(X_train, y_train) 时:

1. Scaler 对 X_train 调 fit_transform。
2. Model 在缩放后的 X_train 上调 fit。

调用 pipe.predict(X_test) 时:

1. Scaler 对 X_test 调 transform(不是 fit_transform)。
2. Model 在缩放后的 X_test 上调 predict。

scaler 在拟合阶段从不看测试数据。这就是全部要点。

ColumnTransformer:不同列走不同流水线

真实数据有数值列和类别列,需要不同预处理。ColumnTransformer 处理这个。

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer

numeric_pipe = Pipeline([
    ("impute", SimpleImputer(strategy="median")),
    ("scale", StandardScaler()),
])

categorical_pipe = Pipeline([
    ("impute", SimpleImputer(strategy="most_frequent")),
    ("encode", OneHotEncoder(handle_unknown="ignore")),
])

preprocessor = ColumnTransformer([
    ("num", numeric_pipe, ["age", "income", "score"]),
    ("cat", categorical_pipe, ["city", "gender", "plan"]),
])

full_pipeline = Pipeline([
    ("preprocess", preprocessor),
    ("model", GradientBoostingClassifier()),
])
​

OneHotEncoder 里的 handle_unknown="ignore" 对生产至关重要。当出现新类别(模型从未见过的城市),它产出一个零向量而非崩溃。

实验追踪

流水线让训练可复现,但你还需要追踪跨实验发生了什么:用了哪些超参数、哪个数据集版本、指标多少、跑的哪份代码。

MLflow 是最常见的开源方案:

import mlflow

with mlflow.start_run():
    mlflow.log_param("max_depth", 5)
    mlflow.log_param("n_estimators", 100)
    mlflow.log_param("learning_rate", 0.1)

    pipe.fit(X_train, y_train)
    accuracy = pipe.score(X_test, y_test)

    mlflow.log_metric("accuracy", accuracy)
    mlflow.sklearn.log_model(pipe, "model")
​

每次运行都记录了参数、指标、产物和完整模型。你可以对比运行、复现任何实验、部署任何模型版本。

Weights & Biases(wandb) 提供同样的功能,配一个托管仪表盘:

import wandb

wandb.init(project="my-pipeline")
wandb.config.update({"max_depth": 5, "n_estimators": 100})

pipe.fit(X_train, y_train)
accuracy = pipe.score(X_test, y_test)

wandb.log({"accuracy": accuracy})
​

模型版本管理

实验追踪之外,你还需要管理模型版本。哪个模型在生产?哪个是预发布?上周的是哪个?

MLflow 的模型注册表(Model Registry)提供:

• 版本追踪:每个保存的模型拿到一个版本号。
• 阶段切换:"Staging"、"Production"、"Archived"。
• 审批流程:模型必须被显式晋升到生产。
• 回滚:瞬间切回前一版本。

用 DVC 做数据版本管理

代码用 git 版本管理。数据也该如此,但 git 处理不了大文件。DVC(Data Version Control)解决这个问题。

dvc init
dvc add data/training.csv
git add data/training.csv.dvc data/.gitignore
git commit -m "Track training data"
dvc push
​

DVC 把实际数据存在远程存储(S3、GCS、Azure),在 git 里保留一个小的 .dvc 文件记录哈希。当你 checkout 一个 git commit,dvc checkout 还原当时用的精确数据。

这意味着每个 git commit 同时钉住了代码和数据。完全可复现。

可复现实验的四要素

一个可复现实验需要四样东西:

1. 固定随机种子:给 numpy、random、框架(torch、sklearn)设种子。
2. 钉死的依赖:requirements.txt 或 poetry.lock,精确版本。
3. 版本化的数据:DVC 或类似工具。
4. 配置文件:所有超参数在配置里,不硬编码。

import numpy as np
import random

def set_seed(seed=42):
    random.seed(seed)
    np.random.seed(seed)
    try:
        import torch
        torch.manual_seed(seed)
        torch.cuda.manual_seed_all(seed)
        torch.backends.cudnn.deterministic = True
    except ImportError:
        pass
​

从 notebook 到生产流水线

典型的演进路径:

1. notebook 探索:快速实验、可视化、特征想法。
2. 抽取函数:把预处理、特征工程、评估搬进模块。
3. 构建 Pipeline:把变换链接成 sklearn Pipeline 或自定义类。
4. 配置管理:所有超参数搬进 YAML/JSON 配置。
5. 实验追踪:加 MLflow 或 wandb 日志。
6. 数据校验:训练前检查 schema、分布、缺失模式。
7. 测试:变换器的单元测试、整条流水线的集成测试。
8. 部署:序列化流水线,包进 API(FastAPI、Flask),容器化。

流水线常见错误

二、从零实现

code/pipeline.py 从零构建一条完整的机器学习流水线。

第 1 步:自定义变换器

class CustomTransformer:
    def __init__(self):
        self.means = None
        self.stds = None

    def fit(self, X):
        self.means = np.mean(X, axis=0)
        self.stds = np.std(X, axis=0)
        self.stds[self.stds == 0] = 1.0
        return self

    def transform(self, X):
        return (X - self.means) / self.stds

    def fit_transform(self, X):
        return self.fit(X).transform(X)
​

第 2 步:从零实现 Pipeline

class PipelineFromScratch:
    def __init__(self, steps):
        self.steps = steps

    def fit(self, X, y=None):
        X_current = X.copy()
        for name, step in self.steps[:-1]:
            X_current = step.fit_transform(X_current)
        name, model = self.steps[-1]
        model.fit(X_current, y)
        return self

    def predict(self, X):
        X_current = X.copy()
        for name, step in self.steps[:-1]:
            X_current = step.transform(X_current)
        name, model = self.steps[-1]
        return model.predict(X_current)
​

注意 fit 时变换器调 fit_transform,而 predict 时只调 transform——这正是防泄漏的关键。

第 3 步:配交叉验证的流水线

代码演示带流水线的交叉验证如何防数据泄漏:scaler 在每一折的训练数据上单独拟合。

第 4 步:用 sklearn 搭完整生产流水线

用 ColumnTransformer、多条预处理路径和一个模型搭一条完整流水线,配恰当的交叉验证和实验日志。

三、框架对比

sklearn 完整流水线

from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.ensemble import GradientBoostingClassifier

numeric_features = ["age", "income", "score"]
categorical_features = ["city", "gender", "plan"]

preprocessor = ColumnTransformer(
    transformers=[
        ("num", Pipeline([
            ("impute", SimpleImputer(strategy="median")),
            ("scale", StandardScaler()),
        ]), numeric_features),
        ("cat", Pipeline([
            ("impute", SimpleImputer(strategy="most_frequent")),
            ("encode", OneHotEncoder(handle_unknown="ignore")),
        ]), categorical_features),
    ]
)

pipe = Pipeline([
    ("preprocess", preprocessor),
    ("model", GradientBoostingClassifier()),
])

pipe.fit(X_train, y_train)
​

流水线 + 交叉验证(防泄漏)

from sklearn.model_selection import cross_val_score

scores = cross_val_score(pipe, X, y, cv=5, scoring="accuracy")
print(f"CV 准确率: {scores.mean():.4f} +/- {scores.std():.4f}")
​

cross_val_score 在每一折内部对训练部分调 fit(从而 fit_transform)、对验证部分只调 transform,彻底杜绝泄漏。

流水线序列化与部署

import joblib

# 训练并保存
pipe.fit(X_train, y_train)
joblib.dump(pipe, "model_pipeline.joblib")

# 在另一个脚本里加载并推理
loaded_pipe = joblib.load("model_pipeline.joblib")
predictions = loaded_pipe.predict(X_new)
​

序列化后整个对象(填补器、缩放器、编码器、模型)作为一个产物存在。推理时无需重建任何变换逻辑,训练/服务偏差无从产生。

第 12 节嵌套调参与流水线结合

from sklearn.model_selection import GridSearchCV, cross_val_score

param_grid = {
    "model__max_depth": [3, 5, 7],
    "model__n_estimators": [50, 100, 200],
    "model__learning_rate": [0.01, 0.1],
}

search = GridSearchCV(pipe, param_grid, cv=5, scoring="accuracy")
search.fit(X_train, y_train)
print(f"最佳参数: {search.best_params_}")
​

注意参数名用 步骤名__参数名 的双下划线语法,这让你能调流水线任意步骤的超参数。

💡 凡是能在 Pipeline 里做的事就别在 Pipeline 外做。任何游离于流水线之外的变换都是训练/服务偏差的潜在来源。

四、可复用产物

本节产出:

• outputs/prompt-ml-pipeline.md——一个帮你搭建和调试机器学习流水线的技能提示词,涵盖防泄漏、列变换、序列化、实验追踪。
• code/pipeline.py——从零到 sklearn 的完整流水线实现。

五、练习

1. 多列类型流水线:搭一条处理 3 个数值列、2 个类别列的流水线。用 ColumnTransformer 对数值列做中位数填补 + 缩放,对类别列做众数填补 + one-hot 编码,用 5 折交叉验证训练。

2. 故意引入泄漏:在切分前对全集拟合 scaler,对比其交叉验证分数(有泄漏)与流水线交叉验证分数(干净)。差距多大?

3. 序列化与一致性:用 joblib.dump 保存流水线,在另一个脚本加载并预测,验证预测完全一致。

4. 加多项式特征变换器:给流水线加一个自定义变换器,为两个最重要的数值列生成二次多项式特征。它该放在流水线的哪个位置?

5. 配 MLflow:给流水线配 MLflow 追踪,跑 5 个不同超参数的实验,用 MLflow UI(mlflow ui)对比运行并挑出最佳模型。

本节要点回顾

1. 模型不是产品,流水线才是:从原始数据到部署预测的每一步都必须可复现,流水线把所有步骤打包成单一对象。
2. 数据泄漏是沉默杀手:在切分前对全集拟合变换器会让测试信息混入训练,虚高准确率。
3. fit_transform vs transform 是防泄漏关键:训练时 fit_transform(学统计量+变换),推理/验证时只 transform(用学到的统计量)。
4. ColumnTransformer 处理异质列:数值列走填补+缩放,类别列走填补+编码,合二为一。
5. handle_unknown="ignore" 对生产至关重要:未见过的类别产出零向量而非崩溃。
6. 交叉验证按折应用流水线:cross_val_score 在每折内部对训练部分 fit、对验证部分只 transform,杜绝微妙泄漏。
7. 序列化整条流水线:joblib 把填补器、缩放器、编码器、模型存为一个产物,消除训练/服务偏差。
8. 双下划线语法调流水线超参数:model__max_depth 让你能调任意步骤的参数。
9. 可复现实验四要素:固定随机种子、钉死依赖、版本化数据(DVC)、配置文件管理超参数。
10. 凡是能在 Pipeline 里做的就别在外做:游离变换是训练/服务偏差的潜在来源;notebook 到生产的路径是抽取函数→Pipeline→配置→追踪→校验→测试→部署。

下一节,我们讲朴素贝叶斯——为何一个数学上「错」的独立性假设,反而能在文本分类上击败更「聪明」的模型。