如何正确堆叠含自定义选择器的Sklearn Pipeline

错误原因 #

• 自定义选择器方法参数错误：原DictionarySelector的transform方法参数为key，但sklearn规范中transform方法需接收输入数据（此处为字典），导致无法正确识别样本数量，触发不匹配报错。
• 硬编码训练数据到选择器：初始化DictionarySelector时直接绑定了训练用的in_dict，导致StackingClassifier在交叉验证过程中无法动态获取对应折的数据，进一步加剧样本数不匹配问题。
• 测试代码笔误：原代码测试pipe_vector时误用了pipe_images进行拟合预测，属于逻辑错误。

修正后的代码 #

import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.neural_network import MLPClassifier
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.ensemble import RandomForestClassifier
from sklearn.ensemble import StackingClassifier
from sklearn.datasets import load_iris
from sklearn.metrics import accuracy_score
from sklearn.pipeline import Pipeline

# 修正后的自定义选择器：仅按key从输入字典提取对应数据
class DictionarySelector(BaseEstimator, TransformerMixin):
    def __init__(self, key):
        self.key = key
        
    def fit(self, X, y=None):
        return self
    
    def transform(self, X):
        # X为传入的字典，返回对应key的数据集
        return X[self.key]

# 生成示例数据
data = load_iris()
X_images = data.data[:, :2]  # 模拟2D图像数据
X_vector = data.data[:, 2:]  # 模拟表格数据
y = data.target

# 拆分训练/测试集
X_images_train, X_images_test, X_vector_train, X_vector_test, y_train, y_test = train_test_split(
    X_images, X_vector, y, test_size=0.2, random_state=42
)

# 定义基础模型
image_model = MLPClassifier(hidden_layer_sizes=(64, 32), activation="relu", max_iter=1000)
vector_model = RandomForestClassifier(n_estimators=100, criterion="gini", max_depth=None)

# 构建Pipeline：动态从传入字典提取数据，不再绑定固定数据集
pipe_images = Pipeline([
    ('select', DictionarySelector('images')),
    ('clf', image_model)
])

pipe_vector = Pipeline([
    ('select', DictionarySelector('vector')),
    ('clf', vector_model)
])

# 创建堆叠模型
stacked_model = StackingClassifier(
    estimators=[
        ("image_mlp", pipe_images),
        ("vector_rf", pipe_vector),
    ],
    final_estimator=RandomForestClassifier(n_estimators=100),
)

# 准备训练用字典并拟合模型
train_dict = {'images': X_images_train, 'vector': X_vector_train}
stacked_model.fit(train_dict, y_train)

# 准备测试用字典并评估模型
test_dict = {'images': X_images_test, 'vector': X_vector_test}
y_pred = stacked_model.predict(test_dict)
print(f"堆叠模型准确率: {accuracy_score(y_test, y_pred):.4f}")

关键修正点 #

• 重构选择器逻辑：移除初始化时的固定字典参数，让transform方法从传入的字典中动态提取对应key的数据，适配StackingClassifier的交叉验证流程。
• Pipeline动态适配数据：Pipeline不再绑定训练集，而是在fit和predict阶段从传入的字典中获取对应数据，保证模型能处理不同阶段的数据集。
• 修正测试笔误：调整测试代码逻辑，确保每个Pipeline独立测试正常。

内容的提问来源于stack exchange，提问作者fednem