如何封装Keras模型适配Scikit-learn Stacking集成学习

问题描述 #

已训练完成的Keras模型通过自定义封装类适配Scikit-learn接口后，在VotingClassifier（软/硬投票模式）中可正常运行，但调用StackingClassifier.fit()时抛出形状不匹配的ValueError，核心报错提示概率数组形状(387348,3)无法广播到索引结果形状(387348,1,3)，同时伴随训练fold中类别数与总类别数不匹配的警告。

原代码与报错信息 #

封装类及调用代码 #

from sklearn.base import BaseEstimator, ClassifierMixin
from sklearn.preprocessing import OneHotEncoder
from sklearn.ensemble import VotingClassifier, StackingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import StratifiedKFold
import numpy as np

class KerasWrapperWithEncoder(BaseEstimator, ClassifierMixin):
    def __init__(self, keras_model, classes_):
        self.keras_model = keras_model
        self.encoder = OneHotEncoder(sparse_output=False)
        self.classes_ = classes_  # 定义可用类别

    def fit(self, X, y):
        # 模型已训练完成，无需重新拟合
        y_reshaped = y.reshape(-1, 1)
        self.encoder.fit(y_reshaped)
        return self

    def predict(self, X):
        predictions = self.keras_model.predict(X)
        np_argmax = np.argmax(predictions, axis=1)
        return np_argmax

    def predict_proba(self, X):
        probabilities = self.keras_model.predict(X)
        print("Shape of probabilities:", probabilities.shape)  # 调试信息
        return probabilities


keras_wrapped_models_with_encoder = [
    (name.replace(' ', '_').replace('__', '_'), KerasWrapperWithEncoder(model, _target_classes_))
    for name, model in keras_models.items()
]

# VotingClassifier可正常运行
voting_clf = VotingClassifier(
         estimators=all_estimators,
         voting='soft', 
         n_jobs=3,  
         verbose=True)
voting_clf.fit(X_train, y_train) 

# StackingClassifier报错
cv = StratifiedKFold(n_splits=3, shuffle=True, random_state=42)
keras_stacking_models_current_year = StackingClassifier(
    estimators=all_estimators,
    final_estimator=LogisticRegression(),
    cv=cv,
    verbose=3,
    # n_jobs=2
)

keras_stacking_models_current_year.fit(X_train, y_train) 

报错信息 #

======================================stacking_models_all_models=============================
12105/12105 [==============================] - 25s 2ms/step
Shape of probabilities: (387348, 3)
Number of classes in training fold (1) does not match total number of classes (3). Results may not be appropriate for your use case. To fix this, use a cross-validation technique resulting in properly stratified folds
_enforce_prediction_order(classes, predictions, n_classes, method)
   1457             dtype=predictions.dtype,
   1458         )
-> 1459         predictions_for_all_classes[:, classes] = predictions
   1460         predictions = predictions_for_all_classes
   1461     return predictions

ValueError: shape mismatch: value array of shape (387348,3) could not be broadcast to indexing result of shape (387348,1,3)

解决方法 #

1. 修复交叉验证的类别分层问题 #

报错的核心原因是某个交叉验证fold中仅包含1个类别，导致Scikit-learn在对齐全类别概率时形状不匹配。解决这个问题的关键是确保每个fold都包含所有类别：

• 检查数据集的类别分布：如果存在样本极少的类别，可考虑合并小类别，或者为小类别增加样本（过采样）。
• 调整交叉验证策略：使用RepeatedStratifiedKFold替代StratifiedKFold，通过重复分层抽样减少出现单类别fold的概率；或者增大n_splits值（但需保证每个类别的样本数足够分配到各个fold）。
  示例代码：

from sklearn.model_selection import RepeatedStratifiedKFold

# 重复3次，每次3折分层抽样
cv = RepeatedStratifiedKFold(n_splits=3, n_repeats=3, random_state=42)

2. 简化Keras模型封装类 #

原封装类中的OneHotEncoder属于冗余代码（模型已预训练，无需重新拟合编码器），且classes_的处理可能与Scikit-learn的期望不一致。简化后的封装类如下：

class KerasWrapper(BaseEstimator, ClassifierMixin):
    def __init__(self, keras_model, classes):
        self.keras_model = keras_model
        # 确保classes_是numpy数组，顺序与Keras模型输出的概率列严格一致
        self.classes_ = np.array(classes)

    def fit(self, X, y):
        # 预训练模型无需重新拟合，直接返回自身
        return self

    def predict(self, X):
        # 关闭Keras的预测进度条，避免干扰
        predictions = self.keras_model.predict(X, verbose=0)
        # 直接映射到类别标签，确保与classes_顺序一致
        return self.classes_[np.argmax(predictions, axis=1)]

    def predict_proba(self, X):
        probabilities = self.keras_model.predict(X, verbose=0)
        # 确保输出的概率矩阵列顺序与self.classes_完全匹配
        return probabilities

修改后重新生成封装模型列表：

keras_wrapped_models = [
    (name.replace(' ', '_').replace('__', '_'), KerasWrapper(model, _target_classes_))
    for name, model in keras_models.items()
]

3. 验证类别一致性 #

确保_target_classes_的顺序与Keras模型输出的概率列顺序完全一致（比如Keras模型最后一层是Dense(3, activation='softmax')，则_target_classes_的顺序要对应这3个类别的预测顺序）。

内容的提问来源于stack exchange，提问作者Mohamed Kounda SACKO