# Pipeline Base Classes

The core pipeline framework provides the base abstractions for stage-based
orchestration. Domain-specific pipelines (LLM, LVM, etc.) subclass these to add
their own configuration and user-facing API.

## Pipeline

- *class* qairt.experimental.pipeline.torch.common.bases.pipeline.Pipeline(*config: PipelineConfigT*)

    - Bases: `ABC`, `Generic`[`PipelineConfigT`]

Generic pipeline base class.

Handles orchestration concerns common across model types (LLM, LVM, LMM, …):
stage resolution, config init, I/O binding, ordering, execution, and export.

- add\_observer(*observer: [StageObserver](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage_observer.StageObserver)*) → None

    - Attach an observer to monitor stage execution.
Multiple observers can be attached. They are called in registration order.

- Parameters

    - **observer** – A `StageObserver` instance to attach.

- Raises

    - **ValueError** – If the observer is already registered.

- *abstract property* config*: PipelineConfigT*

    - Return the domain-specific pipeline configuration.

- construct() → [Pipeline](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.pipeline.Pipeline)

    - Execute all pipeline stages in order, skipping any already completed or pre-executed.

- Raises

    - **AttributeError** – If no stage instances have been initialised.

- export(*export\_dir: Optional[str] = None*, *all\_stages: bool = False*) → None

    - Export pipeline artifacts to disk.

- Parameters

    - - **export\_dir** – Directory to write artifacts to. Defaults to the pipeline
config’s `cache_dir` if not provided.
- **all\_stages** – If `True`, export from every stage that supports export.
If `False` (default), export only from the last stage.

- Raises

    - - **AttributeError** – If `construct()` has not been called yet.
- **RuntimeError** – If the target stage does not support export.

- *classmethod* from\_pretrained(*model\_id\_or\_path: str*, *recipe: Optional[Union[str, Path, dict[str, Any]]] = None*, *\*\*kwargs: Any*) → PipelineT

    - Create a pipeline from a pretrained model identifier or local path.

- Parameters

    - - **model\_id\_or\_path** – HuggingFace model ID or local path.
- **recipe** – YAML file path, dict of recipe fields, or `None` to look
up a recipe by model ID or model type.
- **\*\*kwargs** – Config fields that override values from the recipe.

- Returns

    - A fully initialized instance of the calling pipeline subclass.

- Raises

    - **ValueError** – If *recipe* is `None` and no matching recipe is found.

- get\_observer\_reports() → dict[str, Any]

    - Retrieve the observer report from all observers that implement `get_observer_report()`.

- Returns

    - Dict mapping observer class name to its report object.
Observers without `get_observer_report()` are silently skipped.

- *classmethod* load(*cache\_dir: Union[str, Path]*) → PipelineT

    - Resume a pipeline from a previous run by restoring completed stages from the manifest.

Supports all cases where artifacts were written to disk:
- Caching enabled (with or without checkpoint)
- Caching disabled with checkpoint set

- Parameters

    - **cache\_dir** – Directory from the original run; must contain `pipeline_state/manifest.json`
and `pipeline_state/recipe.yaml`.

- Returns

    - Pipeline ready to resume. Call `construct()` to continue from where it left off.

- Raises

    - - **FileNotFoundError** – If the manifest or cached recipe does not exist.
- **ValueError** – If the manifest is empty, the recipe has no model identifier, or
    the original run had neither caching nor a checkpoint (no artifacts on disk).

- *property* manifest*: dict[str, Any]*

    - Return the pipeline manifest.

- *property* stage\_inputs*: ConfigAccessor*

    - Provide read access to stage inputs after construct.

Deprecated since version Use: `pipe.stages.<name>.input` instead.

- *property* stage\_outputs*: ConfigAccessor*

    - Provide read access to stage outputs after construct.

Deprecated since version Use: `pipe.stages.<name>.output` instead.

- *property* stages*: StagesAccessor*

    - Access pipeline stages with their config, input, and output.

Example:

