main
py 1,463 lines 56.3 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 import typing
4 import weakref
5 from collections.abc import Callable, Iterable
6 from dataclasses import replace
7 from itertools import islice
8
9 import regex as re
10 import torch
11 import torch.nn as nn
12
13 import vllm.envs as envs
14 from vllm._aiter_ops import rocm_aiter_ops
15 from vllm.config import ParallelConfig, VllmConfig
16 from vllm.distributed import (
17 get_pp_group,
18 get_tensor_model_parallel_rank,
19 get_tensor_model_parallel_world_size,
20 )
21 from vllm.logger import init_logger
22 from vllm.model_executor.layers.activation import SiluAndMul, SiluAndMulWithClamp
23 from vllm.model_executor.layers.fused_moe import (
24 FusedMoEFactory,
25 FusedMoEQuantConfig,
26 GateLinear,
27 RoutedExperts,
28 fused_moe_make_expert_params_mapping,
29 )
30 from vllm.model_executor.layers.fused_moe.experts.rocm_aiter_moe import (
31 rocm_aiter_fused_experts,
32 )
33 from vllm.model_executor.layers.fused_moe.utils import (
34 is_model_fused_shared_expert_compatible,
35 )
36 from vllm.model_executor.layers.layernorm import RMSNorm
37 from vllm.model_executor.layers.linear import (
38 MergedColumnParallelLinear,
39 RowParallelLinear,
40 )
41 from vllm.model_executor.layers.logits_processor import LogitsProcessor
42 from vllm.model_executor.layers.mhc import (
43 HAS_AITER_MHC,
44 HAS_TILELANG_MHC,
45 HCHeadOp,
46 MHCFusedPostPreOp,
47 MHCPostOp,
48 MHCPreOp,
49 )
50 from vllm.model_executor.layers.quantization import QuantizationConfig
51 from vllm.model_executor.layers.quantization.mxfp4 import Mxfp4MoEMethod
52 from vllm.model_executor.layers.quantization.utils.config_utils import (
53 is_shared_expert_quant_fse_compatible,
54 )
55 from vllm.model_executor.layers.vocab_parallel_embedding import (
56 ParallelLMHead,
57 VocabParallelEmbedding,
58 )
59 from vllm.model_executor.model_loader.weight_utils import default_weight_loader
60 from vllm.model_executor.models.interfaces import (
61 EagleModelMixin,
62 SupportsEagle3,
63 SupportsPP,
64 )
65 from vllm.model_executor.models.utils import (
66 AutoWeightsLoader,
67 PPMissingLayer,
68 WeightsMapper,
69 extract_layer_index,
70 is_pp_missing_parameter,
71 make_layers,
72 maybe_prefix,
73 )
74 from vllm.models.deepseek_v4.amd.rocm import DeepseekV4ROCMAiterMLAAttention
75 from vllm.platforms import current_platform
76 from vllm.platforms.rocm import on_gfx950
77 from vllm.sequence import IntermediateTensors
78
79 from ..common.mm_preprocess import IMAGE_SENTINEL_BASE_ID
80
81 logger = init_logger(__name__)
82
83
84 class DeepseekV4MLP(nn.Module):
85 def __init__(
86 self,
87 hidden_size: int,
88 intermediate_size: int,
89 hidden_act: str,
90 swiglu_limit: float | None = None,
91 quant_config: QuantizationConfig | None = None,
92 reduce_results: bool = True,
93 is_sequence_parallel: bool = False,
94 prefix: str = "",
95 ) -> None:
96 super().__init__()
97
98 # If is_sequence_parallel, the input and output tensors are sharded
99 # across the ranks within the tp_group. In this case the weights are
100 # replicated and no collective ops are needed.
101 # Otherwise we use standard TP with an allreduce at the end.
102 self.gate_up_proj = MergedColumnParallelLinear(
103 hidden_size,
104 [intermediate_size] * 2,
105 bias=False,
106 quant_config=quant_config,
107 disable_tp=is_sequence_parallel,
108 prefix=f"{prefix}.gate_up_proj",
109 )
110 self.down_proj = RowParallelLinear(
111 intermediate_size,
112 hidden_size,
113 bias=False,
114 quant_config=quant_config,
115 reduce_results=reduce_results,
116 disable_tp=is_sequence_parallel,
117 prefix=f"{prefix}.down_proj",
118 )
119 if hidden_act != "silu":
120 raise ValueError(
121 f"Unsupported activation: {hidden_act}. Only silu is supported for now."
122 )
123 if swiglu_limit is not None:
124 self.act_fn = SiluAndMulWithClamp(swiglu_limit)
125 else:
126 self.act_fn = SiluAndMul()
127
128 # gate_up_proj B-preshuffle (ColumnParallel -> no all-reduce); set at load.
129 self._gateup = rocm_aiter_ops.is_enabled()
130 # Block scale for the preshuffled gate_up weight; None = not preshuffled.
131 self._gateup_scale: torch.Tensor | None = None
132
133 def prepare_gateup_preshuffle(self) -> None:
134 # B-preshuffle the gate_up_proj weight in place (single weight).
135 if not self._gateup:
136 return
137 from vllm.model_executor.layers.quantization.utils.fp8_utils import (
138 _upcast_e8m0_to_fp32,
139 get_fp8_block_weight_scale,
140 )
141 from vllm.model_executor.utils import replace_parameter
142
143 w = getattr(self.gate_up_proj, "weight", None)
144 ws = get_fp8_block_weight_scale(self.gate_up_proj)
145 if w is None or ws is None or w.dim() != 2:
146 return
147 # K % 128 (group-128 quant) and N % 16 (shuffle_weight) must hold.
148 if w.shape[-1] % 128 != 0 or w.shape[0] % 16 != 0:
149 return
150 if ws.dtype == torch.float8_e8m0fnu:
151 ws = _upcast_e8m0_to_fp32(ws).contiguous()
152 replace_parameter(
153 self.gate_up_proj,
154 "weight",
155 rocm_aiter_ops.shuffle_weight(w.data, layout=(16, 16)),
156 )
157 self._gateup_scale = ws
158
159 def forward(self, x):
160 if self._gateup_scale is not None and x.dim() == 2:
161 # gate_up via fp8 group-quant (col-major) + B-preshuffle GEMM.
162 x_fp8, x_scale = rocm_aiter_ops.group_fp8_quant(x, transpose_scale=True)
163 gate_up = rocm_aiter_ops.gemm_a8w8_blockscale_bpreshuffle(
164 x_fp8,
165 self.gate_up_proj.weight,
166 x_scale,
167 self._gateup_scale,
168 output_dtype=x.dtype,
169 )
170 else:
171 gate_up, _ = self.gate_up_proj(x)
172 x = self.act_fn(gate_up)
173 x, _ = self.down_proj(x)
174 return x
175
176
177 def _use_heterogeneous_fhmoe(num_tokens: int) -> bool:
178 return num_tokens > 0 and (
179 rocm_aiter_ops.fused_moe_supports_heterogeneous_shared_expert(num_tokens)
180 is True
181 )
182
183
184 def _validate_heterogeneous_routes(
185 topk_weights: torch.Tensor,
186 topk_ids: torch.Tensor,
187 experts_per_token: int,
188 ) -> None:
189 if (
190 topk_weights.ndim != 2
191 or topk_ids.ndim != 2
192 or topk_weights.shape != topk_ids.shape
193 ):
194 raise ValueError(
195 "Heterogeneous fused MoE routes must have equal two-dimensional shapes."
196 )
197 expected_columns = experts_per_token + 1
198 if topk_weights.shape[1] != expected_columns:
199 raise ValueError(
200 "Heterogeneous fused MoE routes must have exactly "
201 f"{expected_columns} columns, got {topk_weights.shape[1]}."
202 )
203
204
205 def _heterogeneous_shared_expert_enabled(vllm_config: VllmConfig) -> bool:
206 config = vllm_config.model_config.hf_config
207 quant_config = vllm_config.quant_config
208 parallel_config = vllm_config.parallel_config
209 reasons: list[str] = []
210
211 if not rocm_aiter_ops.is_fusion_moe_shared_experts_enabled():
212 return False
213 if not on_gfx950():
214 reasons.append("the device is not gfx950")
215 if parallel_config.enable_expert_parallel:
216 reasons.append("expert parallelism is enabled")
217 if parallel_config.enable_eplb:
218 reasons.append("EPLB is enabled")
219 if parallel_config.tensor_parallel_size != 8:
220 reasons.append("tensor parallelism is not 8")
221 if parallel_config.data_parallel_size != 1:
222 reasons.append("data parallelism is enabled")
223 if parallel_config.prefill_context_parallel_size != 1:
224 reasons.append("prefill context parallelism is enabled")
225 if vllm_config.kernel_config.moe_backend != "aiter":
226 reasons.append("the MoE backend is not AITER")
227 if vllm_config.model_config.dtype != torch.bfloat16:
228 reasons.append("the model dtype is not BF16")
229 if not rocm_aiter_ops.fused_moe_supports_heterogeneous_shared_expert(1):
230 reasons.append(
231 "AITER lacks CSV-backed DSV4 I384 heterogeneous shared-expert support"
232 )
233
234 expected_model = {
235 "n_routed_experts": 384,
236 "num_experts_per_tok": 6,
237 "n_shared_experts": 1,
238 "hidden_size": 7168,
239 "moe_intermediate_size": 3072,
240 "hidden_act": "silu",
241 "expert_dtype": "fp4",
242 "topk_method": "noaux_tc",
243 }
244 for name, expected in expected_model.items():
245 if getattr(config, name, None) != expected:
246 reasons.append(f"{name} is not {expected!r}")
247
248 if quant_config is None or quant_config.get_name() != "deepseek_v4_fp8":
249 reasons.append("the DeepSeek V4 FP8 quantization config is not active")
250 else:
251 if getattr(quant_config, "moe_quant_algo", "").upper() == "NVFP4":
252 reasons.append("routed experts use NVFP4")
253 if getattr(quant_config, "weight_block_size", None) != [128, 128]:
254 reasons.append("shared experts are not block-128 FP8")
255 if not getattr(quant_config, "is_checkpoint_fp8_serialized", False):
256 reasons.append("shared experts are not serialized as FP8")
257 if not getattr(quant_config, "is_scale_e8m0", False):
258 reasons.append("shared-expert scales are not E8M0")
259 if getattr(quant_config, "ignored_layers", None):
260 reasons.append("the quantization config has ignored layers")
261
262 offload_config = getattr(vllm_config, "offload_config", None)
263 if offload_config is not None and (
264 getattr(getattr(offload_config, "uva", None), "cpu_offload_gb", 0) > 0
265 or getattr(
266 getattr(offload_config, "prefetch", None),
267 "offload_group_size",
268 0,
269 )
270 > 0
271 ):
272 reasons.append("weight offloading is enabled")
273
274 if reasons:
275 logger.debug_once(
276 "DeepSeek V4 heterogeneous shared-expert fusion is unavailable: %s. "
277 "Continuing with the existing shared-expert path.",
278 "; ".join(reasons),
279 )
280 return False
281
282 logger.info_once(
283 "Using AITER DeepSeek V4 heterogeneous MoE for CSV-covered MoE input "
284 "rows (M) per invocation (native intermediate width 384)."
285 )
286 return True
287
288
289 @torch.no_grad()
290 def _prepare_native_fp8_shared_expert(
291 w13: torch.Tensor,
292 w2: torch.Tensor,
293 w13_scale: torch.Tensor,
294 w2_scale: torch.Tensor,
295 intermediate_size: int,
296 ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
297 """Expand native block-128 E8M0 scales to FHMoE's 1x32 layout."""
298 if w13.dtype != torch.float8_e4m3fn or w2.dtype != torch.float8_e4m3fn:
299 raise ValueError("Heterogeneous shared-expert weights must be FP8 E4M3.")
300 hidden_size = w13.shape[1]
301 if w13.shape != (2 * intermediate_size, hidden_size):
302 raise ValueError(f"Unexpected shared W13 shape {tuple(w13.shape)}.")
303 if w2.shape != (hidden_size, intermediate_size):
304 raise ValueError(f"Unexpected shared W2 shape {tuple(w2.shape)}.")
305
306 def scale_bytes(scale: torch.Tensor) -> torch.Tensor:
307 if scale.dtype == torch.float8_e8m0fnu:
308 return scale.contiguous().view(torch.uint8)
309 if scale.dtype == torch.uint8:
310 return scale.contiguous()
311 if scale.dtype == torch.float32:
312 bits = scale.contiguous().view(torch.int32)
313 exponent = (bits >> 23).to(torch.uint8)
314 if not torch.equal(bits, exponent.to(torch.int32) << 23):
315 raise ValueError("FP32 shared-expert scales are not exact E8M0.")
316 return exponent
317 raise ValueError(f"Unsupported shared-expert scale dtype {scale.dtype}.")
318
319 w13_scale = scale_bytes(w13_scale)
320 w2_scale = scale_bytes(w2_scale)
321 expected_w13_scale = (2 * intermediate_size // 128, hidden_size // 128)
322 expected_w2_scale = (hidden_size // 128, intermediate_size // 128)
323 if tuple(w13_scale.shape) != expected_w13_scale:
324 raise ValueError(
325 f"Expected shared W13 scale shape {expected_w13_scale}, got "
326 f"{tuple(w13_scale.shape)}."
327 )
328 if tuple(w2_scale.shape) != expected_w2_scale:
329 raise ValueError(
330 f"Expected shared W2 scale shape {expected_w2_scale}, got "
331 f"{tuple(w2_scale.shape)}."
332 )
333
334 w13_scale = w13_scale.repeat_interleave(128, dim=0).repeat_interleave(4, dim=1)
335 w2_scale = w2_scale.repeat_interleave(128, dim=0).repeat_interleave(4, dim=1)
336
337 scale_cols = ((w2_scale.shape[1] + 7) // 8) * 8
338 if scale_cols != w2_scale.shape[1]:
339 padded_w2_scale = torch.full(
340 (w2_scale.shape[0], scale_cols),
341 0x7F,
342 dtype=torch.uint8,
343 device=w2_scale.device,
344 )
345 padded_w2_scale[:, : w2_scale.shape[1]].copy_(w2_scale)
346 w2_scale = padded_w2_scale
347
348 return (
349 w13.unsqueeze(0),
350 w2.unsqueeze(0),
351 w13_scale.view(torch.float8_e8m0fnu),
352 w2_scale.view(torch.float8_e8m0fnu),
353 )
354
355
356 class DeepseekV4HeterogeneousMxfp4MoEMethod(Mxfp4MoEMethod):
357 def process_weights_after_loading(self, layer: RoutedExperts) -> None:
358 super().process_weights_after_loading(layer)
359 assert isinstance(layer, DeepseekV4HeterogeneousSharedRoutedExperts)
360 layer.prepare_heterogeneous_shared_expert()
361
362
363 class DeepseekV4HeterogeneousSharedRoutedExperts(RoutedExperts):
364 def __init__(
365 self,
366 *args,
367 shared_expert: DeepseekV4MLP,
368 shared_expert_id: int,
369 **kwargs,
370 ):
371 self._shared_expert_ref = weakref.ref(shared_expert)
372 self.shared_expert_id = shared_expert_id
373 self._routed_quant_config: FusedMoEQuantConfig | None = None
374 super().__init__(*args, **kwargs)
375 if self.expert_map_manager.num_fused_shared_experts != 1:
376 raise ValueError("Heterogeneous fusion requires one appended expert.")
377 if self.shared_expert_id != self.global_num_experts:
378 raise ValueError("The shared expert ID must name the appended slot.")
379 self.register_buffer("shared_w1", None, persistent=False)
380 self.register_buffer("shared_w2", None, persistent=False)
381 self.register_buffer("shared_w1_scale", None, persistent=False)
382 self.register_buffer("shared_w2_scale", None, persistent=False)
383
384 def _get_quant_method(self, prefix, quant_config, moe_config):
385 quant_method = super()._get_quant_method(prefix, quant_config, moe_config)
386 if not isinstance(quant_method, Mxfp4MoEMethod):
387 raise ValueError("Heterogeneous fusion requires MXFP4 routed experts.")
388 return DeepseekV4HeterogeneousMxfp4MoEMethod(moe_config)
389
390 @torch.no_grad()
391 def prepare_heterogeneous_shared_expert(self) -> None:
392 shared_expert = self._shared_expert_ref()
393 if shared_expert is None:
394 raise RuntimeError("The native shared-expert module was released.")
395
396 shared = _prepare_native_fp8_shared_expert(
397 shared_expert.gate_up_proj.weight,
398 shared_expert.down_proj.weight,
399 shared_expert.gate_up_proj.weight_scale_inv,
400 shared_expert.down_proj.weight_scale_inv,
401 self.moe_config.intermediate_size_per_partition,
402 )
403 self.shared_w1 = rocm_aiter_ops.shuffle_weight_a16w4(shared[0], 16, True)
404 self.shared_w2 = rocm_aiter_ops.shuffle_weight_a16w4(shared[1], 16, False)
405 self.shared_w1_scale = rocm_aiter_ops.shuffle_scale_a16w4(shared[2], 1, True)
406 self.shared_w2_scale = rocm_aiter_ops.shuffle_scale(shared[3])
407
408 self._ensure_moe_quant_config_init()
409 quant_config = self.quant_method.moe_quant_config
410 if quant_config is None:
411 raise RuntimeError("The routed MXFP4 quantization config is unavailable.")
412 if quant_config.w1_scale is None or quant_config.w2_scale is None:
413 raise RuntimeError("The routed MXFP4 scales are unavailable.")
414
415 def routed_scales(scale: torch.Tensor) -> torch.Tensor:
416 scale = scale.reshape(self.local_num_experts, -1, scale.shape[-1])[
417 : self.shared_expert_id
418 ]
419 return scale.reshape(-1, scale.shape[-1]).contiguous()
420
421 self._routed_quant_config = replace(
422 quant_config,
423 _w1=replace(
424 quant_config._w1,
425 scale=routed_scales(quant_config.w1_scale),
426 ),
427 _w2=replace(
428 quant_config._w2,
429 scale=routed_scales(quant_config.w2_scale),
430 ),
431 )
432
433 def forward_modular(
434 self,
435 x: torch.Tensor,
436 topk_weights: torch.Tensor,
437 topk_ids: torch.Tensor,
438 shared_experts=None,
439 shared_experts_input: torch.Tensor | None = None,
440 ) -> torch.Tensor:
441 if shared_experts is not None or shared_experts_input is not None:
442 raise ValueError("The heterogeneous expert owns the shared path.")
443 _validate_heterogeneous_routes(
444 topk_weights,
445 topk_ids,
446 self.moe_config.experts_per_token,
447 )
448 if any(
449 tensor is None
450 for tensor in (
451 self.shared_w1,
452 self.shared_w2,
453 self.shared_w1_scale,
454 self.shared_w2_scale,
455 )
456 ):
457 raise RuntimeError("Heterogeneous shared weights were not prepared.")
458
459 if _use_heterogeneous_fhmoe(x.shape[0]):
460 self._ensure_moe_quant_config_init()
461 quant_config = self.quant_method.moe_quant_config
462 if quant_config is None:
463 raise RuntimeError("The routed MXFP4 config is unavailable.")
464 return rocm_aiter_fused_experts(
465 hidden_states=x,
466 w1=self.w13_weight,
467 w2=self.w2_weight,
468 topk_weights=topk_weights,
469 topk_ids=topk_ids,
470 moe_config=self.moe_config,
471 activation=self.activation,
472 expert_mask=self.expert_map,
473 quant_config=quant_config,
474 output_dtype=x.dtype,
475 moe_sorting_dispatch_policy=(rocm_aiter_ops.get_moe_dispatch_policy()),
476 shared_w1=self.shared_w1,
477 shared_w2=self.shared_w2,
478 shared_w1_scale=self.shared_w1_scale,
479 shared_w2_scale=self.shared_w2_scale,
480 shared_expert_id=self.shared_expert_id,
481 )
482
483 shared_expert = self._shared_expert_ref()
484 if shared_expert is None or self._routed_quant_config is None:
485 raise RuntimeError("The separate shared-expert fallback is unavailable.")
486 shared_out = shared_expert(x)
487 routed_w1 = self.w13_weight[: self.shared_expert_id]
488 routed_w2 = self.w2_weight[: self.shared_expert_id]
489 routed_w1.is_shuffled = True
490 routed_w2.is_shuffled = True
491 routed = rocm_aiter_fused_experts(
492 hidden_states=x,
493 w1=routed_w1,
494 w2=routed_w2,
495 topk_weights=topk_weights[:, :-1].contiguous(),
496 topk_ids=topk_ids[:, :-1].contiguous(),
497 moe_config=self.moe_config,
498 activation=self.activation,
499 expert_mask=self.expert_map,
500 quant_config=self._routed_quant_config,
501 output_dtype=x.dtype,
502 moe_sorting_dispatch_policy=rocm_aiter_ops.get_moe_dispatch_policy(),
503 )
504 return routed + shared_out
505
506
507 def _fuse_shared_experts_enabled(config, parallel_config: ParallelConfig) -> bool:
508 if (
509 getattr(config, "n_shared_experts", None)
510 and envs.VLLM_ROCM_USE_AITER_FUSION_SHARED_EXPERTS
511 and not parallel_config.enable_expert_parallel
512 and parallel_config.data_parallel_size > 1
513 ):
514 # Fused shared experts are not supported under data parallelism:
515 # the fused path cannot load the FP8 shared expert into the MXFP4
516 # routed slot, which fails obscurely during weight loading. Fail
517 # fast with an actionable message instead.
518 raise ValueError(
519 "DeepSeek-V4 fused shared experts "
520 "(VLLM_ROCM_USE_AITER_FUSION_SHARED_EXPERTS=1) are not supported "
521 "with data parallelism (data_parallel_size > 1). Set "
522 "VLLM_ROCM_USE_AITER_FUSION_SHARED_EXPERTS=0 to run the shared "
523 "expert as an unfused FP8 MLP."
524 )
525 return bool(
526 getattr(config, "n_shared_experts", None)
527 and envs.VLLM_ROCM_USE_AITER_FUSION_SHARED_EXPERTS
528 and not parallel_config.enable_expert_parallel
529 )
530
531
532 class DeepseekV4MoE(nn.Module):
533 def __init__(
534 self,
535 vllm_config: VllmConfig,
536 prefix: str = "",
537 fuse_heterogeneous_shared_expert: bool = False,
538 ):
539 super().__init__()
540
541 self.tp_size = get_tensor_model_parallel_world_size()
542 config = vllm_config.model_config.hf_config
543 quant_config = vllm_config.quant_config
544 self.prefix = prefix
545
546 self.routed_scaling_factor = getattr(config, "routed_scaling_factor", 1.0)
547 self.hidden_size = config.hidden_size
548
549 self.n_routed_experts = config.n_routed_experts
550 self.n_activated_experts = config.num_experts_per_tok
551 self.moe_intermediate_size = config.moe_intermediate_size
552 self.swiglu_limit = config.swiglu_limit
553 self.renormalize = config.norm_topk_prob
554 self.scoring_func = getattr(config, "scoring_func", "sqrtsoftplus")
555
556 self.gate = GateLinear(
557 input_size=config.hidden_size,
558 output_size=config.n_routed_experts,
559 bias=False,
560 out_dtype=torch.float32,
561 prefix=f"{prefix}.gate",
562 )
563
564 self.gate.e_score_correction_bias = None
565 self.gate.tid2eid = None
566 self.gate.bias_vl = None
567 self.image_sentinel_lo = (
568 IMAGE_SENTINEL_BASE_ID if getattr(config, "vision_n_layers", 0) > 0 else 0
569 )
570 is_hash_moe = extract_layer_index(prefix) < config.num_hash_layers
571 self.hash_indices_dtype = torch.int32
572 if is_hash_moe:
573 # hash MoE doesn't use e_score_correction_bias
574 # Use randint instead of empty to avoid garbage values causing
575 # invalid memory access in dummy mode (--load-format="dummy")
576 self.gate.tid2eid = nn.Parameter(
577 torch.randint(
578 0,
579 config.n_routed_experts,
580 (config.vocab_size, config.num_experts_per_tok),
581 dtype=self.hash_indices_dtype,
582 ),
583 requires_grad=False,
584 )
585 if getattr(config, "topk_method", None) == "noaux_tc" and (
586 not is_hash_moe or getattr(config, "vision_n_layers", 0) > 0
587 ):
588 self.gate.e_score_correction_bias = nn.Parameter(
589 torch.empty(config.n_routed_experts, dtype=torch.float32),
590 requires_grad=False,
591 )
592
593 if getattr(config, "vision_n_layers", 0) > 0:
594 self.gate.bias_vl = nn.Parameter(
595 torch.empty(config.n_routed_experts, dtype=torch.float32),
596 requires_grad=False,
597 )
598
599 self.n_shared_experts = config.n_shared_experts
600
601 # TODO: Historically, only `VLLM_ROCM_USE_AITER_FUSION_SHARED_EXPERTS=1`
602 # is checked to enable FSE for DeepSeek-v4, despite AITER not being used.
603 # This should be cleaned up and use `resolve_layer_fused_shared_expert`.
604 self.fuse_heterogeneous_shared_expert = fuse_heterogeneous_shared_expert
605 fse_requested = (
606 _fuse_shared_experts_enabled(config, vllm_config.parallel_config)
607 and not self.fuse_heterogeneous_shared_expert
608 )
609 fse_compatible = False
610 if fse_requested:
611 fse_compatible, fse_reason = is_shared_expert_quant_fse_compatible(
612 quant_config,
613 f"{prefix}.experts",
614 f"{prefix}.shared_experts",
615 )
616 if not fse_compatible:
617 logger.warning(
618 "VLLM_ROCM_USE_AITER_FUSION_SHARED_EXPERTS is enabled but "
619 "cannot be enabled: %s.",
620 fse_reason,
621 )
622 self.is_fused_shared_expert_enabled = fse_requested and fse_compatible
623
624 if config.n_shared_experts is None or self.is_fused_shared_expert_enabled:
625 self.shared_experts = None
626 else:
627 intermediate_size = config.moe_intermediate_size * config.n_shared_experts
628
629 self.shared_experts = DeepseekV4MLP(
630 hidden_size=config.hidden_size,
631 intermediate_size=intermediate_size,
632 hidden_act=config.hidden_act,
633 swiglu_limit=self.swiglu_limit,
634 quant_config=quant_config,
635 reduce_results=False,
636 prefix=f"{prefix}.shared_experts",
637 )
638
639 self.tp_rank = get_tensor_model_parallel_rank()
640 assert config.n_routed_experts % self.tp_size == 0
641
642 self.n_local_experts = config.n_routed_experts // self.tp_size
643 self.experts_start_idx = self.tp_rank * self.n_local_experts
644 self.experts_end_idx = self.experts_start_idx + self.n_local_experts
645
646 fuse_shared_into_routed = (
647 self.is_fused_shared_expert_enabled or self.fuse_heterogeneous_shared_expert
648 )
649 self.experts = FusedMoEFactory(
650 shared_experts=(None if fuse_shared_into_routed else self.shared_experts),
651 n_shared_experts=(
652 config.n_shared_experts if fuse_shared_into_routed else None
653 ),
654 fuse_shared_experts=fuse_shared_into_routed,
655 gate=self.gate,
656 num_experts=config.n_routed_experts,
657 top_k=config.num_experts_per_tok,
658 hidden_size=config.hidden_size,
659 intermediate_size=config.moe_intermediate_size,
660 renormalize=config.norm_topk_prob,
661 quant_config=quant_config,
662 prefix=f"{prefix}.experts",
663 scoring_func=self.scoring_func,
664 routed_scaling_factor=self.routed_scaling_factor,
665 e_score_correction_bias=self.gate.e_score_correction_bias,
666 hash_indices_table=self.gate.tid2eid,
667 bias_vl=self.gate.bias_vl,
668 image_sentinel_lo=self.image_sentinel_lo,
669 swiglu_limit=self.swiglu_limit,
670 router_logits_dtype=torch.float32,
671 routed_experts_cls=(
672 DeepseekV4HeterogeneousSharedRoutedExperts
673 if self.fuse_heterogeneous_shared_expert
674 else None
675 ),
676 routed_experts_args=(
677 {
678 "shared_expert": self.shared_experts,
679 "shared_expert_id": config.n_routed_experts,
680 }
681 if self.fuse_heterogeneous_shared_expert
682 else None
683 ),
684 )
685
686 def forward(
687 self, hidden_states: torch.Tensor, input_ids: torch.Tensor | None = None
688 ) -> torch.Tensor:
689 if self.gate.tid2eid is not None and input_ids is None:
690 raise ValueError("DeepSeek V4 hash MoE routing requires input_ids.")
691 if self.gate.bias_vl is not None and input_ids is None:
692 raise ValueError("DeepSeek V4 vision MoE routing requires input_ids.")
693
694 org_shape = hidden_states.shape
695 final_hidden_states = self.experts(
696 hidden_states=hidden_states,
697 router_logits=hidden_states,
698 input_ids=input_ids,
699 )
700
701 return final_hidden_states.view(org_shape)
702
703
704 # Hidden sizes supported by AITER mhc_pre_big_fuse_rmsnorm.
705 _AITER_MHC_FUSED_RMSNORM_SIZES = frozenset({1280, 2560, 4096, 7168})
706
707
708 class DeepseekV4DecoderLayer(nn.Module):
709 def __init__(
710 self,
711 vllm_config,
712 prefix,
713 topk_indices_buffer: torch.Tensor | None = None,
714 aux_stream_list: list[torch.cuda.Stream] | None = None,
715 fuse_heterogeneous_shared_expert: bool = False,
716 ):
717 super().__init__()
718
719 # Lazy import to avoid top-level tilelang dependency.
720 # Registers both torch.ops.vllm.mhc_pre and mhc_post
721 import vllm.model_executor.layers.mhc # noqa: F401
722
723 config = vllm_config.model_config.hf_config
724 self.hidden_size = config.hidden_size
725
726 self.rms_norm_eps = config.rms_norm_eps
727 self.attn = DeepseekV4ROCMAiterMLAAttention(
728 vllm_config,
729 prefix=f"{prefix}.attn",
730 topk_indices_buffer=topk_indices_buffer,
731 aux_stream_list=aux_stream_list,
732 )
733 self.ffn = DeepseekV4MoE(
734 vllm_config,
735 prefix=f"{prefix}.ffn",
736 fuse_heterogeneous_shared_expert=fuse_heterogeneous_shared_expert,
737 )
738
739 self.attn_norm = RMSNorm(self.hidden_size, self.rms_norm_eps)
740 self.ffn_norm = RMSNorm(self.hidden_size, self.rms_norm_eps)
741 self.hc_mult = config.hc_mult
742 self.hc_sinkhorn_iters = config.hc_sinkhorn_iters
743 self.hc_eps = config.hc_eps
744 self.hc_post_alpha = 2.0
745 mix_hc = (2 + self.hc_mult) * self.hc_mult
746 hc_dim = self.hc_mult * self.hidden_size
747 self.hc_attn_fn = nn.Parameter(
748 torch.empty(
749 (mix_hc, hc_dim),
750 dtype=torch.float32,
751 ),
752 requires_grad=False,
753 )
754 self.hc_ffn_fn = nn.Parameter(
755 torch.empty(
756 (mix_hc, hc_dim),
757 dtype=torch.float32,
758 ),
759 requires_grad=False,
760 )
761 self.hc_attn_base = nn.Parameter(
762 torch.empty(
763 mix_hc,
764 dtype=torch.float32,
765 ),
766 requires_grad=False,
767 )
768 self.hc_ffn_base = nn.Parameter(
769 torch.empty(
770 mix_hc,
771 dtype=torch.float32,
772 ),
773 requires_grad=False,
774 )
775 self.hc_attn_scale = nn.Parameter(
776 torch.empty(
777 3,
778 dtype=torch.float32,
779 ),
780 requires_grad=False,
781 )
782 self.hc_ffn_scale = nn.Parameter(
783 torch.empty(
784 3,
785 dtype=torch.float32,
786 ),
787 requires_grad=False,
788 )
789 self.mhc_pre = MHCPreOp()
790 self.mhc_post = MHCPostOp()
791 self.mhc_fused_post_pre = MHCFusedPostPreOp()
792 # AITER mhc kernels (pre/post/fused) require hc_mult == 4.
793 use_aiter_mhc = (
794 HAS_AITER_MHC and self.hidden_size % 256 == 0 and self.hc_mult == 4
795 )
796 # Prefer AITER fused post+pre when eligible; otherwise TileLang.
797 self.use_fused_mhc = use_aiter_mhc or HAS_TILELANG_MHC
798 # Fold attn/ffn RMSNorm into MHC only when the active backend's
799 # fused-rmsnorm path supports this hidden size.
800 if use_aiter_mhc:
801 self.fuse_mhc_rmsnorm = self.hidden_size in _AITER_MHC_FUSED_RMSNORM_SIZES
802 else:
803 self.fuse_mhc_rmsnorm = HAS_TILELANG_MHC and self.use_fused_mhc
804
805 def hc_pre(
806 self,
807 x: torch.Tensor,
808 hc_fn: torch.Tensor,
809 hc_scale: torch.Tensor,
810 hc_base: torch.Tensor,
811 norm_weight: torch.Tensor | None = None,
812 norm_eps: float = 0.0,
813 ):
814 """Reduce HC residual streams to the next sub-layer input.
815
816 When ``norm_weight`` is set, RMSNorm is fused into the pre kernel.
817 """
818 post_mix, res_mix, layer_input = self.mhc_pre(
819 residual=x,
820 fn=hc_fn,
821 hc_scale=hc_scale,
822 hc_base=hc_base,
823 rms_eps=self.rms_norm_eps,
824 hc_pre_eps=self.hc_eps,
825 hc_sinkhorn_eps=self.hc_eps,
826 hc_post_mult_value=self.hc_post_alpha,
827 sinkhorn_repeat=self.hc_sinkhorn_iters,
828 norm_weight=norm_weight,
829 norm_eps=norm_eps,
830 )
831 return layer_input, post_mix, res_mix
832
833 def hc_post(
834 self,
835 x: torch.Tensor,
836 residual: torch.Tensor,
837 post: torch.Tensor,
838 comb: torch.Tensor,
839 ):
840 return self.mhc_post(x, residual, post, comb)
841
842 def _forward_fused_post_pre(
843 self,
844 x: torch.Tensor,
845 positions: torch.Tensor,
846 input_ids: torch.Tensor | None,
847 post_mix: torch.Tensor | None = None,
848 res_mix: torch.Tensor | None = None,
849 residual: torch.Tensor | None = None,
850 ) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor, torch.Tensor]:
851 attn_norm_weight = self.attn_norm.weight if self.fuse_mhc_rmsnorm else None
852 attn_norm_eps = (
853 self.attn_norm.variance_epsilon if self.fuse_mhc_rmsnorm else 0.0
854 )
855 if residual is None:
856 # Run standalone hc_pre on first layer
857 residual = x
858 x, post_mix, res_mix = self.hc_pre(
859 x,
860 self.hc_attn_fn,
861 self.hc_attn_scale,
862 self.hc_attn_base,
863 norm_weight=attn_norm_weight,
864 norm_eps=attn_norm_eps,
865 )
866 else:
867 residual, post_mix, res_mix, x = self.mhc_fused_post_pre(
868 x,
869 residual,
870 post_mix,
871 res_mix,
872 self.hc_attn_fn,
873 self.hc_attn_scale,
874 self.hc_attn_base,
875 self.rms_norm_eps,
876 self.hc_eps,
877 self.hc_eps,
878 self.hc_post_alpha,
879 self.hc_sinkhorn_iters,
880 norm_weight=attn_norm_weight,
881 norm_eps=attn_norm_eps,
882 )
883
884 if not self.fuse_mhc_rmsnorm:
885 x = self.attn_norm(x)
886 x = self.attn(positions, x, None)
887
888 ffn_norm_weight = self.ffn_norm.weight if self.fuse_mhc_rmsnorm else None
889 ffn_norm_eps = self.ffn_norm.variance_epsilon if self.fuse_mhc_rmsnorm else 0.0
890 residual, post_mix, res_mix, x = self.mhc_fused_post_pre(
891 x,
892 residual,
893 post_mix,
894 res_mix,
895 self.hc_ffn_fn,
896 self.hc_ffn_scale,
897 self.hc_ffn_base,
898 self.rms_norm_eps,
899 self.hc_eps,
900 self.hc_eps,
901 self.hc_post_alpha,
902 self.hc_sinkhorn_iters,
903 norm_weight=ffn_norm_weight,
904 norm_eps=ffn_norm_eps,
905 )
906 if not self.fuse_mhc_rmsnorm:
907 x = self.ffn_norm(x)
908 x = self.ffn(x, input_ids)
909 return x, residual, post_mix, res_mix
910
911 def _forward_unfused_post_pre(
912 self,
913 x: torch.Tensor,
914 positions: torch.Tensor,
915 input_ids: torch.Tensor | None,
916 post_mix: torch.Tensor | None = None,
917 res_mix: torch.Tensor | None = None,
918 residual: torch.Tensor | None = None,
919 ) -> tuple[
920 torch.Tensor, torch.Tensor | None, torch.Tensor | None, torch.Tensor | None
921 ]:
922 residual = x
923 x, post, comb = self.hc_pre(
924 x, self.hc_attn_fn, self.hc_attn_scale, self.hc_attn_base
925 )
926 x = self.attn_norm(x)
927 x = self.attn(positions, x, None)
928 x = self.hc_post(x, residual, post, comb)
929
930 residual = x
931 x, post, comb = self.hc_pre(
932 x, self.hc_ffn_fn, self.hc_ffn_scale, self.hc_ffn_base
933 )
934 x = self.ffn_norm(x)
935 x = self.ffn(x, input_ids)
936 x = self.hc_post(x, residual, post, comb)
937 return x, None, None, None
938
939 def forward(
940 self,
941 x: torch.Tensor,
942 positions: torch.Tensor,
943 input_ids: torch.Tensor | None,
944 post_mix: torch.Tensor | None = None,
945 res_mix: torch.Tensor | None = None,
946 residual: torch.Tensor | None = None,
947 ) -> tuple[
948 torch.Tensor, torch.Tensor | None, torch.Tensor | None, torch.Tensor | None
949 ]:
950 if not self.use_fused_mhc:
951 return self._forward_unfused_post_pre(
952 x, positions, input_ids, post_mix, res_mix, residual
953 )
954 return self._forward_fused_post_pre(
955 x, positions, input_ids, post_mix, res_mix, residual
956 )
957
958
959 class DeepseekV4Model(nn.Module, EagleModelMixin):
960 def __init__(self, *, vllm_config: VllmConfig, prefix: str = ""):
961 super().__init__()
962
963 config = vllm_config.model_config.hf_config
964 quant_config = vllm_config.quant_config
965 self.config = config
966 self.vocab_size = config.vocab_size
967 self.hc_eps = config.hc_eps
968 self.hc_mult = config.hc_mult
969 self.hc_dim = self.hc_mult * config.hidden_size
970 self.rms_norm_eps = config.rms_norm_eps
971
972 aux_stream_list = [torch.cuda.Stream() for _ in range(3)]
973
974 self.device = current_platform.device_type
975 # Reserved topk indices buffer for all Indexer layers to reuse.
976 self.topk_indices_buffer = torch.empty(
977 vllm_config.scheduler_config.max_num_batched_tokens,
978 config.index_topk,
979 dtype=torch.int32,
980 device=self.device,
981 )
982
983 if get_pp_group().is_first_rank:
984 self.embed_tokens = VocabParallelEmbedding(
985 config.vocab_size,
986 config.hidden_size,
987 quant_config=quant_config,
988 prefix=f"{prefix}.embed_tokens",
989 )
990 else:
991 self.embed_tokens = PPMissingLayer()
992
993 self.fuse_heterogeneous_shared_expert = _heterogeneous_shared_expert_enabled(
994 vllm_config
995 )
996 self.start_layer, self.end_layer, self.layers = make_layers(
997 config.num_hidden_layers,
998 lambda prefix: DeepseekV4DecoderLayer(
999 vllm_config,
1000 prefix=prefix,
1001 topk_indices_buffer=self.topk_indices_buffer,
1002 aux_stream_list=aux_stream_list,
1003 fuse_heterogeneous_shared_expert=(
1004 self.fuse_heterogeneous_shared_expert
1005 ),
1006 ),
1007 prefix=f"{prefix}.layers",
1008 )
1009 self.is_fused_shared_expert_enabled = is_model_fused_shared_expert_compatible(
1010 self.layers,
1011 DeepseekV4MoE,
1012 "ffn",
1013 )
1014
1015 if get_pp_group().is_last_rank:
1016 self.norm = RMSNorm(config.hidden_size, self.rms_norm_eps)
1017 else:
1018 self.norm = PPMissingLayer()
1019
1020 self.hc_head_fn = nn.Parameter(
1021 torch.empty(
1022 self.hc_mult,
1023 self.hc_dim,
1024 dtype=torch.float32,
1025 ),
1026 requires_grad=False,
1027 )
1028 self.hc_head_base = nn.Parameter(
1029 torch.empty(
1030 self.hc_mult,
1031 dtype=torch.float32,
1032 ),
1033 requires_grad=False,
1034 )
1035 self.hc_head_scale = nn.Parameter(
1036 torch.empty(1, dtype=torch.float32),
1037 requires_grad=False,
1038 )
1039 self.hc_head_op = HCHeadOp()
1040 spec_config = vllm_config.speculative_config
1041 needs_mtp_hidden_states = spec_config is not None and (
1042 spec_config.use_eagle() or spec_config.uses_draft_model()
1043 )
1044 if get_pp_group().is_last_rank and needs_mtp_hidden_states:
1045 self._mtp_hidden_buffer = torch.empty(
1046 vllm_config.scheduler_config.max_num_batched_tokens,
1047 self.hc_dim,
1048 dtype=vllm_config.model_config.dtype,
1049 device=self.device,
1050 )
1051 else:
1052 self._mtp_hidden_buffer = None
1053
1054 def embed_input_ids(self, input_ids: torch.Tensor) -> torch.Tensor:
1055 return self.embed_tokens(input_ids)
1056
1057 def make_empty_intermediate_tensors(
1058 self,
1059 batch_size: int,
1060 dtype: torch.dtype,
1061 device: torch.device,
1062 ) -> IntermediateTensors:
1063 # PP intermediate tensors carry the multi-stream hidden_states
1064 # of shape (num_tokens, hc_mult, hidden_size) — V4 expands the
1065 # token embedding to hc_mult streams before the first decoder
1066 # layer and keeps that shape until hc_head() collapses it.
1067 return IntermediateTensors(
1068 {
1069 "hidden_states": torch.zeros(
1070 (batch_size, self.hc_mult, self.config.hidden_size),
1071 dtype=dtype,
1072 device=device,
1073 ),
1074 }
1075 )
1076
1077 def forward(
1078 self,
1079 input_ids: torch.Tensor,
1080 positions: torch.Tensor,
1081 intermediate_tensors: IntermediateTensors | None,
1082 inputs_embeds: torch.Tensor | None = None,
1083 ) -> torch.Tensor | IntermediateTensors:
1084 if get_pp_group().is_first_rank:
1085 if inputs_embeds is not None:
1086 hidden_states = inputs_embeds
1087 else:
1088 hidden_states = self.embed_input_ids(input_ids)
1089 hidden_states = hidden_states.unsqueeze(-2).repeat(1, self.hc_mult, 1)
1090 else:
1091 assert intermediate_tensors is not None
1092 hidden_states = intermediate_tensors["hidden_states"]
1093
1094 residual, post_mix, res_mix = None, None, None
1095 # EAGLE3 / DSpark / DFlash aux hidden states: reconstructed (post-mhc)
1096 # hidden state at the configured target layers, averaged over the
1097 # hc_mult streams to [T, hidden_size]. Empty unless a draft model set
1098 # aux_hidden_state_layers.
1099 aux_hidden_states: list[torch.Tensor] = []
1100 # On the fused path the final layer's hc_post output is reused below
1101 # (avoids computing hc_post twice when the last layer is also an aux
1102 # layer).
1103 final_aux_recon: torch.Tensor | None = None
1104 for idx, layer in enumerate(
1105 islice(self.layers, self.start_layer, self.end_layer),
1106 start=self.start_layer,
1107 ):
1108 hidden_states, residual, post_mix, res_mix = layer(
1109 hidden_states,
1110 positions,
1111 input_ids,
1112 post_mix,
1113 res_mix,
1114 residual,
1115 )
1116 if (idx + 1) in self.aux_hidden_state_layers:
1117 # On the unfused path the layer already applied hc_post,
1118 # so hidden_states is the reconstructed stream; on the fused
1119 # path reconstruct it via hc_post before averaging.
1120 if layer.use_fused_mhc:
1121 aux_recon = layer.hc_post(
1122 hidden_states, residual, post_mix, res_mix
1123 )
1124 final_aux_recon = aux_recon
1125 else:
1126 aux_recon = hidden_states
1127 aux_hidden_states.append(aux_recon.mean(dim=1))
1128 if layer is not None and layer.use_fused_mhc:
1129 # Reuse the last layer's hc_post output if it was already computed
1130 # for the aux hidden state above; otherwise compute it now.
1131 if (
1132 final_aux_recon is not None
1133 and self.end_layer in self.aux_hidden_state_layers
1134 ):
1135 hidden_states = final_aux_recon
1136 else:
1137 hidden_states = layer.hc_post(
1138 hidden_states, residual, post_mix, res_mix
1139 )
1140
1141 if not get_pp_group().is_last_rank:
1142 return IntermediateTensors({"hidden_states": hidden_states})
1143
1144 if self._mtp_hidden_buffer is not None:
1145 num_tokens = hidden_states.shape[0]
1146 self._mtp_hidden_buffer[:num_tokens].copy_(hidden_states.flatten(1))
1147
1148 hidden_states = self.hc_head_op(
1149 hidden_states,
1150 self.hc_head_fn,
1151 self.hc_head_scale,
1152 self.hc_head_base,
1153 self.rms_norm_eps,
1154 self.hc_eps,
1155 )
1156 hidden_states = self.norm(hidden_states)
1157 if len(aux_hidden_states) > 0:
1158 return hidden_states, aux_hidden_states
1159 return hidden_states
1160
1161 def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
1162 stacked_params_mapping = [
1163 # (param_name, shard_name, shard_id)
1164 ("gate_up_proj", "w1", 0),
1165 ("gate_up_proj", "w3", 1),
1166 ("attn.fused_wqa_wkv", "attn.wq_a", 0),
1167 ("attn.fused_wqa_wkv", "attn.wkv", 1),
1168 ("compressor.fused_wkv_wgate", "compressor.wkv", 0),
1169 ("compressor.fused_wkv_wgate", "compressor.wgate", 1),
1170 ]
1171 params_dict = dict(self.named_parameters())
1172 loaded_params: set[str] = set()
1173
1174 def _resolve_param_name(name: str) -> str:
1175 # Fp8LinearMethod registers block scales as ``weight_scale_inv``.
1176 # Quark checkpoints / QuarkW8A8Fp8PerBlock use ``weight_scale``.
1177 # Alias only when the registered param is the ``_inv`` name.
1178 inv_name = f"{name}_inv"
1179 if name not in params_dict and inv_name in params_dict:
1180 return inv_name
1181 return name
1182
1183 # TP for attention
1184 tp_size = get_tensor_model_parallel_world_size()
1185 tp_rank = get_tensor_model_parallel_rank()
1186 n_head = self.config.num_attention_heads
1187 n_local_head = n_head // tp_size
1188 head_rank_start = n_local_head * tp_rank
1189 head_rank_end = n_local_head * (tp_rank + 1)
1190
1191 # Pre-compute expert mapping ONCE.
1192 expert_mapping = self.get_expert_mapping()
1193
1194 # Use each MoE's own per-layer fusion decision (computed with its prefix
1195 # at init) as the single source of truth, so the redirect below cannot
1196 # diverge from how the module was built if per-layer quantization ever
1197 # mixes fused and non-fused layers.
1198 fuse_by_layer = {
1199 extract_layer_index(mod_name): mod.is_fused_shared_expert_enabled
1200 for mod_name, mod in self.named_modules()
1201 if isinstance(mod, DeepseekV4MoE)
1202 }
1203 n_routed = self.config.n_routed_experts
1204 # The redirect below maps the single shared-expert tensor group to one
1205 # appended slot; multiple shared experts would need per-expert slicing
1206 # (see deepseek_v2.py). DeepSeek-V4 has n_shared_experts == 1.
1207 if any(fuse_by_layer.values()) and self.config.n_shared_experts != 1:
1208 raise NotImplementedError(
1209 "deepseek-v4 fused shared-expert loading supports only "
1210 f"n_shared_experts == 1, got {self.config.n_shared_experts}"
1211 )
1212
1213 for name, loaded_weight in weights:
1214 # Shared-expert fusion: redirect ``.ffn.shared_experts.w{1,2,3}``
1215 # into appended routed-expert slot ``.ffn.experts.{n_routed}``
1216 # so the MXFP4-quantized shared expert loads through the routed
1217 # expert loader (grouped GEMM). Single shared expert only.
1218 if ".ffn.shared_experts.w" in name and fuse_by_layer.get(
1219 extract_layer_index(name), False
1220 ):
1221 name = name.replace(
1222 ".ffn.shared_experts.w",
1223 f".ffn.experts.{n_routed}.w",
1224 )
1225
1226 for param_name, weight_name, shard_id in stacked_params_mapping:
1227 # Skip non-stacked layers and experts (experts handled below).
1228 if ".experts." in name:
1229 continue
1230 if weight_name not in name:
1231 continue
1232 name = name.replace(weight_name, param_name)
1233
1234 if is_pp_missing_parameter(name, self):
1235 break
1236 param_name = _resolve_param_name(name)
1237 param = params_dict[param_name]
1238 weight_loader = param.weight_loader
1239 weight_loader(param, loaded_weight, shard_id)
1240 loaded_params.add(param_name)
1241 break
1242 else:
1243 if ".experts." in name:
1244 # E8M0 scales are stored as float8_e8m0fnu in
1245 # checkpoints but the MoE param is uint8. copy_()
1246 # would do a numeric conversion (e.g. 2^-7 → 0),
1247 # destroying the raw exponent bytes.
1248 if (
1249 "weight_scale" in name
1250 and loaded_weight.dtype == torch.float8_e8m0fnu
1251 ):
1252 loaded_weight = loaded_weight.view(torch.uint8)
1253 for mapping in expert_mapping:
1254 param_name, weight_name, expert_id, expert_shard_id = mapping
1255 if weight_name not in name:
1256 continue
1257 name_mapped = name.replace(weight_name, param_name)
1258 if is_pp_missing_parameter(name_mapped, self):
1259 continue
1260 param = params_dict[name_mapped]
1261 # We should ask the weight loader to return success or not
1262 # here since otherwise we may skip experts with other
1263 # available replicas.
1264 weight_loader = typing.cast(
1265 Callable[..., bool], param.weight_loader
1266 )
1267 success = weight_loader(
1268 param,
1269 loaded_weight,
1270 name_mapped,
1271 shard_id=expert_shard_id,
1272 expert_id=expert_id,
1273 return_success=True,
1274 )
1275 if success:
1276 name = name_mapped
1277 break
1278 loaded_params.add(name_mapped)
1279 continue
1280 elif "attn_sink" in name:
1281 if is_pp_missing_parameter(name, self):
1282 continue
1283 narrow_weight = loaded_weight[head_rank_start:head_rank_end]
1284 n = narrow_weight.shape[0]
1285 params_dict[name][:n].copy_(narrow_weight)
1286 loaded_params.add(name)
1287 continue
1288 else:
1289 if is_pp_missing_parameter(name, self):
1290 continue
1291 param_name = _resolve_param_name(name)
1292 param = params_dict[param_name]
1293 weight_loader = getattr(
1294 param, "weight_loader", default_weight_loader
1295 )
1296 weight_loader(param, loaded_weight)
1297 loaded_params.add(param_name)
1298 continue
1299
1300 return loaded_params
1301
1302 def get_expert_mapping(self) -> list[tuple[str, str, int, str]]:
1303 # Params for weights, fp8 weight scales, fp8 activation scales
1304 # (param_name, weight_name, expert_id, shard_id)
1305 # When fusing shared experts, include the appended slots
1306 # (ids n_routed_experts .. n_routed_experts + n_shared - 1) so the
1307 # redirected shared-expert weights route through the expert loader.
1308 n_shared = getattr(self.config, "n_shared_experts", 0) or 0
1309 num_experts = self.config.n_routed_experts + (
1310 n_shared if self.is_fused_shared_expert_enabled else 0
1311 )
1312 return fused_moe_make_expert_params_mapping(
1313 self,
1314 ckpt_gate_proj_name="w1",
1315 ckpt_down_proj_name="w2",
1316 ckpt_up_proj_name="w3",
1317 num_experts=num_experts,
1318 )
1319
1320
1321 def _make_deepseek_v4_weights_mapper(
1322 expert_dtype: str, fuse_shared_experts: bool = False
1323 ) -> WeightsMapper:
1324 if expert_dtype == "fp4":
1325 # MXFP4 experts use Mxfp4MoEMethod, which registers scales as
1326 # ``w{1,2,3}_weight_scale`` (no _inv suffix). Native DeepSeek FP8
1327 # linear / shared experts use Fp8LinearMethod's ``weight_scale_inv``.
1328 # Quark linear block-FP8 registers ``weight_scale`` and is left
1329 # as-is; load_weights aliases ``.weight_scale`` -> ``.weight_scale_inv``
1330 # only when the latter is the registered parameter.
1331 #
1332 # - DeepSeek native ``.scale``: expert scales -> ``.weight_scale``,
1333 # everything else -> ``.weight_scale_inv``.
1334 scale_regex = {
1335 re.compile(r"(\.experts\.\d+\.w[123])\.scale$"): r"\1.weight_scale",
1336 re.compile(r"\.scale$"): ".weight_scale_inv",
1337 }
1338 else:
1339 # FP8 experts use Fp8MoEMethod (block_quant=True), which registers
1340 # scales as ``w{13,2}_weight_scale_inv``. Map all ``.scale`` keys
1341 # there.
1342 scale_regex = {
1343 re.compile(r"\.scale$"): ".weight_scale_inv",
1344 }
1345 # When shared experts are fused into the routed MXFP4 grouped GEMM, the
1346 # shared_experts tensors are redirected to routed expert slots ; leave
1347 # their names untouched here.
1348 orig_to_new_substr: dict[str, str | None] = (
1349 {}
1350 if fuse_shared_experts
1351 else {".shared_experts.w2": ".shared_experts.down_proj"}
1352 )
1353 orig_to_new_substr["mtp."] = None
1354 return WeightsMapper(
1355 orig_to_new_prefix={
1356 "layers.": "model.layers.",
1357 "embed.": "model.embed.",
1358 "norm.": "model.norm.",
1359 "hc_head": "model.hc_head",
1360 "mtp.": "model.mtp.",
1361 },
1362 orig_to_new_regex=scale_regex,
1363 orig_to_new_suffix={
1364 "head.weight": "lm_head.weight",
1365 "embed.weight": "embed_tokens.weight",
1366 ".ffn.gate.bias": ".ffn.gate.e_score_correction_bias",
1367 ".input_scale": ".input_scale_2",
1368 },
1369 orig_to_new_substr=orig_to_new_substr,
1370 )
1371
1372
1373 class DeepseekV4ForCausalLM(nn.Module, SupportsPP, SupportsEagle3):
1374 model_cls = DeepseekV4Model
1375 finalizes_weights_during_load = False
1376
1377 # Default mapper assumes the original FP4-expert checkpoint layout.
1378 # Overridden per-instance in __init__ when expert_dtype != "fp4".
1379 hf_to_vllm_mapper = _make_deepseek_v4_weights_mapper("fp4")
1380 packed_modules_mapping = {
1381 "gate_up_proj": ["w1", "w3"],
1382 "fused_wqa_wkv": ["wq_a", "wkv"],
1383 }
1384
1385 def __init__(self, *, vllm_config: VllmConfig, prefix: str = ""):
1386 super().__init__()
1387
1388 config = vllm_config.model_config.hf_config
1389 self.config = config
1390 expert_dtype = getattr(config, "expert_dtype", "fp4")
1391 self.model = self.model_cls(
1392 vllm_config=vllm_config, prefix=maybe_prefix(prefix, "model")
1393 )
1394 if expert_dtype != "fp4" or self.model.is_fused_shared_expert_enabled:
1395 self.hf_to_vllm_mapper = _make_deepseek_v4_weights_mapper(
1396 expert_dtype,
1397 fuse_shared_experts=self.model.is_fused_shared_expert_enabled,
1398 )
1399 if get_pp_group().is_last_rank:
1400 self.lm_head = ParallelLMHead(
1401 config.vocab_size,
1402 config.hidden_size,
1403 prefix=maybe_prefix(prefix, "lm_head"),
1404 )
1405 else:
1406 self.lm_head = PPMissingLayer()
1407 self.logits_processor = LogitsProcessor(config.vocab_size)
1408 self.make_empty_intermediate_tensors = ( # type: ignore[method-assign]
1409 self.model.make_empty_intermediate_tensors
1410 )
1411
1412 def embed_input_ids(self, input_ids: torch.Tensor) -> torch.Tensor:
1413 return self.model.embed_input_ids(input_ids)
1414
1415 def compute_logits(
1416 self,
1417 hidden_states: torch.Tensor,
1418 ) -> torch.Tensor | None:
1419 logits = self.logits_processor(self.lm_head, hidden_states)
1420 return logits
1421
1422 def compute_logits_local(self, hidden_states: torch.Tensor) -> torch.Tensor:
1423 return self.logits_processor(self.lm_head, hidden_states, skip_gather=True)
1424
1425 def forward(
1426 self,
1427 input_ids: torch.Tensor,
1428 positions: torch.Tensor,
1429 intermediate_tensors: IntermediateTensors | None = None,
1430 inputs_embeds: torch.Tensor | None = None,
1431 ) -> torch.Tensor | IntermediateTensors:
1432 hidden_states = self.model(
1433 input_ids, positions, intermediate_tensors, inputs_embeds
1434 )
1435 return hidden_states
1436
1437 def get_mtp_target_hidden_states(self) -> torch.Tensor | None:
1438 """Pre-hc_head residual stream buffer (max_num_batched_tokens,
1439 hc_mult * hidden_size) for the MTP draft model. Populated by
1440 forward(); valid after each target step."""
1441 return getattr(self.model, "_mtp_hidden_buffer", None)
1442
1443 def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
1444 loader = AutoWeightsLoader(self)
1445 return loader.load_weights(weights, mapper=self.hf_to_vllm_mapper)
1446
1447 def process_weights_after_loading(self) -> None:
1448 # After per-layer quant finalize, so we preshuffle the final fp8 weights.
1449 fused_compressor_layers = 0
1450 for module in self.modules():
1451 if isinstance(module, DeepseekV4ROCMAiterMLAAttention):
1452 fused_compressor_layers += module.prepare_compressor_gemm_fusion()
1453 module.prepare_attn_preshuffle()
1454 elif isinstance(module, DeepseekV4MLP):
1455 module.prepare_gateup_preshuffle()
1456 if fused_compressor_layers:
1457 logger.info(
1458 "Fused the C4 compressor GEMMs in %d DeepSeek V4 layers",
1459 fused_compressor_layers,
1460 )
1461
1462 def get_expert_mapping(self) -> list[tuple[str, str, int, str]]:
1463 return self.model.get_expert_mapping()