Skip to content

Commit 48741ac

Browse files
qti-horodnicqti-horodnic
authored
Qualcomm AI Engine Direct - [GenAI Pipeline] PR3: Pipeline Orchestrator (pytorch#21149)
## Summary This PR adds the pipeline orchestrator (`GenAIPipeline`) that wires together all the components from PRs 1 and 2. It assembles stages from `EngineProxy`, builds `InputConfig` objects from `OutputConfig` results, executes stages sequentially with timing, and returns `InferenceOutputConfig`. No existing files are modified. ### What's included #### Pipeline orchestrator: - `GenAIPipeline`: main orchestrator class with: - `_STRATEGY_REGISTRY`: maps `(stage_name, engine_type)` → `(StageClass, StrategyClass)` for extensibility - `from_proxy()`: factory method that resolves stages from `EngineProxy` - `invoke()`: executes model_preparation → quantization → compilation → inference - Private `_run_*` methods: each builds an `InputConfig`, calls the stage, returns the `OutputConfig` - Structured logging with `[GenAIPipeline] StageName started/completed in X.Xs` format per LLD Section 5.2 - `executorch_model_preparation_strategy.py`: ExecuTorch model preparation strategy stub (implementation in a subsequent PR) #### Unit tests (13 tests): - `from_proxy()`: creates all stages, skip stages, default engines - `from_proxy()` error paths: unsupported engine for each stage - `invoke()`: full pipeline with mock strategies, compile-only, no stages - Data wiring: quantization receives `soc_model`, compilation receives `backend_type`, inference receives `prompt` ### PR Review Checklist - All new classes follow single responsibility (one class per file) - Yes. - All dependencies are injected via constructor with sensible defaults - Yes. - All external calls are behind injectable interfaces - Yes (via strategy pattern). - Unit tests cover every public method - Yes. - No existing files are modified (Phase 1 constraint) - Yes (only `__init__.py` updated to add `GenAIPipeline` export). - Docstrings on all public classes and methods - Yes. - Type annotations on all function signatures - Yes. - Logging follows the strategy in the LLD - Yes (`[GenAIPipeline]` prefix, timing). - ### Related PRs - PR 1: Core data model, engine routing & exceptions: pytorch#20409 - PR 2: Strategy interfaces & stage wrappers: pytorch#20795. - PR 3: Pipeline orchestrator: this pr. - PR 4: Adapter interfaces & default implementations: pending. - PR 5: Model preparation & quantization strategies: pending - PR 6: Compilation & inference strategy implementations: pending. - PR 7: Integration & E2E tests: pending. ## Test plan ``` python -m pytest \ backends/qualcomm/genai_pipeline/tests/test_genai_pipeline.py \ -v ``` All 13 unit-tests passed. ### Test Coverage Command to run: ``` coverage run --rcfile=backends/qualcomm/.coveragerc \ -m pytest \ backends/qualcomm/genai_pipeline/tests/test_genai_pipeline.py \ -v coverage report --rcfile=backends/qualcomm/.coveragerc \ --include="backends/qualcomm/genai_pipeline/genai_pipeline.py" ``` Result: ``` Name Stmts Miss Branch BrPart Cover Missing ------------------------------------------------------------------------------------------------ backends/qualcomm/genai_pipeline/genai_pipeline.py 98 7 12 1 93% 184-195 ------------------------------------------------------------------------------------------------ TOTAL 98 7 12 1 93% ```
1 parent a439afd commit 48741ac

4 files changed

Lines changed: 606 additions & 0 deletions

File tree

‎backends/qualcomm/genai_pipeline/__init__.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,10 +23,12 @@
2323
PipelineError,
2424
StageError,
2525
)
26+
from executorch.backends.qualcomm.genai_pipeline.genai_pipeline import GenAIPipeline
2627
from executorch.backends.qualcomm.genai_pipeline.pipeline_context import (
2728
PipelineContext,
2829
PipelineContextBuilder,
2930
)
31+
from executorch.backends.qualcomm.genai_pipeline.pipeline_stage import PipelineStage
3032
from executorch.backends.qualcomm.genai_pipeline.pipeline_types import EngineType
3133

