Scikit-learn Pipeline v Pythonu: Od preprocessingu po ladění hyperparametrů

Proč jsou pipeline ve scikit-learn tak důležité

Ruku na srdce — pokud jste někdy psali kód pro strojové učení v Pythonu, tenhle scénář určitě znáte. Načtete data, provedete škálování, zakódujete kategorické proměnné, natrénujete model… a pak vám dojde, že jste omylem aplikovali škálování na celý dataset včetně testovacích dat. Gratuluju, právě jste si zavedli únik dat (data leakage) a vaše metriky jsou k ničemu.

Stává se to překvapivě často. I zkušeným datovým vědcům.

Scikit-learn Pipeline tento problém řeší elegantně. Spojí všechny kroky vašeho ML workflow — preprocessing, výběr příznaků, trénování modelu — do jednoho objektu. Transformace se automaticky aplikují jen na trénovací data a správně se propagují při křížové validaci. Žádný únik dat, žádné zapomenuté kroky. Prostě to funguje.

V tomto průvodci si projdeme vše od úplných základů po pokročilé techniky ladění hyperparametrů. Používáme scikit-learn 1.8 (prosinec 2025), aktuálně nejnovější stabilní verzi, která přinesla podporu Array API pro GPU výpočty a výrazné zrychlení L1-penalizovaných odhadů.

Instalace a příprava prostředí

Než se do toho pustíme, nainstalujte si aktuální verzi scikit-learn a pár dalších knihoven, které budeme potřebovat.

pip install scikit-learn pandas numpy matplotlib

# Ověření verze
import sklearn
print(sklearn.__version__)  # 1.8.x

Pro příklady v tomto článku budeme pracovat s datasetem wine zabudovaným přímo ve scikit-learn. Obsahuje výsledky chemické analýzy 178 vzorků vín ze tří různých odrůd a má 13 numerických příznaků — ideální pro demonstraci klasifikačních pipeline.

import pandas as pd
from sklearn.datasets import load_wine

# Načtení datasetu jako DataFrame
wine = load_wine(as_frame=True)
df = wine.frame

print(f"Rozměry: {df.shape}")
print(f"Třídy: {wine.target_names}")
print(df.head())

Váš první pipeline: krok za krokem

Pipeline se skládá z posloupnosti kroků. Každý krok má název a transformátor (nebo estimátor, pokud jde o poslední krok). Pojďme vytvořit jednoduchý pipeline, který škáluje data a pak použije logistickou regresi.

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

X, y = load_wine(return_X_y=True)

# Rozdělení na trénovací a testovací sadu
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.25, random_state=42, stratify=y
)

# Vytvoření pipeline
pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("classifier", LogisticRegression(max_iter=2000, random_state=42))
])

# Trénování — scaler se fitne jen na trénovací data
pipe.fit(X_train, y_train)

# Predikce — scaler automaticky transformuje testovací data
accuracy = pipe.score(X_test, y_test)
print(f"Přesnost: {accuracy:.4f}")

Všimněte si, co se tady děje pod kapotou: při volání pipe.fit() se StandardScaler naučí průměry a směrodatné odchylky pouze z trénovacích dat a pak je aplikuje na transformaci. Při pipe.score() se testovací data transformují pomocí parametrů naučených z trénovací sady. Žádný únik dat — a přesně tohle je důvod, proč pipeline existují.

Zkrácený zápis s make_pipeline

Pokud nepotřebujete pojmenovávat jednotlivé kroky (třeba při rychlém prototypování), funkce make_pipeline je vytvoří automaticky za vás.

from sklearn.pipeline import make_pipeline

# Automatické pojmenování kroků
pipe = make_pipeline(
    StandardScaler(),
    LogisticRegression(max_iter=2000, random_state=42)
)

# Názvy kroků se odvodí z názvů tříd (malými písmeny)
print(pipe.named_steps)
# {'standardscaler': StandardScaler(), 'logisticregression': LogisticRegression(...)}

ColumnTransformer: různé transformace pro různé sloupce

