vllm.model_executor.layers.quantization.utils.config_utils

Functions:

• find_matching_patterns –

  Return matching patterns for a layer or each shard of a fused layer.

• get_checkpoint_quant_keys –

  Return the (weight, activation) keys a layer is stored with in the

• get_layer_name_after_index –

  Return the suffix following the final numeric component of a layer name.

• get_quark_ocp_mx_group_size –

  Return the OCP MX group size for a quantized Quark linear layer.

• is_equal_or_regex_match –

  Checks whether a value is exactly equal or a regex match for target

• is_shared_expert_quant_fse_compatible –

  Check whether quantization permits fused shared-expert execution.

_get_effective_quant_keys(quant_config, prefix, layer_type)

Return the (weight, activation) keys a layer runs with: its online quantization target if any, else its checkpoint format. None denotes an unquantized weight, or an unknown/unquantized activation.

Raises:

• NotImplementedError –

  If the checkpoint format can not be determined.

Source code in vllm/model_executor/layers/quantization/utils/config_utils.py

def _get_effective_quant_keys(
    quant_config: "QuantizationConfig",
    prefix: str,
    layer_type: "type[torch.nn.Module]",
) -> tuple["QuantKey | None", "QuantKey | None"]:
    """Return the (weight, activation) keys a layer runs with: its online
    quantization target if any, else its checkpoint format. ``None`` denotes an
    unquantized weight, or an unknown/unquantized activation.

    Raises:
        NotImplementedError: If the checkpoint format can not be determined.

    """
    from vllm.model_executor.layers.linear import UnquantizedLinearMethod

    online_quant_config = quant_config.online_quantization_config
    target = (
        online_quant_config.resolve_quant_method_cls(layer_type, prefix)
        if online_quant_config is not None
        else None
    )
    if target is not None:
        _, _, _, quant_spec, method_cls = target
        if method_cls in (None, UnquantizedLinearMethod):
            return None, None
        # NOTE: The online shorthands' `QuantSpec` (`_ONLINE_SHORTHANDS` in
        # quantization.py) do not define the activation key: use the method's.
        return quant_spec.weight, getattr(method_cls, "activation_quant_key", None)
    return get_checkpoint_quant_keys(quant_config, prefix, layer_type)

find_matching_patterns(layer_name, patterns, fused_mapping=MappingProxyType({}), use_fnmatch=False)

Return matching patterns for a layer or each shard of a fused layer.

A pattern matching the fused layer directly takes precedence. Otherwise, return one set of matching patterns for every shard.

Source code in vllm/model_executor/layers/quantization/utils/config_utils.py

def find_matching_patterns(
    layer_name: str,
    patterns: Iterable[str],
    fused_mapping: Mapping[str, list[str]] = MappingProxyType({}),
    use_fnmatch: bool = False,
) -> list[set[str]]:
    """Return matching patterns for a layer or each shard of a fused layer.

    A pattern matching the fused layer directly takes precedence. Otherwise,
    return one set of matching patterns for every shard.
    """
    patterns = list(patterns)
    matches = [
        pattern
        for pattern in patterns
        if is_equal_or_regex_match(layer_name, pattern, use_fnmatch=use_fnmatch)
    ]
    if matches:
        return [set(matches)]

    proj_name = layer_name.split(".")[-1]
    if proj_name not in fused_mapping:
        return [set()]

    shard_names = [
        layer_name.replace(proj_name, shard_proj_name)
        for shard_proj_name in fused_mapping[proj_name]
    ]
    per_shard_matches = [
        {
            pattern
            for pattern in patterns
            if is_equal_or_regex_match(shard_name, pattern, use_fnmatch=use_fnmatch)
        }
        for shard_name in shard_names
    ]
    return per_shard_matches

get_checkpoint_quant_keys(quant_config, prefix, layer_type)

Return the (weight, activation) keys a layer is stored with in the checkpoint. None denotes an unquantized weight, or an unknown/unquantized activation.

Raises:

• NotImplementedError –

  If the checkpoint format can not be determined.

