Python 基础体系 · 第 97/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

scikit-learn 工程基础：Pipeline、预处理、训练、评测和持久化

机器学习工程并不只是调用一个估计器：

model.fit(X_train, y_train)

真正可复用、可评测、可部署的流程通常还包括：

1. 明确特征和目标；
2. 划分训练集、验证集和测试集；
3. 只使用训练数据学习预处理参数；
4. 将不同类型的列分别转换；
5. 训练模型；
6. 使用交叉验证选择超参数；
7. 在未参与选择过程的测试集上评测；
8. 将预处理器和模型作为一个整体持久化；
9. 在加载模型后验证输入契约和预测结果。

scikit-learn 的核心价值之一，是把预处理器、特征组合器、模型选择工具和最终估计器统一到相近的接口中。预处理器通常通过 fit 从训练数据中学习状态，再通过 transform 转换新数据；最终模型通过 fit 学习参数，再通过 predict 或 predict_proba 产生预测。(scikit-learn.org)

本文使用 Python 3.14 语法和常规的 scikit-learn 工程接口。具体可安装版本应以目标平台上可获得的 wheel、NumPy、SciPy 和 Python 版本组合为准；安装后应通过导入和小规模训练进行验证，而不是仅根据版本字符串推断兼容性。scikit-learn 当前稳定文档已经包含 Python 3.14 相关测试和发布说明，但不同 Python 构建方式，尤其是 free-threaded 构建，仍应单独验证。(scikit-learn.org)

---

一、先建立 scikit-learn 的对象模型

1. 估计器、转换器和预测器

scikit-learn 中的估计器是能够从数据中学习的对象，最基本的方法是：

estimator.fit(X, y)

其中：

• X 是样本特征；
• y 是监督学习中的目标；
• fit 会修改对象内部状态；
• 训练完成后，模型通常会出现以下划线结尾的属性，例如 coef_、classes_、mean_。

转换器是能够改变特征表示的估计器，通常实现：

transformer.fit(X_train)
X_train_transformed = transformer.transform(X_train)
X_test_transformed = transformer.transform(X_test)

许多转换器还提供：

X_train_transformed = transformer.fit_transform(X_train)

fit_transform 的语义是先根据输入数据学习转换所需的状态，再转换同一批数据。它通常可以比显式调用 fit 后再调用 transform 更高效，但不能因此把测试数据传给 fit_transform。(scikit-learn.org)

预测器是在 fit 后能够进行预测的估计器，例如：

model.predict(X)
model.predict_proba(X)

分类器的 predict 通常返回离散类别，predict_proba 返回每个类别的概率。回归器通常使用：

model.predict(X)

返回连续值。

2. 训练状态和未训练状态

一个新建的模型还没有训练状态：

from sklearn.preprocessing import StandardScaler

scaler = StandardScaler()

此时不能直接依赖 scaler.mean_：

# AttributeError：尚未调用 fit
print(scaler.mean_)

调用：

scaler.fit(X_train)

后，StandardScaler 会根据 X_train 计算均值和尺度，随后才可以转换数据。

这个状态变化可以表示为：

新建对象
   │
   ├── fit(X_train)
   ▼
已拟合对象
   │
   ├── transform(X_new)
   ├── predict(X_new)
   └── score(X_new, y_new)

同一个对象不应在训练完成后被不受控制地重复 fit。再次调用 fit 通常会用新数据覆盖旧的拟合状态，而不是自动增量合并所有历史数据。是否支持增量训练，应查看具体估计器是否实现 partial_fit，不能把普通的 fit 当成增量接口。

---

二、为什么预处理必须属于训练流程

1. 预处理不是“训练前的无害准备”

假设有一个特征 $x$，标准化变换为：

$$z = \frac{x - \mu}{\sigma}$$

其中：

• $\mu$ 是训练数据的均值；
• $\sigma$ 是训练数据的标准差；
• $z$ 是转换后的特征。

如果用训练集计算：

$$\mu_{\text{train}}, \sigma_{\text{train}}$$

那么测试样本只能使用这两个已经学习到的量：

$$z_{\text{test}} = \frac{x_{\text{test}} - \mu_{\text{train}}} {\sigma_{\text{train}}}$$

不能使用测试集自己的均值和标准差：

$$\frac{x_{\text{test}}-\mu_{\text{test}}}{\sigma_{\text{test}}}$$

后者会让测试集参与预处理参数的计算。即使没有直接使用测试标签，也已经把测试分布的信息带入了训练过程。

2. 数据泄漏的形式化定义

设训练数据为：

$$D_{\text{train}} = \{(x_i, y_i)\}_{i=1}^{n}$$

测试数据为：

$$D_{\text{test}} = \{(x_j, y_j)\}_{j=1}^{m}$$

理想的训练过程是：

$$\theta = \operatorname{fit}(D_{\text{train}})$$

预测时：

$$\hat{y}_{\text{test}} = \operatorname{predict}(x_{\text{test}};\theta)$$

如果预处理参数 $\phi$ 和模型参数 $\theta$ 分开学习，正确流程是：