V reálných projektech máte skoro vždycky mix numerických a kategorických příznaků. Numerické sloupce chcete škálovat, kategorické zakódovat — a na každý typ aplikovat jinou transformaci. Přesně k tomu slouží ColumnTransformer.

Ukážeme si to na datovém souboru typu Titanic — klasický ML benchmark s kombinací numerických a kategorických dat. Je to sice trochu okoukaný příklad, ale pro demonstraci ColumnTransformeru funguje skvěle.

import pandas as pd
import numpy as np
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split

# Simulace datasetu typu Titanic
np.random.seed(42)
n = 500
data = {
    "vek": np.random.normal(30, 12, n),
    "jizne": np.random.exponential(50, n),
    "trida": np.random.choice(["1.", "2.", "3."], n, p=[0.2, 0.3, 0.5]),
    "pohlavi": np.random.choice(["muz", "zena"], n, p=[0.6, 0.4]),
    "nastupiste": np.random.choice(["C", "Q", "S", None], n, p=[0.2, 0.1, 0.6, 0.1]),
    "prezil": np.random.binomial(1, 0.38, n)
}
df = pd.DataFrame(data)
# Přidáme chybějící hodnoty do věku
df.loc[np.random.choice(n, 30, replace=False), "vek"] = np.nan

X = df.drop("prezil", axis=1)
y = df["prezil"]

# Definice sloupců podle typu
num_cols = ["vek", "jizne"]
cat_cols = ["trida", "pohlavi", "nastupiste"]

# Preprocessing pipeline pro numerické sloupce
num_transformer = Pipeline([
    ("imputer", SimpleImputer(strategy="median")),
    ("scaler", StandardScaler())
])

# Preprocessing pipeline pro kategorické sloupce
cat_transformer = Pipeline([
    ("imputer", SimpleImputer(strategy="most_frequent")),
    ("encoder", OneHotEncoder(handle_unknown="ignore", sparse_output=False))
])

# Spojení do ColumnTransformer
preprocessor = ColumnTransformer([
    ("num", num_transformer, num_cols),
    ("cat", cat_transformer, cat_cols)
])

# Kompletní pipeline: preprocessing + model
full_pipe = Pipeline([
    ("preprocessor", preprocessor),
    ("classifier", RandomForestClassifier(n_estimators=100, random_state=42))
])

# Rozdělení a trénování
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42
)

full_pipe.fit(X_train, y_train)
print(f"Přesnost: {full_pipe.score(X_test, y_test):.4f}")

Klíčová výhoda? Celý preprocessing je teď součástí jednoho objektu. Můžete ho serializovat, nasadit do produkce, nebo použít v křížové validaci — a máte jistotu, že se vždy aplikují stejné transformace ve správném pořadí.

Křížová validace s pipeline

Křížová validace (cross-validation) je standardní technika pro spolehlivější odhad výkonu modelu. Místo jediného rozdělení na trénovací a testovací sadu data opakovaně rozdělíte do několika foldů a model vyhodnotíte na každém z nich.

Bez pipeline byste museli ručně zajistit, že preprocessing se provede správně v každém foldu. To je otravné a náchylné k chybám. S pipeline to scikit-learn udělá za vás.

from sklearn.model_selection import cross_val_score, cross_validate

# 5-fold křížová validace
scores = cross_val_score(full_pipe, X, y, cv=5, scoring="accuracy")

print(f"Přesnosti v jednotlivých foldech: {scores}")
print(f"Průměrná přesnost: {scores.mean():.4f} (+/- {scores.std():.4f})")

# Více metrik najednou
results = cross_validate(
    full_pipe, X, y, cv=5,
    scoring=["accuracy", "precision_macro", "recall_macro", "f1_macro"],
    return_train_score=True
)

for metric in ["accuracy", "precision_macro", "recall_macro", "f1_macro"]:
    test_key = f"test_{metric}"
    train_key = f"train_{metric}"
    print(f"{metric}:")
    print(f"  Train: {results[train_key].mean():.4f}")
    print(f"  Test:  {results[test_key].mean():.4f}")

