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_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.

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_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.online.fp8 import OnlineLinearBase
    from vllm.model_executor.layers.quantization.quark.quark import QuarkConfig
    from vllm.model_executor.layers.quantization.utils.quant_utils import (
        is_layer_skipped,
        kMxfp4Static,
    )
    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

    online_quant_config = quant_config.online_quantization_config
    shared_expert_targets = None
    if online_quant_config is not None:
        from vllm.model_executor.layers.linear import LinearBase

        online_quant_config.packed_modules_mapping = quant_config.packed_modules_mapping
        shared_expert_targets = [
            online_quant_config.resolve_quant_method_cls(
                LinearBase, f"{shared_expert_prefix}.{projection_name}"
            )
            for projection_name in projection_names
        ]
        if all(target is None for target in shared_expert_targets):
            shared_expert_targets = None

    if shared_expert_targets is not None:
        from vllm.model_executor.layers.fused_moe import RoutedExperts
        from vllm.model_executor.layers.linear import UnquantizedLinearMethod
        from vllm.model_executor.layers.quantization.online.base import (
            ONLINE_SHARED_EXPERT_QUANTIZERS,
        )

        # NOTE: online shared experts quantization check is only implemented for quark
        # quant method at the moment. This can be extended here for other quant methods.
        # Requirement is:
        # - Retrieve routed weight quant key
        # - Retrieve routed activation quant key
        # and ensure the online quantization set on shared expert match them.
        if isinstance(quant_config, QuarkConfig):
            routed_weight_key, routed_activation_key, _ = (
                quant_config.get_quant_method_target(expert_prefix, RoutedExperts)
            )
        elif (
            isinstance(quant_config, (DeepseekV4FP8Config, DeepseekV41FP8Config))
            and quant_config.expert_dtype == "fp4"
            and moe_backend == "aiter_mega_moe"
        ):
            # AITER MegaMoE quantizes the activations of every slot itself.
            routed_weight_key, routed_activation_key = kMxfp4Static, None
        else:
            reason = (
                "online quantization FSE compatibility check is not implemented for "
                f"{type(quant_config).__name__}"
            )
            logger.warning_once(reason)
            return False, reason

        if routed_weight_key not in ONLINE_SHARED_EXPERT_QUANTIZERS:
            return (
                False,
                "online shared-expert quantization is not supported for routed "
                f"expert weight key {routed_weight_key}",
            )

        for shared_expert_target in shared_expert_targets:
            if shared_expert_target is None:
                return (
                    False,
                    "online quantization targets only part of the shared expert at "
                    f"{shared_expert_prefix}",
                )

            _, _, _, shared_quant_spec, shared_method_cls = shared_expert_target
            if shared_method_cls in (None, UnquantizedLinearMethod):
                return (
                    False,
                    "online quantization excludes FSE weights at "
                    f"{shared_expert_prefix}",
                )
            assert issubclass(shared_method_cls, OnlineLinearBase)

            # NOTE: We can not rely on shared_quant_spec.activation_quant_key
            # as _ONLINE_SHORTHANDS: dict[str, QuantSpec] in quantization.py
            # does not define the default activation quant key.
            # `online_quant_config.resolve_quant_method_cls` should return the
            # activation quant key as well. Only some online methods define it
            # on the class, so compare the weight key first.
            if shared_quant_spec.weight != routed_weight_key or (
                moe_backend != "aiter_mega_moe"
                and shared_method_cls.activation_quant_key != routed_activation_key
            ):
                return (
                    False,
                    "online shared-expert quantization keys do not match the "
                    "routed expert 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):
        from vllm.config import get_current_vllm_config
        from vllm.model_executor.models.utils import extract_layer_index

        if quant_config.expert_dtype != "fp4":
            return False, "DeepSeek-V4 routed experts are not MXFP4"

        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 has no quantization configuration"

        if not quant_config._is_quark_mxfp4_ocp(quantization_config):
            return False, "DeepSeek-v4 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 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 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

    # 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__}",
    )