nirs4all.pipeline.retrainer module

Pipeline retrainer - Handles retraining, transfer learning, and fine-tuning.

This module provides the Retrainer class for retraining pipelines with different modes: full retrain, transfer learning (reuse preprocessing), and fine-tuning.

Phase 7 Implementation:

This module enables retraining trained pipelines on new data with various modes: - full: Train everything from scratch (same pipeline structure) - transfer: Use existing preprocessing artifacts, train new model - finetune: Continue training existing model with new data

Design Principles:

1. Controller-Agnostic: Works with any controller type via per-step mode control

2. Reuses Existing Infrastructure: Leverages resolver, artifact provider, executor

3. Composable: Same infrastructure for all retrain modes

class nirs4all.pipeline.retrainer.ExtractedPipeline(steps: list[Any] = <factory>, trace: ExecutionTrace | None = None, artifact_provider: ArtifactProvider | None = None, model_step_index: int | None = None, preprocessing_chain: str = '', source_pipeline_uid: str = '', metadata: dict[str, ~typing.Any]=<factory>)

Bases: object

Extracted pipeline for inspection and modification.

Represents a pipeline extracted from a trained prediction, ready for inspection, modification, or re-execution.

Variables:

• steps (list[Any]) – List of pipeline steps (can be modified)

• trace (nirs4all.pipeline.trace.execution_trace.ExecutionTrace | None) – Original execution trace (read-only)

• artifact_provider (nirs4all.pipeline.config.context.ArtifactProvider | None) – Provider for original artifacts

• model_step_index (int | None) – Index of the model step

• preprocessing_chain (str) – Summary of preprocessing

• source_pipeline_uid (str) – UID of the source pipeline

• metadata (dict[str, Any]) – Additional metadata

artifact_provider: ArtifactProvider | None = None

get_model_step() → Any | None

Get the model step.

Returns:

Model step configuration or None

get_step(index: int) → Any

Get a step by 0-based index.

Parameters:

index – 0-based step index

Returns:

Step configuration

metadata: dict[str, Any]

model_step_index: int | None = None

preprocessing_chain: str = ''

set_model(model: Any) → None

Replace the model in the model step.

Parameters:

model – New model to use

set_step(index: int, step: Any) → None

Set a step by 0-based index.

Parameters:

• index – 0-based step index

• step – New step configuration

source_pipeline_uid: str = ''

steps: list[Any]

trace: ExecutionTrace | None = None

class nirs4all.pipeline.retrainer.RetrainArtifactProvider(base_provider: ArtifactProvider, retrain_config: RetrainConfig, trace: ExecutionTrace | None = None)

Bases: ArtifactProvider

Artifact provider for retraining that respects step modes.

Provides artifacts only for steps that should use existing artifacts (i.e., mode=’predict’), while returning None for steps that should train.

Variables:

• base_provider – Underlying artifact provider

• retrain_config – Configuration determining which steps use artifacts

• trace – Execution trace for step type detection

get_artifact(step_index: int, fold_id: int | str | None = None) → Any | None

Get a single artifact for a step if applicable.

Parameters:

• step_index – 1-based step index

• fold_id – Optional fold ID for fold-specific artifacts

Returns:

Artifact object or None if step should train

get_artifacts_for_step(step_index: int, branch_path: list[int] | None = None, branch_id: int | None = None, source_index: int | None = None, substep_index: int | None = None) → list[tuple[str, Any]]

Get all artifacts for a step if applicable.

Parameters:

• step_index – 1-based step index

• branch_path – Optional branch path filter

• branch_id – Optional branch ID filter

• source_index – Optional source/dataset index filter for multi-source

• substep_index – Optional substep index filter for branch substeps

Returns:

List of (artifact_id, artifact_object) tuples, or empty if should train

get_fold_artifacts(step_index: int, branch_path: list[int] | None = None) → list[tuple[int, Any]]

Get all fold-specific artifacts for a step if applicable.

Parameters:

• step_index – 1-based step index

• branch_path – Optional branch path filter

Returns:

List of (fold_id, artifact_object) tuples, or empty if should train

has_artifacts_for_step(step_index: int) → bool

Check if artifacts should be used for this step.

Parameters:

step_index – 1-based step index

Returns:

True if artifacts are available and should be used

class nirs4all.pipeline.retrainer.RetrainConfig(mode: RetrainMode = RetrainMode.FULL, step_modes: list[StepMode] = <factory>, new_model: Any | None = None, epochs: int | None = None, learning_rate: float | None = None, freeze_layers: list[str] | None = None, metadata: dict[str, ~typing.Any]=<factory>)

Bases: object

Configuration for retraining operation.

Variables:

• mode (nirs4all.pipeline.retrainer.RetrainMode) – Overall retrain mode (full, transfer, finetune)

• step_modes (list[nirs4all.pipeline.retrainer.StepMode]) – Per-step mode overrides (optional, for fine-grained control)

• new_model (Any | None) – Optional new model to use instead of original (for transfer)

