Pipelines

Uma cadeia real de pré-processamento — imputar, codificar, escalonar e então modelar — é uma sequência de passos que deve ser ajustada apenas nos dados de treino e reproduzida de forma idêntica nos folds de validação, no conjunto de teste e no tráfego de produção. Fazer isso à mão convida a bugs. O Pipeline do scikit-learn empacota toda a cadeia como um único estimador.

O problema que os pipelines resolvem

Sem um pipeline, o fluxo honesto exige uma contabilidade cuidadosa:

# Frágil: cada passo precisa ser ajustado à mão no treino e aplicado ao teste
imputer.fit(X_train)
X_train_i = imputer.transform(X_train)
X_test_i = imputer.transform(X_test)

scaler.fit(X_train_i)
X_train_s = scaler.transform(X_train_i)
X_test_s = scaler.transform(X_test_i)

model.fit(X_train_s, y_train)
model.predict(X_test_s)

Um fit fora do lugar — ou um fit_transform no conjunto completo — e você tem vazamento de dados. Pior: durante a validação cruzada o pré-processamento precisa ser reajustado em cada fold de treino, o que é praticamente impossível de fazer corretamente à mão.

Pipeline: um estimador, vários passos

from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression

pipe = Pipeline([
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler()),
    ('model', LogisticRegression()),
])

pipe.fit(X_train, y_train)      # ajusta imputer → scaler → model, em ordem, no treino
pipe.predict(X_test)            # transforma X_test pelos MESMOS passos ajustados

Regras da composição:

• todo passo, exceto o último, deve ser um transformador (fit + transform);
• o último passo é tipicamente um estimador (fit + predict);
• pipe.fit(X, y) chama fit_transform em cada transformador em sequência e depois fit no estimador;
• pipe.predict(X) chama apenas transform em cada transformador — os parâmetros ficam congelados.

Como o pipeline é um estimador, ele se encaixa diretamente em cross_val_score e GridSearchCV — e o pré-processamento é automaticamente reajustado dentro de cada fold, eliminando o bug de vazamento por construção:

from sklearn.model_selection import cross_val_score
scores = cross_val_score(pipe, X, y, cv=5)   # livre de vazamento por design

Colunas heterogêneas: ColumnTransformer

Tabelas reais misturam colunas numéricas e categóricas que precisam de tratamentos diferentes. O ColumnTransformer roteia subconjuntos de colunas para ramos de pré-processamento paralelos e concatena os resultados:

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder

numeric = ['age', 'income', 'tenure']
categorical = ['city', 'plan', 'device']

preprocess = ColumnTransformer([
    ('num', Pipeline([
        ('imputer', SimpleImputer(strategy='median')),
        ('scaler', StandardScaler()),
    ]), numeric),
    ('cat', Pipeline([
        ('imputer', SimpleImputer(strategy='most_frequent')),
        ('onehot', OneHotEncoder(handle_unknown='ignore')),
    ]), categorical),
])

pipe = Pipeline([
    ('preprocess', preprocess),
    ('model', LogisticRegression(max_iter=1000)),
])

flowchart LR
    X[DataFrame bruto] --> CT{ColumnTransformer}
    CT -->|colunas numéricas| N[imputar mediana → escalonar]
    CT -->|colunas categóricas| C[imputar moda → one-hot]
    N --> J[concatenar]
    C --> J
    J --> M[LogisticRegression]

Ajuste através do pipeline

Os hiperparâmetros de qualquer passo são endereçados como nomedopasso__nomedoparam (duplo sublinhado) — assim, uma única busca em grade pode ajustar escolhas de pré-processamento e hiperparâmetros do modelo juntos, de forma honesta:

from sklearn.model_selection import GridSearchCV

param_grid = {
    'preprocess__num__imputer__strategy': ['mean', 'median'],
    'model__C': [0.01, 0.1, 1, 10],
}
search = GridSearchCV(pipe, param_grid, cv=5, scoring='f1')
search.fit(X_train, y_train)

Mais sobre isso em Seleção de Modelos.

Inspeção e persistência

pipe.named_steps['model'].coef_            # acessar qualquer passo ajustado
pipe[:-1].transform(X_train)               # rodar apenas o pré-processamento
pipe.get_feature_names_out()               # nomes após a expansão one-hot

import joblib
joblib.dump(pipe, 'churn-model.joblib')    # entregar UM artefato: pré-processamento + modelo

Persistir o pipeline inteiro é a base de um MLOps confiável: o código de produção não pode "esquecer" um passo de pré-processamento, porque os passos viajam dentro do artefato.

Material de aula

Quiz