4. Pipelines with Workflow

Chapter 3 ended with a hand-written fold loop containing the line that keeps you honest — fit_transform on train, transform on test. That loop is where leakage bugs live, and rewriting it in every notebook is how they spread.

Workflow is the fix. It packages preprocessing and a model into one object that behaves like a model, and does the fit-on-train-only bookkeeping internally, every time, without being asked.

You will learn:

• how to build a pipeline as an ordered list of steps
• proof that Workflow gets the fold discipline right
• how to inspect a fitted pipeline, step by step
• how to apply a transform to some columns with On
• how to export a pipeline as configuration, and save it to disk

Prerequisites: chapters 2 and 3.

import numpy as np

from tuiml.datasets import load_diabetes
from tuiml.workflow import Workflow, On
from tuiml.preprocessing import SimpleImputer, StandardScaler
from tuiml.algorithms.trees import RandomForestClassifier
from tuiml.algorithms.neighbors import KNearestNeighborsClassifier
from tuiml.evaluation import StratifiedKFold, cross_val_score, accuracy_score

data = load_diabetes()
y = data.y

# The chapter 3 cleanup: impossible zeros become NaN.
IMPOSSIBLE_ZERO = ["plas", "pres", "skin", "insu", "mass"]
columns = [data.feature_names.index(c) for c in IMPOSSIBLE_ZERO]

X = data.X.copy()
for col in columns:
    X[X[:, col] == 0, col] = np.nan

print(f"{int(np.isnan(X).sum())} missing values")

652 missing values

4.1 A pipeline is a list

Workflow takes an ordered list. Transformers run in sequence, and the last entry is the model.

flow = Workflow([
    SimpleImputer(strategy="median"),
    StandardScaler(),
    KNearestNeighborsClassifier(k=5),
])

flow

Workflow([
    SimpleImputer(strategy='median'),
    StandardScaler(with_mean=True, with_std=True),
    KNearestNeighborsClassifier(k=5, distance_weighting='uniform'),
])

That is the whole notation. There is no separate "add a step" call and no name-to-object mapping to maintain — the list order is the pipeline order.

It behaves like a model, so everything from chapter 2 applies unchanged:

from tuiml.evaluation import train_test_split

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

flow.fit(X_train, y_train)

print("predictions:", flow.predict(X_test[:8]))
print("accuracy   :", round(accuracy_score(y_test, flow.predict(X_test)), 4))

predictions: [1 0 0 0 0 0 0 1]
accuracy   : 0.8026

When you call fit, each transformer is fitted on the training data and passes its output to the next. When you call predict, each fitted transformer transforms — it does not refit. That is the entire discipline of chapter 3, enforced by construction.

4.2 Proof that it gets the folds right

Claims about leakage deserve evidence rather than assurance. Here is the hand-written loop from chapter 3, and the same thing via Workflow, on identical folds:

splitter = StratifiedKFold(n_splits=10, shuffle=True, random_state=42)

# The careful version, written out by hand.
manual = []
for train_idx, test_idx in splitter.split(X, y):
    imputer = SimpleImputer(strategy="median")
    X_tr = imputer.fit_transform(X[train_idx].copy())
    X_te = imputer.transform(X[test_idx].copy())

    model = RandomForestClassifier(n_estimators=200, random_state=42)
    model.fit(X_tr, y[train_idx])
    manual.append(accuracy_score(y[test_idx], model.predict(X_te)))

manual_score = np.mean(manual)

# The same thing, as a Workflow.
pipeline = Workflow([
    SimpleImputer(strategy="median"),
    RandomForestClassifier(n_estimators=200, random_state=42),
])
workflow_score = cross_val_score(pipeline, X, y, cv=splitter).mean()

print(f"hand-written fold loop : {manual_score:.4f}")
print(f"cross_val_score(flow)  : {workflow_score:.4f}")
print(f"identical              : {np.isclose(manual_score, workflow_score)}")

hand-written fold loop : 0.7695
cross_val_score(flow)  : 0.7695
identical              : True

Identical to the last digit. The pipeline is being refitted from scratch inside every fold, exactly as the careful version does — and you did not have to write the loop, so you cannot get it wrong.

Remark. This is the real argument for pipelines, and it is not about tidiness. A pipeline makes the correct thing automatic and the incorrect thing difficult to express. You can still leak by transforming your data before handing it over — nothing stops you calling fit_transform on the whole array first — but you have to go out of your way, rather than merely forget.

4.3 Looking inside

A fitted pipeline is inspectable. This matters when a result surprises you and you need to know which step is responsible.

fitted = Workflow([
    SimpleImputer(strategy="median"),
    StandardScaler(),
    RandomForestClassifier(n_estimators=100, random_state=42),
])
fitted.fit(X_train, y_train)

print("steps by name:")
for name, step in fitted.named_steps.items():
    print(f"  {name:26s} {step}")

print()
print("transformers only:", [name for name, _ in fitted.transformers])
print("model            :", fitted.model)