Tady je důležitý detail: srovnáním trénovacího a testovacího skóre odhalíte přeučení (overfitting). Pokud je trénovací přesnost výrazně vyšší než testovací, model se příliš přizpůsobil trénovacím datům a nebude dobře generalizovat.

Stratifikovaná křížová validace

Při klasifikačních úlohách s nevyváženými třídami je dobré použít stratifikovanou křížovou validaci. Ta zachovává poměr tříd v každém foldu, takže výsledky jsou spolehlivější.

from sklearn.model_selection import StratifiedKFold

skf = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)

scores = cross_val_score(full_pipe, X, y, cv=skf, scoring="accuracy")
print(f"Stratifikovaná CV: {scores.mean():.4f} (+/- {scores.std():.4f})")

GridSearchCV: systematické ladění hyperparametrů

Tak, teď se dostáváme k tomu nejzajímavějšímu — ladění hyperparametrů. Hyperparametry jsou parametry modelu, které se nenastavují během trénování, musíte je zvolit předem. Počet stromů v Random Forest, regularizační konstanta C v logistické regresi, hloubka rozhodovacího stromu… Správná volba hyperparametrů může dramaticky ovlivnit výkon modelu.

GridSearchCV systematicky prohledá všechny kombinace zadaných hodnot a pro každou kombinaci provede křížovou validaci. Výsledkem je nejlepší nalezená kombinace. Jednoduché, ale efektivní.

Pojmenování parametrů v pipeline

Pro odkazování na parametry uvnitř pipeline se používá konvence s dvojitým podtržítkem: nazev_kroku__parametr. Pro vnořené pipeline (třeba pipeline uvnitř ColumnTransformeru) se řetězí: preprocessor__num__imputer__strategy. Zpočátku to vypadá trochu krkolomně, ale zvyknete si rychle.

from sklearn.model_selection import GridSearchCV

# Definice parametrové mřížky
param_grid = {
    # Parametry preprocessingu
    "preprocessor__num__imputer__strategy": ["mean", "median"],
    # Parametry modelu
    "classifier__n_estimators": [50, 100, 200],
    "classifier__max_depth": [5, 10, None],
    "classifier__min_samples_split": [2, 5, 10]
}

# Celkový počet kombinací: 2 * 3 * 3 * 3 = 54
# S 5-fold CV: 54 * 5 = 270 fitů modelu

grid_search = GridSearchCV(
    full_pipe,
    param_grid,
    cv=5,
    scoring="accuracy",
    n_jobs=-1,       # využít všechna jádra procesoru
    verbose=1,
    return_train_score=True
)

grid_search.fit(X_train, y_train)

print(f"Nejlepší parametry: {grid_search.best_params_}")
print(f"Nejlepší CV skóre: {grid_search.best_score_:.4f}")
print(f"Skóre na testovací sadě: {grid_search.score(X_test, y_test):.4f}")

Analýza výsledků GridSearchCV

Výsledky hledání jsou uloženy v atributu cv_results_. Pojďme se na ně podívat podrobněji — tady se často najdou zajímavé postřehy o tom, které parametry mají na výkon největší vliv.

import pandas as pd

# Převod výsledků na DataFrame
results_df = pd.DataFrame(grid_search.cv_results_)

# Top 5 kombinací podle testovacího skóre
top5 = results_df.nsmallest(5, "rank_test_score")[
    ["params", "mean_test_score", "std_test_score", "mean_train_score", "rank_test_score"]
]
print(top5.to_string())

# Detekce přeučení — velký rozdíl mezi train a test skóre
results_df["overfit_gap"] = (
    results_df["mean_train_score"] - results_df["mean_test_score"]
)
print(f"\nPrůměrné přeučení: {results_df['overfit_gap'].mean():.4f}")
print(f"Maximální přeučení: {results_df['overfit_gap'].max():.4f}")

RandomizedSearchCV: rychlejší alternativa pro velké prostory

GridSearchCV prohledá všechny kombinace — to je skvělé pro malé prostory parametrů, ale co když máte 4 parametry s 10 hodnotami každý? To je 10 000 kombinací. S 5-fold CV je to 50 000 fitů modelu. Klidně to může trvat hodiny.

