使用VotingClassifier集成自定义PyTorch CNN时遇forward()参数缺失错误

问题 #

使用Scikit-learn的VotingClassifier集成模型时触发错误：TypeError: forward() missing 1 required positional argument: 'x'。

相关代码：

leNetModel = mycnn()
leNetK = KerasClassifier(build_fn=leNetModel , epochs=100, batch_size=256, verbose=0)
leNetK._estimator_type = "classifier"
mlpModel = MLPClassifier(hidden_layer_sizes=(150,100,50), max_iter=300,activation = 'relu',solver='adam',random_state=1)
svmModel = SVC(kernel= 'linear', C=1.0, probability=True)

models = [('leNet', leNetK), ('mlp', mlpModel), ('svm', svmModel)]

其中mycnn是基于PyTorch实现的自定义CNN类（代码如下），尝试用KerasClassifier包装它解决之前的“非classifier类型”错误，但运行时触发上述参数缺失错误。移除models列表中的leNet后，集成模型可正常运行，且该自定义CNN单独使用无问题。无法控制VotingClassifier对forward()的调用逻辑，求助解决。

自定义CNN代码：

from torch.nn import Module
from torch import nn


class Model(Module):
    def __init__(self):
        super(Model, self).__init__()
        self.conv1 = nn.Conv1d(1, 10, 5)
        self.relu1 = nn.ReLU()
        self.pool1 = nn.MaxPool1d(2)
        self.conv2 = nn.Conv1d(10, 100, 5)
        self.relu2 = nn.ReLU()
        self.pool2 = nn.MaxPool1d(2)
        self.fc1 = nn.Linear(200, 100)
        self.relu3 = nn.ReLU()
        self.fc2 = nn.Linear(100, 25)
        self.relu4 = nn.ReLU()
        self.fc3 = nn.Linear(25, 2) # 二分类
        #self.fc3 = nn.Linear(25, 5) # 五分类
        self.relu5 = nn.ReLU()

    def forward(self, x):
        y = self.conv1(x)
        y = self.relu1(y)
        y = self.pool1(y)
        y = self.conv2(y)
        y = self.relu2(y)
        y = self.pool2(y)
        y = y.view(y.shape[0], -1)
        y = self.fc1(y)
        y = self.relu3(y)
        y = self.fc2(y)
        y = self.relu4(y)
        y = self.fc3(y)
        y = self.relu5(y)
        return y

错误栈信息：