steps by name:
  simpleimputer              SimpleImputer(strategy='median')
  standardscaler             StandardScaler(with_mean=True, with_std=True)
  randomforestclassifier     RandomForestClassifier(n_estimators=100)

transformers only: ['simpleimputer', 'standardscaler']
model            : RandomForestClassifier(n_estimators=100)

There are two views of the steps, and the difference matters.

steps, named_steps, transformers and model are the templates — the unfitted objects you handed to the constructor. steps_ and model_, with the trailing underscore, are the fitted copies produced by fit. This follows the usual convention that a trailing underscore means "learned from data", and it is why a Workflow can be refitted on new data without you rebuilding it.

Reach for the underscore versions when you want to inspect what was actually learned:

imputer = fitted.steps_[0]          # fitted, not the template

sample = X_train[:3].copy()
print("three rows, before imputation:")
print(sample.round(1))
print("\nthe same rows after this one step:")
print(imputer.transform(sample).round(1))

three rows, before imputation:
[[  6.  102.   82.    nan   nan  30.8   0.2  36. ]
 [  4.  122.   68.    nan   nan  35.    0.4  29. ]
 [ 10.  129.   62.   36.    nan  41.2   0.4  38. ]]

the same rows after this one step:
[[  6.  102.   82.   29.  129.5  30.8   0.2  36. ]
 [  4.  122.   68.   29.  129.5  35.    0.4  29. ]
 [ 10.  129.   62.   36.  129.5  41.2   0.4  38. ]]

Pushing a few rows through one step at a time is the most reliable way to localise a surprising result: keep going until the numbers stop looking like what you expected, and the step you just ran is the culprit.

Remark — the templates are not fitted. fitted.named_steps["simpleimputer"].transform(...) raises RuntimeError: SimpleImputer must be fitted, which is confusing the first time you hit it on an obviously fitted pipeline. named_steps is for reading the configuration; steps_ is for using the fitted step.

Remark — every step in steps_ was fitted on X_train alone. Nothing in this pipeline has seen X_test. In chapter 3 that was a property of your discipline; here it is a property of the object.

4.4 Applying a step to some columns: On

So far every transformer has hit every column. That is often wrong.

Recall the distinction chapter 1 made: preg has real zeros, the other five do not. Suppose we want to impute only the five medical columns and leave preg and pedi and age untouched. On wraps a transformer and restricts it to a column subset.

selective = Workflow([
    On(columns=columns, transformer=SimpleImputer(strategy="median")),
    RandomForestClassifier(n_estimators=200, random_state=42),
])

selective

Workflow([
    On([1, 2, 3, 4, 5], SimpleImputer(strategy='median')),
    RandomForestClassifier(n_estimators=200),
])

score_selective = cross_val_score(selective, X, y, cv=splitter).mean()
print(f"impute only the 5 medical columns: {score_selective:.4f}")
print(f"impute everything                : {workflow_score:.4f}")

impute only the 5 medical columns: 0.7591
impute everything                : 0.7695

Those should be the same number. The three columns we excluded contain no NaNs, so the imputer had nothing to do there — and yet the scores differ by a full point.

The reason is worth knowing, because it will bite you eventually.

one_step = On(columns=columns, transformer=SimpleImputer(strategy="median"))
reordered = one_step.fit_transform(X.copy())

print("original column order:", data.feature_names)
print()
print("row 0 in,  original order:", X[0].round(1))
print("row 0 out, after On      :", reordered[0].round(1))

original column order: ['preg', 'plas', 'pres', 'skin', 'insu', 'mass', 'pedi', 'age']

row 0 in,  original order: [  6.  148.   72.   35.    nan  33.6   0.6  50. ]
row 0 out, after On      : [148.   72.   35.  125.   33.6   6.    0.6  50. ]

On does not put the columns back where it found them. The columns it transformed come out first, in the order you listed them, and the passed-through remainder is appended after. preg went in at position 0 and came out at position 5.

The values are all still there and none of them changed, so the information content is identical — but a random forest samples features by position, so a different column order means different trees and a slightly different score. That is the whole of the one-point gap: not better or worse modelling, just a different arbitrary tie-break.

This matches how scikit-learn's ColumnTransformer behaves, so it is a convention rather than a defect. It has one sharp edge:

Warning — column indices shift after every On. If you chain two On steps and address the second one using the original column numbers, it will silently transform the wrong columns. Indices in the second step refer to the output order of the first, not to your source data.

import numpy as np

demo = np.arange(24, dtype=float).reshape(4, 6)
print("original          :", demo[0])

after_first = On(columns=[0, 1], transformer=StandardScaler()).fit_transform(demo.copy())
print("after On([0, 1])  :", after_first[0].round(2))

after_second = On(columns=[4, 5], transformer=StandardScaler()).fit_transform(after_first.copy())
print("then On([4, 5])   :", after_second[0].round(2))
print()
print("The second step meant original columns 4 and 5. It got positions 4 and 5")
print("of the *reordered* array, which are original columns 2 and 3.")

