Pipeline

Центральной сущностью фреймворка является класс Pipeline. Его ответственность — хранить последовательность шагов (объекты Step), а также предоставлять возможность запуска обучения и предсказания для описанного пайплайна.

Шаги задаются в виде словаря steps, где ключ — уникальное имя шага (например, "nail_localization"), а значение — объект Step. Начиная с Python 3.7, словари сохраняют порядок вставки, поэтому такой способ описания безопасен и гарантирует уникальность имён.

В режиме предсказания пайплайн может работать в двух режимах:

shift_result=True (по умолчанию): первый шаг получает вход из predict(), каждый следующий — результат предыдущего. Аналогично scikit-learn.
shift_result=False: каждый шаг использует свой список inputs; пайплайн подставляет результаты именованных шагов перед вызовом. Аналогично Kedro.

Каждый этап представлен объектом Step, который хранит вызываемое action, опциональные inputs, description и preprocessor. Для специализированных ML-задач используйте готовые функторы:

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="Локализация опухоли",
        ),
        "tumor_classify": Step(
            ImageClassifier(source_mode="roi"),
            inputs=["tumor_localize"],
            description="Классификация типа опухоли",
        ),
    },
    shift_result=False,
)

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

Обучение на датасете, а не на выходах шагов

Каждая модель обучается на датасете, переданном в Trainer, а не на результатах предыдущего шага. Шаги определяют поток данных только при инференсе.

Документация¶

`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`

Pipeline

Документация¶

Pipeline ¶

__init__(steps, *, shift_result=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) ¶

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

Step ¶

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

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

`Pipeline` ¶

`init(steps, *, shift_result=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)` ¶

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

`Step` ¶

`call(*args, previous_result=NO_VALUE, **kwargs)` ¶

`init(action, action_args=None, action_kwargs=None, *, inputs=None, name=None, description=None, input_prepare=None)` ¶