Scikit-learn Pipeline中聚类步骤fit_transform参数不匹配问题排查

问题 #

尝试在Scikit-learn的Pipeline中加入聚类步骤，将聚类结果作为额外特征。调用fit_transform()时出现参数不匹配错误，原transformer工作正常，问题出在聚类环节和特征融合过程。

代码 #

import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.base import TransformerMixin
from sklearn.cluster import KMeans
from sklearn.preprocessing import LabelBinarizer
from sklearn.pipeline import FeatureUnion

df = pd.DataFrame({'text': [
    'Here is an example of some text data',
    'here is another sentence of text',
    'yet another random string'],
    'label': [1, 1, 0]
})
X, y = df['text'], df['label']

class KMeans_foo(KMeans):
    def fit_transform(self, X, y = None):
        return self.fit_predict(X)

class ModelTransformer(TransformerMixin):

    def __init__(self, model):
        self.model = model

    def fit(self, *args, **kwargs):
        self.model.fit(*args, **kwargs)
        return self

    def transform(self, X, **transform_params):
        return pd.DataFrame(self.model.predict(X))

tf_idf_pipe = Pipeline(steps = [
    ('tf-idf', TfidfVectorizer(min_df = 0.10, stop_words = 'english'))
])

transformer = ColumnTransformer([
    ('text_pipe', tf_idf_pipe, 'text')],
    sparse_threshold = 0,
    remainder = 'drop'
)

transformer_cluster = Pipeline([
    ('orig', transformer),
    ('cluster', ModelTransformer(KMeans_foo(3))),
    ('binarize', LabelBinarizer())
])

feature_union = FeatureUnion([
    ('orig', transformer),
    ('cluster', transformer_cluster)
])

print('\nTransformed Training Data:')
print(pd.DataFrame(feature_union.fit_transform(pd.DataFrame(X))))
# print(pd.DataFrame(transformer.fit_transform(pd.DataFrame(X))))

# Transform test data
X_test = pd.DataFrame({'text': [
    'Here we have some test data',
    'and another one with text'
]})

print('\nTransformed Test Data:')
print(pd.DataFrame(feature_union.transform(X_test)))
# print(pd.DataFrame(transformer.transform(X_test)))

报错信息 #

