diff --git a/docs/src/index.md b/docs/src/index.md
index 608cfc4..cb59743 100644
--- a/docs/src/index.md
+++ b/docs/src/index.md
@@ -42,6 +42,7 @@ utils/mixed_precision
utils/activation_comparison
utils/casting
utils/coreai_compression
+utils/composite_op_quantization
```
```{toctree}
diff --git a/docs/src/quantization/overview.md b/docs/src/quantization/overview.md
index 45286f2..753a539 100644
--- a/docs/src/quantization/overview.md
+++ b/docs/src/quantization/overview.md
@@ -204,7 +204,7 @@ The two modes are expected to produce very similar models for weight-only quanti
A few scenarios where `eager` mode may need to be used instead of `graph`:
- If you run into any errors during the `prepare` call which, under the hood, invokes the `torch.export.export` and `torchao`'s `prepare_qat_pt2e`/`convert_pt2e` APIs. See [Graph Mode Troubleshooting](../debugging/graph_mode_troubleshooting.md) for common export errors and workarounds before falling back to eager mode.
-- When `torch.nn.Module` needs to be provided as an input, instead of `ExportedProgram` to the conversion API of [coreai-torch](https://github.com/apple/coreai-torch). This happens when the `coreai-torch` conversion needs to "externalize" certain sub-modules to map them to _composite ops_ for better runtime performance.
+- When `torch.nn.Module` needs to be provided as an input, instead of `ExportedProgram` to the conversion API of [coreai-torch](https://github.com/apple/coreai-torch). Note that models whose submodules must be "externalized" to map them to _composite ops_ for better runtime performance can still be quantized in graph mode. See [Quantizing Models with Core AI Composite Ops in Graph Mode](../utils/composite_op_quantization.md).
#### Weights and activations quantization
diff --git a/docs/src/utils/composite_op_quantization.md b/docs/src/utils/composite_op_quantization.md
new file mode 100644
index 0000000..5978792
--- /dev/null
+++ b/docs/src/utils/composite_op_quantization.md
@@ -0,0 +1,251 @@
+# Quantizing Models with Core AI Composite Ops in Graph Mode
+
+Core AI recognizes certain well-known building blocks, such as SDPA or RMSNorm, as _composite ops_ and applies optimized implementations for them.
+`coreai-torch` establishes those boundaries through _externalization_.
+Refer to the [Externalization](https://apple.github.io/coreai-torch/main/guides/externalization.html) guide for details.
+Here, we will discuss the steps required to quantize a model in `graph` mode using `coreai-opt`.
+
+`graph`-mode quantization invokes `torch.export.export` under the hood, which decomposes a submodule's `forward` into aten ops.
+In order to preserve the composite op structure during this process for externalization, the following APIs are provided:
+
+- `_patch_model_for_externalization`: Patch the model **before** `quantizer.prepare`, so that the composite op call sites survive export and all subsequent quantization passes as opaque nodes.
+- `_subexport_and_restore`: The submodule bodies of the composite ops themselves are then exported and restored before lowering to `CoreAI`.
+
+Quantization treats each composite op as opaque, i.e., no fake-quantize op is placed inside the composite body.
+The composite's input and output boundary can still be quantized, see [Quantizing the composite op boundary](#quantizing-the-composite-op-boundary) below for details.
+
+:::{warning}
+The externalization APIs used below, `_patch_model_for_externalization` and `_subexport_and_restore` in `coreai-torch` are currently experimental.
+:::
+
+```mermaid
+---
+title: "Graph mode Quantization Workflow with Externalization"
+---
+flowchart LR
+ model["Full Precision
Model"] --> patch["Patch Model for
Externalization"]
+ patch --> prepare["Prepare and
Calibrate"]
+ prepare --> qfin["Finalize and
Export"]
+ qfin --> sub["Sub-export
and Restore"]
+ sub --> convert["Convert to
Core AI"]
+ style model fill:#f9f9f9,stroke:#999
+ style patch fill:#e8f0fe,stroke:#4285f4
+ style sub fill:#e8f0fe,stroke:#4285f4
+```
+
+## Step 1: Patch the model before prepare
+
+`_patch_model_for_externalization` replaces the `forward` of every matching submodule in the model with a `torch.library.custom_op`, in place.
+Call it before constructing the `Quantizer`.
+The example below demonstrates this using the same `RMSNormComposite` op from the [Externalization](https://apple.github.io/coreai-torch/main/guides/externalization.html) guide, however, the same process applies for all composite ops with their respective `ExternalizeSpec`s.
+
+```python
+import torch
+import torch.nn as nn
+from coreai_torch import ExternalizeSpec, _patch_model_for_externalization
+
+
+# The composite op
+class RMSNormComposite(nn.Module):
+ def __init__(self, axes=-1, eps=1e-5, version=1):
+ super().__init__()
+ self.axes = axes
+ self.eps = eps
+ self.version = version
+
+ def forward(self, input: torch.Tensor, scale: torch.Tensor) -> torch.Tensor:
+ x_f32 = input.to(torch.float32)
+ inv_rms = torch.rsqrt((x_f32 * x_f32).mean(self.axes, keepdim=True) + self.eps)
+ return (input * inv_rms).to(input.dtype) * scale
+
+
+# A model that uses the composite op
+class Model(nn.Module):
+ def __init__(self, dim=32):
+ super().__init__()
+ self.proj = nn.Linear(dim, dim)
+ self.norm = RMSNormComposite()
+ self.norm_weight = nn.Parameter(torch.ones(dim))
+ self.out = nn.Linear(dim, dim)
+
+ def forward(self, x):
+ return self.out(self.norm(self.proj(x), self.norm_weight))
+
+
+model = Model().eval()
+example_inputs = (torch.randn(1, 32),)
+
+# Patch the model in-place
+# to externalize the RMSNormComposite
+_patch_model_for_externalization(
+ model,
+ targets=[
+ ExternalizeSpec(
+ target_class=RMSNormComposite,
+ composite_op_name="rms_norm",
+ composite_attrs=["axes", "eps", "version"],
+ )
+ ],
+)
+```
+
+## Step 2: Prepare, calibrate and finalize
+
+Nothing about the quantizer configuration or the calibration workflow changes.
+The composite op holds no weights of its own here, so weight quantization applies to the surrounding `Linear` layers only.
+
+```python
+import coreai_opt as opt
+from coreai_opt.quantization import ModuleQuantizerConfig, Quantizer, QuantizerConfig
+from coreai_opt.quantization.spec import (
+ default_activation_quantization_spec,
+ default_weight_quantization_spec,
+)
+
+global_config = ModuleQuantizerConfig(
+ op_state_spec={"weight": default_weight_quantization_spec()},
+ op_input_spec={"*": default_activation_quantization_spec()},
+ op_output_spec={"*": default_activation_quantization_spec()},
+)
+quant_config = QuantizerConfig(global_config=global_config)
+
+quantizer = Quantizer(model, quant_config)
+prepared_model = quantizer.prepare(example_inputs)
+
+with quantizer.calibration_mode():
+ for batch in calibration_dataloader:
+ prepared_model(batch)
+
+final_model = quantizer.finalize(backend=opt.ExportBackend.CoreAI)
+```
+
+## Step 3: Export and convert to Core AI
+
+After quantization is complete and the model is finalized, `_subexport_and_restore` API exports each patched composite op and restores the original `forward` method in the model.
+Note that the first argument to `_subexport_and_restore` is the original module that was patched in Step 1, not the finalized `GraphModule`.
+
+```python
+import coreai_torch
+from coreai_torch import TorchConverter, _subexport_and_restore
+
+exported_program = torch.export.export(final_model, example_inputs).run_decompositions(
+ coreai_torch.get_decomp_table()
+)
+externalized = _subexport_and_restore(model, exported_program)
+
+coreai_program = (
+ TorchConverter()
+ .add_exported_program(
+ exported_program, _externalized_exported_programs=externalized
+ )
+ .to_coreai()
+)
+```
+
+In the Core AI graph, the composite op is emitted as a separate private graph that `@main` reaches through `coreai.invoke`:
+
+```text
+// composite op body
+coreai.graph private noinline @norm_57e2d4a8(%arg0: tensor<1x32xf32> {coreai.name = "input"}, %arg1: tensor<32xf32> {coreai.name = "scale"}) -> (tensor<1x32xf32>) attributes {composite_decl = ...} {
+ %2 = coreai.decomposable.broadcasting_mul %0, %1 : (tensor<1x32xf32>, tensor<1x32xf32>) -> tensor<1x32xf32>
+ %4 = coreai.reduce_mean %2, %3 : (tensor<1x32xf32>, tensor<1xsi32>) -> tensor<1x1xf32>
+ %8 = coreai.decomposable.broadcasting_add %6, %7 : (tensor<1x1xf32>, tensor) -> tensor<1x1xf32>
+ %9 = coreai.rsqrt %8 : tensor<1x1xf32> -> tensor<1x1xf32>
+ %12 = coreai.decomposable.broadcasting_mul %10, %11 : (tensor<1x32xf32>, tensor<1x1xf32>) -> tensor<1x32xf32>
+ %15 = coreai.decomposable.broadcasting_mul %13, %14 : (tensor<1x32xf32>, tensor<32xf32>) -> tensor<1x32xf32>
+ coreai.output %15 : tensor<1x32xf32>
+}
+
+coreai.graph @main(%arg0: tensor<1x32xf32> {coreai.name = "x"}) -> (tensor<1x32xf32>) {
+ %44 = coreai.decomposable.broadcasting_add %43, %2 : (tensor<1x32xf32>, tensor<32xf32>) -> tensor<1x32xf32>
+ %53 = coreai.quantize %44, ... : (tensor<1x32xf32>, ...) -> tensor<1x32xsi8>
+ %62 = coreai.dequantize %53, ... : (tensor<1x32xsi8>, ...) -> tensor<1x32xf32>
+
+ // externalized composite op invocation
+ %63 = coreai.invoke @norm_57e2d4a8(%62, %0) : (tensor<1x32xf32>, tensor<32xf32>) -> tensor<1x32xf32>
+ %72 = coreai.quantize %63, ... : (tensor<1x32xf32>, ...) -> tensor<1x32xsi8>
+ %81 = coreai.dequantize %72, ... : (tensor<1x32xsi8>, ...) -> tensor<1x32xf32>
+ %84 = coreai.decomposable.broadcasting_batch_matmul %81, %83 : (tensor<1x32xf32>, tensor<32x32xf32>) -> tensor<1x32xf32>
+}
+```
+
+(`coreai.cast`, `coreai.constant` and `coreai.reshape` ops omitted above for brevity.)
+
+The composite body carries no `coreai.quantize` or `coreai.dequantize` op and stays in full precision.
+
+## Quantizing the composite op boundary
+
+The `coreai.quantize` pairs surrounding the `coreai.invoke` above come from the global config. They are the output quantizer of the preceding `Linear` and the input quantizer of the following one.
+The composite op boundary itself is not targeted by the global config.
+
+To target the boundary specifically, use `module_input_spec` and `module_output_spec` on a {class}`~coreai_opt.quantization.config.ModuleQuantizerConfig` scoped by `module_type_configs` or `module_name_configs`.
+
+To see this in isolation, the following example uses a model with the composite op alone and specifies a module level spec to quantize it's boundary.
+
+```python
+from coreai_opt.quantization.spec import (
+ PerTensorGranularity,
+ QuantizationScheme,
+ QuantizationSpec,
+)
+
+
+class RMSNormOnly(nn.Module):
+ def __init__(self, dim=32):
+ super().__init__()
+ self.norm = RMSNormComposite()
+ self.norm_weight = nn.Parameter(torch.ones(dim))
+
+ def forward(self, x):
+ return self.norm(x, self.norm_weight)
+
+
+boundary_spec = QuantizationSpec(
+ dtype=torch.int8,
+ qscheme=QuantizationScheme.SYMMETRIC,
+ granularity=PerTensorGranularity(),
+)
+quant_config = QuantizerConfig(
+ module_type_configs={
+ RMSNormComposite: ModuleQuantizerConfig(
+ module_input_spec={"*": boundary_spec},
+ module_output_spec={"*": boundary_spec},
+ )
+ },
+)
+```
+
+Running the same patch, prepare, calibrate, finalize and convert steps as above, gives a `@main` graph containing just the boundary quantization and the composite call.
+
+```text
+coreai.graph private noinline @norm_20ea9665(%arg0: tensor<1x32xf32> {coreai.name = "input"}, %arg1: tensor<32xf32> {coreai.name = "scale"}) -> (tensor<1x32xf32>) attributes {composite_decl = ...} {
+ %2 = coreai.decomposable.broadcasting_mul %0, %1 : (tensor<1x32xf32>, tensor<1x32xf32>) -> tensor<1x32xf32>
+ %4 = coreai.reduce_mean %2, %3 : (tensor<1x32xf32>, tensor<1xsi32>) -> tensor<1x1xf32>
+ %8 = coreai.decomposable.broadcasting_add %6, %7 : (tensor<1x1xf32>, tensor) -> tensor<1x1xf32>
+ %9 = coreai.rsqrt %8 : tensor<1x1xf32> -> tensor<1x1xf32>
+ %12 = coreai.decomposable.broadcasting_mul %10, %11 : (tensor<1x32xf32>, tensor<1x1xf32>) -> tensor<1x32xf32>
+ %15 = coreai.decomposable.broadcasting_mul %13, %14 : (tensor<1x32xf32>, tensor<32xf32>) -> tensor<1x32xf32>
+ coreai.output %15 : tensor<1x32xf32>
+}
+
+coreai.graph @main(%arg0: tensor<1x32xf32> {coreai.name = "x"}) -> (tensor<1x32xf32>) {
+
+ // Input boundary quantizers for the composite op
+ %13 = coreai.quantize %arg0, ... : (tensor<1x32xf32>, ...) -> tensor<1x32xsi8>
+ %22 = coreai.dequantize %13, ... : (tensor<1x32xsi8>, ...) -> tensor<1x32xf32>
+
+ // externalized composite op invocation
+ %23 = coreai.invoke @norm_20ea9665(%22, %0) : (tensor<1x32xf32>, tensor<32xf32>) -> tensor<1x32xf32>
+
+ // Output boundary quantizers for the composite op
+ %32 = coreai.quantize %23, ... : (tensor<1x32xf32>, ...) -> tensor<1x32xsi8>
+ %41 = coreai.dequantize %32, ... : (tensor<1x32xsi8>, ...) -> tensor<1x32xf32>
+ coreai.output %41 : tensor<1x32xf32>
+}
+```
+
+(`coreai.cast`, `coreai.constant` and `coreai.reshape` ops omitted above for brevity.)
+
+## Notes
+
+- The same set of APIs and steps apply for Quantization Aware Training in `graph` mode as well.
diff --git a/pyproject.toml b/pyproject.toml
index 2a172f9..5f1d5d8 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -57,7 +57,7 @@ name = "Apple Core AI Optimization Team"
[project.optional-dependencies]
coreai = [
"coreai-core==1.0.0b2",
- "coreai-torch==0.4.1",
+ "coreai-torch>=0.4.1",
"scikit-learn>=1.7.2",
]
coreml = [
@@ -205,6 +205,12 @@ conflicts = [
],
]
[tool.uv.sources]
+# TEMPORARY: resolve coreai-torch from the module externalization API branch
+# instead of the PyPI release. That branch adds the `_patch_model_for_externalization`
+# and `_subexport_and_restore` entry points used by the externalization tests.
+# It lives on a fork; apple/coreai-torch does not carry it yet. Drop this entry
+# (and re-pin the versions above) once the work lands upstream and is released.
+coreai-torch = { git = "https://github.com/gokulkrishna98/coreai-torch.git", branch = "dev/gokul/module-externalization-api" }
torch = [
{ index = "pytorch-cpu", marker = "sys_platform != 'linux'" },
{ index = "pytorch-cu128", marker = "sys_platform == 'linux'" },
diff --git a/tests/_test_artifacts/mnist/mnist_composite_rmsnorm_pretrained_1epoch_08132026.pt b/tests/_test_artifacts/mnist/mnist_composite_rmsnorm_pretrained_1epoch_08132026.pt
new file mode 100644
index 0000000..b707e36
Binary files /dev/null and b/tests/_test_artifacts/mnist/mnist_composite_rmsnorm_pretrained_1epoch_08132026.pt differ
diff --git a/tests/conftest.py b/tests/conftest.py
index 71e4afa..a441b67 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -32,6 +32,7 @@
"tests.models.mnist",
"tests.models.resnet",
"tests.models.simple",
+ "tests.models.composite",
]
_DEFAULT_SEED: int = 42
diff --git a/tests/export/export_utils.py b/tests/export/export_utils.py
index d622324..5203702 100644
--- a/tests/export/export_utils.py
+++ b/tests/export/export_utils.py
@@ -436,10 +436,14 @@ def convert(
self,
traced_model: torch.export.ExportedProgram,
input_data: torch.Tensor,
+ externalized_model: Any = None,
**kwargs: Any,
) -> AIProgram:
_, _ = input_data, kwargs
- coreai_program = self._lower_to_coreai(traced_model)
+ coreai_program = self._lower_to_coreai(
+ traced_model,
+ externalized_model=externalized_model,
+ )
assert type(coreai_program) is AIProgram
return coreai_program
@@ -526,10 +530,25 @@ def _verify_custom_ops_in_torch_program(
@staticmethod
def _lower_to_coreai(
exported_program: torch.export.ExportedProgram,
+ externalized_model: Any = None,
) -> AIProgram:
- """Lower exported program to Core AI."""
+ """Lower exported program to Core AI.
+
+ Args:
+ exported_program: The exported program to lower.
+ externalized_model: Optional ``torch.nn.Module`` that was marked in
+ place by ``coreai_torch._patch_model_for_externalization``.
+ """
converter = coreai_torch.TorchConverter()
- converter.add_exported_program(exported_program)
+ externalized_exported_programs = (
+ coreai_torch._subexport_and_restore(externalized_model, exported_program)
+ if externalized_model is not None
+ else None
+ )
+ converter.add_exported_program(
+ exported_program,
+ _externalized_exported_programs=externalized_exported_programs,
+ )
return converter.to_coreai()
@@ -554,6 +573,7 @@ def convert_and_verify(
expected_ops: Mapping[str, int],
export_backend: ExportBackend,
prepared_model_output: torch.Tensor | tuple[torch.Tensor, ...],
+ externalized_model: torch.nn.Module | None = None,
snr_thresh: float = 20.0,
psnr_thresh: float = 22.0,
skip_finalized_model_verify: bool = False,
@@ -568,6 +588,9 @@ def convert_and_verify(
export_backend: Target inference stack (CoreML or CoreAI)
prepared_model_output: Pre-computed reference output from the prepared
PyTorch model (single tensor or tuple).
+ externalized_model: Optional ``torch.nn.Module`` that was patched in place by
+ ``coreai_torch._patch_model_for_externalization``. Only supported by the
+ CoreAI backend.
snr_thresh: Minimum acceptable SNR value
psnr_thresh: Minimum acceptable PSNR value
skip_finalized_model_verify: If True, skip forward pass verification on
@@ -578,9 +601,20 @@ def convert_and_verify(
Returns:
The converted model in the specified format
+ Raises:
+ ValueError: If externalized_model is given for a non-CoreAI backend.
+
"""
converter = create_converter(export_backend)
+ if externalized_model is not None:
+ if export_backend is not ExportBackend.CoreAI:
+ msg = (
+ f"externalized_model is only supported by the CoreAI backend, got {export_backend}"
+ )
+ raise ValueError(msg)
+ converter_kwargs["externalized_model"] = externalized_model
+
# Run finalized model forward pass BEFORE tracing. torch.export.export()
# (called in trace) may mutate the model (e.g., strip parametrizations on
# older PyTorch versions), so the forward pass must happen first.
diff --git a/tests/export/test_graph_mode_mlir_export.py b/tests/export/test_graph_mode_mlir_export.py
index 6d6ea7a..7de0585 100644
--- a/tests/export/test_graph_mode_mlir_export.py
+++ b/tests/export/test_graph_mode_mlir_export.py
@@ -10,6 +10,7 @@
import pytest
import torch
+from coreai_torch import ExternalizeSpec, _patch_model_for_externalization
from coreai_opt import ExportBackend
from coreai_opt.palettization.kmeans import KMeansPalettizer
@@ -28,7 +29,15 @@
from tests.fixtures.compression import ParametrizedP4A8CompressionConfigs
from tests.fixtures.fp4 import ParametrizedFP4Configs
from tests.fixtures.fp8 import ParametrizedFP8Configs
-from tests.fixtures.quantization import ParametrizedQuantConfigs
+from tests.fixtures.quantization import (
+ ParametrizedQuantConfigs,
+ make_graph_mode_module_boundary_config,
+ make_quant_config,
+)
+from tests.models.composite import (
+ CompositeSDPAModel,
+ sdpa_externalize_spec,
+)
from . import export_utils
@@ -39,6 +48,8 @@ def _run_graph_mode_mlir_export_test_ex(
config: QuantizerConfig,
expected_ops: Mapping[str, int],
model_dtype: torch.dtype | None = None,
+ calibrate: bool = False,
+ externalized_model: torch.nn.Module | None = None,
) -> None:
"""Run graph-mode Core AI export test with expanded configuration parameters.
@@ -48,6 +59,10 @@ def _run_graph_mode_mlir_export_test_ex(
config: graph-mode quantization configuration
model_dtype: Model dtype (float16, float32, bfloat16, or None for no conversion)
expected_ops: Expected operation counts in converted model
+ calibrate: If True, run one calibration pass under
+ ``quantizer.calibration_mode()`` before the reference forward.
+ externalized_model: The model patched in place by
+ ``coreai_torch._patch_model_for_externalization``.
"""
if model_dtype is not None:
model = model.to(dtype=model_dtype)
@@ -57,6 +72,10 @@ def _run_graph_mode_mlir_export_test_ex(
quantizer = Quantizer(model, config)
prepared_model = quantizer.prepare((input_data,))
+ if calibrate:
+ with quantizer.calibration_mode(), torch.no_grad():
+ prepared_model(input_data)
+
with torch.no_grad():
prepared_model_output = prepared_model(input_data)
@@ -68,6 +87,7 @@ def _run_graph_mode_mlir_export_test_ex(
expected_ops=expected_ops,
export_backend=ExportBackend.CoreAI,
prepared_model_output=prepared_model_output,
+ externalized_model=externalized_model,
)
@@ -438,3 +458,75 @@ def test_integer_quant_minval_export(
"dequantize": 4 if has_activation_quant else 0,
},
)
+
+
+# Composite-op externalize export coverage
+
+
+@pytest.mark.parametrize("config_kind", ["w8", "w8a8", "w8a8-boundary"])
+@pytest.mark.parametrize(
+ # (model, externalize spec, composite submodule path, expected coreai.quantize
+ # count per config kind)
+ "model_cls, externalize_spec, composite_module, expected_quantize_counts",
+ [
+ pytest.param(
+ CompositeSDPAModel,
+ sdpa_externalize_spec(),
+ "composite",
+ {"w8": 0, "w8a8": 4, "w8a8-boundary": 8},
+ id="sdpa",
+ ),
+ ],
+)
+def test_composite_externalize_export(
+ model_cls: type[torch.nn.Module],
+ externalize_spec: ExternalizeSpec,
+ composite_module: str,
+ expected_quantize_counts: Mapping[str, int],
+ config_kind: str,
+) -> None:
+ """End-to-end CoreAI export of a model with an externalized composite op.
+
+ Marks the composite op for externalization, runs graph-mode PTQ, then lowers the
+ finalized graph to a .aimodel and runs it. ``convert_and_verify`` handles SNR /
+ PSNR on the runtime output and op-count verification on the exported program
+ (``constexpr_blockwise_shift_scale`` for weight quantizers and ``quantize`` /
+ ``dequantize`` for activation quantizers).
+
+ Three configs:
+
+ - ``w8`` / ``w8a8``: global config only.
+ - ``w8a8-boundary``: adds a module-scoped ``module_input_spec`` /
+ ``module_output_spec`` so the composite op's i/o edges are quantized
+ """
+ model = model_cls().eval().half()
+ input_data = torch.randn(2, 4, 32, dtype=torch.float16)
+
+ _patch_model_for_externalization(model, [externalize_spec])
+
+ if config_kind == "w8a8-boundary":
+ # uint8 boundary edges stay distinguishable from the int8 global ones.
+ config = make_graph_mode_module_boundary_config(
+ module_boundary_dtype=torch.uint8,
+ module_name=composite_module,
+ )
+ else:
+ config = make_quant_config(
+ weight_dtype=torch.int8,
+ act_dtype=torch.int8 if config_kind == "w8a8" else None,
+ execution_mode="graph",
+ )
+
+ expected_quantize_count = expected_quantize_counts[config_kind]
+ _run_graph_mode_mlir_export_test_ex(
+ model=model,
+ input_data=input_data,
+ config=config,
+ expected_ops={
+ "constexpr_blockwise_shift_scale": 2,
+ "quantize": expected_quantize_count,
+ "dequantize": expected_quantize_count,
+ },
+ calibrate=config_kind != "w8",
+ externalized_model=model,
+ )
diff --git a/tests/fixtures/quantization.py b/tests/fixtures/quantization.py
index 4ee165b..84ea8f7 100644
--- a/tests/fixtures/quantization.py
+++ b/tests/fixtures/quantization.py
@@ -81,6 +81,65 @@ def _spec(dtype: torch.dtype | str) -> QuantizationSpec:
)
+def make_graph_mode_module_boundary_config(
+ *,
+ module_boundary_dtype: torch.dtype,
+ module_name: str | None = None,
+ module_type: type | None = None,
+ module_input_spec: dict | None = None,
+ global_dtype: torch.dtype = torch.int8,
+) -> QuantizerConfig:
+ """Build a graph-mode config that also quantizes one module's own i/o boundary.
+
+ The global part comes from ``make_quant_config``; this adds a module-scoped
+ ``module_input_spec`` / ``module_output_spec`` on top of it.
+
+ Args:
+ module_boundary_dtype: Activation dtype for the boundary spec.
+ module_name: Target the module at this path (``module_name_configs``).
+ module_type: Target modules of this type (``module_type_configs``).
+ Exactly one of module_name / module_type must be given.
+ module_input_spec: Override the boundary input spec, e.g.
+ ``{0: spec, 2: spec}`` to select individual positional args.
+ Defaults to the ``"*"`` wildcard over every boundary input.
+ global_dtype: Weight and activation dtype for the global config.
+
+ Returns:
+ QuantizerConfig: the global config plus a module-scoped boundary spec.
+
+ Raises:
+ ValueError: If not exactly one of module_name / module_type is provided.
+ """
+ if (module_name is None) == (module_type is None):
+ msg = "pass exactly one of module_name / module_type"
+ raise ValueError(msg)
+
+ def _boundary_spec() -> QuantizationSpec:
+ return QuantizationSpec(
+ dtype=module_boundary_dtype,
+ qscheme=QuantizationScheme.SYMMETRIC,
+ granularity=PerTensorGranularity(),
+ )
+
+ boundary_config = ModuleQuantizerConfig(
+ module_input_spec=module_input_spec or {"*": _boundary_spec()},
+ module_output_spec={"*": _boundary_spec()},
+ )
+ scope = (
+ {"module_name_configs": {module_name: boundary_config}}
+ if module_name is not None
+ else {"module_type_configs": {module_type: boundary_config}}
+ )
+ base = make_quant_config(
+ weight_dtype=global_dtype, act_dtype=global_dtype, execution_mode="graph"
+ )
+ return QuantizerConfig(
+ global_config=base.global_config,
+ execution_mode="graph",
+ **scope,
+ )
+
+
@dataclass
class ParametrizedQuantConfigs:
"""Container for parametrized Eager and PT2E quantization configs.
diff --git a/tests/models/composite.py b/tests/models/composite.py
new file mode 100644
index 0000000..2bb8985
--- /dev/null
+++ b/tests/models/composite.py
@@ -0,0 +1,132 @@
+# Copyright 2026 Apple Inc.
+#
+# Use of this source code is governed by a BSD-3-Clause license that can
+# be found in the LICENSE file or at https://opensource.org/licenses/BSD-3-Clause
+
+"""Test models with composite ops, for externalization test coverage."""
+
+from __future__ import annotations
+
+import pytest
+import torch
+import torch.nn as nn
+
+from tests.utils import test_artifact_path
+
+# Externalize specs shared by the externalization test modules.
+
+
+def rmsnorm_externalize_spec():
+ """ExternalizeSpec targeting the RMSNormImpl composite."""
+ from coreai_torch import ExternalizeSpec # noqa: PLC0415
+ from coreai_torch.composite_ops import RMSNormImpl # noqa: PLC0415
+
+ return ExternalizeSpec(
+ target_class=RMSNormImpl,
+ composite_op_name="rms_norm",
+ composite_attrs=["axes", "eps"],
+ )
+
+
+def sdpa_externalize_spec():
+ """ExternalizeSpec targeting the SDPA composite."""
+ from coreai_torch import ExternalizeSpec # noqa: PLC0415
+ from coreai_torch.composite_ops import SDPA # noqa: PLC0415
+
+ return ExternalizeSpec(
+ target_class=SDPA,
+ composite_op_name="scaled_dot_product_attention",
+ composite_attrs=["scale", "is_causal", "window_size"],
+ )
+
+
+class MNISTCompositeRMSNormModel(nn.Module):
+ """Tiny MNIST classifier with an embedded RMSNormImpl composite op.
+
+ Architecture: Flatten -> Linear(28*28, 128) -> RMSNormImpl ->
+ ReLU -> Linear(128, 10) -> LogSoftmax.
+ """
+
+ def __init__(
+ self,
+ hidden: int = 128,
+ num_classes: int = 10,
+ eps: float = 1e-5,
+ ) -> None:
+ from coreai_torch.composite_ops import RMSNormImpl # noqa: PLC0415
+
+ super().__init__()
+ self.flatten = nn.Flatten()
+ self.fc1 = nn.Linear(28 * 28, hidden)
+ self.norm = RMSNormImpl(eps=eps)
+ self.scale = nn.Parameter(torch.ones(hidden))
+ self.relu = nn.ReLU()
+ self.fc2 = nn.Linear(hidden, num_classes)
+ self.softmax = nn.LogSoftmax(dim=-1)
+
+ def forward(self, x: torch.Tensor) -> torch.Tensor:
+ x = self.flatten(x)
+ x = self.fc1(x)
+ x = self.norm(x, self.scale)
+ x = self.relu(x)
+ x = self.fc2(x)
+ return self.softmax(x)
+
+
+# Function scoped so each test gets an independently mutable model.
+@pytest.fixture(scope="function")
+def mnist_composite_rmsnorm_pretrained_model() -> MNISTCompositeRMSNormModel:
+ """Load the committed 1-epoch MNISTCompositeRMSNormModel checkpoint.
+
+ Trained with seed 42 for one epoch over the MNIST train split: Adam,
+ lr=1e-3, batch_size=128, shuffle=False, nll_loss.
+ """
+ model = MNISTCompositeRMSNormModel()
+ model.load_state_dict(
+ torch.load(
+ test_artifact_path("mnist/mnist_composite_rmsnorm_pretrained_1epoch_08132026.pt")
+ )
+ )
+ return model
+
+
+@pytest.fixture
+def mnist_composite_rmsnorm_example_input() -> torch.Tensor:
+ """A canonical example-input tensor matching MNIST shape, for prepare()."""
+ return torch.ones(1, 1, 28, 28, dtype=torch.float32)
+
+
+class CompositeRMSNormOnlyModel(nn.Module):
+ """A single RMSNormImpl composite and nothing else."""
+
+ def __init__(self, dim: int = 32, eps: float = 1e-5) -> None:
+ from coreai_torch.composite_ops import RMSNormImpl # noqa: PLC0415
+
+ super().__init__()
+ self.norm = RMSNormImpl(eps=eps)
+ self.scale = nn.Parameter(torch.ones(dim))
+
+ def forward(self, x: torch.Tensor) -> torch.Tensor:
+ return self.norm(x, self.scale)
+
+
+class CompositeSDPAModel(nn.Module):
+ """proj -> SDPA(composite) -> output_proj, fp16, single-head fake-dim."""
+
+ def __init__(self, dim: int = 32) -> None:
+ from coreai_torch.composite_ops import SDPA # noqa: PLC0415
+
+ super().__init__()
+ self.qkv = nn.Linear(dim, dim * 3, bias=False)
+ self.composite = SDPA(scale=None, is_causal=True)
+ self.out = nn.Linear(dim, dim, bias=False)
+
+ def forward(self, x: torch.Tensor) -> torch.Tensor:
+ qkv = self.qkv(x)
+ q, k, v = qkv.chunk(3, dim=-1)
+ # Insert a single fake head dim so SDPA sees rank-4 inputs.
+ q = q.unsqueeze(1)
+ k = k.unsqueeze(1)
+ v = v.unsqueeze(1)
+ attn = self.composite(q, k, v).squeeze(1)
+ return self.out(attn)
diff --git a/tests/quantization/test_composite_op_externalize.py b/tests/quantization/test_composite_op_externalize.py
new file mode 100644
index 0000000..87c4583
--- /dev/null
+++ b/tests/quantization/test_composite_op_externalize.py
@@ -0,0 +1,296 @@
+# Copyright 2026 Apple Inc.
+#
+# Use of this source code is governed by a BSD-3-Clause license that can
+# be found in the LICENSE file or at https://opensource.org/licenses/BSD-3-Clause
+
+"""Externalize-specific structural tests for _patch_model_for_externalization
+in presence of coreai-opt graph mode quantization.
+
+Test structural assertions: after ``_patch_model_for_externalization``
+patches a composite submodule's forward into a ``torch.library.custom_op``,
+the resulting opaque call_function node survives Graph-mode ``prepare`` + ``finalize``.
+"""
+
+from __future__ import annotations
+
+import pytest
+import torch
+import torch.nn as nn
+from coreai_torch import ExternalizeSpec, _patch_model_for_externalization
+
+from coreai_opt import ExportBackend
+from coreai_opt.quantization import (
+ Quantizer,
+ QuantizerConfig,
+)
+from coreai_opt.quantization.spec import (
+ PerTensorGranularity,
+ QuantizationScheme,
+ QuantizationSpec,
+)
+from tests.fixtures.quantization import (
+ make_graph_mode_module_boundary_config,
+ make_quant_config,
+)
+from tests.models.composite import (
+ CompositeRMSNormOnlyModel,
+ CompositeSDPAModel,
+ rmsnorm_externalize_spec,
+ sdpa_externalize_spec,
+)
+from tests.test_utils.general import (
+ assert_single_call_function_node,
+ get_quantize_dtype,
+ is_coreai_dequantize,
+ is_coreai_quantize,
+)
+
+
+@pytest.mark.parametrize(
+ "quantize_activations",
+ [
+ pytest.param(False, id="w8-weight-only"),
+ pytest.param(True, id="w8a8"),
+ ],
+)
+def test_composite_op_survives_prepare_and_finalize(
+ quantize_activations: bool,
+) -> None:
+ """The externalized composite must remain a single opaque
+ call_function node end-to-end, under both w8 and w8a8.
+ """
+ model = CompositeSDPAModel().eval().half()
+ sample = torch.randn(2, 4, 32, dtype=torch.float16)
+
+ _patch_model_for_externalization(model, [sdpa_externalize_spec()])
+ op_name = model.composite._externalize_op_name
+ target_substr = f"coreai_torch_ext.{op_name}"
+
+ quantizer = Quantizer(
+ model,
+ make_quant_config(
+ weight_dtype=torch.int8,
+ act_dtype=torch.int8 if quantize_activations else None,
+ execution_mode="graph",
+ ),
+ )
+ prepared = quantizer.prepare((sample,))
+ assert_single_call_function_node(prepared, target_substr, stage="prepared")
+
+ finalized = quantizer.finalize(backend=ExportBackend.CoreAI)
+ assert_single_call_function_node(finalized, target_substr, stage="finalized")
+
+
+# Composite op I/O boundary quantization
+
+
+class TestCompositeOpIOQuantization:
+ """Ensure externalized composite's I/O boundary can be quantized
+ via a module-level config, by name and by type.
+
+ A global config quantizes the rest of the model with the default activation
+ dtype (int8) and the composite config provides a distinct boundary dtype
+ (uint8) on the composite's edges. Module config outranks global, so the
+ composite boundary must carry the composite dtype while every other quantized
+ edge carries the global dtype.
+ """
+
+ @staticmethod
+ def _composite_act_spec(boundary_dtype: torch.dtype) -> QuantizationSpec:
+ return QuantizationSpec(
+ dtype=boundary_dtype,
+ qscheme=QuantizationScheme.SYMMETRIC,
+ granularity=PerTensorGranularity(),
+ )
+
+ @staticmethod
+ def _config(
+ spec: ExternalizeSpec,
+ module_name: str,
+ target_by: str,
+ boundary_dtype: torch.dtype,
+ module_input_spec: dict | None = None,
+ ) -> QuantizerConfig:
+ return make_graph_mode_module_boundary_config(
+ module_boundary_dtype=boundary_dtype,
+ module_name=module_name if target_by == "name" else None,
+ module_type=spec.target_class if target_by == "type" else None,
+ module_input_spec=module_input_spec,
+ )
+
+ def _quantize_with_externalization_and_verify(
+ self,
+ model: nn.Module,
+ sample: torch.Tensor,
+ spec: ExternalizeSpec,
+ module_name: str,
+ config: QuantizerConfig,
+ ) -> tuple[torch.fx.GraphModule, str]:
+ """Run graph-mode PTQ on the model and verify the composite op stays opaque.
+
+ Patches the composite for externalization, prepares, finalizes, and asserts
+ the composite is still a single opaque call_function node after each of
+ those two stages.
+
+ Returns the finalized graph and the substring identifying that node.
+ """
+ _patch_model_for_externalization(model, [spec])
+ op_name = model.get_submodule(module_name)._externalize_op_name
+ target_substr = f"coreai_torch_ext.{op_name}"
+
+ quantizer = Quantizer(model, config)
+ prepared = quantizer.prepare((sample,))
+ assert_single_call_function_node(prepared, target_substr, stage="prepared")
+
+ finalized = quantizer.finalize(backend=ExportBackend.CoreAI)
+ assert_single_call_function_node(finalized, target_substr, stage="finalized")
+ return finalized, target_substr
+
+ def _assert_boundary_quantized(
+ self,
+ finalized: torch.fx.GraphModule,
+ target_substr: str,
+ num_tensor_inputs: int,
+ boundary_dtype: torch.dtype,
+ ) -> None:
+ composite = assert_single_call_function_node(finalized, target_substr, stage="finalized")
+
+ # A composite's non-tensor captured attributes appear either as baked-in
+ # constants (SDPA's scale / is_causal / window_size) or as a get_attr arg
+ # (RMSNorm's scale), and neither is a quantized activation edge, so
+ # filter get_attr out rather than indexing fixed arg positions.
+ tensor_inputs = [
+ a for a in composite.args if isinstance(a, torch.fx.Node) and a.op != "get_attr"
+ ]
+ assert len(tensor_inputs) == num_tensor_inputs, (
+ f"Expected {num_tensor_inputs} tensor inputs to {composite.name}, "
+ f"got {[n.name for n in tensor_inputs]}"
+ )
+ for act_input in tensor_inputs:
+ assert is_coreai_dequantize(act_input.target)
+ assert get_quantize_dtype(act_input.args[0]) == boundary_dtype
+
+ users = list(composite.users)
+ assert len(users) == 1
+ assert is_coreai_quantize(users[0].target)
+ assert get_quantize_dtype(users[0]) == boundary_dtype
+
+ composite_dtype_quant = [
+ n
+ for n in finalized.graph.nodes
+ if is_coreai_quantize(n.target) and get_quantize_dtype(n) == boundary_dtype
+ ]
+ assert len(composite_dtype_quant) == num_tensor_inputs + 1
+
+ @pytest.mark.parametrize("target_by", ["name", "type"])
+ @pytest.mark.parametrize(
+ # (model class, externalize spec, submodule attribute name, tensor input count).
+ # Both models default to dim=32 and accept the same rank-3 fp16 sample.
+ "model_cls, spec, module_name, num_tensor_inputs",
+ [
+ pytest.param(
+ CompositeRMSNormOnlyModel,
+ rmsnorm_externalize_spec(),
+ "norm",
+ 1,
+ id="rmsnorm-only",
+ ),
+ pytest.param(
+ CompositeSDPAModel,
+ sdpa_externalize_spec(),
+ "composite",
+ 3,
+ id="sdpa-qkv",
+ ),
+ ],
+ )
+ def test_composite_boundary_quantized(
+ self,
+ model_cls: type[nn.Module],
+ spec: ExternalizeSpec,
+ module_name: str,
+ num_tensor_inputs: int,
+ target_by: str,
+ ) -> None:
+ # uint8 boundary edges stay distinguishable from the int8 global ones.
+ boundary_dtype = torch.uint8
+ model = model_cls().eval().half()
+ sample = torch.randn(2, 4, 32, dtype=torch.float16)
+ config = self._config(spec, module_name, target_by, boundary_dtype)
+ finalized, target_substr = self._quantize_with_externalization_and_verify(
+ model, sample, spec, module_name, config
+ )
+ self._assert_boundary_quantized(finalized, target_substr, num_tensor_inputs, boundary_dtype)
+
+ @pytest.mark.parametrize("target_by", ["name", "type"])
+ def test_composite_boundary_input_index_selects_those_args(self, target_by: str) -> None:
+ """Integer keys in ``module_input_spec`` quantize exactly those positional args
+ for composite ops.
+
+ The unselected input is left unquantized rather than falling
+ back to the global spec, because the composite is opaque to the
+ op-pattern annotator and only a module-level config reaches its edges.
+ """
+ # uint8 boundary edges stay distinguishable from the int8 global ones.
+ boundary_dtype = torch.uint8
+ quantized_indices = (0, 2)
+ num_tensor_inputs = 3
+ model = CompositeSDPAModel().eval().half()
+ sample = torch.randn(2, 4, 32, dtype=torch.float16)
+ spec = sdpa_externalize_spec()
+ config = self._config(
+ spec,
+ "composite",
+ target_by,
+ boundary_dtype,
+ module_input_spec={
+ i: self._composite_act_spec(boundary_dtype) for i in quantized_indices
+ },
+ )
+
+ finalized, target_substr = self._quantize_with_externalization_and_verify(
+ model, sample, spec, "composite", config
+ )
+
+ composite = assert_single_call_function_node(finalized, target_substr, stage="finalized")
+ tensor_inputs = [
+ a for a in composite.args if isinstance(a, torch.fx.Node) and a.op != "get_attr"
+ ]
+ assert len(tensor_inputs) == num_tensor_inputs, (
+ f"Expected {num_tensor_inputs} tensor inputs to {composite.name}, "
+ f"got {[n.name for n in tensor_inputs]}"
+ )
+
+ for index, act_input in enumerate(tensor_inputs):
+ if index in quantized_indices:
+ assert is_coreai_dequantize(act_input.target), (
+ f"input {index} was selected by module_input_spec but is not fed by "
+ f"a dequantize: {act_input.target}"
+ )
+ input_dtype = get_quantize_dtype(act_input.args[0])
+ assert input_dtype == boundary_dtype, (
+ f"input {index} was selected by module_input_spec but is quantized as "
+ f"{input_dtype}, expected the composite dtype {boundary_dtype}"
+ )
+ else:
+ assert not is_coreai_dequantize(act_input.target), (
+ f"input {index} was not selected by module_input_spec but is fed by "
+ f"a dequantize: {act_input.target}"
+ )
+
+ # module_output_spec stays the wildcard, so the composite's consumer is
+ # quantized with the composite dtype regardless of which inputs were selected.
+ consumers = list(composite.users)
+ assert len(consumers) == 1, (
+ f"expected the composite to have exactly one consumer, got "
+ f"{[n.name for n in consumers]}"
+ )
+ consumer = consumers[0]
+ assert is_coreai_quantize(consumer.target), (
+ f"the composite's consumer is not a quantize node: {consumer.target}"
+ )
+ consumer_dtype = get_quantize_dtype(consumer)
+ assert consumer_dtype == boundary_dtype, (
+ f"the composite's output is quantized as {consumer_dtype}, expected the "
+ f"composite dtype {boundary_dtype}"
+ )
diff --git a/tests/quantization/test_graph_mode_quantizer.py b/tests/quantization/test_graph_mode_quantizer.py
index 901cb24..ef95594 100644
--- a/tests/quantization/test_graph_mode_quantizer.py
+++ b/tests/quantization/test_graph_mode_quantizer.py
@@ -50,6 +50,7 @@
StaticQParamsCalculator,
)
from tests.models.simple import SimpleModel
+from tests.test_utils.general import get_fake_quant_nodes
@pytest.fixture
@@ -88,16 +89,6 @@ def weight_only_config():
)
-def get_fake_quant_nodes(model: torch.fx.GraphModule) -> list[torch.fx.Node]:
- """
- Returns list of fake quant nodes present in the input model
- """
- fake_quant_nodes = [
- node for node in model.graph.nodes if "activation_post_process" in node.name
- ]
- return fake_quant_nodes
-
-
class TestGraphModeQuantizer:
"""Test cases for GraphQuantizer class."""
@@ -1223,125 +1214,6 @@ def test_train_eval_compatible_with_context_managers(
assert prepared_model.training
-class TestCompositeOpQuantization:
- """
- Tests that models with COREAI CompositeOps can be quantized
- """
-
- class SDPAModule(torch.nn.Module):
- def forward(self, query, key, value):
- from coreai_torch.composite_ops._sdpa import ( # noqa: PLC0415
- scaled_dot_product_attention as _scaled_dot_product_attention,
- )
-
- return _scaled_dot_product_attention(query, key, value, is_causal=True)
-
- class SimpleSDPAModel(torch.nn.Module):
- def __init__(self):
- super().__init__()
- self.proj = torch.nn.Linear(16, 48)
- self.sdpa = TestCompositeOpQuantization.SDPAModule()
-
- def forward(self, x):
- qkv = self.proj(x)
- q, k, v = qkv.chunk(3, dim=-1)
- b, s, _ = q.shape
- q = q.reshape(b, 1, s, 16)
- k = k.reshape(b, 1, s, 16)
- v = v.reshape(b, 1, s, 16)
- return self.sdpa(q, k, v)
-
- @pytest.fixture
- def model(self):
- return self.SimpleSDPAModel()
-
- @pytest.fixture
- def example_input(self):
- return torch.randn(1, 4, 16)
-
- @pytest.mark.xfail(reason="tracked by coreai-torch issue #309")
- def test_composite_op_io_quantization(self, model, example_input):
- """
- Verify CompositeOps boundaries can be quantized
- """
- qspec_dict = {
- "dtype": "int8",
- "qscheme": "symmetric",
- "granularity": {"type": "per_tensor"},
- }
- config = QuantizerConfig.from_dict(
- {
- "quantization_config": {
- "global_config": {"op_state_spec": {"weight": qspec_dict}},
- "module_name_configs": {
- "sdpa": {
- "op_input_spec": None,
- "op_output_spec": None,
- "op_state_spec": None,
- "module_input_spec": {
- 0: qspec_dict,
- 1: qspec_dict,
- 2: qspec_dict,
- },
- "module_output_spec": {
- "*": qspec_dict,
- },
- }
- },
- }
- }
- )
-
- quantizer = Quantizer(model, config)
- prepared_model = quantizer.prepare((example_input,))
- assert isinstance(prepared_model, torch.fx.GraphModule)
-
- def _is_composite_op(node):
- return (
- node.op == "call_function"
- and isinstance(node.target, torch._ops.OpOverload)
- and node.target.namespace == "CompositeOps"
- )
-
- # Find the SDPA node
- sdpa_nodes = [
- n
- for n in prepared_model.graph.nodes
- if n.op == "call_function"
- and isinstance(n.target, torch._ops.OpOverload)
- and n.target._opname == "scaled_dot_product_attention"
- ]
- assert len(sdpa_nodes) == 1, f"Expected 1 SDPA node, got {len(sdpa_nodes)}"
- sdpa_node = sdpa_nodes[0]
-
- composite_op_input_nodes = [arg for arg in sdpa_node.args if _is_composite_op(arg)]
- composite_op_output_nodes = [user for user in sdpa_node.users if _is_composite_op(user)]
-
- assert len(composite_op_input_nodes) == 3, (
- f"Expected 3 custom input nodes (q/k/v), got {len(composite_op_input_nodes)}"
- )
- assert len(composite_op_output_nodes) == 1, "Expected 1 composite op output node"
-
- for node in composite_op_input_nodes + composite_op_output_nodes:
- assert "name" in node.kwargs, f"kwargs not restored for {node.name}"
- assert "op_name" in node.kwargs, f"kwargs not restored for {node.name}"
- assert node.kwargs["op_name"] == "scaled_dot_product_attention"
-
- for node in composite_op_input_nodes:
- input_node = node.args[0]
- assert "activation_post_process" in input_node.name, (
- f"Expected fake quant before {node.name} "
- f"(input={node.kwargs['name']}), "
- f"got {input_node.name} instead"
- )
-
- for node in composite_op_output_nodes:
- users = list(node.users.keys())
- assert any("activation_post_process" in u.name for u in users), (
- f"Expected fake quant after {node.name}, got users: {[u.name for u in users]}"
- )
-
-
class TestFP4MLIRExportValidation:
"""Test that FP4 export validation rejects unsupported configurations."""
diff --git a/tests/quantization/test_graph_mode_quantizer_mnist.py b/tests/quantization/test_graph_mode_quantizer_mnist.py
index adcc3ce..a23b5fc 100644
--- a/tests/quantization/test_graph_mode_quantizer_mnist.py
+++ b/tests/quantization/test_graph_mode_quantizer_mnist.py
@@ -7,6 +7,7 @@
import pytest
import torch
+from coreai_torch import _patch_model_for_externalization
import tests.utils as utils
from coreai_opt import ExportBackend
@@ -20,6 +21,9 @@
PerChannelGranularity,
PerTensorGranularity,
)
+from tests.export import export_utils
+from tests.models.composite import rmsnorm_externalize_spec
+from tests.test_utils.general import assert_single_call_function_node
image_size = 28
batch_size = 128
@@ -327,3 +331,97 @@ def test_weight_and_activation_qat_mnist(mnist_pretrained_model, mnist_dataset,
# Accuracy before and after finalize should match
assert post_qat_accuracy == finalized_accuracy
+
+
+@pytest.mark.slow
+@pytest.mark.seed
+def test_weight_and_activation_qat_mnist_with_externalized_composite(
+ mnist_composite_rmsnorm_pretrained_model,
+ mnist_composite_rmsnorm_example_input,
+ mnist_dataset,
+):
+ """QAT on an MNIST classifier with an externalized RMSNormImpl composite.
+
+ Graph-mode QAT flow: baseline accuracy -> mark
+ for externalization -> prepare -> post-prepare drop -> train under
+ ``training_mode()`` -> post-QAT recovery -> finalize -> finalized
+ accuracy matches post-QAT accuracy.
+ """
+ train_loader, test_loader = utils.setup_data_loaders(mnist_dataset, batch_size)
+
+ model = mnist_composite_rmsnorm_pretrained_model
+ accuracy = utils.eval_model(model, test_loader)
+ assert accuracy > 92.0, (
+ f"expect pretrained MNIST-composite model accuracy > 92%, got {accuracy:.2f}%"
+ )
+
+ _patch_model_for_externalization(model, [rmsnorm_externalize_spec()])
+ op_name = model.norm._externalize_op_name
+ target_substr = f"coreai_torch_ext.{op_name}"
+
+ config = QuantizerConfig(global_config=ModuleQuantizerConfig())
+ quantizer = Quantizer(model, config)
+
+ prepared_model = quantizer.prepare(
+ example_inputs=(mnist_composite_rmsnorm_example_input,),
+ )
+
+ post_prepare_accuracy = utils.eval_model(prepared_model, test_loader)
+ assert post_prepare_accuracy < 90.0, (
+ f"Expect accuracy to drop below 90% after preparation with an all ones data sample; "
+ f"got {post_prepare_accuracy:.2f}% (baseline {accuracy:.2f}%)"
+ )
+
+ optimizer = torch.optim.Adam(prepared_model.parameters(), eps=1e-3, weight_decay=1e-4)
+ with quantizer.training_mode():
+ for batch_idx, (data, target) in enumerate(train_loader):
+ utils.train_step(
+ prepared_model,
+ optimizer,
+ train_loader,
+ data,
+ target,
+ batch_idx,
+ epoch=0,
+ )
+
+ post_qat_accuracy = utils.eval_model(prepared_model, test_loader)
+ assert post_qat_accuracy > 94.0, (
+ f"Expect accuracy to climb above 94% after QAT, got {post_qat_accuracy:.2f}%"
+ )
+
+ finalized_model = quantizer.finalize(backend=ExportBackend.CoreAI)
+ finalized_accuracy = utils.eval_model(finalized_model, test_loader)
+ assert post_qat_accuracy == finalized_accuracy, (
+ f"post-QAT accuracy ({post_qat_accuracy:.2f}%) must match "
+ f"post-finalize accuracy ({finalized_accuracy:.2f}%)"
+ )
+
+ assert_single_call_function_node(finalized_model, target_substr, stage="finalized")
+
+ # Lower to .aimodel and verify the runtime output matches the
+ # finalized torch output. expected_ops pins the weight constexpr and
+ # activation q-dq node counts graph mode inserts for the
+ # flatten -> Linear -> composite -> ReLU -> Linear -> LogSoftmax graph
+ # under the default w8a8 (int8) global config:
+ # - 2 constexpr_blockwise_shift_scale: the two Linear weights.
+ # - 5 quantize / 5 dequantize: the five annotated activation edges,
+ # namely the model input, fc1's input and output, and fc2's input
+ # and output.
+ # ReLU, LogSoftmax and the composite carry no quantization annotation
+ # and contribute no q-dq nodes.
+ sample_input = mnist_composite_rmsnorm_example_input
+ with torch.no_grad():
+ prepared_for_export_output = prepared_model(sample_input)
+ export_utils.convert_and_verify(
+ finalized_model=finalized_model,
+ input_data=sample_input,
+ expected_ops={
+ "constexpr_blockwise_shift_scale": 2,
+ "quantize": 5,
+ "dequantize": 5,
+ },
+ export_backend=ExportBackend.CoreAI,
+ prepared_model_output=prepared_for_export_output,
+ externalized_model=model,
+ )
diff --git a/tests/test_utils/general.py b/tests/test_utils/general.py
index b6ba93d..6a42f42 100644
--- a/tests/test_utils/general.py
+++ b/tests/test_utils/general.py
@@ -9,6 +9,7 @@
from decimal import ROUND_FLOOR, Decimal
import torch
+from torch.ops import coreai
COREAI_AVAILABLE = importlib.util.find_spec("coreai") is not None
@@ -70,7 +71,6 @@ def compute_snr_psnr(
Returns:
Tuple of (SNR, PSNR) values
-
"""
assert len(data) == len(reference), f"Tensor length mismatch: {len(data)} vs {len(reference)}"
@@ -113,3 +113,51 @@ def verify_snr_psnr(
if snr <= snr_thresh or psnr <= psnr_thresh:
raise SNRBelowThresholdError(snr, psnr, snr_thresh, psnr_thresh, prefix)
+
+
+def get_fake_quant_nodes(model: torch.fx.GraphModule) -> list[torch.fx.Node]:
+ """Return the activation_post_process observer nodes in a prepared graph."""
+ return [node for node in model.graph.nodes if "activation_post_process" in node.name]
+
+
+def assert_single_call_function_node(
+ gm: torch.fx.GraphModule, target_substr: str, *, stage: str = ""
+) -> torch.fx.Node:
+ """Assert exactly one call_function node's target contains a substring.
+
+ Current usage: externalized composite ops get a process-unique op name (sanitized module
+ path plus a uuid4 suffix), so there is no stable target object to compare
+ against and a substring match on ``str(node.target)`` is required.
+
+ Args:
+ gm: The graph module to search
+ target_substr: Substring to match against ``str(node.target)``
+ stage: Optional pipeline stage name, used only in the error message
+
+ Returns:
+ The single matching node
+ """
+ matches = [
+ n for n in gm.graph.nodes if n.op == "call_function" and target_substr in str(n.target)
+ ]
+ where = f" in the {stage.upper()} graph" if stage else ""
+ assert len(matches) == 1, (
+ f"Expected exactly one call_function matching '{target_substr}'"
+ f"{where}; got {[n.name for n in matches]}"
+ )
+ return matches[0]
+
+
+def is_coreai_quantize(target: object) -> bool:
+ """Whether an FX node target is the ``coreai::quantize`` op."""
+ return target is coreai.quantize or target is coreai.quantize.default
+
+
+def is_coreai_dequantize(target: object) -> bool:
+ """Whether an FX node target is the ``coreai::dequantize`` op."""
+ return target is coreai.dequantize or target is coreai.dequantize.default
+
+
+def get_quantize_dtype(node: torch.fx.Node) -> torch.dtype | None:
+ """Return the quantized dtype carried in an FX node's args, else None."""
+ return next((a for a in node.args if isinstance(a, torch.dtype)), None)