pipe.stages.model_loader.config   # stage config
    pipe.stages.model_loader.output   # stage output (after construct)
    pipe.stages.model_loader.input    # stage input (after construct)
    pipe.stages.model_loader.run(...)  # delegate to stage instance
    Copy to clipboard

## PipelineConfig

- *class* qairt.experimental.pipeline.torch.common.bases.pipeline.PipelineConfig(*\**, *model\_id\_or\_path: str*, *cache\_dir: str = './workspace'*, *backend: [BackendType](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-api-configs.html#qairt.api.configs.common.BackendType) = BackendType.HTP*, *soc\_details: Optional[Union[qti.aisw.tools.core.utilities.devices.api.device\_definitions.SocDetails, str]] = None*, *log\_level: Optional[str] = None*, *enable\_cache: bool = False*, *checkpoint: Optional[str] = None*, *enable\_observers: bool = False*, *observers: dict[str, Any] = None*, *stage\_info: OrderedDict[str, dict[str, Any]] = None*)

    - Bases: [`PipelineContext`](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.pipeline_context.PipelineContext)

Base configuration shared by all pipeline types.

- *field* checkpoint*: Optional[str]*  *= None*

    - 

- *field* enable\_cache*: bool*  *= False*

    - 

- *field* enable\_observers*: bool*  *= False*

    - 

- model\_computed\_fields*: ClassVar[dict[str, ComputedFieldInfo]]*  *= {}*

    - A dictionary of computed field names and their corresponding <cite>ComputedFieldInfo</cite> objects.

- model\_config*: ClassVar[ConfigDict]*  *= {'arbitrary\_types\_allowed': True, 'protected\_namespaces': ()}*

    - Configuration for the model, should be a dictionary conforming to [<cite>ConfigDict</cite>][pydantic.config.ConfigDict].

- model\_fields*: ClassVar[dict[str, FieldInfo]]*  *= {'backend': FieldInfo(annotation=BackendType, required=False, default=&lt;BackendType.HTP: 'HTP'&gt;), 'cache\_dir': FieldInfo(annotation=str, required=False, default='./workspace'), 'checkpoint': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'enable\_cache': FieldInfo(annotation=bool, required=False, default=False), 'enable\_observers': FieldInfo(annotation=bool, required=False, default=False), 'log\_level': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'model\_id\_or\_path': FieldInfo(annotation=str, required=True), 'observers': FieldInfo(annotation=dict[str, Any], required=False, default\_factory=dict), 'soc\_details': FieldInfo(annotation=Union[SocDetails, str, NoneType], required=False, default=None), 'stage\_info': FieldInfo(annotation=OrderedDict[str, dict[str, Any]], required=False, default\_factory=OrderedDict)}*

    - Metadata about the fields defined on the model,
mapping of field names to [<cite>FieldInfo</cite>][pydantic.fields.FieldInfo].

This replaces <cite>Model.__fields__</cite> from Pydantic V1.

- model\_post\_init(*context: Any*, */*) → None

    - We need to both initialize private attributes and call the user-defined model\_post\_init
method.

- *field* observers*: dict[str, Any]*  *[Optional]*

    - 

- *field* stage\_info*: OrderedDict[str, dict[str, Any]]*  *[Optional]*

    - 

- to\_context() → [PipelineContext](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.pipeline_context.PipelineContext)

    - Return a context-only view of this config, stripping orchestration fields.

## Stage

- *class* qairt.experimental.pipeline.torch.common.bases.stage.Stage

    - Bases: `ABC`, `Generic`[`InputT`, `ConfigT`, `OutputT`]

Base class that all pipeline stages must implement.

A stage is defined by three Pydantic models that form its contract:

- `Input` - upstream data injected by the pipeline.
- `Config`- user-provided settings from recipe/code.
- `Output`- data produced and made available downstream.

- Config*: Type[ConfigT]*

    - 

- Input

    - alias of [`StageInput`](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageInput)

- Output*: Type[OutputT]*

    - 

- artifact\_dirname(*config: ConfigT*, *key: str*) → str

    - Return the directory name for this stage’s cached artifacts.

Default: `{self.name}_{key[:8]}`. Override in subclasses for
human-readable names.

- can\_evaluate*: bool*  *= False*

    - 

- can\_export*: bool*  *= True*

    - 

- can\_generate*: bool*  *= False*

    - 

- can\_start*: bool*  *= False*

    - 

- evaluate(*stage\_output: OutputT*, *config: ConfigT*, *\*\*kwargs*) → Any

    - Optional method for stages that evaluate their stage output (e.g. perplexity).
Override this method to implement custom evaluation logic and set
`can_evaluate = True`. By default, raises `NotImplementedError`.
Keyword-only beyond `stage_output`/`config` (no positional variadic args).

- *abstract classmethod* export(*stage\_output: OutputT*, *output\_path: Path*, *\*\*kwargs*) → None

    - Export stage artifacts to disk.

- Parameters

    - - **stage\_output** – The output of the stage to be exported.
- **output\_path** – The file path where the stage output should be exported.

- generate(*stage\_output: OutputT*, *config: ConfigT*, *\*args*, *\*\*kwargs*) → Any

    - Optional method for stages that perform generation using their stage output.
Override this method to implement custom generation logic.
By default, raises `NotImplementedError`.

- *classmethod* get\_stage\_dependency() → [StageDependencies](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageDependencies)

    - Declare dependencies for this stage.
Override in subclasses; default is no dependencies.

- *classmethod* load\_from\_cache(*cache\_path: Path*, *\*\*kwargs: Any*) → OutputT

    - Restore a `StageOutput` from artifacts written by `export()`.

Override in stages that support caching; raises `NotImplementedError` by default.

- Parameters

    - - **cache\_path** – Directory where this stage’s artifacts were previously exported.
- **\*\*kwargs** – Additional context for reconstruction (e.g. `lora_cfg`).

- Returns

    - A `StageOutput` equivalent to what `_execute()` would have produced.

- name*: ClassVar[str]*

    - 

- *final* run(*input: InputT*, *config: ConfigT*, *\**, *artifact\_path: Optional[Path] = None*) → OutputT

    - - Orchestrates stage execution by running hooks and core logic in order:
    - 1. `_pre_hook`:  preprocessing
2. `_execute`:   core stage logic
3. `_post_hook`: postprocessing

- Parameters

    - - **input** – Upstream data constructed by the pipeline from prior stage outputs.
- **config** – Validated stage configuration.
- **artifact\_path** – Optional path where the runner will persist this stage’s artifacts.

- Returns

    - The final stage output after all hooks have been applied.

- Return type

    - `OutputT`

- Raises

    - **StageExecutionError** – If an error occurs during stage execution.

## StageInput

- *class* qairt.experimental.pipeline.torch.common.bases.stage.StageInput(*\*\*extra\_data: Any*)

    - Bases: `BaseModel`

Base input for all pipeline stages.

Populated by the pipeline from the immediate predecessor’s output.
Subclass this to declare the upstream fields a stage needs. Required fields
(no default) will be validated by the pipeline before execution; optional
fields (default=None) are injected when available.

Example:

class QuantizationInput(StageInput):
        model: torch.nn.Module          # required - must be provided by upstream
        tokenizer: Any | None = None    # optional - injected if available
    Copy to clipboard

- model\_computed\_fields*: ClassVar[dict[str, ComputedFieldInfo]]*  *= {}*

    - A dictionary of computed field names and their corresponding <cite>ComputedFieldInfo</cite> objects.

- model\_config*: ClassVar[ConfigDict]*  *= {'arbitrary\_types\_allowed': True, 'extra': 'allow', 'protected\_namespaces': ()}*

    - Configuration for the model, should be a dictionary conforming to [<cite>ConfigDict</cite>][pydantic.config.ConfigDict].

- model\_fields*: ClassVar[dict[str, FieldInfo]]*  *= {}*

    - Metadata about the fields defined on the model,
mapping of field names to [<cite>FieldInfo</cite>][pydantic.fields.FieldInfo].

This replaces <cite>Model.__fields__</cite> from Pydantic V1.

## StageConfig

- *class* qairt.experimental.pipeline.torch.common.bases.stage.StageConfig(*\**, *execution\_environment: ExecutionEnvironment = ExecutionEnvironment.CPU*, *generator\_config: Optional[[GeneratorConfig](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-common.html#qairt.experimental.pipeline.torch.common.configs.GeneratorConfig)] = None*, *evaluator\_config: Optional[[EvaluatorConfig](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-common.html#qairt.experimental.pipeline.torch.common.configs.EvaluatorConfig)] = None*, *exporter\_config: Optional[[ExporterConfig](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-common.html#qairt.experimental.pipeline.torch.common.configs.ExporterConfig)] = None*, *\*\*extra\_data: Any*)

    - Bases: `BaseModel`

Base configuration for all pipeline stages. Set by users via recipe or code.

- *field* evaluator\_config*: Optional[[EvaluatorConfig](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-common.html#qairt.experimental.pipeline.torch.common.configs.EvaluatorConfig)]*  *= None*

    - 

- *field* execution\_environment*: ExecutionEnvironment*  *= ExecutionEnvironment.CPU*

    - 

- *field* exporter\_config*: Optional[[ExporterConfig](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-common.html#qairt.experimental.pipeline.torch.common.configs.ExporterConfig)]*  *= None*

    - 

- *field* generator\_config*: Optional[[GeneratorConfig](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-common.html#qairt.experimental.pipeline.torch.common.configs.GeneratorConfig)]*  *= None*

    - 

- model\_computed\_fields*: ClassVar[dict[str, ComputedFieldInfo]]*  *= {}*

    - A dictionary of computed field names and their corresponding <cite>ComputedFieldInfo</cite> objects.

- model\_config*: ClassVar[ConfigDict]*  *= {'arbitrary\_types\_allowed': True, 'extra': 'allow', 'protected\_namespaces': ()}*

    - Configuration for the model, should be a dictionary conforming to [<cite>ConfigDict</cite>][pydantic.config.ConfigDict].

- model\_fields*: ClassVar[dict[str, FieldInfo]]*  *= {'evaluator\_config': FieldInfo(annotation=Union[EvaluatorConfig, NoneType], required=False, default=None), 'execution\_environment': FieldInfo(annotation=ExecutionEnvironment, required=False, default=&lt;ExecutionEnvironment.CPU: 'cpu'&gt;), 'exporter\_config': FieldInfo(annotation=Union[ExporterConfig, NoneType], required=False, default=None), 'generator\_config': FieldInfo(annotation=Union[GeneratorConfig, NoneType], required=False, default=None)}*

    - Metadata about the fields defined on the model,
mapping of field names to [<cite>FieldInfo</cite>][pydantic.fields.FieldInfo].

This replaces <cite>Model.__fields__</cite> from Pydantic V1.

- model\_post\_init(*context: Any*, */*) → None

    - This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

- Parameters

    - - **self** – The BaseModel instance.
- **context** – The context.

## StageOutput

- *class* qairt.experimental.pipeline.torch.common.bases.stage.StageOutput(*\*\*extra\_data: Any*)

    - Bases: `BaseModel`

Base output for all pipeline stages. Produced by a stage and made available to downstream stages.

- model\_computed\_fields*: ClassVar[dict[str, ComputedFieldInfo]]*  *= {}*

    - A dictionary of computed field names and their corresponding <cite>ComputedFieldInfo</cite> objects.

- model\_config*: ClassVar[ConfigDict]*  *= {'arbitrary\_types\_allowed': True, 'extra': 'allow', 'protected\_namespaces': ()}*

    - Configuration for the model, should be a dictionary conforming to [<cite>ConfigDict</cite>][pydantic.config.ConfigDict].

- model\_fields*: ClassVar[dict[str, FieldInfo]]*  *= {}*

    - Metadata about the fields defined on the model,
mapping of field names to [<cite>FieldInfo</cite>][pydantic.fields.FieldInfo].

This replaces <cite>Model.__fields__</cite> from Pydantic V1.

## StageDependencies

- *class* qairt.experimental.pipeline.torch.common.bases.stage.StageDependencies(*\*\*data: Any*)

    - Bases: `BaseModel`

Defines ordering and conflict constraints for a pipeline stage.

- *field* conflict*: List[Type['Stage']] | List[str]*  *= []*

    - 

- model\_computed\_fields*: ClassVar[dict[str, ComputedFieldInfo]]*  *= {}*

    - A dictionary of computed field names and their corresponding <cite>ComputedFieldInfo</cite> objects.

- model\_config*: ClassVar[ConfigDict]*  *= {'arbitrary\_types\_allowed': True}*

    - Configuration for the model, should be a dictionary conforming to [<cite>ConfigDict</cite>][pydantic.config.ConfigDict].

- model\_fields*: ClassVar[dict[str, FieldInfo]]*  *= {'conflict': FieldInfo(annotation=Union[List[Type[Stage]], List[str]], required=False, default=[]), 'optional': FieldInfo(annotation=List[Union[Type[Stage], str]], required=False, default=[]), 'requires': FieldInfo(annotation=List[Union[Type[Stage], str]], required=False, default=[])}*

    - Metadata about the fields defined on the model,
mapping of field names to [<cite>FieldInfo</cite>][pydantic.fields.FieldInfo].

This replaces <cite>Model.__fields__</cite> from Pydantic V1.

- *field* optional*: List[Union[Type['Stage'], str]]*  *= []*

    - 

- *field* requires*: List[Union[Type['Stage'], str]]*  *= []*

    -

## StageRegistry

- *class* qairt.experimental.pipeline.torch.common.bases.stage\_registry.StageRegistry

    - Bases: `object`

Registry for managing pipeline stages.

The registry maps stage names (or tuples of identifiers) to stage classes.
It distinguishes between built-in stages and custom stages, and provides
methods for registration, removal, and listing of stages.

Example

>>> registry = StageRegistry()
    >>> registry.register("quantization", QuantizationStage, built_in=True)
    >>> registry.register(("custom", "my_stage"), MyCustomStage)
    >>> stages = registry.list_stages()
    Copy to clipboard

- clear(*custom\_only: bool = False*) → None

    - Clear the registry.

- Parameters

    - **custom\_only** – If True, only clear custom stages (keep built-in stages)

Example

>>> registry.clear(custom_only=True)  # Remove only custom stages
    >>> registry.clear()  # Remove all stages
    Copy to clipboard

- get(*key: Union[str, [BuiltInStage](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage_registry.BuiltInStage)]*) → Optional[Type[[Stage](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.Stage)]]

    - Retrieve a stage class by its key.

- Parameters

    - **key** – Stage identifier

- Returns

    - The stage class, or None if not found

Example

>>> stage_class = registry.get("quantization")
    Copy to clipboard

- is\_built\_in(*key: Union[str, [BuiltInStage](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage_registry.BuiltInStage)]*) → bool

    - Check if a stage is a built-in stage.

- Parameters

    - **key** – Stage identifier

- Returns

    - True if the stage is built-in, False otherwise

Example

>>> registry.is_built_in("quantization")
    True
    Copy to clipboard

- is\_custom(*key: Union[str, [BuiltInStage](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage_registry.BuiltInStage)]*) → bool

    - Check if a stage is a custom stage.

- Parameters

    - **key** – Stage identifier

- Returns

    - True if the stage is custom, False otherwise

Example

>>> registry.is_custom(("custom", "my_stage"))
    True
    Copy to clipboard

- list\_stages(*built\_in\_only: bool = False*, *custom\_only: bool = False*) → dict[Union[str, [qairt.experimental.pipeline.torch.common.bases.stage\_registry.BuiltInStage](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage_registry.BuiltInStage)], Type[[qairt.experimental.pipeline.torch.common.bases.stage.Stage](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.Stage)]]

    - List all registered stages.

- Parameters

    - - **built\_in\_only** – If True, return only built-in stages
- **custom\_only** – If True, return only custom stages

- Returns

    - Dictionary mapping stage keys to stage classes

Example

>>> all_stages = registry.list_stages()
    >>> built_in = registry.list_stages(built_in_only=True)
    >>> custom = registry.list_stages(custom_only=True)
    Copy to clipboard

- register(*key: Union[str, [BuiltInStage](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage_registry.BuiltInStage)]*, *stage\_class: Type[[Stage](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.Stage)]*, *built\_in: bool = False*) → None

    - Register a stage in the registry.

- Parameters

    - - **key** – Stage identifier (string name or tuple for extra delimiters)
- **stage\_class** – The stage class to register
- **built\_in** – Whether this is a built-in stage (default: False)

- Raises

    - **ValueError** – If the key is already registered

Example

>>> registry.register("quantization", QuantizationStage, built_in=True)
    >>> registry.register(("custom", "adapter"), CustomAdapterStage)
    Copy to clipboard

- unregister(*key: Union[str, [BuiltInStage](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage_registry.BuiltInStage)]*) → None

    - Remove a stage from the registry.

- Parameters

    - **key** – Stage identifier to remove

- Raises

    - **KeyError** – If the key is not found in the registry

Example

>>> registry.unregister("my_custom_stage")
    Copy to clipboard

## BuiltInStage

- *class* qairt.experimental.pipeline.torch.common.bases.stage\_registry.BuiltInStage(*value*)

    - Bases: `str`, `Enum`

Built-in stage identifiers for the pipeline framework.

These are internal pipeline stage names used for dependency resolution
and registration. They are NOT model identifiers (e.g. HuggingFace model IDs)
or external framework names — they identify pipeline processing steps.

- DM\_MODEL\_LOADER *= 'dm\_model\_loader'*

    - 

- DM\_QUANTIZATION *= 'dm\_quantization'*

    - 

- GENAI\_BUILDER *= 'genai\_builder'*

    - 

- MODEL\_LOADER *= 'model\_loader'*

    - 

- QUANTIZATION *= 'quantization'*

    - 

- QUANTIZATION\_OPT *= 'quantization\_opt'*

    -

## PipelineContext

- *class* qairt.experimental.pipeline.torch.common.bases.pipeline\_context.PipelineContext(*\**, *model\_id\_or\_path: str*, *cache\_dir: str = './workspace'*, *backend: [BackendType](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-api-configs.html#qairt.api.configs.common.BackendType) = BackendType.HTP*, *soc\_details: Optional[Union[qti.aisw.tools.core.utilities.devices.api.device\_definitions.SocDetails, str]] = None*, *log\_level: Optional[str] = None*)

    - Bases: `BaseModel`

Stage-visible pipeline context.

Contains only the fields that pipeline stages are permitted to access.
Injected into `StageConfig._pipeline_context` by the pipeline at
construction time.

- *field* backend*: [BackendType](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-api-configs.html#qairt.api.configs.common.BackendType)*  *= BackendType.HTP*

    - - Validated by

    - - [`normalize_backend`](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.pipeline_context.PipelineContext.normalize_backend)

- *field* cache\_dir*: str*  *= './workspace'*

    - 

- *field* log\_level*: Optional[str]*  *= None*

    - 

- model\_computed\_fields*: ClassVar[dict[str, ComputedFieldInfo]]*  *= {}*

    - A dictionary of computed field names and their corresponding <cite>ComputedFieldInfo</cite> objects.

- model\_config*: ClassVar[ConfigDict]*  *= {'arbitrary\_types\_allowed': True, 'protected\_namespaces': ()}*

    - Configuration for the model, should be a dictionary conforming to [<cite>ConfigDict</cite>][pydantic.config.ConfigDict].

- model\_fields*: ClassVar[dict[str, FieldInfo]]*  *= {'backend': FieldInfo(annotation=BackendType, required=False, default=&lt;BackendType.HTP: 'HTP'&gt;), 'cache\_dir': FieldInfo(annotation=str, required=False, default='./workspace'), 'log\_level': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'model\_id\_or\_path': FieldInfo(annotation=str, required=True), 'soc\_details': FieldInfo(annotation=Union[SocDetails, str, NoneType], required=False, default=None)}*

    - Metadata about the fields defined on the model,
mapping of field names to [<cite>FieldInfo</cite>][pydantic.fields.FieldInfo].

This replaces <cite>Model.__fields__</cite> from Pydantic V1.

- *field* model\_id\_or\_path*: str*  *[Required]*

    - - Validated by

    - - [`validate_model_id`](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.pipeline_context.PipelineContext.validate_model_id)

- model\_post\_init(*context: Any*, */*) → None

    - We need to both initialize private attributes and call the user-defined model\_post\_init
method.

- *validator* normalize\_backend*»* [*backend*](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.pipeline_context.PipelineContext.backend)

    - 

- *field* soc\_details*: Optional[Union[SocDetails, str]]*  *= None*

    - 

- *validator* validate\_model\_id*»* [*model\_id\_or\_path*](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.pipeline_context.PipelineContext.model_id_or_path)

    -

## StageObserver

- *class* qairt.experimental.pipeline.torch.common.bases.stage\_observer.StageObserver

    - Bases: `ABC`

Base class for stage observers.
Observers can monitor stage execution and collect data before and after each stage runs.
Multiple observers can be attached; they are called in registration order.

Example:

# Using the built-in observer:
    from qairt.experimental.pipeline.common.observers import StageProfilerObserver
    
    observer = StageProfilerObserver()
    pipe.add_observer(observer)
    pipe.construct()
    
    print(observer.get_execution_times())
    # {'model_loading': 1.23, 'quantization': 45.67}
    
    # Or implement a custom observer:
    class CheckpointObserver(StageObserver):
        def run_before_stage(self, stage, stage_config, stage_input):
            pass
    
        def run_after_stage(self, stage, stage_config, stage_input, stage_output):
            stage_output.save_checkpoint(stage.name)
    
    pipe.add_observer(CheckpointObserver())
    pipe.construct()
    Copy to clipboard

- get\_report() → dict

    - Return a structured report of the observer’s findings.

Override in subclasses that collect per-stage data (e.g. timings,
metrics).

- Returns

    - Dictionary mapping stage name to the captured value(s) for that stage.

- *abstract* run\_after\_stage(*stage: [Stage](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.Stage)*, *stage\_config: [StageConfig](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageConfig)*, *stage\_input: [StageInput](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageInput)*, *stage\_output: [StageOutput](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageOutput)*) → None

    - Called after stage execution.

- Parameters

    - - **stage** – The stage instance that was executed.
- **stage\_config** – The stage configuration.
- **stage\_input** – The input that was passed to the stage.
- **stage\_output** – The output produced by the stage.

- *abstract* run\_before\_stage(*stage: [Stage](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.Stage)*, *stage\_config: [StageConfig](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageConfig)*, *stage\_input: [StageInput](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageInput)*) → None

    - Called before stage execution.

- Parameters

    - - **stage** – The stage instance about to be executed.
- **stage\_config** – The stage configuration in its declared state.
- **stage\_input** – The input constructed for the stage from upstream outputs.

## Runners

Stage runner chain: \_CheckpointRunner → \_CachingRunner → \_ObservingRunner → \_DirectRunner → stage.run()

- *class* qairt.experimental.pipeline.torch.common.bases.runners.RunResult(*output: [StageOutput](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageOutput)*, *artifact\_path: Optional[Path] = None*, *is\_terminal: bool = False*, *cache\_key: Optional[str] = None*)

    - Bases: `object`

Return value of all stage runners, carrying output and where artifacts landed.

- artifact\_path*: Optional[Path]*  *= None*

    - 

- cache\_key*: Optional[str]*  *= None*

    - 

- is\_terminal*: bool*  *= False*

    - 

- output*: [StageOutput](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageOutput)*

    -

Last Published: Jul 08, 2026

[Previous Topic
Pipeline (Experimental)](https://docs.qualcomm.com/bundle/publicresource/80-87189-2/topics/qairt-pipeline.md)