text_encoder/fused_moe_ops.py

# Copyright 2025 Bytedance Ltd. and/or its affiliates
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Standalone, inference-only VeOmni v0.1.0 fused-MoE compatibility shim.

This module preserves the ``veomni.ops.fused_moe_forward`` call signature used
by VeOmni v0.1.0 while removing VeOmni's training, Expert Parallelism (EP), NPU,
and Seed-kernel dependencies. It is intended for single-device inference only.

The CUDA fast path uses a small Triton grouped-linear kernel. If Triton is not
available, the tensors are not on CUDA, or ``LLADA_MOE_BACKEND=eager`` is set,
the implementation falls back to ordinary PyTorch operations.

Replace the original model-code import with, for example,
``from .fused_moe_v010 import fused_moe_forward``.

Derived from ByteDance-Seed/VeOmni v0.1.0.post1:
https://github.com/ByteDance-Seed/VeOmni/tree/v0.1.0.post1
"""

from __future__ import annotations

import os

import torch
import torch.nn.functional as F

try:
    import triton
    import triton.language as tl
except ImportError:  # The eager fallback does not require Triton.
    triton = None
    tl = None


_SUPPORTED_TRITON_DTYPES = (torch.float16, torch.bfloat16)


if triton is not None:

    @triton.jit
    def _grouped_linear_kernel(
        input_ptr,
        weight_ptr,
        output_ptr,
        expert_cumsum_ptr,
        N: tl.constexpr,
        K: tl.constexpr,
        BLOCK_M: tl.constexpr,
        BLOCK_N: tl.constexpr,
        BLOCK_K: tl.constexpr,
    ):
        """Compute per-expert ``input @ weight.T`` for contiguous tensors."""
        block_m = tl.program_id(axis=0)
        block_n = tl.program_id(axis=1)
        expert = tl.program_id(axis=2)

        expert_start = tl.load(expert_cumsum_ptr + expert - 1, mask=expert > 0, other=0)
        expert_end = tl.load(expert_cumsum_ptr + expert)
        expert_tokens = expert_end - expert_start

        if block_m * BLOCK_M >= expert_tokens:
            return

        row_offsets = block_m * BLOCK_M + tl.arange(0, BLOCK_M)
        col_offsets = block_n * BLOCK_N + tl.arange(0, BLOCK_N)
        k_offsets = tl.arange(0, BLOCK_K)

        input_ptrs = (
            input_ptr
            + (expert_start + row_offsets[:, None]) * K
            + k_offsets[None, :]
        )
        weight_ptrs = (
            weight_ptr
            + expert * N * K
            + col_offsets[None, :] * K
            + k_offsets[:, None]
        )

        accumulator = tl.zeros((BLOCK_M, BLOCK_N), dtype=tl.float32)
        for k_block in range(0, tl.cdiv(K, BLOCK_K)):
            remaining_k = K - k_block * BLOCK_K
            inputs = tl.load(
                input_ptrs,
                mask=(row_offsets[:, None] < expert_tokens) & (k_offsets[None, :] < remaining_k),
                other=0.0,
            )
            weights = tl.load(
                weight_ptrs,
                mask=(col_offsets[None, :] < N) & (k_offsets[:, None] < remaining_k),
                other=0.0,
            )
            accumulator += tl.dot(inputs, weights)
            input_ptrs += BLOCK_K
            weight_ptrs += BLOCK_K

        output_ptrs = (
            output_ptr
            + (expert_start + row_offsets[:, None]) * N
            + col_offsets[None, :]
        )
        tl.store(
            output_ptrs,
            accumulator,
            mask=(row_offsets[:, None] < expert_tokens) & (col_offsets[None, :] < N),
        )


def _validate_inputs(
    num_experts: int,
    routing_weights: torch.Tensor,
    selected_experts: torch.Tensor,
    hidden_states: torch.Tensor,
    fc1_1_weight: torch.Tensor,
    fc1_2_weight: torch.Tensor,
    fc2_weight: torch.Tensor,
) -> None:
    if num_experts <= 0:
        raise ValueError(f"num_experts must be positive, got {num_experts}")
    if torch.is_grad_enabled():
        raise RuntimeError(
            "This standalone fused_moe_forward is inference-only. Call it under "
            "torch.no_grad() or torch.inference_mode()."
        )
    if hidden_states.ndim != 2:
        raise ValueError(f"hidden_states must have shape [tokens, hidden], got {tuple(hidden_states.shape)}")
    if routing_weights.ndim != 2 or selected_experts.shape != routing_weights.shape:
        raise ValueError(
            "routing_weights and selected_experts must have the same [tokens, top_k] shape, got "
            f"{tuple(routing_weights.shape)} and {tuple(selected_experts.shape)}"
        )
    if routing_weights.shape[1] == 0:
        raise ValueError("top_k must be positive")
    if routing_weights.shape[0] != hidden_states.shape[0]:
        raise ValueError("routing_weights and hidden_states must contain the same number of tokens")
    if selected_experts.dtype not in (torch.int32, torch.int64):
        raise TypeError(f"selected_experts must be int32 or int64, got {selected_experts.dtype}")
    if fc1_1_weight.ndim != 3 or fc1_2_weight.ndim != 3 or fc2_weight.ndim != 3:
        raise ValueError("expert weights must be rank-3 tensors")
    if fc1_1_weight.shape != fc1_2_weight.shape:
        raise ValueError("fc1_1_weight and fc1_2_weight must have identical shapes")

    experts, intermediate_size, hidden_size = fc1_1_weight.shape
    expected_fc2_shape = (experts, hidden_size, intermediate_size)
    if experts != num_experts:
        raise ValueError(f"num_experts={num_experts}, but the weights contain {experts} experts")
    if hidden_states.shape[1] != hidden_size:
        raise ValueError(f"hidden size is {hidden_states.shape[1]}, but the weights expect {hidden_size}")
    if tuple(fc2_weight.shape) != expected_fc2_shape:
        raise ValueError(f"fc2_weight must have shape {expected_fc2_shape}, got {tuple(fc2_weight.shape)}")
    if selected_experts.numel():
        # These scalar checks synchronize CUDA once, before launching harder-to-debug kernels.
        min_expert = int(selected_experts.min().item())
        max_expert = int(selected_experts.max().item())
        if min_expert < 0 or max_expert >= num_experts:
            raise ValueError(f"selected expert IDs must be in [0, {num_experts}), got [{min_expert}, {max_expert}]")

    devices = {
        hidden_states.device,
        routing_weights.device,
        selected_experts.device,
        fc1_1_weight.device,
        fc1_2_weight.device,
        fc2_weight.device,
    }
    if len(devices) != 1:
        raise ValueError(f"all inputs and weights must be on one device, got {sorted(map(str, devices))}")


def _route_tokens(
    num_experts: int,
    routing_weights: torch.Tensor,
    selected_experts: torch.Tensor,
    hidden_states: torch.Tensor,
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
    """Sort routed token copies by expert and return the inverse permutation."""
    top_k = selected_experts.shape[1]
    flat_experts = selected_experts.reshape(-1).to(torch.int64)
    order = torch.argsort(flat_experts, stable=True)
    sorted_hidden_states = hidden_states[torch.div(order, top_k, rounding_mode="floor")].contiguous()
    sorted_routing_weights = routing_weights.reshape(-1)[order].contiguous()
    tokens_per_expert = torch.bincount(flat_experts, minlength=num_experts)
    expert_cumsum = torch.cumsum(tokens_per_expert, dim=0, dtype=torch.int32).contiguous()
    return sorted_hidden_states, sorted_routing_weights, expert_cumsum, order


def _unroute_tokens(
    sorted_outputs: torch.Tensor,
    order: torch.Tensor,
    num_tokens: int,
    top_k: int,
) -> torch.Tensor:
    restored = torch.empty_like(sorted_outputs)
    restored[order] = sorted_outputs
    # VeOmni's v0.1.0 gather kernel accumulates the top-k outputs in FP32.
    return restored.view(num_tokens, top_k, -1).sum(dim=1, dtype=torch.float32).to(sorted_outputs.dtype)


def _grouped_linear_triton(
    inputs: torch.Tensor,
    weights: torch.Tensor,
    expert_cumsum: torch.Tensor,
) -> torch.Tensor:
    if triton is None:  # pragma: no cover - guarded by the caller
        raise RuntimeError("Triton is not available")
    if not inputs.is_contiguous() or not weights.is_contiguous():
        raise ValueError("the Triton path requires contiguous inputs and expert weights")

    num_experts, output_size, input_size = weights.shape
    if inputs.shape[1] != input_size:
        raise ValueError(f"input width is {inputs.shape[1]}, but the weights expect {input_size}")

    output = torch.empty((inputs.shape[0], output_size), dtype=inputs.dtype, device=inputs.device)
    block_m, block_n, block_k = 128, 128, 32
    grid = (
        triton.cdiv(inputs.shape[0], block_m),
        triton.cdiv(output_size, block_n),
        num_experts,
    )
    with torch.cuda.device(inputs.device):
        _grouped_linear_kernel[grid](
            inputs,
            weights,
            output,
            expert_cumsum,
            N=output_size,
            K=input_size,
            BLOCK_M=block_m,
            BLOCK_N=block_n,
            BLOCK_K=block_k,
            num_warps=8,
            num_stages=3,
        )
    return output


def _triton_moe_forward(
    num_experts: int,
    routing_weights: torch.Tensor,
    selected_experts: torch.Tensor,
    hidden_states: torch.Tensor,
    fc1_1_weight: torch.Tensor,
    fc1_2_weight: torch.Tensor,
    fc2_weight: torch.Tensor,
) -> torch.Tensor:
    sorted_hidden, sorted_routing, expert_cumsum, order = _route_tokens(
        num_experts, routing_weights, selected_experts, hidden_states
    )
    gate = _grouped_linear_triton(sorted_hidden, fc1_1_weight, expert_cumsum)
    up = _grouped_linear_triton(sorted_hidden, fc1_2_weight, expert_cumsum)
    intermediate = F.silu(gate) * up
    intermediate.mul_(sorted_routing.unsqueeze(-1))
    sorted_outputs = _grouped_linear_triton(intermediate.contiguous(), fc2_weight, expert_cumsum)
    return _unroute_tokens(sorted_outputs, order, hidden_states.shape[0], selected_experts.shape[1])


def _eager_moe_forward(
    num_experts: int,
    routing_weights: torch.Tensor,
    selected_experts: torch.Tensor,
    hidden_states: torch.Tensor,
    fc1_1_weight: torch.Tensor,
    fc1_2_weight: torch.Tensor,
    fc2_weight: torch.Tensor,
) -> torch.Tensor:
    sorted_hidden, sorted_routing, expert_cumsum, order = _route_tokens(
        num_experts, routing_weights, selected_experts, hidden_states
    )
    expert_ends = expert_cumsum.to(device="cpu", dtype=torch.int64).tolist()
    outputs: list[torch.Tensor] = []
    start = 0
    for expert, end in enumerate(expert_ends):
        if end > start:
            expert_inputs = sorted_hidden[start:end]
            gate = F.linear(expert_inputs, fc1_1_weight[expert])
            up = F.linear(expert_inputs, fc1_2_weight[expert])
            intermediate = F.silu(gate) * up
            intermediate.mul_(sorted_routing[start:end].unsqueeze(-1))
            outputs.append(F.linear(intermediate, fc2_weight[expert]))
        start = end

    sorted_outputs = torch.cat(outputs, dim=0) if outputs else hidden_states.new_empty((0, hidden_states.shape[1]))
    return _unroute_tokens(sorted_outputs, order, hidden_states.shape[0], selected_experts.shape[1])


def fused_moe_forward(
    module: torch.nn.Module,
    num_experts: int,
    routing_weights: torch.Tensor,
    selected_experts: torch.Tensor,
    hidden_states: torch.Tensor,
    fc1_1_weight: torch.Tensor,
    fc1_2_weight: torch.Tensor,
    fc2_weight: torch.Tensor,
) -> torch.Tensor:
    """Run the VeOmni v0.1.0 split-weight MoE operation for inference.

    ``module`` is retained for call-site compatibility. Like VeOmni's original
    non-EP implementation, this function does not use it.

    Set ``LLADA_MOE_BACKEND`` to ``auto`` (default), ``triton``, or ``eager``.
    The ``triton`` setting fails loudly if its requirements are not met;
    ``auto`` falls back to the PyTorch implementation.
    """
    del module
    _validate_inputs(
        num_experts,
        routing_weights,
        selected_experts,
        hidden_states,
        fc1_1_weight,
        fc1_2_weight,
        fc2_weight,
    )

    backend = os.getenv("LLADA_MOE_BACKEND", "auto").lower()
    if backend not in {"auto", "triton", "eager"}:
        raise ValueError(f"LLADA_MOE_BACKEND must be auto, triton, or eager; got {backend!r}")

    compute_dtype = fc1_1_weight.dtype
    if fc1_2_weight.dtype != compute_dtype or fc2_weight.dtype != compute_dtype:
        raise TypeError("all expert weights must have the same dtype")
    hidden_states = hidden_states.to(dtype=compute_dtype)
    routing_weights = routing_weights.to(dtype=compute_dtype)

    if hidden_states.shape[0] == 0:
        return hidden_states

    can_use_triton = (
        triton is not None
        and hidden_states.is_cuda
        and compute_dtype in _SUPPORTED_TRITON_DTYPES
        and fc1_1_weight.is_contiguous()
        and fc1_2_weight.is_contiguous()
        and fc2_weight.is_contiguous()
    )
    if backend == "triton" and not can_use_triton:
        raise RuntimeError(
            "The Triton backend requires Triton, CUDA tensors, contiguous expert weights, "
            "and float16 or bfloat16 weights."
        )
    if backend != "eager" and can_use_triton:
        return _triton_moe_forward(
            num_experts,
            routing_weights,
            selected_experts,
            hidden_states,
            fc1_1_weight,
            fc1_2_weight,
            fc2_weight,
        )
    return _eager_moe_forward(
        num_experts,
        routing_weights,
        selected_experts,
        hidden_states,
        fc1_1_weight,
        fc1_2_weight,
        fc2_weight,
    )


__all__ = ["fused_moe_forward"]