Source code in vllm/model_executor/layers/quantization/utils/config_utils.py

def get_checkpoint_quant_keys(
    quant_config: "QuantizationConfig",
    prefix: str,
    layer_type: "type[torch.nn.Module]",
) -> tuple["QuantKey | None", "QuantKey | None"]:
    """Return the (weight, activation) keys a layer is stored with in the
    checkpoint. ``None`` denotes an unquantized weight, or an unknown/unquantized
    activation.

    Raises:
        NotImplementedError: If the checkpoint format can not be determined.

    """
    from vllm.model_executor.layers.fused_moe import RoutedExperts
    from vllm.model_executor.layers.quantization.modelopt import ModelOptMxFp8Config
    from vllm.model_executor.layers.quantization.quark.quark import QuarkConfig
    from vllm.model_executor.layers.quantization.utils.quant_utils import kMxfp8Static

    if isinstance(quant_config, QuarkConfig):
        weight_key, activation_key, _ = quant_config.get_quant_method_target(
            prefix, layer_type
        )
        return weight_key, activation_key
    if isinstance(quant_config, ModelOptMxFp8Config):
        # Non-excluded linears are MXFP8. TODO: activation keys are left unknown.
        if quant_config.is_layer_excluded(prefix):
            return None, None
        if issubclass(layer_type, RoutedExperts):
            return quant_config.routed_expert_weight_key(), None
        return kMxfp8Static, None
    raise NotImplementedError(
        f"Cannot determine the checkpoint format of {type(quant_config).__name__}."
    )

get_layer_name_after_index(layer_name)

Return the suffix following the final numeric component of a layer name.

Source code in vllm/model_executor/layers/quantization/utils/config_utils.py

def get_layer_name_after_index(layer_name: str) -> str:
    """Return the suffix following the final numeric component of a layer name."""
    parts = layer_name.split(".")
    for index in range(len(parts) - 1, -1, -1):
        if parts[index].isdigit():
            return ".".join(parts[index + 1 :])
    return layer_name

get_quark_ocp_mx_group_size(quant_config, layer_name)

Return the OCP MX group size for a quantized Quark linear layer.

Source code in vllm/model_executor/layers/quantization/utils/config_utils.py

def get_quark_ocp_mx_group_size(
    quant_config: "QuantizationConfig | None",
    layer_name: str,
) -> int | None:
    """Return the OCP MX group size for a quantized Quark linear layer."""
    if quant_config is None:
        return None

    from vllm.model_executor.layers.quantization.quark.quark import QuarkConfig
    from vllm.model_executor.layers.quantization.quark.utils import should_ignore_layer

    if not isinstance(quant_config, QuarkConfig):
        return None

    if should_ignore_layer(
        layer_name,
        ignore=quant_config.quant_config.get("exclude") or [],
        fused_mapping=quant_config.packed_modules_mapping,
    ):
        return None

    layer_quant_config = (
        quant_config.get_layer_quant_config_from_name(layer_name)
        or quant_config.quant_config.get("global_quant_config")
        or {}
    )
    weight_quant = layer_quant_config.get("weight")
    input_quant = layer_quant_config.get("input_tensors")
    if not quant_config._is_w_ocp_mx_a_x(weight_quant, input_quant):
        return None

    assert weight_quant is not None
    group_size = weight_quant.get("group_size")
    if group_size is None:
        block_size = weight_quant.get("block_size")
        if isinstance(block_size, list | tuple) and len(block_size) >= 2:
            group_size = block_size[1]
    return int(group_size) if group_size is not None else None

is_equal_or_regex_match(value, target, check_contains=False, use_fnmatch=False)

Checks whether a value is exactly equal or a regex match for target if target starts with 're:'. If check_contains is set to True, additionally checks if the target string is contained within the value. If use_fnmatch is set, supports shell-style patterns in target.

Source code in vllm/model_executor/layers/quantization/utils/config_utils.py

