# Common Utilities

Shared infrastructure used across the pipeline framework: recipes, manifest
tracking, caching, and cross-cutting configuration models.

## Recipe

Recipe class for pipeline configuration.

A Recipe is a declarative specification that defines pipeline configuration,
including global settings, features, and per-stage parameters.

- *class* qairt.experimental.pipeline.torch.common.recipe.Recipe(*\*args*, *schema: Optional[dict[str, Any]] = None*, *\*\*kwargs*)

    - Bases: `dict`

Recipe class that stores pipeline configuration, allowing arbitrary keys and values
while maintaining insertion order. It provides convenient methods for
saving to and loading from YAML files, and supports schema validation.

Example

>>> recipe = Recipe({
    ...     "model_id_or_path": "meta-llama/Llama-3.2-1B",
    ...     "backend": "HTP",
    ...     "stages": {"quantization": {"recipe": "lpbq_mse.yaml"}}
    ... })
    >>> recipe.to_yaml("my_recipe.yaml")
    >>> loaded = Recipe.from_yaml("my_recipe.yaml")
    Copy to clipboard

- *classmethod* from\_file(*path: str | pathlib.Path*, *schema: Optional[dict[str, Any]] = None*) → [Recipe](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-common.html#qairt.experimental.pipeline.torch.common.recipe.Recipe)

    - Load a recipe from a file, dispatching on the file extension.

Supported extensions: `.yaml`, `.yml`

- Parameters

    - - **path** – File path to the recipe file.
- **schema** – Optional schema dictionary for validation.

- Returns

    - Recipe instance loaded from the file.

- Raises

    - - **ValueError** – If the file extension is not supported.
- **FileNotFoundError** – If the file does not exist.

- *classmethod* from\_yaml(*path: str | pathlib.Path*, *schema: Optional[dict[str, Any]] = None*) → [Recipe](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-common.html#qairt.experimental.pipeline.torch.common.recipe.Recipe)

    - Load a recipe from a YAML file.

- Parameters

    - - **path** – File path to the YAML recipe file
- **schema** – Optional schema dictionary for validation

- Returns

    - Recipe instance loaded from the YAML file

- Raises

    - - **FileNotFoundError** – If the file does not exist
- **yaml.YAMLError** – If the file is not valid YAML

- *property* schema*: Optional[dict[str, Any]]*

    - Get the current schema for this recipe.

- to\_yaml(*path: str | pathlib.Path*, *\*\*kwargs*) → None

    - Save the recipe to a YAML file.

- Parameters

    - - **path** – File path where the recipe should be saved
- **\*\*kwargs** – Additional keyword arguments forwarded to `yaml.dump`

- validate(*schema: Optional[dict[str, Any]] = None*) → None

    - Validate the recipe against a schema.

- Parameters

    - **schema** – Schema dictionary to validate against. If None, uses the
schema provided during initialization.

- Raises

    - **ValueError** – If validation fails

- qairt.experimental.pipeline.torch.common.recipe.recipe\_representer(*dumper*, *data*)

    - Represent a Recipe as a plain YAML mapping

## PipelineManifest

Pipeline Manifest

Tracks which stages have completed and where their artifacts live.

Manifest structure on disk (`{cache_dir}/pipeline_state/manifest.json`):

{
        "stages": {
            "<stage_name>": {
                "artifact_path": "<path>",
                "timestamp":     "<iso8601>",
                "config":        "<json string>",   # optional, exclude-unset fields only
            }
        }
    }
    Copy to clipboard

- *class* qairt.experimental.pipeline.torch.common.manifest.PipelineManifest(*cache\_dir: Union[str, Path]*)

    - Bases: `object`

Reads and writes the pipeline-level manifest file.

The manifest records each completed stage’s artifact path, timestamp, and
config snapshot.

- *property* path*: Path*

    - Absolute path to `manifest.json`.

- read() → dict[str, Any]

    - Return the manifest as a dict, or an empty structure if absent.

Manifest structure:

{
        "stages": {
            "<stage_name>": {
                "artifact_path": "<path>",
                "timestamp": "<iso8601>",
                "config": "<json string>",   # optional
            }
        }
    }
    Copy to clipboard

- remove(*\*stage\_names: str*) → None

    - Remove one or more stage entries from the manifest.

Stages not present in the manifest are silently ignored.

- Parameters

    - **\*stage\_names** – Names of the stages to remove.

- update(*stage\_name: str*, *artifact\_path: Optional[Path]*, *stage\_config: Optional[[StageConfig](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageConfig)] = None*, *cache\_key: Optional[str] = None*) → None

    - Add or update the manifest entry for a single stage.

- Parameters

    - - **stage\_name** – Name of the stage.
- **artifact\_path** – Directory where artifacts were written, or `None` if no
artifacts were persisted (caching disabled, no checkpoint).
- **stage\_config** – Optional; only explicitly-set fields are serialised.
- **cache\_key** – Optional; the key-chain key for this stage, used by
`_preload_manifest_stages` to validate resume correctness.

## StageCache

Cache implementation for pipeline stages. Artifacts are stored on disk via
`stage.export()` and restored on cache hits via `load_from_cache()`.

Storage layout:

{cache_dir}/pipeline_state/
        manifest.json               # execution record used by Pipeline.load() to resume
        metrics.json                # per-stage runtime metrics (written at end of pipeline run)
        {stage_name}/{stage_name}_{key[:8]}/  # directory name uses first 8 chars of SHA-256 key
            cache_entry.json        # StageCache entry: full key, stage_name, timestamp, artifact_path
            <stage artifacts>       # written by stage.export() (may include stage-specific metadata.json)
    Copy to clipboard

- *class* qairt.experimental.pipeline.torch.common.cache.StageCache(*cache\_dir: Union[str, Path]*)

    - Bases: `object`

Cache store for pipeline stage outputs.

Keys are SHA-256 digests of (stage\_name, StageConfig). On a hit,
`get()` calls `stage_class.load_from_cache()`; on a miss, the caller runs
the stage and calls `put()` to persist it.

- *property* cache\_dir*: Path*

    - Return the pipeline state directory (`{cache_dir}/pipeline_state`).

- clear(*stage\_name: Optional[str] = None*) → None

    - Delete cached entries from disk and the in-memory index.

- Parameters

    - **stage\_name** – If provided, delete only entries for this stage; otherwise clears all.

- *static* compute\_key(*stage\_name: str*, *stage\_config: [StageConfig](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageConfig)*, *upstream\_key: Optional[str] = None*) → str

    - Return a deterministic SHA-256 hex digest for `(name, config, upstream_key)`.

Keys form a chain: `key_N = hash(name_N | config_N | key_{N-1})`.
This means any config change at stage N propagates to all downstream keys,
so a downstream stage’s cache hit implies the entire upstream config is unchanged.

- Parameters

    - - **stage\_name** – Name of the stage (`Stage.name`).
- **stage\_config** – The `StageConfig` instance for this run.
- **upstream\_key** – The cache key of the immediately preceding stage, or `None`
for the first stage in the pipeline.

- Returns

    - A 64-character lowercase hex string.

- get(*key: str*, *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)]*, *\*\*kwargs: Any*) → Optional[[StageOutput](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageOutput)]

    - Return the cached `StageOutput` for `key`, or `None` on a miss or corruption.

