Skip to content

vllm.model_executor.layers.fused_moe.oracle.mxfp8

Classes:

Functions:

Mxfp8MoEKernelOracle

Bases: MoEKernelOracle[Fp8MoeBackend]

MXFP8 MoE backend selection for one deployment configuration.

Source code in vllm/model_executor/layers/fused_moe/oracle/mxfp8.py
class Mxfp8MoEKernelOracle(MoEKernelOracle[Fp8MoeBackend]):
    """MXFP8 MoE backend selection for one deployment configuration."""

    name = "Mxfp8"

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

_mxfp8_backend_to_kernel_cls(backend)

Resolve the MXFP8 expert classes for a backend.

DeepGEMM resolves directly to DeepGemmExperts (not the TritonOrDeepGemmExperts wrapper, whose Triton fallback cannot handle the MXFP8 1x32 scheme); all other backends defer to the FP8 resolver.

Source code in vllm/model_executor/layers/fused_moe/oracle/mxfp8.py
def _mxfp8_backend_to_kernel_cls(
    backend: Fp8MoeBackend,
) -> list[type[mk.FusedMoEExperts]]:
    """Resolve the MXFP8 expert classes for a backend.

    DeepGEMM resolves directly to ``DeepGemmExperts`` (not the
    ``TritonOrDeepGemmExperts`` wrapper, whose Triton fallback cannot handle the
    MXFP8 1x32 scheme); all other backends defer to the FP8 resolver.
    """
    if backend == Fp8MoeBackend.DEEPGEMM:
        from vllm.model_executor.layers.fused_moe.experts.deep_gemm_moe import (
            DeepGemmExperts,
        )

        return [DeepGemmExperts]
    if backend == Fp8MoeBackend.AITER_MXFP8:
        from vllm.model_executor.layers.fused_moe.experts.aiter_mxfp8_moe import (
            AiterMxfp8Experts,
        )

        return [AiterMxfp8Experts]
    if backend == Fp8MoeBackend.TRITON_MXFP8:
        from vllm.model_executor.layers.fused_moe.experts.mxfp8_native_moe import (
            Mxfp8NativeTritonExperts,
        )

        return [Mxfp8NativeTritonExperts]
    if backend == Fp8MoeBackend.EMULATION:
        from vllm.model_executor.layers.fused_moe.experts.mxfp8_emulation_moe import (
            Mxfp8EmulationTritonExperts,
        )

        return [Mxfp8EmulationTritonExperts]
    return backend_to_kernel_cls(backend)

_select_kernel_cls(backend, config)

Select the first supported expert class for the MXFP8 config.

Source code in vllm/model_executor/layers/fused_moe/oracle/mxfp8.py
def _select_kernel_cls(
    backend: Fp8MoeBackend,
    config: FusedMoEConfig,
) -> type[mk.FusedMoEExperts]:
    """Select the first supported expert class for the MXFP8 config."""
    _, experts_cls = Mxfp8MoEKernelOracle(
        config, kMxfp8Static, kMxfp8Dynamic
    ).return_or_raise(backend)
    return experts_cls

select_mxfp8_moe_backend(config)

Select the MXFP8 MoE backend and the best expert class.

Returns:

Source code in vllm/model_executor/layers/fused_moe/oracle/mxfp8.py
def select_mxfp8_moe_backend(
    config: FusedMoEConfig,
) -> tuple[Fp8MoeBackend, type[mk.FusedMoEExperts]]:
    """Select the MXFP8 MoE backend and the best expert class.

    Returns:
        A tuple of (fp8_backend, experts_cls).

    """
    runner_backend = config.moe_backend
    if runner_backend != "auto":
        backend = _BACKEND_NAME_MAP.get(runner_backend)
        if backend is None:
            raise ValueError(
                f"moe_backend='{runner_backend}' is not supported for "
                f"MXFP8 MoE. Expected one of "
                f"{list(_BACKEND_NAME_MAP.keys())}."
            )
        return backend, _select_kernel_cls(backend, config)

    # Auto-select: pick the first supported backend.
    backends = prioritize_humming(list(_SUPPORTED_BACKENDS))
    for backend in backends:
        try:
            experts_cls = _select_kernel_cls(backend, config)
        except ValueError:
            continue
        return backend, experts_cls

    raise ValueError("No MXFP8 MoE backends available.")