def is_equal_or_regex_match(
    value: str,
    target: str,
    check_contains: bool = False,
    use_fnmatch: bool = False,
) -> bool:
    """Checks whether a value is exactly equal or a regex match for target
    if target starts with 're:'. If check_contains is set to True,
    additionally checks if the target string is contained within the value.
    If use_fnmatch is set, supports shell-style patterns in target.
    """
    if target.startswith("re:"):
        pattern = target[3:]
        if re.match(pattern, value):
            return True
    elif check_contains:
        if target.lower() in value.lower():
            return True
    elif target == value or use_fnmatch and fnmatch(value, target):
        return True
    return False

is_shared_expert_quant_fse_compatible(quant_config, expert_prefix, shared_expert_prefix, projection_names=None, moe_backend=None)

Check whether quantization permits fused shared-expert execution.

Parameters:

• quant_config

  (QuantizationConfig | None) –

  Model quantization configuration.

• expert_prefix

  (str) –

  Routed-expert module prefix.

• shared_expert_prefix

  (str) –

  Shared-expert module prefix.

• projection_names

  (list[str] | None, default: None ) –

  Shared-expert projection names.

• moe_backend

  (str | None, default: None ) –

  MoE backend that runs the fused shared expert.

Returns:

• tuple[bool, str | None] –

  A compatibility flag and, when incompatible, the reason.

Source code in vllm/model_executor/layers/quantization/utils/config_utils.py

