Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
286 changes: 286 additions & 0 deletions test/quantization/recipes/test_export_qwen3_vl.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,286 @@
# Copyright (c) 2026 Samsung Electronics Co., Ltd. All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

try:
from quantization.recipes.optional_dependency_stubs import (
install_optional_dependency_stubs,
)
except ModuleNotFoundError:
from optional_dependency_stubs import install_optional_dependency_stubs

install_optional_dependency_stubs()

import tempfile
import unittest
from types import SimpleNamespace
from unittest.mock import patch

import tico.quantization.recipes.adapters.qwen3_vl as qwen_adapter_mod
import tico.quantization.recipes.export.qwen3_vl as qwen_export

import torch
from tico.quantization.recipes.adapters.qwen3_vl import Qwen3VLAdapter
from tico.quantization.recipes.context import RecipeContext


class FakePTQWrapper(torch.nn.Module):
"""Minimal PTQWrapper-like container."""

def __init__(self, wrapped):
super().__init__()
self.wrapped = wrapped

def forward(self, *args, **kwargs):
"""Forward to the wrapped module when a test executes the adapter."""
return self.wrapped(*args, **kwargs)


class FakePatchEmbed(torch.nn.Module):
"""Expose the patch geometry used to build vision export inputs."""

in_channels = 3
temporal_patch_size = 2
patch_size = 2

def forward(self, value):
"""Return a placeholder tensor."""
return value


class FakeDecoderLayer(torch.nn.Module):
"""Expose prefill and decode export modules."""

def as_export_module(self, mode, *, return_kv=True):
"""Return a trivial module for the requested export mode."""
del mode, return_kv
return torch.nn.Identity()


class FakeVision(torch.nn.Module):
"""Minimal fixed-grid Qwen3-VL vision wrapper."""

def __init__(self):
super().__init__()
self.register_buffer(
"vision_grid_thw",
torch.tensor([[1, 4, 4]], dtype=torch.long),
persistent=False,
)
self.spatial_merge_size = 2
self.patch_embed = FakePTQWrapper(FakePatchEmbed())
self.deepstack_merger_list = torch.nn.ModuleList([torch.nn.Identity()])


class FakeText(torch.nn.Module):
"""Minimal Qwen3-VL text wrapper hierarchy."""

def __init__(self):
super().__init__()
self.config = SimpleNamespace(
hidden_size=8,
head_dim=4,
num_attention_heads=2,
num_key_value_heads=1,
num_hidden_layers=1,
max_position_embeddings=4,
vocab_size=32,
)
self.embed_tokens = torch.nn.Embedding(32, 8)
self.layers = torch.nn.ModuleList([FakePTQWrapper(FakeDecoderLayer())])
self.norm = torch.nn.Identity()
self.rotate_embedding = None


class FakeQwenModel(torch.nn.Module):
"""Minimal multimodal wrapper hierarchy."""

def __init__(self):
super().__init__()
self.visual = FakePTQWrapper(FakeVision())
self.language_model = FakePTQWrapper(FakeText())
self.visual_start_idx = 0


class FakeTopLevelQwen(torch.nn.Module):
"""Minimal conditional-generation wrapper hierarchy."""

def __init__(self):
super().__init__()
self.model = FakePTQWrapper(FakeQwenModel())
self.lm_head = torch.nn.Linear(8, 32, bias=False)
self.rotate_lm_head = None


class FakeExportModel(torch.nn.Module):
"""Outer PTQWrapper-like model returned by prepare/convert."""

def __init__(self):
super().__init__()
self.wrapped = FakeTopLevelQwen()


def _model_args():
"""Return the fixed vision contract used by exporter tests."""
return {
"vision": {
"grid_thw": [1, 4, 4],
"visual_start_idx": 0,
"spatial_merge_size": 2,
}
}


class TestQwen3VLPerLayerExport(unittest.TestCase):
def test_exports_all_static_runtime_stages(self):
"""Qwen3-VL staged export should emit prefill, decode, and DeepStack graphs."""
calls = []
export_model = FakeExportModel()
dynamic_shapes = {"input_ids": {1: "S"}}

def fake_convert_and_save(module, example_inputs, save_path, **kwargs):
del module, example_inputs
calls.append((save_path.name, kwargs.get("dynamic_shapes")))