RandomizedSearchCV tohle řeší elegantně. Místo prohledávání všech kombinací náhodně vzorkuje z prostoru parametrů. Nastavíte počet iterací (n_iter) a algoritmus vybere daný počet náhodných kombinací. Z mé zkušenosti RandomizedSearchCV skoro vždy najde srovnatelně dobré řešení za zlomek času — a to i na relativně malých prostorech.

from sklearn.model_selection import RandomizedSearchCV
from scipy.stats import randint, uniform

# Definice distribucí — ne seznamů hodnot, ale statistických rozdělení
param_distributions = {
    "classifier__n_estimators": randint(50, 500),
    "classifier__max_depth": [3, 5, 10, 15, None],
    "classifier__min_samples_split": randint(2, 20),
    "classifier__min_samples_leaf": randint(1, 10),
    "classifier__max_features": ["sqrt", "log2", None],
    "preprocessor__num__imputer__strategy": ["mean", "median"]
}

random_search = RandomizedSearchCV(
    full_pipe,
    param_distributions,
    n_iter=50,          # počet náhodných kombinací k vyzkoušení
    cv=5,
    scoring="accuracy",
    n_jobs=-1,
    random_state=42,
    verbose=1
)

random_search.fit(X_train, y_train)

print(f"Nejlepší parametry: {random_search.best_params_}")
print(f"Nejlepší CV skóre: {random_search.best_score_:.4f}")
print(f"Skóre na testovací sadě: {random_search.score(X_test, y_test):.4f}")

Kdy použít GridSearchCV vs RandomizedSearchCV

• GridSearchCV — malý prostor parametrů (desítky kombinací), potřebujete prohledat opravdu všechny možnosti, máte dostatek výpočetního času.
• RandomizedSearchCV — velký prostor parametrů (stovky až tisíce kombinací), chcete rychlé výsledky, máte spojité parametry (kde stejně nemá smysl testovat každou hodnotu).

V praxi se často začíná s RandomizedSearchCV pro hrubé prohledání a pak se výsledky zpřesní pomocí GridSearchCV na zúženém prostoru kolem nejlepších nalezených hodnot. Tento dvoustupňový přístup je podle mě nejefektivnější strategie.

HalvingSearchCV: nejrychlejší hledání hyperparametrů

Scikit-learn 1.8 obsahuje experimentální třídy HalvingGridSearchCV a HalvingRandomSearchCV, které implementují metodu successive halving (postupné půlení). Princip je jednoduchý a upřímně docela geniální: začněte s mnoha kandidáty a málo zdrojů (malý vzorek dat), v každé iteraci odstraňte nejhorší kandidáty a zbývajícím přidělte víc zdrojů.

from sklearn.experimental import enable_halving_search_cv  # nutný import
from sklearn.model_selection import HalvingRandomSearchCV
from scipy.stats import randint

param_distributions = {
    "classifier__n_estimators": randint(50, 500),
    "classifier__max_depth": [3, 5, 10, None],
    "classifier__min_samples_split": randint(2, 20),
}

halving_search = HalvingRandomSearchCV(
    full_pipe,
    param_distributions,
    factor=3,          # v každé iteraci zůstane 1/3 kandidátů
    cv=5,
    scoring="accuracy",
    random_state=42,
    n_jobs=-1
)

halving_search.fit(X_train, y_train)

print(f"Nejlepší parametry: {halving_search.best_params_}")
print(f"Nejlepší CV skóre: {halving_search.best_score_:.4f}")
print(f"Počet iterací: {halving_search.n_iterations_}")
print(f"Počet kandidátů v poslední iteraci: {halving_search.n_candidates_[-1]}")

Hlavní výhoda: u velkých datasetů a rozsáhlých prostorů parametrů může být HalvingSearchCV řádově rychlejší než GridSearchCV nebo RandomizedSearchCV.

Nevýhoda? Protože na začátku pracuje s malým vzorkem dat, může přehlédnout kandidáty, kteří fungují dobře až s více daty. Pro většinu praktických případů ale výsledky odpovídají klasickým metodám.

