diff --git a/auto_round/algorithms/transforms/awq/base.py b/auto_round/algorithms/transforms/awq/base.py index d2e3999927..310b375e2f 100644 --- a/auto_round/algorithms/transforms/awq/base.py +++ b/auto_round/algorithms/transforms/awq/base.py @@ -29,8 +29,6 @@ from __future__ import annotations -import inspect -import re import sys from typing import TYPE_CHECKING, Any @@ -46,11 +44,16 @@ resolve_mappings, ) from auto_round.algorithms.transforms.awq.qdq import QDQTool -from auto_round.algorithms.transforms.base import BasePreprocessor -from auto_round.data_type.utils import ( - reshape_pad_tensor_by_group_size, - revert_tensor_by_pad, +from auto_round.algorithms.transforms.awq.smooth import ( + AWQSmoothStrategy, + compute_layer_means, + fold_scales_into_smooth_layer, + get_grid_search_params, + smooth_group_from_mapping, ) +from auto_round.algorithms.transforms.base import BasePreprocessor +from auto_round.algorithms.transforms.smoothing.engine import SmoothEngine +from auto_round.algorithms.transforms.smoothing.errors import NoFiniteCandidateError from auto_round.logger import logger from auto_round.utils.model import move_to_device @@ -58,53 +61,6 @@ from auto_round.algorithms.composer import AlgorithmComposer, BlockContext -# Known normalization classes whose ``forward`` computes -# ``output = (1 + weight) * x_norm`` (Gemma-style "unit-offset" RMSNorm) rather -# than the standard ``output = weight * x_norm``. Folding an AWQ smoothing scale -# ``s`` into such a layer requires ``weight <- (1 + weight) / s - 1`` instead of -# ``weight <- weight / s``; using the wrong fold silently breaks AWQ's output -# invariance and severely degrades accuracy (e.g. Qwen3.5, Gemma2/3, Qwen3-Next). -_UNIT_OFFSET_RMSNORM_NAMES = frozenset( - { - "GemmaRMSNorm", - "Gemma2RMSNorm", - "Gemma3RMSNorm", - "Gemma3TextRMSNorm", - "Qwen3_5RMSNorm", - "Qwen3_5MoeRMSNorm", - "Qwen3NextRMSNorm", - } -) - -# Detects ``1 + self.weight`` / ``self.weight + 1`` in a norm's forward source. -_UNIT_OFFSET_SRC_RE = re.compile(r"1(\.0)?\s*\+\s*self\.weight|self\.weight(\.float\(\))?\s*\+\s*1") - -# Cache the unit-offset decision per norm class to avoid repeated source parsing. -_unit_offset_cache: dict[type, bool] = {} - - -def _rmsnorm_has_unit_offset(module: torch.nn.Module) -> bool: - """Return True if ``module`` applies a Gemma-style ``(1 + weight)`` gain. - - Uses a fast class-name allowlist, falling back to source inspection of the - module's ``forward`` so newly-added Gemma-style norms are detected without a - code change. Result is cached per class. - """ - cls = type(module) - cached = _unit_offset_cache.get(cls) - if cached is not None: - return cached - result = cls.__name__ in _UNIT_OFFSET_RMSNORM_NAMES - if not result: - try: - src = inspect.getsource(cls.forward) - result = bool(_UNIT_OFFSET_SRC_RE.search(src)) - except (OSError, TypeError): - result = False - _unit_offset_cache[cls] = result - return result - - def _slice_seq_tensor(v: Any, actual_seq: int, seqlen: int) -> Any: """Slice a value's sequence dimension to *seqlen*, recursing into containers.""" if isinstance(v, torch.Tensor): @@ -659,19 +615,7 @@ def _smooth_block(self, block_prefix: str, block_mappings: list) -> None: self._parent_args_cache.pop(mapping.parent, None) def _get_grid_search_params(self) -> list[tuple[float, bool]]: - """Return (ratio, use_duo_scaling) tuples for the grid search.""" - match self.duo_scaling: - case "both": - n = max(int(self.n_grid / 2), 2) - return [(idx / (n - 1), duo) for idx in range(n) for duo in [False, True]] - case False: - n = max(self.n_grid, 2) - return [(idx / (n - 1), False) for idx in range(n)] - case True: - n = max(self.n_grid, 3) - return [(0.0, False)] + [(idx / (n - 2), True) for idx in range(n - 1)] - case _: - raise ValueError(f"Unexpected duo_scaling value: {self.duo_scaling!r}") + return get_grid_search_params(self.n_grid, self.duo_scaling) @staticmethod def _normalize_group_size(group_size: int | None, fallback: int) -> int: @@ -685,14 +629,7 @@ def _normalize_group_size(group_size: int | None, fallback: int) -> int: @staticmethod def _compute_layer_means(layers: list[torch.nn.Module], group_size: int) -> torch.Tensor: - """Per-channel mean of normalised weights across all balance layers.""" - weight = torch.cat([m.weight.detach().float() for m in layers], dim=0) - org_shape = weight.shape - gs = AWQTransform._normalize_group_size(group_size, org_shape[1]) - weight, _, pad_len = reshape_pad_tensor_by_group_size(weight, gs) - w_scale = weight.abs() / (weight.abs().amax(dim=1, keepdim=True) + 1e-6) - w_scale = revert_tensor_by_pad(w_scale, orig_shape=org_shape, pad_len=pad_len) - return w_scale.mean(0) + return compute_layer_means(layers, group_size) @torch.no_grad() def _grid_search_scales( @@ -701,93 +638,19 @@ def _grid_search_scales( x_mean: torch.Tensor, ) -> torch.Tensor | None: """Find the best scaling ratio for *mapping* via output-based loss.""" - device = mapping.balance_layers[0].weight.device - x_mean = x_mean.to(device) - - bl_params = {bl: self._qdq_tool.resolve_params(bl) for bl in mapping.balance_layers} - group_size = self._normalize_group_size(bl_params[mapping.balance_layers[0]]["group_size"], -1) - if self.duo_scaling is not False: - w_mean = self._compute_layer_means(mapping.balance_layers, group_size).to(device) - - parent_kwargs_list = self._parent_args_cache.get(mapping.parent, []) - use_parent_forward = len(parent_kwargs_list) > 0 - - if use_parent_forward: - fp16_outputs = self._run_parent_samples( - mapping.parent, - parent_kwargs_list, - offload_to_cpu=self._smooth_batch_size is not None, + engine = SmoothEngine() + group = smooth_group_from_mapping(mapping) + strategy = AWQSmoothStrategy(self, mapping) + with engine.session(strategy): + try: + result = engine.search(group, x_mean, strategy, require_finite=False, log_search=False) + except NoFiniteCandidateError: + logger.warning("AWQ: grid search failed for '%s': no finite error.", mapping.smooth_name) + return None + logger.debug( + "AWQ '%s': best_ratio=%.2f, best_error=%.3e", mapping.smooth_name, result.candidate[0], result.error ) - if not fp16_outputs or all(f.numel() == 0 for f in fp16_outputs): - use_parent_forward = False - - orig_state = {bl: bl.weight.data.clone() for bl in mapping.balance_layers} - if not use_parent_forward: - orig_weights = orig_state # same reference is fine - - # Resolve each balance layer's quant functions once, then reuse them in - # the grid-search loop. Normal AWQ flow requires one mapping to have - # compatible quant params, but keeping this per-layer avoids hidden - # coupling to the first layer and makes direct calls robust. - bl_quant_funcs = {bl: self._qdq_tool.resolve_quant_funcs(bl_params[bl]) for bl in mapping.balance_layers} - - best_error = float("inf") - best_scales = None - best_ratio = -1 - - for ratio, use_duo in self._get_grid_search_params(): - if use_duo: - scales = (x_mean.pow(ratio) / (w_mean.pow(1 - ratio) + 1e-4)).clamp(min=1e-4) - else: - scales = x_mean.pow(ratio).clamp(min=1e-4).view(-1) - scales = scales / (scales.max() * scales.min()).sqrt() - scales[torch.isinf(scales)] = 1 - scales[torch.isnan(scales)] = 1 - scales_view = scales.view(1, -1).to(device) - - if use_parent_forward: - # Quantize each balance layer's smoothed weight and write the - # de-smoothed result back, so the parent forward below sees the - # weights the layer would actually compute with. - for bl in mapping.balance_layers: - quant_func, opt_quant_func = bl_quant_funcs[bl] - w_qdq = self._qdq_tool.qdq( - orig_state[bl] * scales_view, - bl_params[bl], - quant_func=quant_func, - opt_quant_func=opt_quant_func, - imatrix=getattr(bl, "imatrix", None), - ) - bl.weight.data = (w_qdq / scales_view).to(bl.weight.dtype) - - total_loss = self._compute_parent_loss(mapping.parent, parent_kwargs_list, fp16_outputs) - for bl in mapping.balance_layers: - bl.weight.data.copy_(orig_state[bl]) - else: - total_loss = 0.0 - for bl in mapping.balance_layers: - quant_func, opt_quant_func = bl_quant_funcs[bl] - w_orig = orig_weights[bl].to(device) - w_qdq = self._qdq_tool.qdq( - w_orig * scales_view, - bl_params[bl], - quant_func=quant_func, - opt_quant_func=opt_quant_func, - imatrix=getattr(bl, "imatrix", None), - ) - total_loss += (w_orig - w_qdq / scales_view).pow(2).sum().item() - - if total_loss < best_error: - best_error = total_loss - best_scales = scales.clone() - best_ratio = ratio - - if best_ratio < 0: - logger.warning("AWQ: grid search failed for '%s': no finite error.", mapping.smooth_name) - return None - - logger.debug("AWQ '%s': best_ratio=%.2f, best_error=%.3e", mapping.smooth_name, best_ratio, best_error) - return best_scales + return result.candidate[1].clone() def _iter_parent_calls(self, stored_args: tuple, stored_kwargs: dict): """Yield full or microbatched parent-call args from one cached calibration batch.""" @@ -908,29 +771,7 @@ def _apply_scales(self, mapping: ResolvedMapping, scales: torch.Tensor) -> None: @staticmethod @torch.no_grad() def _fold_scales_into_smooth_layer(smooth: torch.nn.Module, scales: torch.Tensor) -> None: - """Divide a smooth layer's output by ``scales`` to offset balance scaling. - - Dispatches on the smooth layer's weight layout: - - * 1-D norm weight with a Gemma-style ``(1 + weight)`` gain: folded as - ``weight <- (1 + weight) / s - 1`` to preserve output invariance. - * 1-D standard norm weight: folded as ``weight <- weight / s``. - * 2-D linear weight: its trailing ``s.numel()`` output rows are divided. - - Any bias is always divided by ``s``. - """ - s = scales.to(smooth.weight.device) - weight = smooth.weight.data - if weight.ndim == 1: - if _rmsnorm_has_unit_offset(smooth): - weight.copy_((1.0 + weight) / s - 1.0) - else: - weight.div_(s) - else: - weight[-s.size(0) :].div_(s.view(-1, 1)) - - if getattr(smooth, "bias", None) is not None: - smooth.bias.data.div_(s) + fold_scales_into_smooth_layer(smooth, scales) # ── Weight clipping (search best per-group clip + hard-clamp) ───────────── diff --git a/auto_round/algorithms/transforms/awq/smooth.py b/auto_round/algorithms/transforms/awq/smooth.py new file mode 100644 index 0000000000..68e2d61c89 --- /dev/null +++ b/auto_round/algorithms/transforms/awq/smooth.py @@ -0,0 +1,249 @@ +# Copyright (c) 2026 Intel Corporation +# +# 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. +from __future__ import annotations + +import inspect +import re +from typing import TYPE_CHECKING + +import torch + +from auto_round.algorithms.transforms.smoothing.groups import SmoothGroup +from auto_round.data_type.utils import reshape_pad_tensor_by_group_size, revert_tensor_by_pad + +if TYPE_CHECKING: + from auto_round.algorithms.transforms.awq.base import AWQTransform + from auto_round.algorithms.transforms.awq.mappings import ResolvedMapping + + +# Known normalization classes whose ``forward`` computes +# ``output = (1 + weight) * x_norm`` (Gemma-style "unit-offset" RMSNorm) rather +# than the standard ``output = weight * x_norm``. Folding an AWQ smoothing scale +# ``s`` into such a layer requires ``weight <- (1 + weight) / s - 1`` instead of +# ``weight <- weight / s``; using the wrong fold silently breaks AWQ's output +# invariance and severely degrades accuracy (e.g. Qwen3.5, Gemma2/3, Qwen3-Next). +_UNIT_OFFSET_RMSNORM_NAMES = frozenset( + { + "GemmaRMSNorm", + "Gemma2RMSNorm", + "Gemma3RMSNorm", + "Gemma3TextRMSNorm", + "Qwen3_5RMSNorm", + "Qwen3_5MoeRMSNorm", + "Qwen3NextRMSNorm", + } +) + +# Detects ``1 + self.weight`` / ``self.weight + 1`` in a norm's forward source. +_UNIT_OFFSET_SRC_RE = re.compile(r"1(\.0)?\s*\+\s*self\.weight|self\.weight(\.float\(\))?\s*\+\s*1") + +# Cache the unit-offset decision per norm class to avoid repeated source parsing. +_unit_offset_cache: dict[type, bool] = {} + + +def _rmsnorm_has_unit_offset(module: torch.nn.Module) -> bool: + """Return True if ``module`` applies a Gemma-style ``(1 + weight)`` gain. + + Uses a fast class-name allowlist, falling back to source inspection of the + module's ``forward`` so newly-added Gemma-style norms are detected without a + code change. Result is cached per class. + """ + cls = type(module) + cached = _unit_offset_cache.get(cls) + if cached is not None: + return cached + result = cls.__name__ in _UNIT_OFFSET_RMSNORM_NAMES + if not result: + try: + src = inspect.getsource(cls.forward) + result = bool(_UNIT_OFFSET_SRC_RE.search(src)) + except (OSError, TypeError): + result = False + _unit_offset_cache[cls] = result + return result + + +def smooth_group_from_mapping(mapping: ResolvedMapping) -> SmoothGroup: + return SmoothGroup( + key=mapping.smooth_name, + projection_names=tuple(mapping.balance_names), + projections=tuple(mapping.balance_layers), + projection_input_key=mapping.activation_hook_target or mapping.balance_names[0], + projection_input_module=mapping.balance_layers[0], + evaluation_input_key=mapping.parent_name, + evaluation_module=mapping.parent, + ) + + +class AWQSmoothStrategy: + """AWQ statistics, ratio candidates, QDQ loss and scale folding policy.""" + + def __init__(self, owner: AWQTransform, mapping: ResolvedMapping) -> None: + self.owner = owner + self.mapping = mapping + self._candidates = None + self._score = None + + def prepare(self, group: SmoothGroup, x_mean: torch.Tensor) -> None: + mapping = self.mapping + owner = self.owner + device = mapping.balance_layers[0].weight.device + x_mean = x_mean.to(device) + + bl_params = {bl: owner._qdq_tool.resolve_params(bl) for bl in mapping.balance_layers} + group_size = owner._normalize_group_size(bl_params[mapping.balance_layers[0]]["group_size"], -1) + if owner.duo_scaling is not False: + w_mean = owner._compute_layer_means(mapping.balance_layers, group_size).to(device) + + parent_kwargs_list = owner._parent_args_cache.get(mapping.parent, []) + use_parent_forward = len(parent_kwargs_list) > 0 + + if use_parent_forward: + fp16_outputs = owner._run_parent_samples( + mapping.parent, + parent_kwargs_list, + offload_to_cpu=owner._smooth_batch_size is not None, + ) + if not fp16_outputs or all(f.numel() == 0 for f in fp16_outputs): + use_parent_forward = False + + orig_state = {bl: bl.weight.data.clone() for bl in mapping.balance_layers} + if not use_parent_forward: + orig_weights = orig_state # same reference is fine + + # Resolve each balance layer's quant functions once, then reuse them in + # the grid-search loop. Normal AWQ flow requires one mapping to have + # compatible quant params, but keeping this per-layer avoids hidden + # coupling to the first layer and makes direct calls robust. + bl_quant_funcs = {bl: owner._qdq_tool.resolve_quant_funcs(bl_params[bl]) for bl in mapping.balance_layers} + + def candidates(): + for ratio, use_duo in owner._get_grid_search_params(): + if use_duo: + scales = (x_mean.pow(ratio) / (w_mean.pow(1 - ratio) + 1e-4)).clamp(min=1e-4) + else: + scales = x_mean.pow(ratio).clamp(min=1e-4).view(-1) + scales = scales / (scales.max() * scales.min()).sqrt() + scales[torch.isinf(scales)] = 1 + scales[torch.isnan(scales)] = 1 + yield ratio, scales + + def score(candidate): + _, scales = candidate + scales_view = scales.view(1, -1).to(device) + if use_parent_forward: + # Quantize each balance layer's smoothed weight and write the + # de-smoothed result back, so the parent forward below sees the + # weights the layer would actually compute with. + for bl in mapping.balance_layers: + quant_func, opt_quant_func = bl_quant_funcs[bl] + w_qdq = owner._qdq_tool.qdq( + orig_state[bl] * scales_view, + bl_params[bl], + quant_func=quant_func, + opt_quant_func=opt_quant_func, + imatrix=getattr(bl, "imatrix", None), + ) + bl.weight.data = (w_qdq / scales_view).to(bl.weight.dtype) + + total_loss = owner._compute_parent_loss(mapping.parent, parent_kwargs_list, fp16_outputs) + for bl in mapping.balance_layers: + bl.weight.data.copy_(orig_state[bl]) + else: + total_loss = 0.0 + for bl in mapping.balance_layers: + quant_func, opt_quant_func = bl_quant_funcs[bl] + w_orig = orig_weights[bl].to(device) + w_qdq = owner._qdq_tool.qdq( + w_orig * scales_view, + bl_params[bl], + quant_func=quant_func, + opt_quant_func=opt_quant_func, + imatrix=getattr(bl, "imatrix", None), + ) + total_loss += (w_orig - w_qdq / scales_view).pow(2).sum().item() + + return total_loss + + self._candidates = candidates + self._score = score + + def candidates(self): + return self._candidates() + + def score(self, candidate): + return self._score(candidate) + + def apply(self, candidate) -> None: + self.owner._apply_scales(self.mapping, candidate[1]) + + def clear(self) -> None: + # Drop closure-owned snapshots and outputs; shared parent calls remain + # available to subsequent mappings and smoothing passes. + self._candidates = None + self._score = None + + +def get_grid_search_params(n_grid: int, duo_scaling: bool | str) -> list[tuple[float, bool]]: + """Return (ratio, use_duo_scaling) tuples for the grid search.""" + match duo_scaling: + case "both": + n = max(int(n_grid / 2), 2) + return [(idx / (n - 1), duo) for idx in range(n) for duo in [False, True]] + case False: + n = max(n_grid, 2) + return [(idx / (n - 1), False) for idx in range(n)] + case True: + n = max(n_grid, 3) + return [(0.0, False)] + [(idx / (n - 2), True) for idx in range(n - 1)] + case _: + raise ValueError(f"Unexpected duo_scaling value: {duo_scaling!r}") + + +def compute_layer_means(layers: list[torch.nn.Module], group_size: int) -> torch.Tensor: + """Per-channel mean of normalised weights across all balance layers.""" + weight = torch.cat([m.weight.detach().float() for m in layers], dim=0) + org_shape = weight.shape + gs = group_size if group_size is not None and group_size > 0 else org_shape[1] + weight, _, pad_len = reshape_pad_tensor_by_group_size(weight, gs) + w_scale = weight.abs() / (weight.abs().amax(dim=1, keepdim=True) + 1e-6) + w_scale = revert_tensor_by_pad(w_scale, orig_shape=org_shape, pad_len=pad_len) + return w_scale.mean(0) + + +@torch.no_grad() +def fold_scales_into_smooth_layer(smooth: torch.nn.Module, scales: torch.Tensor) -> None: + """Divide a smooth layer's output by ``scales`` to offset balance scaling. + + Dispatches on the smooth layer's weight layout: + + * 1-D norm weight with a Gemma-style ``(1 + weight)`` gain: folded as + ``weight <- (1 + weight) / s - 1`` to preserve output invariance. + * 1-D standard norm weight: folded as ``weight <- weight / s``. + * 2-D linear weight: its trailing ``s.numel()`` output rows are divided. + + Any bias is always divided by ``s``. + """ + s = scales.to(smooth.weight.device) + weight = smooth.weight.data + if weight.ndim == 1: + if _rmsnorm_has_unit_offset(smooth): + weight.copy_((1.0 + weight) / s - 1.0) + else: + weight.div_(s) + else: + weight[-s.size(0) :].div_(s.view(-1, 1)) + + if getattr(smooth, "bias", None) is not None: + smooth.bias.data.div_(s) diff --git a/auto_round/algorithms/transforms/smoothing/__init__.py b/auto_round/algorithms/transforms/smoothing/__init__.py new file mode 100644 index 0000000000..b5da2b00c8 --- /dev/null +++ b/auto_round/algorithms/transforms/smoothing/__init__.py @@ -0,0 +1,20 @@ +# Copyright (c) 2026 Intel Corporation +# +# 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. + +"""Extensible smoothing infrastructure independent of model and quantizer.""" + +from auto_round.algorithms.transforms.smoothing.engine import SmoothEngine, SmoothResult, SmoothStrategy +from auto_round.algorithms.transforms.smoothing.groups import SmoothGroup + +__all__ = ["SmoothEngine", "SmoothGroup", "SmoothResult", "SmoothStrategy"] diff --git a/auto_round/algorithms/transforms/smoothing/calibration.py b/auto_round/algorithms/transforms/smoothing/calibration.py new file mode 100644 index 0000000000..396ad3eb03 --- /dev/null +++ b/auto_round/algorithms/transforms/smoothing/calibration.py @@ -0,0 +1,99 @@ +# Copyright (c) 2026 Intel Corporation +# +# 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. + +from __future__ import annotations + +import random +from collections.abc import MutableMapping +from dataclasses import dataclass, field +from typing import Any, Generic, TypeVar + +import torch + +from auto_round.logger import logger +from auto_round.utils.model import map_nested_tensors + +_GroupT = TypeVar("_GroupT") + + +def detach_to_cpu(value: Any) -> Any: + return map_nested_tensors(value, lambda tensor: tensor.detach().to("cpu", copy=True)) + + +def move_to_device(value: Any, device: torch.device, dtype: torch.dtype | None = None) -> Any: + def move(tensor: torch.Tensor) -> torch.Tensor: + target_dtype = dtype if dtype is not None and tensor.is_floating_point() else tensor.dtype + return tensor.to(device=device, dtype=target_dtype) + + return map_nested_tensors(value, move) + + +@dataclass +class CapturedEvaluation: + args: tuple[Any, ...] + kwargs: dict[str, Any] + output: Any + + +@dataclass +class SmoothGroupCalibration(Generic[_GroupT]): + group: _GroupT + limit: int + projection_inputs: list[torch.Tensor] = field(default_factory=list) + evaluation_calls: list[CapturedEvaluation] = field(default_factory=list) + seen_calls: int = 0 + pending_slot: int | None = None + pending_input: torch.Tensor | None = None + random: random.Random = field(default_factory=lambda: random.Random(0)) + + def begin_call(self, inputs: torch.Tensor) -> None: + self.seen_calls += 1 + if len(self.projection_inputs) < self.limit: + slot = len(self.projection_inputs) + else: + candidate = self.random.randrange(self.seen_calls) + slot = candidate if candidate < self.limit else None + self.pending_slot = slot + self.pending_input = detach_to_cpu(inputs) if slot is not None else None + + def finish_call(self, args: tuple[Any, ...], kwargs: dict[str, Any], output: Any) -> None: + slot = self.pending_slot + captured_input = self.pending_input + self.pending_slot = None + self.pending_input = None + if slot is None or captured_input is None: + return + captured = CapturedEvaluation(detach_to_cpu(args), detach_to_cpu(kwargs), detach_to_cpu(output)) + if slot == len(self.projection_inputs): + self.projection_inputs.append(captured_input) + self.evaluation_calls.append(captured) + else: + self.projection_inputs[slot] = captured_input + self.evaluation_calls[slot] = captured + + def clear(self) -> None: + """Release retained and pending calls, including failed captures.""" + self.projection_inputs.clear() + self.evaluation_calls.clear() + self.pending_input = None + self.pending_slot = None + self.seen_calls = 0 + + +def clear_caches(*caches: MutableMapping, scope: str) -> None: + """Release algorithm-owned caches with common lifecycle logging.""" + entries = sum(len(cache) for cache in caches) + for cache in caches: + cache.clear() + logger.debug("Smooth caches cleared for %s: entries=%d", scope, entries) diff --git a/auto_round/algorithms/transforms/smoothing/engine.py b/auto_round/algorithms/transforms/smoothing/engine.py new file mode 100644 index 0000000000..1907ad6a09 --- /dev/null +++ b/auto_round/algorithms/transforms/smoothing/engine.py @@ -0,0 +1,113 @@ +# Copyright (c) 2026 Intel Corporation +# +# 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. + +from __future__ import annotations + +from collections.abc import Iterable, Iterator +from contextlib import contextmanager +from dataclasses import dataclass +from typing import Generic, Protocol, TypeVar + +import torch + +from auto_round.algorithms.transforms.smoothing.groups import SmoothGroup +from auto_round.algorithms.transforms.smoothing.search import search_candidates + +_CalibrationT_contra = TypeVar("_CalibrationT_contra", contravariant=True) +_CandidateT = TypeVar("_CandidateT") +_AppliedT_co = TypeVar("_AppliedT_co", covariant=True) + + +class SmoothStrategy(Protocol[_CalibrationT_contra, _CandidateT, _AppliedT_co]): + """Extension contract; statistics, candidates, loss and deployment are policies. + + Each group uses a separate strategy instance. Strategies own temporary + mutation handling. clear() releases only strategy-owned + state, preserving calibration caches shared with other groups or passes. + """ + + def prepare(self, group: SmoothGroup, calibration: _CalibrationT_contra) -> None: ... + + def candidates(self) -> Iterable[_CandidateT]: ... + + def score(self, candidate: _CandidateT) -> float | torch.Tensor: ... + + def apply(self, candidate: _CandidateT) -> _AppliedT_co: ... + + def clear(self) -> None: ... + + +@dataclass(frozen=True) +class SmoothResult(Generic[_CandidateT]): + group: SmoothGroup + candidate: _CandidateT + error: float + + +class SmoothEngine: + """Algorithm-independent smoothing with separate search and apply phases.""" + + @contextmanager + def session( + self, strategy: SmoothStrategy[_CalibrationT_contra, _CandidateT, _AppliedT_co] + ) -> Iterator[SmoothStrategy[_CalibrationT_contra, _CandidateT, _AppliedT_co]]: + try: + yield strategy + finally: + strategy.clear() + + @torch.no_grad() + def search( + self, + group: SmoothGroup, + calibration: _CalibrationT_contra, + strategy: SmoothStrategy[_CalibrationT_contra, _CandidateT, _AppliedT_co], + *, + require_finite: bool = True, + log_search: bool = True, + ) -> SmoothResult[_CandidateT]: + strategy.prepare(group, calibration) + candidate, error = search_candidates( + strategy.candidates(), + strategy.score, + module_name=group.key, + require_finite=require_finite, + log_search=log_search, + ) + return SmoothResult(group, candidate, error) + + @torch.no_grad() + def apply( + self, + group: SmoothGroup, + strategy: SmoothStrategy[_CalibrationT_contra, _CandidateT, _AppliedT_co], + result: SmoothResult[_CandidateT], + ) -> _AppliedT_co: + """Apply a result using the same group and strategy instances used by search(). + + Equivalent reconstructed groups are not interchangeable; the strategy + retains state prepared for the original group. + """ + if result.group is not group: + raise ValueError("Smooth result belongs to a different group.") + return strategy.apply(result.candidate) + + def run( + self, + group: SmoothGroup, + calibration: _CalibrationT_contra, + strategy: SmoothStrategy[_CalibrationT_contra, _CandidateT, _AppliedT_co], + ) -> _AppliedT_co: + with self.session(strategy): + return self.apply(group, strategy, self.search(group, calibration, strategy)) diff --git a/auto_round/algorithms/transforms/smoothing/errors.py b/auto_round/algorithms/transforms/smoothing/errors.py new file mode 100644 index 0000000000..a2d2b41793 --- /dev/null +++ b/auto_round/algorithms/transforms/smoothing/errors.py @@ -0,0 +1,23 @@ +# Copyright (c) 2026 Intel Corporation +# +# 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. + +"""Failure categories shared by smoothing algorithms.""" + + +class InvalidSmoothCandidateError(ValueError): + """A numerical candidate cannot be deployed; search may skip it.""" + + +class NoFiniteCandidateError(ValueError): + """No candidate is usable; the algorithm decides its fallback.""" diff --git a/auto_round/algorithms/transforms/smoothing/groups.py b/auto_round/algorithms/transforms/smoothing/groups.py new file mode 100644 index 0000000000..625a283111 --- /dev/null +++ b/auto_round/algorithms/transforms/smoothing/groups.py @@ -0,0 +1,46 @@ +# Copyright (c) 2026 Intel Corporation +# +# 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. + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +import torch + +from auto_round.algorithms.transforms.smoothing.replay import filter_supported_kwargs, normalize_tensors + + +@dataclass(frozen=True) +class SmoothGroup: + """Targets sharing an input scale and an evaluation boundary. + + Model adapters own discovery. Shared scale does not imply shared low-rank + factors, and strategies own target-specific dimension checks.""" + + key: str + projection_names: tuple[str, ...] + projections: tuple[torch.nn.Module, ...] + projection_input_key: str + projection_input_module: torch.nn.Module + evaluation_input_key: str + evaluation_module: torch.nn.Module + output_indices: tuple[int, ...] | None = None + + def filter_evaluation_kwargs(self, kwargs: Mapping[str, Any]) -> dict[str, Any]: + return filter_supported_kwargs(self.evaluation_module, kwargs) + + def normalize_output(self, output: Any) -> tuple[torch.Tensor, ...]: + return normalize_tensors(output, self.output_indices) diff --git a/auto_round/algorithms/transforms/smoothing/replay.py b/auto_round/algorithms/transforms/smoothing/replay.py new file mode 100644 index 0000000000..5441046870 --- /dev/null +++ b/auto_round/algorithms/transforms/smoothing/replay.py @@ -0,0 +1,89 @@ +# Copyright (c) 2026 Intel Corporation +# +# 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. + +from __future__ import annotations + +import inspect +from collections.abc import Mapping +from contextlib import contextmanager +from typing import Any + +import torch + + +def normalize_tensors(output: Any, indices: tuple[int, ...] | None = None) -> tuple[torch.Tensor, ...]: + """Return floating-point tensors used by the smooth output-error objective.""" + if indices is not None: + if not isinstance(output, (tuple, list)): + raise TypeError("Indexed Smooth output must be a tuple or list.") + output = tuple(output[index] for index in indices) + + tensors = [] + + def collect(value): + if torch.is_tensor(value): + if value.is_floating_point(): + tensors.append(value) + elif isinstance(value, Mapping): + for item in value.values(): + collect(item) + elif isinstance(value, (tuple, list)): + for item in value: + collect(item) + + collect(output) + if not tensors: + raise ValueError("Smooth evaluation produced no floating-point tensors.") + return tuple(tensors) + + +def filter_supported_kwargs(module: torch.nn.Module, kwargs: Mapping[str, Any]) -> dict[str, Any]: + parameters = inspect.signature(module.forward).parameters + if any(parameter.kind is inspect.Parameter.VAR_KEYWORD for parameter in parameters.values()): + return dict(kwargs) + return {name: value for name, value in kwargs.items() if name in parameters} + + +def output_squared_error( + actual: tuple[torch.Tensor, ...], + reference: tuple[torch.Tensor, ...], + *, + accumulator: torch.Tensor | None = None, +) -> torch.Tensor: + """Sum output errors using FP32 differences and an FP64 CPU accumulator. + + When supplied, accumulator must be a scalar CPU tensor of dtype torch.float64. + It is updated in place and returned; otherwise a zero accumulator is created. + """ + if len(actual) != len(reference): + raise ValueError("Smooth output tensor count changed.") + error = torch.zeros((), dtype=torch.float64) if accumulator is None else accumulator + for actual_tensor, reference_tensor in zip(actual, reference): + if actual_tensor.shape != reference_tensor.shape: + raise ValueError("Smooth output tensor shape changed.") + error += torch.sum((actual_tensor.float() - reference_tensor.float()).square()).double().cpu() + return error + + +@contextmanager +def temporary_modules(root: torch.nn.Module, replacements: list[tuple[str, torch.nn.Module]]): + """Restore original modules even when installation or evaluation fails.""" + originals = [(name, root.get_submodule(name)) for name, _ in replacements] + try: + for name, module in replacements: + root.set_submodule(name, module) + yield + finally: + for name, module in originals: + root.set_submodule(name, module) diff --git a/auto_round/algorithms/transforms/smoothing/scale.py b/auto_round/algorithms/transforms/smoothing/scale.py new file mode 100644 index 0000000000..31422ac08e --- /dev/null +++ b/auto_round/algorithms/transforms/smoothing/scale.py @@ -0,0 +1,77 @@ +# Copyright (c) 2026 Intel Corporation +# +# 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. + +from __future__ import annotations + +from dataclasses import dataclass + +import torch + +from auto_round.algorithms.transforms.smoothing.errors import InvalidSmoothCandidateError + + +@dataclass(frozen=True) +class SmoothScaleStats: + minimum: float + maximum: float + ratio: float + below_min_count: int + above_max_count: int + + +def absmax_channel_span(tensor: torch.Tensor, channels_dim: int) -> torch.Tensor: + """Return per-channel absolute maxima over every non-channel dimension.""" + if tensor.ndim == 0: + raise ValueError("Cannot calculate a channel span for a scalar tensor.") + channels_dim %= tensor.ndim + moved = tensor.detach().movedim(channels_dim, -1) + return moved.abs().reshape(-1, moved.shape[-1]).amax(dim=0).to(torch.float32) + + +def validate_smooth_scale_for_deployment( + scale: torch.Tensor, + *, + dtype: torch.dtype, + module_name: str, +) -> torch.Tensor: + """Validate a smooth scale after materialization in its deployment dtype.""" + deployed = scale.to(dtype=dtype) + reciprocal = deployed.reciprocal() + if ( + not bool(torch.isfinite(deployed).all()) + or not bool((deployed > 0).all()) + or not bool(torch.isfinite(reciprocal).all()) + or not bool((reciprocal > 0).all()) + ): + raise InvalidSmoothCandidateError(f"Smooth scale is not deployable for {module_name!r} in dtype {dtype}.") + return deployed + + +def summarize_smooth_scale( + scale: torch.Tensor, + *, + low_threshold: float = 1e-3, + high_threshold: float = 20.0, +) -> SmoothScaleStats: + """Summarize factor range and deployment-risk threshold counts.""" + values = scale.detach().to(device="cpu", dtype=torch.float32) + minimum = values.amin().item() + maximum = values.amax().item() + return SmoothScaleStats( + minimum=minimum, + maximum=maximum, + ratio=maximum / minimum, + below_min_count=int((values < low_threshold).sum().item()), + above_max_count=int((values > high_threshold).sum().item()), + ) diff --git a/auto_round/algorithms/transforms/smoothing/search.py b/auto_round/algorithms/transforms/smoothing/search.py new file mode 100644 index 0000000000..7a86c67bdf --- /dev/null +++ b/auto_round/algorithms/transforms/smoothing/search.py @@ -0,0 +1,99 @@ +# Copyright (c) 2026 Intel Corporation +# +# 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. + +from __future__ import annotations + +import math +from collections.abc import Callable, Iterable +from typing import TypeVar + +import torch + +from auto_round.algorithms.transforms.smoothing.errors import InvalidSmoothCandidateError, NoFiniteCandidateError +from auto_round.logger import logger + +_CandidateT = TypeVar("_CandidateT") + + +def _select_best_layer_result( + candidates: Iterable[tuple[_CandidateT, float | torch.Tensor]], + *, + module_name: str, + require_finite: bool = True, +) -> tuple[_CandidateT, float]: + """Keep the candidate and its score together, including a valid None candidate.""" + best = None + for candidate, error in candidates: + error_value = error.item() if torch.is_tensor(error) else float(error) + if (not require_finite or math.isfinite(error_value)) and error_value < (best[1] if best else float("inf")): + best = (candidate, error_value) + if best is None: + raise NoFiniteCandidateError(f"Smooth search produced no finite candidate for {module_name!r}.") + return best + + +def select_best_layer_candidate( + candidates: Iterable[tuple[_CandidateT, float | torch.Tensor]], + *, + module_name: str, +) -> _CandidateT: + """Select the lowest finite error, retaining the first exact tie.""" + return _select_best_layer_result(candidates, module_name=module_name)[0] + + +def search_candidates( + candidates: Iterable[_CandidateT], + score: Callable[[_CandidateT], float | torch.Tensor], + *, + module_name: str, + on_invalid: Callable[[_CandidateT, Exception], None] | None = None, + require_finite: bool = True, + log_search: bool = True, +) -> tuple[_CandidateT, float]: + """Score candidates incrementally and retain only the first lowest-error candidate. + + Candidate validity and diagnostics can preserve algorithm-specific behavior. + Execution errors propagate; temporary mutation handling belongs to the strategy. + """ + + def scored_candidates(): + for index, candidate in enumerate(candidates): + try: + error = score(candidate) + error = error.item() if torch.is_tensor(error) else float(error) + if require_finite and not math.isfinite(error): + raise InvalidSmoothCandidateError(f"Nonfinite candidate error: {error}.") + except InvalidSmoothCandidateError as exc: + if log_search: + logger.debug("Smooth search skipped candidate %d for %s: %s", index, module_name, exc) + if on_invalid is not None: + on_invalid(candidate, exc) + error = float("inf") + except Exception: + if log_search: + logger.error("Smooth search failed at candidate %d for %s", index, module_name, exc_info=True) + raise + yield candidate, error + + try: + selected, error = _select_best_layer_result( + scored_candidates(), module_name=module_name, require_finite=require_finite + ) + except NoFiniteCandidateError: + if log_search: + logger.warning("Smooth search produced no finite candidate for %s", module_name) + raise + if log_search: + logger.debug("Smooth search selected a candidate for %s: error=%.6g", module_name, error) + return selected, error diff --git a/auto_round/algorithms/transforms/svdquant/apply.py b/auto_round/algorithms/transforms/svdquant/apply.py index fffff0beba..9fd6739255 100644 --- a/auto_round/algorithms/transforms/svdquant/apply.py +++ b/auto_round/algorithms/transforms/svdquant/apply.py @@ -15,8 +15,7 @@ from __future__ import annotations import math -import random -from dataclasses import dataclass, field +from contextlib import ExitStack from functools import partial from typing import Any @@ -25,6 +24,13 @@ import auto_round.algorithms.transforms.svdquant.residual as residual_module from auto_round.algorithms.registry import register_pipeline_member from auto_round.algorithms.transforms.base import BasePreprocessor +from auto_round.algorithms.transforms.smoothing.calibration import ( + SmoothGroupCalibration, + clear_caches, +) +from auto_round.algorithms.transforms.smoothing.calibration import move_to_device as _move_to_device +from auto_round.algorithms.transforms.smoothing.engine import SmoothEngine +from auto_round.algorithms.transforms.smoothing.replay import output_squared_error, temporary_modules from auto_round.algorithms.transforms.svdquant.config import SVDQuantConfig from auto_round.algorithms.transforms.svdquant.residual import ( ActivationQuantScheme, @@ -34,19 +40,14 @@ ) from auto_round.algorithms.transforms.svdquant.smooth import ( SmoothCandidate, - absmax_channel_span, - build_alpha_beta_candidates, - build_smooth_scale, - select_best_layer_candidate, + SVDQuantSmoothStrategy, summarize_smooth_scale, - validate_smooth_scale_for_deployment, ) from auto_round.algorithms.transforms.svdquant.smooth_adapters import SmoothSearchGroup, discover_svdquant_groups from auto_round.algorithms.transforms.svdquant.wrapper import SVDQuantLinear from auto_round.logger import logger from auto_round.schemes import QuantizationScheme from auto_round.utils.device_manager import device_manager -from auto_round.utils.model import map_nested_tensors _SCHEME_ATTRS = set(QuantizationScheme.get_attributes()) _RUNTIME_QUANT_ATTRS = {"scale_dtype", "weight_global_scale", "tuning_device"} @@ -73,62 +74,6 @@ def _select_svd_driver(device: torch.device) -> str | None: return driver -def _detach_to_cpu(value: Any) -> Any: - return map_nested_tensors(value, lambda tensor: tensor.detach().to("cpu", copy=True)) - - -def _move_to_device(value: Any, device: torch.device, dtype: torch.dtype | None = None) -> Any: - def move(tensor: torch.Tensor) -> torch.Tensor: - target_dtype = dtype if dtype is not None and tensor.is_floating_point() else tensor.dtype - return tensor.to(device=device, dtype=target_dtype) - - return map_nested_tensors(value, move) - - -@dataclass -class CapturedEvaluation: - args: tuple[Any, ...] - kwargs: dict[str, Any] - output: Any - - -@dataclass -class SmoothGroupCalibration: - group: SmoothSearchGroup - limit: int - projection_inputs: list[torch.Tensor] = field(default_factory=list) - evaluation_calls: list[CapturedEvaluation] = field(default_factory=list) - seen_calls: int = 0 - pending_slot: int | None = None - pending_input: torch.Tensor | None = None - random: random.Random = field(default_factory=lambda: random.Random(0)) - - def begin_call(self, inputs: torch.Tensor) -> None: - self.seen_calls += 1 - if len(self.projection_inputs) < self.limit: - slot = len(self.projection_inputs) - else: - candidate = self.random.randrange(self.seen_calls) - slot = candidate if candidate < self.limit else None - self.pending_slot = slot - self.pending_input = _detach_to_cpu(inputs) if slot is not None else None - - def finish_call(self, args: tuple[Any, ...], kwargs: dict[str, Any], output: Any) -> None: - slot = self.pending_slot - captured_input = self.pending_input - self.pending_slot = None - self.pending_input = None - if slot is None or captured_input is None: - return - captured = CapturedEvaluation(_detach_to_cpu(args), _detach_to_cpu(kwargs), _detach_to_cpu(output)) - if slot == len(self.projection_inputs): - self.projection_inputs.append(captured_input) - self.evaluation_calls.append(captured) - else: - self.projection_inputs[slot] = captured_input - self.evaluation_calls[slot] = captured - - @register_pipeline_member(SVDQuantConfig) class SVDQuantTransform(BasePreprocessor): """Split target linears into a quantized residual and an FP low-rank branch.""" @@ -288,68 +233,57 @@ def finalize_run(self) -> None: self._block_groups.clear() def _clear_smooth_calibration(self) -> None: - self._smooth_calibration.clear() + for calibration in self._smooth_calibration.values(): + calibration.clear() + clear_caches(self._smooth_calibration, scope="SVDQuant calibration") def _pre_quantize_smoothed_block(self, block: torch.nn.Module, groups: list[SmoothSearchGroup]) -> None: if not self._smooth_calibration: raise ValueError("SVDQuant smooth calibration inputs are missing for the current block.") local_names = {id(module): name for name, module in block.named_modules() if name} + engine = SmoothEngine() try: - selected_scales = {} - for group in groups: - calibration = self._smooth_calibration.get(group.key) - if calibration is None or not calibration.projection_inputs or not calibration.evaluation_calls: - raise ValueError(f"SVDQuant smooth calibration inputs are missing for group {group.key!r}.") - selected_scales[group.key] = self._search_group_scale(calibration, block, local_names) - - replacements = [] - for group in groups: - calibration = self._smooth_calibration[group.key] - wrappers = self._decompose_smoothed_group(calibration, selected_scales[group.key], block, local_names) - for projection, wrapper in zip(group.projections, wrappers): - local_name = local_names.get(id(projection)) - if local_name is None: - raise ValueError(f"SVDQuant could not locate projection {self._module_name(projection)!r}.") - replacements.append((local_name, wrapper)) - for local_name, wrapper in replacements: - _set_child_module(block, local_name, wrapper) + with ExitStack() as sessions: + self._search_and_apply_smooth_groups(engine, sessions, block, groups, local_names) finally: self._clear_smooth_calibration() + def _search_and_apply_smooth_groups(self, engine, sessions, block, groups, local_names) -> None: + selected = {} + for group in groups: + calibration = self._smooth_calibration.get(group.key) + if calibration is None or not calibration.projection_inputs or not calibration.evaluation_calls: + raise ValueError(f"SVDQuant smooth calibration inputs are missing for group {group.key!r}.") + strategy = SVDQuantSmoothStrategy(self, block, local_names) + sessions.enter_context(engine.session(strategy)) + result = engine.search(group, calibration, strategy) + self._log_selected_smooth_candidate(group.key, result.candidate, result.error) + selected[group.key] = (strategy, result) + + replacements = [] + for group in groups: + strategy, result = selected[group.key] + wrappers = engine.apply(group, strategy, result) + for projection, wrapper in zip(group.projections, wrappers): + local_name = local_names.get(id(projection)) + if local_name is None: + raise ValueError(f"SVDQuant could not locate projection {self._module_name(projection)!r}.") + replacements.append((local_name, wrapper)) + for local_name, wrapper in replacements: + _set_child_module(block, local_name, wrapper) + def _search_group_scale( self, calibration: SmoothGroupCalibration, block: torch.nn.Module, local_names: dict[int, str], ) -> torch.Tensor: - group = calibration.group - device = group.projections[0].weight.device - x_span = torch.stack([absmax_channel_span(inputs, -1) for inputs in calibration.projection_inputs], dim=0).amax( - dim=0 - ) - weights = [ - projection.weight.detach().to(device=device, dtype=torch.float32) for projection in group.projections - ] - w_span = absmax_channel_span(torch.cat(weights, dim=0), 1).cpu() - scored = [] - for alpha, beta in build_alpha_beta_candidates(self.config.smooth_num_grids): - scale = build_smooth_scale(x_span, w_span, alpha, beta, eps=self.config.smooth_eps) - try: - scale = validate_smooth_scale_for_deployment( - scale, dtype=group.projections[0].weight.dtype, module_name=group.key - ).to(torch.float32) - error = self._score_group_wrappers( - calibration, self._candidate_group_wrappers(group, scale), block, local_names - ) - except (RuntimeError, ValueError, TypeError) as exc: - logger.debug("Skipping SVDQuant smooth candidate (%s, %s) for %s: %s", alpha, beta, group.key, exc) - error = float("inf") - candidate = SmoothCandidate(alpha, beta, scale) - scored.append((candidate, error)) - selected = select_best_layer_candidate(scored, module_name=group.key) - error = next(error for candidate, error in reversed(scored) if candidate is selected) - self._log_selected_smooth_candidate(group.key, selected, error) - return selected.scale + engine = SmoothEngine() + strategy = SVDQuantSmoothStrategy(self, block, local_names) + with engine.session(strategy): + result = engine.search(calibration.group, calibration, strategy) + self._log_selected_smooth_candidate(calibration.group.key, result.candidate, result.error) + return result.candidate.scale @staticmethod def _log_selected_smooth_candidate(module_name: str, candidate: SmoothCandidate, error: float) -> None: @@ -388,10 +322,8 @@ def _score_group_wrappers( local_name = local_names.get(id(projection)) if local_name is None: raise ValueError(f"SVDQuant could not locate projection {self._module_name(projection)!r}.") - replacements.append((local_name, projection, wrapper)) - try: - for local_name, _, wrapper in replacements: - _set_child_module(block, local_name, wrapper) + replacements.append((local_name, wrapper)) + with temporary_modules(block, replacements): error = torch.zeros((), dtype=torch.float64) for call in calibration.evaluation_calls: evaluation_module = group.evaluation_module @@ -403,16 +335,8 @@ def _score_group_wrappers( kwargs = group.filter_evaluation_kwargs(_move_to_device(call.kwargs, device, dtype)) actual = group.normalize_output(evaluation_module(*args, **kwargs)) reference = tuple(tensor.to(device) for tensor in group.normalize_output(call.output)) - if len(actual) != len(reference): - raise ValueError("SVDQuant smooth output tensor count changed.") - for actual_tensor, reference_tensor in zip(actual, reference): - if actual_tensor.shape != reference_tensor.shape: - raise ValueError("SVDQuant smooth output tensor shape changed.") - error += torch.sum((actual_tensor.float() - reference_tensor.float()).square()).double().cpu() + output_squared_error(actual, reference, accumulator=error) return error.item() - finally: - for local_name, projection, _ in replacements: - _set_child_module(block, local_name, projection) def _candidate_group_wrappers(self, group: SmoothSearchGroup, scale: torch.Tensor) -> list[SVDQuantLinear]: weights = [ @@ -511,11 +435,15 @@ def _iterate_smoothed_group_residual( activation_scheme=activation_scheme, ) error = self._score_group_wrappers(calibration, wrappers, block, local_names) - accepted = math.isfinite(error) and error <= best_error + accepted = math.isfinite(error) and error < best_error if accepted: best = (deployed_down.clone(), deployed_up.clone(), iteration) best_error = error - elif self.config.residual_early_stop and best is not None: + elif ( + self.config.residual_early_stop + and best is not None + and (not math.isfinite(error) or error > best_error) + ): logger.info( "SVDQuant residual early stop for %s at iteration %d: output error %.6g > best %.6g", group.key, diff --git a/auto_round/algorithms/transforms/svdquant/residual.py b/auto_round/algorithms/transforms/svdquant/residual.py index 461f6763b4..60a4e6b6d3 100644 --- a/auto_round/algorithms/transforms/svdquant/residual.py +++ b/auto_round/algorithms/transforms/svdquant/residual.py @@ -231,14 +231,14 @@ def iterate_residual_decomposition( deployed_up = up.to(low_rank_dtype) deployed_low_rank = deployed_up.to(weight.dtype) @ deployed_down.to(weight.dtype) error = torch.sum((weight - (quantized_residual + deployed_low_rank)).square()).item() - accepted = math.isfinite(error) and error <= best_error + accepted = math.isfinite(error) and error < best_error if accepted: best_down = deployed_down.clone() best_up = deployed_up.clone() best_error = error best_iteration = iteration - elif early_stop and best_iteration is not None: + elif early_stop and best_iteration is not None and (not math.isfinite(error) or error > best_error): break if best_down is None or best_up is None or best_iteration is None: diff --git a/auto_round/algorithms/transforms/svdquant/smooth.py b/auto_round/algorithms/transforms/svdquant/smooth.py index c3d0d4d9f2..996b0bfbdc 100644 --- a/auto_round/algorithms/transforms/svdquant/smooth.py +++ b/auto_round/algorithms/transforms/svdquant/smooth.py @@ -14,13 +14,37 @@ from __future__ import annotations -import math -from collections.abc import Iterable from dataclasses import dataclass -from typing import TypeVar +from typing import TYPE_CHECKING import torch +# Compatibility exports for existing callers. +from auto_round.algorithms.transforms.smoothing.scale import ( + SmoothScaleStats, + absmax_channel_span, + summarize_smooth_scale, + validate_smooth_scale_for_deployment, +) +from auto_round.algorithms.transforms.smoothing.search import select_best_layer_candidate + +if TYPE_CHECKING: + from auto_round.algorithms.transforms.smoothing.calibration import SmoothGroupCalibration + from auto_round.algorithms.transforms.smoothing.groups import SmoothGroup + from auto_round.algorithms.transforms.svdquant.apply import SVDQuantTransform + +__all__ = [ + "SVDQuantSmoothStrategy", + "SmoothCandidate", + "SmoothScaleStats", + "absmax_channel_span", + "build_alpha_beta_candidates", + "build_smooth_scale", + "select_best_layer_candidate", + "summarize_smooth_scale", + "validate_smooth_scale_for_deployment", +] + @dataclass(frozen=True) class SmoothCandidate: @@ -29,15 +53,6 @@ class SmoothCandidate: scale: torch.Tensor -@dataclass(frozen=True) -class SmoothScaleStats: - minimum: float - maximum: float - ratio: float - below_min_count: int - above_max_count: int - - def build_alpha_beta_candidates(num_grids: int) -> list[tuple[float, float]]: """Build identity, activation-only, and activation/weight balanced candidates.""" if type(num_grids) is not int or num_grids < 2: @@ -46,15 +61,6 @@ def build_alpha_beta_candidates(num_grids: int) -> list[tuple[float, float]]: return [(0.0, 0.0), *[(alpha, 0.0) for alpha in choices], *[(alpha, 1.0 - alpha) for alpha in choices]] -def absmax_channel_span(tensor: torch.Tensor, channels_dim: int) -> torch.Tensor: - """Return per-channel absolute maxima over every non-channel dimension.""" - if tensor.ndim == 0: - raise ValueError("Cannot calculate a channel span for a scalar tensor.") - channels_dim %= tensor.ndim - moved = tensor.detach().movedim(channels_dim, -1) - return moved.abs().reshape(-1, moved.shape[-1]).amax(dim=0).to(torch.float32) - - def build_smooth_scale( x_span: torch.Tensor, w_span: torch.Tensor, @@ -98,60 +104,59 @@ def build_smooth_scale( return scale -def validate_smooth_scale_for_deployment( - scale: torch.Tensor, - *, - dtype: torch.dtype, - module_name: str, -) -> torch.Tensor: - """Validate a smooth scale after materialization in its deployment dtype.""" - deployed = scale.to(dtype=dtype) - reciprocal = deployed.reciprocal() - if ( - not bool(torch.isfinite(deployed).all()) - or not bool((deployed > 0).all()) - or not bool(torch.isfinite(reciprocal).all()) - or not bool((reciprocal > 0).all()) - ): - raise ValueError(f"SVDQuant smooth scale is not deployable for {module_name!r} in dtype {dtype}.") - return deployed - - -def summarize_smooth_scale( - scale: torch.Tensor, - *, - low_threshold: float = 1e-3, - high_threshold: float = 20.0, -) -> SmoothScaleStats: - """Summarize factor range and deployment-risk threshold counts.""" - values = scale.detach().to(device="cpu", dtype=torch.float32) - minimum = values.amin().item() - maximum = values.amax().item() - return SmoothScaleStats( - minimum=minimum, - maximum=maximum, - ratio=maximum / minimum, - below_min_count=int((values < low_threshold).sum().item()), - above_max_count=int((values > high_threshold).sum().item()), - ) - - -_CandidateT = TypeVar("_CandidateT") - - -def select_best_layer_candidate( - candidates: Iterable[tuple[_CandidateT, float | torch.Tensor]], - *, - module_name: str, -) -> _CandidateT: - """Select the lowest finite-error candidate, preferring the later exact tie.""" - best_candidate = None - best_error = float("inf") - for candidate, error in candidates: - error_value = error.item() if torch.is_tensor(error) else float(error) - if math.isfinite(error_value) and error_value <= best_error: - best_candidate = candidate - best_error = error_value - if best_candidate is None: - raise ValueError(f"SVDQuant smooth search produced no finite candidate for {module_name!r}.") - return best_candidate +class SVDQuantSmoothStrategy: + """Absmax candidates and low-rank-aware reconstruction scoring policy.""" + + def __init__(self, owner: SVDQuantTransform, block: torch.nn.Module, local_names: dict[int, str]) -> None: + self.owner = owner + self.block = block + self.local_names = local_names + self._candidates = None + self._score = None + self.calibration = None + + def prepare(self, group: SmoothGroup, calibration: SmoothGroupCalibration) -> None: + self.calibration = calibration + owner = self.owner + block = self.block + local_names = self.local_names + group = calibration.group + device = group.projections[0].weight.device + x_span = torch.stack([absmax_channel_span(inputs, -1) for inputs in calibration.projection_inputs], dim=0).amax( + dim=0 + ) + weights = [ + projection.weight.detach().to(device=device, dtype=torch.float32) for projection in group.projections + ] + w_span = absmax_channel_span(torch.cat(weights, dim=0), 1).cpu() + candidates = ( + SmoothCandidate(alpha, beta, build_smooth_scale(x_span, w_span, alpha, beta, eps=owner.config.smooth_eps)) + for alpha, beta in build_alpha_beta_candidates(owner.config.smooth_num_grids) + ) + + def score(candidate): + scale = validate_smooth_scale_for_deployment( + candidate.scale, dtype=group.projections[0].weight.dtype, module_name=group.key + ).to(torch.float32) + # Keep the deployed precision in the selected candidate. + candidate.scale.copy_(scale) + return owner._score_group_wrappers( + calibration, owner._candidate_group_wrappers(group, scale), block, local_names + ) + + self._candidates = candidates + self._score = score + + def candidates(self): + return self._candidates + + def score(self, candidate: SmoothCandidate) -> float: + return self._score(candidate) + + def apply(self, candidate: SmoothCandidate): + return self.owner._decompose_smoothed_group(self.calibration, candidate.scale, self.block, self.local_names) + + def clear(self) -> None: + self._candidates = None + self._score = None + self.calibration = None diff --git a/auto_round/algorithms/transforms/svdquant/smooth_adapters/base.py b/auto_round/algorithms/transforms/svdquant/smooth_adapters/base.py index 9380fda9e6..230e5452af 100644 --- a/auto_round/algorithms/transforms/svdquant/smooth_adapters/base.py +++ b/auto_round/algorithms/transforms/svdquant/smooth_adapters/base.py @@ -14,61 +14,19 @@ from __future__ import annotations -import inspect -from collections.abc import Callable, Mapping +from collections.abc import Callable from dataclasses import dataclass -from typing import Any import torch -TargetPredicate = Callable[[str, torch.nn.Module], bool] - - -def normalize_tensors(output: Any, indices: tuple[int, ...] | None = None) -> tuple[torch.Tensor, ...]: - """Return floating-point tensors used by the smooth output-error objective.""" - if indices is not None: - if not isinstance(output, (tuple, list)): - raise TypeError("Indexed SVDQuant smooth output must be a tuple or list.") - output = tuple(output[index] for index in indices) - - tensors = [] - - def collect(value): - if torch.is_tensor(value): - if value.is_floating_point(): - tensors.append(value) - elif isinstance(value, Mapping): - for item in value.values(): - collect(item) - elif isinstance(value, (tuple, list)): - for item in value: - collect(item) - - collect(output) - if not tensors: - raise ValueError("SVDQuant smooth evaluation produced no floating-point tensors.") - return tuple(tensors) +from auto_round.algorithms.transforms.smoothing.groups import SmoothGroup - -def filter_supported_kwargs(module: torch.nn.Module, kwargs: Mapping[str, Any]) -> dict[str, Any]: - parameters = inspect.signature(module.forward).parameters - if any(parameter.kind is inspect.Parameter.VAR_KEYWORD for parameter in parameters.values()): - return dict(kwargs) - return {name: value for name, value in kwargs.items() if name in parameters} +TargetPredicate = Callable[[str, torch.nn.Module], bool] @dataclass(frozen=True) -class SmoothSearchGroup: - """Projection group that shares an input scale and low-rank down factor.""" - - key: str - projection_names: tuple[str, ...] - projections: tuple[torch.nn.Linear, ...] - projection_input_key: str - projection_input_module: torch.nn.Module - evaluation_input_key: str - evaluation_module: torch.nn.Module - output_indices: tuple[int, ...] | None = None +class SmoothSearchGroup(SmoothGroup): + """SVDQuant compatibility group with a shared linear input width.""" def __post_init__(self) -> None: if not self.projections: @@ -78,12 +36,6 @@ def __post_init__(self) -> None: if len({projection.in_features for projection in self.projections}) != 1: raise ValueError(f"SVDQuant group {self.key!r} projections must share an input width.") - def filter_evaluation_kwargs(self, kwargs: Mapping[str, Any]) -> dict[str, Any]: - return filter_supported_kwargs(self.evaluation_module, kwargs) - - def normalize_output(self, output: Any) -> tuple[torch.Tensor, ...]: - return normalize_tensors(output, self.output_indices) - def module_global_name(block: torch.nn.Module, local_name: str) -> str: prefix = str(getattr(block, "global_name", block.__class__.__name__)) diff --git a/test/unit/common/algorithms/test_smoothing.py b/test/unit/common/algorithms/test_smoothing.py new file mode 100644 index 0000000000..d95a447fb1 --- /dev/null +++ b/test/unit/common/algorithms/test_smoothing.py @@ -0,0 +1,185 @@ +# Copyright (c) 2026 Intel Corporation +# +# 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. + +import pytest +import torch + +from auto_round.algorithms.transforms.smoothing.calibration import SmoothGroupCalibration, move_to_device +from auto_round.algorithms.transforms.smoothing.replay import ( + normalize_tensors, + output_squared_error, + temporary_modules, +) +from auto_round.algorithms.transforms.smoothing.search import ( + InvalidSmoothCandidateError, + NoFiniteCandidateError, + search_candidates, + select_best_layer_candidate, +) + + +def test_calibration_reservoir_keeps_paired_calls_and_detached_copies(): + capture = SmoothGroupCalibration(group="custom", limit=3) + for index in range(10): + inputs = torch.full((1, 2), float(index), requires_grad=True) + capture.begin_call(inputs) + capture.finish_call((inputs,), {"nested": [inputs]}, {"output": inputs * 2}) + with torch.no_grad(): + inputs.fill_(-1) + assert capture.seen_calls == 10 + assert len(capture.projection_inputs) == len(capture.evaluation_calls) == 3 + assert capture.pending_input is None + assert capture.pending_slot is None + for inputs, call in zip(capture.projection_inputs, capture.evaluation_calls): + assert inputs.grad_fn is None and not inputs.requires_grad + assert inputs.min() >= 0 + torch.testing.assert_close(inputs, call.args[0]) + torch.testing.assert_close(inputs, call.kwargs["nested"][0]) + torch.testing.assert_close(inputs * 2, call.output["output"]) + + +def test_replay_preserves_integer_masks_and_nested_outputs(): + values = {"hidden": [torch.ones(2)], "mask": torch.ones(2, dtype=torch.int64)} + moved = move_to_device(values, torch.device("cpu"), torch.bfloat16) + assert moved["hidden"][0].dtype == torch.bfloat16 + assert moved["mask"].dtype == torch.int64 + assert normalize_tensors(moved) == (moved["hidden"][0],) + + +def test_search_keeps_order_invalid_candidates_and_first_ties(): + invalid = [] + + def score(candidate): + if candidate == "invalid": + raise InvalidSmoothCandidateError("trial failed") + return {"nan": float("nan"), "first": 1, "later": 1}[candidate] + + selected, error = search_candidates( + ["first", "invalid", "nan", "later"], + score, + module_name="proj", + on_invalid=lambda candidate, exc: invalid.append(candidate), + ) + assert (selected, error) == ("first", 1) + assert invalid == ["invalid", "nan"] + + +@pytest.mark.parametrize("candidate", [None, object()]) +def test_search_keeps_the_winning_score_for_repeated_candidates(candidate): + errors = iter([1.0, 2.0]) + assert search_candidates([candidate, candidate], lambda _: next(errors), module_name="proj") == (candidate, 1.0) + assert select_best_layer_candidate([(candidate, 1.0)], module_name="proj") is candidate + + +@pytest.mark.parametrize("fail", [False, True]) +def test_temporary_modules_restore_originals(fail): + root = torch.nn.Sequential(torch.nn.Linear(2, 2), torch.nn.Linear(2, 2)) + originals = list(root) + trial = torch.nn.Identity() + try: + with temporary_modules(root, [("0", trial), ("1", trial)]): + assert root[0] is trial and root[1] is trial + if fail: + raise RuntimeError("evaluation failed") + except RuntimeError: + assert fail + assert list(root) == originals + + +def test_temporary_modules_restore_after_partial_install_failure(): + class FailOnce(torch.nn.Sequential): + def set_submodule(self, name, module): + if name == "1" and isinstance(module, torch.nn.Identity): + raise RuntimeError("installation failed") + return super().set_submodule(name, module) + + root = FailOnce(torch.nn.Linear(2, 2), torch.nn.Linear(2, 2)) + originals = list(root) + with ( + pytest.raises(RuntimeError, match="installation failed"), + temporary_modules(root, [("0", torch.nn.Identity()), ("1", torch.nn.Identity())]), + ): + pytest.fail("partial installation must not reach evaluation") + assert list(root) == originals + + +def test_error_reduction_matches_original_svdquant_accumulation(): + torch.manual_seed(7) + actual = (torch.randn(2, 3), torch.randn(1, 4)) + reference = tuple(torch.randn_like(tensor) for tensor in actual) + expected = torch.zeros((), dtype=torch.float64) + for value, target in zip(actual, reference): + expected += torch.sum((value.float() - target.float()).square()).double().cpu() + torch.testing.assert_close(output_squared_error(actual, reference), expected, rtol=0, atol=0) + with pytest.raises(ValueError, match="count changed"): + output_squared_error(actual, reference[:1]) + with pytest.raises(ValueError, match="shape changed"): + output_squared_error(actual[:1], (torch.zeros(1),)) + + +@pytest.mark.parametrize("errors", [[], [float("nan"), float("inf")]]) +def test_no_finite_candidate_has_distinct_algorithm_fallback(errors): + with pytest.raises(NoFiniteCandidateError): + search_candidates(errors, lambda error: error, module_name="proj") + + +@pytest.mark.parametrize("error_type", [RuntimeError, torch.OutOfMemoryError]) +def test_execution_errors_are_never_skipped(error_type): + def fail(_): + raise error_type("execution failure") + + with pytest.raises(error_type, match="execution failure"): + search_candidates([0, 1], fail, module_name="proj") + + +def test_calibration_clear_releases_pending_and_retained_calls(): + capture = SmoothGroupCalibration("group", 2) + inputs = torch.ones(1, 2) + capture.begin_call(inputs) + capture.finish_call((inputs,), {}, inputs) + capture.begin_call(inputs) + capture.clear() + assert not capture.projection_inputs and not capture.evaluation_calls + assert capture.pending_input is None and capture.pending_slot is None + assert capture.seen_calls == 0 + + +@pytest.mark.parametrize("require_finite, expected", [(True, 1.0), (False, -float("inf"))]) +def test_search_preserves_algorithm_specific_loss_validity(require_finite, expected): + candidate, error = search_candidates( + [float("nan"), float("inf"), -float("inf"), 1.0], + lambda value: value, + module_name="proj", + require_finite=require_finite, + ) + assert candidate == expected and error == expected + + +def test_search_releases_losing_candidates(): + import weakref + + references = [] + + def candidates(): + for index in range(10): + candidate = torch.tensor(float(index)) + references.append(weakref.ref(candidate)) + yield candidate + + def score(candidate): + assert sum(reference() is not None for reference in references) <= 3 + return candidate.item() + + candidate, error = search_candidates(candidates(), score, module_name="proj") + assert candidate.item() == error == 0.0 diff --git a/test/unit/common/algorithms/test_smoothing_awq_compatibility.py b/test/unit/common/algorithms/test_smoothing_awq_compatibility.py new file mode 100644 index 0000000000..1d952dc0c0 --- /dev/null +++ b/test/unit/common/algorithms/test_smoothing_awq_compatibility.py @@ -0,0 +1,112 @@ +# Copyright (c) 2026 Intel Corporation +# +# 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. + +import pytest +import torch + +from auto_round.algorithms.transforms.awq.base import AWQTransform +from auto_round.algorithms.transforms.awq.config import AWQConfig +from auto_round.algorithms.transforms.awq.mappings import ResolvedMapping + + +class Parent(torch.nn.Module): + def __init__(self): + super().__init__() + self.norm = torch.nn.LayerNorm(8) + self.q = torch.nn.Linear(8, 8, bias=False) + self.k = torch.nn.Linear(8, 8, bias=False) + + def forward(self, inputs): + hidden = self.norm(inputs) + output = self.q(hidden) + self.k(hidden) + # AWQ evaluates only the first tuple output. + return output, output.square() * 100 + + +@pytest.mark.parametrize("duo_scaling", [False, True, "both"]) +@pytest.mark.parametrize("parent_forward", [False, True]) +@torch.no_grad() +def test_shared_search_preserves_awq_candidates_and_loss(duo_scaling, parent_forward): + torch.manual_seed(41) + parent = Parent() + parent.q.global_name = "block.q" + parent.k.global_name = "block.k" + mapping = ResolvedMapping( + smooth_name="block.norm", + smooth_layer=parent.norm, + balance_names=["block.q", "block.k"], + balance_layers=[parent.q, parent.k], + parent_name="block", + parent=parent, + ) + transform = AWQTransform( + AWQConfig( + bits=4, + group_size=8, + sym=True, + data_type="int", + duo_scaling=duo_scaling, + n_grid=6, + smooth_batch_size=1, + ) + ) + calls = [((torch.randn(batch, 3, 8),), {}) for batch in (2, 1)] + if parent_forward: + transform._parent_args_cache[parent] = calls + x_mean = torch.cat([parent.norm(args[0]).reshape(-1, 8) for args, _ in calls]).abs().mean(0) + original = {layer: layer.weight.detach().clone() for layer in mapping.balance_layers} + actual = transform._grid_search_scales(mapping, x_mean) + assert actual is not None + params = {layer: transform._qdq_tool.resolve_params(layer) for layer in mapping.balance_layers} + funcs = {layer: transform._qdq_tool.resolve_quant_funcs(params[layer]) for layer in mapping.balance_layers} + w_mean = transform._compute_layer_means(mapping.balance_layers, 8) + reference = transform._run_parent_samples(parent, calls, offload_to_cpu=True) if parent_forward else None + candidates = [] + for ratio, use_duo in transform._get_grid_search_params(): + if use_duo: + scales = (x_mean.pow(ratio) / (w_mean.pow(1 - ratio) + 1e-4)).clamp(min=1e-4) + else: + scales = x_mean.pow(ratio).clamp(min=1e-4).view(-1) + scales = scales / (scales.max() * scales.min()).sqrt() + scales[torch.isinf(scales)] = 1 + scales[torch.isnan(scales)] = 1 + candidates.append(scales) + + def score(scales): + error = 0.0 + try: + for layer in mapping.balance_layers: + quant_func, opt_quant_func = funcs[layer] + quantized = transform._qdq_tool.qdq( + original[layer] * scales.view(1, -1), + params[layer], + quant_func=quant_func, + opt_quant_func=opt_quant_func, + imatrix=getattr(layer, "imatrix", None), + ) / scales.view(1, -1) + if parent_forward: + layer.weight.copy_(quantized) + else: + error += (original[layer] - quantized).pow(2).sum().item() + if parent_forward: + return transform._compute_parent_loss(parent, calls, reference) + return error + finally: + for layer, weight in original.items(): + layer.weight.copy_(weight) + + expected = min(candidates, key=score) + torch.testing.assert_close(actual, expected, rtol=0, atol=0) + for layer, weight in original.items(): + torch.testing.assert_close(layer.weight, weight, rtol=0, atol=0) diff --git a/test/unit/common/algorithms/test_smoothing_engine.py b/test/unit/common/algorithms/test_smoothing_engine.py new file mode 100644 index 0000000000..81f271f6f8 --- /dev/null +++ b/test/unit/common/algorithms/test_smoothing_engine.py @@ -0,0 +1,86 @@ +# Copyright (c) 2026 Intel Corporation +# +# 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. + +import pytest +import torch + +from auto_round.algorithms.transforms.smoothing import SmoothEngine, SmoothGroup + + +class IndependentStrategy: + """A third-party policy with no dependency on AWQ or SVDQuant.""" + + def __init__(self): + self.target = None + self.applied = False + + def prepare(self, group, calibration): + self.target = sum(calibration) / len(calibration) + + def candidates(self): + return [1.0, 2.0, 3.0] + + def score(self, candidate): + return (candidate - self.target) ** 2 + + def apply(self, candidate): + self.applied = True + return candidate + + def clear(self): + self.target = None + + +def make_group(key="custom"): + # Generic groups also support non-linear targets; algorithms own constraints. + module = torch.nn.Identity() + return SmoothGroup(key, (key,), (module,), key, module, key, module) + + +def test_independent_strategy_can_extend_engine(): + strategy = IndependentStrategy() + assert SmoothEngine().run(make_group(), [1.0, 3.0], strategy) == 2.0 + assert strategy.target is None + + +def test_engine_cleans_strategy_after_failure(monkeypatch): + strategy = IndependentStrategy() + + def fail(candidate): + raise RuntimeError("score failed") + + monkeypatch.setattr(strategy, "score", fail) + with pytest.raises(RuntimeError, match="score failed"): + SmoothEngine().run(make_group(), [2.0], strategy) + assert strategy.target is None + + +def test_search_does_not_apply_until_requested(): + engine = SmoothEngine() + group = make_group() + strategy = IndependentStrategy() + with engine.session(strategy): + result = engine.search(group, [2.0], strategy) + assert not strategy.applied + assert engine.apply(group, strategy, result) == 2.0 + assert strategy.applied + + +def test_result_cannot_be_applied_to_another_group(): + engine = SmoothEngine() + strategy = IndependentStrategy() + with engine.session(strategy): + result = engine.search(make_group("source"), [2.0], strategy) + with pytest.raises(ValueError, match="different group"): + engine.apply(make_group("other"), strategy, result) diff --git a/test/unit/common/algorithms/test_svdquant.py b/test/unit/common/algorithms/test_svdquant.py index 876bb36a54..fac1adf4a9 100644 --- a/test/unit/common/algorithms/test_svdquant.py +++ b/test/unit/common/algorithms/test_svdquant.py @@ -81,6 +81,12 @@ def _prepare_model(): @pytest.mark.parametrize("smooth", [False, True], ids=["nosmooth", "smooth"]) @pytest.mark.parametrize("terminal", ["rtn", "signround"]) def test_svdquant_pipeline_smoke(monkeypatch, smooth, terminal): + if not smooth: + from auto_round.algorithms.transforms.smoothing import SmoothEngine + + monkeypatch.setattr( + SmoothEngine, "search", lambda *args, **kwargs: pytest.fail("Disabled smoothing must bypass the engine") + ) monkeypatch.setattr(device_manager, "_device_map", "cpu") monkeypatch.setattr(device_manager, "_device_list", ["cpu"]) monkeypatch.setattr(device_manager, "_major_device", "cpu") diff --git a/test/unit/common/algorithms/test_svdquant_residual.py b/test/unit/common/algorithms/test_svdquant_residual.py index c35dbb5767..53344a61f2 100644 --- a/test/unit/common/algorithms/test_svdquant_residual.py +++ b/test/unit/common/algorithms/test_svdquant_residual.py @@ -133,3 +133,25 @@ def qdq_then_degrade(residual, _scheme): assert calls == 2 assert result.selected_iteration == 1 + + +def test_residual_ties_keep_first_without_early_stopping(monkeypatch): + calls = 0 + + def exact_qdq(residual, scheme): + nonlocal calls + calls += 1 + return residual + + monkeypatch.setattr(residual_module, "rtn_qdq_residual", exact_qdq) + result = iterate_residual_decomposition( + torch.ones(4, 32), + rank=0, + scheme=ResidualQuantScheme(data_type="mx_fp4e2m1", bits=4, group_size=32, sym=True), + iterations=3, + early_stop=True, + residual_dtype=torch.float32, + low_rank_dtype=torch.float32, + ) + assert result.selected_iteration == 1 + assert calls == 3 diff --git a/test/unit/common/algorithms/test_svdquant_smooth.py b/test/unit/common/algorithms/test_svdquant_smooth.py index 0d44ff8c65..beafcaacb7 100644 --- a/test/unit/common/algorithms/test_svdquant_smooth.py +++ b/test/unit/common/algorithms/test_svdquant_smooth.py @@ -19,7 +19,6 @@ absmax_channel_span, build_alpha_beta_candidates, build_smooth_scale, - select_best_layer_candidate, validate_smooth_scale_for_deployment, ) @@ -69,7 +68,82 @@ def test_deployment_validation_rejects_bfloat16_reciprocal_overflow(): ) -def test_candidate_selection_keeps_later_exact_tie_and_skips_nonfinite(): - candidates = [("nan", float("nan")), ("first", 1.0), ("later", 1.0), ("worse", 2.0)] - - assert select_best_layer_candidate(candidates, module_name="blocks.0.qkv") == "later" +@pytest.mark.parametrize("dtype", [torch.float32, torch.bfloat16]) +@torch.no_grad() +def test_shared_search_matches_original_svdquant_search(dtype): + from auto_round.algorithms.transforms.smoothing.calibration import SmoothGroupCalibration + from auto_round.algorithms.transforms.svdquant.apply import SVDQuantTransform + from auto_round.algorithms.transforms.svdquant.config import SVDQuantConfig + from auto_round.algorithms.transforms.svdquant.smooth_adapters.base import generic_linear_groups + from auto_round.schemes import PRESET_SCHEMES + + torch.manual_seed(17) + block = torch.nn.Sequential(torch.nn.Linear(32, 16, dtype=dtype)) + projection = block[0] + for name, value in PRESET_SCHEMES["MXFP4"].to_dict().items(): + setattr(projection, name, value) + projection.scale_dtype = torch.float32 + group = generic_linear_groups(block, lambda name, module: True)[0] + capture = SmoothGroupCalibration(group, limit=1) + inputs = torch.randn(1, 3, 32, dtype=dtype) + capture.begin_call(inputs) + capture.finish_call((inputs,), {}, projection(inputs)) + transform = SVDQuantTransform( + SVDQuantConfig( + rank=2, + smooth_enabled=True, + smooth_num_grids=4, + low_rank_dtype="fp32" if dtype == torch.float32 else "bf16", + ) + ) + local_names = {id(projection): "0"} + x_span = torch.stack([absmax_channel_span(inputs, -1) for inputs in capture.projection_inputs]).amax(dim=0) + w_span = absmax_channel_span(projection.weight.float(), 1) + scored = [] + # Keep the original candidate scoring as an independent numerical reference. + for alpha, beta in build_alpha_beta_candidates(4): + scale = build_smooth_scale(x_span, w_span, alpha, beta, eps=transform.config.smooth_eps) + scale = validate_smooth_scale_for_deployment(scale, dtype=dtype, module_name=group.key).float() + error = transform._score_group_wrappers( + capture, transform._candidate_group_wrappers(group, scale), block, local_names + ) + scored.append((scale, error)) + # Use the agreed first-tie policy; the original SVDQuant search kept the last tie. + expected = min(scored, key=lambda candidate: candidate[1])[0] + actual = transform._search_group_scale(capture, block, local_names) + torch.testing.assert_close(actual, expected, rtol=0, atol=0) + assert block[0] is projection + + +@torch.no_grad() +def test_svdquant_later_group_failure_does_not_install_earlier_wrappers(monkeypatch): + from auto_round.algorithms.transforms.svdquant.apply import SVDQuantTransform + from auto_round.algorithms.transforms.svdquant.config import SVDQuantConfig + from auto_round.schemes import PRESET_SCHEMES + + block = torch.nn.Sequential(torch.nn.Linear(32, 32), torch.nn.Linear(32, 32)) + originals = list(block) + for projection in block: + for name, value in PRESET_SCHEMES["MXFP4"].to_dict().items(): + setattr(projection, name, value) + projection.scale_dtype = torch.float32 + transform = SVDQuantTransform( + SVDQuantConfig(rank=2, smooth_enabled=True, smooth_num_grids=2, low_rank_dtype="fp32", model_adapter="identity") + ) + hooks = transform.register_fp_input_forward_hooks(block) + block(torch.randn(1, 2, 32)) + for hook in hooks: + hook.remove() + groups = [capture.group for capture in transform._smooth_calibration.values()] + decompose = transform._decompose_smoothed_group + + def fail_later_group(calibration, *args): + if calibration.group is groups[1]: + raise RuntimeError("second group failed") + return decompose(calibration, *args) + + monkeypatch.setattr(transform, "_decompose_smoothed_group", fail_later_group) + with pytest.raises(RuntimeError, match="second group failed"): + transform._pre_quantize_smoothed_block(block, groups) + assert list(block) == originals + assert not transform._smooth_calibration