vllm.model_executor.layers.fused_moe.oracle.nvfp4

Classes:

• NvFp4MoEKernelOracle –

  NVFP4 MoE backend selection for one deployment configuration.

Functions:

• map_nvfp4_backend –

  Map user's MoEBackend to NvFp4MoeBackend.

• select_nvfp4_moe_backend –

  Select the primary NvFP4 MoE backend

NvFp4MoEKernelOracle

Bases: MoEKernelOracle[NvFp4MoeBackend]

NVFP4 MoE backend selection for one deployment configuration.

Source code in vllm/model_executor/layers/fused_moe/oracle/nvfp4.py

class NvFp4MoEKernelOracle(MoEKernelOracle[NvFp4MoeBackend]):
    """NVFP4 MoE backend selection for one deployment configuration."""

    name = "NvFp4"

    def backend_to_kernel_cls(
        self, backend: NvFp4MoeBackend
    ) -> list[type[mk.FusedMoEExperts]]:
        return backend_to_kernel_cls(backend)

map_nvfp4_backend(runner_backend)

Map user's MoEBackend to NvFp4MoeBackend.

Source code in vllm/model_executor/layers/fused_moe/oracle/nvfp4.py

def map_nvfp4_backend(runner_backend: MoEBackend) -> NvFp4MoeBackend:
    """Map user's MoEBackend to NvFp4MoeBackend."""
    mapping = {
        "b12x": NvFp4MoeBackend.B12X,
        "cutlass": NvFp4MoeBackend.VLLM_CUTLASS,
        "flashinfer_trtllm": NvFp4MoeBackend.FLASHINFER_TRTLLM,
        "flashinfer_cutlass": NvFp4MoeBackend.FLASHINFER_CUTLASS,
        "flashinfer_cutedsl": NvFp4MoeBackend.FLASHINFER_CUTEDSL,
        "flashinfer_moe_ep_cutedsl": NvFp4MoeBackend.FLASHINFER_MOE_EP_CUTEDSL,
        "flashinfer_b12x": NvFp4MoeBackend.FLASHINFER_B12X,
        "marlin": NvFp4MoeBackend.MARLIN,
        "humming": NvFp4MoeBackend.HUMMING,
        "emulation": NvFp4MoeBackend.EMULATION,
    }
    if backend := mapping.get(runner_backend):
        return backend
    raise ValueError(
        f"moe_backend='{runner_backend}' is not supported for NvFP4 MoE. "
        f"Expected one of {list(mapping.keys())}."
    )

select_nvfp4_moe_backend(config, weight_key, activation_key)

Select the primary NvFP4 MoE backend Note: Shape-specific fallbacks may still occur at runtime.

Source code in vllm/model_executor/layers/fused_moe/oracle/nvfp4.py

def select_nvfp4_moe_backend(
    config: FusedMoEConfig,
    weight_key: QuantKey | None,
    activation_key: QuantKey | None,
) -> tuple[NvFp4MoeBackend, type[mk.FusedMoEExperts]]:
    """Select the primary NvFP4 MoE backend
    Note: Shape-specific fallbacks may still occur at runtime.
    """
    from vllm.model_executor.layers.quantization.online.requantization import (
        is_requantization_source,
    )

    # TODO: remove and let the oracles have an option to skip init.
    if is_requantization_source():
        from vllm.model_executor.layers.fused_moe.experts.nvfp4_emulation_moe import (
            Nvfp4QuantizationEmulationTritonExperts,
        )

        return NvFp4MoeBackend.EMULATION, Nvfp4QuantizationEmulationTritonExperts

    # NOTE: the kernels are selected in the following order.
    # FLASHINFER_B12X is intentionally excluded from auto-selection until
    # the upstream CUTLASS SM121 MMA op guard is resolved; use
    # moe_backend="flashinfer_b12x" to opt in explicitly.
    AVAILABLE_BACKENDS = [
        NvFp4MoeBackend.FLASHINFER_TRTLLM,
        NvFp4MoeBackend.FLASHINFER_CUTEDSL,
        NvFp4MoeBackend.FLASHINFER_CUTEDSL_BATCHED,
        NvFp4MoeBackend.FLASHINFER_CUTLASS,
        NvFp4MoeBackend.VLLM_CUTLASS,
        NvFp4MoeBackend.MARLIN,
        NvFp4MoeBackend.HUMMING,
        NvFp4MoeBackend.EMULATION,
    ]
    AVAILABLE_BACKENDS = prioritize_humming(AVAILABLE_BACKENDS)

    NVFP4_BACKENDS_WITH_CLAMP = {
        NvFp4MoeBackend.B12X,
        NvFp4MoeBackend.FLASHINFER_TRTLLM,
        NvFp4MoeBackend.FLASHINFER_CUTLASS,
        NvFp4MoeBackend.FLASHINFER_CUTEDSL,
        NvFp4MoeBackend.FLASHINFER_MOE_EP_CUTEDSL,
        NvFp4MoeBackend.VLLM_CUTLASS,
        NvFp4MoeBackend.MARLIN,
        NvFp4MoeBackend.EMULATION,
        NvFp4MoeBackend.HUMMING,
    }

    if config.swiglu_limit is not None:
        AVAILABLE_BACKENDS = [
            b for b in AVAILABLE_BACKENDS if b in NVFP4_BACKENDS_WITH_CLAMP
        ]

    activation_format = config.activation_format

    runner_backend = config.moe_backend
    requested_backend = (
        map_nvfp4_backend(runner_backend) if runner_backend != "auto" else None
    )
    # B12X can run with BF16 activations when the env forces a16.
    if requested_backend is not None and _use_a16(requested_backend, False):
        activation_key = None
    oracle = NvFp4MoEKernelOracle(config, weight_key, activation_key)

    # Handle explicit moe_backend from user.
    if requested_backend is not None:
        # For batched activation format, use batched variant if available.
        if (
            activation_format.is_batched
            and requested_backend == NvFp4MoeBackend.FLASHINFER_CUTEDSL
        ):
            requested_backend = NvFp4MoeBackend.FLASHINFER_CUTEDSL_BATCHED
        if (
            config.swiglu_limit is not None
            and requested_backend not in NVFP4_BACKENDS_WITH_CLAMP
        ):
            raise ValueError(
                f"Model sets swiglu_limit={config.swiglu_limit}, but the "
                f"explicitly requested moe_backend={runner_backend!r} does "
                f"not apply the SwiGLU clamp. Use 'flashinfer_trtllm', "
                f"'flashinfer_cutlass', 'flashinfer_cutedsl', "
                f"'flashinfer_moe_ep_cutedsl', 'cutlass', 'b12x', 'marlin', "
                f"or 'humming' instead."
            )
        return oracle.return_or_raise(requested_backend, AVAILABLE_BACKENDS)

    # Select kernels in order of backend.
    for backend in AVAILABLE_BACKENDS:
        kernel_cls, _ = oracle.try_backend(backend)
        if kernel_cls is not None:
            logger.info_once(oracle.make_log_backend(backend, AVAILABLE_BACKENDS))
            return backend, kernel_cls

    raise NotImplementedError(
        "No NvFp4 MoE backend supports the deployment configuration."
    )