$$\phi = \operatorname{fit\_transformer}(X_{\text{train}})$$

$$X'_{\text{train}} = T(X_{\text{train}};\phi)$$

$$\theta = \operatorname{fit\_model}(X'_{\text{train}}, y_{\text{train}})$$

$$X'_{\text{test}} = T(X_{\text{test}};\phi)$$

$$\hat{y}_{\text{test}} = \operatorname{predict\_model}(X'_{\text{test}};\theta)$$

而错误流程是：

$$\phi = \operatorname{fit\_transformer}(X_{\text{train}} \cup X_{\text{test}})$$

这会导致评测结果偏乐观。模型看到了测试集的分布统计，只是没有直接看到测试标签。

3. 一个具体反例

下面的代码看起来合理，但存在泄漏：

from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

scaler = StandardScaler()
X_all_scaled = scaler.fit_transform(
    # 错误：这里不应把训练集和测试集一起 fit
    X
)

X_train_scaled = X_all_scaled[X_train.index]
X_test_scaled = X_all_scaled[X_test.index]

model = LogisticRegression()
model.fit(X_train_scaled, y_train)

问题不在于 transform 本身，而在于 fit_transform(X) 使用了全部样本。

正确写法是：

scaler = StandardScaler()

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

对于均值、标准差、中位数、类别集合、词表、特征选择阈值、降维方向等所有从数据中学习的状态，都应遵循同一原则。

---

三、Pipeline：把转换和模型组成一个整体

1. Pipeline 的结构

Pipeline 是一系列按顺序执行的数据转换器，最后可以连接一个预测器。中间步骤必须实现 fit 和 transform，最后一步至少需要实现 fit。(scikit-learn.org)

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

pipeline = Pipeline([
    ("scale", StandardScaler()),
    ("model", LogisticRegression()),
])

它的逻辑相当于：

X
 │
 ▼
StandardScaler.fit_transform(X_train)
 │
 ▼
LogisticRegression.fit(X_scaled_train, y_train)

调用：

pipeline.fit(X_train, y_train)

时，Pipeline 会：

1. 对第一个步骤调用 fit_transform；
2. 把转换结果交给下一个步骤；
3. 对中间步骤重复这一过程；
4. 对最后一个步骤调用 fit。

调用：

pipeline.predict(X_test)

时，Pipeline 会：

1. 对输入数据依次调用中间步骤的 transform；
2. 把最终转换后的数据交给模型；
3. 调用模型的 predict。

因此，预处理器和模型的状态可以放在同一个对象中，减少“训练时做了一套处理、预测时忘了做处理”的错误。

2. 为什么 Pipeline 能避免交叉验证泄漏

假设要进行 $k$ 折交叉验证。在第 $r$ 折中：

• $D_r^{\text{train}}$ 是该折的训练子集；
• $D_r^{\text{valid}}$ 是该折的验证子集。

正确的每折训练流程是：

$$\phi_r = \operatorname{fit\_transformer}(D_r^{\text{train}})$$

$$\theta_r = \operatorname{fit\_model} \left( T(D_r^{\text{train}};\phi_r), y_r^{\text{train}} \right)$$

然后：

$$\hat{y}_r^{\text{valid}} = \operatorname{predict\_model} \left( T(D_r^{\text{valid}};\phi_r); \theta_r \right)$$

每一折都必须重新学习 $\phi_r$。如果在交叉验证之前就对全部数据进行标准化、缺失值填充或特征选择，那么验证折已经参与了转换器状态的学习。

Pipeline 把转换器和模型交给交叉验证工具，使每个折可以独立拟合完整流程，而不是只重新拟合最后一个模型。scikit-learn 的模型选择工具也支持通过 步骤名__参数名 的形式搜索嵌套 Pipeline 的参数。(scikit-learn.org)

3. Pipeline 参数命名

pipeline.get_params().keys()

通常会包含：

scale__with_mean
model__C
model__max_iter

因此可以这样设置：

pipeline.set_params(
    scale__with_mean=False,
    model__C=0.5,
)

也可以在网格搜索中使用：

param_grid = {
    "model__C": [0.1, 1.0, 10.0],
    "model__class_weight": [None, "balanced"],
}

这里的双下划线不是 Python 属性访问，而是 scikit-learn 组合估计器用于定位内部参数的约定。

4. named_steps 与拟合后的对象

pipeline.fit(X_train, y_train)

fitted_scaler = pipeline.named_steps["scale"]
fitted_model = pipeline.named_steps["model"]

print(fitted_scaler.mean_)
print(fitted_model.classes_)

不要依赖最初传入 Pipeline 的对象来观察拟合状态，尤其是在启用缓存时。Pipeline 的缓存会克隆转换器，实际拟合对象应通过 named_steps 或 steps 访问。(scikit-learn.org)

---

四、异构数据预处理：数值列和类别列分开处理

真实表格数据通常同时包含：

• 数值列：年龄、金额、次数；
• 类别列：城市、设备类型、会员等级；
• 缺失值；
• 可能需要删除的标识列。

不能把所有列直接交给 StandardScaler。类别字符串也不能直接交给大多数数值模型。

1. 常用转换器

SimpleImputer

用于填充缺失值：

SimpleImputer(strategy="median")
SimpleImputer(strategy="most_frequent")
SimpleImputer(strategy="constant", fill_value="missing")

例如：

• 数值列常用中位数；
• 类别列可以使用最高频值或固定字符串；
• 填充值必须只从训练数据学习，不能让测试集参与统计。

StandardScaler

标准化数值特征：

$$x' = \frac{x-u}{s}$$

其中 $u$ 和 $s$ 通常由训练数据估计。它常用于逻辑回归、线性模型、支持向量机、最近邻等对特征尺度敏感的模型。

OneHotEncoder

将类别列转换为多个 0/1 特征。例如：

city
----
杭州
上海
杭州

可能转换为：

city_上海  city_杭州
0          1
1          0
0          1

训练阶段未出现、预测阶段新出现的类别是一个重要边界。设置：

OneHotEncoder(handle_unknown="ignore")

时，新类别不会导致转换失败，而是对应的已知类别列全部为 0。这样可以提高线上输入的容错性，但也意味着“未知类别”可能与某些真实的全零组合共享表示，业务上需要监控未知类别比例。

2. ColumnTransformer

ColumnTransformer 用于对不同列应用不同的转换流程。官方示例也将其用于混合文本、数值或类别数据的表格管道。(scikit-learn.org)

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

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

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

categorical_pipeline = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("onehot", OneHotEncoder(handle_unknown="ignore")),
])