- Parameters

    - - **key** – Cache key returned by `compute_key()`.
- **stage\_class** – The `Stage` subclass whose `load_from_cache()` reconstructs the output.
- **\*\*kwargs** – Additional context forwarded to `load_from_cache()` (e.g. `lora_cfg`).

- Returns

    - A `StageOutput` instance, or `None` on miss or corruption.

- has(*key: str*) → bool

    - Return `True` if a valid cache entry exists for this key.

- Parameters

    - **key** – Cache key returned by `compute_key()`.

- put(*key: str*, *stage\_output: [StageOutput](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.StageOutput)*, *stage: [Stage](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-base.html#qairt.experimental.pipeline.torch.common.bases.stage.Stage)*, *artifact\_path: Path*) → None

    - Persist a `StageOutput` to disk and register it in the cache.

Calls `stage.export()` to write artifacts and writes `metadata.json`.

- Parameters

    - - **key** – Cache key returned by `compute_key()`.
- **stage\_output** – The output to cache.
- **stage** – Stage instance used to call `export()`.
- **artifact\_path** – Directory to write artifacts into; created if absent.

- write\_metrics(*metrics: dict[str, Any]*) → None

    - Write per-stage runtime metrics to `metrics.json` in the cache directory.

- Parameters

    - **metrics** – Dict mapping stage name to a dict of metric name → value,
as returned by `MetricsObserver.get_metrics()`.

## Configs

Common pipeline configuration classes

- *pydantic model* qairt.experimental.pipeline.torch.common.configs.EvaluatorConfig

    - Bases: `BaseModel`

Pipeline-level evaluator configuration. Parallel to GeneratorConfig.

The `metrics` list drives evaluation. Each entry is a dict with a `name`
key that selects the metric class from the registry; every other key is
forwarded as `**kwargs` to that metric’s `__init__`.

- *field* context\_length*: int*  *= 2048*

    - 

- *field* metrics*: list[dict[str, Any]]*  *[Optional]*

    - 

- *field* output\_dir*: Optional[str]*  *= None*

    -

- *pydantic model* qairt.experimental.pipeline.torch.common.configs.ExporterConfig

    - Bases: `BaseModel`

Global exporter configuration.

- *pydantic model* qairt.experimental.pipeline.torch.common.configs.GeneratorConfig

    - Bases: `BaseModel`

Configuration for generation.

- *field* context\_length*: Optional[int]*  *= None*

    - 

- *field* sequence\_length*: Optional[int]*  *= None*

    -

- *pydantic model* qairt.experimental.pipeline.torch.common.configs.LoRAAdapterConfig

    - Bases: `BaseModel`

Configuration for a single LoRA adapter.

- *field* path*: DirectoryPath*  *[Required]*

    - - Constraints

    - - **path\_type** = dir

- *pydantic model* qairt.experimental.pipeline.torch.common.configs.LoRAAdapterMetadata

    - Bases: `BaseModel`

Aggregated LoRA adapter metadata produced during model loading.

Groups peft configs, alpha scalings, and max scale into a single object
so downstream stages can receive it as Optional[LoRAAdapterMetadata].

- *field* alpha\_scalings*: dict[str, list[float]]*  *[Required]*

    - 

- *field* max\_alpha\_scale*: float*  *[Required]*

    - 

- *field* peft\_configs*: dict[str, 'PeftConfig']*  *[Required]*

    -

- *pydantic model* qairt.experimental.pipeline.torch.common.configs.LoRAConcurrencyConfig

    - Bases: `BaseModel`

Configuration for a LoRA concurrency group (adapter subset + calibration).

- Validators

    - - `_check_scaling_length` » `all fields`

- *field* adapters*: list[str]*  *[Required]*

    - - Validated by

    - - `_check_scaling_length`

- *field* calibration\_dataset*: str*  *[Required]*

    - - Validated by

    - - `_check_scaling_length`

- *field* lora\_scaling*: list[float]*  *[Required]*

    - - Validated by

    - - `_check_scaling_length`

- *pydantic model* qairt.experimental.pipeline.torch.common.configs.LoRAFeatureConfig

    - Bases: `BaseModel`

Top-level LoRA feature configuration for the pipeline.

- Validators

    - - `_check_concurrency_adapter_refs` » `all fields`

- *field* adapters*: dict[str, [LoRAAdapterConfig](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-common.html#qairt.experimental.pipeline.torch.common.configs.LoRAAdapterConfig)]*  *[Optional]*

    - - Validated by

    - - `_check_concurrency_adapter_refs`

- *field* concurrencies*: dict[str, [LoRAConcurrencyConfig](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-common.html#qairt.experimental.pipeline.torch.common.configs.LoRAConcurrencyConfig)]*  *[Optional]*

    - - Validated by

    - - `_check_concurrency_adapter_refs`

- *field* quant\_updatable\_mode*: Literal['adapter\_only', 'all', 'none']*  *= 'adapter\_only'*

    - - Validated by

    - - `_check_concurrency_adapter_refs`

- *pydantic model* qairt.experimental.pipeline.torch.common.configs.OnnxExporterConfig

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

ONNX exporter configuration.

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

    -

## Observers

Stage Profiler Observer

A single observer backed by one Profiler instance that captures both timing
and memory metrics for each pipeline stage.

- *class* qairt.experimental.pipeline.torch.common.observers.profiler\_observer.StageProfilerObserver

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

Records timing and RAM/GPU memory metrics for each pipeline stage.

Uses a single ResourceProfiler instance so timing and memory are captured in one
snapshot pair per stage. Results are available via `get_execution_times()`,
`get_metrics()`, and `get_observer_report()`.

- get\_execution\_times() → dict[str, float]

    - Return a copy of the wall-clock duration for all completed stages.

- Returns

    - Dict mapping stage name to elapsed duration in seconds.
Example: `{'model_loading': 1.23, 'quantization': 45.67}`

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

    - Return a copy of the collected per-stage memory metrics.

- Returns

    - Dict mapping stage name to a dict of metric name to value.
GPU metrics are always present; they are `0.0` when CUDA is unavailable.
Example:

{
        'model_loading': {
            'ram_pss_entry_mb': 512.0,
            'ram_pss_mb': 128.5,
            'gpu_memory_allocated_mb': 0.0,
            'gpu_memory_peak_mb': 0.0,
        },
    }
    Copy to clipboard

- get\_observer\_report() → Report

    - Return a MemoryProfilingReport with full timing and memory event data.

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

    - Return timing and memory data as a dict (StageObserver interface).

- Returns

    - Dict with keys `"execution_times"` and `"metrics"`.

- reset() → None

    - Reset all collected data and profiler store.

- 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

    - Record timing and memory metrics after stage completion.

- 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

    - Capture a snapshot before stage execution.

Last Published: Jun 19, 2026

[Previous Topic
run\_evaluation()](https://docs.qualcomm.com/bundle/publicresource/80-87189-2/topics/qairt-pipeline-generation.md) [Next Topic
Appendix](https://docs.qualcomm.com/bundle/publicresource/80-87189-2/topics/appendix.md)