main
py 566 lines 19.8 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 """NemotronH-MTP model with attention layers."""
4
5 import typing
6 from collections.abc import Callable, Iterable
7
8 import torch
9 import torch.nn as nn
10
11 from vllm.compilation.decorators import support_torch_compile
12 from vllm.config import CacheConfig, ModelConfig, VllmConfig
13 from vllm.config.parallel import ParallelConfig
14 from vllm.model_executor.layers.fused_moe import (
15 fused_moe_make_expert_params_mapping,
16 )
17 from vllm.model_executor.layers.layernorm import RMSNorm
18 from vllm.model_executor.layers.linear import ColumnParallelLinear
19 from vllm.model_executor.layers.logits_processor import LogitsProcessor
20 from vllm.model_executor.layers.quantization import QuantizationConfig
21 from vllm.model_executor.layers.vocab_parallel_embedding import (
22 ParallelLMHead,
23 VocabParallelEmbedding,
24 )
25 from vllm.model_executor.model_loader.weight_utils import (
26 default_weight_loader,
27 maybe_remap_kv_scale_name,
28 )
29 from vllm.model_executor.models.utils import (
30 WeightsMapper,
31 get_draft_quant_config,
32 make_empty_intermediate_tensors_factory,
33 maybe_prefix,
34 )
35 from vllm.sequence import IntermediateTensors
36 from vllm.transformers_utils.configs.nemotron_h import NemotronHConfig
37
38 from .interfaces import SupportsPP, SupportsQuant
39 from .nemotron_h import (
40 NemotronHAttentionDecoderLayer,
41 NemotronHMoEDecoderLayer,
42 )
43
44
45 class NemotronHMTPAttentionDecoderLayer(NemotronHAttentionDecoderLayer):
46 def __init__(
47 self,
48 config: NemotronHConfig,
49 layer_idx: int,
50 model_config: ModelConfig | None = None,
51 cache_config: CacheConfig | None = None,
52 quant_config: QuantizationConfig | None = None,
53 parallel_config: ParallelConfig | None = None,
54 prefix: str = "",
55 has_start_projections: bool = False,
56 has_end_norm: bool = False,
57 ) -> None:
58 super().__init__(
59 config=config,
60 layer_idx=layer_idx,
61 model_config=model_config,
62 cache_config=cache_config,
63 quant_config=quant_config,
64 parallel_config=parallel_config,
65 prefix=prefix,
66 )
67 self.has_start_projections = has_start_projections
68 self.has_end_norm = has_end_norm
69
70 if has_start_projections:
71 self.enorm = RMSNorm(config.hidden_size, eps=config.layer_norm_epsilon)
72 self.hnorm = RMSNorm(config.hidden_size, eps=config.layer_norm_epsilon)
73
74 # Fusion layer to combine embeddings with target hidden states
75 self.eh_proj = ColumnParallelLinear(
76 input_size=config.hidden_size * 2,
77 output_size=config.hidden_size,
78 bias=False,
79 gather_output=True,
80 params_dtype=config.dtype
81 if hasattr(config, "dtype")
82 else torch.bfloat16,
83 quant_config=quant_config,
84 prefix=f"{prefix}.eh_proj",
85 )
86
87 if has_end_norm:
88 self.final_layernorm = RMSNorm(
89 config.hidden_size,
90 eps=getattr(config, "layer_norm_epsilon", 1e-5),
91 )
92
93 def forward(
94 self,
95 positions: torch.Tensor,
96 hidden_states: torch.Tensor,
97 residual: torch.Tensor | None = None,
98 *,
99 inputs_embeds: torch.Tensor | None = None,
100 **kwargs: object,
101 ) -> tuple[torch.Tensor, torch.Tensor | None]:
102 # Start projections (Fusion)
103 if self.has_start_projections:
104 # Normalize both inputs before fusion
105 assert inputs_embeds is not None
106 inputs_embeds_normed = self.enorm(inputs_embeds)
107 previous_hidden_states_normed = self.hnorm(hidden_states)
108
109 # Fuse via concatenation and linear projection
110 fused = torch.cat(
111 [inputs_embeds_normed, previous_hidden_states_normed], dim=-1
112 )
113 hidden_states, _ = self.eh_proj(fused)
114
115 # Call parent forward (Attention)
116 # Parent forward expects: hidden_states, residual
117 hidden_states, residual = super().forward(
118 positions=positions,
119 hidden_states=hidden_states,
120 residual=residual,
121 )
122
123 # End norm
124 if self.has_end_norm:
125 if residual is not None:
126 hidden_states = hidden_states + residual
127 residual = None # Consumed residual
128
129 hidden_states = self.final_layernorm(hidden_states)
130
131 return hidden_states, residual
132
133
134 class NemotronHMTPMoEDecoderLayer(NemotronHMoEDecoderLayer):
135 def __init__(
136 self,
137 config: NemotronHConfig,
138 layer_idx: int,
139 model_config: ModelConfig | None = None,
140 cache_config: CacheConfig | None = None,
141 quant_config: QuantizationConfig | None = None,
142 parallel_config: ParallelConfig | None = None,
143 prefix: str = "",
144 has_start_projections: bool = False,
145 has_end_norm: bool = False,
146 ) -> None:
147 super().__init__(
148 config=config,
149 layer_idx=layer_idx,
150 model_config=model_config,
151 cache_config=cache_config,
152 quant_config=quant_config,
153 parallel_config=parallel_config,
154 prefix=prefix,
155 )
156 self.has_start_projections = has_start_projections
157 self.has_end_norm = has_end_norm
158
159 if has_start_projections:
160 self.enorm = RMSNorm(config.hidden_size, eps=config.layer_norm_epsilon)
161 self.hnorm = RMSNorm(config.hidden_size, eps=config.layer_norm_epsilon)
162
163 # Fusion layer to combine embeddings with target hidden states
164 self.eh_proj = ColumnParallelLinear(
165 input_size=config.hidden_size * 2,
166 output_size=config.hidden_size,
167 bias=False,
168 gather_output=True,
169 params_dtype=config.dtype
170 if hasattr(config, "dtype")
171 else torch.bfloat16,
172 quant_config=quant_config,
173 prefix=f"{prefix}.eh_proj",
174 )
175
176 if has_end_norm:
177 self.final_layernorm = RMSNorm(
178 config.hidden_size,
179 eps=getattr(config, "layer_norm_epsilon", 1e-5),
180 )
181
182 def forward(
183 self,
184 hidden_states: torch.Tensor,
185 residual: torch.Tensor | None = None,
186 *,
187 inputs_embeds: torch.Tensor | None = None,
188 **kwargs: object,
189 ) -> tuple[torch.Tensor, torch.Tensor | None]:
190 # Start projections (Fusion)
191 if self.has_start_projections:
192 # Normalize both inputs before fusion
193 assert inputs_embeds is not None
194 inputs_embeds_normed = self.enorm(inputs_embeds)
195 previous_hidden_states_normed = self.hnorm(hidden_states)
196
197 # Fuse via concatenation and linear projection
198 fused = torch.cat(
199 [inputs_embeds_normed, previous_hidden_states_normed], dim=-1
200 )
201 hidden_states, _ = self.eh_proj(fused)
202
203 # Call parent forward (MoE)
204 hidden_states, residual = super().forward(
205 hidden_states=hidden_states,
206 residual=residual,
207 )
208
209 # End norm
210 if self.has_end_norm:
211 if residual is not None:
212 hidden_states = hidden_states + residual
213 residual = None # Consumed residual
214
215 hidden_states = self.final_layernorm(hidden_states)
216
217 return hidden_states, residual
218
219
220 @support_torch_compile
221 class NemotronHMultiTokenPredictor(nn.Module):
222 """MTP predictor with NemotronH layers."""
223
224 def __init__(
225 self,
226 *,
227 vllm_config: VllmConfig,
228 quant_config: QuantizationConfig | None = None,
229 prefix: str = "",
230 ):
231 super().__init__()
232
233 speculative_config = vllm_config.speculative_config
234 assert speculative_config is not None
235 draft_model_config = speculative_config.draft_model_config
236 assert draft_model_config is not None
237 config = draft_model_config.hf_config.get_text_config()
238 if quant_config is None:
239 quant_config = get_draft_quant_config(vllm_config)
240
241 self.config = config
242 self.vocab_size = config.vocab_size
243 self.org_vocab_size = config.vocab_size
244
245 self.mtp_start_layer_idx = config.num_hidden_layers
246 self.num_mtp_layers = getattr(config, "num_nextn_predict_layers", 1)
247 assert self.num_mtp_layers == 1, (
248 "Only one MTP layer is supported for NemotronH-MTP"
249 )
250
251 self.pattern_str = config.mtp_hybrid_override_pattern
252 self.pattern_len = len(self.pattern_str)
253 assert self.pattern_len > 0
254
255 self.embed_tokens = VocabParallelEmbedding(
256 self.vocab_size,
257 config.hidden_size,
258 )
259
260 # Build flat list of layers
261 self.layers = torch.nn.ModuleDict()
262
263 # Total number of physical layers = num_steps * pattern_len
264 total_layers = self.num_mtp_layers * self.pattern_len
265 for i in range(total_layers):
266 step_rel_idx = i % self.pattern_len
267
268 char = self.pattern_str[step_rel_idx]
269
270 is_start_of_step = step_rel_idx == 0
271 is_end_of_step = step_rel_idx == self.pattern_len - 1
272
273 layer_prefix = f"{prefix}.layers.{i}"
274
275 # TODO smor- remove double layers formation
276 common_kwargs = dict(
277 config=config,
278 layer_idx=self.mtp_start_layer_idx + i,
279 model_config=draft_model_config,
280 cache_config=vllm_config.cache_config,
281 quant_config=quant_config,
282 parallel_config=vllm_config.parallel_config,
283 prefix=layer_prefix,
284 has_start_projections=is_start_of_step,
285 has_end_norm=is_end_of_step,
286 )
287
288 if char == "*":
289 self.layers[str(i)] = NemotronHMTPAttentionDecoderLayer(**common_kwargs)
290 elif char == "E":
291 self.layers[str(i)] = NemotronHMTPMoEDecoderLayer(**common_kwargs)
292 else:
293 raise NotImplementedError(
294 f"Pattern char '{char}' in {self.pattern_str} not implemented"
295 )
296
297 self.make_empty_intermediate_tensors: Callable[..., IntermediateTensors] = (
298 make_empty_intermediate_tensors_factory(
299 ["hidden_states", "residual"], config.hidden_size
300 )
301 )
302
303 def get_input_embeddings(self, input_ids: torch.Tensor) -> torch.Tensor:
304 assert self.embed_tokens is not None, (
305 "embed_tokens not initialized - must be shared from target model"
306 )
307 return self.embed_tokens(input_ids)
308
309 def forward(
310 self,
311 input_ids: torch.Tensor | None,
312 positions: torch.Tensor,
313 hidden_states: torch.Tensor,
314 intermediate_tensors: IntermediateTensors | None = None,
315 inputs_embeds: torch.Tensor | None = None,
316 ) -> torch.Tensor | IntermediateTensors:
317 if inputs_embeds is None:
318 assert input_ids is not None
319 inputs_embeds = self.get_input_embeddings(input_ids)
320
321 residual = None
322
323 for i in range(self.pattern_len):
324 hidden_states, residual = self.layers[str(i)](
325 inputs_embeds=inputs_embeds,
326 positions=positions,
327 hidden_states=hidden_states,
328 residual=residual,
329 )
330 return hidden_states
331
332
333 class NemotronHMTP(nn.Module, SupportsPP, SupportsQuant):
334 """NemotronH MTP model."""
335
336 hf_to_vllm_mapper = WeightsMapper(
337 orig_to_new_prefix={"language_model.": ""},
338 orig_to_new_substr={"embeddings": "embed_tokens"},
339 )
340
341 packed_modules_mapping = {
342 "qkv_proj": [
343 "q_proj",
344 "k_proj",
345 "v_proj",
346 ],
347 }
348
349 def __init__(self, *, vllm_config: VllmConfig, prefix: str = ""):
350 super().__init__()
351 speculative_config = vllm_config.speculative_config
352 assert speculative_config is not None
353 draft_model_config = speculative_config.draft_model_config
354 assert draft_model_config is not None
355 config = draft_model_config.hf_config.get_text_config()
356 self.vllm_config = vllm_config
357 self.config = config
358 # Needed for load_weights mapping
359 self.mtp_start_layer_idx = config.num_hidden_layers
360
361 # EPLB config for experts
362 self.num_redundant_experts = 0
363 if vllm_config.parallel_config and vllm_config.parallel_config.eplb_config:
364 self.num_redundant_experts = (
365 vllm_config.parallel_config.eplb_config.num_redundant_experts
366 )
367
368 # MTP predictor
369 self.model = NemotronHMultiTokenPredictor(
370 vllm_config=vllm_config,
371 quant_config=self.quant_config,
372 prefix=maybe_prefix(prefix, "mtp"),
373 )
374
375 # LM head for generating logits
376 self.has_own_lm_head = False
377 self.lm_head = ParallelLMHead(
378 self.config.vocab_size,
379 self.config.hidden_size,
380 quant_config=self.quant_config,
381 prefix=maybe_prefix(prefix, "lm_head"),
382 )
383
384 self.logits_processor = LogitsProcessor(self.config.vocab_size)
385
386 self.make_empty_intermediate_tensors = (
387 self.model.make_empty_intermediate_tensors
388 )
389
390 @staticmethod
391 def _find_quant_config(*args, **kwargs) -> QuantizationConfig | None:
392 vllm_config = kwargs.get("vllm_config")
393 assert isinstance(vllm_config, VllmConfig)
394 return get_draft_quant_config(vllm_config)
395
396 def get_input_embeddings(self, input_ids: torch.Tensor) -> torch.Tensor:
397 return self.model.get_input_embeddings(input_ids)
398
399 def forward(
400 self,
401 input_ids: torch.Tensor | None,
402 positions: torch.Tensor,
403 hidden_states: torch.Tensor | None = None,
404 intermediate_tensors: IntermediateTensors | None = None,
405 inputs_embeds: torch.Tensor | None = None,
406 **kwargs: object,
407 ) -> torch.Tensor:
408 """Forward - applies attention-based MTP."""
409 assert hidden_states is not None
410 hidden_states = self.model(
411 input_ids,
412 positions,
413 hidden_states,
414 intermediate_tensors,
415 inputs_embeds,
416 )
417 return hidden_states
418
419 def compute_logits(
420 self,
421 hidden_states: torch.Tensor,
422 ) -> torch.Tensor | None:
423 """Compute logits for DRAFT token generation."""
424 assert self.lm_head is not None, (
425 "lm_head not initialized - must be shared from target model"
426 )
427 return self.logits_processor(self.lm_head, hidden_states)
428
429 def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
430 """Load MTP weights with proper name remapping."""
431 stacked_params_mapping = [
432 # (param_name, shard_name, shard_id)
433 ("qkv_proj", "q_proj", "q"),
434 ("qkv_proj", "k_proj", "k"),
435 ("qkv_proj", "v_proj", "v"),
436 ]
437
438 expert_params_mapping = []
439 num_experts = getattr(self.config, "n_routed_experts", None)
440 if getattr(self.config, "model_type", None) == "nemotron_h_puzzle":
441 num_experts = self.config.mtp_n_routed_experts
442 if num_experts is not None:
443 expert_params_mapping = fused_moe_make_expert_params_mapping(
444 self,
445 ckpt_gate_proj_name="up_proj",
446 ckpt_down_proj_name="down_proj",
447 ckpt_up_proj_name="", # Empty - non-gated MoE
448 num_experts=num_experts,
449 num_redundant_experts=self.num_redundant_experts,
450 )
451
452 params_dict = dict(self.named_parameters())
453 loaded_params: set[str] = set()
454
455 for name, loaded_weight in weights:
456 # MTP weights are nested in "language_model."
457 # in Multimodal Nemotron-H checkpoints.
458 name = name.removeprefix("language_model.")
459 is_lm_head_weight = name.startswith("lm_head.")
460 if is_lm_head_weight:
461 self.has_own_lm_head = True
462 # Only process MTP and LM head weights -
463 # skip all non-MTP and non-LM head weights
464 if (
465 not name.startswith("mtp.")
466 and "embeddings" not in name
467 and not is_lm_head_weight
468 ):
469 continue
470 # Skip rotary embeddings (computed, not loaded)
471 if "rotary_emb.inv_freq" in name:
472 continue
473
474 name = name.replace("mtp.layers.", "model.layers.")
475
476 if "embeddings" in name:
477 name = name.replace("embeddings", "embed_tokens")
478 if name.startswith("backbone."):
479 name = name.replace("backbone.", "model.")
480
481 if "scale" in name or "zero_point" in name:
482 remapped_name = maybe_remap_kv_scale_name(name, params_dict)
483 if remapped_name is None:
484 continue
485 name = remapped_name
486
487 # Handle stacked parameters (qkv_proj) for attention layers
488 is_stacked = False
489 for param_name, weight_name, shard_id in stacked_params_mapping:
490 if weight_name not in name:
491 continue
492 # Must be in a mixer (attention layer)
493 if ".mixer." not in name:
494 continue
495
496 is_stacked = True
497 stacked_name = name.replace(weight_name, param_name)
498
499 if stacked_name.endswith(".bias") and stacked_name not in params_dict:
500 continue
501
502 if stacked_name not in params_dict:
503 # Might be that mapping failed or param doesn't exist
504 continue
505
506 param = params_dict[stacked_name]
507 weight_loader = getattr(param, "weight_loader", None)
508 if weight_loader is not None:
509 weight_loader(param, loaded_weight, shard_id)
510 loaded_params.add(stacked_name)
511 break
512
513 if is_stacked:
514 continue
515
516 is_expert_weight = False
517 for mapping in expert_params_mapping:
518 param_name, weight_name, expert_id, shard_id = mapping
519 # weight_name is like "experts.0.up_proj."
520 if weight_name not in name:
521 continue
522
523 is_expert_weight = True
524
525 # Replace the expert-specific weight name with fused parameter name
526 # e.g., "experts.0.up_proj." -> "experts.w13_"
527 name_mapped = name.replace(weight_name, param_name)
528
529 if name_mapped not in params_dict:
530 continue
531
532 param = params_dict[name_mapped]
533 weight_loader = typing.cast(Callable[..., bool], param.weight_loader)
534 success = weight_loader(
535 param,
536 loaded_weight,
537 name_mapped,
538 shard_id=shard_id,
539 expert_id=expert_id,
540 return_success=True,
541 )
542 if success:
543 loaded_params.add(name_mapped)
544 break
545
546 if is_expert_weight:
547 continue
548
549 if name.endswith(".bias") and name not in params_dict:
550 continue
551
552 if name not in params_dict:
553 continue
554
555 param = params_dict[name]
556 weight_loader = getattr(param, "weight_loader", default_weight_loader)
557 assert weight_loader is not None
558 weight_loader(param, loaded_weight)
559 loaded_params.add(name)
560
561 if not self.has_own_lm_head:
562 loaded_params.update(
563 name for name in params_dict if name.startswith("lm_head.")
564 )
565
566 return loaded_params