Traceback (most recent call last):
  File "/Home/test.py", line 59, in <module>
    print(pd.DataFrame(feature_union.fit_transform(pd.DataFrame(X))))
  File "/Home/venv/lib/python3.9/site-packages/sklearn/pipeline.py", line 1154, in fit_transform
    results = self._parallel_func(X, y, fit_params, _fit_transform_one)
  File "/Home/venv/lib/python3.9/site-packages/sklearn/pipeline.py", line 1176, in _parallel_func
    return Parallel(n_jobs=self.n_jobs)(
  File "/Home/venv/lib/python3.9/site-packages/joblib/parallel.py", line 1046, in __call__
    while self.dispatch_one_batch(iterator):
  File "/Home/venv/lib/python3.9/site-packages/joblib/parallel.py", line 861, in dispatch_one_batch
    self._dispatch(tasks)
  File "/Home/venv/lib/python3.9/site-packages/joblib/parallel.py", line 779, in _dispatch
    job = self._backend.apply_async(batch, callback=cb)
  File "/Home/venv/lib/python3.9/site-packages/joblib/_parallel_backends.py", line 208, in apply_async
    result = ImmediateResult(func)
  File "/Home/venv/lib/python3.9/site-packages/joblib/_parallel_backends.py", line 572, in __init__
    self.results = batch()
  File "/Home/venv/lib/python3.9/site-packages/joblib/parallel.py", line 262, in __call__
    return [func(*args, **kwargs)
  File "/Home/venv/lib/python3.9/site-packages/joblib/parallel.py", line 262, in <listcomp>
    return [func(*args, **kwargs)
  File "/Home/venv/lib/python3.9/site-packages/sklearn/utils/fixes.py", line 117, in __call__
    return self.function(*args, **kwargs)
  File "/Home/venv/lib/python3.9/site-packages/sklearn/pipeline.py", line 870, in _fit_transform_one
    res = transformer.fit_transform(X, y, **fit_params)
  File "/Home/venv/lib/python3.9/site-packages/sklearn/pipeline.py", line 422, in fit_transform
    return last_step.fit_transform(Xt, y, **fit_params_last_step)
TypeError: fit_transform() takes 2 positional arguments but 3 were given

期望输出（不含聚类特征） #

Transformed Training Data:
          0         1         2         3         4         5
0  0.622766  0.622766  0.000000  0.000000  0.000000  0.473630
1  0.000000  0.000000  0.000000  0.795961  0.000000  0.605349
2  0.000000  0.000000  0.707107  0.000000  0.707107  0.000000

Transformed Test Data:
     0    1    2    3    4    5
0  1.0  0.0  0.0  0.0  0.0  0.0
1  0.0  0.0  0.0  0.0  0.0  1.0

问题分析与修复 #

核心问题 #

1. LabelBinarizer使用场景错误：LabelBinarizer是专门处理**目标变量（y）**的编码工具，它的fit_transform仅接受一个参数，但Pipeline会同时传递前一步输出的X和原始y参数，导致参数不匹配。
2. 自定义KMeans子类冗余：无需重写KMeans的fit_transform，直接用predict就能获取聚类标签。
3. 特征格式不兼容：ModelTransformer返回pd.DataFrame可能引发后续步骤格式冲突，应返回numpy数组。
4. 共享transformer实例风险：FeatureUnion两个分支复用同一个transformer实例，并行处理时会出现状态冲突。

修复代码 #

import pandas as pd
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.base import TransformerMixin
from sklearn.cluster import KMeans
from sklearn.preprocessing import OneHotEncoder
from sklearn.pipeline import FeatureUnion

df = pd.DataFrame({'text': [
    'Here is an example of some text data',
    'here is another sentence of text',
    'yet another random string'],
    'label': [1, 1, 0]
})
X, y = df['text'], df['label']

class ModelTransformer(TransformerMixin):
    def __init__(self, model):
        self.model = model

    def fit(self, X, y=None, **kwargs):
        self.model.fit(X, y=None, **kwargs)
        return self

    def transform(self, X, **transform_params):
        # 返回二维numpy数组，适配后续特征处理流程
        return np.array(self.model.predict(X)).reshape(-1, 1)

def create_text_transformer():
    # 封装为函数，每次调用生成新实例，避免共享状态冲突
    tf_idf_pipe = Pipeline(steps=[
        ('tf-idf', TfidfVectorizer(min_df=0.10, stop_words='english'))
    ])
    return ColumnTransformer([
        ('text_pipe', tf_idf_pipe, 'text')],
        sparse_threshold=0,
        remainder='drop'
    )

# 聚类分支：用OneHotEncoder处理聚类标签（特征处理工具，适配Pipeline流程）
transformer_cluster = Pipeline([
    ('orig', create_text_transformer()),
    ('cluster', ModelTransformer(KMeans(n_clusters=3, random_state=42))),
    ('binarize', OneHotEncoder(sparse_output=False, handle_unknown='ignore'))
])

# 特征融合：两个分支使用独立的transformer实例
feature_union = FeatureUnion([
    ('orig', create_text_transformer()),
    ('cluster', transformer_cluster)
])

print('\nTransformed Training Data:')
train_result = feature_union.fit_transform(pd.DataFrame(X))
print(pd.DataFrame(train_result))

# 测试数据转换
X_test = pd.DataFrame({'text': [
    'Here we have some test data',
    'and another one with text'
]})

print('\nTransformed Test Data:')
test_result = feature_union.transform(X_test)
print(pd.DataFrame(test_result))

修复说明 #

• 替换LabelBinarizer为OneHotEncoder：OneHotEncoder是专门用于特征编码的工具，接受X作为输入，完全适配Pipeline的特征处理流程。
• 移除冗余的KMeans_foo类：直接使用原生KMeans，通过predict方法获取聚类标签即可。
• 调整ModelTransformer返回格式：返回二维numpy数组，确保与后续特征处理步骤兼容。
• 新增create_text_transformer函数：每次调用生成独立的文本特征转换实例，避免FeatureUnion并行处理时的状态冲突。

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