vllm.model_executor.layers.quantization.online.base

Classes:

• OnlineQuantizationConfig –

  Model-level config for online quantization (quantize fp16/bf16 weights

• OnlineQuantizationSource –

  Supported online quantization configuration sources.

OnlineQuantizationConfig

Bases: QuantizationConfig

Model-level config for online quantization (quantize fp16/bf16 weights during model loading, without requiring a pre-quantized checkpoint).

Methods:

• resolve_quant_method_cls –

  Resolve quantization metadata and method class without instantiating it.

Source code in vllm/model_executor/layers/quantization/online/base.py

class OnlineQuantizationConfig(QuantizationConfig):
    """Model-level config for online quantization (quantize fp16/bf16 weights
    during model loading, without requiring a pre-quantized checkpoint)."""

    def __init__(
        self,
        args: QuantizationConfigArgs,
    ) -> None:
        super().__init__()
        if args.linear is None and args.moe is None and args.targets is None:
            raise ValueError(
                "OnlineQuantizationConfig requires at least one of "
                "quantization_config.linear, quantization_config.moe, or "
                "quantization_config.targets to be set."
            )
        self.args = args
        self.ignored_layers: list[str] = args.ignore
        self.quantized_layers: dict[str, tuple[str, str, str | None]] = {}
        self.online_quantization_time = 0.0

    @property
    def quantized_layer_summaries(self) -> list[str]:
        counts: dict[tuple[str, str, str | None, str], int] = {}
        for layer_name, (
            source,
            quant_key_str,
            target_pattern,
        ) in self.quantized_layers.items():
            key = (
                get_layer_name_after_index(layer_name),
                source,
                target_pattern,
                quant_key_str,
            )
            counts[key] = counts.get(key, 0) + 1

        summaries = []
        # Build summary entries as
        # `self_attn.o_proj: 24 (from targets: re:.*self_attn\.o_proj, mxfp4`
        for (layer_type, source, target_pattern, quant_key_str), count in sorted(
            counts.items()
        ):
            pattern_prefix = f"{target_pattern}, " if target_pattern else ""
            summaries.append(
                f"{layer_type}: {count} "
                f"(from {source}: {pattern_prefix}{quant_key_str})"
            )
        return summaries

    @classmethod
    def get_name(cls) -> QuantizationMethods:
        return "online"

    @classmethod
    def get_supported_act_dtypes(cls) -> list[torch.dtype]:
        return [torch.bfloat16, torch.half]

    @classmethod
    def get_min_capability(cls) -> int:
        # Note: as more online quant schemes will be added, this
        # value will become the minimum across all supported schemes.
        return 75

    @classmethod
    def get_config_filenames(cls) -> list[str]:
        return []

    @classmethod
    def from_config(cls, config: dict[str, Any]) -> "OnlineQuantizationConfig":
        raise NotImplementedError(
            "OnlineQuantizationConfig does not support loading from a "
            "checkpoint config. Use quantization_config or "
            "quantization='fp8_per_tensor'/'fp8_per_block' instead."
        )

    def _get_method_cls(
        self,
        spec: QuantSpec | None,
        table: dict[QuantKey, type],
        layer_cls: type[torch.nn.Module],
    ) -> type | None:
        """Resolve the online method class for a layer's quantization spec.

        Args:
            spec: Quantization specification to resolve.
            table: Mapping from weight quantization keys to method classes.
            layer_cls: Type of the layer that will use the resolved method.

        Returns:
            The matching method class, or None when ``spec`` has no weight
            quantization.

        """
        if spec is None or spec.weight is None:
            return None
        # Load-time gfx942 requant, not online conversion. Mxfp4MoEMethod owns it.
        if spec.weight == kInt4Static32:
            return None
        cls = table.get(spec.weight)
        if cls is None:
            raise ValueError(
                f"online quantization for {layer_cls.__name__} with "
                f"weight={spec.weight} is not supported; supported weight "
                f"keys: {sorted(str(k) for k in table)}"
            )
        # TODO: InputHadamardLinearMethod is format-agnostic; allow the
        # remaining linear methods and MoE once validated.
        if (
            spec.hadamard_size is not None
            and cls not in _ONLINE_HADAMARD_LINEAR_METHODS
        ):
            raise NotImplementedError(
                f"hadamard_size is not yet supported for online {cls.__name__}; "
                "supported: "
                f"{sorted(c.__name__ for c in _ONLINE_HADAMARD_LINEAR_METHODS)}"
            )
        return cls

    def resolve_quant_method_cls(
        self, layer_cls: type[torch.nn.Module], prefix: str
    ) -> tuple[OnlineQuantizationSource, str, str | None, QuantSpec, type] | None:
        """Resolve quantization metadata and method class without instantiating it.

        Args:
            layer_cls: Type of layer for which to resolve online quantization.
            prefix: Fully qualified layer name.

        Returns:
            A tuple of source, quantization key string, target pattern, spec
            with its resolved activation key, and method class. Returns None
            when online quantization does not apply to the layer. The returned
            spec's `fields_set` is not user input.

        """
        quant_spec: QuantSpec | None
        if self.args.targets is not None:
            resolved_pattern = self._resolve_targets_quant_method_metadata(
                prefix, layer_cls
            )
            if resolved_pattern is None:
                return None
            source, quant_key_str, target_pattern, quant_spec, table = resolved_pattern
        else:
            if issubclass(layer_cls, LinearBase):
                source = OnlineQuantizationSource.linear
                quant_spec = self.args.linear
                table = _ONLINE_LINEAR_METHODS
            elif issubclass(layer_cls, RoutedExperts):
                source = OnlineQuantizationSource.moe
                quant_spec = self.args.moe
                table = _ONLINE_MOE_METHODS
            else:
                return None

            if should_ignore_layer(
                prefix,
                ignore=self.ignored_layers,
                fused_mapping=self.packed_modules_mapping,
                use_fnmatch=True,
            ):
                return None
            quant_key_str = str(quant_spec)
            target_pattern = None

        quant_method_cls = self._get_method_cls(quant_spec, table, layer_cls)
        if quant_method_cls is None:
            return None
        assert quant_spec is not None
        assert issubclass(quant_method_cls, (OnlineMoEMethodBase, OnlineLinearBase))
        explicit = "activation" in quant_spec.fields_set
        quant_spec = QuantSpec(
            weight=quant_spec.weight,
            activation=quant_method_cls.resolve_activation_quant_key(
                quant_spec.activation
                if explicit
                else getattr(quant_method_cls, "activation_quant_key", None),
                explicit=explicit,
            ),
            # Carried so `get_quant_method` can wrap the method in
            # `InputHadamardLinearMethod`.
            hadamard_size=quant_spec.hadamard_size,
        )
        return source, quant_key_str, target_pattern, quant_spec, quant_method_cls

    def _resolve_targets_quant_method_metadata(
        self, prefix: str, layer_cls: type[torch.nn.Module]
    ) -> (
        tuple[OnlineQuantizationSource, str, str, QuantSpec, dict[QuantKey, type]]
        | None
    ):
        """Resolve target-pattern quantization metadata and dispatch table.

        Args:
            prefix: Fully qualified layer name.
            layer_cls: Type of layer matched against configured target patterns.

        Returns:
            A tuple of source, quantization key string, target pattern, spec,
            and dispatch table. Returns None when no pattern applies or the
            layer is ignored.

        """
        assert self.args.targets is not None
        ignored = should_ignore_layer(
            prefix,
            ignore=self.ignored_layers,
            fused_mapping=self.packed_modules_mapping,
            use_fnmatch=True,
        )
        matches = _find_matching_targets(
            prefix, self.args.targets, fused_mapping=self.packed_modules_mapping
        )
        if ignored and matches:
            raise ValueError(
                f"Layer {prefix} matches both quantization_config.ignore "
                f"and quantization_config.targets ({matches}); a layer may "
                f"not be referenced by both."
            )
        if ignored or not matches:
            return None
        if len(matches) > 1:
            raise ValueError(
                f"Layer {prefix} matches multiple quantization_config."
                f"targets patterns: {matches}. Each layer may match at most "
                f"one target."
            )
        target_pattern = matches[0]
        target = self.args.targets[target_pattern]
        if isinstance(target, str):
            shorthand = _ONLINE_SHORTHANDS[target]
            quant_key_str = target
            quant_spec = (
                shorthand.linear if issubclass(layer_cls, LinearBase) else shorthand.moe
            )
        else:
            quant_spec = target
            quant_key_str = str(quant_spec)
        if issubclass(layer_cls, LinearBase):
            table = _ONLINE_LINEAR_METHODS
        elif issubclass(layer_cls, RoutedExperts):
            table = _ONLINE_MOE_METHODS
        else:
            raise ValueError(
                f"Layer {prefix} was matched by quantization_config.targets "
                f"({target_pattern}), but online quantization is not supported for "
                f"{layer_cls.__name__}."
            )
        if quant_spec is None:
            raise ValueError(
                f"targets pattern {target_pattern} = {target} does "
                f"not define a QuantSpec for {layer_cls.__name__} layers "
                f"(matched at {prefix})."
            )
        return (
            OnlineQuantizationSource.targets,
            quant_key_str,
            target_pattern,
            quant_spec,
            table,
        )

    def get_quant_method(
        self, layer: torch.nn.Module, prefix: str
    ) -> "QuantizeMethodBase | None":
        # `targets` takes precedence over `moe` and `linear` and is exclusive.
        resolved = self.resolve_quant_method_cls(type(layer), prefix)
        if resolved is not None:
            source, quant_key_str, target_pattern, quant_spec, quant_method_cls = (
                resolved
            )
            self.quantized_layers[prefix] = (
                source.value,
                quant_key_str,
                target_pattern,
            )
            assert issubclass(quant_method_cls, (OnlineMoEMethodBase, OnlineLinearBase))
            kwargs = {}
            if quant_method_cls.accepts_activation_override:
                kwargs["activation_quant_key"] = quant_spec.activation
            if isinstance(layer, RoutedExperts):
                assert issubclass(quant_method_cls, OnlineMoEMethodBase)
                return quant_method_cls(moe=layer.moe_config, **kwargs)

            assert issubclass(quant_method_cls, OnlineLinearBase)
            if quant_spec.hadamard_size is not None:
                return InputHadamardLinearMethod(
                    quant_method_cls(**kwargs),
                    quant_spec.hadamard_size,
                    rotate_weight=True,
                )
            return quant_method_cls(**kwargs)

        if isinstance(layer, LinearBase):
            return UnquantizedLinearMethod()
        if isinstance(layer, RoutedExperts):
            return UnquantizedFusedMoEMethod(layer.moe_config)
        return None

_get_method_cls(spec, table, layer_cls)

Resolve the online method class for a layer's quantization spec.

Parameters:

• spec

  (QuantSpec | None) –

  Quantization specification to resolve.

• table

  (dict[QuantKey, type]) –

  Mapping from weight quantization keys to method classes.

• layer_cls

  (type[Module]) –

  Type of the layer that will use the resolved method.

Returns:

• type | None –

  The matching method class, or None when spec has no weight

• type | None –

  quantization.

Source code in vllm/model_executor/layers/quantization/online/base.py

def _get_method_cls(
    self,
    spec: QuantSpec | None,
    table: dict[QuantKey, type],
    layer_cls: type[torch.nn.Module],
) -> type | None:
    """Resolve the online method class for a layer's quantization spec.

    Args:
        spec: Quantization specification to resolve.
        table: Mapping from weight quantization keys to method classes.
        layer_cls: Type of the layer that will use the resolved method.

    Returns:
        The matching method class, or None when ``spec`` has no weight
        quantization.

    """
    if spec is None or spec.weight is None:
        return None
    # Load-time gfx942 requant, not online conversion. Mxfp4MoEMethod owns it.
    if spec.weight == kInt4Static32:
        return None
    cls = table.get(spec.weight)
    if cls is None:
        raise ValueError(
            f"online quantization for {layer_cls.__name__} with "
            f"weight={spec.weight} is not supported; supported weight "
            f"keys: {sorted(str(k) for k in table)}"
        )
    # TODO: InputHadamardLinearMethod is format-agnostic; allow the
    # remaining linear methods and MoE once validated.
    if (
        spec.hadamard_size is not None
        and cls not in _ONLINE_HADAMARD_LINEAR_METHODS
    ):
        raise NotImplementedError(
            f"hadamard_size is not yet supported for online {cls.__name__}; "
            "supported: "
            f"{sorted(c.__name__ for c in _ONLINE_HADAMARD_LINEAR_METHODS)}"
        )
    return cls

_resolve_targets_quant_method_metadata(prefix, layer_cls)

Resolve target-pattern quantization metadata and dispatch table.

Parameters:

• prefix

  (str) –

  Fully qualified layer name.

• layer_cls

  (type[Module]) –

  Type of layer matched against configured target patterns.

Returns:

• tuple[OnlineQuantizationSource, str, str, QuantSpec, dict[QuantKey, type]] | None –

  A tuple of source, quantization key string, target pattern, spec,

• tuple[OnlineQuantizationSource, str, str, QuantSpec, dict[QuantKey, type]] | None –

  and dispatch table. Returns None when no pattern applies or the

• tuple[OnlineQuantizationSource, str, str, QuantSpec, dict[QuantKey, type]] | None –

  layer is ignored.

Source code in vllm/model_executor/layers/quantization/online/base.py

def _resolve_targets_quant_method_metadata(
    self, prefix: str, layer_cls: type[torch.nn.Module]
) -> (
    tuple[OnlineQuantizationSource, str, str, QuantSpec, dict[QuantKey, type]]
    | None
):
    """Resolve target-pattern quantization metadata and dispatch table.

    Args:
        prefix: Fully qualified layer name.
        layer_cls: Type of layer matched against configured target patterns.

    Returns:
        A tuple of source, quantization key string, target pattern, spec,
        and dispatch table. Returns None when no pattern applies or the
        layer is ignored.

    """
    assert self.args.targets is not None
    ignored = should_ignore_layer(
        prefix,
        ignore=self.ignored_layers,
        fused_mapping=self.packed_modules_mapping,
        use_fnmatch=True,
    )
    matches = _find_matching_targets(
        prefix, self.args.targets, fused_mapping=self.packed_modules_mapping
    )
    if ignored and matches:
        raise ValueError(
            f"Layer {prefix} matches both quantization_config.ignore "
            f"and quantization_config.targets ({matches}); a layer may "
            f"not be referenced by both."
        )
    if ignored or not matches:
        return None
    if len(matches) > 1:
        raise ValueError(
            f"Layer {prefix} matches multiple quantization_config."
            f"targets patterns: {matches}. Each layer may match at most "
            f"one target."
        )
    target_pattern = matches[0]
    target = self.args.targets[target_pattern]
    if isinstance(target, str):
        shorthand = _ONLINE_SHORTHANDS[target]
        quant_key_str = target
        quant_spec = (
            shorthand.linear if issubclass(layer_cls, LinearBase) else shorthand.moe
        )
    else:
        quant_spec = target
        quant_key_str = str(quant_spec)
    if issubclass(layer_cls, LinearBase):
        table = _ONLINE_LINEAR_METHODS
    elif issubclass(layer_cls, RoutedExperts):
        table = _ONLINE_MOE_METHODS
    else:
        raise ValueError(
            f"Layer {prefix} was matched by quantization_config.targets "
            f"({target_pattern}), but online quantization is not supported for "
            f"{layer_cls.__name__}."
        )
    if quant_spec is None:
        raise ValueError(
            f"targets pattern {target_pattern} = {target} does "
            f"not define a QuantSpec for {layer_cls.__name__} layers "
            f"(matched at {prefix})."
        )
    return (
        OnlineQuantizationSource.targets,
        quant_key_str,
        target_pattern,
        quant_spec,
        table,
    )

resolve_quant_method_cls(layer_cls, prefix)

Resolve quantization metadata and method class without instantiating it.

Parameters:

• layer_cls

  (type[Module]) –

  Type of layer for which to resolve online quantization.

• prefix

  (str) –

  Fully qualified layer name.

Returns:

• tuple[OnlineQuantizationSource, str, str | None, QuantSpec, type] | None –

  A tuple of source, quantization key string, target pattern, spec

• tuple[OnlineQuantizationSource, str, str | None, QuantSpec, type] | None –

  with its resolved activation key, and method class. Returns None

• tuple[OnlineQuantizationSource, str, str | None, QuantSpec, type] | None –

  when online quantization does not apply to the layer. The returned

• tuple[OnlineQuantizationSource, str, str | None, QuantSpec, type] | None –

  spec's fields_set is not user input.

Source code in vllm/model_executor/layers/quantization/online/base.py

def resolve_quant_method_cls(
    self, layer_cls: type[torch.nn.Module], prefix: str
) -> tuple[OnlineQuantizationSource, str, str | None, QuantSpec, type] | None:
    """Resolve quantization metadata and method class without instantiating it.

    Args:
        layer_cls: Type of layer for which to resolve online quantization.
        prefix: Fully qualified layer name.

    Returns:
        A tuple of source, quantization key string, target pattern, spec
        with its resolved activation key, and method class. Returns None
        when online quantization does not apply to the layer. The returned
        spec's `fields_set` is not user input.

    """
    quant_spec: QuantSpec | None
    if self.args.targets is not None:
        resolved_pattern = self._resolve_targets_quant_method_metadata(
            prefix, layer_cls
        )
        if resolved_pattern is None:
            return None
        source, quant_key_str, target_pattern, quant_spec, table = resolved_pattern
    else:
        if issubclass(layer_cls, LinearBase):
            source = OnlineQuantizationSource.linear
            quant_spec = self.args.linear
            table = _ONLINE_LINEAR_METHODS
        elif issubclass(layer_cls, RoutedExperts):
            source = OnlineQuantizationSource.moe
            quant_spec = self.args.moe
            table = _ONLINE_MOE_METHODS
        else:
            return None

        if should_ignore_layer(
            prefix,
            ignore=self.ignored_layers,
            fused_mapping=self.packed_modules_mapping,
            use_fnmatch=True,
        ):
            return None
        quant_key_str = str(quant_spec)
        target_pattern = None

    quant_method_cls = self._get_method_cls(quant_spec, table, layer_cls)
    if quant_method_cls is None:
        return None
    assert quant_spec is not None
    assert issubclass(quant_method_cls, (OnlineMoEMethodBase, OnlineLinearBase))
    explicit = "activation" in quant_spec.fields_set
    quant_spec = QuantSpec(
        weight=quant_spec.weight,
        activation=quant_method_cls.resolve_activation_quant_key(
            quant_spec.activation
            if explicit
            else getattr(quant_method_cls, "activation_quant_key", None),
            explicit=explicit,
        ),
        # Carried so `get_quant_method` can wrap the method in
        # `InputHadamardLinearMethod`.
        hadamard_size=quant_spec.hadamard_size,
    )
    return source, quant_key_str, target_pattern, quant_spec, quant_method_cls

OnlineQuantizationSource

Bases: str, Enum

Supported online quantization configuration sources.

Source code in vllm/model_executor/layers/quantization/online/base.py

class OnlineQuantizationSource(str, Enum):
    """Supported online quantization configuration sources."""

    linear = "linear"  # LinearBase
    moe = "moe"  # RoutedExperts
    targets = "targets"

_resolved_target(target)

Comparable identity of a quantization_config.targets value.

Returns one (weight, activation, activation was explicit, hadamard_size) tuple per layer kind (linear, MoE), or None for a kind the target does not define: a shorthand name and the equivalent QuantSpec therefore compare equal, while an omitted activation still differs from an explicit null.

Source code in vllm/model_executor/layers/quantization/online/base.py

def _resolved_target(target: str | QuantSpec) -> tuple:
    """Comparable identity of a `quantization_config.targets` value.

    Returns one `(weight, activation, activation was explicit, hadamard_size)`
    tuple per layer kind (linear, MoE), or `None` for a kind the target does not
    define: a shorthand name and the equivalent `QuantSpec` therefore compare
    equal, while an omitted activation still differs from an explicit `null`.
    """
    if isinstance(target, str):
        shorthand = _ONLINE_SHORTHANDS[target]
        # A one-sided shorthand applies its spec to either layer kind, like the
        # equivalent `QuantSpec` does, so compare it that way.
        specs = (
            shorthand.linear or shorthand.moe,
            shorthand.moe or shorthand.linear,
        )
    else:
        specs = (target, target)
    return tuple(
        None
        if spec is None
        else (
            spec.weight,
            spec.activation,
            "activation" in spec.fields_set,
            spec.hadamard_size,
        )
        for spec in specs
    )