preprocessor = ColumnTransformer([
    ("numeric", numeric_pipeline, numeric_features),
    ("categorical", categorical_pipeline, categorical_features),
])

数据流如下：

flowchart LR
    X[原始 DataFrame] --> N[数值列]
    X --> C[类别列]

    N --> NI[中位数填充]
    NI --> NS[标准化]

    C --> CI[最高频值填充]
    CI --> CO[One-Hot 编码]

    NS --> U[列拼接]
    CO --> U
    U --> M[分类模型]

ColumnTransformer 的关键不是“把几列传给几个对象”这么简单，而是建立了一个固定的特征契约：

1. 输入列名必须存在；
2. 每类列使用自己的转换逻辑；
3. 输出列顺序由转换器决定；
4. 模型只接收转换后的矩阵；
5. 预测时必须经过完全相同的列选择和转换。

3. 保持输出为 DataFrame

部分转换器可以配置输出容器：

preprocessor.set_output(transform="pandas")

或者：

from sklearn import set_config

set_config(transform_output="pandas")

这样可以更容易检查转换后的列名和形状。Pipeline 的 set_output 也支持配置转换结果的容器类型；当前接口包括默认格式、pandas 和 polars。(scikit-learn.org)

但不要把“输出为 DataFrame”误解为所有估计器都保留原始语义。经过 One-Hot 编码后，列已经是展开后的特征；经过降维后，列也不再对应原始字段。

---

五、一个可运行的端到端分类示例

下面使用 scikit-learn 自带的乳腺癌分类数据集。为了演示异构表格数据，先将数组转换为 DataFrame，并人为构造两个类别列和少量缺失值。

1. 安装和环境检查

python3.14 -m venv .venv

Linux 或 macOS：

source .venv/bin/activate

Windows PowerShell：

.venv\Scripts\Activate.ps1

安装：

python -m pip install -U scikit-learn pandas joblib

验证：

python -c "import sys, sklearn, pandas; print(sys.version); print(sklearn.__version__); print(pandas.__version__)"

如果导入失败，应先确认：

• python 和 pip 是否来自同一个虚拟环境；
• 当前平台是否有对应 Python 3.14 的二进制依赖；
• NumPy、SciPy 和 scikit-learn 是否被混用了不兼容版本；
• 是否在 free-threaded Python 构建中运行。

2. 完整代码

from __future__ import annotations

from pathlib import Path

import joblib
import numpy as np
import pandas as pd

from sklearn.compose import ColumnTransformer
from sklearn.datasets import load_breast_cancer
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
    accuracy_score,
    classification_report,
    confusion_matrix,
    f1_score,
    precision_score,
    recall_score,
    roc_auc_score,
)
from sklearn.model_selection import (
    GridSearchCV,
    StratifiedKFold,
    train_test_split,
)
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler


def build_dataframe() -> tuple[pd.DataFrame, pd.Series]:
    data = load_breast_cancer(as_frame=True)

    X = data.data.copy()
    y = data.target.copy()

    # 构造两个用于演示的类别列。
    X["city"] = np.where(X["mean radius"] > X["mean radius"].median(), "Hangzhou", "Ningbo")
    X["plan"] = np.where(X["mean texture"] > X["mean texture"].median(), "pro", "basic")

    # 构造缺失值，演示填充器必须位于 Pipeline 内。
    X.loc[X.index[::37], "mean radius"] = np.nan
    X.loc[X.index[::53], "city"] = None

    return X, y