with tempfile.TemporaryDirectory() as tmpdir, patch.object(
qwen_export,
"_prepare_qwen3_vl_export_model",
return_value=(export_model, "q"),
), patch.object(
qwen_export,
"make_token_embedding_dynamic_shapes",
return_value=dynamic_shapes,
), patch.object(
qwen_export,
"register_fake_quant_meta_kernels_for_dynamic_export",
) as register_meta, patch.object(
qwen_export, "_convert_and_save", fake_convert_and_save
):
qwen_export.export_qwen3_vl_per_layer(
q_model=torch.nn.Identity(),
max_seq_len=4,
output_dir=tmpdir,
model_args=_model_args(),
prefill_decode=True,
)

self.assertEqual(
[name for name, _ in calls],
[
"vision_prefill.q.circle",
"token_embedding.q.circle",
"multimodal_embedding_prefill.q.circle",
"decoder_layer_prefill_0.q.circle",
"deepstack_fusion_0.q.circle",
"decoder_layer_decode_0.q.circle",
"lm_head.q.circle",
],
)
token_embedding_calls = [
shapes for name, shapes in calls if name == "token_embedding.q.circle"
]
self.assertEqual(token_embedding_calls, [dynamic_shapes])
register_meta.assert_called_once_with()

def test_prefill_only_export_uses_unsuffixed_stage_names(self):
"""Disabling decode export should omit all decode artifacts."""
names = []
export_model = FakeExportModel()

def fake_convert_and_save(module, example_inputs, save_path, **kwargs):
del module, example_inputs, kwargs
names.append(save_path.name)

with tempfile.TemporaryDirectory() as tmpdir, patch.object(
qwen_export,
"_prepare_qwen3_vl_export_model",
return_value=(export_model, "f32"),
), patch.object(qwen_export, "_convert_and_save", fake_convert_and_save):
qwen_export.export_qwen3_vl_per_layer(
q_model=torch.nn.Identity(),
max_seq_len=4,
output_dir=tmpdir,
model_args=_model_args(),
prefill_decode=False,
)

self.assertEqual(
names,
[
"vision_prefill.f32.circle",
"token_embedding.f32.circle",
"multimodal_embedding.f32.circle",
"decoder_layer_0.f32.circle",
"deepstack_fusion_0.f32.circle",
"lm_head.f32.circle",
],
)

def test_rejects_visual_span_larger_than_static_sequence(self):
"""The fixed visual span must fit inside max_seq_len."""
export_model = FakeExportModel()
args = _model_args()
args["vision"]["visual_start_idx"] = 1
export_model.wrapped.model.wrapped.visual_start_idx = 1

with tempfile.TemporaryDirectory() as tmpdir, patch.object(
qwen_export,
"_prepare_qwen3_vl_export_model",
return_value=(export_model, "q"),
):
with self.assertRaisesRegex(ValueError, "visual-token span"):
qwen_export.export_qwen3_vl_per_layer(
q_model=torch.nn.Identity(),
max_seq_len=4,
output_dir=tmpdir,
model_args=args,
)

def test_adapter_routes_circle_per_layer_artifact(self):
"""The Qwen adapter should dispatch the generic Circle artifact key."""
model = torch.nn.Identity()
ctx = RecipeContext(
cfg={
"calibration": {"seq_len": 2048},
"model_args": _model_args(),
"export": {
"enabled": True,
"output_dir": "./out/qwen",
"max_seq_len": 1024,
"prefill_decode": True,
"strict": True,
"artifacts": ["circle_per_layer"],
},
},
adapter=Qwen3VLAdapter(),
model=model,
)

with patch.object(
qwen_adapter_mod,
"export_qwen3_vl_per_layer",
) as export_per_layer:
Qwen3VLAdapter().export(ctx)

export_per_layer.assert_called_once_with(
q_model=model,
max_seq_len=1024,
output_dir=qwen_adapter_mod.Path("./out/qwen"),
model_args=_model_args(),
prefill_decode=True,
strict=True,
)


if __name__ == "__main__":
unittest.main()
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,30 @@ def test_forward_diff(self):
self.assertLess(diff, 0.1)
self.assertEqual(fp_out.shape, q_out.shape)