• epochs (int | None) – Optional epochs for fine-tuning

• learning_rate (float | None) – Optional learning rate for fine-tuning

• freeze_layers (list[str] | None) – Optional list of layers to freeze during fine-tuning

• metadata (dict[str, Any]) – Additional metadata for the retrain operation

epochs: int | None = None

freeze_layers: list[str] | None = None

get_step_mode(step_index: int) → StepMode | None

Get mode override for a specific step.

Parameters:

step_index – 1-based step index

Returns:

StepMode if override exists, None otherwise

learning_rate: float | None = None

metadata: dict[str, Any]

mode: RetrainMode = 'full'

new_model: Any | None = None

should_train_step(step_index: int, is_model: bool = False) → bool

Determine if a step should train based on mode and overrides.

Parameters:

• step_index – 1-based step index

• is_model – Whether this is the model step

Returns:

True if step should train

step_modes: list[StepMode]

class nirs4all.pipeline.retrainer.RetrainMode(value)

Bases: StrEnum

Mode of retraining operation.

Variables:

• FULL – Train everything from scratch (same pipeline structure)

• TRANSFER – Use existing preprocessing artifacts, train new model

• FINETUNE – Continue training existing model with new data

FINETUNE = 'finetune'

FULL = 'full'

TRANSFER = 'transfer'

class nirs4all.pipeline.retrainer.Retrainer(runner: PipelineRunner)

Bases: object

Handles retraining pipelines with various modes.

This class manages the retrain workflow: loading saved pipelines, determining which steps to retrain vs. reuse, and executing the modified pipeline on new data.

Phase 7 Implementation:

The Retrainer enables three modes: - full: Train from scratch with same pipeline structure - transfer: Use existing preprocessing, train new model - finetune: Continue training existing model

Variables:

• runner – Parent PipelineRunner instance

• resolver – Prediction resolver for loading sources

extract(source: dict[str, Any] | str | Path | Any, verbose: int = 0) → ExtractedPipeline

Extract a pipeline for inspection or modification.

Returns an ExtractedPipeline object that can be inspected, modified, and then executed with runner.run().

Parameters:

• source – Prediction source (dict, folder, Run, artifact_id, bundle)

• verbose – Verbosity level

Returns:

ExtractedPipeline for inspection/modification

Example

>>> extracted = retrainer.extract(best_pred)
>>> print(extracted.steps)
>>> extracted.steps[-1] = {"model": RandomForestRegressor()}
>>> preds, _ = runner.run(extracted.steps, new_data)

retrain(source: dict[str, Any] | str | Path | Any, dataset: DatasetConfigs | SpectroDataset | ndarray | tuple[ndarray, ...] | dict | list[dict] | str | list[str], mode: str | RetrainMode = 'full', dataset_name: str = 'retrain_dataset', new_model: Any | None = None, epochs: int | None = None, step_modes: list[StepMode] | None = None, verbose: int = 0, **kwargs) → tuple[Predictions, dict[str, Any]]

Retrain a pipeline on new data.

Parameters:

• source – Prediction source (dict, folder, Run, artifact_id, bundle)

• dataset – New dataset to train on

• mode – Retrain mode (‘full’, ‘transfer’, ‘finetune’)

• dataset_name – Name for the dataset

• new_model – Optional new model for transfer mode

• epochs – Optional epochs for fine-tuning

• step_modes – Optional per-step mode overrides

• verbose – Verbosity level

• **kwargs – Additional parameters (learning_rate, freeze_layers, etc.)

Returns:

Tuple of (predictions, dataset_predictions_dict)

Example

>>> retrainer = Retrainer(runner)
>>>
>>> # Full retrain
>>> preds, _ = retrainer.retrain(best_pred, new_data, mode='full')
>>>
>>> # Transfer: use preprocessing, new model
>>> preds, _ = retrainer.retrain(best_pred, new_data, mode='transfer')
>>>
>>> # Finetune: continue training
>>> preds, _ = retrainer.retrain(best_pred, new_data, mode='finetune', epochs=10)

class nirs4all.pipeline.retrainer.StepMode(step_index: int, mode: str = 'train', artifact_id: str | None = None, kwargs: dict[str, ~typing.Any]=<factory>)

Bases: object

Mode override for a specific step during retraining.

Enables fine-grained control over which steps train vs. use existing artifacts.

Variables:

• step_index (int) – 1-based step index to apply this mode to

• mode (str) – How to execute this step (‘train’, ‘predict’, ‘skip’)

• artifact_id (str | None) – For ‘predict’ mode, specific artifact to use

• kwargs (dict[str, Any]) – Additional step-specific parameters (e.g., epochs for finetune)

artifact_id: str | None = None

is_predict() → bool

Check if this step should use existing artifacts.

Returns:

True if step should use existing artifacts (predict mode)

is_train() → bool

Check if this step should train.

Returns:

True if step should be trained

kwargs: dict[str, Any]

mode: str = 'train'

step_index: int