diff --git a/modelopt/torch/export/__init__.py b/modelopt/torch/export/__init__.py index 004788bbf14..9a4624a25e7 100644 --- a/modelopt/torch/export/__init__.py +++ b/modelopt/torch/export/__init__.py @@ -21,6 +21,7 @@ from .model_utils import * from .moe_utils import * from .plugins import * +from .registry import * from .transformer_engine import * from .unified_export_hf import * from .unified_export_megatron import * diff --git a/modelopt/torch/export/registry.py b/modelopt/torch/export/registry.py new file mode 100644 index 00000000000..df7bc3f4497 --- /dev/null +++ b/modelopt/torch/export/registry.py @@ -0,0 +1,148 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# 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. + +"""Registry dispatching per-module export logic for the unified HF export path. + +This mirrors the registration-and-dispatch idiom of +:class:`QuantModuleRegistry `, +but not its mechanism: quantization registers replacement classes and converts modules +in place, whereas export registers :class:`ModuleExporter` handlers that emit compressed +weights and scale buffers for a module without changing its class. + +Handlers are matched per module during the export walk. Registering a handler for a new +module type replaces what previously required editing if/elif chains inside +``unified_export_hf.py``. +""" + +from collections.abc import Callable +from dataclasses import dataclass, field + +import torch +import torch.nn as nn + +__all__ = ["ExportContext", "ExportModuleRegistry", "ModuleExporter"] + + +@dataclass +class ExportContext: + """Shared state for a single export invocation, passed to every handler call. + + The tied-weight dedup caches must be scoped to one export invocation: a + process-global cache would carry stale entries whose ``data_ptr`` keys can be + recycled by PyTorch's allocator across exports, causing silent false-positive + aliasing. ``tied_cache`` (int keys) holds dense Linear / per-expert wrapper + dedup; ``moe_tied_cache`` (tuple keys) holds MoE fused-experts module dedup. + """ + + model: nn.Module + dtype: torch.dtype + is_modelopt_qlora: bool = False + tied_cache: dict[int, nn.Module] = field(default_factory=dict) + moe_tied_cache: dict[tuple[int, int], nn.Module] = field(default_factory=dict) + + +class ModuleExporter: + """Base class for per-module export handlers. + + Subclasses are registered on :data:`ExportModuleRegistry` and dispatched during the + export walk. Both hooks default to no-ops so a handler only implements the phases + that apply to its module type. + """ + + def prepare_moe_inputs(self, name: str, moe_module: nn.Module, ctx: ExportContext) -> None: + """Fill missing expert input-quantizer amax values before fusion and compression. + + Called once per MoE block whose ``.experts`` container matched this entry. + ``moe_module`` is the MoE block itself, not the experts container. + """ + + def export(self, name: str, module: nn.Module, ctx: ExportContext) -> None: + """Emit compressed weights and scale buffers for ``module``, in place.""" + + +class _ExportModuleRegistryCls: + """Ordered, first-match-wins registry mapping modules to :class:`ModuleExporter`. + + An entry can match a module by any combination of: + + - a class key: the registered class appears in ``type(module).__mro__``, so + dynamically generated quantized classes (e.g. ``QuantLinear``) match through + their original base class; + - a class-name string key: the string equals the ``__name__`` of a class in the + MRO — for classes that cannot be imported statically (trust_remote_code models + or on-the-fly generated quantized classes); + - a predicate on the module instance, for structural detection. + + When keys and a predicate are both given, both must match. Entries are tried in + registration order and the first match wins, so more specific handlers must be + registered before generic ones. The built-in handlers end with broad structural + catch-alls (e.g. any iterable module), so external handlers should register with + ``prepend=True`` to take precedence over them. + """ + + def __init__(self) -> None: + self._entries: list[ + tuple[tuple[type | str, ...], Callable[[nn.Module], bool] | None, ModuleExporter] + ] = [] + + def register( + self, + *keys: type | str, + predicate: Callable[[nn.Module], bool] | None = None, + prepend: bool = False, + ): + """Return a decorator registering a :class:`ModuleExporter` subclass. + + Re-registering the same exporter class (e.g. on module reload) replaces its + existing entry in place instead of appending a duplicate. + + Usage:: + + @ExportModuleRegistry.register("Llama4TextExperts", "GptOssExperts") + class _BmmExpertsExporter(ModuleExporter): ... + """ + assert keys or predicate is not None, "register() requires at least one key or a predicate" + + def decorator(exporter_cls: type[ModuleExporter]) -> type[ModuleExporter]: + entry = (keys, predicate, exporter_cls()) + identity = (exporter_cls.__module__, exporter_cls.__qualname__) + for i, (_, _, existing) in enumerate(self._entries): + if (type(existing).__module__, type(existing).__qualname__) == identity: + self._entries[i] = entry + return exporter_cls + if prepend: + self._entries.insert(0, entry) + else: + self._entries.append(entry) + return exporter_cls + + return decorator + + def match(self, module: nn.Module) -> ModuleExporter | None: + """Return the first registered exporter matching ``module``, or None.""" + mro = type(module).__mro__ + mro_names = {cls.__name__ for cls in mro} + for keys, predicate, exporter in self._entries: + if keys and not any( + key in mro_names if isinstance(key, str) else key in mro for key in keys + ): + continue + if predicate is not None and not predicate(module): + continue + return exporter + return None + + +ExportModuleRegistry = _ExportModuleRegistryCls() diff --git a/modelopt/torch/export/unified_export_hf.py b/modelopt/torch/export/unified_export_hf.py index 64dfb5e12c2..5ab12d75fd7 100644 --- a/modelopt/torch/export/unified_export_hf.py +++ b/modelopt/torch/export/unified_export_hf.py @@ -112,6 +112,7 @@ sync_tied_input_amax, to_quantized_weight, ) +from .registry import ExportContext, ExportModuleRegistry, ModuleExporter __all__ = ["export_hf_checkpoint", "export_speculative_decoding"] @@ -779,6 +780,178 @@ def _export_quantized_weight( torch.cuda.empty_cache() +def _has_fused_experts_quantizers(module: nn.Module) -> bool: + first_proj_attr = getattr(module, "_first_proj_attr", "gate_up_proj") + return hasattr(module, f"{first_proj_attr}_weight_quantizers") + + +@ExportModuleRegistry.register("QuantMoELinear", predicate=lambda m: hasattr(m, "experts")) +class _MoELinearExporter(ModuleExporter): + """Step-3.5 QuantMoELinear reconstructs packed MoE tensors from child expert QuantLinears. + + Fill missing input amax here, before named_modules() reaches those children, so every + expert emits input_scale. The child QuantLinears export their own weights when the walk + reaches them; the packed tensors are rebuilt afterwards in _reconstruct_fused_moe_linear. + """ + + def export(self, name: str, module: nn.Module, ctx: ExportContext) -> None: + set_expert_quantizer_amax(list(module.experts), quantizer_attrs="input_quantizer") + + +# Keyed on the mixin class name too: the generated class is normally named +# "QuantDbrxExperts", but _DMRegistryCls falls back to a module-prefixed name on +# collision, while "_QuantDbrxExperts" is always in the generated class's MRO. +@ExportModuleRegistry.register("QuantDbrxExperts", "_QuantDbrxExperts") +class _DbrxExpertsExporter(ModuleExporter): + """DBRX experts: per-expert linears live as ModuleLists on ``experts.mlp``.""" + + def prepare_moe_inputs(self, name: str, moe_module: nn.Module, ctx: ExportContext) -> None: + experts_mlp = moe_module.experts.mlp + for linear_name in get_expert_linear_names(moe_module): + if hasattr(experts_mlp, linear_name): + linear_modulelist = getattr(experts_mlp, linear_name) + if hasattr(linear_modulelist, "__iter__"): + set_expert_quantizer_amax( + modules=list(linear_modulelist), + quantizer_attrs=["input_quantizer"], + ) + + +@ExportModuleRegistry.register(predicate=_has_fused_experts_quantizers) +class _FusedExpertsExporter(ModuleExporter): + """_QuantFusedExperts uses plural ``_weight_quantizers`` (ModuleList). + + get_quantization_format's singular-weight_quantizer check misses these, so they are + matched structurally so fused-experts get split + quantized. The input-amax fallback + is handled inside _export_fused_experts, hence no prepare_moe_inputs here. + """ + + def export(self, name: str, module: nn.Module, ctx: ExportContext) -> None: + with fsdp2_aware_weight_update(ctx.model, module, reshard=False): + _export_fused_experts( + module, + ctx.dtype, + _moe_tied_cache=ctx.moe_tied_cache, + _tied_cache=ctx.tied_cache, + ) + + +@ExportModuleRegistry.register("Llama4TextExperts", "GptOssExperts") +class _BmmExpertsExporter(ModuleExporter): + """Fused BMM-style experts: 3-D gate_up_proj/down_proj weights with singular quantizers.""" + + def prepare_moe_inputs(self, name: str, moe_module: nn.Module, ctx: ExportContext) -> None: + # Both use gate_up_proj and down_proj with singular input quantizers + # (gate_up_proj_input_quantizer/down_proj_input_quantizer); the weight-side + # amax fallback and the weight export happen in export() below. + for linear_name in ["gate_up_proj", "down_proj"]: + if hasattr(moe_module.experts, linear_name): + linear_module = getattr(moe_module.experts, linear_name) + if hasattr(linear_module, "input_quantizer"): + set_expert_quantizer_amax( + modules=[linear_module], + quantizer_attrs=["input_quantizer"], + ) + + def export(self, name: str, module: nn.Module, ctx: ExportContext) -> None: + if get_quantization_format(module) == QUANTIZATION_NONE: + return + # TODO: consolidate uncalibrated experts handling logic + # Handle weight quantizers amax values using smart fallback logic + set_expert_quantizer_amax( + modules=module, + quantizer_attrs=["gate_up_proj_weight_quantizer", "down_proj_weight_quantizer"], + ) + # Handle input quantizers amax values using smart fallback logic + set_expert_quantizer_amax( + modules=module, + quantizer_attrs=["gate_up_proj_input_quantizer", "down_proj_input_quantizer"], + ) + # Export the quantized weights + with fsdp2_aware_weight_update(ctx.model, module, reshard=False): + for weight_name in ["gate_up_proj", "down_proj"]: + _export_quantized_weight(module, ctx.dtype, weight_name, _tied_cache=ctx.tied_cache) + + +@ExportModuleRegistry.register(predicate=is_quantlinear) +class _QuantLinearExporter(ModuleExporter): + """Standard quantized linear layers (QuantLinear, QuantCompressedLinear, QuantFP8Linear).""" + + def export(self, name: str, module: nn.Module, ctx: ExportContext) -> None: + if get_quantization_format(module) == QUANTIZATION_NONE: + return + try: + with fsdp2_aware_weight_update(ctx.model, module, reshard=False): + _export_quantized_weight(module, ctx.dtype, _tied_cache=ctx.tied_cache) + except AssertionError as e: + raise AssertionError( + f"Failed to export module '{name}' (type={type(module).__name__}): {e}" + ) from e + + +@ExportModuleRegistry.register(nn.Embedding, predicate=lambda m: hasattr(m, "weight_quantizer")) +class _QuantEmbeddingExporter(ModuleExporter): + """Quantized nn.Embedding: pack the embedding table the same way as Linear weights. + + Downstream loaders then see the NVFP4/FP8/INT-packed bytes + scales. + """ + + def export(self, name: str, module: nn.Module, ctx: ExportContext) -> None: + if get_quantization_format(module) == QUANTIZATION_NONE: + return + # Skip packing when the embedding's weight is tied to another module + # (e.g. tied_word_embeddings → lm_head): _export_quantized_weight reassigns + # the .weight attribute to a new uint8 Parameter, which severs the Python- + # level tie and leaves the other module pointing at a stale float Parameter. + tied_to = [ + other_name + for other_name, other_module in ctx.model.named_modules() + if other_module is not module and getattr(other_module, "weight", None) is module.weight + ] + if tied_to: + warnings.warn( + f"Skipping quantized weight packing for embedding '{name}': its " + f"weight Parameter is shared with {tied_to} (weight tying). Packing " + "would break the tie and produce stale weights in the tied module(s). " + "The embedding will be exported as its fake-quantized float weight." + ) + return + try: + with fsdp2_aware_weight_update(ctx.model, module, reshard=False): + _export_quantized_weight(module, ctx.dtype, _tied_cache=ctx.tied_cache) + except AssertionError as e: + raise AssertionError( + f"Failed to export embedding '{name}' (type={type(module).__name__}): {e}" + ) from e + + +@ExportModuleRegistry.register(predicate=lambda m: isinstance(m, collections.abc.Iterable)) +class _IterableExpertsExporter(ModuleExporter): + """MoE blocks with iterable per-expert submodules (e.g. Mixtral's ModuleList of experts).""" + + def prepare_moe_inputs(self, name: str, moe_module: nn.Module, ctx: ExportContext) -> None: + expert_linear_names = get_expert_linear_names(moe_module) + linear_name = None + try: + for linear_name in expert_linear_names: + set_expert_quantizer_amax( + modules=[getattr(expert, linear_name) for expert in moe_module.experts], + quantizer_attrs=["input_quantizer"], + ) + except AttributeError as e: + # Provide more helpful debugging information + expert_types = [type(expert).__name__ for expert in moe_module.experts] + raise AttributeError( + f"Failed to access attribute '{linear_name}' on experts. " + f"MoE module type: {type(moe_module).__name__}, " + f"Expert types: {expert_types}, " + f"Expected linear names: {expert_linear_names}. " + f"This suggests the get_expert_linear_names function may need " + f"to be updated for this model architecture. " + f"Original error: {e}" + ) from e + + def _process_quantized_modules( model: nn.Module, dtype: torch.dtype, @@ -786,9 +959,9 @@ def _process_quantized_modules( ) -> None: """Process all quantized modules in model, export weights in-place. - This function iterates through all modules in the model and exports quantized weights - for modules that have quantization enabled. It handles both standard linear layers - and specialized expert modules (Llama4TextExperts, GptOssExperts). + This function iterates through all modules in the model and dispatches each one to + the first matching :class:`ModuleExporter` in :data:`ExportModuleRegistry`, which + exports quantized weights in place. Modules matching no handler are left untouched. Args: model: The model containing quantized modules. @@ -796,14 +969,10 @@ def _process_quantized_modules( is_modelopt_qlora: Whether the model is a modelopt-trained QLoRA model. If True, modules with base_layer attribute are skipped. """ - # Per-call tied-weight dedup caches. Created fresh on every invocation - # so cache state is scoped to one export and cannot leak into a later - # call (a process-global cache would carry stale entries whose data_ptr - # keys can be recycled by PyTorch's allocator across exports — silent - # false-positive aliasing). int keys hold dense Linear / per-expert - # wrapper dedup; tuple keys hold MoE fused-experts module dedup. - _tied_cache: dict[int, nn.Module] = {} - _moe_tied_cache: dict[tuple[int, int], nn.Module] = {} + # Per-call tied-weight dedup caches inside the context. Created fresh on + # every invocation so cache state is scoped to one export and cannot leak + # into a later call (see ExportContext). + ctx = ExportContext(model=model, dtype=dtype, is_modelopt_qlora=is_modelopt_qlora) fsdp_module_to_reshard = None for name, sub_module in model.named_modules(): @@ -818,97 +987,19 @@ def _process_quantized_modules( fsdp_module_to_reshard = sub_module # We skip QuantLoraLinear module for modelopt QLoRA - if is_modelopt_qlora and (hasattr(sub_module, "base_layer")): - continue - - # Step-3.5 QuantMoELinear reconstructs packed MoE tensors from child - # expert QuantLinears after export. Fill missing input amax here, before - # named_modules() reaches those children, so every expert emits input_scale. - if type(sub_module).__name__ == "QuantMoELinear" and hasattr(sub_module, "experts"): - set_expert_quantizer_amax(list(sub_module.experts), quantizer_attrs="input_quantizer") + if ctx.is_modelopt_qlora and hasattr(sub_module, "base_layer"): continue # Preprocessing: restore unpacked weight so the export path can read - # the live quantizer state. Falls through to the export branches below. + # the live quantizer state. Falls through to the handler dispatch below. if hasattr(sub_module, "weight_packed") or ( "QuantFP8Linear" in type(sub_module).__name__ and sub_module.weight.element_size() <= 1 ): sub_module.unpack_weight() - first_proj_attr = getattr(sub_module, "_first_proj_attr", "gate_up_proj") - if hasattr(sub_module, f"{first_proj_attr}_weight_quantizers"): - # _QuantFusedExperts uses plural `_weight_quantizers` - # (ModuleList), which get_quantization_format's singular-weight_quantizer - # check misses. Handle it explicitly before the format gate so fused-experts - # get split + quantized. - with fsdp2_aware_weight_update(model, sub_module, reshard=False): - _export_fused_experts( - sub_module, - dtype, - _moe_tied_cache=_moe_tied_cache, - _tied_cache=_tied_cache, - ) - elif get_quantization_format(sub_module) != QUANTIZATION_NONE: - # Skip QuantMoELinear - it's handled separately in _reconstruct_fused_moe_linear - if type(sub_module).__name__ == "QuantMoELinear": - continue - if is_quantlinear(sub_module): - try: - with fsdp2_aware_weight_update(model, sub_module, reshard=False): - _export_quantized_weight(sub_module, dtype, _tied_cache=_tied_cache) - except AssertionError as e: - raise AssertionError( - f"Failed to export module '{name}' (type={type(sub_module).__name__}): {e}" - ) from e - elif isinstance(sub_module, nn.Embedding) and hasattr(sub_module, "weight_quantizer"): - # Quantized nn.Embedding: pack the embedding table the same way as Linear - # weights so downstream loaders see the NVFP4/FP8/INT-packed bytes + scales. - # Skip packing when the embedding's weight is tied to another module - # (e.g. tied_word_embeddings → lm_head): _export_quantized_weight reassigns - # the .weight attribute to a new uint8 Parameter, which severs the Python- - # level tie and leaves the other module pointing at a stale float Parameter. - tied_to = [ - other_name - for other_name, other_module in model.named_modules() - if other_module is not sub_module - and getattr(other_module, "weight", None) is sub_module.weight - ] - if tied_to: - warnings.warn( - f"Skipping quantized weight packing for embedding '{name}': its " - f"weight Parameter is shared with {tied_to} (weight tying). Packing " - "would break the tie and produce stale weights in the tied module(s). " - "The embedding will be exported as its fake-quantized float weight." - ) - else: - try: - with fsdp2_aware_weight_update(model, sub_module, reshard=False): - _export_quantized_weight(sub_module, dtype, _tied_cache=_tied_cache) - except AssertionError as e: - raise AssertionError( - f"Failed to export embedding '{name}' (type={type(sub_module).__name__}): {e}" - ) from e - elif ( - "Llama4TextExperts" in type(sub_module).__name__ - or "GptOssExperts" in type(sub_module).__name__ - ): - # TODO: consolidate uncalibrated experts handling logic - # Handle weight quantizers amax values using smart fallback logic - set_expert_quantizer_amax( - modules=sub_module, - quantizer_attrs=["gate_up_proj_weight_quantizer", "down_proj_weight_quantizer"], - ) - # Handle input quantizers amax values using smart fallback logic - set_expert_quantizer_amax( - modules=sub_module, - quantizer_attrs=["gate_up_proj_input_quantizer", "down_proj_input_quantizer"], - ) - # Export the quantized weights - with fsdp2_aware_weight_update(model, sub_module, reshard=False): - for weight_name in ["gate_up_proj", "down_proj"]: - _export_quantized_weight( - sub_module, dtype, weight_name, _tied_cache=_tied_cache - ) + exporter = ExportModuleRegistry.match(sub_module) + if exporter is not None: + exporter.export(name, sub_module, ctx) def _export_transformers_checkpoint( @@ -940,71 +1031,19 @@ def _export_transformers_checkpoint( accelerator = kwargs.get("accelerator") - # Handle input quantizers of experts that are not calibrated - for _, sub_module in model.named_modules(): + # Handle input quantizers of experts that are not calibrated. Each MoE block is + # dispatched by its experts container to the matching ModuleExporter. + prepare_ctx = ExportContext(model=model, dtype=dtype, is_modelopt_qlora=is_modelopt_qlora) + for name, sub_module in model.named_modules(): if is_moe(sub_module) and hasattr(sub_module, "experts"): - expert_linear_names = get_expert_linear_names(sub_module) - first_proj_attr = getattr(sub_module.experts, "_first_proj_attr", "gate_up_proj") - has_fused_experts_quantizers = hasattr( - sub_module.experts, f"{first_proj_attr}_weight_quantizers" - ) - for linear_name in expert_linear_names: - # Handle DBRX experts specifically - if "QuantDbrxExperts" in type(sub_module.experts).__name__: - # For DBRX, experts are in sub_module.experts.mlp and linear layers are ModuleLists - experts_mlp = sub_module.experts.mlp - if hasattr(experts_mlp, linear_name): - linear_modulelist = getattr(experts_mlp, linear_name) - if hasattr(linear_modulelist, "__iter__"): - set_expert_quantizer_amax( - modules=list(linear_modulelist), - quantizer_attrs=["input_quantizer"], - ) - elif has_fused_experts_quantizers: - # _QuantFusedExperts: amax fallback is handled in _export_fused_experts - break - elif ( - "QuantGptOssExperts" in type(sub_module.experts).__name__ - or "QuantLlama4TextExperts" in type(sub_module.experts).__name__ - ): - # Handle GPT-OSS / Llama4 fused experts specifically. - # Both use gate_up_proj and down_proj with singular input quantizers - # (gate_up_proj_input_quantizer/down_proj_input_quantizer); the actual - # amax fallback and weight export is performed in _process_quantized_modules. - gpt_oss_linear_names = ["gate_up_proj", "down_proj"] - for linear_name in gpt_oss_linear_names: - if hasattr(sub_module.experts, linear_name): - linear_module = getattr(sub_module.experts, linear_name) - if hasattr(linear_module, "input_quantizer"): - set_expert_quantizer_amax( - modules=[linear_module], - quantizer_attrs=["input_quantizer"], - ) - elif isinstance(sub_module.experts, collections.abc.Iterable): - # For other MoE models (like Mixtral) with iterable experts - try: - set_expert_quantizer_amax( - modules=[getattr(expert, linear_name) for expert in sub_module.experts], - quantizer_attrs=["input_quantizer"], - ) - except AttributeError as e: - # Provide more helpful debugging information - expert_types = [type(expert).__name__ for expert in sub_module.experts] - raise AttributeError( - f"Failed to access attribute '{linear_name}' on experts. " - f"MoE module type: {type(sub_module).__name__}, " - f"Expert types: {expert_types}, " - f"Expected linear names: {expert_linear_names}. " - f"This suggests the get_expert_linear_names function may need " - f"to be updated for this model architecture. " - f"Original error: {e}" - ) from e - else: - # Unsupported MoE model structure - raise NotImplementedError( - f"MoE model with experts type '{type(sub_module.experts).__name__}' is not supported in export." - f"Please file an issue or add support for this model architecture." - ) + exporter = ExportModuleRegistry.match(sub_module.experts) + if exporter is None: + # Unsupported MoE model structure + raise NotImplementedError( + f"MoE model with experts type '{type(sub_module.experts).__name__}' is not supported in export." + f"Please file an issue or add support for this model architecture." + ) + exporter.prepare_moe_inputs(name, sub_module, prepare_ctx) # Resmooth and requantize fused layers # TODO: Handle mixed precision diff --git a/tests/unit/torch/export/test_export_registry.py b/tests/unit/torch/export/test_export_registry.py new file mode 100644 index 00000000000..b2368883277 --- /dev/null +++ b/tests/unit/torch/export/test_export_registry.py @@ -0,0 +1,249 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 +# +# 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. + +"""Tests for the export module registry dispatching the unified HF export.""" + +import pytest +import torch +import torch.nn as nn +from _test_utils.torch.export.utils import ToyModel, partial_fp8_config + +import modelopt.torch.quantization as mtq +from modelopt.torch.export.registry import ( + ExportContext, + ExportModuleRegistry, + ModuleExporter, + _ExportModuleRegistryCls, +) +from modelopt.torch.export.unified_export_hf import ( + _BmmExpertsExporter, + _DbrxExpertsExporter, + _FusedExpertsExporter, + _IterableExpertsExporter, + _MoELinearExporter, + _process_quantized_modules, + _QuantEmbeddingExporter, + _QuantLinearExporter, +) + + +class _Experts(nn.Module): + pass + + +def _make_dynamic_subclass(base: type[nn.Module], prefix: str = "Quant") -> type[nn.Module]: + """Mimic _DMRegistryCls's on-the-fly class generation (e.g. QuantLinear).""" + return type(f"{prefix}{base.__name__}", (base,), {}) + + +def test_class_key_matches_generated_subclass_via_mro(): + registry = _ExportModuleRegistryCls() + + @registry.register(nn.Linear) + class _LinearHandler(ModuleExporter): + pass + + quant_linear_cls = _make_dynamic_subclass(nn.Linear) + module = nn.Linear(2, 2) + module.__class__ = quant_linear_cls + + assert isinstance(registry.match(module), _LinearHandler) + assert isinstance(registry.match(nn.Linear(2, 2)), _LinearHandler) + assert registry.match(nn.Embedding(2, 2)) is None + + +def test_name_key_matches_original_and_generated_class(): + registry = _ExportModuleRegistryCls() + + @registry.register("_Experts") + class _ExpertsHandler(ModuleExporter): + pass + + raw = _Experts() + generated = _Experts() + generated.__class__ = _make_dynamic_subclass(_Experts) + + # The original class name appears in the generated class's MRO. + assert isinstance(registry.match(raw), _ExpertsHandler) + assert isinstance(registry.match(generated), _ExpertsHandler) + assert registry.match(nn.Linear(2, 2)) is None + + +def test_keys_and_predicate_are_both_required(): + registry = _ExportModuleRegistryCls() + + @registry.register("_Experts", predicate=lambda m: hasattr(m, "experts")) + class _ExpertsHandler(ModuleExporter): + pass + + without_experts = _Experts() + with_experts = _Experts() + with_experts.experts = nn.ModuleList([nn.Linear(2, 2)]) + + assert registry.match(without_experts) is None + assert isinstance(registry.match(with_experts), _ExpertsHandler) + + +def test_first_registered_entry_wins(): + registry = _ExportModuleRegistryCls() + + @registry.register(predicate=lambda m: isinstance(m, nn.Linear)) + class _SpecificHandler(ModuleExporter): + pass + + @registry.register(nn.Module) + class _GenericHandler(ModuleExporter): + pass + + assert isinstance(registry.match(nn.Linear(2, 2)), _SpecificHandler) + assert isinstance(registry.match(nn.Embedding(2, 2)), _GenericHandler) + + +def test_register_requires_key_or_predicate(): + registry = _ExportModuleRegistryCls() + with pytest.raises(AssertionError): + registry.register()(ModuleExporter) + + +def test_prepend_registers_before_existing_entries(): + registry = _ExportModuleRegistryCls() + + @registry.register(predicate=lambda m: True) + class _CatchAllHandler(ModuleExporter): + pass + + @registry.register(nn.Linear, prepend=True) + class _SpecificHandler(ModuleExporter): + pass + + assert isinstance(registry.match(nn.Linear(2, 2)), _SpecificHandler) + assert isinstance(registry.match(nn.Embedding(2, 2)), _CatchAllHandler) + + +def test_reregistering_same_class_replaces_entry_in_place(): + registry = _ExportModuleRegistryCls() + + @registry.register(nn.Linear) + class _FirstHandler(ModuleExporter): + pass + + @registry.register(predicate=lambda m: True) + class _CatchAllHandler(ModuleExporter): + pass + + # Simulate a module reload re-running the decorator on the same class: + # the entry is replaced in place, keeping its (winning) position. + registry.register(nn.Linear)(_FirstHandler) + assert len(registry._entries) == 2 + assert isinstance(registry.match(nn.Linear(2, 2)), _FirstHandler) + + +def _named_module(name: str, base_name: str | None = None, **attrs) -> nn.Module: + """Create a module instance whose class (and optionally base class) has a given name.""" + base = type(base_name, (nn.Module,), {}) if base_name else nn.Module + module = type(name, (base,), {})() + for attr, value in attrs.items(): + setattr(module, attr, value) + return module + + +def test_builtin_dispatch_covers_all_handler_shapes(): + # Step-3.5 QuantMoELinear wrapper: exact leaf-class name plus experts attr. + moe_linear = _named_module("QuantMoELinear", experts=nn.ModuleList([nn.Linear(2, 2)])) + assert isinstance(ExportModuleRegistry.match(moe_linear), _MoELinearExporter) + # Without the experts attr the entry must not match. + assert not isinstance( + ExportModuleRegistry.match(_named_module("QuantMoELinear")), _MoELinearExporter + ) + + # DBRX experts container: the usual generated class name... + assert isinstance( + ExportModuleRegistry.match(_named_module("QuantDbrxExperts", base_name="DbrxExperts")), + _DbrxExpertsExporter, + ) + # ...and the _DMRegistryCls collision-fallback name, where only the mixin + # class name ("_QuantDbrxExperts") remains recognizable in the MRO. + fallback = _named_module( + "transformers_modules_modeling_dbrx_QuantDbrxExperts", base_name="_QuantDbrxExperts" + ) + assert isinstance(ExportModuleRegistry.match(fallback), _DbrxExpertsExporter) + + # Fused experts (plural per-expert weight quantizers) must win over the BMM + # name keys even when the class name also matches a BMM entry. + fused = _named_module( + "QuantGptOssExperts", + base_name="GptOssExperts", + gate_up_proj_weight_quantizers=nn.ModuleList(), + ) + assert isinstance(ExportModuleRegistry.match(fused), _FusedExpertsExporter) + + # BMM-style experts by class name: raw and quant-generated variants. + assert isinstance( + ExportModuleRegistry.match(_named_module("GptOssExperts")), _BmmExpertsExporter + ) + assert isinstance( + ExportModuleRegistry.match( + _named_module("QuantLlama4TextExperts", base_name="Llama4TextExperts") + ), + _BmmExpertsExporter, + ) + + # Iterable containers (e.g. Mixtral's ModuleList of experts) hit the catch-all. + assert isinstance(ExportModuleRegistry.match(nn.ModuleList()), _IterableExpertsExporter) + + # Opaque experts containers match nothing — the prepass turns this into + # an actionable NotImplementedError. + assert ExportModuleRegistry.match(_named_module("OpaqueExperts")) is None + + +def test_builtin_registry_dispatches_quantized_modules(): + model = ToyModel(dims=[16, 32, 16]) + mtq.quantize(model, partial_fp8_config, lambda m: m(torch.randn(2, 4, 16))) + + quantized = [m for m in model.modules() if type(m).__name__ == "QuantLinear"] + assert quantized, "expected at least one quantized linear in the toy model" + for module in quantized: + assert isinstance(ExportModuleRegistry.match(module), _QuantLinearExporter) + + # Plain (unquantized) modules match no handler. + assert ExportModuleRegistry.match(nn.Linear(2, 2)) is None + + # A quantized embedding dispatches to the embedding handler. + embedding = nn.Embedding(4, 4) + embedding.weight_quantizer = None + assert isinstance(ExportModuleRegistry.match(embedding), _QuantEmbeddingExporter) + + +def test_process_quantized_modules_exports_via_registry(): + model = ToyModel(dims=[16, 32, 16]) + mtq.quantize(model, partial_fp8_config, lambda m: m(torch.randn(2, 4, 16))) + + _process_quantized_modules(model, torch.float16) + + state_dict = model.state_dict() + fp8_weights = [k for k in state_dict if k.endswith("weight_scale")] + assert fp8_weights, "expected weight_scale buffers registered by the linear exporter" + for key in fp8_weights: + weight = state_dict[key.replace("weight_scale", "weight")] + assert weight.dtype == torch.float8_e4m3fn + + +def test_export_context_caches_are_per_instance(): + model = nn.Linear(2, 2) + ctx_a = ExportContext(model=model, dtype=torch.float16) + ctx_b = ExportContext(model=model, dtype=torch.float16) + ctx_a.tied_cache[123] = model + assert ctx_b.tied_cache == {} + assert ctx_b.moe_tied_cache == {}