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

- *static* excluded\_active\_fields(*stage\_config: Any*) → dict[str, Any]

    - Return `{field_name: value}` for `exclude=True` fields that are active.

Excluded fields (e.g. diagnostic hooks) don’t affect the cache key, so a
cache hit or manifest restore skips `_execute()` and never runs them. Callers
use this to warn when that happens instead of silently doing nothing.

A field is “active” if it’s truthy and, when it exposes an `enabled`
flag (attribute or dict key, as config objects like
`PostPrepareEvaluatorConfig` do), that flag is also truthy — a
present-but-disabled config object is not active.

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

    -

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

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

Evaluator configuration for the prepared model, run inside quantization.

Extends [`EvaluatorConfig`](https://docs.qualcomm.com/doc/80-87189-2/topic/qairt-pipeline-common.html#qairt.experimental.pipeline.torch.common.configs.EvaluatorConfig) (reusing `metrics`, `context_length`,
and `output_dir`) to evaluate the model produced by `prepare_llm` before
quantsim creation. Unlike `evaluator_config` — which only drives on-demand
`pipe.evaluate()` — this runs inline during quantization when `enabled`.
It reports metric scores (persisted as JSON); it does not compare against a
baseline.

- *field* enabled*: bool*  *= False*

    - Whether to run the evaluation. Opt-in.

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

    - Bases: `BaseModel`

Configuration for golden test vector generation after quantization.

Test vectors pair one sample input with the model’s floating-point output
(`fp_<index>.pkl`) and quantized output (`qt_<index>.pkl`), plus
input/output tensors recorded at regex-selected internal layers. They are
replayed on-device to localize where the on-device path diverges.

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

    - Overrides the generator’s total context window size.

- *field* dataset*: Optional[str | dict[str, Any]]*  *= None*

    - Dataset to draw samples from when no dataloader is passed; same string-or-dict shape the recipes accept. `None` defaults to wikitext.

- *field* num\_test\_vectors*: int*  *= 1*

    - Number of dataloader samples to record.

- *field* output\_dir*: str*  *[Required]*

    - Destination for the written vectors. `fp_N.pkl` / `qt_N.pkl` are written under `<output_dir>/test_vectors/`.

- *field* output\_index\_mapping*: dict[int, str]*  *[Optional]*

    - Maps each model output position to a name used as the key in the saved pickle. Its length must equal the model’s actual output count.

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

    - Overrides the generator’s autoregressive window size. Needed on the technique-level path where no export generator exists.

- *field* slice\_num*: int*  *= 1*

    - Index of the autoregressive window to record. Windows before it are run first to warm up the KV cache. `0` records the first window with no warm-up (not covered by legacy parity).

- *field* split\_candidate\_layers*: list[str]*  *[Optional]*

    - Regex patterns matched against module *names*; every matching layer’s inputs and outputs are recorded.

- *field* tokenizer*: Optional[Any]*  *= None*

    - Explicit tokenizer for the technique-level path, where the result carries none.

## 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)*, *stage\_key: Optional[str] = None*) → 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)*, *stage\_key: Optional[str] = None*) → None

    - Capture a snapshot before stage execution.

Last Published: Sep 17, 2026

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