def test_static_attention_split_sizes_match_dynamic_cu_seqlens(self):
"""Static export split sizes should preserve eager attention results."""
split_sizes = (4, 8)
seq_len = sum(split_sizes)
cu_seqlens = torch.tensor([0, split_sizes[0], seq_len])
qattn = QuantQwen3VLVisionAttention(self.fp_attn)

x = torch.randn(seq_len, self.hidden_size)
pos = self._rand_rope(seq_len)
with torch.no_grad():
dynamic_output = qattn(
x,
cu_seqlens=cu_seqlens,
position_embeddings=pos,
)
static_output = qattn(
x,
cu_seqlens=cu_seqlens,
position_embeddings=pos,
attention_split_sizes=split_sizes,
)

torch.testing.assert_close(static_output, dynamic_output)

def test_per_projection_override(self):
cfg = make_affine_ptq_config(
dtype=DType.uint(8),
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,9 @@
from tico.quantization.config.ptq import PTQConfig
from tico.quantization.wrapq.mode import Mode
from tico.quantization.wrapq.utils.version import has_transformers_for
from tico.quantization.wrapq.wrappers.qwen_vl.export_adapters import (
Qwen3VLVisionPrefillExportAdapter,
)
from tico.quantization.wrapq.wrappers.qwen_vl.quant_vision_model import (
QuantQwen3VLVisionModel,
)
Expand Down Expand Up @@ -222,6 +225,26 @@ def test_init_with_valid_config(self):
len(q_model.deepstack_merger_list), len(self.fp_model.deepstack_merger_list)
)

def test_non_strict_export_with_temporal_grid(self):
"""Fixed temporal grids should not create data-dependent split sizes."""
grid_thw = (2, 4, 4)
ptq_config = self._make_ptq_config(grid_thw)
q_model = QuantQwen3VLVisionModel(
self.fp_model,
qcfg=ptq_config,
fp_name="test_temporal_export",
).eval()
export_module = Qwen3VLVisionPrefillExportAdapter(q_model).eval()
hidden_states, grid_tensor = self._create_test_inputs(grid_thw)

exported_program = torch.export.export(
export_module,
(hidden_states, grid_tensor),
strict=False,
)

self.assertIsNotNone(exported_program)

def test_init_missing_vision_grid_thw(self):
"""Test initialization fails without vision_grid_thw."""
ptq_config = PTQConfig()
Expand Down
31 changes: 30 additions & 1 deletion tico/quantization/examples/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ are ignored by `evaluate.py`.

Use `export.py` to export either a floating-point model loaded through its
adapter or an already saved checkpoint. It writes configured artifacts such as
LLaMA per-layer Circle files and does **not** run `pipeline` stages.
LLaMA and Qwen3-VL staged Circle files and does **not** run `pipeline` stages.

Use `inspector.py` for debug-oriented workflows such as trace, parity, runtime
inspection, and wrapper-level smoke checks.
Expand Down Expand Up @@ -284,6 +284,35 @@ python -m tico.quantization.examples.export \
--checkpoint ./out/llama_quantized/quantized_model.pt
```

Qwen3-VL uses the same generic `circle_per_layer` artifact key. The exporter
keeps the fixed-grid vision model as one prefill stage and emits each text
decoder layer separately for prefill and decode:

```bash
python -m tico.quantization.examples.export \
--config tico/quantization/examples/configs/qwen3_vl_export.yaml \
--checkpoint ./out/qwen3_vl_quantized/quantized_model.pt
```

The fixed values under `model_args.vision` must match the values used when the
checkpoint was wrapped and quantized. A default prefill/decode export writes:

```text
vision_prefill.q.circle
token_embedding.q.circle
multimodal_embedding_prefill.q.circle
decoder_layer_prefill_<index>.q.circle
deepstack_fusion_<index>.q.circle
decoder_layer_decode_<index>.q.circle
lm_head.q.circle
```

`token_embedding` has a dynamic sequence dimension and is shared by prefill
and decode. The runtime owns mRoPE lookup, additive attention-mask
construction, and KV cache storage/update. It passes those tensors to the
exported decoder-layer graphs. Exporting with `--source model` writes the
same artifact set with the `.f32.circle` suffix.

`llama_export.yaml` exports both prefill and decode layer artifacts by default.
Use an override only when decode layer export is not needed:

Expand Down
Loading
Loading