main
py 528 lines 20.8 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 """MTP draft model for DeepSeek V4 (internal codename: DeepseekV4).
4
5 Split from ``deepseek_mtp.py`` because the V4 architecture introduces several
6 pieces that have no analogue in V3/V32:
7 * separate ``e_proj`` / ``h_proj`` with fp8 linear quantization (instead of
8 the fused ``eh_proj``);
9 * ``hc_head`` hypercompressed vocab projection applied in ``compute_logits``;
10 * ``DeepseekV4DecoderLayer`` with its own aux-stream management;
11 * V4-specific checkpoint weight-name remapping in ``load_weights``.
12 """
13
14 import typing
15 from collections.abc import Callable, Iterable
16
17 import regex as re
18 import torch
19 import torch.nn as nn
20
21 from vllm.config import VllmConfig
22 from vllm.distributed import (
23 get_tensor_model_parallel_rank,
24 get_tensor_model_parallel_world_size,
25 )
26 from vllm.logger import init_logger
27 from vllm.model_executor.layers.fused_moe import fused_moe_make_expert_params_mapping
28 from vllm.model_executor.layers.layernorm import RMSNorm
29 from vllm.model_executor.layers.linear import ReplicatedLinear
30 from vllm.model_executor.layers.logits_processor import LogitsProcessor
31 from vllm.model_executor.layers.mhc import HCHeadOp
32 from vllm.model_executor.layers.vocab_parallel_embedding import (
33 VocabParallelEmbedding,
34 )
35 from vllm.model_executor.model_loader.mtp_validation import (
36 is_mtp_completeness_check_enabled,
37 )
38 from vllm.model_executor.model_loader.weight_utils import default_weight_loader
39 from vllm.model_executor.models.deepseek_mtp import SharedHead
40 from vllm.model_executor.models.deepseek_v2 import get_spec_layer_idx_from_weight_name
41 from vllm.model_executor.models.utils import maybe_prefix
42 from vllm.models.deepseek_v4.common.ops.fused_mtp_input_rmsnorm import (
43 _FUSED_MTP_INPUT_RMSNORM_KERNEL,
44 _MTP_SHARED_HEAD_RMSNORM_KERNEL,
45 )
46 from vllm.platforms import current_platform
47 from vllm.sequence import IntermediateTensors
48
49 from .model import (
50 DeepseekV4DecoderLayer,
51 make_deepseek_v4_expert_params_mapping,
52 )
53
54 logger = init_logger(__name__)
55
56 # MoE expert scales are fused into per-layer w13/w2 tensors. The exact
57 # parameter suffix depends on which FusedMoEFactory method handles the experts:
58 # - fp4 experts (Mxfp4MoEMethod) register ``w{1,2,3}_weight_scale``;
59 # - fp8 experts (Fp8MoEMethod with block_quant=True) register
60 # ``w{1,2,3}_weight_scale_inv``.
61 # Other FP8 linear scales (including shared experts) always use
62 # ``.weight_scale_inv``. Mirrors the per-instance mapper built by
63 # ``_make_deepseek_v4_weights_mapper`` in deepseek_v4.py.
64 _EXPERT_SCALE_RE = re.compile(r"\.experts\.\d+\.w[123]\.scale$")
65
66
67 class DeepSeekV4MultiTokenPredictorLayer(nn.Module):
68 def __init__(
69 self,
70 vllm_config: VllmConfig,
71 topk_indices_buffer: torch.Tensor,
72 prefix: str,
73 aux_stream_list: list | None = None,
74 ) -> None:
75 super().__init__()
76
77 assert vllm_config.speculative_config is not None
78 config = vllm_config.speculative_config.draft_model_config.hf_config
79 self.config = config
80 quant_config = vllm_config.quant_config
81 self.rms_norm_eps = config.rms_norm_eps
82
83 self.enorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
84 self.hnorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
85
86 # V4 keeps e_ and h_ proj separate (with fp8 linear quant) rather than
87 # fusing them the way V3 does with eh_proj.
88 self.e_proj = ReplicatedLinear(
89 config.hidden_size,
90 config.hidden_size,
91 bias=False,
92 return_bias=False,
93 quant_config=quant_config,
94 prefix=f"{prefix}.e_proj",
95 )
96 self.h_proj = ReplicatedLinear(
97 config.hidden_size,
98 config.hidden_size,
99 bias=False,
100 return_bias=False,
101 quant_config=quant_config,
102 prefix=f"{prefix}.h_proj",
103 )
104
105 self.hc_eps = config.hc_eps
106 self.hc_mult = config.hc_mult
107 self.hc_dim = self.hc_mult * config.hidden_size
108 self.hc_head_fn = nn.Parameter(
109 torch.empty(self.hc_mult, self.hc_dim, dtype=torch.float32),
110 requires_grad=False,
111 )
112 self.hc_head_base = nn.Parameter(
113 torch.empty(self.hc_mult, dtype=torch.float32),
114 requires_grad=False,
115 )
116 self.hc_head_scale = nn.Parameter(
117 torch.empty(1, dtype=torch.float32),
118 requires_grad=False,
119 )
120
121 self.shared_head = SharedHead(
122 config=config, prefix=prefix, quant_config=quant_config
123 )
124 self.mtp_block = DeepseekV4DecoderLayer(
125 vllm_config,
126 prefix,
127 topk_indices_buffer=topk_indices_buffer,
128 aux_stream_list=aux_stream_list,
129 )
130
131 self.hc_head_op = HCHeadOp()
132
133 if vllm_config.kernel_config.enable_jit_warmup:
134 _FUSED_MTP_INPUT_RMSNORM_KERNEL.register_warmup()
135 _MTP_SHARED_HEAD_RMSNORM_KERNEL.register_warmup()
136
137 def forward(
138 self,
139 input_ids: torch.Tensor,
140 positions: torch.Tensor,
141 previous_hidden_states: torch.Tensor,
142 inputs_embeds: torch.Tensor | None = None,
143 spec_step_index: int = 0,
144 ) -> torch.Tensor:
145 assert inputs_embeds is not None
146 # Target stashes pre-hc_head residual as flat (T, hc_mult * D);
147 # reshape to (T, hc_mult, D) — the training-time layout — before
148 # the fused norm pass so both inputs are 3D-friendly.
149 previous_hidden_states = previous_hidden_states.view(
150 -1, self.hc_mult, self.config.hidden_size
151 )
152 # Fused: mask inputs at position 0 (not needed by MTP), enorm, hnorm.
153 inputs_embeds, previous_hidden_states = _FUSED_MTP_INPUT_RMSNORM_KERNEL(
154 inputs_embeds,
155 positions,
156 previous_hidden_states,
157 self.enorm.weight.data,
158 self.hnorm.weight.data,
159 self.enorm.variance_epsilon,
160 self.hc_mult,
161 )
162 hidden_states = self.h_proj(previous_hidden_states) + self.e_proj(
163 inputs_embeds
164 ).unsqueeze(-2)
165 hidden_states, residual, post_mix, res_mix = self.mtp_block(
166 positions=positions, x=hidden_states, input_ids=None
167 )
168 hidden_states = self.mtp_block.hc_post(
169 hidden_states, residual, post_mix, res_mix
170 )
171 # Return the flat pre-hc_head residual so it can be re-fed as the
172 # next spec step's `previous_hidden_states` when
173 # num_speculative_tokens > 1. hc_head is deferred to compute_logits.
174 return hidden_states.flatten(1)
175
176
177 class DeepSeekV4MultiTokenPredictor(nn.Module):
178 def __init__(self, *, vllm_config: VllmConfig, prefix: str = ""):
179 super().__init__()
180 config = vllm_config.model_config.hf_config
181 self.mtp_start_layer_idx = config.num_hidden_layers
182 self.num_mtp_layers = config.num_nextn_predict_layers
183 self.device = current_platform.device_type
184
185 topk_tokens = config.index_topk
186 self.topk_indices_buffer = torch.empty(
187 vllm_config.scheduler_config.max_num_batched_tokens,
188 topk_tokens,
189 dtype=torch.int32,
190 device=self.device,
191 )
192
193 # Disable aux streams on XPU — no multi-stream overlap support.
194 aux_stream_list = None
195
196 # to map the exact layer index from weights
197 self.layers = torch.nn.ModuleDict(
198 {
199 str(idx): DeepSeekV4MultiTokenPredictorLayer(
200 vllm_config,
201 self.topk_indices_buffer,
202 f"{prefix}.layers.{idx}",
203 aux_stream_list=aux_stream_list,
204 )
205 for idx in range(
206 self.mtp_start_layer_idx,
207 self.mtp_start_layer_idx + self.num_mtp_layers,
208 )
209 }
210 )
211 self.embed_tokens = VocabParallelEmbedding(
212 config.vocab_size,
213 config.hidden_size,
214 prefix=maybe_prefix(prefix, "embed_tokens"),
215 )
216 self.logits_processor = LogitsProcessor(config.vocab_size)
217
218 def embed_input_ids(self, input_ids: torch.Tensor) -> torch.Tensor:
219 return self.embed_tokens(input_ids)
220
221 def forward(
222 self,
223 input_ids: torch.Tensor,
224 positions: torch.Tensor,
225 previous_hidden_states: torch.Tensor,
226 inputs_embeds: torch.Tensor | None = None,
227 spec_step_idx: int = 0,
228 ) -> torch.Tensor:
229 if inputs_embeds is None:
230 inputs_embeds = self.embed_tokens(input_ids)
231 current_step_idx = spec_step_idx % self.num_mtp_layers
232 return self.layers[str(self.mtp_start_layer_idx + current_step_idx)](
233 input_ids,
234 positions,
235 previous_hidden_states,
236 inputs_embeds,
237 current_step_idx,
238 )
239
240 def compute_logits(
241 self,
242 hidden_states: torch.Tensor,
243 spec_step_idx: int = 0,
244 ) -> torch.Tensor:
245 current_step_idx = spec_step_idx % self.num_mtp_layers
246 mtp_layer = self.layers[str(self.mtp_start_layer_idx + current_step_idx)]
247 # MTP forward returns the pre-hc_head residual (T, hc_mult * D); apply
248 # hc_head here so logits are computed from the dense hidden state.
249 hidden_states = hidden_states.view(
250 -1, mtp_layer.hc_mult, mtp_layer.config.hidden_size
251 )
252 hidden_states = mtp_layer.hc_head_op(
253 hidden_states,
254 mtp_layer.hc_head_fn,
255 mtp_layer.hc_head_scale,
256 mtp_layer.hc_head_base,
257 mtp_layer.rms_norm_eps,
258 mtp_layer.hc_eps,
259 )
260 hidden_states = _MTP_SHARED_HEAD_RMSNORM_KERNEL(
261 hidden_states,
262 mtp_layer.shared_head.norm.weight.data,
263 mtp_layer.shared_head.norm.variance_epsilon,
264 )
265 logits = self.logits_processor(mtp_layer.shared_head.head, hidden_states)
266 return logits
267
268
269 class DeepSeekV4MTP(nn.Module):
270 def __init__(self, *, vllm_config: VllmConfig, prefix: str = ""):
271 super().__init__()
272 self.config = vllm_config.model_config.hf_config
273 self.quant_config = vllm_config.quant_config
274 self.model = DeepSeekV4MultiTokenPredictor(
275 vllm_config=vllm_config, prefix=maybe_prefix(prefix, "model")
276 )
277
278 def embed_input_ids(self, input_ids: torch.Tensor) -> torch.Tensor:
279 return self.model.embed_input_ids(input_ids)
280
281 def forward(
282 self,
283 input_ids: torch.Tensor | None,
284 positions: torch.Tensor,
285 hidden_states: torch.Tensor,
286 intermediate_tensors: IntermediateTensors | None = None,
287 inputs_embeds: torch.Tensor | None = None,
288 spec_step_idx: int = 0,
289 ) -> torch.Tensor:
290 hidden_states = self.model(
291 input_ids, positions, hidden_states, inputs_embeds, spec_step_idx
292 )
293 return hidden_states
294
295 def compute_logits(
296 self,
297 hidden_states: torch.Tensor,
298 spec_step_idx: int = 0,
299 ) -> torch.Tensor | None:
300 return self.model.compute_logits(hidden_states, spec_step_idx)
301
302 def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
303 # Weight name remapping for checkpoint compatibility.
304 # Maps checkpoint weight paths to model parameter paths.
305 WEIGHT_NAME_REMAPPING: dict[str, str] = {
306 ".emb.tok_emb.weight": ".embed_tokens.weight",
307 ".head.weight": ".shared_head.head.weight",
308 ".norm.weight": ".shared_head.norm.weight",
309 }
310
311 def _remap_weight_name(name: str) -> str:
312 """Remap checkpoint weight names to model parameter names."""
313 for old_pattern, new_pattern in WEIGHT_NAME_REMAPPING.items():
314 if old_pattern in name:
315 name = name.replace(old_pattern, new_pattern)
316 return name
317
318 def _find_mtp_layer_idx(name: str) -> int:
319 subnames = name.split(".")
320 for subname in subnames:
321 try:
322 # we return the first encountered integer
323 return int(subname)
324 except ValueError:
325 continue
326 return 0
327
328 stacked_params_mapping = [
329 # (param_name, shard_name, shard_id)
330 ("gate_up_proj", "w1", 0),
331 ("gate_up_proj", "w3", 1),
332 ("attn.fused_wqa_wkv", "attn.wq_a", 0),
333 ("attn.fused_wqa_wkv", "attn.wkv", 1),
334 ]
335 params_dict = dict(self.named_parameters())
336 loaded_params: set[str] = set()
337
338 # TP for attention
339 tp_size = get_tensor_model_parallel_world_size()
340 tp_rank = get_tensor_model_parallel_rank()
341 n_head = self.config.num_attention_heads
342 n_local_head = n_head // tp_size
343 head_rank_start = n_local_head * tp_rank
344 head_rank_end = n_local_head * (tp_rank + 1)
345
346 # Pre-compute expert mapping ONCE.
347 first_layer = next(iter(self.model.layers.values()))
348 if first_layer.mtp_block.ffn.use_mega_moe:
349 expert_mapping = make_deepseek_v4_expert_params_mapping(
350 self.config.n_routed_experts
351 )
352 else:
353 expert_mapping = fused_moe_make_expert_params_mapping(
354 self,
355 ckpt_gate_proj_name="w1",
356 ckpt_down_proj_name="w2",
357 ckpt_up_proj_name="w3",
358 num_experts=self.config.n_routed_experts,
359 )
360
361 # FP8 experts register ``..._weight_scale_inv`` (block_quant) while
362 # FP4/MXFP4 experts register ``..._weight_scale``. Choose the suffix
363 # for the rename below based on the model's expert dtype.
364 expert_scale_suffix = (
365 ".weight_scale"
366 if getattr(self.config, "expert_dtype", "fp4") == "fp4"
367 else ".weight_scale_inv"
368 )
369
370 for name, loaded_weight in weights:
371 mtp_layer_idx = _find_mtp_layer_idx(name)
372 # V4 checkpoints store MTP weights as `mtp.{i}.*`; remap to
373 # `model.layers.{num_hidden_layers + i}.*` so that
374 # get_spec_layer_idx_from_weight_name can identify them.
375 name = name.replace(
376 f"mtp.{mtp_layer_idx}.",
377 f"model.layers.{self.config.num_hidden_layers + mtp_layer_idx}.",
378 )
379
380 spec_layer = get_spec_layer_idx_from_weight_name(self.config, name)
381 if spec_layer is None:
382 continue
383
384 name = _remap_weight_name(name)
385 name = self._rewrite_spec_layer_name(spec_layer, name)
386
387 if spec_layer != self.model.mtp_start_layer_idx and ".layers" not in name:
388 continue
389 if name.endswith(".scale"):
390 suffix = (
391 expert_scale_suffix
392 if _EXPERT_SCALE_RE.search(name)
393 else ".weight_scale_inv"
394 )
395 name = name.removesuffix(".scale") + suffix
396 for param_name, weight_name, shard_id in stacked_params_mapping:
397 # Skip non-stacked layers and experts (experts handled below).
398 if ".experts." in name:
399 continue
400 if weight_name not in name:
401 continue
402 name = name.replace(weight_name, param_name)
403
404 param = params_dict[name]
405 weight_loader = param.weight_loader
406 weight_loader(param, loaded_weight, shard_id)
407 loaded_params.add(name)
408 break
409 else:
410 if ".experts." in name:
411 # Reinterpret E8M0 scales as uint8 to preserve raw
412 # exponent bytes; numeric copy_() would zero them.
413 # Mirrors the main DeepseekV4 loader.
414 if (
415 "weight_scale" in name
416 and loaded_weight.dtype == torch.float8_e8m0fnu
417 ):
418 loaded_weight = loaded_weight.view(torch.uint8)
419 for mapping in expert_mapping:
420 param_name, weight_name, expert_id, expert_shard_id = mapping
421 if weight_name not in name:
422 continue
423 name_mapped = name.replace(weight_name, param_name)
424 param = params_dict[name_mapped]
425 # We should ask the weight loader to return success or not
426 # here since otherwise we may skip experts with other
427 # available replicas.
428 weight_loader = typing.cast(
429 Callable[..., bool], param.weight_loader
430 )
431 success = weight_loader(
432 param,
433 loaded_weight,
434 name_mapped,
435 shard_id=expert_shard_id,
436 expert_id=expert_id,
437 return_success=True,
438 )
439 if success:
440 name = name_mapped
441 loaded_params.add(name_mapped)
442 break
443 continue
444 elif "attn_sink" in name:
445 narrow_weight = loaded_weight[head_rank_start:head_rank_end]
446 n = narrow_weight.shape[0]
447 params_dict[name][:n].copy_(narrow_weight)
448 loaded_params.add(name)
449 continue
450 else:
451 if ".shared_experts.w2" in name:
452 name = name.replace(
453 ".shared_experts.w2", ".shared_experts.down_proj"
454 )
455 if name.endswith(".ffn.gate.bias"):
456 # ``e_score_correction_bias`` lives on the gate
457 # under a different attribute name.
458 name = name.replace(
459 ".ffn.gate.bias",
460 ".ffn.gate.e_score_correction_bias",
461 )
462 param = params_dict[name]
463 weight_loader = getattr(
464 param, "weight_loader", default_weight_loader
465 )
466 weight_loader(param, loaded_weight)
467 loaded_params.add(name)
468 continue
469
470 loaded_layers: set[int] = set()
471 for param_name in loaded_params:
472 spec_layer = get_spec_layer_idx_from_weight_name(self.config, param_name)
473 if spec_layer is not None:
474 loaded_layers.add(spec_layer)
475 for layer_idx in range(
476 self.model.mtp_start_layer_idx,
477 self.model.mtp_start_layer_idx + self.model.num_mtp_layers,
478 ):
479 if layer_idx not in loaded_layers and is_mtp_completeness_check_enabled():
480 raise ValueError(
481 f"MTP speculative decoding layer {layer_idx} weights "
482 f"missing from checkpoint. The checkpoint may have "
483 f"been quantized without including the MTP layers. "
484 f"Use a checkpoint that includes MTP layer weights, "
485 f"or disable speculative decoding."
486 )
487 self.finalize_mega_moe_weights()
488 logger.info_once("MTP draft model loaded: %d params", len(loaded_params))
489 return loaded_params
490
491 def finalize_mega_moe_weights(self) -> None:
492 for layer in self.model.layers.values():
493 layer.mtp_block.ffn.finalize_mega_moe_weights()
494
495 def _rewrite_spec_layer_name(self, spec_layer: int, name: str) -> str:
496 """Rewrite the weight name to match the format of the original model.
497 Add .mtp_block for modules in transformer layer block for spec layer
498 and rename shared layer weights to be top level.
499 """
500 spec_layer_weight_names = [
501 "embed_tokens",
502 "enorm",
503 "hnorm",
504 "h_proj",
505 "e_proj",
506 "shared_head",
507 "hc_head_fn",
508 "hc_head_base",
509 "hc_head_scale",
510 ]
511 shared_weight_names = ["embed_tokens"]
512 spec_layer_weight = False
513 shared_weight = False
514 for weight_name in spec_layer_weight_names:
515 if weight_name in name:
516 spec_layer_weight = True
517 if weight_name in shared_weight_names:
518 shared_weight = True
519 break
520 if not spec_layer_weight:
521 # treat rest weights as weights for transformer layer block
522 name = name.replace(
523 f"model.layers.{spec_layer}.", f"model.layers.{spec_layer}.mtp_block."
524 )
525 elif shared_weight:
526 # treat shared weights as top level weights
527 name = name.replace(f"model.layers.{spec_layer}.", "model.")
528 return name