def build_pipeline(feature_names: list[str]) -> Pipeline:
    numeric_features = [
        name for name in feature_names
        if name not in {"city", "plan"}
    ]
    categorical_features = ["city", "plan"]

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

    categorical_pipeline = Pipeline([
        ("imputer", SimpleImputer(strategy="most_frequent")),
        ("onehot", OneHotEncoder(handle_unknown="ignore")),
    ])

    preprocessor = ColumnTransformer([
        ("numeric", numeric_pipeline, numeric_features),
        ("categorical", categorical_pipeline, categorical_features),
    ])

    classifier = LogisticRegression(
        max_iter=2_000,
        random_state=42,
    )

    return Pipeline([
        ("preprocessor", preprocessor),
        ("classifier", classifier),
    ])


def main() -> None:
    X, y = build_dataframe()

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

    pipeline = build_pipeline(X.columns.tolist())

    cv = StratifiedKFold(
        n_splits=5,
        shuffle=True,
        random_state=42,
    )

    search = GridSearchCV(
        estimator=pipeline,
        param_grid={
            "classifier__C": [0.1, 1.0, 10.0],
            "classifier__class_weight": [None, "balanced"],
        },
        scoring="roc_auc",
        cv=cv,
        n_jobs=-1,
        refit=True,
    )

    search.fit(X_train, y_train)

    best_pipeline = search.best_estimator_

    y_pred = best_pipeline.predict(X_test)
    y_proba = best_pipeline.predict_proba(X_test)[:, 1]

    print("best_params:", search.best_params_)
    print("best_cv_roc_auc:", round(search.best_score_, 4))
    print("test_accuracy:", round(accuracy_score(y_test, y_pred), 4))
    print("test_precision:", round(precision_score(y_test, y_pred), 4))
    print("test_recall:", round(recall_score(y_test, y_pred), 4))
    print("test_f1:", round(f1_score(y_test, y_pred), 4))
    print("test_roc_auc:", round(roc_auc_score(y_test, y_proba), 4))
    print("confusion_matrix:")
    print(confusion_matrix(y_test, y_pred))
    print(classification_report(y_test, y_pred))

    artifact_dir = Path("artifacts")
    artifact_dir.mkdir(exist_ok=True)

    model_path = artifact_dir / "breast_cancer_pipeline.joblib"
    joblib.dump(best_pipeline, model_path)

    loaded_pipeline: Pipeline = joblib.load(model_path)
    loaded_pred = loaded_pipeline.predict(X_test)

    if not np.array_equal(y_pred, loaded_pred):
        raise RuntimeError("加载后的模型预测结果与保存前不一致")

    print("saved_to:", model_path)
    print("reload_check: passed")


if __name__ == "__main__":
    main()

3. 每一步为什么成立

数据划分

train_test_split(
    X,
    y,
    test_size=0.2,
    stratify=y,
    random_state=42,
)

test_size=0.2 将约 20% 的样本留作最终测试。stratify=y 使训练集和测试集尽量保持相似的类别比例。random_state 固定随机过程，便于复现。

测试集不能参与 GridSearchCV 的参数选择，否则最终测试分数就不再是独立评估。

预处理器

数值管道先填充缺失值，再标准化：

数值列
  → 中位数填充
  → 标准化

类别管道先填充缺失值，再独热编码：

类别列
  → 最高频值填充
  → One-Hot 编码

两条管道由 ColumnTransformer 并行执行，最后拼接成模型输入。

网格搜索

"scoring": "roc_auc"

表示用 ROC AUC 作为选择指标，而不是使用分类器默认的 score。这很重要，因为“优化什么指标”决定了参数选择方向。

"refit": True

表示搜索完成后，用最优参数在整个 X_train 上重新拟合一个 best_estimator_。因此最终保存的是完整的“预处理器 + 分类器”对象，而不是某一折交叉验证中的临时模型。多指标搜索时，refit 也可以指定要用于选择和重新拟合的指标名称。(scikit-learn.org)

---

六、分类评测：指标不是越多越好，而是要对应错误代价

1. 混淆矩阵

二分类混淆矩阵通常包含：

实际 / 预测	负类	正类
负类	TN	FP
正类	FN	TP

四个量的含义是：

• TP：实际为正，预测为正；
• TN：实际为负，预测为负；
• FP：实际为负，却预测为正；
• FN：实际为正，却预测为负。

2. Accuracy

$$\text{Accuracy} = \frac{TP+TN}{TP+TN+FP+FN}$$

Accuracy 衡量总体预测正确比例。类别平衡时它比较直观，但类别极不平衡时可能误导。

例如，1000 个样本中只有 10 个正类。模型把全部样本都预测为负类：

$$\text{Accuracy} = \frac{990}{1000}=99\%$$

但它一个正类也没有识别出来。

3. Precision

$$\text{Precision} = \frac{TP}{TP+FP}$$

Precision 回答：

所有被模型判定为正类的样本中，真正为正的比例是多少？