3234
__all__ = [
@@ -36,6 +38,7 @@
3638
"EngineNotAvailableError",
3739
"EngineProxy",
3840
"EngineType",
41+
"GenAIPipeline",
3942
"InferenceInputConfig",
4043
"InferenceOutputConfig",
4144
"ModelPreparationInputConfig",

‎backends/qualcomm/genai_pipeline/genai_pipeline.py‎

Lines changed: 270 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,270 @@
1+
# Copyright (c) Qualcomm Innovation Center, Inc.
2+
# All rights reserved
3+
#
4+
# This source code is licensed under the BSD-style license found in the
5+
# LICENSE file in the root directory of this source tree.
6+
7+
import logging
8+
import time
9+
from pathlib import Path
10+
from typing import Dict, Optional, Set, Tuple, Type
11+
12+
from executorch.backends.qualcomm.genai_pipeline.configs.compilation_input_config import (
13+
CompilationInputConfig,
14+
)
15+
from executorch.backends.qualcomm.genai_pipeline.configs.compilation_output_config import (
16+
CompilationOutputConfig,
17+
)
18+
from executorch.backends.qualcomm.genai_pipeline.configs.inference_input_config import (
19+
InferenceInputConfig,
20+
)
21+
from executorch.backends.qualcomm.genai_pipeline.configs.inference_output_config import (
22+
InferenceOutputConfig,
23+
)
24+
from executorch.backends.qualcomm.genai_pipeline.configs.model_preparation_input_config import (
25+
ModelPreparationInputConfig,
26+
)
27+
from executorch.backends.qualcomm.genai_pipeline.configs.model_preparation_output_config import (
28+
ModelPreparationOutputConfig,
29+
)
30+
from executorch.backends.qualcomm.genai_pipeline.configs.quantization_input_config import (
31+
QuantizationInputConfig,
32+
)
33+
from executorch.backends.qualcomm.genai_pipeline.configs.quantization_output_config import (
34+
QuantizationOutputConfig,
35+
)
36+
from executorch.backends.qualcomm.genai_pipeline.engine_proxy import EngineProxy
37+
from executorch.backends.qualcomm.genai_pipeline.pipeline_context import PipelineContext
38+
from executorch.backends.qualcomm.genai_pipeline.pipeline_types import (
39+
EngineType,
40+
STAGE_COMPILATION,
41+
STAGE_INFERENCE,
42+
STAGE_MODEL_PREPARATION,
43+
STAGE_QUANTIZATION,
44+
)
45+
from executorch.backends.qualcomm.genai_pipeline.stages.compilation_stage import (
46+
CompilationStage,
47+
)
48+
from executorch.backends.qualcomm.genai_pipeline.stages.inference_stage import (
49+
InferenceStage,
50+
)
51+
from executorch.backends.qualcomm.genai_pipeline.stages.model_preparation_stage import (
52+
ModelPreparationStage,
53+
)
54+
from executorch.backends.qualcomm.genai_pipeline.stages.quantization_stage import (
55+
QuantizationStage,
56+
)
57+
from executorch.backends.qualcomm.genai_pipeline.strategies.compilation.executorch_compilation_strategy import (
58+
ExecuTorchCompilationStrategy,
59+
)
60+
from executorch.backends.qualcomm.genai_pipeline.strategies.inference.executorch_inference_strategy import (
61+
ExecuTorchInferenceStrategy,
62+
)
63+
from executorch.backends.qualcomm.genai_pipeline.strategies.model_preparation.executorch_model_preparation_strategy import (
64+
ExecuTorchModelPreparationStrategy,
65+
)
66+
from executorch.backends.qualcomm.genai_pipeline.strategies.quantization.executorch_quantization_strategy import (
67+
ExecuTorchQuantizationStrategy,
68+
)
69+
70+
logger = logging.getLogger("genai_pipeline")
71+
72+
# Maps (stage_name, engine_type) → (StageClass, StrategyClass)
73+
_STRATEGY_REGISTRY: Dict[Tuple[str, EngineType], Tuple[Type, Type]] = {
74+
(STAGE_MODEL_PREPARATION, EngineType.EXECUTORCH): (
75+
ModelPreparationStage,
76+
ExecuTorchModelPreparationStrategy,
77+
),
78+
(STAGE_QUANTIZATION, EngineType.EXECUTORCH): (
79+
QuantizationStage,
80+
ExecuTorchQuantizationStrategy,
81+
),
82+
(STAGE_COMPILATION, EngineType.EXECUTORCH): (
83+
CompilationStage,
84+
ExecuTorchCompilationStrategy,
85+
),
86+
(STAGE_INFERENCE, EngineType.EXECUTORCH): (
87+
InferenceStage,
88+
ExecuTorchInferenceStrategy,
89+
),
90+
}
91+
92+
93+
class GenAIPipeline:
94+
"""Main pipeline orchestrator for GenAI model workflows.
95+
96+
Assembles stages from EngineProxy, wires data flow between
97+
InputConfig → Stage → OutputConfig, and executes sequentially:
98+
model_preparation → quantization → compilation → inference.
99+
"""
100+
101+
def __init__(
102+
self,
103+
model_preparation_stage: Optional[ModelPreparationStage],
104+
quantization_stage: Optional[QuantizationStage],
105+
compilation_stage: Optional[CompilationStage],
106+
inference_stage: Optional[InferenceStage],
107+
engine_proxy: EngineProxy,
108+
) -> None:
109+
self._model_preparation_stage = model_preparation_stage
110+
self._quantization_stage = quantization_stage
111+
self._compilation_stage = compilation_stage
112+
self._inference_stage = inference_stage
113+
self._engine_proxy = engine_proxy
114+
115+
@classmethod
116+
def from_proxy(
117+
cls,
118+
engine_proxy: EngineProxy,
119+
skip_stages: Optional[Set[str]] = None,
120+
) -> "GenAIPipeline":
121+
"""Create a pipeline from an EngineProxy configuration.
122+
123+
Args:
124+
engine_proxy: The EngineProxy with per-stage engine assignments.
125+
skip_stages: Optional set of stage names to skip.
126+
127+
Returns:
128+
A fully configured GenAIPipeline.
129+
"""
130+
skip = skip_stages or set()
131+
132+
model_prep_stage = cls._resolve_stage(
133+
engine_proxy, STAGE_MODEL_PREPARATION, skip
134+
)
135+
quant_stage = cls._resolve_stage(engine_proxy, STAGE_QUANTIZATION, skip)
136+
compile_stage = cls._resolve_stage(engine_proxy, STAGE_COMPILATION, skip)
137+
infer_stage = cls._resolve_stage(engine_proxy, STAGE_INFERENCE, skip)
138+
139+
return cls(
140+
model_preparation_stage=model_prep_stage,
141+
quantization_stage=quant_stage,
142+
compilation_stage=compile_stage,
143+
inference_stage=infer_stage,
144+
engine_proxy=engine_proxy,
145+
)
146+
147+
@staticmethod
148+
def _resolve_stage(
149+
engine_proxy: EngineProxy,
150+
stage_name: str,
151+
skip: set,
152+
):
153+
if stage_name in skip:
154+
return None
155+
156+
engine = engine_proxy.get_engine(stage_name)
157+
key = (stage_name, engine)
158+
159+
if key not in _STRATEGY_REGISTRY:
160+
raise ValueError(f"No {stage_name} strategy for engine: {engine}")
161+
162+
stage_cls, strategy_cls = _STRATEGY_REGISTRY[key]
163+
return stage_cls(strategy_cls())
164+
165+
def invoke(self, context: PipelineContext) -> InferenceOutputConfig:
166+
"""Execute the full pipeline sequentially.
167+
168+
Args:
169+
context: The pipeline context with user inputs.
170+
171+
Returns:
172+
InferenceOutputConfig with inference results and metrics.
173+
"""
174+
logger.info(
175+
"[GenAIPipeline] Pipeline started for model '%s'", context.model_name
176+
)
177+
178+
model_prep_output = self._run_model_preparation(context)
179+
quant_output = self._run_quantization(context, model_prep_output)
180+
compile_output = self._run_compilation(context, model_prep_output, quant_output)
181+
result = self._run_inference(context, model_prep_output, compile_output)
182+
183+
logger.info("[GenAIPipeline] Pipeline completed")
184+
return result
185+
186+
def _run_model_preparation(
187+
self, context: PipelineContext
188+
) -> ModelPreparationOutputConfig:
189+
if self._model_preparation_stage is not None:
190+
logger.info("[GenAIPipeline] ModelPreparationStage started")
191+
start = time.monotonic()
192+
input_config = ModelPreparationInputConfig(
193+
model_name=context.model_name,
194+
soc_model=context.soc_model,
195+
)
196+
output = self._model_preparation_stage.invoke(context, input_config)
197+
elapsed = time.monotonic() - start
198+
logger.info(
199+
"[GenAIPipeline] ModelPreparationStage completed in %.1fs", elapsed
200+
)
201+
return output
202+
return ModelPreparationOutputConfig()
203+
204+
def _run_quantization(
205+
self,
206+
context: PipelineContext,
207+
model_prep_output: ModelPreparationOutputConfig,
208+
) -> QuantizationOutputConfig:
209+
if self._quantization_stage is not None:
210+
logger.info("[GenAIPipeline] QuantizationStage started")
211+
start = time.monotonic()
212+
input_config = QuantizationInputConfig(
213+
soc_model=context.soc_model,
214+
backend_type=self._engine_proxy.backend_type,
215+
model_module=model_prep_output.model_module,
216+
calibration_data=model_prep_output.calibration_data,
217+
)
218+
output = self._quantization_stage.invoke(context, input_config)
219+
elapsed = time.monotonic() - start
220+
logger.info("[GenAIPipeline] QuantizationStage completed in %.1fs", elapsed)
221+
return output
222+
return QuantizationOutputConfig()
223+
224+
def _run_compilation(
225+
self,
226+
context: PipelineContext,
227+
model_prep_output: ModelPreparationOutputConfig,
228+
quant_output: QuantizationOutputConfig,
229+
) -> CompilationOutputConfig:
230+
if self._compilation_stage is not None:
231+
logger.info("[GenAIPipeline] CompilationStage started")
232+
start = time.monotonic()
233+
# Fall back to the prepared module when quantization was skipped
234+
# (FP16 / GPU flows), so the model isn't silently dropped.
235+
model = quant_output.quantized_model or model_prep_output.model_module
236+
input_config = CompilationInputConfig(
237+
# Note: context.soc_model is str; str→QcomChipset conversion is
238+
# deferred to the strategy/adapter layer (see PR5).
239+
soc_model=context.soc_model,
240+
backend_type=self._engine_proxy.backend_type,
241+
model=model,
242+
artifact_dir=Path(context.artifact_dir),
243+
)
244+
output = self._compilation_stage.invoke(context, input_config)
245+
elapsed = time.monotonic() - start
246+
logger.info("[GenAIPipeline] CompilationStage completed in %.1fs", elapsed)
247+
return output
248+
return CompilationOutputConfig()
249+
250+
def _run_inference(
251+
self,
252+
context: PipelineContext,
253+
model_prep_output: ModelPreparationOutputConfig,
254+
compile_output: CompilationOutputConfig,
255+
) -> InferenceOutputConfig:
256+
if self._inference_stage is not None:
257+
logger.info("[GenAIPipeline] InferenceStage started")
258+
start = time.monotonic()
259+
input_config = InferenceInputConfig(
260+
artifact_paths=compile_output.artifact_paths,
261+
tokenizer=model_prep_output.tokenizer,
262+
runtime_tokenizer_path=model_prep_output.runtime_tokenizer_path,
263+
prompt=context.prompt,
264+
soc_model=context.soc_model,
265+
)
266+
output = self._inference_stage.invoke(context, input_config)
267+
elapsed = time.monotonic() - start
268+
logger.info("[GenAIPipeline] InferenceStage completed in %.1fs", elapsed)
269+
return output
270+
return InferenceOutputConfig()

‎backends/qualcomm/genai_pipeline/strategies/model_preparation/executorch_model_preparation_strategy.py‎

Lines changed: 43 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
# Copyright (c) Qualcomm Innovation Center, Inc.
2+
# All rights reserved
3+
#
4+
# This source code is licensed under the BSD-style license found in the
5+
# LICENSE file in the root directory of this source tree.
6+
7+
from executorch.backends.qualcomm.genai_pipeline.configs.model_preparation_input_config import (
8+
ModelPreparationInputConfig,
9+
)
10+
from executorch.backends.qualcomm.genai_pipeline.configs.model_preparation_output_config import (
11+
ModelPreparationOutputConfig,
12+
)
13+
from executorch.backends.qualcomm.genai_pipeline.pipeline_context import PipelineContext
14+
from executorch.backends.qualcomm.genai_pipeline.strategies.model_preparation.model_preparation_strategy import (
15+
ModelPreparationStrategy,
16+
)
17+
18+
19+
class ExecuTorchModelPreparationStrategy(ModelPreparationStrategy):
20+
"""ExecuTorch-based model preparation using HuggingFace transformers.
21+
22+
Loads model weights, tokenizer, and generates calibration data
23+
for downstream quantization and compilation stages.
24+
"""
25+
26+
def invoke(
27+
self,
28+
context: PipelineContext,
29+
input_config: ModelPreparationInputConfig,
30+
) -> ModelPreparationOutputConfig:
31+
"""Prepare the model, tokenizer, and calibration data.
32+
33+
Args:
34+
context: The pipeline context.
35+
input_config: The model preparation input configuration.
36+
37+
Returns:
38+
ModelPreparationOutputConfig with model, tokenizer, and calibration data.
39+
"""
40+
raise NotImplementedError(
41+
"ExecuTorchModelPreparationStrategy.invoke() is a stub. "
42+
"Implementation will be added in a subsequent PR."
43+
)

0 commit comments

Comments
 (0)