vllm.model_executor.layers.fused_moe.experts.fused_humming_moe ¶

class BatchedHummingGroupedExperts(HummingExpertsBase):
    def finalize_weight_and_reduce_impl(self) -> mk.TopKWeightAndReduce:
        return TopKWeightAndReduceDelegate()

    @staticmethod
    def activation_format() -> mk.FusedMoEActivationFormat:
        return mk.FusedMoEActivationFormat.BatchedExperts

    @staticmethod
    def humming_gemm_type() -> "HummingGemmType":
        from vllm.utils.humming import GemmType as HummingGemmType

        return HummingGemmType.GROUPED_MASKED

    def apply(
        self,
        output: torch.Tensor,
        hidden_states: torch.Tensor,
        w1: torch.Tensor,
        w2: torch.Tensor,
        topk_weights: torch.Tensor,
        topk_ids: torch.Tensor,
        activation: MoEActivation,
        global_num_experts: int,
        expert_map: torch.Tensor | None,
        a1q_scale: torch.Tensor | None,
        a2_scale: torch.Tensor | None,
        workspace13: torch.Tensor,
        workspace2: torch.Tensor,
        expert_tokens_meta: mk.ExpertTokensMetadata | None,
        apply_router_weight_on_input: bool,
    ) -> None:
        """
        Standard apply implementation for Humming batched grouped experts.

        Note: Humming kernels handle weights internally through the layer
        object, so w1, w2, a2_scale are unused. a1q_scale is None on the usual
        path (Humming quantizes the w13 input itself); for block-FP8 activations
        it carries the scale computed before dispatch, which is forwarded to
        may_quant_input so Humming skips the redundant w13 quantization. The
        output is written into workspace13 via the buffer management.
        """
        assert not apply_router_weight_on_input
        assert expert_tokens_meta is not None

        hidden_states = hidden_states.view(-1, hidden_states.size(-1))
        # Keep the (batched) block-FP8 scale row-aligned with the flattened
        # [num_experts * max_tokens, K] hidden states above.
        if a1q_scale is not None and a1q_scale.dim() == 3:
            a1q_scale = a1q_scale.view(-1, a1q_scale.size(-1))
        valid_shape_m = self.estimate_local_valid_shape_m(topk_ids)
        expert_num_tokens = expert_tokens_meta.expert_num_tokens

        buffers = self.prepare_buffers(
            workspace13,
            workspace2,
            topk_ids.size(0),
            topk_ids.size(1),
            activation,
        )
        buffers["down_output"] = output

        inputs, input_scale = self.quantize_input(
            "w13",
            inputs=hidden_states,
            input_scale=a1q_scale,
            quanted_input=buffers.get("quanted_gate_up_input", None),
        )

        self.humming_forward(
            "w13",
            inputs=inputs,
            weight=w1,
            input_scale=input_scale,
            outputs=buffers["gate_up_output"],
            valid_shape_m=valid_shape_m,
            expert_layout=expert_num_tokens,
            compute_config=self.compute_config_str,
            tuning_config=self.w13_tuning_config_str,
        )

        self.apply_activation(
            activation=activation,
            input=buffers["gate_up_output"],
            output=buffers["activation_output"],
        )

        inputs, input_scale = self.quantize_input(
            "w2",
            inputs=buffers["activation_output"],
            quanted_input=buffers.get("quanted_down_input", None),
        )

        self.humming_forward(
            "w2",
            inputs=inputs,
            weight=w2,
            input_scale=input_scale,
            outputs=output.view(-1, hidden_states.size(-1)),
            valid_shape_m=valid_shape_m,
            expert_layout=expert_num_tokens,
            compute_config=self.compute_config_str,
            tuning_config=self.w2_tuning_config_str,
        )