Traceback (most recent call last):
  File "./votingClassifier.py", line 202, in <module>
    main()
  File "./votingClassifier.py", line 130, in main
    model.fit(X_train, Y_train)
  File ".../.local/lib/python3.8/site-packages/sklearn/ensemble/_voting.py", line 324, in fit
    return super().fit(X, transformed_y, sample_weight)
  File ".../.local/lib/python3.8/site-packages/sklearn/ensemble/_voting.py", line 74, in fit
    self.estimators_ = Parallel(n_jobs=self.n_jobs)(
  File ".../.local/lib/python3.8/site-packages/joblib/parallel.py", line 1043, in __call__
    if self.dispatch_one_batch(iterator):
  File ".../.local/lib/python3.8/site-packages/joblib/parallel.py", line 861, in dispatch_one_batch
    self._dispatch(tasks)
  File "/home/izabela/.local/lib/python3.8/site-packages/joblib/parallel.py", line 779, in _dispatch
    job = self._backend.apply_async(batch, callback=cb)
  File ".../.local/lib/python3.8/site-packages/joblib/_parallel_backends.py", line 208, in apply_async
    result = ImmediateResult(func)
  File ".../.local/lib/python3.8/site-packages/joblib/_parallel_backends.py", line 572, in __init__
    self.results = batch()
  File ".../.local/lib/python3.8/site-packages/joblib/parallel.py", line 262, in __call__
    return [func(*args, **kwargs)
  File ".../.local/lib/python3.8/site-packages/joblib/parallel.py", line 262, in <listcomp>
    return [func(*args, **kwargs)
  File "/home/izabela/.local/lib/python3.8/site-packages/sklearn/utils/fixes.py", line 216, in __call__
    return self.function(*args, **kwargs)
  File ".../.local/lib/python3.8/site-packages/sklearn/ensemble/_base.py", line 42, in _fit_single_estimator
    estimator.fit(X, y)
  File ".../.local/lib/python3.8/site-packages/keras/wrappers/scikit_learn.py", line 248, in fit
    return super().fit(x, y, **kwargs)
  File ".../.local/lib/python3.8/site-packages/keras/wrappers/scikit_learn.py", line 160, in fit
    self.model = self.build_fn(
  File ".../.local/lib/python3.8/site-packages/torch/nn/modules/module.py", line 1110, in _call_impl
    return forward_call(*input, **kwargs)
TypeError: forward() missing 1 required positional argument: 'x'

解决方案 #

问题根源 #

KerasClassifier是专门用于包装Keras模型的工具，不能直接兼容PyTorch模型。你传入的是已经实例化的PyTorch模型对象，KerasClassifier会把它当成创建模型的可调用函数执行，直接触发模型的__call__方法（对应PyTorch的forward），但此时没有传入输入参数，就会报x缺失的错误。

正确做法：自定义Scikit-learn兼容的PyTorch模型包装类 #

不需要依赖KerasClassifier，直接实现一个符合Scikit-learn分类器接口的包装类，把PyTorch模型封装进去：

import torch
import numpy as np
from sklearn.base import BaseEstimator, ClassifierMixin
from torch.utils.data import TensorDataset, DataLoader


class TorchClassifier(BaseEstimator, ClassifierMixin):
    def __init__(self, model_class, epochs=100, batch_size=256, verbose=0, device="cuda" if torch.cuda.is_available() else "cpu"):
        self.model_class = model_class
        self.epochs = epochs
        self.batch_size = batch_size
        self.verbose = verbose
        self.device = device
        self.model = None
        self.loss_fn = torch.nn.CrossEntropyLoss()
        self.optimizer = None

    def fit(self, X, y):
        # 转换输入数据为PyTorch张量
        X_tensor = torch.tensor(X, dtype=torch.float32).to(self.device)
        # 适配Conv1d输入格式：(batch_size, in_channels, seq_len)
        if len(X_tensor.shape) == 2:
            X_tensor = X_tensor.unsqueeze(1)
        y_tensor = torch.tensor(y, dtype=torch.long).to(self.device)
        
        # 初始化模型与优化器
        self.model = self.model_class().to(self.device)
        self.optimizer = torch.optim.Adam(self.model.parameters())
        
        # 创建数据加载器
        dataset = TensorDataset(X_tensor, y_tensor)
        dataloader = DataLoader(dataset, batch_size=self.batch_size, shuffle=True)
        
        # 训练循环
        self.model.train()
        for epoch in range(self.epochs):
            total_loss = 0.0
            for batch_X, batch_y in dataloader:
                self.optimizer.zero_grad()
                outputs = self.model(batch_X)
                loss = self.loss_fn(outputs, batch_y)
                loss.backward()
                self.optimizer.step()
                total_loss += loss.item()
            
            if self.verbose > 0 and (epoch + 1) % 10 == 0:
                print(f"Epoch {epoch+1}/{self.epochs}, Loss: {total_loss/len(dataloader):.4f}")
        
        return self

    def predict(self, X):
        self.model.eval()
        with torch.no_grad():
            X_tensor = torch.tensor(X, dtype=torch.float32).to(self.device)
            if len(X_tensor.shape) == 2:
                X_tensor = X_tensor.unsqueeze(1)
            outputs = self.model(X_tensor)
            _, preds = torch.max(outputs, 1)
            return preds.cpu().numpy()

    def predict_proba(self, X):
        self.model.eval()
        with torch.no_grad():
            X_tensor = torch.tensor(X, dtype=torch.float32).to(self.device)
            if len(X_tensor.shape) == 2:
                X_tensor = X_tensor.unsqueeze(1)
            outputs = self.model(X_tensor)
            # 转换为概率分布
            probs = torch.nn.functional.softmax(outputs, dim=1)
            return probs.cpu().numpy()

修改集成代码 #

把原来的KerasClassifier替换为自定义包装类：

# 传入Model类而非实例
leNetK = TorchClassifier(model_class=Model, epochs=100, batch_size=256, verbose=0)
mlpModel = MLPClassifier(hidden_layer_sizes=(150,100,50), max_iter=300,activation = 'relu',solver='adam',random_state=1)
svmModel = SVC(kernel= 'linear', C=1.0, probability=True)

models = [('leNet', leNetK), ('mlp', mlpModel), ('svm', svmModel)]

额外说明 #

• 该包装类实现了Scikit-learn分类器的核心接口，包括VotingClassifier必需的predict_proba方法（用于权重投票计算）。
• 代码中自动处理了输入维度适配，确保符合Conv1d的输入要求。
• 训练逻辑（优化器、损失函数、训练轮次等）可根据你的需求自行调整。

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