当误报代价高时，例如人工审核资源有限，Precision 比单独看 Accuracy 更有意义。

4. Recall

$$\text{Recall} = \frac{TP}{TP+FN}$$

Recall 也称真正率，回答：

所有实际正类中，模型找到了多少？

当漏报代价高时，例如疾病筛查、风险拦截，Recall 通常更重要。

5. F1

$$F_1 = 2 \cdot \frac{\text{Precision}\cdot\text{Recall}} {\text{Precision}+\text{Recall}}$$

F1 是 Precision 和 Recall 的调和平均。调和平均会强烈惩罚其中一个指标很低的情况。

如果：

Precision = 0.9
Recall    = 0.1

那么 F1 不会接近 0.5，而约为：

$$F_1 = 2 \times \frac{0.9\times0.1}{0.9+0.1} =0.18$$

这符合“不能只顾其中一边”的直觉。

6. ROC AUC 与概率阈值

二分类模型通常先产生一个分数或概率：

$$p = P(y=1 \mid x)$$

默认情况下常用阈值 $0.5$：

$$\hat y = \begin{cases} 1, & p \ge 0.5 \\ 0, & p < 0.5 \end{cases}$$

但阈值不是自然规律。可以根据业务需要调整：

threshold = 0.3
y_pred_custom = (y_proba >= threshold).astype(int)

降低阈值通常会增加正类预测数，Recall 可能上升，Precision 可能下降；但具体结果取决于分数分布。

ROC AUC 衡量模型排序正负样本的能力，通常不依赖某一个固定阈值。它适合比较排序能力，但不能代替阈值确定，也不能直接说明概率是否校准。

因此：

• 选择模型时可以用 ROC AUC；
• 线上决策时仍需明确阈值；
• 业务关心正类比例和概率可信度时，还应检查 Precision-Recall 曲线、概率校准和不同阈值下的成本。

---

七、交叉验证和独立测试集的边界

1. 训练集、交叉验证和测试集分别做什么

一个清晰的流程是：

全部数据
   │
   ├── 训练集：用于交叉验证、选参数、拟合最终训练模型
   │
   └── 测试集：最后只评测一次

交叉验证在训练集内部继续切分：

训练集
   ├── 第 1 折验证
   ├── 第 2 折验证
   ├── 第 3 折验证
   ├── 第 4 折验证
   └── 第 5 折验证

在第 $r$ 折中，模型只使用该折训练部分拟合，然后在验证部分计算指标。最终交叉验证分数通常是各折分数的平均：

$$\bar{s} = \frac{1}{k}\sum_{r=1}^{k}s_r$$

还应关注折间波动：

$$s_{\text{std}} = \sqrt{ \frac{1}{k} \sum_{r=1}^{k} (s_r-\bar{s})^2 }$$

平均分较高但标准差很大的模型，可能对数据划分敏感。

2. 为什么使用 StratifiedKFold

分类任务中，如果某一折正类过少，指标会变得不稳定。StratifiedKFold 会尽量保持每一折的类别比例与整体相近：

cv = StratifiedKFold(
    n_splits=5,
    shuffle=True,
    random_state=42,
)

但它不是时间序列切分工具。如果数据具有时间顺序，随机打乱可能把未来信息带入过去，应使用符合业务时间边界的切分方法。

同样，如果同一个用户、设备、患者或订单会产生多行记录，普通随机切分可能让同一实体同时出现在训练集和测试集中。这会造成实体级泄漏，此时应考虑按实体分组的切分。

---

八、回归任务的对应流程

Pipeline 并不只适用于分类。回归任务的结构相同，只是最终估计器和评测指标不同。

from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.linear_model import Ridge
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

numeric_features = ["area", "rooms"]
categorical_features = ["district", "orientation"]

preprocessor = ColumnTransformer([
    (
        "numeric",
        Pipeline([
            ("imputer", SimpleImputer(strategy="median")),
            ("scaler", StandardScaler()),
        ]),
        numeric_features,
    ),
    (
        "categorical",
        Pipeline([
            ("imputer", SimpleImputer(strategy="most_frequent")),
            ("onehot", OneHotEncoder(handle_unknown="ignore")),
        ]),
        categorical_features,
    ),
])

regression_pipeline = Pipeline([
    ("preprocessor", preprocessor),
    ("model", Ridge(alpha=1.0)),
])

regression_pipeline.fit(X_train, y_train)
y_pred = regression_pipeline.predict(X_test)

mae = mean_absolute_error(y_test, y_pred)
rmse = mean_squared_error(y_test, y_pred) ** 0.5
r2 = r2_score(y_test, y_pred)

print("MAE:", mae)
print("RMSE:", rmse)
print("R2:", r2)

1. MAE

$$\text{MAE} = \frac{1}{n} \sum_{i=1}^{n} |y_i-\hat y_i|$$

MAE 的单位与目标值相同。例如目标是房价，MAE 也是房价单位，解释比较直接。

2. RMSE

$$\text{RMSE} = \sqrt{ \frac{1}{n} \sum_{i=1}^{n} (y_i-\hat y_i)^2 }$$

