diff --git a/modelopt/torch/quantization/algorithms.py b/modelopt/torch/quantization/algorithms.py index 7beeef6ad7f..7708c6b04a3 100644 --- a/modelopt/torch/quantization/algorithms.py +++ b/modelopt/torch/quantization/algorithms.py @@ -23,7 +23,7 @@ from abc import ABC, abstractmethod from collections import defaultdict from collections.abc import Callable, Sequence -from contextlib import nullcontext +from contextlib import ExitStack, nullcontext from typing import Any import regex as re @@ -413,8 +413,9 @@ def __init__( self.allow_no_quant = allow_no_quant self.is_fixed = fixed_recipe is not None - self.quant_modules = list(set(quant_modules or [])) - self.score_modules = list(set(score_modules or self.quant_modules)) + # Module hashes depend on object identity, so sets can produce different orders per rank. + self.quant_modules = list(dict.fromkeys(quant_modules or [])) + self.score_modules = list(dict.fromkeys(score_modules or self.quant_modules)) fixed_quantizers = ( { @@ -467,11 +468,11 @@ def __init__( quant_recipe: dict.fromkeys(self.score_modules) for quant_recipe in self.choices } - # Attach this hparam to each score_module's set of hparams it scores + # Registration order follows the rank-stable runtime-group construction order. for score_module in self.score_modules: if not hasattr(score_module, "_hparams_for_scoring"): - score_module._hparams_for_scoring = set() - score_module._hparams_for_scoring.add(self) + score_module._hparams_for_scoring = [] + score_module._hparams_for_scoring.append(self) @property def active(self) -> HPType: @@ -532,6 +533,15 @@ def get_score(self, recipe: QuantRecipe) -> float: continue parallel_state = getattr(score_module, "parallel_state", None) + if parallel_state is None: + parallel_state = next( + ( + state + for module in self.quant_modules + if (state := getattr(module, "parallel_state", None)) is not None + ), + None, + ) if parallel_state is None: total_score += importance.cpu().item() @@ -1438,7 +1448,211 @@ def _add_auto_quantize_score(grad_output, output_diff, score_tensor): score_tensor += _get_auto_quantize_score(grad_output, output_diff) -class AutoQuantizeGradientSearcher(_AutoQuantizeBaseSearcher): +class _AutoQuantizeBackwardScoringSession(ABC): + """Manage temporary model state used by activation-backward scoring.""" + + def __init__( + self, + model: nn.Module, + score_modules: Sequence[nn.Module], + is_param_grad_enabled: Callable, + verbose: bool = False, + ) -> None: + self.model = model + self.score_modules = tuple(score_modules) + self.is_param_grad_enabled = is_param_grad_enabled + self.verbose = verbose + self._stack = ExitStack() + self._original_forwards: dict[nn.Module, Callable] = {} + self._grad_accumulators: list[Any] = [] + + def __enter__(self): + """Install scoring hooks and parameter settings.""" + try: + hparams = list( + dict.fromkeys( + hparam + for module in self.score_modules + for hparam in module._hparams_for_scoring + ) + ) + for hparam in hparams: + self._stack.callback(setattr, hparam, "active", hparam.active) + + def patched_forward(module, *args, **kwargs): + return self.forward(module, *args, **kwargs) + + for module in self.score_modules: + original_forward = module.forward + self._original_forwards[module] = original_forward + had_instance_forward = "forward" in module.__dict__ + instance_forward = module.__dict__.get("forward") + module.forward = types.MethodType(patched_forward, module) + if had_instance_forward: + self._stack.callback(setattr, module, "forward", instance_forward) + else: + self._stack.callback(module.__dict__.pop, "forward", None) + hook = module.register_full_backward_hook(self.backward_hook) + self._stack.callback(hook.remove) + + for name, param in self.model.named_parameters(): + requires_grad = param.requires_grad + enable_grad = self.is_param_grad_enabled(name, self.model) + param.requires_grad = enable_grad + self._stack.callback(setattr, param, "requires_grad", requires_grad) + if not enable_grad: + continue + if self.verbose: + print_rank_0(f"AutoQuantize: Enabling gradient for param {name}.") + accumulator, hook = create_param_grad_clear_hook(param) + self._grad_accumulators.append(accumulator) + self._stack.callback(hook.remove) + except Exception: + self._stack.close() + raise + return self + + def __exit__(self, exc_type, exc_value, traceback) -> None: + """Restore all model state changed for scoring.""" + self._stack.close() + self._original_forwards.clear() + self._grad_accumulators.clear() + + def original_forward(self, module: nn.Module) -> Callable: + """Return the forward method saved before scoring.""" + return self._original_forwards[module] + + @abstractmethod + def forward(self, module: nn.Module, *args, **kwargs): + """Run a score module forward pass and collect method-specific state.""" + + @abstractmethod + def backward_hook(self, module: nn.Module, grad_input, grad_output) -> None: + """Accumulate scores from a score module's output gradient.""" + + +class _AutoQuantizeGradientScoringSession(_AutoQuantizeBackwardScoringSession): + """Collect gradient-based scores while candidate recipes are replayed.""" + + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + self._output_diffs: dict[nn.Module, dict] = {} + + def forward(self, module: nn.Module, *args, **kwargs): + """Run the reference forward and cache each recipe's output perturbation.""" + no_quant_recipe = QuantRecipe(quant_cfg=None) + for hparam in module._hparams_for_scoring: + if hparam.is_configurable: + hparam.active = no_quant_recipe + + output = self.original_forward(module)(*args, **kwargs) + + # Checkpointed modules recompute with gradients enabled during backward. + if not torch.is_grad_enabled(): + return output + + output_diffs = {hparam: {} for hparam in module._hparams_for_scoring} + self._output_diffs[module] = output_diffs + with torch.no_grad(): + for hparam in module._hparams_for_scoring: + if not hparam.is_configurable: + continue + for recipe in hparam.choices: + if recipe == no_quant_recipe: + continue + hparam.active = recipe + replay = self.original_forward(module)(*args, **kwargs) + output_diff = ( + replay[0] - output[0] if isinstance(replay, tuple) else replay - output + ) + output_diffs[hparam][recipe] = output_diff.detach() + hparam.active = no_quant_recipe + + return output + + def backward_hook(self, module: nn.Module, grad_input, grad_output) -> None: + """Accumulate squared gradient-weighted output perturbations.""" + for hparam, output_diffs in self._output_diffs[module].items(): + for recipe, output_diff in output_diffs.items(): + importance = hparam._importance_dict[recipe][module] + if importance is None: + hparam._importance_dict[recipe][module] = _get_auto_quantize_score( + grad_output[0], output_diff + ) + else: + _add_auto_quantize_score(grad_output[0], output_diff, importance) + + def __exit__(self, exc_type, exc_value, traceback) -> None: + self._output_diffs.clear() + super().__exit__(exc_type, exc_value, traceback) + + +class _AutoQuantizeBackwardScoringSearcher(_AutoQuantizeBaseSearcher): + """Share orchestration used by activation-backward scoring methods.""" + + score_module_rules = [ + # Score MoE projections together at their enclosing MLP or mixer output. + r"^(.*?\.mlp)\.experts\.\d+\.(gate_proj|up_proj|down_proj)$", + r"^(.*?\.mixer)\.experts\.\d+\.(up_proj|down_proj)$", + r"^(.*?)\.(\d+\.(w1|w2|w3))$", + r"^(.*?)\.((w1_linear|w2_linear|w3_linear)\.\d+)$", + ] + + _custom_support: list[tuple[Callable, Callable, Callable]] = [] + + @classmethod + def register_custom_support( + cls, + is_supported_checker: Callable, + grad_ckpt_context: Callable, + is_param_grad_enabled: Callable, + ) -> None: + """Register optional hooks for memory-efficient backward scoring. + + `is_supported_checker` selects models that use these hooks. + `grad_ckpt_context` enables their gradient-checkpointing context, and + `is_param_grad_enabled` selects the minimum parameters needed to propagate + activation gradients. + """ + cls._custom_support.append((is_supported_checker, grad_ckpt_context, is_param_grad_enabled)) + + def _configurable_score_modules(self) -> list[nn.Module]: + return [ + module + for module in self.model.modules() + if hasattr(module, "_hparams_for_scoring") + and any(hparam.is_configurable for hparam in module._hparams_for_scoring) + ] + + @abstractmethod + def _estimate_auto_quantize_scores(self, is_param_grad_enabled: Callable) -> None: + """Estimate scores while activation gradients are enabled.""" + + def estimate_sensitivity_scores(self) -> None: + """Run backward scoring with the first matching model-specific support hook.""" + self.model.eval() + + def default_is_param_grad_enabled(_name, _model): + return True + + grad_checkpointing_context = None + is_param_grad_enabled = default_is_param_grad_enabled + for is_supported, context_candidate, grad_candidate in self._custom_support: + if is_supported(self.model): + grad_checkpointing_context = context_candidate + is_param_grad_enabled = grad_candidate + break + + context = ( + grad_checkpointing_context(self.model) + if grad_checkpointing_context is not None + else nullcontext() + ) + with context: + self._estimate_auto_quantize_scores(is_param_grad_enabled) + + +class AutoQuantizeGradientSearcher(_AutoQuantizeBackwardScoringSearcher): """A searcher for AutoQuantize algorithm that uses gradient based score estimation. In AutoQuantize, we search for the best per-layer quantization configuration that minimizes the sum of per-layer @@ -1472,17 +1686,6 @@ class AutoQuantizeGradientSearcher(_AutoQuantizeBaseSearcher): method_name = "gradient" - score_module_rules = [ - # Use MLP layer output for gate_proj, up_proj, down_proj for Qwen3 like MoE models (local and shared experts) - r"^(.*?\.mlp)\.experts\.\d+\.(gate_proj|up_proj|down_proj)$", - r"^(.*?\.mixer)\.experts\.\d+\.(up_proj|down_proj)$", # NemotronH MoE experts - r"^(.*?)\.(\d+\.(w1|w2|w3))$", # mixtral experts - r"^(.*?)\.((w1_linear|w2_linear|w3_linear)\.\d+)$", # dbrx experts - ] - - # See `register_custom_support` for details - _custom_support: list[tuple[Callable, Callable, Callable]] = [] - @property def default_search_config(self): """Get the default config for the searcher.""" @@ -1511,30 +1714,6 @@ def sanitize_search_config(self, config: SearchConfig | None) -> SearchConfig: return config - @classmethod - def register_custom_support( - cls, - is_supported_checker: Callable, - grad_ckpt_context: Callable, - is_param_grad_enabled: Callable, - ) -> None: - """(Optional) Register custom support for `AutoQuantize` score estimation. - - This custom support is used to enable memory/compute efficient backward gradient propagation. This involves: - - - `grad_ckpt_context`: backward pass with gradient checkpointing enabled - - `is_param_grad_enabled`: AutoQuantize only needs activation gradients to be computed (not weight - gradients). `is_param_grad_enabled` is used to select which parameters should have gradients enabled, - limiting gradient computation to only what's needed for activation gradients. For LLMs, to trigger all - activation gradient computation, just enabling the embedding layer weight gradient is sufficient. This will - enable gradient computation for all the activation gradients downstream. - - If the `is_supported_checker(model)` returns True, the `grad_ckpt_context(model)` will be - used to enable gradient checkpointing and `is_param_grad_enabled(pname, model)` - will be used to select which parameters have gradients enabled to minimize gradient computation. - """ - cls._custom_support.append((is_supported_checker, grad_ckpt_context, is_param_grad_enabled)) - def _get_default_forward_backward_step(self): def forward_backward_step(model, data): output = self.config["forward_step"](model, data) @@ -1552,143 +1731,29 @@ def forward_backward_step(model, data): @torch.enable_grad() def _estimate_auto_quantize_scores(self, is_param_grad_enabled): - # TODO: remove the no-quant recipe - def auto_quantize_score_estimate_forward(module, input, *args, **kwargs): - for hparam in module._hparams_for_scoring: - if hparam.is_configurable: - hparam.active = QuantRecipe(quant_cfg=None) - - output = module._forward_original(input, *args, **kwargs) - - # If gradient checkpointing is enabled, gradient will not be enabled in the global forward pass. - # With gradient checkpointing, gradients are computed in the local forward pass during backward pass - - # Lets compute the output_diff and save it in memory only if gradient is enabled to be memory efficient - if not torch.is_grad_enabled(): - return output - - module.output_diff_dict = {hparam: {} for hparam in module._hparams_for_scoring} - with torch.no_grad(): - for hparam in module._hparams_for_scoring: - if not hparam.is_configurable: - continue - for recipe in hparam.choices: - if recipe == QuantRecipe(quant_cfg=None): - continue - hparam.active = recipe - output_diff = module._forward_original(input, *args, **kwargs) - - if isinstance(output_diff, tuple): - output_diff = output_diff[0] - output[0] - else: - output_diff -= output - module.output_diff_dict[hparam][recipe] = output_diff.detach() - - # Disable the configurable hparam now that we have computed the diff - hparam.active = QuantRecipe(quant_cfg=None) - - return output - - def backward_hook(module, grad_input, grad_output): - for hparam, output_diff_dict in module.output_diff_dict.items(): - for recipe, output_diff in output_diff_dict.items(): - if hparam._importance_dict[recipe][module] is None: - hparam._importance_dict[recipe][module] = _get_auto_quantize_score( - grad_output[0], output_diff - ) - else: - _add_auto_quantize_score( - grad_output[0], output_diff, hparam._importance_dict[recipe][module] - ) - - def setup_params_for_score_estimation(name, param, params_metadata, enable_grad=True): - # Let us delete the gradient as soon as they are computed to save memory - params_metadata[name] = {"requires_grad": param.requires_grad} - param.requires_grad = enable_grad - if not enable_grad: - return - if self.config.get("verbose", False): - print_rank_0(f"AutoQuantize: Enabling gradient for param {name}.") - accum_grad, handle = create_param_grad_clear_hook(param) - params_metadata[name]["accum_grad"] = accum_grad # We need to keep the accum_grad alive - params_metadata[name]["handle"] = handle - - def setup_module_for_score_estimation(module): - module._forward_original = module.forward - module.forward = types.MethodType(auto_quantize_score_estimate_forward, module) - module._backward_hook_handle = module.register_full_backward_hook(backward_hook) - - def cleanup_module_after_score_estimation(module): - module.forward = module._forward_original - del module._forward_original - - module._backward_hook_handle.remove() - - def cleanup_params_after_score_estimation(name, param, params_metadata): - param.requires_grad = params_metadata[name]["requires_grad"] - handle = params_metadata[name].get("handle", None) - if handle is not None: - handle.remove() - - score_modules = set() - for name, module in self.model.named_modules(): - if ( - hasattr(module, "_hparams_for_scoring") - and any(hparam.is_configurable for hparam in module._hparams_for_scoring) - and module not in score_modules - ): - # Monkey patch the forward methods to cache (Q(Y) - Y) - setup_module_for_score_estimation(module) - score_modules.add(module) - - params_metadata = {} - for name, param in self.model.named_parameters(): - setup_params_for_score_estimation( - name, param, params_metadata, is_param_grad_enabled(name, self.model) + score_modules = self._configurable_score_modules() + with _AutoQuantizeGradientScoringSession( + self.model, + score_modules, + is_param_grad_enabled, + verbose=self.config.get("verbose", False), + ): + gc.collect() + if torch.cuda.is_available(): + torch.cuda.reset_peak_memory_stats() + report_memory("AutoQuantize: starting score estimation, ") + + self._run_func( + self.config["forward_backward_step"], + num_iters=self.config["num_score_steps"], + desc="Estimating auto_quantize scores", ) - gc.collect() - if torch.cuda.is_available(): - torch.cuda.reset_peak_memory_stats() - report_memory("AutoQuantize: starting score estimation, ") - - self._run_func( - self.config["forward_backward_step"], - num_iters=self.config["num_score_steps"], - desc="Estimating auto_quantize scores", - ) - - if torch.cuda.is_available(): - report_memory("AutoQuantize: After score estimation") - - for module in score_modules: - cleanup_module_after_score_estimation(module) + if torch.cuda.is_available(): + report_memory("AutoQuantize: After score estimation") - for name, param in self.model.named_parameters(): - cleanup_params_after_score_estimation(name, param, params_metadata) - - # Delete the params_metadata - del params_metadata gc.collect() - def estimate_sensitivity_scores(self) -> None: - """Estimate sensitivity scores using hessian approximation.""" - self.model.eval() - - def _default_is_param_grad_enabled(pname, model): - return True - - grad_checkpointing_ctxt = None - is_param_grad_enabled = _default_is_param_grad_enabled - for is_supported_checker, ctxt_candidate, grad_enabled_candidate in self._custom_support: - if is_supported_checker(self.model): - grad_checkpointing_ctxt = ctxt_candidate - is_param_grad_enabled = grad_enabled_candidate - break - - with grad_checkpointing_ctxt(self.model) if grad_checkpointing_ctxt else nullcontext(): - self._estimate_auto_quantize_scores(is_param_grad_enabled) - def run_search_with_stats(self, max_weight_size, verbose=False): """Linear Programming Solve for gradient based auto_quantize. diff --git a/tests/unit/torch/quantization/test_autoquant.py b/tests/unit/torch/quantization/test_autoquant.py index e83f7fa0a70..ba3c104f731 100644 --- a/tests/unit/torch/quantization/test_autoquant.py +++ b/tests/unit/torch/quantization/test_autoquant.py @@ -36,6 +36,7 @@ QuantRecipe, QuantRecipeHparam, _AutoQuantizeBaseSearcher, + _AutoQuantizeGradientScoringSession, _module_search_space_signature, estimate_quant_compression, ) @@ -98,6 +99,47 @@ def get_input(self): return torch.randn(1, 4, 32) +class _ScoredMoeExpert(torch.nn.Module): + def __init__(self): + super().__init__() + self.gate_proj = torch.nn.Linear(8, 8) + self.up_proj = torch.nn.Linear(8, 8) + self.down_proj = torch.nn.Linear(8, 8) + + def forward(self, x): + return self.down_proj(self.gate_proj(x) + self.up_proj(x)) + + +class _ScoredMoeMlp(torch.nn.Module): + def __init__(self): + super().__init__() + self.experts = torch.nn.ModuleList([_ScoredMoeExpert(), _ScoredMoeExpert()]) + + def forward(self, x): + output = torch.zeros_like(x) + for expert in self.experts: + output = output + expert(x) + return output + + +class _ScoredMoeLayer(torch.nn.Module): + def __init__(self): + super().__init__() + self.mlp = _ScoredMoeMlp() + + def forward(self, x): + return self.mlp(x) + + +class _ScoredMoeModel(torch.nn.Module): + def __init__(self): + super().__init__() + self.layers = torch.nn.ModuleList([_ScoredMoeLayer()]) + + def forward(self, x): + return self.layers[0](x) + + @pytest.mark.parametrize( ("quant_cfg", "other_quant_cfg", "is_less_than"), [ @@ -905,6 +947,141 @@ def test_data_parallel_auto_quantize(skip_on_windows): spawn_multiprocess_job(2, _test_data_parallel_auto_quantize, backend="gloo") +def _test_data_parallel_moe_score_module(rank, size): + torch.manual_seed(1234) + model = _ScoredMoeModel() + data_loader = [torch.randn(2, 3, 8) for _ in range(2)] + model, search_history = mtq.auto_quantize( + model, + constraints={"effective_bits": 12.0}, + quantization_formats=[mtq.INT8_DEFAULT_CFG], + data_loader=data_loader, + forward_step=lambda model, batch: model(batch), + loss_func=lambda output, data: output.square().mean(), + num_calib_steps=2, + num_score_steps=2, + ) + + hparam = model.layers[0].mlp.experts[0].gate_proj.get_hparam("quant_recipe") + assert hparam.score_modules == [model.layers[0].mlp] + assert isinstance(model.layers[0].mlp._hparams_for_scoring, list) + + recipe = QuantRecipe(mtq.INT8_DEFAULT_CFG) + local_score = sum( + hparam._importance_dict[recipe][score_module].item() + for score_module in hparam.score_modules + ) + candidate = next( + candidate + for candidate in search_history["candidate_stats"].values() + if "layers.0.mlp.experts.0.gate_proj" in candidate["module_names"] + ) + recipe_idx = candidate["formats"].index(recipe) + assert candidate["scores"][recipe_idx] == pytest.approx(local_score * size) + + scores = { + name: candidate["scores"] for name, candidate in search_history["candidate_stats"].items() + } + rank_zero_scores = DistributedProcessGroup.get_dist_syncd_obj( + scores if rank == 0 else None, + DistributedProcessGroup(None), + lambda values: values[0], + ) + assert scores == rank_zero_scores + + +def test_data_parallel_moe_score_module(skip_on_windows): + spawn_multiprocess_job(2, _test_data_parallel_moe_score_module, backend="gloo") + + +def test_score_hparam_registration_preserves_order(): + quant_modules = [ + mtq.quantize(torch.nn.Linear(4, 4), mtq.INT8_DEFAULT_CFG), + mtq.quantize(torch.nn.Linear(4, 4), mtq.INT8_DEFAULT_CFG), + ] + score_module = torch.nn.Identity() + recipe = QuantRecipe(mtq.INT8_DEFAULT_CFG) + + first = QuantRecipeHparam( + [recipe], + quant_modules=[quant_modules[0], quant_modules[1], quant_modules[0]], + score_modules=[score_module, score_module], + ) + second = QuantRecipeHparam( + [recipe], + quant_modules=[quant_modules[1]], + score_modules=[score_module], + ) + + assert first.quant_modules == quant_modules + assert first.score_modules == [score_module] + assert score_module._hparams_for_scoring == [first, second] + + +def test_gradient_scoring_restores_model_after_failure(): + model = SimpleLinear() + patched_modules = [] + + def fail_during_scoring(model, data): + model(data) + patched_modules.extend( + module + for module in model.modules() + if getattr(module.forward, "__name__", None) == "patched_forward" + ) + raise RuntimeError("stop after scoring forward") + + with pytest.raises(RuntimeError, match="stop after scoring forward"): + mtq.auto_quantize( + model, + constraints={"effective_bits": 12.0}, + quantization_formats=[mtq.INT8_DEFAULT_CFG], + data_loader=[model.get_input()], + forward_step=lambda model, batch: model(batch), + forward_backward_step=fail_during_scoring, + num_calib_steps=1, + num_score_steps=1, + ) + + assert patched_modules + assert all( + getattr(module.forward, "__name__", None) != "patched_forward" for module in patched_modules + ) + assert all("forward" not in module.__dict__ for module in patched_modules) + assert all(not module._backward_hooks for module in patched_modules) + assert all(param.requires_grad for param in model.parameters()) + for module in model.modules(): + for hparam in getattr(module, "_hparams_for_scoring", []): + assert hparam.active == hparam.original + + +@pytest.mark.parametrize("instance_override", [False, True]) +def test_gradient_scoring_restores_forward_attribute_layout(instance_override): + module = torch.nn.Identity() + module._hparams_for_scoring = [] + + def original_forward(x): + return x + 1 + + original_override = original_forward + if instance_override: + module.forward = original_override + + session = _AutoQuantizeGradientScoringSession( + module, + [module], + lambda _name, _model: False, + ) + with pytest.raises(RuntimeError, match="stop during scoring"), session: + assert module.__dict__["forward"] is not original_override + raise RuntimeError("stop during scoring") + + if instance_override: + assert module.__dict__["forward"] is original_override + else: + assert "forward" not in module.__dict__ + + def test_auto_quantize_budget_uses_no_quant_candidate_cost(monkeypatch): class _BudgetCaptureSearcher(AutoQuantizeGradientSearcher): def run_search_with_stats(self, max_weight_size, verbose=False):