Перейти к содержанию

Pipeline

The central entity of the framework is the Pipeline class. It stores an ordered sequence of steps (Step objects) and exposes methods to train models and run inference on the described workflow.

Steps are passed as a dictionary steps where the key is a unique step name (for example, "nail_localization") and the value is a Step instance. Since Python 3.7, dict insertion order is preserved, so this layout is safe and guarantees unique names.

During inference, the pipeline supports two execution modes:

  • shift_result=True (default): the first step receives the input passed to predict(), and each following step automatically receives the previous step's output. Similar to scikit-learn pipelines.
  • shift_result=False: each step uses its own inputs list; the pipeline resolves dependencies and injects results from named steps before calling the action. Similar to Kedro.

Each pipeline stage is represented by a Step. It stores a callable action, optional inputs (step names or "__input__"), a human-readable description, and an optional preprocessor for incoming data.

The action can be any callable: a plain function, a lambda, or an object with __call__. For specialized ML tasks, pass dedicated functors — the framework adjusts training and inference accordingly:

from hyppopipe.data.image import Image
from hyppopipe.pipeline import Pipeline, Step
from hyppopipe.pipeline.image.classification import ImageClassifier
from hyppopipe.pipeline.image.localization import ImageLocalizer

brain_pipeline = Pipeline(
    {
        "tumor_localize": Step(
            ImageLocalizer(),
            description="Locate tumor region",
        ),
        "tumor_classify": Step(
            ImageClassifier(source_mode="roi"),
            inputs=["tumor_localize"],
            description="Classify tumor type",
        ),
    },
    shift_result=False,
)

image = Image.from_path("mri_slice.dcm")
result = brain_pipeline.predict(image, bundle_path="artifacts/")

Training on datasets, not on step outputs

Each model is trained on the dataset you provide to Trainer, not on the outputs of the previous pipeline step. Steps only define inference-time data flow.

Documentation

Pipeline

Ordered graph of :class:~hyppopipe.pipeline.step.Step for train and predict.

Example

Build from a mapping, train selected steps, then run inference::

pipeline = Pipeline({"detect": detect_step, "classify": classify_step})
pipeline.train({"detect": trainer_det, "classify": trainer_cls}, data=splits)
pred = pipeline.predict(image, train_result=result)  # or predict(image) for transform-only steps

__init__(steps, *, shift_result=True) 

__init__(steps: Iterable[Step], *, shift_result: bool = True)

__init__(steps: Mapping[str, Step], *, shift_result: bool = True)

__init__(steps: Pipeline, *, shift_result: bool = True)

Build a pipeline from steps in dependency order.

Parameters:

Name Type Description Default
steps Iterable[Step] | Mapping[str, Step] | Pipeline

Named mapping, iterable of Step, or another Pipeline to copy.

 required
shift_result bool

If True, predict chains each step's output into the next step instead of resolving inputs from Step.inputs.

 True

predict(image, *, bundle=None, bundle_path=None, train_result=None, run_index_by_step=None, device=None, score_thresh=0.5, step_base_models=None, return_all=False)

Run inference for every step in topological order.

Parameters:

Name Type Description Default
image Image

Input image stored under registry key __input__.

 required
bundle PredictBundle | None

Pre-exported :class:~hyppopipe.train.bundle.PredictBundle.

 None
bundle_path Path | str | None

Directory containing manifest.json and weights.

 None
train_result TrainResult | None

Ephemeral bundle built from a recent :class:~hyppopipe.train.result.TrainResult.

 None
run_index_by_step dict[str, int] | None

Per-step index into StepTrainResult.runs when using train_result.

 None
device str | device | None

Torch device string or instance; auto-detected if None.

 None
score_thresh float

Minimum detection/segmentation score for ROI selection.

 0.5
step_base_models Mapping[str, Module] | None

Optional pre-built base modules keyed by step name.

 None
return_all bool

If True, include __input__ in returned outputs.

 False

Returns:

Type Description
PipelinePrediction

Predictions keyed by step name (and optionally __input__).

Raises:

Type Description
ValueError

If trained steps lack exactly one artifact source, or artifact sources are passed when no step needs them.
KeyError

If a trained step has no artifacts in the bundle.

train(step_config, data=None, config=None, *, log_to=None)

Train one or more pipeline steps with the given trainers.

Parameters:

Name Type Description Default
step_config Mapping[str, 'Trainer']

Map from step name to :class:~hyppopipe.train.trainer.Trainer.

 required
data SplitData | None

Default split data when a trainer has no data set.

 None
config TrainingConfig | None

Optional override of each trainer's :class:~hyppopipe.train.config.TrainingConfig.

 None
log_to Path | str | LogConfig | None

Per-run logging target (see :func:~hyppopipe.package_logging.run_logging).

 None

Returns:

Type Description
TrainResult

Aggregated training results per step.

Raises:

Type Description
KeyError

If step_config references unknown step names.
ValueError

If training data is missing for a step.

Step

Wraps a callable action with optional named inputs and metadata.

__call__(*args, previous_result=NO_VALUE, **kwargs)

Invoke action, optionally threading a prior step result.

Parameters:

Name Type Description Default
*args

Positional arguments when previous_result is not set.

 ()
previous_result Any

Output of the previous step in chained predict mode.

 NO_VALUE
**kwargs

Keyword arguments forwarded to action.

 {}

Returns:

Type Description
Any

Whatever action returns.

__init__(action, action_args=None, action_kwargs=None, *, inputs=None, name=None, description=None, input_prepare=None)

Configure a pipeline step.

Parameters:

Name Type Description Default
action Callable

Functor (e.g. ImageClassifier) or any callable executed at runtime.

 required
action_args tuple[Any, ...] | None

Positional arguments forwarded to action when invoked directly.

 None
action_kwargs dict[str, Any] | None

Keyword arguments forwarded to action when invoked directly.

 None
inputs Sequence[str] | None

Registry keys required before this step runs ("__input__" for the raw image).

 None
name str | None

Stable step identifier; auto-generated when the pipeline is built from a list.

 None
description str | None

Human-readable label for logging and str(step).

 None
input_prepare Callable[[tuple[Any, ...]], tuple[Any, ...]] | None

Optional transform applied to resolved inputs before inference.

 None