Důležité upozornění: tyto třídy jsou stále označené jako experimentální. Jejich API se může v budoucích verzích změnit, proto musíte explicitně importovat enable_halving_search_cv. Mějte to na paměti, pokud je chcete používat v produkčním kódu.

Praktický příklad: kompletní ML workflow od začátku do konce

Tak, pojďme spojit všechno dohromady. Vytvoříme kompletní pipeline pro klasifikaci vín — od načtení dat přes preprocessing, výběr modelu a ladění hyperparametrů až po finální vyhodnocení. Tohle je v podstatě šablona, kterou (s drobnými úpravami) můžete použít na většinu klasifikačních úloh.

import numpy as np
import pandas as pd
from sklearn.datasets import load_wine
from sklearn.model_selection import train_test_split, RandomizedSearchCV, cross_val_score
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.svm import SVC
from sklearn.metrics import classification_report, ConfusionMatrixDisplay
from scipy.stats import randint, uniform
import matplotlib.pyplot as plt

# 1. Načtení dat
X, y = load_wine(return_X_y=True)
target_names = load_wine().target_names

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

print(f"Trénovací sada: {X_train.shape[0]} vzorků")
print(f"Testovací sada: {X_test.shape[0]} vzorků")

# 2. Pipeline s preprocessingem a PCA
pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("pca", PCA()),
    ("classifier", SVC())
])

# 3. Prohledávání různých modelů i preprocessingu
param_distributions = [
    {
        "pca__n_components": randint(3, 13),
        "classifier": [SVC()],
        "classifier__C": uniform(0.1, 100),
        "classifier__kernel": ["rbf", "linear"],
    },
    {
        "pca__n_components": randint(3, 13),
        "classifier": [RandomForestClassifier(random_state=42)],
        "classifier__n_estimators": randint(50, 300),
        "classifier__max_depth": [5, 10, None],
    },
    {
        "pca__n_components": randint(3, 13),
        "classifier": [GradientBoostingClassifier(random_state=42)],
        "classifier__n_estimators": randint(50, 300),
        "classifier__learning_rate": uniform(0.01, 0.3),
        "classifier__max_depth": [3, 5, 7],
    }
]

# 4. RandomizedSearchCV přes všechny modely
search = RandomizedSearchCV(
    pipe,
    param_distributions,
    n_iter=60,
    cv=5,
    scoring="accuracy",
    n_jobs=-1,
    random_state=42
)

search.fit(X_train, y_train)

# 5. Výsledky
print(f"\nNejlepší model: {search.best_params_['classifier'].__class__.__name__}")
print(f"Nejlepší parametry: {search.best_params_}")
print(f"Nejlepší CV skóre: {search.best_score_:.4f}")

# 6. Vyhodnocení na testovací sadě
y_pred = search.predict(X_test)
print(f"\nSkóre na testovací sadě: {search.score(X_test, y_test):.4f}")
print("\nKlasifikační zpráva:")
print(classification_report(y_test, y_pred, target_names=target_names))

Tento příklad demonstruje jednu z nejsilnějších vlastností scikit-learn pipeline — možnost zaměnit celý krok pipeline (včetně klasifikátoru) jako hyperparametr. Místo ručního zkoušení různých modelů necháte RandomizedSearchCV, aby systematicky porovnal SVM, Random Forest a Gradient Boosting s různými nastaveními preprocessingu. Docela mocný nástroj, když se nad tím zamyslíte.

Tipy a osvědčené postupy pro produkční nasazení

Serializace pipeline

Natrénovaný pipeline můžete uložit a později znovu načíst pomocí knihovny joblib. Tenhle krok je zásadní — v produkci nechcete model pokaždé trénovat znovu.

import joblib

# Uložení
joblib.dump(search.best_estimator_, "best_model_pipeline.joblib")

# Načtení
loaded_pipe = joblib.load("best_model_pipeline.joblib")
prediction = loaded_pipe.predict(X_test[:5])
print(f"Predikce: {prediction}")

Vlastní transformátor v pipeline