平方项会放大大误差，因此 RMSE 对异常大误差更敏感。

3. $R^2$

$$R^2 = 1- \frac{\sum_i(y_i-\hat y_i)^2} {\sum_i(y_i-\bar{y})^2}$$

其中 $\bar{y}$ 是真实目标的均值。$R^2$ 可以为负数：这表示模型还不如直接预测训练目标均值的基线。

---

九、超参数搜索：搜索的是完整流程

1. 参数和超参数

模型参数是通过 fit 从数据中学习的量，例如线性模型的系数：

model.coef_
model.intercept_

超参数是在训练前指定或搜索的量，例如：

LogisticRegression(C=1.0)
Ridge(alpha=1.0)

C 和 alpha 不是从每次 fit 自动确定的模型参数，而是控制模型行为的超参数。

2. 搜索预处理参数

可以同时搜索预处理和模型参数：

param_grid = {
    "preprocessor__numeric__imputer__strategy": [
        "mean",
        "median",
    ],
    "classifier__C": [
        0.1,
        1.0,
        10.0,
    ],
}

参数路径逐层表示：

Pipeline
└── preprocessor
    └── numeric
        └── imputer
            └── strategy

这体现了组合估计器的工程价值：搜索过程可以重新构建和评测完整的数据流，而不是只改变最后一个模型。

3. 用一个网格替换模型

还可以搜索不同类型的最终模型：

from sklearn.ensemble import RandomForestClassifier

param_grid = [
    {
        "classifier": [LogisticRegression(max_iter=2000, random_state=42)],
        "classifier__C": [0.1, 1.0, 10.0],
    },
    {
        "classifier": [RandomForestClassifier(random_state=42)],
        "classifier__n_estimators": [100, 300],
        "classifier__max_depth": [None, 10],
    },
]

这要求两个模型都能接收预处理后的特征矩阵，并且网格中的参数路径分别对应各自的类。

但模型类型越多，搜索成本越高，指标解释也越复杂。搜索空间应由明确的建模假设驱动，而不是无边界地堆参数。

---

十、特征名、稀疏矩阵和形状诊断

1. 查看转换后的特征名

当 Pipeline 的预处理器支持特征名输出时，可以尝试：

preprocessor = best_pipeline.named_steps["preprocessor"]

feature_names = preprocessor.get_feature_names_out()
print(feature_names[:10])
print("feature_count:", len(feature_names))

常见输出类似：

[
    "numeric__mean radius",
    "numeric__mean texture",
    "categorical__city_Hangzhou",
    "categorical__city_Ningbo",
]

这些名称带有转换器前缀，能够帮助定位特征来源。

2. 稀疏输出

OneHotEncoder 可能产生稀疏矩阵。当类别数量很多时，稀疏矩阵可以避免存储大量零值。

某些模型支持稀疏输入，某些模型不支持或会隐式转换为稠密矩阵。若遇到内存暴涨或类型错误，应检查：

Xt = preprocessor.fit_transform(X_train)
print(type(Xt))
print(Xt.shape)

还可以显式控制 One-Hot 编码器的输出格式：

OneHotEncoder(
    handle_unknown="ignore",
    sparse_output=True,
)

具体参数名和支持情况应以所安装版本的 API 文档为准；版本升级时尤其不要假设旧参数名称永远有效。

3. 常见形状错误

以下错误通常来自把一维数据当成二维特征矩阵：

x = [1, 2, 3, 4]
model.fit(x, y)

通常应改为：

x = [[1], [2], [3], [4]]

或者：

import numpy as np

x = np.array([1, 2, 3, 4]).reshape(-1, 1)

scikit-learn 通常约定：

X.shape == (n_samples, n_features)
y.shape == (n_samples,)

如果使用 pandas，预测时也应尽量保持列名和列语义一致：

new_data = pd.DataFrame([
    {
        "age": 42,
        "income": 18000,
        "city": "Hangzhou",
        "plan": "pro",
    }
])

prediction = best_pipeline.predict(new_data)

不要把训练时的 DataFrame 随意转换成按位置排列的数组后再传入，除非已经明确验证列顺序完全一致。

---

十一、模型持久化：保存的不只是模型参数

1. 为什么必须保存完整 Pipeline

错误做法：

preprocessor.fit(X_train)
X_train_transformed = preprocessor.transform(X_train)

model.fit(X_train_transformed, y_train)

joblib.dump(model, "model.joblib")

这个文件只保存了模型，没有保存：

• 缺失值填充规则；
• 标准化均值和尺度；
• 类别到整数列的映射；
• 未知类别处理策略；
• 输入列选择顺序。

部署时如果忘记复现其中任何一步，预测结果就可能错误。

正确做法：

full_pipeline = Pipeline([
    ("preprocessor", preprocessor),
    ("model", model),
])

full_pipeline.fit(X_train, y_train)
joblib.dump(full_pipeline, "full_pipeline.joblib")

保存对象应代表完整预测函数：

