# 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\_export*: bool*  *= True*

    - 

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

    - 

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

    - 

- *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.

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

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

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

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

- PipelineContext.normalize\_backend

    - 

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

    - 

- PipelineContext.validate\_model\_id

    -

## 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: Jun 19, 2026

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