class HummingExpertsBase(mk.FusedMoEExpertsModular):
    def __init__(
        self,
        moe_config: FusedMoEConfig,
        quant_config: FusedMoEQuantConfig,
        max_num_tokens: int | None = None,
        num_dispatchers: int | None = None,
    ):
        humming_quant_config = cast("HummingMoEQuantConfig", quant_config)
        self.humming_configs: dict[str, HummingLayerConfig] = {
            "w13": humming_quant_config.w1_humming_config,
            "w2": humming_quant_config.w2_humming_config,
        }
        self.locks = torch.zeros(1024, dtype=torch.int32, device=moe_config.device)
        self.num_experts = moe_config.num_local_experts
        self.global_num_experts = moe_config.num_experts
        self.quant_config = quant_config
        self.init_humming_moe()

        if self.is_batched():
            assert max_num_tokens is not None and num_dispatchers is not None

        super().__init__(
            moe_config=moe_config,
            quant_config=quant_config,
            max_num_tokens=max_num_tokens,
            num_dispatchers=num_dispatchers,
        )
        self._permute_scratch: dict[int, MoEPermuteScratch] = {}

    def init_humming_moe(self):
        from vllm.utils.humming import get_heuristics_config

        self.compute_config = {
            "use_batch_invariant": envs.VLLM_BATCH_INVARIANT,
            "use_f16_accum": envs.VLLM_HUMMING_USE_F16_ACCUM,
            "gemm_type": self.humming_gemm_type().value,
        }
        self.w13_tuning_config = get_heuristics_config(
            layer_config=self.humming_configs["w13"],
            use_f16_accum=envs.VLLM_HUMMING_USE_F16_ACCUM,
            use_batch_invariant=envs.VLLM_BATCH_INVARIANT,
            gemm_type=self.humming_gemm_type(),
        )
        self.w2_tuning_config = get_heuristics_config(
            layer_config=self.humming_configs["w2"],
            use_f16_accum=envs.VLLM_HUMMING_USE_F16_ACCUM,
            use_batch_invariant=envs.VLLM_BATCH_INVARIANT,
            gemm_type=self.humming_gemm_type(),
        )
        self.compute_config_str = json.dumps(self.compute_config)
        # Fix up the heuristic-chosen tiles before freezing to JSON -- see
        # _fixup_moe_tuning_config.
        _fixup_moe_tuning_config(self.w13_tuning_config)
        _fixup_moe_tuning_config(self.w2_tuning_config)
        self.w13_tuning_config_str = json.dumps(self.w13_tuning_config)
        self.w2_tuning_config_str = json.dumps(self.w2_tuning_config)

    def quantize_input(
        self,
        sublayer_name: str,
        inputs: torch.Tensor,
        quanted_input: torch.Tensor | None,
        input_scale: torch.Tensor | None = None,
    ) -> tuple[torch.Tensor, torch.Tensor | None]:
        from vllm.utils.humming import may_quant_input

        # input_scale is set for block-FP8 (group-128) activations that were
        # quantized before the EP dispatch: may_quant_input then skips the
        # redundant w13 quantization and forwards the pre-computed scale.
        return may_quant_input(
            self.humming_configs[sublayer_name],
            inputs=inputs,
            input_scale=input_scale,
            quanted_input=quanted_input,
        )

    def humming_forward(
        self,
        sublayer_name: str,
        inputs: torch.Tensor,
        weight: torch.Tensor,
        input_scale: torch.Tensor | None,
        outputs: torch.Tensor,
        **kwargs: Any,
    ) -> torch.Tensor:
        from vllm.utils.humming import humming_forward

        is_w13 = sublayer_name == "w13"
        return humming_forward(
            self.humming_configs[sublayer_name],
            inputs=inputs,
            weight=weight,
            weight_scale=(
                self.quant_config.w1_scale if is_w13 else self.quant_config.w2_scale
            ),
            zero_point=(self.quant_config.w1_zp if is_w13 else self.quant_config.w2_zp),
            bias=(self.quant_config.w1_bias if is_w13 else self.quant_config.w2_bias),
            weight_scale_2=(
                self.quant_config.g1_alphas if is_w13 else self.quant_config.g2_alphas
            ),
            input_scale=input_scale,
            outputs=outputs,
            locks=self.locks,
            **kwargs,
        )

    def _get_permute_scratch(self, topk: int) -> MoEPermuteScratch | None:
        if not moe_permute_unpermute_supported():
            return None

        scratch = self._permute_scratch.get(topk)
        if scratch is None:
            max_expanded_rows = (
                self.moe_config.max_num_tokens
                * self.moe_config.dp_size
                * self.moe_config.experts_per_token
            )
            scratch = MoEPermuteScratch(
                max_num_tokens=math.ceil(max_expanded_rows / topk),
                topk=topk,
                num_experts=self.moe_config.num_experts,
                num_local_experts=self.moe_config.num_local_experts,
                device=torch.device(self.moe_config.device),
                hidden_size=self.moe_config.hidden_dim,
                hidden_dtype=self.moe_config.in_dtype,
            )
            self._permute_scratch[topk] = scratch
        return scratch

    def get_global_valid_shape_m(self, topk_ids: torch.Tensor):
        ctx = get_forward_context()
        if ctx.dp_metadata is not None:
            num_tokens = ctx.dp_metadata.num_tokens_across_dp_cpu.sum().item()
            return num_tokens * self.moe_config.experts_per_token

        return topk_ids.size(0) * topk_ids.size(1)

    def estimate_local_valid_shape_m(self, topk_ids: torch.Tensor):
        # estimate shape_m for kernel tuning
        global_valid_shape_m = self.get_global_valid_shape_m(topk_ids)
        num_experts = self.num_experts
        global_num_experts = self.global_num_experts
        return math.ceil(global_valid_shape_m * num_experts / global_num_experts)

    @staticmethod
    def humming_gemm_type() -> "HummingGemmType":
        raise NotImplementedError

    @classmethod
    def is_batched(cls) -> bool:
        return cls.activation_format() == mk.FusedMoEActivationFormat.BatchedExperts

    @staticmethod
    def _supports_quant_scheme(
        weight_key: QuantKey | None,
        activation_key: QuantKey | None,
    ) -> bool:
        SUPPORTED_W_A = [
            (kMxfp4Static, None),
            (kMxfp4Static, kMxfp4Dynamic),
            (kMxfp4Static, kMxfp8Dynamic),
            (kMxfp4Static, kFp8DynamicTokenSym),
            # MXFP4 weight (group-32 e8m0) with block-FP8 activation
            # (group-128 float32). Runs via WGMMA software dequant, so it
            # works on Hopper (SM90/H200) as well as Blackwell.
            (kMxfp4Static, kFp8Dynamic128Sym),
            (kNvfp4Static, None),
            (kNvfp4Static, kFp8DynamicTokenSym),
            (kMxfp8Static, None),
            (kMxfp8Static, kFp8DynamicTokenSym),
            (kFp8StaticChannelSym, None),
            (kFp8StaticChannelSym, kFp8DynamicTokenSym),
            (kFp8Static128BlockSym, None),
            (kFp8Static128BlockSym, kFp8DynamicTokenSym),
            (kInt4Static, None),
            (kInt4Static, kFp8DynamicTokenSym),
            (kInt8Static, None),
            (kInt8Static, kFp8DynamicTokenSym),
            # Checkpoint-driven (weight, activation) pairs the dense/MoE oracles
            # pass. Humming defers input quant (see expects_unquantized_inputs),
            # so the activation key does not constrain support.
            # fp8 (compressed-tensors / native / modelopt)
            (kFp8StaticChannelSym, kFp8StaticTensorSym),
            (kFp8StaticChannelSym, kFp8Dynamic128Sym),
            (kFp8StaticTensorSym, None),
            (kFp8StaticTensorSym, kFp8DynamicTokenSym),
            (kFp8StaticTensorSym, kFp8StaticTensorSym),
            (kFp8StaticTensorSym, kFp8Dynamic128Sym),
            (kFp8Static128BlockSym, kFp8Dynamic128Sym),
            # int8 (compressed-tensors w8a8 / experts_int8)
            (kInt8StaticChannelSym, None),
            (kInt8StaticChannelSym, kInt8DynamicTokenSym),
            # nvfp4 (compressed-tensors / modelopt / quark)
            (kNvfp4Static, kNvfp4Dynamic),
            # mxfp8 (compressed-tensors / modelopt / online)
            (kMxfp8Static, kMxfp8Dynamic),
        ]
        return (weight_key, activation_key) in SUPPORTED_W_A or (
            activation_key in (None, kFp8DynamicTokenSym)
            and _is_supported_wna16_weight_key(weight_key)
        )

    def _prequantizes_dispatch_activation(self) -> bool:
        """
        Whether the prepare/finalize step should quantize activations before
        the (EP all-to-all) dispatch instead of leaving it to Humming.

        This is enabled only for block-FP8 (group-128) activations: quantizing
        to FP8 before dispatch sends FP8 rather than BF16 over the interconnect,
        and Humming then consumes the pre-quantized FP8 + scale directly (see
        the apply() methods, which forward the dispatch scale into
        HummingMethod.may_quant_input, and may_quant_input itself, which is a
        no-op when an input scale is already supplied). The scale layout
        produced by vLLM's block-FP8 quantization ([M, K // 128] float32,
        row-major) matches what the Humming WGMMA grouped GEMM expects.
        """
        quant_config = self.quant_config
        return (
            quant_config.is_block_quantized
            and quant_config.quant_dtype == current_platform.fp8_dtype()
        )

    @property
    def expects_unquantized_inputs(self) -> bool:
        """
        Whether the prepare/finalize step should defer input quantization to
        the experts (by setting defer_input_quant=True and passing unquantized
        inputs).

        Humming normally quantizes inputs internally via
        HummingMethod.may_quant_input() in apply(), so we defer quantization
        (return True) to avoid quantizing twice -- once in prepare and once in
        Humming's apply().

        The exception is block-FP8 (group-128) activations, which are quantized
        before the dispatch to save interconnect bandwidth (see
        _prequantizes_dispatch_activation): for those we must NOT defer.
        """
        return not self._prequantizes_dispatch_activation()

    @staticmethod
    def _supports_current_device() -> bool:
        platform = current_platform
        return (
            has_humming()
            and platform.is_cuda()
            and platform.has_device_capability((7, 5))
        )

    @staticmethod
    def _supports_no_act_and_mul() -> bool:
        return True

    @staticmethod
    def _supports_activation(activation: MoEActivation) -> bool:
        # Humming uses apply_moe_activation() callback for activation,
        # so any activation supported there can be used here.
        return apply_moe_activation_supported(activation)

    @staticmethod
    def _supports_parallel_config(moe_parallel_config: FusedMoEParallelConfig) -> bool:
        return True

    @staticmethod
    def _supports_batch_invariance() -> bool:
        return True

    def moe_problem_size(
        self,
        a1: torch.Tensor,
        w1: torch.Tensor,
        w2: torch.Tensor,
        topk_ids: torch.Tensor,
    ) -> tuple[int, int, int, int, int]:
        meta1 = self.humming_configs["w13"]
        meta2 = self.humming_configs["w2"]

        assert meta1.num_experts == meta2.num_experts

        num_experts = meta1.num_experts
        top_k = topk_ids.size(1)
        assert w1.size(0) == num_experts
        assert w2.size(0) == num_experts

        if not self.is_batched():
            num_tokens = a1.size(0)
            assert topk_ids.size(0) == num_tokens
        else:
            assert a1.dim() == 3
            assert a1.size(0) == num_experts
            num_tokens = a1.size(1)

        intermediate_dim = meta2.shape_k - meta2.pad_shape_k
        hidden_dim = meta1.shape_k - meta1.pad_shape_k
        return meta1.num_experts, num_tokens, intermediate_dim, hidden_dim, top_k

    def get_buffer_metas(self, M: int, topk: int, activation: MoEActivation):
        from vllm.utils.humming import GemmType as HummingGemmType
        from vllm.utils.humming import dtypes

        num_experts = self.num_experts
        w13_config = self.humming_configs["w13"]
        w2_config = self.humming_configs["w2"]
        gate_up_dim = w13_config.shape_n - w13_config.pad_shape_n
        intermediate_dim = w2_config.shape_k - w2_config.pad_shape_k
        K = w13_config.shape_k - w13_config.pad_shape_k
        assert isinstance(num_experts, int)
        assert isinstance(gate_up_dim, int)
        assert isinstance(intermediate_dim, int)
        assert isinstance(K, int)
        assert intermediate_dim == self.adjust_N_for_activation(gate_up_dim, activation)

        # hidden_states
        # (-> quanted_gate_up_input) (if not BF16/FP16 activation)
        # -> gate_up_output
        # -> activation_output
        # (-> quanted_down_input) (if not BF16/FP16 activation)
        # -> down_output
        # (-> output) (if not is_batched)
        # Neighboring nodes are required to utilize distinct workspaces.
        # The final output buffer is supplied by the modular kernel and may
        # alias workspace1.

        output_shape: tuple[int, ...]
        if self.is_batched():
            max_num_tokens = self.max_num_tokens
            num_dispatchers = self.num_dispatchers
            assert max_num_tokens is not None and num_dispatchers is not None
            input_shape_m = num_experts * max_num_tokens
            real_shape_m = num_experts * max_num_tokens * num_dispatchers
            output_shape = (num_experts, max_num_tokens * num_dispatchers, K)
        else:
            input_shape_m = M
            if self.humming_gemm_type() != HummingGemmType.INDEXED:
                input_shape_m = M * topk
            real_shape_m = M * topk
            output_shape = (M, K)

        a_dtype = self.humming_configs["w13"].a_dtype
        c_dtype = self.humming_configs["w13"].c_dtype
        num_bits = a_dtype.num_bits
        torch_dtype_map = {
            dtypes.float16: torch.float16,
            dtypes.bfloat16: torch.bfloat16,
            dtypes.float32: torch.float32,
            dtypes.float8e4m3: torch.float8_e4m3fn,
            dtypes.float8e5m2: torch.float8_e5m2,
            dtypes.int8: torch.int8,
            dtypes.int4: torch.uint8,
        }

        buffer_metas = {
            "quanted_gate_up_input": {
                "shape": (input_shape_m, K),
                "dtype": torch_dtype_map[a_dtype],
            },
            "gate_up_output": {
                "shape": (real_shape_m, gate_up_dim),
                "dtype": torch_dtype_map[c_dtype],
            },
            "activation_output": {
                "shape": (real_shape_m, intermediate_dim),
                "dtype": torch_dtype_map[c_dtype],
            },
            "quanted_down_input": {
                "shape": (real_shape_m, intermediate_dim),
                "dtype": torch_dtype_map[a_dtype],
            },
            "down_output": {
                "shape": output_shape if self.is_batched() else (real_shape_m, K),
                "dtype": torch_dtype_map[c_dtype],
            },
            "output": {
                "shape": output_shape,
                "dtype": torch_dtype_map[c_dtype],
            },
        }

        for key in buffer_metas:
            meta = buffer_metas[key]
            if "quanted" in key and a_dtype.num_bits == 4:
                last_dim = meta["shape"][-1]
                if last_dim % 2 != 0:
                    raise ValueError(
                        f"Int4 packing requires last dimension to be even, "
                        f"got {last_dim} for buffer '{key}'"
                    )
                meta["shape"] = meta["shape"][:-1] + (last_dim // 2,)

        if num_bits == 16:
            required_buffers = ["gate_up_output", "activation_output", "down_output"]
        else:
            required_buffers = [
                "quanted_gate_up_input",
                "gate_up_output",
                "activation_output",
                "quanted_down_input",
                "down_output",
            ]

            # Note: The fused SITU+FP8 quant goes gate_up_output ->
            #        quanted_down_input directly, never materializing
            #        activation_output. Dropping activation_output from the chain
            #        flips the even/odd workspace 2-coloring below so
            #        gate_up_output and quanted_down_input land
            #        on DIFFERENT workspaces.
            if self.fused_situ_quant_enabled(activation):
                required_buffers.remove("activation_output")

        # batched moe use down_output as output
        if not self.is_batched():
            required_buffers.append("output")

        return buffer_metas, required_buffers

    def _workspace_shapes(self, M: int, topk: int, activation: MoEActivation):
        buffer_metas, required_buffers = self.get_buffer_metas(M, topk, activation)

        workspace1_nbytes = 0
        workspace2_nbytes = 0

        for index, name in enumerate(required_buffers[::-1]):
            buffer_meta = buffer_metas[name]
            nelement = math.prod(buffer_meta["shape"])
            nbytes = nelement * buffer_meta["dtype"].itemsize
            if index % 2 == 0:
                workspace1_nbytes = max(workspace1_nbytes, nbytes)
            else:
                workspace2_nbytes = max(workspace2_nbytes, nbytes)

        output_key = "down_output" if self.is_batched() else "output"
        output_shape = buffer_metas[output_key]["shape"]
        elem_size = self.moe_config.in_dtype.itemsize

        return (
            (workspace1_nbytes // elem_size,),
            (workspace2_nbytes // elem_size,),
            output_shape,
        )

    def workspace_shapes(
        self,
        M: int,
        N: int,
        K: int,
        topk: int,
        global_num_experts: int,
        local_num_experts: int,
        expert_tokens_meta: mk.ExpertTokensMetadata | None,
        activation: MoEActivation,
    ) -> tuple[tuple[int, ...], tuple[int, ...], tuple[int, ...]]:
        return self._workspace_shapes(M, topk, activation)

    def make_workspaces(self, M: int, topk: int, activation: MoEActivation):
        shapes = self._workspace_shapes(M, topk, activation)
        workspace1_shape, workspace2_shape, output_shape = shapes
        torch_dtype = self.moe_config.in_dtype
        workspace1, workspace2 = current_workspace_manager().get_simultaneous(
            (workspace1_shape, torch_dtype),
            (workspace2_shape, torch_dtype),
        )
        output = _resize_cache(workspace1, output_shape)
        return workspace1, workspace2, output

    def prepare_buffers(
        self,
        workspace1: torch.Tensor,
        workspace2: torch.Tensor,
        M: int,
        topk: int,
        activation: MoEActivation,
    ) -> dict[str, torch.Tensor]:
        buffer_metas, required_buffers = self.get_buffer_metas(M, topk, activation)
        buffers = {}
        for index, name in enumerate(required_buffers[::-1]):
            buffer_meta = buffer_metas[name]
            workspace = workspace1 if index % 2 == 0 else workspace2
            workspace = workspace.view(buffer_meta["dtype"])
            buffers[name] = _resize_cache(workspace, buffer_meta["shape"])

        return buffers

    # Note: apply method is implemented by subclasses following the
    # standard FusedMoEExpertsModular.apply signature

    @staticmethod
    def is_supported_config(
        cls: type[mk.FusedMoEExperts],
        moe_config: FusedMoEConfig,
        weight_key: QuantKey | None,
        activation_key: QuantKey | None,
        activation_format: mk.FusedMoEActivationFormat,
    ) -> tuple[bool, str | None]:
        supported, reason = mk.FusedMoEExpertsModular.is_supported_config(
            cls,
            moe_config,
            weight_key,
            activation_key,
            activation_format,
        )

        if supported:
            assert hasattr(cls, "humming_gemm_type")
            gemm_type = cls.humming_gemm_type().value.lower()
            preferred_gemm_type = get_humming_moe_gemm_type(moe_config)
            supported = preferred_gemm_type.lower() == gemm_type
            if not supported:
                reason = (
                    f"preferred gemm type {preferred_gemm_type} != "
                    f"supported gemm type {gemm_type}"
                )

        return supported, reason

    def apply_activation(
        self,
        activation: MoEActivation,
        output: torch.Tensor,
        input: torch.Tensor,
        valid_token_counts: torch.Tensor | None = None,
    ) -> None:
        activation_kwargs: dict[str, Any] = dict(
            activation=activation,
            input=input,
            output=output,
        )
        if valid_token_counts is not None:
            activation_kwargs["valid_token_counts"] = valid_token_counts
        self.activation(**activation_kwargs)

    def fused_situ_quant_enabled(self, activation: MoEActivation) -> bool:
        """Whether the SITU activation + w2 quant can be fused into one kernel.

        Fused only for k-major block-FP8 group-128 e4m3 with float32 scales --
        the sole layout situ_and_mul_quant supports. A float32 as_dtype rules out
        MXMMA (which uses e8m0 m-major scales), so a group-128 scale is
        guaranteed k-major. Everything else (m-major MXFP8, other group sizes,
        16-bit passthrough, non-SITU) falls back to the separate
        situ_and_mul + quantize_input path.
        """
        if activation != MoEActivation.SITU:
            return False
        w2cfg = self.humming_configs["w2"]
        # Report the specific blocker once so a silent fallback to the unfused
        # situ_and_mul + quant pair is never a mystery in a trace.
        reason: str | None = None
        if not (w2cfg.a_dtype.num_bits == 8 and str(w2cfg.a_dtype) == "float8e4m3"):
            reason = f"w2 a_dtype is {w2cfg.a_dtype} (need float8e4m3)"
        elif w2cfg.input_scale_group_size != 128:
            reason = (
                f"w2 input_scale_group_size is {w2cfg.input_scale_group_size} "
                "(need 128)"
            )
        elif not (w2cfg.as_dtype is not None and str(w2cfg.as_dtype) == "float32"):
            reason = f"w2 as_dtype is {w2cfg.as_dtype} (need float32, k-major)"
        if reason is not None:
            logger.warning_once(
                "Humming fused SITU+FP8 quant disabled, using unfused "
                "situ_and_mul + quant: %s",
                reason,
            )
            return False
        return True

    def fused_situ_quant(
        self,
        gate_up_output: torch.Tensor,
        quanted_down_input: torch.Tensor,
        num_valid_tokens: torch.Tensor | None,
        topk: int = 1,
    ) -> tuple[torch.Tensor, torch.Tensor]:
        """Run the fused SITU activation + FP8 quant for the w2 input.

        Returns ``(quanted_down_input, input_scale)`` in the same layout the
        unfused ``apply_activation`` + ``quantize_input("w2")`` pair produces: an
        fp8 [M, d] tensor plus a k-major block-FP8 float32 [M, d // 128] group
        scale. Only reached when ``fused_situ_quant_enabled`` (group_size 128).
        """
        from vllm.model_executor.layers.fused_moe.activation import (
            situ_and_mul_quant,
        )

        cfg = self.activation_config
        assert cfg.activation_situ_beta is not None, (
            "SITU requires activation_situ_beta from FusedMoEConfig"
        )
        group_size = self.humming_configs["w2"].input_scale_group_size
        m, d = quanted_down_input.size(0), quanted_down_input.size(1)
        input_scale = torch.empty(
            (m, d // group_size),
            dtype=torch.float32,
            device=quanted_down_input.device,
        )
        situ_and_mul_quant(
            quanted_down_input,
            input_scale,
            gate_up_output,
            beta=cfg.activation_situ_beta,
            linear_beta=cfg.activation_situ_linear_beta,
            group_size=group_size,
            num_valid_tokens=num_valid_tokens,
            topk=topk,
        )
        return quanted_down_input, input_scale