original          : [0. 1. 2. 3. 4. 5.]
after On([0, 1])  : [-1.34 -1.34  2.    3.    4.    5.  ]
then On([4, 5])   : [-1.34 -1.34 -1.34 -1.34  2.    3.  ]

The second step meant original columns 4 and 5. It got positions 4 and 5
of the *reordered* array, which are original columns 2 and 3.

So: use a single On per pipeline where you can, and where you cannot, work out the indices against the previous step's output rather than against your DataFrame. The typical use is mixed-type data with one pass:

Workflow([
    On(columns=[0, 3, 7], transformer=OneHotEncoder()),
    RandomForestClassifier(),
])

On takes remainder="passthrough" by default, so untouched columns are carried through unchanged rather than dropped.

4.5 Order is part of the pipeline

Chapter 3 destroyed the insulin column by imputing before clipping. In a Workflow, that ordering is written down where you can see it — which is the point.

from tuiml.preprocessing import IQROutlierDetector

impute_first = Workflow([
    SimpleImputer(strategy="median"),
    IQROutlierDetector(factor=1.5, action="clip"),
    RandomForestClassifier(n_estimators=200, random_state=42),
])

clip_first = Workflow([
    IQROutlierDetector(factor=1.5, action="clip"),
    SimpleImputer(strategy="median"),
    RandomForestClassifier(n_estimators=200, random_state=42),
])

print(f"impute then clip: {cross_val_score(impute_first, X, y, cv=splitter).mean():.4f}")
print(f"clip then impute: {cross_val_score(clip_first, X, y, cv=splitter).mean():.4f}")

impute then clip: 0.7669

clip then impute: 0.7617

Two pipelines with the same three components and different orders are two different models. Chapter 8 shows how to benchmark variants like these properly instead of eyeballing two numbers.

4.6 A pipeline is also data

to_config() converts a Workflow into a plain dictionary.

import json

config = clip_first.to_config()
print(json.dumps(config, indent=2))

{
  "model": {
    "name": "RandomForestClassifier",
    "params": {
      "n_estimators": 200,
      "random_state": 42
    }
  },
  "pipeline": [
    {
      "name": "IQROutlierDetector"
    },
    {
      "name": "SimpleImputer",
      "params": {
        "strategy": "median"
      }
    }
  ]
}

Look carefully at that output, because it is the same shape as the spec you passed to tuiml.train() in chapter 0. That is not a coincidence — it is the design. A pipeline you built by hand out of objects can be dumped to configuration, checked into git, diffed against last week's, sent to a colleague, or handed to train() verbatim:

import tuiml

replayed = tuiml.train({
    **config,
    "data": {"X": X, "y": y},
    "evaluation": {"cv": 10},
    "random_seed": 42,
})

print(f"replayed from config: {replayed.metrics_['cv_accuracy_score_mean']:.4f}")

replayed from config: 0.7643

Chapter 9 takes this idea seriously. For now the point is that objects and configuration are two views of the same pipeline, and you can move between them in either direction.

4.7 Saving a fitted pipeline

save writes the fitted pipeline — transformers, learned statistics, model, and all — to one file.

import tempfile
from pathlib import Path

path = Path(tempfile.mkdtemp()) / "diabetes_pipeline.joblib"
fitted.save(path)

print(f"wrote {path.stat().st_size / 1024:.0f} KB")

loaded = Workflow.load(path)
same = (loaded.predict(X_test) == fitted.predict(X_test)).all()
print("reloaded pipeline predicts identically:", same)

wrote 3598 KB

reloaded pipeline predicts identically: True

Saving the pipeline rather than the model is what makes this useful. The imputer's learned fill values travel with it, so raw data at serving time gets exactly the treatment it got at training time. Save only the model and you have to reimplement the preprocessing in your serving code — and that reimplementation drifting out of sync is one of the most common ways a model that tested well behaves badly in production. Chapter 13 serves this file directly.

Recap

• Workflow([step, step, model]) — transformers in order, model last. The list order is the pipeline order.
• It behaves like a model: fit, predict, predict_proba, score, save, serve.
• Inside cross-validation it refits every step per fold, giving results identical to a correct hand-written loop.
• named_steps / transformers / model are the unfitted templates; steps_ and model_ are the fitted objects. Use the underscore versions to inspect or reuse a fitted step.
• On(columns=..., transformer=...) restricts a step to a column subset; the rest passes through — but transformed columns come out first and the remainder is appended, so indices shift. Chain On steps with care.
• Step order changes the model. Write it down where you can see it.
• to_config() turns a pipeline into the same dictionary tuiml.train() consumes.
• save/load persist the whole fitted pipeline, preprocessing included.

Next: chapter 5 builds new features and selects among them — and demonstrates the leakage that chapter 3 promised, where selecting features outside the fold produces a model that scores well above chance on pure noise.