Někdy potřebujete transformaci, kterou scikit-learn standardně nenabízí. V takovém případě si můžete vytvořit vlastní transformátor pomocí FunctionTransformer nebo děděním z BaseEstimator a TransformerMixin.

from sklearn.preprocessing import FunctionTransformer
import numpy as np

# Jednoduchý vlastní transformátor — logaritmická transformace
log_transformer = FunctionTransformer(
    func=np.log1p,
    inverse_func=np.expm1
)

pipe_with_custom = Pipeline([
    ("log_transform", log_transformer),
    ("scaler", StandardScaler()),
    ("classifier", LogisticRegression(max_iter=2000))
])

Cachování mezikroků

U náročného preprocessingu (třeba TF-IDF na velkém textovém korpusu) můžete zapnout cachování, aby se transformace neopakovaly při každém fitu. Ušetří to spoustu času.

from tempfile import mkdtemp

cachedir = mkdtemp()
cached_pipe = Pipeline(
    [("scaler", StandardScaler()), ("svm", SVC())],
    memory=cachedir
)

Kontrolní seznam pro produkci

• Vždy používejte pipeline — i pro jednoduchou standardizaci. Vyhnete se úniku dat a ušetříte si budoucí bolesti hlavy.
• Nastavte random_state u všech stochastických komponent pro reprodukovatelnost.
• Používejte n_jobs=-1 pro paralelizaci, ale pozor na spotřebu paměti u velkých datasetů.
• Ukládejte celý pipeline, ne jen model — potřebujete i preprocessing kroky.
• Pro ladění hyperparametrů začněte s RandomizedSearchCV a pak zpřesněte GridSearchCV.
• Verzujte si natrénované modely — joblib soubory patří do verzovacího systému nebo model registry (ne do Gitu přímo, ale třeba do DVC nebo MLflow).

Často kladené otázky

Jaký je rozdíl mezi Pipeline a make_pipeline ve scikit-learn?

Pipeline vyžaduje, abyste každému kroku přiřadili explicitní název — například ("scaler", StandardScaler()). Funkce make_pipeline názvy generuje automaticky z názvů tříd (malými písmeny). Používejte Pipeline když potřebujete vlastní názvy kroků (zejména při ladění hyperparametrů s GridSearchCV), a make_pipeline pro rychlé prototypování.

Jak zabránit úniku dat (data leakage) při preprocessingu?

Nejspolehlivější ochrana je zařadit veškerý preprocessing do pipeline a používat křížovou validaci přes celý pipeline — ne jen přes model. Pokud provedete škálování nebo imputaci chybějících hodnot na celém datasetu před rozdělením na trénovací a testovací sadu, testovací data ovlivní parametry preprocessingu a vaše metriky budou nadhodnocené. Pipeline tomuto problému zabrání automaticky.

Jak zjistím, které hyperparametry mohu ladit uvnitř pipeline?

Zavolejte pipeline.get_params() — vrátí slovník všech parametrů včetně vnořených. Pro GridSearchCV nebo RandomizedSearchCV pak použijte konvenci nazev_kroku__parametr. U vnořených pipeline (ColumnTransformer s pipeline uvnitř) se řetězí: preprocessor__num__scaler__with_mean.

Je lepší GridSearchCV nebo RandomizedSearchCV?

Záleží na velikosti prostoru parametrů. GridSearchCV prohledá všechny kombinace — je vhodný pro malé prostory (desítky kombinací). RandomizedSearchCV náhodně vzorkuje a je výrazně rychlejší pro velké prostory. Nová experimentální třída HalvingRandomSearchCV ve scikit-learn 1.8 nabízí ještě rychlejší alternativu metodou postupného půlení, ale její API se může v budoucnu změnit.

Mohu v pipeline porovnávat různé modely najednou?

Ano, a je to překvapivě jednoduché. V parametrové mřížce zadáte seznam různých estimátorů jako hodnoty pro krok classifier. RandomizedSearchCV nebo GridSearchCV pak vyzkouší každý model s odpovídajícími hyperparametry. Stačí definovat parametrovou mřížku jako seznam slovníků — každý slovník pro jiný model s jeho specifickými parametry.