def is_shared_expert_quant_fse_compatible(
    quant_config: "QuantizationConfig | None",
    expert_prefix: str,
    shared_expert_prefix: str,
    projection_names: list[str] | None = None,
    moe_backend: str | None = None,
) -> tuple[bool, str | None]:
    """Check whether quantization permits fused shared-expert execution.

    Args:
        quant_config: Model quantization configuration.
        expert_prefix: Routed-expert module prefix.
        shared_expert_prefix: Shared-expert module prefix.
        projection_names: Shared-expert projection names.
        moe_backend: MoE backend that runs the fused shared expert.

    Returns:
        A compatibility flag and, when incompatible, the reason.

    """
    from vllm.model_executor.layers.quantization.fp8 import Fp8Config
    from vllm.model_executor.layers.quantization.modelopt import ModelOptMxFp8Config
    from vllm.model_executor.layers.quantization.quark.quark import QuarkConfig
    from vllm.model_executor.layers.quantization.utils.quant_utils import (
        is_layer_skipped,
        kMxfp4Static,
        kMxfp8Static,
    )
    from vllm.models.deepseek_v4.quant_config import DeepseekV4FP8Config
    from vllm.models.deepseek_v41.quant_config import (
        DeepseekV4FP8Config as DeepseekV41FP8Config,
    )

    if projection_names is None:
        projection_names = ["gate_up_proj", "down_proj"]

    if quant_config is None:
        return True, None

    # Re-quantization of a pre-quantized checkpoint.
    online_quant_config = quant_config.online_quantization_config
    if online_quant_config is not None:
        from vllm.model_executor.layers.fused_moe import RoutedExperts
        from vllm.model_executor.layers.linear import LinearBase

        online_quant_config.packed_modules_mapping = quant_config.packed_modules_mapping
        shared_prefixes = [
            f"{shared_expert_prefix}.{projection_name}"
            for projection_name in projection_names
        ]
        is_online_quantized = any(
            online_quant_config.resolve_quant_method_cls(layer_type, prefix) is not None
            for layer_type, prefix in [
                (RoutedExperts, expert_prefix),
                *((LinearBase, prefix) for prefix in shared_prefixes),
            ]
        )
        if is_online_quantized:
            # Only check that the routed and shared experts run with the same
            # keys. Unsupported loading paths (e.g. the format the shared expert
            # is stored in) raise NotImplementedError at weight loading.
            try:
                if (
                    isinstance(
                        quant_config, (DeepseekV4FP8Config, DeepseekV41FP8Config)
                    )
                    and quant_config.expert_dtype == "fp4"
                    and moe_backend == "aiter_mega_moe"
                ):
                    routed_keys: tuple[QuantKey | None, QuantKey | None] = (
                        kMxfp4Static,
                        None,
                    )
                else:
                    routed_keys = _get_effective_quant_keys(
                        quant_config, expert_prefix, RoutedExperts
                    )
                shared_keys = {
                    _get_effective_quant_keys(quant_config, prefix, LinearBase)
                    for prefix in shared_prefixes
                }
            except NotImplementedError as e:
                logger.warning_once(str(e))
                return False, str(e)
            if moe_backend == "aiter_mega_moe":
                # AITER MegaMoE quantizes the activations of every slot itself.
                routed_keys = (routed_keys[0], None)
                shared_keys = {(weight_key, None) for weight_key, _ in shared_keys}
            if shared_keys != {routed_keys}:
                return (
                    False,
                    f"shared expert (weight, activation) keys {shared_keys} at "
                    f"{shared_expert_prefix} do not match the routed expert keys "
                    f"{routed_keys}",
                )
            return True, None

    if moe_backend == "aiter_mega_moe":
        return (
            False,
            "AITER MegaMoE fuses the shared experts only when they are "
            "requantized to MXFP4, e.g. with --quantization-config.targets "
            '\'{"*shared_experts*": "mxfp4"}\'',
        )

    if (
        isinstance(quant_config, (DeepseekV4FP8Config, DeepseekV41FP8Config))
        and quant_config.expert_dtype == "fp4"
    ):
        from vllm.config import get_current_vllm_config
        from vllm.model_executor.models.utils import extract_layer_index

        hf_config = get_current_vllm_config().model_config.hf_config

        # TODO: This is adapted from former `_shared_experts_are_fp4`, and
        # needs to be cleaned up this . There should not be Quark-specific
        # logic in DeepseekV4FP8Config.
        quantization_config = getattr(hf_config, "quantization_config", None)
        if quantization_config is None:
            return False, "DeepSeek-V4/V4.1 has no quantization configuration"

        if not quant_config._is_quark_mxfp4_ocp(quantization_config):
            return (
                False,
                "DeepSeek-V4/V4.1 FSE is only implemented/tested with Quark MXFP4",
            )

        layer_idx = extract_layer_index(shared_expert_prefix)
        if layer_idx >= hf_config.num_hidden_layers:
            shared_expert_prefix = (
                f"mtp.{layer_idx - hf_config.num_hidden_layers}.ffn.shared_experts"
            )
        else:
            shared_expert_prefix = f"layers.{layer_idx}.ffn.shared_experts"

        if any(
            entry.startswith(shared_expert_prefix)
            for entry in quantization_config.get("exclude") or []
            if isinstance(entry, str)
        ):
            return (
                False,
                f"DeepSeek-V4/V4.1 excludes shared experts at {shared_expert_prefix}",
            )

        shared_expert_weight_name = f"{shared_expert_prefix}.w1"
        layer_quant_config = quantization_config.get("layer_quant_config") or {}
        layer_config = layer_quant_config.get(shared_expert_weight_name)
        if layer_config is None:
            layer_config = next(
                (
                    config
                    for pattern, config in layer_quant_config.items()
                    if isinstance(pattern, str)
                    and pattern.startswith("re:")
                    and re.fullmatch(
                        pattern.removeprefix("re:"), shared_expert_weight_name
                    )
                ),
                None,
            )
        shared_weight_config = (
            layer_config or quantization_config.get("global_quant_config") or {}
        ).get("weight") or {}
        if shared_weight_config.get("dtype") == "fp4":
            return True, None
        return (
            False,
            f"DeepSeek-V4/V4.1 shared experts at {shared_expert_prefix} are not MXFP4",
        )

    if isinstance(quant_config, QuarkConfig):
        from vllm.model_executor.layers.quantization.quark.utils import (
            should_ignore_layer,
        )

        # TODO: layer_type_quant_config is not taken into account here.
        assert "exclude" in quant_config.quant_config
        assert "global_quant_config" in quant_config.quant_config

        exclude = quant_config.quant_config["exclude"]

        # should_ignore_layer raises a rightful ValueError in case different shards
        # have a different ignore policy, as unsupported.
        is_excluded = any(
            should_ignore_layer(
                f"{shared_expert_prefix}.{projection_name}",
                ignore=exclude,
                fused_mapping=quant_config.packed_modules_mapping,
            )
            for projection_name in projection_names
        )
        if is_excluded:
            return False, f"Quark excludes shared experts at {shared_expert_prefix}"

        global_quant_config = quant_config.quant_config["global_quant_config"]

        def get_projection_quant_configs(layer_name: str) -> list[object]:
            module_prefix, _, projection_name = layer_name.rpartition(".")
            packed_projection_names = quant_config.packed_modules_mapping.get(
                projection_name, [projection_name]
            )
            return [
                quant_config.get_layer_quant_config_from_name(
                    f"{module_prefix}.{packed_projection_name}"
                )
                or global_quant_config
                for packed_projection_name in packed_projection_names
            ]

        expert_quant_config = (
            quant_config.get_layer_quant_config_from_name(expert_prefix)
            or global_quant_config
        )
        shared_expert_quant_configs = [
            config
            for projection_name in projection_names
            for config in get_projection_quant_configs(
                f"{shared_expert_prefix}.{projection_name}"
            )
        ]
        if all(config == expert_quant_config for config in shared_expert_quant_configs):
            return True, None
        return (
            False,
            "Quark uses different quantization configurations for routed and "
            f"shared experts at {shared_expert_prefix}",
        )

    if isinstance(quant_config, Fp8Config):
        if quant_config.store_dtype is not None:
            return (
                False,
                f"FP8 stores routed experts as {quant_config.store_dtype}, which "
                f"is not supported for fused shared experts at "
                f"{shared_expert_prefix}",
            )

        # Serialized per-tensor checkpoints store 0-D or size-1 scales, which
        # the shared-expert weight chunker cannot slice into the appended expert
        # slots; online FP8 is simply untested. Both lack a weight block size.
        if quant_config.weight_block_size is None:
            return (
                False,
                "FP8 shared-expert FSE is only implemented for block-quantized "
                "checkpoints",
            )

        def is_ignored(layer_name: str) -> bool:
            return is_layer_skipped(
                prefix=layer_name,
                ignored_layers=quant_config.ignored_layers,
                fused_mapping=quant_config.packed_modules_mapping,
                match_mode=quant_config.ignored_layers_match_mode,
            )

        expert_ignored = is_ignored(expert_prefix)
        if any(
            is_ignored(f"{shared_expert_prefix}.{projection_name}") != expert_ignored
            for projection_name in projection_names
        ):
            return (
                False,
                "FP8 ignores routed and shared experts inconsistently at "
                f"{shared_expert_prefix}",
            )

        return True, None

    if isinstance(quant_config, ModelOptMxFp8Config):
        expert_excluded = quant_config.is_layer_excluded(expert_prefix)
        if any(
            quant_config.is_layer_excluded(f"{shared_expert_prefix}.{projection_name}")
            != expert_excluded
            for projection_name in projection_names
        ):
            return (
                False,
                "MXFP8 excludes routed and shared experts inconsistently at "
                f"{shared_expert_prefix}",
            )
        # The shared expert is stored as MXFP8, so it can take an expert slot
        # only when the routed experts are MXFP8 too (not NVFP4-routed hybrids).
        if (
            not expert_excluded
            and quant_config.routed_expert_weight_key() != kMxfp8Static
        ):
            return (
                False,
                "routed experts do not run as MXFP8 like the shared expert at "
                f"{shared_expert_prefix}",
            )

        return True, None

    # TODO: Extend FSE support detection to other quantization methods. Typically,
    # one would check that the experts and shared_experts use the same
    # quantization config. This may be refactored as part of QuantizationConfig later.

    return (
        False,
        "shared-expert FSE quantization compatibility is not implemented for "
        f"{type(quant_config).__name__}",
    )