Transformadores personalizados Sklearn con tubería: todas las dimensio

Transformadores personalizados Sklearn con tubería: todas las dimensiones de la matriz de entrada para el eje de concatenación deben coincidir exactamente

Estoy aprendiendo sobre sklearn transformadores personalizados y leí sobre las dos formas principales de crear transformadores personalizados:

  1. configurando una clase personalizada que hereda de BaseEstimator y TransformerMixin, o
  2. creando un método de transformación y pasándolo a FunctionTransformer.

Quería comparar estos dos enfoques implementando una funcionalidad de "meta-vectorizador": un vectorizador que admite CountVectorizer o TfidfVectorizer y transforma los datos de entrada de acuerdo con el vectorizador especificado tipo.

Sin embargo, parece que no puedo hacer que ninguno de los dos funcione cuando los paso a un sklearn.pipeline.Pipeline. Recibo el siguiente mensaje de error en el paso fit_transform():

ValueError: all the input array dimensions for the concatenation axis must match 
exactly, but along dimension 0, the array at index 0 has size 6 and the array 
at index 1 has size 1

Mi código para la opción 1 (usando una clase personalizada):

class Vectorizer(BaseEstimator, TransformerMixin):
    def __init__(self, vectorizer:Callable=CountVectorizer(), ngram_range:tuple=(1,1)) -> None:
        super().__init__()
        self.vectorizer = vectorizer
        self.ngram_range = ngram_range
    def fit(self, X, y=None):
        return self 
    def transform(self, X, y=None):
        X_vect_ = self.vectorizer.fit_transform(X.copy())
        return X_vect_.toarray()

pipe = Pipeline([
    ('column_transformer', ColumnTransformer([
        ('lesson_type_category', OneHotEncoder(), ['Type']),
        ('comment_text_vectorizer', Vectorizer(), ['Text'])],
        remainder='drop')),
    ('model', LogisticRegression())])

param_dict = {'column_transformer__comment_text_vectorizer__vectorizer': \
[CountVectorizer(), TfidfVectorizer()]
}

randsearch = GridSearchCV(pipe, param_dict, cv=2, scoring='f1',).fit(X_train, y_train)

Y mi código para la opción 2 (crear un transformador personalizado a partir de una función usando FunctionTransformer):

def vectorize_text(X, vectorizer: Callable):
    X_vect_ = vectorizer.fit_transform(X)
    return X_vect_.toarray()

vectorizer_transformer = FunctionTransformer(vectorize_text, kw_args={'vectorizer': TfidfVectorizer()})

pipe = Pipeline([
    ('column_transformer', ColumnTransformer([
        ('lesson_type_category', OneHotEncoder(), ['Type']),
        ('comment_text_vectorizer', vectorizer_transformer, ['Text'])],
        remainder='drop')),
    ('model', LogisticRegression())])

param_dict = {'column_transformer__comment_text_vectorizer__kw_args': \
    [{'vectorizer':CountVectorizer()}, {'vectorizer': TfidfVectorizer()}]
}

randsearch = GridSearchCV(pipe, param_dict, cv=2, scoring='f1').fit(X_train, y_train)

Importaciones y datos de muestra:

import pandas as pd 
from typing import Callable
import sklearn
from sklearn.preprocessing import OneHotEncoder, FunctionTransformer
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LogisticRegression
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.feature_extraction.text import CountVectorizer, TfidfVectorizer
from sklearn.model_selection import GridSearchCV

df = pd.DataFrame([
    ['A99', 'hi i love python very much', 'c', 1],
    ['B07', 'which programming language should i learn', 'b', 0],
    ['A12', 'what is the difference between python django flask', 'b', 1],
    ['A21', 'i want to be a programmer one day', 'c', 0],
    ['B11', 'should i learn java or python', 'b', 1],
    ['C01', 'how much can i earn as a programmer with python', 'a', 0]
], columns=['Src', 'Text', 'Type', 'Target'])

Notas:

  • Como se recomienda en esta pregunta , transformé todas las matrices dispersas en arreglos densos después de la vectorización, como puede ver en ambos casos: X_vect_.toarray().
Mostrar la mejor respuesta

El problema es que tanto CountVectorizer como TfidfVectorizer requieren que su entrada sea 1D (y no 2D). En tales casos, el doc de ColumnTransformer establece que el parámetro columns de la tupla transformers debe pasarse como una cadena en lugar de como una lista.

columnas: str, tipo matriz de str, int, tipo matriz de int, tipo matriz de bool, rebanada o invocable

Indexa los datos en su segundo eje. Los enteros se interpretan como columnas posicionales, mientras que las cadenas pueden hacer referencia a las columnas DataFrame por su nombre. Se debe usar una cadena escalar o int donde el transformador espera que X sea una matriz 1d (vector), de lo contrario, se pasará una matriz 2d al transformador. Un invocable se pasa los datos de entrada X y puede devolver cualquiera de los anteriores. Para seleccionar varias columnas por nombre o tipo, puede usar make_column_selector.

Por lo tanto, lo siguiente funcionará en su caso (es decir, cambiar ['Text'] por 'Text').

class Vectorizer(BaseEstimator, TransformerMixin):
    def __init__(self, vectorizer:Callable=CountVectorizer(), ngram_range:tuple=(1,1)) -> None:
        super().__init__()
        self.vectorizer = vectorizer
        self.ngram_range = ngram_range
    def fit(self, X, y=None):
        return self 
    def transform(self, X, y=None):
        X_vect_ = self.vectorizer.fit_transform(X.copy())
        return X_vect_.toarray()

pipe = Pipeline([
    ('column_transformer', ColumnTransformer([
        ('lesson_type_category', OneHotEncoder(handle_unknown='ignore'), ['Type']),
        ('comment_text_vectorizer', Vectorizer(), 'Text')], remainder='drop')),
    ('model', LogisticRegression())])

param_dict = {'column_transformer__comment_text_vectorizer__vectorizer': [CountVectorizer(), TfidfVectorizer()]
}

randsearch = GridSearchCV(pipe, param_dict, cv=2, scoring='f1',).fit(X_train, y_train)

Puede ajustar el ejemplo con FunctionTransformer según corresponda. Observe, como comentario final, que tuve que pasar handle_unknown='ignore' a OneHotEncoder para evitar la posibilidad de que hubiera surgido un error en caso de categorías desconocidas vistas durante la fase de prueba de su cruce -validación (y no visto durante la fase de entrenamiento).