$$f(x) = \operatorname{model} \left( T(x;\phi); \theta \right)$$

而不是只保存：

$$\operatorname{model}(x;\theta)$$

2. joblib、pickle、skops 和 ONNX

joblib

import joblib

joblib.dump(pipeline, "model.joblib")
pipeline_loaded = joblib.load("model.joblib")

joblib 对包含大型 NumPy 数组的对象通常比较方便，也支持内存映射等能力。但它基于 pickle 机制，加载不可信文件可能执行任意代码。(scikit-learn.org)

pickle

import pickle

with open("model.pkl", "wb") as file:
    pickle.dump(pipeline, file, protocol=5)

with open("model.pkl", "rb") as file:
    loaded = pickle.load(file)

它是 Python 标准库方案，但同样不能加载不可信来源的文件。

skops.io

如果需要比 pickle 系列更谨慎地检查模型文件，可以使用 skops.io：

import skops.io as sio

sio.dump(pipeline, "model.skops")

unknown_types = sio.get_untrusted_types(file="model.skops")
print(unknown_types)

loaded = sio.load(
    "model.skops",
    trusted=unknown_types,
)

实际生产环境不应盲目把所有未知类型自动加入信任列表，而应审查文件内容后再决定允许哪些类型。skops.io 不会像 pickle 那样在加载时默认执行任意代码，但仍需要与训练环境相匹配，并且支持的对象类型比 pickle 少。(scikit-learn.org)

ONNX

ONNX 更适合只需要推理、不需要恢复 Python 对象的场景。它可以让推理环境不依赖完整 Python 环境，但不是所有 scikit-learn 模型都能直接转换，第三方或自定义估计器还可能需要编写转换器。(scikit-learn.org)

3. 持久化的版本边界

模型文件不等于永久兼容格式。使用 joblib、pickle 或 skops.io 保存的对象，通常依赖：

• Python 版本；
• scikit-learn 版本；
• NumPy、SciPy 等依赖版本；
• 自定义类和函数所在的可导入模块；
• CPU、操作系统和底层二进制环境。

scikit-learn 官方文档明确指出，不支持把模型从一个 scikit-learn 版本加载到另一个版本；即使文件实际上能够加载，也不能因此认为跨版本行为得到保证。(scikit-learn.org)

因此，模型制品至少应同时记录：

model.joblib
metadata.json
requirements-lock.txt

例如：

{
  "model_name": "breast-cancer-classifier",
  "python": "3.14.x",
  "scikit_learn": "1.x.y",
  "numpy": "x.y.z",
  "pandas": "x.y.z",
  "feature_columns": [
    "mean radius",
    "mean texture",
    "city",
    "plan"
  ],
  "positive_class": 1,
  "training_data_version": "dataset-2026-01"
}

具体版本号应由训练环境实际生成，不能手写成示例值后当作真实环境信息。

---

十二、加载模型后的验证

仅仅成功执行 joblib.load 不代表模型可以安全上线。加载后应至少验证四件事。

1. 能否处理一条合法输入

sample = X_test.iloc[:1]
prediction = loaded_pipeline.predict(sample)
probability = loaded_pipeline.predict_proba(sample)

print(prediction)
print(probability)

2. 输出形状是否符合接口

if prediction.shape != (1,):
    raise RuntimeError(f"unexpected prediction shape: {prediction.shape}")

if probability.shape[0] != 1:
    raise RuntimeError("unexpected probability row count")

3. 保存前后结果是否一致

before = best_pipeline.predict(X_test)
after = loaded_pipeline.predict(X_test)

if not np.array_equal(before, after):
    raise RuntimeError("model artifact is not behaviorally identical")

对于浮点概率，不一定要求逐位完全相同，可以使用容差：

before_proba = best_pipeline.predict_proba(X_test)
after_proba = loaded_pipeline.predict_proba(X_test)

if not np.allclose(before_proba, after_proba, rtol=1e-10, atol=1e-12):
    raise RuntimeError("probabilities changed after reload")

4. 输入列变化是否被发现

expected_columns = set(X_train.columns)
actual_columns = set(sample.columns)

missing = expected_columns - actual_columns
extra = actual_columns - expected_columns

if missing or extra:
    raise ValueError({
        "missing_columns": sorted(missing),
        "extra_columns": sorted(extra),
    })

这类检查属于应用层契约，不是 Pipeline 自动替你完成的全部工作。模型能够接受某个输入，并不意味着这个输入在业务语义上是正确的。

---

十三、失败路径和诊断方法

1. NotFittedError

表现：

This ... instance is not fitted yet

原因通常是：

• 忘记调用 fit；
• 保存了未拟合对象；
• 在 Pipeline 外使用了错误的转换器实例；
• 训练过程异常中断后仍尝试预测。

诊断：

from sklearn.utils.validation import check_is_fitted

check_is_fitted(loaded_pipeline)

对于 Pipeline，还应检查：

print(loaded_pipeline.named_steps)

2. 未知类别错误

如果 OneHotEncoder 没有设置：

handle_unknown="ignore"

