5. Build a pipeline you can explain
Your goal: understand every node’s effect, generate a small comparison, then build feature fusion, stacking and a two-source residual model. A pipeline grows by adding a reasoned operation, not by copying a complicated graph.
The clickable node map
Click a family to see its parameters, equivalent code tabs, an illustration and the expected output. The complete node catalogue lists individual operations.
SNV, scatter correction, smoothing, scaling and dimensionality reduction. What changes in X?
Scale y for fitting; return predictions in original scientific units.
Choose training/validation membership, preserve groups and score held-out observations.
Regressors, classifiers, PLS, Ridge and training/search settings.
Enumerate alternatives, ranges, grids, paired choices and constrained combinations.
Send observations or sources through several paths, then join features or predictions.
Identify observations with a tag; remove permitted training rows with an exclusion.
Create derived training observations, extra feature views or concatenated transforms.
Reshape repeated scans while retaining their specimen relationships.
Additional node pages: merge, exclude, concat_transform, charts, auto-transfer, and residual model.
Read inputs and outputs before choosing a node
Operation |
Input |
Expected result |
|---|---|---|
SNV/MSC |
Spectral rows |
Corrected spectral rows; same dimensions |
PCA or variable selection |
Feature matrix |
Fewer or selected feature columns |
|
Training targets |
Transformed targets; predictions inversely transformed |
Splitter |
Observation/group identities |
Training and validation memberships |
Model |
Prepared X and training y |
Learned relation; one output per requested target |
Generator |
Recipe with alternatives |
Several separate concrete recipes |
|
Shared rows or named sources |
Several paths inside one recipe |
|
Aligned transformed feature blocks |
More columns, same observation IDs |
|
Held-out base-model outputs |
Prediction columns for a final learner |
|
Observations and a rule |
An annotation / permitted removal from fitting |
Augmentation |
Training rows and perturbation |
Derived rows or extra views, with origins retained |
Repetition conversion |
Related measurements |
A changed representation of the same specimens |
Chart |
Inputs or result |
An inspectable figure; no model improvement by itself |
1 · Compare two preprocessors and three model sizes
SNV normalizes each row using its own mean and spread. MSC compares each spectrum to a reference estimated from training spectra. They are alternatives here. PLS n_components chooses how many target-related directions the model uses.
Read the expansion before fitting. SNV and MSC each pair with 2, 4 and 6 components. Six candidates × three folds = eighteen fold-model fits, followed by refit of the selected recipe. This is a search, not a six-model ensemble.
{
"pipeline": [
{
"_or_": [
{
"class": "nirs4all.operators.transforms.StandardNormalVariate"
},
{
"class": "nirs4all.operators.transforms.MultiplicativeScatterCorrection"
}
]
},
{
"class": "sklearn.model_selection.KFold",
"params": {
"n_splits": 3,
"shuffle": true,
"random_state": 17
}
},
{
"model": {
"class": "sklearn.cross_decomposition.PLSRegression",
"params": {
"n_components": {
"_range_": [
2,
6,
2
]
}
}
}
}
]
}
pipeline:
- _or_:
- class: nirs4all.operators.transforms.StandardNormalVariate
- class: nirs4all.operators.transforms.MultiplicativeScatterCorrection
- class: sklearn.model_selection.KFold
params:
n_splits: 3
shuffle: true
random_state: 17
- model:
class: sklearn.cross_decomposition.PLSRegression
params:
n_components:
_range_:
- 2
- 6
- 2
from nirs4all.operators.transforms import SNV, MSC
from sklearn.model_selection import KFold
from sklearn.cross_decomposition import PLSRegression
pipeline = [
{"_or_": [SNV(), MSC()]},
KFold(n_splits=3, shuffle=True, random_state=17),
{"model": {"class": PLSRegression,
"params": {"n_components": {"_range_": [2, 6, 2]}}}},
]
# Run against the dataset from the first Python SDK exercise:
# result = nirs4all.run(pipeline, (X, y), engine="dag-ml", refit=True)
This example uses Python SDK objects. An equivalent public operation is unavailable in this host. Use the shared native recipe in 10. Run the same recipe in your language for the executable cross-language path.
This example uses Python SDK objects. An equivalent public operation is unavailable in this host. Use the shared native recipe in 10. Run the same recipe in your language for the executable cross-language path.
This example uses Python SDK objects. An equivalent public operation is unavailable in this host. Use the shared native recipe in 10. Run the same recipe in your language for the executable cross-language path.
Expected result: six candidate recipes, each evaluated with the same folds and target. _range_: [2, 6, 2] includes 2, 4 and 6. Keep the component count within the feature rank and smallest training-fold size. Use the synthetic SDK arrays from 1. Run your first complete experiment, or supply your actual dataset.
Enumerate the generator vocabulary
Need |
Keyword |
What to expect |
|---|---|---|
Try A or B |
|
One separate candidate per alternative |
Integer/linear values |
|
A start/end/step sweep |
Logarithmic values |
|
Multiplicative-scale choices |
Combine parameter settings |
|
Cartesian parameter combinations |
Combine pipeline stages |
|
Every compatible stage combination |
Keep matched settings together |
|
Paired choices, instead of all combinations |
Join ordered specifications |
|
Concatenated ordered expansions |
Sample a space |
|
A reproducible bounded sample |
Remove incompatible choices |
|
Explicit compatibility rules |
Reuse a named definition |
|
Expanded registered settings |
The generator page shows each mechanism; the exact keyword reference gives schemas and defaults. Some advanced expansions have runtime-specific support; inspect the generated values and preflight the chosen engine.
2 · Keep two views inside one recipe
Now use both normalization and smoothing as inputs to one model. Two paths see the same 31-column spectra. SNV returns 31 columns and smoothing returns 31, so merge: features produces 62 columns for Ridge. Forty rows train the recipe; eight separate rows test its refit.
Two kinds of combination. Early fusion joins feature blocks before fitting one model. Stacking joins held-out prediction columns before fitting a final learner. Count the columns and keep the observation IDs aligned.
{
"pipeline": [
{
"split": {
"class": "sklearn.model_selection.KFold",
"params": {
"n_splits": 3,
"shuffle": true,
"random_state": 17
}
}
},
{
"branch": {
"normalized": [
{
"class": "nirs4all.operators.transforms.StandardNormalVariate"
}
],
"smoothed": [
{
"class": "nirs4all.operators.transforms.SavitzkyGolay",
"params": {
"window_length": 7,
"polyorder": 2,
"deriv": 0
}
}
]
}
},
{
"merge": "features"
},
{
"model": {
"class": "sklearn.linear_model.Ridge",
"params": {
"alpha": 1.0
}
}
}
]
}
pipeline:
- split:
class: sklearn.model_selection.KFold
params: {n_splits: 3, shuffle: true, random_state: 17}
- branch:
normalized:
- class: nirs4all.operators.transforms.StandardNormalVariate
smoothed:
- class: nirs4all.operators.transforms.SavitzkyGolay
params: {window_length: 7, polyorder: 2, deriv: 0}
- merge: features
- model:
class: sklearn.linear_model.Ridge
params: {alpha: 1.0}
import tempfile
import numpy as np
import nirs4all
from sklearn.linear_model import Ridge
from sklearn.model_selection import KFold
from nirs4all.data.dataset import SpectroDataset
from nirs4all.operators.transforms import SNV, SavitzkyGolay
rng = np.random.default_rng(17)
X = rng.normal(size=(48, 31))
y = 2 * X[:, 10] - X[:, 20] + rng.normal(scale=0.1, size=48)
dataset = SpectroDataset("doc_branch_feature_join")
dataset.add_samples(X[:40], {"partition": "train"},
headers=[str(i) for i in range(31)])
dataset.add_samples(X[40:], {"partition": "test"})
dataset.add_targets(y.reshape(-1, 1))
pipeline = [
KFold(n_splits=3, shuffle=True, random_state=17),
{"branch": {
"normalized": [SNV()],
"smoothed": [SavitzkyGolay(window_length=7, polyorder=2, deriv=0)],
}},
{"merge": "features"},
{"model": Ridge(alpha=1.0)},
]
with tempfile.TemporaryDirectory() as workspace:
result = nirs4all.run(
pipeline=pipeline, dataset=dataset, engine="dag-ml", refit=True,
workspace_path=workspace, save_charts=False, verbose=0,
)
print(result.execution_engine, result.best_rmse)
result.close()
This example uses Python SDK objects. An equivalent public operation is unavailable in this host. Use the shared native recipe in 10. Run the same recipe in your language for the executable cross-language path.
This example uses Python SDK objects. An equivalent public operation is unavailable in this host. Use the shared native recipe in 10. Run the same recipe in your language for the executable cross-language path.
This example uses Python SDK objects. An equivalent public operation is unavailable in this host. Use the shared native recipe in 10. Run the same recipe in your language for the executable cross-language path.
Expected result: the model receives 62 columns for every observation. More columns do not guarantee lower held-out error. Compare this recipe with the simpler baseline under the same validation protocol.
The Python tab is self-contained; JSON/YAML describe the same recipe, with the same data generated in Python. Fully qualified class paths construct Python objects. For native R/Octave/WASM training, use the supported finite recipe in 10. Run the same recipe in your language or the native multimodal tutorial.
3 · Combine models by stacking
Put a PLS model in the SNV path and Ridge in the MSC path. For each held-out row, each path contributes one prediction. merge: predictions therefore makes two input columns for a final Ridge learner, rather than 62 spectral columns.
The next Python tab continues the preceding example and reuses X, y and its imports. It creates a fresh dataset for this separate experiment. Its JSON/YAML tabs declare exactly the same stacking recipe.
{
"pipeline": [
{
"split": {
"class": "sklearn.model_selection.KFold",
"params": {
"n_splits": 3,
"shuffle": true,
"random_state": 17
}
}
},
{
"branch": {
"pls": [
{
"class": "nirs4all.operators.transforms.StandardNormalVariate"
},
{
"model": {
"class": "sklearn.cross_decomposition.PLSRegression",
"params": {
"n_components": 3
}
}
}
],
"ridge": [
{
"class": "nirs4all.operators.transforms.MultiplicativeScatterCorrection"
},
{
"model": {
"class": "sklearn.linear_model.Ridge",
"params": {
"alpha": 1.0
}
}
}
]
}
},
{
"merge": "predictions"
},
{
"model": {
"class": "sklearn.linear_model.Ridge",
"params": {
"alpha": 0.1
}
}
}
]
}
pipeline:
- split:
class: sklearn.model_selection.KFold
params:
n_splits: 3
shuffle: true
random_state: 17
- branch:
pls:
- class: nirs4all.operators.transforms.StandardNormalVariate
- model:
class: sklearn.cross_decomposition.PLSRegression
params:
n_components: 3
ridge:
- class: nirs4all.operators.transforms.MultiplicativeScatterCorrection
- model:
class: sklearn.linear_model.Ridge
params:
alpha: 1.0
- merge: predictions
- model:
class: sklearn.linear_model.Ridge
params:
alpha: 0.1
from pathlib import Path
from sklearn.cross_decomposition import PLSRegression
from nirs4all.operators.transforms import MSC
stacking_data = SpectroDataset("doc_stacking")
stacking_data.add_samples(X[:40], {"partition": "train"},
headers=[str(i) for i in range(31)])
stacking_data.add_samples(X[40:], {"partition": "test"})
stacking_data.add_targets(y.reshape(-1, 1))
stacking = [
KFold(n_splits=3, shuffle=True, random_state=17),
{"branch": {
"pls": [SNV(), {"model": PLSRegression(n_components=3)}],
"ridge": [MSC(), {"model": Ridge(alpha=1.0)}],
}},
{"merge": "predictions"},
{"model": Ridge(alpha=0.1)},
]
with tempfile.TemporaryDirectory() as workspace:
result = nirs4all.run(
pipeline=stacking, dataset=stacking_data, engine="dag-ml", refit=True,
workspace_path=workspace, save_charts=False, verbose=0,
)
archive = result.export(Path(workspace) / "stacked.n4a")
replay = nirs4all.predict(archive, X[40:])
assert len(replay.y_pred) == 8
print(result.execution_engine, result.best_rmse)
result.close()
This example uses Python SDK objects. An equivalent public operation is unavailable in this host. Use the shared native recipe in 10. Run the same recipe in your language for the executable cross-language path.
This example uses Python SDK objects. An equivalent public operation is unavailable in this host. Use the shared native recipe in 10. Run the same recipe in your language for the executable cross-language path.
This example uses Python SDK objects. An equivalent public operation is unavailable in this host. Use the shared native recipe in 10. Run the same recipe in your language for the executable cross-language path.
Expected result: two base predictors and a meta-predictor are learned; exported replay returns eight values for the eight test rows. Meta-training inputs must be predictions made without training the base model on that observation. In-sample base predictions give the final learner overly favorable inputs.
This exercise uses duplication branches, default prediction merge and one downstream learner. Combining source routing with additional prediction branches is a richer graph and may be refused by DAG-ML. Choose a documented supported composition; merge: decide what to bring back together explains the available forms.
4 · Add a second source and a residual correction
Use 31 spectral measurements and three laboratory markers for the same specimens. Normalize the spectral source, standardize marker columns, and merge the two sources into 34 features. PLS makes a first estimate; Ridge learns what remains unexplained using held-out residuals. The final prediction is the base estimate plus its learned correction.
{
"pipeline": [
{
"split": {
"class": "sklearn.model_selection.KFold",
"params": {
"n_splits": 3,
"shuffle": true,
"random_state": 17
}
}
},
{
"branch": {
"by_source": true,
"steps": {
"source_0": [
{
"class": "nirs4all.operators.transforms.StandardNormalVariate"
}
],
"source_1": [
{
"class": "sklearn.preprocessing.StandardScaler"
}
]
}
}
},
{
"merge": {
"sources": "concat"
}
},
{
"model": {
"class": "nirs4all.operators.models.ResidualModel",
"params": {
"base": {
"class": "sklearn.cross_decomposition.PLSRegression",
"params": {
"n_components": 3
}
},
"learner": {
"class": "sklearn.linear_model.Ridge",
"params": {
"alpha": 1.0
}
},
"gate": false
}
}
}
]
}
pipeline:
- split:
class: sklearn.model_selection.KFold
params: {n_splits: 3, shuffle: true, random_state: 17}
- branch:
by_source: true
steps:
source_0:
- class: nirs4all.operators.transforms.StandardNormalVariate
source_1:
- class: sklearn.preprocessing.StandardScaler
- merge:
sources: concat
- model:
class: nirs4all.operators.models.ResidualModel
params:
base:
class: sklearn.cross_decomposition.PLSRegression
params: {n_components: 3}
learner:
class: sklearn.linear_model.Ridge
params: {alpha: 1.0}
gate: false
import tempfile
from pathlib import Path
import numpy as np
import nirs4all
from sklearn.cross_decomposition import PLSRegression
from sklearn.linear_model import Ridge
from sklearn.model_selection import KFold
from sklearn.preprocessing import StandardScaler
from nirs4all.data.dataset import SpectroDataset
from nirs4all.operators.models import ResidualModel
from nirs4all.operators.transforms import SNV
rng = np.random.default_rng(17)
spectra = rng.normal(size=(48, 31))
markers = rng.normal(size=(48, 3))
y = (2 * spectra[:, 10] - spectra[:, 20] + markers[:, 0]
+ rng.normal(scale=0.1, size=48))
dataset = SpectroDataset("doc_multimodal_residual")
dataset.add_samples(
[spectra[:40], markers[:40]], {"partition": "train"},
headers=[[str(i) for i in range(31)],
["marker_0", "marker_1", "marker_2"]],
)
dataset.add_samples([spectra[40:], markers[40:]], {"partition": "test"})
dataset.add_targets(y.reshape(-1, 1))
pipeline = [
KFold(n_splits=3, shuffle=True, random_state=17),
{"branch": {"by_source": True, "steps": {
"source_0": [SNV()],
"source_1": [StandardScaler()],
}}},
{"merge": {"sources": "concat"}},
{"model": ResidualModel(
base=PLSRegression(n_components=3),
learner=Ridge(alpha=1.0), gate=False,
)},
]
with tempfile.TemporaryDirectory() as workspace:
result = nirs4all.run(
pipeline=pipeline, dataset=dataset, engine="dag-ml", refit=True,
workspace_path=workspace, save_charts=False, verbose=0,
)
archive = result.export(Path(workspace) / "multimodal-residual.n4a")
# Supply raw columns in the original source order; replay owns preprocessing.
raw_new = np.column_stack([spectra[40:], markers[40:]])
replay = nirs4all.predict(archive, raw_new)
assert len(replay.y_pred) == 8
assert np.isfinite(replay.y_pred).all()
print(result.execution_engine, result.best_rmse)
result.close()
This example uses Python SDK objects. An equivalent public operation is unavailable in this host. Use the shared native recipe in 10. Run the same recipe in your language for the executable cross-language path.
This example uses Python SDK objects. An equivalent public operation is unavailable in this host. Use the shared native recipe in 10. Run the same recipe in your language for the executable cross-language path.
This example uses Python SDK objects. An equivalent public operation is unavailable in this host. Use the shared native recipe in 10. Run the same recipe in your language for the executable cross-language path.
Expected result: the final exported predictor keeps source preprocessing, the PLS base and the Ridge correction together. Replay consumes raw source columns in their original order, and returns eight finite predictions. The two sources must describe the same observations; real files should be joined by sample IDs before loading.
Run all three advanced checkpoints
Download D08 documented workflows. It constructs synthetic data and runs feature fusion, stacking, and multisource residual learning, including export/replay checks. The final recipe is also available as YAML and JSON.
The examples demonstrate the data flow. They are not evidence that a more complex model improves a real spectroscopy problem. Follow 6. Measure whether your model answers the question to choose the test that answers your deployment question.
Checkpoint: explain why a generator makes separate candidates, a feature branch makes more input columns, and stacking makes prediction columns. Draw the final model’s inputs before proceeding.