Files
wehub-resource-sync 59a0a3844c
PR Test AMD / cancel-on-close (push) Has been skipped
PR Test NVIDIA ARM / scan (push) Has been skipped
PR Test NVIDIA / cancel-on-close (push) Has been skipped
PR Test AMD / scan (push) Has been skipped
PR Test NVIDIA ARM / cancel-on-close (push) Has been skipped
PR Test NVIDIA / scan (push) Has been skipped
Release Docker Images / build (cu129-torch-2.11.0) (push) Has been skipped
Release Docker Images / build (cu130-torch-2.11.0) (push) Has been skipped
Release PyPI / publish (push) Has been skipped
Scheduler Python Test / test (push) Successful in 27m19s
Docs / build (push) Successful in 28m8s
Scheduler C++ Test / test (push) Successful in 28m19s
Scheduler C++ Test / test-flat (push) Successful in 28m18s
Docs / deploy (push) Has been cancelled
PR Test AMD / finish (push) Has been cancelled
PR Test NVIDIA / finish (push) Has been cancelled
PR Test NVIDIA ARM / finish (push) Has been cancelled
PR Test NVIDIA ARM / ${{ matrix.name }} (${{ matrix.runner }}) (push) Has been cancelled
PR Test AMD / ${{ matrix.name }} (${{ matrix.runner }}) (push) Has been cancelled
PR Test NVIDIA / ${{ matrix.name }} (${{ matrix.runner }}) (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:32:31 +08:00

934 lines
30 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# Copyright (c) 2026 LightSeek Foundation
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in
# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
"""Inference-only MiniMax-M2 family model compatible with HuggingFace weights."""
# ruff: noqa: E402
import logging
from collections.abc import Iterable
from typing import Any, cast
import torch
import triton
import triton.language as tl
from tokenspeed_kernel.ops.communication.trtllm import (
minimax_allreduce_rms_qk,
trtllm_create_ipc_workspace_for_minimax,
)
from tokenspeed_kernel.platform import current_platform
from tokenspeed_kernel.torch_compile import get_compiler_backend
from torch import nn
from tokenspeed.runtime.configs.minimax_m2_config import MiniMaxM2Config
from tokenspeed.runtime.distributed.comm_ops import all_reduce
from tokenspeed.runtime.distributed.mapping import Mapping
from tokenspeed.runtime.execution.context import ForwardContext
from tokenspeed.runtime.layers.linear import (
QKVParallelLinear,
ReplicatedLinear,
RowParallelLinear,
)
from tokenspeed.runtime.layers.logits_processor import LogitsProcessor
from tokenspeed.runtime.layers.moe import (
ExpertCheckpointSchema,
build_moe_checkpoint_loader,
)
from tokenspeed.runtime.layers.moe.topk import TopK
from tokenspeed.runtime.layers.moe.utils import RoutingMethodType
from tokenspeed.runtime.layers.paged_attention import PagedAttention
from tokenspeed.runtime.layers.quantization.base_config import QuantizationConfig
from tokenspeed.runtime.layers.rotary_embedding import get_rope
from tokenspeed.runtime.layers.vocab_parallel_embedding import ParallelLMHead
from tokenspeed.runtime.model_loader.weight_utils import (
default_weight_loader,
sharded_weight_loader,
)
from tokenspeed.runtime.models.base import (
BaseCausalLM,
BaseMoEDecoderLayer,
BaseTransformerModel,
)
from tokenspeed.runtime.models.utils import (
create_fused_set_kv_buffer_arg,
validate_attention_partition,
)
from tokenspeed.runtime.moe.expert_location import ModelConfigForExpertLocation
from tokenspeed.runtime.utils import (
LazyValue,
add_prefix,
set_weight_attrs,
)
from tokenspeed.runtime.utils.env import envs, global_server_args_dict
from tokenspeed.runtime.utils.pdl import pdl_enabled
logger = logging.getLogger(__name__)
_is_nvidia = current_platform().is_nvidia
if _is_nvidia:
from tokenspeed_kernel.thirdparty.cuda import fp32_router_gemm
from tokenspeed.runtime.layers.moe.expert import MoELayer as _MoELayer
MoELayer = _MoELayer
class MiniMaxM2SparseMoeBlock(nn.Module):
def __init__(
self,
config: MiniMaxM2Config,
mapping: Mapping,
quant_config: QuantizationConfig | None = None,
layer_index: int = -1,
prefix: str = "",
):
super().__init__()
self.mapping = mapping
self.layer_index = layer_index
self.tp_size = mapping.world_size
if self.tp_size > config.num_local_experts:
raise ValueError(
f"Tensor parallel size {self.tp_size} is greater than "
f"the number of experts {config.num_local_experts}."
)
self.gate = ReplicatedLinear(
config.hidden_size,
config.num_local_experts,
bias=False,
quant_config=None,
params_dtype=torch.float32,
prefix=add_prefix("gate", prefix),
)
if config.use_routing_bias:
self.routing_bias = nn.Parameter(
torch.zeros(config.num_local_experts, dtype=torch.float32)
)
else:
self.routing_bias = None
self.use_fp32_router_gemm = (
current_platform().is_hopper_plus
and config.hidden_size == 3072
and config.num_local_experts == 256
)
routing_config = {
"n_group": 1,
"topk_group": 1,
"routed_scaling_factor": 1.0,
"normalize_topk_weights": True,
"correction_bias": self.routing_bias,
"routing_method_type": RoutingMethodType.MiniMax2,
}
self.experts = MoELayer(
top_k=config.num_experts_per_tok,
num_experts=config.num_local_experts
+ global_server_args_dict["ep_num_redundant_experts"],
hidden_size=config.hidden_size,
intermediate_size=config.intermediate_size,
quant_config=quant_config,
layer_index=layer_index,
prefix=prefix,
tp_rank=self.mapping.moe.tp_rank,
tp_size=self.mapping.moe.tp_size,
ep_rank=self.mapping.moe.ep_rank,
ep_size=self.mapping.moe.ep_size,
routing_config=routing_config,
)
self.topk = TopK(
top_k=config.num_experts_per_tok,
renormalize=True,
use_grouped_topk=True,
num_expert_group=1,
topk_group=1,
correction_bias=self.routing_bias,
routed_scaling_factor=1.0,
output_format=self.experts.topk_output_format,
)
def get_moe_routed_weights(self):
return [
x.data
for name, x in self.experts.named_parameters()
if name not in ["correction_bias"]
]
def forward(
self,
hidden_states: torch.Tensor,
num_global_tokens: int,
max_num_tokens_per_gpu: int,
) -> torch.Tensor:
num_tokens, hidden_dim = hidden_states.shape
hidden_states = hidden_states.view(-1, hidden_dim)
# FP32 Router GEMM.
if self.use_fp32_router_gemm and hidden_states.shape[0] > 0:
router_logits = fp32_router_gemm(hidden_states, self.gate.weight)
else:
router_logits, _ = self.gate(hidden_states.to(torch.float32))
if hidden_states.shape[0] > 0:
topk_output = self.topk(hidden_states, router_logits)
else:
topk_output = self.topk.empty_topk_output(
hidden_states.device,
hidden_states=hidden_states,
router_logits=router_logits,
)
# Experts.
final_hidden_states = self.experts(
hidden_states=hidden_states,
topk_output=topk_output,
num_global_tokens=num_global_tokens,
max_num_tokens_per_gpu=max_num_tokens_per_gpu,
)
return final_hidden_states.view(num_tokens, hidden_dim)
@triton.jit
def _rmsnorm_sumsq_kernel(
x1_ptr,
x2_ptr,
stride_x1,
stride_x2,
sum_sq_ptr,
B,
D1,
D2,
BLOCK_SIZE1: tl.constexpr,
BLOCK_SIZE2: tl.constexpr,
):
row_id = tl.program_id(0)
x1_row = x1_ptr + row_id * stride_x1
x2_row = x2_ptr + row_id * stride_x2
offsets1 = tl.arange(0, BLOCK_SIZE1)
offsets2 = tl.arange(0, BLOCK_SIZE2)
x1 = tl.load(x1_row + offsets1, mask=offsets1 < D1, other=0.0).to(tl.float32)
x2 = tl.load(x2_row + offsets2, mask=offsets2 < D2, other=0.0).to(tl.float32)
tl.store(sum_sq_ptr + row_id, tl.sum(x1 * x1, axis=0))
tl.store(sum_sq_ptr + row_id + B, tl.sum(x2 * x2, axis=0))
@triton.jit
def _rmsnorm_apply_kernel(
x1_ptr,
x2_ptr,
w1_ptr,
w2_ptr,
sum_sq_ptr,
out1_ptr,
out2_ptr,
B,
D1,
D2,
stride_x1,
stride_x2,
tp_world,
eps,
BLOCK_SIZE1: tl.constexpr,
BLOCK_SIZE2: tl.constexpr,
):
row_id = tl.program_id(0)
x1_row = x1_ptr + row_id * stride_x1
x2_row = x2_ptr + row_id * stride_x2
out1_row = out1_ptr + row_id * stride_x1
out2_row = out2_ptr + row_id * stride_x2
inv_rms1 = tl.rsqrt(tl.load(sum_sq_ptr + row_id) / D1 / tp_world + eps)
inv_rms2 = tl.rsqrt(tl.load(sum_sq_ptr + row_id + B) / D2 / tp_world + eps)
offsets1 = tl.arange(0, BLOCK_SIZE1)
offsets2 = tl.arange(0, BLOCK_SIZE2)
mask1 = offsets1 < D1
mask2 = offsets2 < D2
x1 = tl.load(x1_row + offsets1, mask=mask1, other=0.0)
w1 = tl.load(w1_ptr + offsets1, mask=mask1, other=1.0)
x2 = tl.load(x2_row + offsets2, mask=mask2, other=0.0)
w2 = tl.load(w2_ptr + offsets2, mask=mask2, other=1.0)
tl.store(
out1_row + offsets1,
(x1.to(tl.float32) * inv_rms1 * w1.to(tl.float32)).to(x1.dtype),
mask=mask1,
)
tl.store(
out2_row + offsets2,
(x2.to(tl.float32) * inv_rms2 * w2.to(tl.float32)).to(x2.dtype),
mask=mask2,
)
@torch.compile(dynamic=True, backend=get_compiler_backend())
def fused_qk_rmsnorm_triton(
q: torch.Tensor,
k: torch.Tensor,
q_weight: torch.Tensor,
k_weight: torch.Tensor,
tp_size: int,
tp_rank: int,
tp_group: tuple[int, ...],
eps: float = 1e-6,
) -> tuple[torch.Tensor, torch.Tensor]:
"""Fused QK RMSNorm: sumsq → allreduce → apply, using 2 Triton kernels."""
q = q.contiguous()
k = k.contiguous()
B, D1 = q.shape
_, D2 = k.shape
BLOCK_SIZE1 = triton.next_power_of_2(D1)
BLOCK_SIZE2 = triton.next_power_of_2(D2)
# Pad for allreduce alignment (16-byte = 4 floats)
B_padded = (B + B + 3) // 4 * 4
sum_sq = torch.empty(B_padded, device=q.device, dtype=torch.float32)
_rmsnorm_sumsq_kernel[(B,)](
q,
k,
q.stride(0),
k.stride(0),
sum_sq,
B,
D1,
D2,
BLOCK_SIZE1,
BLOCK_SIZE2,
)
if tp_size > 1:
sum_sq = all_reduce(sum_sq, tp_group)
out1 = torch.empty_like(q)
out2 = torch.empty_like(k)
_rmsnorm_apply_kernel[(B,)](
q,
k,
q_weight,
k_weight,
sum_sq,
out1,
out2,
B,
D1,
D2,
q.stride(0),
k.stride(0),
tp_size,
eps,
BLOCK_SIZE1,
BLOCK_SIZE2,
)
return out1, out2
def _minimax_fast_path_available(
q: torch.Tensor,
k: torch.Tensor,
tp_size: int,
) -> bool:
"""Fast-path CUDA kernel (Lamport AR fused with RMSNorm) is usable only for
TP in {2,4,8,16} and global head dims (Q, K) == (6144, 1024)."""
if tp_size not in (2, 4, 8, 16):
return False
if q.dim() != 2 or k.dim() != 2:
return False
if q.shape[-1] * tp_size != 6144 or k.shape[-1] * tp_size != 1024:
return False
if q.dtype not in (torch.float16, torch.bfloat16):
return False
return True
class _MinimaxARWorkspace:
"""Singleton holder for the dedicated MiniMax AR+RMSNorm IPC workspace.
One workspace per (tp_group, dtype_elem_size, max_token_num). Lifetime is
tied to the process; it lives as long as the model.
"""
def __init__(self) -> None:
self._entries: dict[tuple[tuple[int, ...], int, int], dict[str, Any]] = {}
def get_or_create(
self,
tp_rank: int,
tp_group: tuple[int, ...],
max_token_num: int,
dtype_elem_size: int,
) -> torch.Tensor | None:
key = (tp_group, dtype_elem_size, max_token_num)
# Grow max_token_num if needed: find any existing entry for the same
# (group, dtype) and check whether we can reuse it.
for (g, sz, cap), entry in self._entries.items():
if g == tp_group and sz == dtype_elem_size and cap >= max_token_num:
return entry["workspace"]
from tokenspeed.runtime.distributed.process_group_manager import (
process_group_manager as pg_manager,
)
device_group = pg_manager.get_process_group("nccl", tp_group)
try:
ipc_handles, workspace = trtllm_create_ipc_workspace_for_minimax(
tp_rank=tp_rank,
tp_size=len(tp_group),
max_token_num=max_token_num,
group=device_group,
dtype_elem_size=dtype_elem_size,
)
except Exception:
logger.exception("Failed to create MiniMax AR+RMSNorm IPC workspace")
return None
self._entries[key] = {
"ipc_handles": ipc_handles,
"workspace": workspace,
"device_group": device_group,
}
return workspace
_minimax_ar_workspace = _MinimaxARWorkspace()
_FORCE_TRITON_AR_RMSNORM = envs.TOKENSPEED_MINIMAX_AR_USE_TRITON.get()
def fused_qk_rmsnorm(
q: torch.Tensor,
k: torch.Tensor,
q_weight_fp32: torch.Tensor,
k_weight_fp32: torch.Tensor,
q_weight_bf16: torch.Tensor | None,
k_weight_bf16: torch.Tensor | None,
tp_size: int,
tp_rank: int,
tp_group: tuple[int, ...],
eps: float = 1e-6,
) -> tuple[torch.Tensor, torch.Tensor]:
"""Route to the Lamport fused-AR QK RMSNorm kernel when its shape
constraints hold, else fall back to the Triton sumsq/apply path.
Setting TOKENSPEED_MINIMAX_AR_USE_TRITON=1 forces the Triton path (A/B debug)."""
if (
not _FORCE_TRITON_AR_RMSNORM
and q_weight_bf16 is not None
and k_weight_bf16 is not None
and _minimax_fast_path_available(q, k, tp_size)
):
num_tokens = q.shape[0]
# Allocate once with a generous ceiling so batch-size changes never
# force reallocation. 16384 tokens × TP=16 fits in ~6MB of lamport buffer.
_MINIMAX_WORKSPACE_CAP = 16384
workspace = _minimax_ar_workspace.get_or_create(
tp_rank=tp_rank,
tp_group=tp_group,
max_token_num=max(num_tokens, _MINIMAX_WORKSPACE_CAP),
dtype_elem_size=q.element_size(),
)
if workspace is not None:
# Kernel reads q/k at their row stride (q_row_stride_f4), so a
# non-contiguous slice from a fused-QKV split is fine.
return minimax_allreduce_rms_qk(
q=q,
k=k,
norm_weight_q=q_weight_bf16,
norm_weight_k=k_weight_bf16,
workspace_ptrs=workspace,
rank=tp_rank,
nranks=tp_size,
eps=eps,
trigger_completion_at_end=True,
launch_with_pdl=pdl_enabled(),
)
return fused_qk_rmsnorm_triton(
q, k, q_weight_fp32, k_weight_fp32, tp_size, tp_rank, tp_group, eps
)
class MiniMaxM2RMSNormTP(nn.Module):
"""Tensor-parallel RMSNorm for MiniMax Q/K normalization."""
def __init__(
self,
global_hidden_size: int,
tp_rank: int,
tp_size: int,
tp_group: tuple[int, ...],
eps: float = 1e-6,
) -> None:
super().__init__()
if global_hidden_size % tp_size != 0:
raise ValueError(
f"global_hidden_size={global_hidden_size} must be divisible by tp_size={tp_size}."
)
self.local_hidden_size = global_hidden_size // tp_size
self.tp_rank = tp_rank
self.tp_size = tp_size
self.tp_group = tp_group
self.variance_epsilon = eps
self.weight = nn.Parameter(torch.ones(self.local_hidden_size))
self._weight_bf16: torch.Tensor | None = None
self._weight_bf16_src_ptr: int = 0
set_weight_attrs(
self.weight, {"weight_loader": sharded_weight_loader(0, self.tp_rank)}
)
def bf16_weight(self) -> torch.Tensor:
# The Lamport-fused AR kernel requires bf16 gamma. Cache a bf16 copy
# of the fp32 Parameter; refresh if the backing storage ever changes
# (e.g. weights reloaded).
src_ptr = self.weight.data_ptr()
if self._weight_bf16 is None or self._weight_bf16_src_ptr != src_ptr:
self._weight_bf16 = self.weight.detach().to(torch.bfloat16).contiguous()
self._weight_bf16_src_ptr = src_ptr
return self._weight_bf16
def remap_minimax_weight_name(name: str) -> str:
"""Map HF checkpoint-only MiniMax names to local parameter names."""
if "e_score_correction_bias" in name:
name = name.replace("e_score_correction_bias", "routing_bias")
if "block_sparse_moe" in name:
name = name.replace("block_sparse_moe", "mlp")
return name
def get_spec_layer_idx_from_weight_name(
config: MiniMaxM2Config, weight_name: str
) -> int | None:
"""Return the extra speculative layer index encoded after main layers.
Public MiniMax-M2 configs can carry speculative-decoding metadata even when
the released checkpoints do not include those extra layer weights. The
serving model instantiated here is main-model only, so extra layers beyond
``num_hidden_layers`` should be ignored if a checkpoint ever includes them.
"""
num_spec_modules = int(getattr(config, "num_mtp_modules", 0) or 0)
layers_per_spec_module = int(getattr(config, "mtp_transformer_layers", 1) or 1)
num_spec_layers = num_spec_modules * layers_per_spec_module
start_layer = int(config.num_hidden_layers)
for i in range(num_spec_layers):
layer_idx = start_layer + i
if weight_name.startswith(f"model.layers.{layer_idx}."):
return layer_idx
return None
class MiniMaxM2Attention(nn.Module):
def __init__(
self,
hidden_size: int,
num_heads: int,
num_kv_heads: int,
mapping: Mapping,
layer_id: int = 0,
rope_theta: float = 10000,
rope_scaling: dict[str, Any] | None = None,
max_position_embeddings: int = 8192,
head_dim: int | None = None,
rotary_dim: int | None = None,
rms_norm_eps: float = 1e-06,
attention_bias: bool = False,
quant_config: QuantizationConfig | None = None,
prefix: str = "",
) -> None:
super().__init__()
self.layer_id = layer_id
self.mapping = mapping
self.hidden_size = hidden_size
self.attn_tp_size = mapping.attn.tp_size
self.attn_tp_rank = mapping.attn.tp_rank
self.attn_tp_group = mapping.attn.tp_group
self.total_num_heads = num_heads
self.total_num_kv_heads = num_kv_heads
validate_attention_partition(
self.total_num_heads,
self.total_num_kv_heads,
self.attn_tp_size,
)
self.num_heads = self.total_num_heads // self.attn_tp_size
self.num_kv_heads = max(1, self.total_num_kv_heads // self.attn_tp_size)
self.head_dim = head_dim or hidden_size // self.total_num_heads
self.rotary_dim = rotary_dim or self.head_dim
self.q_size = self.num_heads * self.head_dim
self.kv_size = self.num_kv_heads * self.head_dim
self.scaling = self.head_dim**-0.5
self.qkv_proj = QKVParallelLinear(
hidden_size,
self.head_dim,
self.total_num_heads,
self.total_num_kv_heads,
bias=attention_bias,
quant_config=quant_config,
tp_rank=self.attn_tp_rank,
tp_size=self.attn_tp_size,
tp_group=self.attn_tp_group,
prefix=add_prefix("qkv_proj", prefix),
)
self.o_proj = RowParallelLinear(
self.total_num_heads * self.head_dim,
hidden_size,
bias=attention_bias,
quant_config=quant_config,
reduce_results=False,
tp_rank=self.attn_tp_rank,
tp_size=self.attn_tp_size,
tp_group=self.attn_tp_group,
prefix=add_prefix("o_proj", prefix),
)
self.rotary_emb = get_rope(
self.head_dim,
rotary_dim=self.rotary_dim,
max_position=max_position_embeddings,
base=int(rope_theta),
rope_scaling=rope_scaling,
)
self.attn = PagedAttention(
self.num_heads,
self.head_dim,
self.scaling,
num_kv_heads=self.num_kv_heads,
layer_id=layer_id,
)
self.q_norm = MiniMaxM2RMSNormTP(
self.total_num_heads * self.head_dim,
tp_rank=self.attn_tp_rank,
tp_size=self.attn_tp_size,
tp_group=self.attn_tp_group,
eps=rms_norm_eps,
)
self.k_norm = MiniMaxM2RMSNormTP(
self.total_num_kv_heads * self.head_dim,
tp_rank=self.attn_tp_rank,
tp_size=self.attn_tp_size,
tp_group=self.attn_tp_group,
eps=rms_norm_eps,
)
def forward(
self,
positions: torch.Tensor,
hidden_states: torch.Tensor,
ctx: ForwardContext,
out_cache_loc: torch.Tensor,
) -> torch.Tensor:
if hidden_states.shape[0] == 0:
return hidden_states
qkv, _ = self.qkv_proj(hidden_states)
q, k, v = qkv.split([self.q_size, self.kv_size, self.kv_size], dim=-1)
q, k = fused_qk_rmsnorm(
q,
k,
self.q_norm.weight,
self.k_norm.weight,
self.q_norm.bf16_weight(),
self.k_norm.bf16_weight(),
self.q_norm.tp_size,
self.q_norm.tp_rank,
self.q_norm.tp_group,
self.q_norm.variance_epsilon,
)
fused_kv_arg = None
if ctx.attn_backend.support_kv_cache_prewrite():
n = q.shape[0]
v_3d = v.view(n, self.num_kv_heads, self.head_dim)
fused_kv_arg = create_fused_set_kv_buffer_arg(
value=v_3d,
layer=self.attn,
out_cache_loc=out_cache_loc,
token_to_kv_pool=ctx.token_to_kv_pool,
)
if fused_kv_arg is not None:
q_rope = torch.empty((n, self.q_size), dtype=q.dtype, device=q.device)
q, k = self.rotary_emb(
positions,
q,
k,
fused_set_kv_buffer_arg=fused_kv_arg,
output_q_rope=q_rope,
enable_pdl=pdl_enabled(),
)
attn_output = self.attn(
q_rope,
None,
None,
save_kv_cache=False,
ctx=ctx,
out_cache_loc=out_cache_loc,
)
else:
q, k = self.rotary_emb(positions, q, k)
q = q.view(-1, self.num_heads, self.head_dim)
k = k.view(-1, self.num_kv_heads, self.head_dim)
v = v.view(-1, self.num_kv_heads, self.head_dim)
attn_output = self.attn(q, k, v, ctx=ctx, out_cache_loc=out_cache_loc)
output, _ = self.o_proj(attn_output)
return output
class MiniMaxM2DecoderLayer(BaseMoEDecoderLayer):
def __init__(
self,
config: MiniMaxM2Config,
layer_id: int,
mapping: Mapping,
quant_config: QuantizationConfig | None = None,
prefix: str = "",
) -> None:
self._config = config
self._mapping = mapping
self._quant_config = quant_config
super().__init__(
config=config,
layer_id=layer_id,
mapping=mapping,
quant_config=quant_config,
prefix=prefix,
)
def resolve_attn(self, prefix: str) -> nn.Module:
config = self._config
return MiniMaxM2Attention(
hidden_size=config.hidden_size,
num_heads=config.num_attention_heads,
num_kv_heads=config.num_key_value_heads,
mapping=self._mapping,
layer_id=self.layer_id,
rope_theta=config.rope_theta,
rope_scaling=getattr(config, "rope_scaling", None),
max_position_embeddings=config.max_position_embeddings,
head_dim=config.head_dim,
rotary_dim=config.rotary_dim,
rms_norm_eps=config.rms_norm_eps,
attention_bias=config.attention_bias,
quant_config=self._quant_config,
prefix=add_prefix("self_attn", prefix),
)
def resolve_mlp(self, prefix: str) -> nn.Module:
return MiniMaxM2SparseMoeBlock(
config=self._config,
mapping=self._mapping,
quant_config=self._quant_config,
layer_index=self.layer_id,
prefix=add_prefix("block_sparse_moe", prefix),
)
class MiniMaxM2Model(BaseTransformerModel):
layer_cls = MiniMaxM2DecoderLayer
class MiniMaxM2ForCausalLM(BaseCausalLM):
model_cls = MiniMaxM2Model
fall_back_to_pt_during_load = False
def __init__(
self,
config: MiniMaxM2Config,
mapping: Mapping,
quant_config: QuantizationConfig | None = None,
prefix: str = "",
) -> None:
super().__init__(config, mapping, quant_config, prefix)
self._routed_experts_weights_of_layer = LazyValue(
lambda: {
layer_id: cast(
MiniMaxM2DecoderLayer, self.model.layers[layer_id]
).mlp.get_moe_routed_weights()
for layer_id in range(len(self.model.layers))
}
)
@property
def routed_experts_weights_of_layer(self):
return self._routed_experts_weights_of_layer.value
def resolve_lm_head(self, config, quant_config, prefix):
if self.mapping.attn.has_dp:
return ReplicatedLinear(
config.hidden_size,
config.vocab_size,
bias=False,
prefix=add_prefix("lm_head", prefix),
)
return ParallelLMHead(
config.vocab_size,
config.hidden_size,
quant_config=quant_config,
prefix=add_prefix("lm_head", prefix),
tp_rank=self.mapping.attn.tp_rank,
tp_size=self.mapping.attn.tp_size,
tp_group=self.mapping.attn.tp_group,
)
def resolve_logits_processor(self, config):
return LogitsProcessor(
config,
skip_all_gather=self.mapping.attn.has_dp,
tp_rank=self.mapping.attn.tp_rank,
tp_size=self.mapping.attn.tp_size,
tp_group=self.mapping.attn.tp_group,
)
def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]], **kwargs):
stacked_params_mapping = [
("qkv_proj", "q_proj", "q"),
("qkv_proj", "k_proj", "k"),
("qkv_proj", "v_proj", "v"),
]
# Skip loading extra parameters for GPTQ/nvfp4 models.
ignore_suffixes = (
".bias",
"_bias",
".k_scale",
"_k_scale",
".v_scale",
"_v_scale",
".weight_scale",
"_weight_scale",
".weight_scale_2",
"_weight_scale_2",
".input_scale",
"_input_scale",
)
loaded_params: set[str] = set()
params_dict = dict(self.named_parameters(remove_duplicate=False))
moe_loader = build_moe_checkpoint_loader(
params_dict=params_dict,
expert_schema=ExpertCheckpointSchema(
gate_proj_name="w1",
down_proj_name="w2",
up_proj_name="w3",
),
num_experts=self.config.num_local_experts,
ep_rank=self.mapping.moe.ep_rank,
ep_size=self.mapping.moe.ep_size,
)
for name, loaded_weight in weights:
if "rotary_emb.inv_freq" in name:
continue
if get_spec_layer_idx_from_weight_name(self.config, name) is not None:
continue
name = remap_minimax_weight_name(name)
for param_name, weight_name, shard_id in stacked_params_mapping:
if weight_name not in name:
continue
if "mlp.experts." in name:
continue
name = name.replace(weight_name, param_name)
if name.endswith(ignore_suffixes) and name not in params_dict:
continue
if name not in params_dict:
continue
param = params_dict[name]
weight_loader = param.weight_loader
weight_loader(param, loaded_weight, shard_id)
break
else:
# moe_loader.matches must be checked BEFORE the
# ignore_suffixes gate. Expert scale names end with
# `.weight_scale` / `.weight_scale_2` / `.input_scale` — those
# match ignore_suffixes, and the pre-remap checkpoint name
# (e.g. `experts.10.w1.weight_scale`) is not in params_dict,
# so the ignore gate would otherwise silently drop every FP4
# expert scale and leave the layer with uninitialized scales.
if moe_loader.matches(name):
name = moe_loader.load(name, loaded_weight)
else:
if name.endswith(ignore_suffixes) and name not in params_dict:
continue
if name not in params_dict:
continue
param = params_dict[name]
weight_loader = getattr(
param, "weight_loader", default_weight_loader
)
weight_loader(param, loaded_weight)
loaded_params.add(name)
return loaded_params
@classmethod
def get_model_config_for_expert_location(cls, config):
return ModelConfigForExpertLocation(
num_layers=config.num_hidden_layers,
num_logical_experts=config.num_local_experts,
num_groups=None,
)
EntryClass = MiniMaxM2ForCausalLM