线上出现训练阶段未见过的类别时，可能在 transform 阶段失败。

修复方法不是简单地在预测前删除该列，而是明确选择：

• 训练时建立稳定的类别字典；
• 对未知类别使用忽略策略；
• 或在输入校验层拒绝未知值。

选择哪一种取决于未知类别代表数据错误，还是正常的新业务状态。

3. 特征数量或列顺序错误

线性模型常见错误：

X has 10 features, but ... is expecting 12 features

原因可能是：

• 预测时少了一列；
• One-Hot 编码方式不同；
• 训练和预测使用了不同的列集合；
• 手工拼接特征时顺序变化。

如果完整 Pipeline 仍然报错，应先打印：

print(X_train.columns.tolist())
print(X_new.columns.tolist())

Xt = loaded_pipeline.named_steps["preprocessor"].transform(X_new)
print(Xt.shape)

4. NaN 或无限值错误

如果模型不接受缺失值，确认：

print(X.isna().sum())
print(np.isinf(numeric_array).sum())

并确认缺失值填充器处于 Pipeline 内，而不是只对训练集单独调用过。

还要注意：字符串 "NaN"、空字符串 "" 和真正的 np.nan 不是同一个值。数据清洗阶段应将业务中的缺失表示统一起来。

5. 评分指标不适合任务

如果正类极少，Accuracy 很高并不一定说明模型有效。应同时检查：

print(confusion_matrix(y_test, y_pred))
print(classification_report(y_test, y_pred, zero_division=0))

如果模型输出概率，还应按不同阈值重新计算 Precision、Recall 和业务成本，而不是只保存默认阈值下的一个分数。

---

十四、训练、评测和持久化的生产边界

1. 训练阶段

训练阶段负责：

读取带标签数据
  → 数据划分
  → Pipeline + 交叉验证
  → 选择超参数
  → 在训练集整体上重新拟合
  → 在独立测试集上评测
  → 保存完整制品和元数据

此时允许使用标签、交叉验证和训练集统计量。

2. 推理阶段

推理阶段通常只负责：

接收无标签输入
  → 校验字段和类型
  → 调用已加载 Pipeline
  → 返回预测及必要的概率
  → 记录版本、延迟和异常

推理服务不应重新调用：

pipeline.fit(...)

否则服务会把线上请求误当成训练数据，并且可能产生竞态、状态漂移和不可复现结果。

3. 重新训练阶段

当数据分布、标签定义或业务规则发生变化时，应创建新的模型制品，而不是覆盖旧文件：

model-v001.joblib
model-v002.joblib
model-v003.joblib

部署时通过明确的模型版本切换，保留回滚能力。每个版本都应能够追溯到：

• 训练数据版本；
• 特征列和转换逻辑；
• 代码提交版本；
• 依赖环境；
• 评测结果；
• 模型文件校验值。

4. 并发和缓存

加载后的 Pipeline 通常被多个请求共享读取，但不应在多个请求中同时调用 fit 或修改其参数。若使用 Pipeline 的 memory 缓存，需要理解缓存会克隆转换器，并且缓存目录可能产生磁盘占用、旧缓存失效和并发访问问题。官方接口说明中，memory 用于缓存拟合后的转换器，最终估计器不会被缓存。(scikit-learn.org)

---

十五、一个最小但完整的工程检查清单

在提交模型前，至少确认以下事实：

• X 的形状是 (样本数, 特征数)；
• y 的长度与 X 的样本数一致；
• 测试集没有参与任何 fit；
• 缺失值填充器位于完整训练 Pipeline 内；
• 类别编码器位于完整训练 Pipeline 内；
• 交叉验证的每一折都重新拟合预处理器；
• 超参数通过 步骤名__参数名 正确传递；
• 测试集只用于最终评测；
• 指标与类别不平衡及业务错误代价匹配；
• 保存的是完整 Pipeline，而不是裸模型；
• 加载模型的环境与训练环境一致；
• 不加载来源不可信的 pickle/joblib 文件；
• 模型制品带有版本、特征列和依赖元数据；
• 加载后通过样例输入、形状和保存前后预测一致性检查；
• 线上预测路径不会调用 fit。

Pipeline 的核心不是减少几行代码，而是把“数据如何变成模型输入”变成模型的一部分。预处理参数、类别映射、特征顺序和最终预测器共同构成了可执行的预测函数。只有当这些状态在训练、交叉验证、评测、保存和加载之间保持一致，模型分数才有意义，线上预测才具有可解释的工程边界。

---

系列导航与关联阅读

• 系列入口：Python 完整学习路线：从语言模型、并发到 Web、数据、AI 与生产交付
• 上一篇：Python 数据可视化：Matplotlib、Seaborn、坐标、布局和可解释图表
• 下一篇：Python PyTorch 工程：Tensor、Autograd、Dataset、训练和检查点
• 延伸：pandas 完整基础：Series、DataFrame、索引、缺失值、分组和连接

官方资料

• PyTorch Documentation
• OpenAI Python SDK

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。