Repository: vllm-project/vllm
File: vllm/model_executor/layers/quantization/quark/quark_moe.py (Python)

Code before:
        layer_quant_config = quant_config._find_matched_config(layer_name, module)

        if layer_quant_config.get("output_tensors") or layer_quant_config.get("bias"):
            raise NotImplementedError(
                "Currently, Quark models with "
                "output_tensors and bias "
                "quantized are not supported"
            )

        emulate = not current_platform.supports_mx() or not (
            rocm_aiter_ops.is_fused_moe_enabled()
        )
        weight_config = layer_quant_config.get("weight")
        input_config = layer_quant_config.get("input_tensors")

        if quant_config._is_fp8_w4a8(weight_config, input_config):
            return QuarkW4A8Fp8MoEMethod(weight_config, input_config, module.moe_config)
        elif quant_config._is_fp8_w8a8(weight_config, input_config):
            return QuarkW8A8Fp8MoEMethod(weight_config, input_config, module.moe_config)
        elif quant_config._is_w_ocp_mx_a_x(weight_config, input_config):
            from vllm.config import get_current_vllm_config

            model_type = getattr(
                get_current_vllm_config().model_config.hf_config, "model_type", None
            )
            if model_type == "gpt_oss" and not emulate:
                return QuarkOCP_MX_MoEMethod_OSS(
                    weight_config, input_config, module.moe_config
                )
            else:
                return QuarkOCP_MX_MoEMethod(
                    weight_config, input_config, module.moe_config
                )
        else:
            raise RuntimeError("Unsupported FusedMoe scheme")


class QuarkW8A8Fp8MoEMethod(QuarkMoEMethod):
    def __init__(
        self,
        weight_config: dict[str, Any],
        input_config: dict[str, Any],
        moe: FusedMoEConfig,
    ):
        super().__init__(moe)
        self.weight_quant = weight_config
        self.input_quant = input_config

        self.weight_qscheme = self.weight_quant.get("qscheme")
        self.input_qscheme = self.input_quant.get("qscheme")
        self.weight_dtype = self.weight_quant.get("dtype", "").replace(

Review comment (suggestion):
best to avoid conditioning on model type, ideally something like `if self.ocp_mx_scheme == "w_mxfp4_a_fp8" and not emulate:`

Code after:

        if layer_quant_config.get("output_tensors") or layer_quant_config.get("bias"):
            raise NotImplementedError(
                "Currently, Quark models with "
                "output_tensors and bias "
                "quantized are not supported"
            )

        weight_config = layer_quant_config.get("weight")
        input_config = layer_quant_config.get("input_tensors")

        if quant_config._is_fp8_w4a8(weight_config, input_config):
            return QuarkW4A8Fp8MoEMethod(weight_config, input_config, module.moe_config)
        elif quant_config._is_fp8_w8a8(weight_config, input_config):
            return QuarkW8A8Fp8MoEMethod(weight_config, input_config, module.moe_config)
        elif quant_config._is_w_ocp_mx_a_x(weight_config, input_config):
            emulate = not current_platform.supports_mx() or not (
                rocm_aiter_ops.is_fused_moe_enabled()
            )
            if (
                input_config.get("dtype") == "fp8_e4m3"
                and not input_config.get("is_dynamic")
                and not emulate
            ):
                return QuarkOCP_MX_MoEMethod_OSS(
                    weight_config, input_config, module.moe_config
                )
            else:
                return QuarkOCP_MX_MoEMethod(
                    weight_config, input_config, module.moe_config
                )
        else:
            raise RuntimeError("Unsupported FusedMoe scheme")


class QuarkW8A8Fp8MoEMethod(QuarkMoEMethod):
    def __init__(
        self,
        weight_config: dict[str, Any],
        input_config: dict[str, Any],
        moe: FusedMoEConfig,
    ):
        super().__init__(moe)
        self.weight_quant = weight_config
        self.input_quant = input_config

        self.weight_qscheme = self.weight_quant.get("qscheme")
        self.input_qscheme = self.input_quant.get("qscheme")
        self.weight_dtype = self.weight_quant.get("dtype", "").replace(
            "fp8_e4m3", "fp8"
        )