main
py 2,009 lines 77 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
4 # Adapted from
5 # https://github.com/huggingface/transformers/blob/v4.28.0/src/transformers/models/llama/modeling_llama.py
6 # Copyright 2023 The vLLM team.
7 # Copyright 2023 DeepSeek-AI and the HuggingFace Inc. team. All rights reserved.
8 #
9 # This code is based on EleutherAI's GPT-NeoX library and the GPT-NeoX
10 # and OPT implementations in this library. It has been modified from its
11 # original forms to accommodate minor architectural differences compared
12 # to GPT-NeoX and OPT used by the Meta AI team that trained the model.
13 #
14 # Licensed under the Apache License, Version 2.0 (the "License");
15 # you may not use this file except in compliance with the License.
16 # You may obtain a copy of the License at
17 #
18 # http://www.apache.org/licenses/LICENSE-2.0
19 #
20 # Unless required by applicable law or agreed to in writing, software
21 # distributed under the License is distributed on an "AS IS" BASIS,
22 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
23 # See the License for the specific language governing permissions and
24 # limitations under the License.
25 """Inference-only DeepseekV2/DeepseekV3 model."""
26
27 import typing
28 from collections.abc import Callable, Iterable
29 from itertools import islice
30
31 import torch
32 from torch import nn
33 from transformers import DeepseekV2Config, DeepseekV3Config
34
35 import vllm._custom_ops as ops
36 import vllm.envs as envs
37 from vllm._aiter_ops import rocm_aiter_ops
38 from vllm.compilation.decorators import support_torch_compile
39 from vllm.config import CacheConfig, ParallelConfig, VllmConfig, get_current_vllm_config
40 from vllm.distributed import (
41 get_ep_group,
42 get_pp_group,
43 get_tensor_model_parallel_rank,
44 get_tensor_model_parallel_world_size,
45 tensor_model_parallel_all_gather,
46 tensor_model_parallel_reduce_scatter,
47 )
48 from vllm.logger import init_logger
49 from vllm.model_executor.layers.activation import SiluAndMul
50 from vllm.model_executor.layers.attention import (
51 Attention,
52 EncoderOnlyAttention,
53 RSWAAttention,
54 )
55 from vllm.model_executor.layers.attention_layer_base import AttentionLayerBase
56 from vllm.model_executor.layers.fused_moe import (
57 FusedMoEFactory,
58 GateLinear,
59 fused_moe_make_expert_params_mapping,
60 )
61 from vllm.model_executor.layers.fused_moe.utils import (
62 is_model_fused_shared_expert_compatible,
63 resolve_layer_fused_shared_expert,
64 )
65 from vllm.model_executor.layers.layernorm import LayerNorm, RMSNorm
66 from vllm.model_executor.layers.linear import (
67 ColumnParallelLinear,
68 DCPGroupColumnParallelLinear,
69 MergedColumnParallelLinear,
70 QKVParallelLinear,
71 ReplicatedLinear,
72 RowParallelLinear,
73 )
74 from vllm.model_executor.layers.logits_processor import LogitsProcessor
75 from vllm.model_executor.layers.mla import MLAModules, MultiHeadLatentAttentionWrapper
76 from vllm.model_executor.layers.quantization import QuantizationConfig
77 from vllm.model_executor.layers.quantization.utils.fp8_utils import (
78 per_token_group_quant_fp8,
79 )
80 from vllm.model_executor.layers.quantization.utils.quant_utils import (
81 GroupShape,
82 scaled_dequantize,
83 )
84 from vllm.model_executor.layers.rotary_embedding import get_rope
85 from vllm.model_executor.layers.sparse_attn_indexer import (
86 SparseAttnIndexer,
87 fused_indexer_q_rope_quant,
88 )
89 from vllm.model_executor.layers.vocab_parallel_embedding import (
90 ParallelLMHead,
91 VocabParallelEmbedding,
92 )
93 from vllm.model_executor.model_loader.weight_utils import (
94 default_weight_loader,
95 maybe_remap_kv_scale_name,
96 )
97 from vllm.model_executor.models.utils import (
98 AutoWeightsLoader,
99 extract_layer_index,
100 sequence_parallel_chunk,
101 )
102 from vllm.platforms import current_platform
103 from vllm.sequence import IntermediateTensors
104 from vllm.utils.torch_utils import direct_register_custom_op
105 from vllm.v1.attention.backend import AttentionBackend, AttentionType
106 from vllm.v1.attention.backends.mla.index_group import (
107 SparseMLAIndexGroupBuilder,
108 get_sparse_mla_index_group_max_rows,
109 )
110 from vllm.v1.attention.backends.mla.indexer import (
111 DeepseekV32IndexerBackend,
112 )
113 from vllm.v1.kv_cache_interface import KVCacheSpec, MLAAttentionSpec, SparseCacheRole
114
115 from .interfaces import (
116 MixtureOfExperts,
117 SupportsEagle,
118 SupportsEagle3,
119 SupportsLoRA,
120 SupportsPP,
121 )
122 from .utils import (
123 PPMissingLayer,
124 get_pp_missing_layer_names,
125 get_spec_layer_idx_from_weight_name,
126 is_pp_missing_parameter,
127 make_empty_intermediate_tensors_factory,
128 make_layers,
129 maybe_prefix,
130 )
131
132 logger = init_logger(__name__)
133
134
135 def _get_moe_router_dtype(
136 config: DeepseekV2Config | DeepseekV3Config,
137 ) -> torch.dtype | None:
138 router_dtype = getattr(config, "moe_router_dtype", None)
139 if getattr(config, "model_type", None) == "glm_moe_dsa":
140 # Older GLM-5/5.2 configs require fp32 routing but do not expose
141 # moe_router_dtype yet.
142 return torch.float32
143 if router_dtype == "float32":
144 return torch.float32
145 return None
146
147
148 class DeepseekAttention(nn.Module):
149 """Normal MHA implementation used by Deepseek v1."""
150
151 def __init__(
152 self,
153 vllm_config: VllmConfig,
154 config: DeepseekV2Config | DeepseekV3Config,
155 hidden_size: int,
156 num_heads: int,
157 max_position_embeddings: int = 8192,
158 cache_config: CacheConfig | None = None,
159 quant_config: QuantizationConfig | None = None,
160 reduce_results: bool = True,
161 prefix: str = "",
162 **kwargs,
163 ) -> None:
164 super().__init__()
165 self.hidden_size = hidden_size
166 tp_size = get_tensor_model_parallel_world_size()
167 self.total_num_heads = num_heads
168 assert self.total_num_heads % tp_size == 0
169 self.num_heads = self.total_num_heads // tp_size
170 self.total_num_kv_heads = config.num_key_value_heads
171 if self.total_num_kv_heads >= tp_size:
172 # Number of KV heads is greater than TP size, so we partition
173 # the KV heads across multiple tensor parallel GPUs.
174 assert self.total_num_kv_heads % tp_size == 0
175 else:
176 # Number of KV heads is less than TP size, so we replicate
177 # the KV heads across multiple tensor parallel GPUs.
178 assert tp_size % self.total_num_kv_heads == 0
179 self.num_kv_heads = max(1, self.total_num_kv_heads // tp_size)
180 self.head_dim = hidden_size // self.total_num_heads
181 self.q_size = self.num_heads * self.head_dim
182 self.kv_size = self.num_kv_heads * self.head_dim
183 self.scaling = self.head_dim**-0.5
184 self.max_position_embeddings = max_position_embeddings
185
186 self.qkv_proj = QKVParallelLinear(
187 hidden_size,
188 self.head_dim,
189 self.total_num_heads,
190 self.total_num_kv_heads,
191 bias=False,
192 quant_config=quant_config,
193 )
194
195 self.o_proj = RowParallelLinear(
196 self.total_num_heads * self.head_dim,
197 hidden_size,
198 bias=False,
199 reduce_results=reduce_results,
200 quant_config=quant_config,
201 )
202
203 self.rotary_emb = get_rope(
204 self.head_dim,
205 max_position=max_position_embeddings,
206 rope_parameters=config.rope_parameters,
207 )
208 rswa_window = getattr(vllm_config.model_config.hf_config, "rswa_window", None)
209 if rswa_window is not None:
210 self.attn = RSWAAttention(
211 self.num_heads,
212 self.head_dim,
213 self.scaling,
214 num_kv_heads=self.num_kv_heads,
215 cache_config=cache_config,
216 quant_config=quant_config,
217 prefix=f"{prefix}.attn",
218 rswa_window=rswa_window,
219 )
220 else:
221 self.attn = Attention(
222 self.num_heads,
223 self.head_dim,
224 self.scaling,
225 num_kv_heads=self.num_kv_heads,
226 cache_config=cache_config,
227 quant_config=quant_config,
228 prefix=f"{prefix}.attn",
229 )
230
231 def forward(
232 self,
233 positions: torch.Tensor,
234 hidden_states: torch.Tensor,
235 ) -> torch.Tensor:
236 qkv, _ = self.qkv_proj(hidden_states)
237 q, k, v = qkv.split([self.q_size, self.kv_size, self.kv_size], dim=-1)
238 q, k = self.rotary_emb(positions, q, k)
239 attn_output = self.attn(q, k, v)
240 output, _ = self.o_proj(attn_output)
241 return output
242
243
244 class DeepseekV2MLP(nn.Module):
245 def __init__(
246 self,
247 hidden_size: int,
248 intermediate_size: int,
249 hidden_act: str,
250 quant_config: QuantizationConfig | None = None,
251 reduce_results: bool = True,
252 is_sequence_parallel=False,
253 prefix: str = "",
254 ) -> None:
255 super().__init__()
256
257 # If is_sequence_parallel, the input and output tensors are sharded
258 # across the ranks within the tp_group. In this case the weights are
259 # replicated and no collective ops are needed.
260 # Otherwise we use standard TP with an allreduce at the end.
261 self.gate_up_proj = MergedColumnParallelLinear(
262 hidden_size,
263 [intermediate_size] * 2,
264 bias=False,
265 quant_config=quant_config,
266 disable_tp=is_sequence_parallel,
267 prefix=f"{prefix}.gate_up_proj",
268 )
269 self.down_proj = RowParallelLinear(
270 intermediate_size,
271 hidden_size,
272 bias=False,
273 quant_config=quant_config,
274 reduce_results=reduce_results,
275 disable_tp=is_sequence_parallel,
276 prefix=f"{prefix}.down_proj",
277 )
278 if hidden_act != "silu":
279 raise ValueError(
280 f"Unsupported activation: {hidden_act}. Only silu is supported for now."
281 )
282 self.act_fn = SiluAndMul()
283
284 def forward(self, x):
285 gate_up, _ = self.gate_up_proj(x)
286 x = self.act_fn(gate_up)
287 x, _ = self.down_proj(x)
288 return x
289
290
291 class DeepseekV2MoE(nn.Module):
292 def __init__(
293 self,
294 config: DeepseekV2Config | DeepseekV3Config,
295 parallel_config: ParallelConfig,
296 quant_config: QuantizationConfig | None = None,
297 reduce_results: bool = True,
298 prefix: str = "",
299 apply_routed_scale_to_output: bool = False,
300 ):
301 super().__init__()
302 self.tp_size = get_tensor_model_parallel_world_size()
303 self.tp_rank = get_tensor_model_parallel_rank()
304
305 self.routed_scaling_factor = getattr(config, "routed_scaling_factor", 1.0)
306
307 self.ep_group = get_ep_group().device_group
308 self.ep_size = self.ep_group.size()
309 self.n_routed_experts: int = config.n_routed_experts
310 self.n_shared_experts: int = config.n_shared_experts
311
312 self.is_sequence_parallel = parallel_config.use_sequence_parallel_moe
313
314 if config.hidden_act != "silu":
315 raise ValueError(
316 f"Unsupported activation: {config.hidden_act}. "
317 "Only silu is supported for now."
318 )
319
320 self.router_dtype = _get_moe_router_dtype(config)
321 self.gate = GateLinear(
322 config.hidden_size,
323 config.n_routed_experts,
324 out_dtype=self.router_dtype,
325 prefix=f"{prefix}.gate",
326 )
327 if getattr(config, "topk_method", None) == "noaux_tc":
328 self.gate.e_score_correction_bias = nn.Parameter(
329 torch.empty(config.n_routed_experts, dtype=torch.float32)
330 )
331 else:
332 self.gate.e_score_correction_bias = None
333
334 # Load balancing settings.
335 eplb_config = parallel_config.eplb_config
336 self.enable_eplb = parallel_config.enable_eplb
337
338 self.n_redundant_experts = eplb_config.num_redundant_experts
339 self.n_logical_experts = self.n_routed_experts
340 self.n_physical_experts = self.n_logical_experts + self.n_redundant_experts
341 self.n_local_physical_experts = self.n_physical_experts // self.ep_size
342
343 self.is_rocm_aiter_moe_enabled = rocm_aiter_ops.is_fused_moe_enabled()
344
345 self.is_fused_shared_expert_enabled = False
346 if config.n_shared_experts is not None:
347 self.is_fused_shared_expert_enabled = resolve_layer_fused_shared_expert(
348 quant_config, prefix
349 )
350
351 if (
352 self.is_rocm_aiter_moe_enabled
353 and self.gate.e_score_correction_bias is not None
354 and self.gate.out_dtype is None
355 ):
356 # Accumulates in fp32; avoids bf16->fp32 cast.
357 self.gate.set_out_dtype(self.gate.weight.dtype)
358
359 if config.n_shared_experts is None or self.is_fused_shared_expert_enabled:
360 self.shared_experts = None
361 else:
362 intermediate_size = config.moe_intermediate_size * config.n_shared_experts
363
364 self.shared_experts = DeepseekV2MLP(
365 hidden_size=config.hidden_size,
366 intermediate_size=intermediate_size,
367 hidden_act=config.hidden_act,
368 quant_config=quant_config,
369 is_sequence_parallel=self.is_sequence_parallel,
370 reduce_results=False,
371 prefix=f"{prefix}.shared_experts",
372 )
373
374 self.experts = FusedMoEFactory(
375 shared_experts=self.shared_experts,
376 gate=self.gate,
377 num_experts=config.n_routed_experts,
378 top_k=config.num_experts_per_tok,
379 hidden_size=config.hidden_size,
380 intermediate_size=config.moe_intermediate_size,
381 renormalize=config.norm_topk_prob,
382 quant_config=quant_config,
383 use_grouped_topk=True,
384 num_expert_group=getattr(config, "n_group", 1),
385 topk_group=getattr(config, "topk_group", 1),
386 prefix=f"{prefix}.experts",
387 scoring_func=getattr(config, "scoring_func", "softmax"),
388 routed_scaling_factor=self.routed_scaling_factor,
389 apply_routed_scale_to_output=apply_routed_scale_to_output,
390 e_score_correction_bias=self.gate.e_score_correction_bias,
391 enable_eplb=self.enable_eplb,
392 num_redundant_experts=self.n_redundant_experts,
393 is_sequence_parallel=self.is_sequence_parallel,
394 reduce_results=reduce_results,
395 n_shared_experts=config.n_shared_experts
396 if self.is_fused_shared_expert_enabled
397 else None,
398 fuse_shared_experts=self.is_fused_shared_expert_enabled,
399 router_logits_dtype=self.gate.out_dtype,
400 )
401
402 if (
403 self.is_rocm_aiter_moe_enabled
404 and self.gate.e_score_correction_bias is not None
405 ):
406 self.gate.e_score_correction_bias.data = (
407 self.gate.e_score_correction_bias.data.to(self.gate.out_dtype)
408 )
409
410 def forward(
411 self,
412 hidden_states: torch.Tensor,
413 already_sequence_parallel: bool = False,
414 ) -> torch.Tensor:
415 num_tokens, hidden_dim = hidden_states.shape
416 hidden_states = hidden_states.view(-1, hidden_dim)
417
418 # Chunk the hidden states so they aren't replicated across TP ranks.
419 # This avoids duplicate computation in self.experts.
420 if self.is_sequence_parallel and not already_sequence_parallel:
421 hidden_states = sequence_parallel_chunk(hidden_states)
422
423 final_hidden_states = self.experts(
424 hidden_states=hidden_states, router_logits=hidden_states
425 )
426
427 if self.is_sequence_parallel and not already_sequence_parallel:
428 final_hidden_states = tensor_model_parallel_all_gather(
429 final_hidden_states, 0
430 )
431 final_hidden_states = final_hidden_states[:num_tokens]
432
433 return final_hidden_states.view(num_tokens, hidden_dim)
434
435
436 def yarn_get_mscale(scale: float = 1, mscale: float = 1) -> float:
437 import math
438
439 if scale <= 1:
440 return 1.0
441 return 0.1 * mscale * math.log(scale) + 1.0
442
443
444 def _get_llama_4_scaling(
445 original_max_position_embeddings: int, scaling_beta: float, positions: torch.Tensor
446 ) -> torch.Tensor:
447 scaling = 1 + scaling_beta * torch.log(
448 1 + torch.floor(positions / original_max_position_embeddings)
449 )
450 # Broadcast over num_heads and head_dim
451 return scaling[..., None, None]
452
453
454 class DeepseekV2Attention(nn.Module):
455 def __init__(
456 self,
457 vllm_config: VllmConfig,
458 config: DeepseekV2Config | DeepseekV3Config,
459 hidden_size: int,
460 num_heads: int,
461 qk_nope_head_dim: int,
462 qk_rope_head_dim: int,
463 v_head_dim: int,
464 q_lora_rank: int | None,
465 kv_lora_rank: int,
466 max_position_embeddings: int = 8192,
467 cache_config: CacheConfig | None = None,
468 quant_config: QuantizationConfig | None = None,
469 topk_indices_buffer: torch.Tensor | None = None,
470 reduce_results: bool = True,
471 prefix: str = "",
472 ) -> None:
473 super().__init__()
474 self.hidden_size = hidden_size
475 self.qk_nope_head_dim = qk_nope_head_dim
476 self.qk_rope_head_dim = qk_rope_head_dim
477 self.qk_head_dim = qk_nope_head_dim + qk_rope_head_dim
478 self.v_head_dim = v_head_dim
479 self.q_lora_rank = q_lora_rank
480 self.kv_lora_rank = kv_lora_rank
481 self.num_heads = num_heads
482 tp_size = get_tensor_model_parallel_world_size()
483 assert num_heads % tp_size == 0
484 self.num_local_heads = num_heads // tp_size
485 self.scaling = self.qk_head_dim**-0.5
486 self.max_position_embeddings = max_position_embeddings
487 assert topk_indices_buffer is None, (
488 "topk_indices_buffer is not \
489 supported for DeepseekV2Attention"
490 )
491
492 if self.q_lora_rank is not None:
493 self.q_a_proj = ReplicatedLinear(
494 self.hidden_size,
495 self.q_lora_rank,
496 bias=False,
497 quant_config=quant_config,
498 prefix=f"{prefix}.q_a_proj",
499 )
500 self.q_a_layernorm = RMSNorm(self.q_lora_rank, eps=config.rms_norm_eps)
501 self.q_b_proj = ColumnParallelLinear(
502 self.q_lora_rank,
503 self.num_heads * self.qk_head_dim,
504 bias=False,
505 quant_config=quant_config,
506 prefix=f"{prefix}.q_b_proj",
507 )
508 else:
509 self.q_proj = ColumnParallelLinear(
510 self.hidden_size,
511 self.num_heads * self.qk_head_dim,
512 bias=False,
513 quant_config=quant_config,
514 prefix=f"{prefix}.q_proj",
515 )
516
517 self.kv_a_proj_with_mqa = ReplicatedLinear(
518 self.hidden_size,
519 self.kv_lora_rank + self.qk_rope_head_dim,
520 bias=False,
521 quant_config=quant_config,
522 prefix=f"{prefix}.kv_a_proj_with_mqa",
523 )
524 self.kv_a_layernorm = RMSNorm(self.kv_lora_rank, eps=config.rms_norm_eps)
525 self.kv_b_proj = ColumnParallelLinear(
526 self.kv_lora_rank,
527 self.num_heads * (self.qk_nope_head_dim + self.v_head_dim),
528 bias=False,
529 quant_config=quant_config,
530 prefix=f"{prefix}.kv_b_proj",
531 )
532 # O projection.
533 self.o_proj = RowParallelLinear(
534 self.num_heads * self.v_head_dim,
535 self.hidden_size,
536 bias=False,
537 reduce_results=reduce_results,
538 quant_config=quant_config,
539 prefix=f"{prefix}.o_proj",
540 )
541 if config.rope_parameters["rope_type"] != "default":
542 config.rope_parameters["rope_type"] = (
543 "deepseek_llama_scaling"
544 if config.rope_parameters.get("attention_factor") == 1.0
545 else "deepseek_yarn"
546 )
547
548 self.rotary_emb = get_rope(
549 qk_rope_head_dim,
550 max_position=max_position_embeddings,
551 rope_parameters=config.rope_parameters,
552 is_neox_style=False,
553 )
554
555 if (
556 config.rope_parameters["rope_type"] != "default"
557 and config.rope_parameters["rope_type"] == "deepseek_yarn"
558 ):
559 mscale_all_dim = config.rope_parameters.get("mscale_all_dim", False)
560 scaling_factor = config.rope_parameters["factor"]
561 mscale = yarn_get_mscale(scaling_factor, float(mscale_all_dim))
562 self.scaling = self.scaling * mscale * mscale
563
564 # DeepSeek is causal by default (decoder-only). Bidirectional
565 # (encoder-only) attention is used by embedding models derived from
566 # this architecture, which set `is_causal=False` on the HF config.
567 # This only applies on the non-MLA path, since the MLA kernels are
568 # causal-only; `ModelConfig.use_mla` already disables MLA for these
569 # models.
570 if getattr(config, "is_causal", True):
571 attn_type = AttentionType.DECODER
572 else:
573 attn_type = AttentionType.ENCODER_ONLY
574 attn_cls = (
575 EncoderOnlyAttention
576 if attn_type == AttentionType.ENCODER_ONLY
577 else Attention
578 )
579 self.attn = attn_cls(
580 self.num_local_heads,
581 self.qk_head_dim,
582 self.scaling,
583 num_kv_heads=self.num_local_heads,
584 cache_config=cache_config,
585 quant_config=quant_config,
586 prefix=f"{prefix}.attn",
587 attn_type=attn_type,
588 )
589
590 def forward(
591 self,
592 positions: torch.Tensor,
593 hidden_states: torch.Tensor,
594 llama_4_scaling: torch.Tensor | None,
595 ) -> torch.Tensor:
596 if self.q_lora_rank is not None:
597 q = self.q_a_proj(hidden_states)[0]
598 q = self.q_a_layernorm(q)
599 q = self.q_b_proj(q)[0].view(-1, self.num_local_heads, self.qk_head_dim)
600 else:
601 q = self.q_proj(hidden_states)[0].view(
602 -1, self.num_local_heads, self.qk_head_dim
603 )
604 q_nope, q_pe = q.split([self.qk_nope_head_dim, self.qk_rope_head_dim], dim=-1)
605 latent_cache = self.kv_a_proj_with_mqa(hidden_states)[0]
606 kv_a, _ = latent_cache.split([self.kv_lora_rank, self.qk_rope_head_dim], dim=-1)
607 latent_cache = latent_cache.unsqueeze(1)
608 kv_a = self.kv_a_layernorm(kv_a)
609 kv = self.kv_b_proj(kv_a)[0]
610 kv = kv.view(-1, self.num_local_heads, self.qk_nope_head_dim + self.v_head_dim)
611 k_nope, v = kv.split([self.qk_nope_head_dim, self.v_head_dim], dim=-1)
612 k_pe = latent_cache[:, :, self.kv_lora_rank :]
613
614 q_pe, k_pe = self.rotary_emb(positions, q_pe, k_pe)
615
616 q[..., self.qk_nope_head_dim :] = q_pe
617 k = torch.empty_like(q)
618 k[..., : self.qk_nope_head_dim] = k_nope
619 k[..., self.qk_nope_head_dim :] = k_pe
620
621 # Apply llama 4 scaling if provided
622 if llama_4_scaling is not None:
623 q *= llama_4_scaling
624
625 # padding value to qk_head_dim for alignment
626 v = torch.nn.functional.pad(
627 v, [0, self.qk_head_dim - self.v_head_dim], value=0
628 ).view(-1, self.num_local_heads * self.qk_head_dim)
629 attn_output = self.attn(q, k, v)
630 attn_output = attn_output.view(-1, self.num_local_heads, self.qk_head_dim)[
631 ..., : self.v_head_dim
632 ].reshape(-1, self.num_local_heads * self.v_head_dim)
633 output, _ = self.o_proj(attn_output)
634 return output
635
636
637 class DeepseekV32IndexerCache(torch.nn.Module, AttentionLayerBase):
638 def __init__(
639 self, head_dim: int, dtype: torch.dtype, prefix: str, cache_config: CacheConfig
640 ):
641 super().__init__()
642 self.kv_cache = torch.tensor([])
643 self.head_dim = head_dim
644 self.prefix = prefix
645 self.cache_config = cache_config
646 self.dtype = dtype
647 compilation_config = get_current_vllm_config().compilation_config
648 if prefix in compilation_config.static_forward_context:
649 raise ValueError(f"Duplicate layer name: {prefix}")
650 compilation_config.static_forward_context[prefix] = self
651
652 def bind_kv_cache(self, kv_cache: torch.Tensor) -> None:
653 # [B, H=1, N, C] -> [B, N, C]: the indexer kernels and
654 # kv_cache_as_quant_view index a 3-D block-major cache.
655 self.kv_cache = kv_cache.squeeze(1)
656
657 def get_kv_cache_spec(self, vllm_config: VllmConfig) -> KVCacheSpec:
658 return MLAAttentionSpec(
659 block_size=self.cache_config.block_size,
660 num_kv_heads=1,
661 max_tp_shards=1,
662 head_size=self.head_dim,
663 dtype=self.dtype,
664 cache_role=SparseCacheRole.INDEXER,
665 ) # Only has one vector instead of K + V
666
667 def forward(self): ...
668
669 @property
670 def uses_shuffled_layout(self) -> bool:
671 """Whether this cache's reader expects the shuffled value layout."""
672 return False
673
674 def get_attn_backend(self) -> type[AttentionBackend]:
675 return DeepseekV32IndexerBackend
676
677
678 class Indexer(nn.Module):
679 def __init__(
680 self,
681 vllm_config: VllmConfig,
682 config: DeepseekV2Config | DeepseekV3Config,
683 hidden_size: int,
684 q_lora_rank: int,
685 quant_config: QuantizationConfig | None,
686 cache_config: CacheConfig | None,
687 topk_indices_buffer: torch.Tensor | None,
688 prefix: str = "",
689 is_inplace_rope: bool = False,
690 ):
691 super().__init__()
692 self.vllm_config = vllm_config
693 self.config = config
694 self.quant_config = quant_config
695 # self.indexer_cfg = config.attn_module_list_cfg[0]["attn_index"]
696 self.topk_tokens = config.index_topk
697 self.n_head = config.index_n_heads # 64
698 self.head_dim = config.index_head_dim # 128
699 self.rope_dim = config.qk_rope_head_dim # 64
700 self.q_lora_rank = q_lora_rank # 1536
701 # no tensor parallel, just replicated
702 self.wq_b = ReplicatedLinear(
703 self.q_lora_rank,
704 self.head_dim * self.n_head,
705 bias=False,
706 quant_config=quant_config,
707 prefix=f"{prefix}.wq_b",
708 )
709 # Fused wk + weights_proj: single GEMM producing [head_dim + n_head].
710 # FP8 wk weights are upcasted to BF16 during loading to maintain fusion.
711 self.wk_weights_proj = MergedColumnParallelLinear(
712 hidden_size,
713 [self.head_dim, self.n_head],
714 bias=False,
715 quant_config=None,
716 disable_tp=True,
717 prefix=f"{prefix}.wk_weights_proj",
718 )
719 self.k_norm = LayerNorm(self.head_dim, eps=1e-6, dtype=torch.float32)
720 self.softmax_scale = self.head_dim**-0.5
721
722 self.scale_fmt = "ue8m0"
723 self.quant_block_size = 128 # TODO: get from config
724 self.topk_indices_buffer = topk_indices_buffer
725
726 # NOTE: (zyongye) we use fp8 naive cache,
727 # where we store value in fp8 and scale in fp32
728 # per self.quant_block_size element
729 assert cache_config is not None
730 self.k_cache = DeepseekV32IndexerCache(
731 head_dim=self.head_dim + self.head_dim // self.quant_block_size * 4,
732 dtype=torch.uint8,
733 prefix=f"{prefix}.k_cache",
734 cache_config=cache_config,
735 )
736 self.max_model_len = vllm_config.model_config.max_model_len
737 self.prefix = prefix
738 from vllm.v1.attention.backends.mla.indexer import get_max_prefill_buffer_size
739
740 self.max_total_seq_len = get_max_prefill_buffer_size(vllm_config)
741 self.indexer_op = SparseAttnIndexer(
742 self.k_cache,
743 self.quant_block_size,
744 self.scale_fmt,
745 self.topk_tokens,
746 self.head_dim,
747 self.max_model_len,
748 self.max_total_seq_len,
749 self.topk_indices_buffer,
750 )
751
752 self.is_inplace_rope = is_inplace_rope
753 self.n_head_scale = self.n_head**-0.5
754 self.use_fused_indexer_q = (
755 current_platform.is_cuda()
756 and self.quant_block_size == self.head_dim
757 and self.head_dim == 128
758 and self.rope_dim == 64
759 and self.scale_fmt is not None
760 )
761
762 def forward(
763 self, hidden_states: torch.Tensor, qr: torch.Tensor, positions, rotary_emb
764 ) -> torch.Tensor:
765 q, _ = self.wq_b(qr)
766 q = q.view(-1, self.n_head, self.head_dim)
767
768 if current_platform.is_rocm() and self.is_inplace_rope:
769 # This path should works on all platform, will remove extra
770 # branches in the future
771 # This fast path relies on rotary_emb mutating q and k inplace.
772 # On ROCm, this is only valid for kernels used as custom ops.
773 # In pytorch-native rope for inductor fusion, rotated q/k tensors
774 # are not mutated inplace but returned as new tensors.
775 # Fused wk + weights_proj: one GEMM, then split
776 kw, _ = self.wk_weights_proj(hidden_states)
777 k = kw[:, : self.head_dim]
778 weights = kw[:, self.head_dim :]
779
780 k = self.k_norm(k)
781
782 rotary_emb(
783 positions, q[..., : self.rope_dim], k[..., : self.rope_dim].unsqueeze(1)
784 )
785 elif self.use_fused_indexer_q and q.dtype == torch.bfloat16:
786 # fused wk + weights_proj: one GEMM, then split
787 kw, _ = self.wk_weights_proj(hidden_states)
788 k = kw[:, : self.head_dim]
789 weights = kw[:, self.head_dim :]
790
791 k = self.k_norm(k)
792 k_pe, k_nope = torch.split(
793 k, [self.rope_dim, self.head_dim - self.rope_dim], dim=-1
794 )
795
796 q_fp8, weights = fused_indexer_q_rope_quant(
797 positions,
798 q,
799 rotary_emb.cos_sin_cache,
800 weights,
801 self.softmax_scale,
802 self.n_head_scale,
803 rotary_emb.is_neox_style,
804 )
805
806 # rotate only the MQA K
807 k_pe = k_pe.unsqueeze(1)
808 q_dummy = torch.empty_like(k_pe)
809 _, k_pe = rotary_emb(positions, q_dummy, k_pe)
810 k_pe = k_pe.reshape(-1, self.rope_dim)
811 k = torch.cat([k_pe, k_nope], dim=-1)
812
813 return self.indexer_op(hidden_states, q_fp8, k, weights)
814 else:
815 q_pe, q_nope = torch.split(
816 q, [self.rope_dim, self.head_dim - self.rope_dim], dim=-1
817 )
818 # Fused wk + weights_proj: one GEMM, then split
819 kw, _ = self.wk_weights_proj(hidden_states)
820 k = kw[:, : self.head_dim]
821 weights = kw[:, self.head_dim :]
822
823 k = self.k_norm(k)
824 k_pe, k_nope = torch.split(
825 k, [self.rope_dim, self.head_dim - self.rope_dim], dim=-1
826 )
827
828 q_pe, k_pe = rotary_emb(positions, q_pe, k_pe.unsqueeze(1))
829 # Note: RoPE (NeoX) can introduce extra leading dimensions during
830 # compilation so we need to reshape back to token-flattened shapes
831 q_pe = q_pe.reshape(-1, self.n_head, self.rope_dim)
832 k_pe = k_pe.reshape(-1, self.rope_dim)
833
834 # `rotary_emb` is shape-preserving; `q_pe` is already
835 # [num_tokens, n_head, rope_dim].
836 q = torch.cat([q_pe, q_nope], dim=-1)
837 # `k_pe` is [num_tokens, rope_dim] (MQA).
838 k = torch.cat([k_pe, k_nope], dim=-1)
839
840 # we only quant q here since k quant is fused with cache insertion
841 q = q.view(-1, self.head_dim)
842 q_fp8, q_scale = per_token_group_quant_fp8(
843 q,
844 self.quant_block_size,
845 column_major_scales=False,
846 use_ue8m0=self.scale_fmt is not None,
847 )
848 q_fp8 = q_fp8.view(-1, self.n_head, self.head_dim)
849 q_scale = q_scale.view(-1, self.n_head)
850
851 weights = weights * q_scale * self.softmax_scale * self.n_head_scale
852
853 return self.indexer_op(hidden_states, q_fp8, k, weights)
854
855
856 def _try_load_fp8_indexer_wk(
857 name, tensor, buf, params_dict, loaded_params, pp_missing_layer_names
858 ):
859 """We fuse the WK and weights_proj projections, but in some checkpoints WK is stored
860 in FP8 with a separate weight_scale_inv, while weights_proj is stored in BF16.
861 Upcasting to BF16 during loading enables the fusion. This function loads the FP8 WK
862 weights and scale, and when both are available, dequantizes to BF16 and stores into
863 the fused wk_weights_proj.weight parameter.
864 """
865 if "indexer.wk." not in name or "wk_weights" in name:
866 return False # Weight is not an isolated WK weight for the indexer, ignore.
867 is_weight = name.endswith(".weight") and tensor.dtype == torch.float8_e4m3fn
868 is_scale = "weight_scale" in name
869 if not is_weight and not is_scale:
870 return False # WK is not in FP8 format, ignore.
871 # Buffer this tensor (weight or scale) until both have arrived.
872 layer_prefix = name.rsplit(".wk.", 1)[0] # e.g. "model.layers.0.self_attn.indexer"
873 fused_name = f"{layer_prefix}.wk_weights_proj.weight"
874 if any(
875 name.startswith(missing_layer_name)
876 for missing_layer_name in pp_missing_layer_names
877 ):
878 return True
879 entry = buf.setdefault(layer_prefix, {})
880 entry["weight" if is_weight else "scale"] = tensor
881 if "weight" not in entry or "scale" not in entry:
882 return True # still waiting for the other param
883
884 # We have both weight and scale: dequantize FP8 to BF16.
885 weight_fp8, scale_inv = entry["weight"], entry["scale"]
886 del buf[layer_prefix]
887 if scale_inv.dtype in (torch.uint8, torch.float8_e8m0fnu):
888 # MXFP8 checkpoints store power-of-two E8M0 scales, decode them to float.
889 scale_inv = scale_inv.view(torch.float8_e8m0fnu).to(torch.float32)
890 if scale_inv.ndim == 1:
891 # Per-channel scale: one scale per row of [out, in]
892 group_shape = GroupShape(1, weight_fp8.shape[1])
893 else:
894 # Derive the block size independently per dim so that per-channel
895 # ([out, 1]), MX ([out, in / 32]) and 2D block ([out / 128, in / 128])
896 # scale layouts are all handled.
897 group_shape = GroupShape(
898 weight_fp8.shape[0] // scale_inv.shape[0],
899 weight_fp8.shape[1] // scale_inv.shape[1],
900 )
901 weight_bf16 = scaled_dequantize(
902 weight_fp8,
903 scale_inv,
904 group_shape=group_shape,
905 out_dtype=torch.bfloat16,
906 )
907
908 # Load the dequantized weight into shard 0 of the fused buffer.
909 param = params_dict[fused_name]
910 param.weight_loader(param, weight_bf16, 0)
911 loaded_params.add(fused_name)
912 return True
913
914
915 def _min_latency_fused_qkv_a_proj_impl(
916 input_: torch.Tensor,
917 weight: torch.Tensor,
918 ) -> torch.Tensor:
919 """Dynamically run min-latency gemm if num_tokens <= 16.
920 This must be wrapped in a custom op because our torch.compile integration
921 does not support runtime dispatching on num_tokens.
922 """
923 num_tokens = input_.shape[0]
924 if 0 < num_tokens <= 16:
925 output = torch.empty(
926 num_tokens,
927 weight.shape[0],
928 dtype=torch.bfloat16,
929 device=input_.device,
930 )
931 ops.dsv3_fused_a_gemm(output, input_, weight.T)
932 return output
933 else:
934 return torch.nn.functional.linear(input_, weight)
935
936
937 def _min_latency_fused_qkv_a_proj_fake(
938 input_: torch.Tensor,
939 weight: torch.Tensor,
940 ) -> torch.Tensor:
941 return input_.new_empty(input_.shape[0], weight.shape[0])
942
943
944 direct_register_custom_op(
945 op_name="min_latency_fused_qkv_a_proj",
946 op_func=_min_latency_fused_qkv_a_proj_impl,
947 mutates_args=[],
948 fake_impl=_min_latency_fused_qkv_a_proj_fake,
949 )
950
951
952 class DeepSeekV2FusedQkvAProjLinear(MergedColumnParallelLinear):
953 def __init__(
954 self,
955 input_size: int,
956 output_size: list[int],
957 quant_config: QuantizationConfig | None = None,
958 prefix: str = "",
959 ):
960 super().__init__(
961 input_size,
962 output_size,
963 bias=False,
964 quant_config=quant_config,
965 disable_tp=True,
966 prefix=prefix,
967 )
968
969 # Check if the DeepSeek V3 fused A GEMM kernel can be used.
970 # This kernel supports PDL and is optimized for low batch size.
971 self._use_min_latency_gemm = (
972 hasattr(self, "weight")
973 and self.weight.dtype == torch.bfloat16
974 and self.weight.shape[0] == 2112
975 and self.weight.shape[1] == 7168
976 and current_platform.is_cuda()
977 and (
978 current_platform.is_device_capability(90)
979 or current_platform.is_device_capability_family(100)
980 )
981 )
982
983 def forward(
984 self,
985 input_,
986 ) -> torch.Tensor | tuple[torch.Tensor, torch.nn.Parameter | None]:
987 if self._use_min_latency_gemm:
988 output = torch.ops.vllm.min_latency_fused_qkv_a_proj(input_, self.weight)
989 if not self.return_bias:
990 return output
991 output_bias = self.bias if self.skip_bias_add else None
992 return output, output_bias
993 else:
994 # Fallback to the standard forward method when
995 # the fused A GEMM kernel cannot be used.
996 return super().forward(input_)
997
998
999 class DeepseekV2MLAAttention(nn.Module):
1000 """Main reference: DeepseekV2 paper, and FlashInfer Implementation
1001 (https://arxiv.org/abs/2405.04434 and https://github.com/flashinfer-ai/flashinfer/pull/551).
1002
1003 For more info see MLACommonImpl in:
1004 vllm/v1/attention/backends/mla/utils.py
1005 """
1006
1007 def __init__(
1008 self,
1009 vllm_config: VllmConfig,
1010 config: DeepseekV2Config | DeepseekV3Config,
1011 hidden_size: int,
1012 num_heads: int,
1013 qk_nope_head_dim: int,
1014 qk_rope_head_dim: int,
1015 v_head_dim: int,
1016 q_lora_rank: int | None,
1017 kv_lora_rank: int,
1018 max_position_embeddings: int = 8192,
1019 cache_config: CacheConfig | None = None,
1020 quant_config: QuantizationConfig | None = None,
1021 prefix: str = "",
1022 topk_indices_buffer: torch.Tensor | None = None,
1023 index_group_builder: SparseMLAIndexGroupBuilder | None = None,
1024 input_size: int | None = None,
1025 reduce_results: bool = True,
1026 non_causal_multi_token_decode: bool = False,
1027 ) -> None:
1028 super().__init__()
1029 self.hidden_size = hidden_size
1030 self.qk_nope_head_dim = qk_nope_head_dim
1031 self.qk_rope_head_dim = qk_rope_head_dim
1032 self.qk_head_dim = qk_nope_head_dim + qk_rope_head_dim
1033 self.v_head_dim = v_head_dim
1034
1035 self.q_lora_rank = q_lora_rank
1036 self.kv_lora_rank = kv_lora_rank
1037
1038 self.num_heads = num_heads
1039 tp_size = get_tensor_model_parallel_world_size()
1040 assert num_heads % tp_size == 0
1041 self.num_local_heads = num_heads // tp_size
1042
1043 self.scaling = self.qk_head_dim**-0.5
1044 self.max_position_embeddings = max_position_embeddings
1045
1046 # Use input_size for projection input dimensions if provided,
1047 # otherwise default to hidden_size (used in Eagle3 Deepseek with MLA)
1048 proj_input_size = input_size if input_size is not None else self.hidden_size
1049
1050 if self.q_lora_rank is not None:
1051 self.fused_qkv_a_proj = DeepSeekV2FusedQkvAProjLinear(
1052 proj_input_size,
1053 [self.q_lora_rank, self.kv_lora_rank + self.qk_rope_head_dim],
1054 quant_config=quant_config,
1055 prefix=f"{prefix}.fused_qkv_a_proj",
1056 )
1057 else:
1058 self.kv_a_proj_with_mqa = ReplicatedLinear(
1059 proj_input_size,
1060 self.kv_lora_rank + self.qk_rope_head_dim,
1061 bias=False,
1062 quant_config=quant_config,
1063 prefix=f"{prefix}.kv_a_proj_with_mqa",
1064 )
1065
1066 # The env var predates the config field and still wins if set explicitly.
1067 qrep_requested = (
1068 envs.VLLM_DCP_Q_REPLICATE
1069 if envs.is_set("VLLM_DCP_Q_REPLICATE")
1070 else bool(vllm_config.parallel_config.dcp_q_replicate)
1071 )
1072 qrep_enabled = (
1073 qrep_requested
1074 and vllm_config.parallel_config.decode_context_parallel_size > 1
1075 and vllm_config.parallel_config.prefill_context_parallel_size <= 1
1076 )
1077 q_proj_cls = (
1078 DCPGroupColumnParallelLinear if qrep_enabled else ColumnParallelLinear
1079 )
1080 if self.q_lora_rank is not None:
1081 self.q_a_layernorm = RMSNorm(self.q_lora_rank, eps=config.rms_norm_eps)
1082 self.q_b_proj = q_proj_cls(
1083 self.q_lora_rank,
1084 self.num_heads * self.qk_head_dim,
1085 bias=False,
1086 quant_config=quant_config,
1087 prefix=f"{prefix}.q_b_proj",
1088 )
1089 else:
1090 self.q_proj = q_proj_cls(
1091 proj_input_size,
1092 self.num_heads * self.qk_head_dim,
1093 bias=False,
1094 quant_config=quant_config,
1095 prefix=f"{prefix}.q_proj",
1096 )
1097 self.kv_a_layernorm = RMSNorm(self.kv_lora_rank, eps=config.rms_norm_eps)
1098 self.kv_b_proj = ColumnParallelLinear(
1099 self.kv_lora_rank,
1100 self.num_heads * (self.qk_nope_head_dim + self.v_head_dim),
1101 bias=False,
1102 quant_config=quant_config,
1103 prefix=f"{prefix}.kv_b_proj",
1104 )
1105 self.o_proj = RowParallelLinear(
1106 self.num_heads * self.v_head_dim,
1107 self.hidden_size,
1108 bias=False,
1109 reduce_results=reduce_results,
1110 quant_config=quant_config,
1111 prefix=f"{prefix}.o_proj",
1112 )
1113
1114 if config.rope_parameters["rope_type"] != "default":
1115 config.rope_parameters["rope_type"] = (
1116 "deepseek_llama_scaling"
1117 if config.rope_parameters.get("attention_factor") == 1.0
1118 else "deepseek_yarn"
1119 )
1120
1121 self.rotary_emb = get_rope(
1122 qk_rope_head_dim,
1123 max_position=max_position_embeddings,
1124 rope_parameters=config.rope_parameters,
1125 is_neox_style=False,
1126 )
1127
1128 if (
1129 config.rope_parameters["rope_type"] != "default"
1130 and config.rope_parameters["rope_type"] == "deepseek_yarn"
1131 ):
1132 mscale_all_dim = config.rope_parameters.get("mscale_all_dim", False)
1133 scaling_factor = config.rope_parameters["factor"]
1134 mscale = yarn_get_mscale(scaling_factor, float(mscale_all_dim))
1135 self.scaling = self.scaling * mscale * mscale
1136
1137 self.is_v32 = hasattr(config, "index_topk")
1138
1139 # IndexCache config
1140 # Refer: https://arxiv.org/abs/2603.12201 for more details.
1141 _skip_topk = False
1142 is_mtp_layer = False
1143 if self.is_v32:
1144 _index_topk_freq = getattr(config, "index_topk_freq", 1)
1145 _index_topk_pattern = getattr(config, "index_topk_pattern", None)
1146 _index_skip_topk_offset = getattr(config, "index_skip_topk_offset", 2)
1147 layer_id = extract_layer_index(prefix)
1148
1149 if _index_topk_pattern is None:
1150 _skip_topk = (
1151 max(layer_id - _index_skip_topk_offset + 1, 0) % _index_topk_freq
1152 != 0
1153 )
1154 elif 0 <= layer_id < len(_index_topk_pattern):
1155 _skip_topk = _index_topk_pattern[layer_id] == "S"
1156
1157 # The skip pattern only governs backbone layers. MTP/nextn
1158 # layers (layer_id >= num_hidden_layers) always build a full
1159 # indexer: they compute indices at draft step 0 and toggle
1160 # at runtime via set_skip_topk
1161 # (index_share_for_mtp_iteration).
1162 _num_hidden_layers = getattr(config, "num_hidden_layers", None)
1163 is_mtp_layer = (
1164 _num_hidden_layers is not None and layer_id >= _num_hidden_layers
1165 )
1166
1167 if self.is_v32 and (not _skip_topk or is_mtp_layer):
1168 assert q_lora_rank is not None
1169 assert cache_config is not None
1170 self.indexer_rope_emb: nn.Module | None
1171 self.indexer: Indexer | None
1172 self.indexer_rope_emb = get_rope(
1173 qk_rope_head_dim,
1174 max_position=max_position_embeddings,
1175 rope_parameters=config.rope_parameters,
1176 is_neox_style=not getattr(config, "indexer_rope_interleave", False),
1177 )
1178 self.indexer = Indexer(
1179 vllm_config,
1180 config,
1181 hidden_size,
1182 q_lora_rank,
1183 quant_config,
1184 cache_config,
1185 topk_indices_buffer,
1186 f"{prefix}.indexer",
1187 is_inplace_rope=self.indexer_rope_emb.enabled(),
1188 )
1189 else:
1190 self.indexer_rope_emb = None
1191 self.indexer = None
1192
1193 mla_modules = MLAModules(
1194 kv_a_layernorm=self.kv_a_layernorm,
1195 kv_b_proj=self.kv_b_proj,
1196 rotary_emb=self.rotary_emb,
1197 o_proj=self.o_proj,
1198 fused_qkv_a_proj=self.fused_qkv_a_proj
1199 if self.q_lora_rank is not None
1200 else None,
1201 kv_a_proj_with_mqa=self.kv_a_proj_with_mqa
1202 if self.q_lora_rank is None
1203 else None,
1204 q_a_layernorm=self.q_a_layernorm if self.q_lora_rank is not None else None,
1205 q_b_proj=self.q_b_proj if self.q_lora_rank is not None else None,
1206 q_proj=self.q_proj if self.q_lora_rank is None else None,
1207 indexer=self.indexer,
1208 indexer_rotary_emb=self.indexer_rope_emb,
1209 is_sparse=self.is_v32,
1210 topk_indices_buffer=topk_indices_buffer,
1211 index_group_builder=index_group_builder,
1212 )
1213
1214 self.mla_attn = MultiHeadLatentAttentionWrapper(
1215 self.hidden_size,
1216 self.num_local_heads,
1217 self.scaling,
1218 self.qk_nope_head_dim,
1219 self.qk_rope_head_dim,
1220 self.v_head_dim,
1221 self.q_lora_rank,
1222 self.kv_lora_rank,
1223 mla_modules,
1224 cache_config,
1225 quant_config,
1226 prefix,
1227 # MTP layers must never start with skip_topk=True: their indexer
1228 # computes indices at draft step 0, and the runtime toggle
1229 # (set_skip_topk, index_share_for_mtp_iteration) only exists in
1230 # the V1 proposer. A frozen True would leave the draft reading a
1231 # never-written topk buffer.
1232 skip_topk=_skip_topk and not is_mtp_layer,
1233 non_causal_multi_token_decode=non_causal_multi_token_decode,
1234 # Do not skip scoring for MTP layers: their top-k buffer may be
1235 # reused by later draft iterations through index sharing.
1236 allow_short_prefill_indexer_scoring_skip=not is_mtp_layer,
1237 )
1238
1239 def forward(
1240 self,
1241 positions: torch.Tensor,
1242 hidden_states: torch.Tensor,
1243 llama_4_scaling: torch.Tensor | None,
1244 ) -> torch.Tensor:
1245 return self.mla_attn(positions, hidden_states, llama_4_scaling)
1246
1247
1248 class DeepseekV2DecoderLayer(nn.Module):
1249 def __init__(
1250 self,
1251 vllm_config: VllmConfig,
1252 prefix: str,
1253 config: DeepseekV2Config | None = None,
1254 topk_indices_buffer: torch.Tensor | None = None,
1255 index_group_builder: SparseMLAIndexGroupBuilder | None = None,
1256 ) -> None:
1257 super().__init__()
1258
1259 if config is None:
1260 config = vllm_config.model_config.hf_config
1261 model_config = vllm_config.model_config
1262 cache_config = vllm_config.cache_config
1263 quant_config = vllm_config.quant_config
1264 parallel_config = vllm_config.parallel_config
1265
1266 self.hidden_size = config.hidden_size
1267 max_position_embeddings = getattr(config, "max_position_embeddings", 8192)
1268 moe_layer_freq = getattr(config, "moe_layer_freq", 1)
1269 # DecoderLayers are created with `make_layers` which passes the prefix
1270 # with the layer's index.
1271 layer_idx = int(prefix.split(sep=".")[-1])
1272 self.layer_idx = layer_idx
1273
1274 # verify MLA attention specific fields
1275 qk_nope_head_dim = getattr(config, "qk_nope_head_dim", 0)
1276 qk_rope_head_dim = getattr(config, "qk_rope_head_dim", 0)
1277 v_head_dim = getattr(config, "v_head_dim", 0)
1278 kv_lora_rank = getattr(config, "kv_lora_rank", 0)
1279 use_mha = config.model_type == "deepseek" or all(
1280 dim == 0 for dim in (qk_nope_head_dim, qk_rope_head_dim)
1281 )
1282
1283 self.use_mha = use_mha
1284
1285 attn_cls: typing.Any
1286 if use_mha:
1287 attn_cls = DeepseekAttention
1288 elif model_config.use_mla:
1289 attn_cls = DeepseekV2MLAAttention
1290 else:
1291 attn_cls = DeepseekV2Attention
1292 is_moe_layer = (
1293 config.n_routed_experts is not None
1294 and layer_idx >= config.first_k_dense_replace
1295 and layer_idx % moe_layer_freq == 0
1296 )
1297 # TODO(wentao): enable SP MoE with PP after the PP boundary logic can safely
1298 # send/receive sequence-parallel hidden_states across stages.
1299 self.use_sequence_parallel_moe = (
1300 parallel_config.use_sequence_parallel_moe
1301 and parallel_config.pipeline_parallel_size == 1
1302 and is_moe_layer
1303 )
1304 attn_kwargs = (
1305 {"index_group_builder": index_group_builder}
1306 if attn_cls is DeepseekV2MLAAttention
1307 else {}
1308 )
1309 self.self_attn = attn_cls(
1310 vllm_config=vllm_config,
1311 config=config,
1312 hidden_size=self.hidden_size,
1313 num_heads=config.num_attention_heads,
1314 qk_nope_head_dim=qk_nope_head_dim,
1315 qk_rope_head_dim=qk_rope_head_dim,
1316 v_head_dim=v_head_dim,
1317 q_lora_rank=config.q_lora_rank if hasattr(config, "q_lora_rank") else None,
1318 kv_lora_rank=kv_lora_rank,
1319 max_position_embeddings=max_position_embeddings,
1320 cache_config=cache_config,
1321 quant_config=quant_config,
1322 prefix=f"{prefix}.self_attn",
1323 topk_indices_buffer=topk_indices_buffer,
1324 reduce_results=not self.use_sequence_parallel_moe,
1325 **attn_kwargs,
1326 )
1327
1328 if is_moe_layer:
1329 self.mlp = DeepseekV2MoE(
1330 config=config,
1331 parallel_config=parallel_config,
1332 quant_config=quant_config,
1333 prefix=f"{prefix}.mlp",
1334 # aiter applies routed_scaling_factor internally
1335 apply_routed_scale_to_output=not rocm_aiter_ops.is_fused_moe_enabled(),
1336 )
1337 else:
1338 self.mlp = DeepseekV2MLP(
1339 hidden_size=config.hidden_size,
1340 intermediate_size=config.intermediate_size,
1341 hidden_act=config.hidden_act,
1342 quant_config=quant_config,
1343 prefix=f"{prefix}.mlp",
1344 )
1345 self.input_layernorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
1346 self.post_attention_layernorm = RMSNorm(
1347 config.hidden_size, eps=config.rms_norm_eps
1348 )
1349 self.routed_scaling_factor = getattr(config, "routed_scaling_factor", 1.0)
1350
1351 def forward(
1352 self,
1353 positions: torch.Tensor,
1354 hidden_states: torch.Tensor,
1355 residual: torch.Tensor | None,
1356 llama_4_scaling: torch.Tensor | None = None,
1357 ) -> torch.Tensor:
1358 full_num_tokens = positions.shape[0]
1359 input_is_sequence_parallel = (
1360 self.use_sequence_parallel_moe
1361 and residual is not None
1362 and hidden_states.shape[0] != full_num_tokens
1363 )
1364
1365 # Self Attention
1366 if residual is None:
1367 residual = hidden_states
1368 hidden_states = self.input_layernorm(hidden_states)
1369 else:
1370 hidden_states, residual = self.input_layernorm(hidden_states, residual)
1371
1372 if input_is_sequence_parallel:
1373 hidden_states = tensor_model_parallel_all_gather(hidden_states, 0)
1374 hidden_states = hidden_states[:full_num_tokens]
1375
1376 if self.use_mha:
1377 hidden_states = self.self_attn(positions, hidden_states)
1378 else:
1379 hidden_states = self.self_attn(positions, hidden_states, llama_4_scaling)
1380
1381 if (
1382 not isinstance(self.self_attn, DeepseekAttention)
1383 and hidden_states.dtype == torch.float16
1384 ):
1385 # Fix FP16 overflow
1386 # We scale both hidden_states and residual before
1387 # rmsnorm, and rmsnorm result would not affect by scale.
1388 hidden_states *= 1.0 / self.routed_scaling_factor
1389 if self.layer_idx == 0:
1390 # The residual is shared by all layers, we only scale it on
1391 # first layer.
1392 residual *= 1.0 / self.routed_scaling_factor
1393
1394 if self.use_sequence_parallel_moe:
1395 tp_world_size = get_tensor_model_parallel_world_size()
1396 # small trick using minus, eg. -17 % 8 = 7
1397 sp_pad = (-hidden_states.shape[0]) % tp_world_size
1398 # pad if not divisible by world size
1399 hidden_states = torch.nn.functional.pad(hidden_states, (0, 0, 0, sp_pad))
1400 hidden_states = tensor_model_parallel_reduce_scatter(hidden_states, 0)
1401 if not input_is_sequence_parallel:
1402 residual = sequence_parallel_chunk(residual)
1403
1404 # Fully Connected
1405 hidden_states, residual = self.post_attention_layernorm(hidden_states, residual)
1406 if self.use_sequence_parallel_moe:
1407 hidden_states = self.mlp(
1408 hidden_states,
1409 already_sequence_parallel=True,
1410 )
1411 else:
1412 hidden_states = self.mlp(hidden_states)
1413
1414 if isinstance(self.mlp, DeepseekV2MLP) and hidden_states.dtype == torch.float16:
1415 # Fix FP16 overflow
1416 # Scaling the DeepseekV2MLP output, it is the input of
1417 # input_layernorm of next decoder layer.
1418 # The scaling of DeepseekV2MOE output would be done in the forward
1419 # of DeepseekV2MOE
1420 hidden_states *= 1.0 / self.routed_scaling_factor
1421
1422 return hidden_states, residual
1423
1424
1425 @support_torch_compile
1426 class DeepseekV2Model(nn.Module):
1427 fall_back_to_pt_during_load = False
1428
1429 def __init__(self, *, vllm_config: VllmConfig, prefix: str = ""):
1430 super().__init__()
1431
1432 config = vllm_config.model_config.hf_config
1433 quant_config = vllm_config.quant_config
1434 self.config = config
1435 self.device = current_platform.device_type
1436 self.hidden_size = config.hidden_size
1437 self.vocab_size = config.vocab_size
1438 self.is_v32 = hasattr(config, "index_topk")
1439 if self.is_v32:
1440 topk_tokens = config.index_topk
1441 topk_indices_buffer = torch.empty(
1442 vllm_config.scheduler_config.max_num_batched_tokens,
1443 topk_tokens,
1444 dtype=torch.int32,
1445 device=self.device,
1446 )
1447 else:
1448 topk_indices_buffer = None
1449 index_group_builder = (
1450 SparseMLAIndexGroupBuilder(
1451 topk_indices_buffer,
1452 get_sparse_mla_index_group_max_rows(vllm_config),
1453 )
1454 if topk_indices_buffer is not None
1455 else None
1456 )
1457
1458 if get_pp_group().is_first_rank:
1459 self.embed_tokens = VocabParallelEmbedding(
1460 config.vocab_size,
1461 self.hidden_size,
1462 quant_config=quant_config,
1463 prefix=f"{prefix}.embed_tokens",
1464 )
1465 else:
1466 self.embed_tokens = PPMissingLayer()
1467 self.start_layer, self.end_layer, self.layers = make_layers(
1468 config.num_hidden_layers,
1469 lambda prefix: DeepseekV2DecoderLayer(
1470 vllm_config=vllm_config,
1471 prefix=prefix,
1472 topk_indices_buffer=topk_indices_buffer,
1473 index_group_builder=index_group_builder,
1474 ),
1475 prefix=f"{prefix}.layers",
1476 )
1477
1478 self.is_fused_shared_expert_enabled = is_model_fused_shared_expert_compatible(
1479 self.layers,
1480 DeepseekV2MoE,
1481 "mlp",
1482 )
1483
1484 if get_pp_group().is_last_rank:
1485 self.norm = RMSNorm(self.hidden_size, eps=config.rms_norm_eps)
1486 else:
1487 self.norm = PPMissingLayer()
1488 self.make_empty_intermediate_tensors = make_empty_intermediate_tensors_factory(
1489 ["hidden_states", "residual"], self.hidden_size
1490 )
1491
1492 self.aux_hidden_state_layers = tuple[int, ...]()
1493
1494 # Needed by load_weights
1495 qk_nope_head_dim = getattr(config, "qk_nope_head_dim", 0)
1496 qk_rope_head_dim = getattr(config, "qk_rope_head_dim", 0)
1497 self.use_mha = config.model_type == "deepseek" or all(
1498 dim == 0 for dim in (qk_nope_head_dim, qk_rope_head_dim)
1499 )
1500 self.num_redundant_experts = (
1501 vllm_config.parallel_config.eplb_config.num_redundant_experts
1502 )
1503
1504 def embed_input_ids(self, input_ids: torch.Tensor) -> torch.Tensor:
1505 return self.embed_tokens(input_ids)
1506
1507 def forward(
1508 self,
1509 input_ids: torch.Tensor | None,
1510 positions: torch.Tensor,
1511 intermediate_tensors: IntermediateTensors | None,
1512 inputs_embeds: torch.Tensor | None = None,
1513 ) -> torch.Tensor | IntermediateTensors:
1514 if get_pp_group().is_first_rank:
1515 if inputs_embeds is not None:
1516 hidden_states = inputs_embeds
1517 else:
1518 if input_ids is None:
1519 raise ValueError(
1520 "Either input_ids or inputs_embeds must be provided "
1521 "to DeepseekV2Model.forward"
1522 )
1523 hidden_states = self.embed_input_ids(input_ids)
1524 residual = None
1525 else:
1526 assert intermediate_tensors is not None
1527 hidden_states = intermediate_tensors["hidden_states"]
1528 residual = intermediate_tensors["residual"]
1529
1530 # Compute llama 4 scaling once per forward pass if enabled
1531 llama_4_scaling_config = getattr(self.config, "llama_4_scaling", None)
1532 llama_4_scaling: torch.Tensor | None
1533 if llama_4_scaling_config is not None:
1534 llama_4_scaling = _get_llama_4_scaling(
1535 original_max_position_embeddings=llama_4_scaling_config[
1536 "original_max_position_embeddings"
1537 ],
1538 scaling_beta=llama_4_scaling_config["beta"],
1539 positions=positions,
1540 )
1541 else:
1542 llama_4_scaling = None
1543
1544 aux_hidden_states = []
1545 for idx, layer in enumerate(
1546 islice(self.layers, self.start_layer, self.end_layer),
1547 start=self.start_layer,
1548 ):
1549 # all gather if we need to use the whole states
1550 if (
1551 hidden_states.shape[0] != positions.shape[0]
1552 and not layer.use_sequence_parallel_moe
1553 ):
1554 combined_states = torch.cat([hidden_states, residual], dim=-1)
1555 combined_states = tensor_model_parallel_all_gather(combined_states, 0)
1556 combined_states = combined_states[: positions.shape[0]]
1557 hidden_states, residual = combined_states.split(
1558 [self.hidden_size, self.hidden_size], dim=-1
1559 )
1560 if idx in self.aux_hidden_state_layers:
1561 aux_hidden_state = hidden_states + residual
1562 if aux_hidden_state.shape[0] != positions.shape[0]:
1563 aux_hidden_state = tensor_model_parallel_all_gather(
1564 aux_hidden_state, 0
1565 )
1566 aux_hidden_state = aux_hidden_state[: positions.shape[0]]
1567 aux_hidden_states.append(aux_hidden_state)
1568 hidden_states, residual = layer(
1569 positions, hidden_states, residual, llama_4_scaling
1570 )
1571
1572 if not get_pp_group().is_last_rank:
1573 return IntermediateTensors(
1574 {"hidden_states": hidden_states, "residual": residual}
1575 )
1576
1577 if hidden_states.shape[0] != positions.shape[0]:
1578 combined_states = torch.cat([hidden_states, residual], dim=-1)
1579 combined_states = tensor_model_parallel_all_gather(combined_states, 0)
1580 combined_states = combined_states[: positions.shape[0]]
1581 hidden_states, residual = combined_states.split(
1582 [self.hidden_size, self.hidden_size], dim=-1
1583 )
1584
1585 if self.end_layer in self.aux_hidden_state_layers:
1586 aux_hidden_states.append(hidden_states + residual)
1587
1588 hidden_states, _ = self.norm(hidden_states, residual)
1589 if len(aux_hidden_states) > 0:
1590 return hidden_states, aux_hidden_states
1591 return hidden_states
1592
1593 def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
1594 stacked_params_mapping: list[tuple[str, str, int | str]] = [
1595 # (param_name, shard_name, shard_id)
1596 ("gate_up_proj", "gate_proj", 0),
1597 ("gate_up_proj", "up_proj", 1),
1598 ]
1599 mla_params_mapping = [
1600 ("fused_qkv_a_proj", "q_a_proj", 0),
1601 ("fused_qkv_a_proj", "kv_a_proj_with_mqa", 1),
1602 ]
1603 mha_params_mapping = [
1604 ("qkv_proj", "q_proj", "q"),
1605 ("qkv_proj", "k_proj", "k"),
1606 ("qkv_proj", "v_proj", "v"),
1607 ]
1608 # Fused indexer wk + weights_proj (shard 0 = wk, shard 1 = weights_proj)
1609 _pending_wk_fp8 = getattr(self, "_pending_indexer_wk_fp8", None)
1610 if _pending_wk_fp8 is None:
1611 self._pending_indexer_wk_fp8 = _pending_wk_fp8 = {}
1612
1613 indexer_fused_mapping = [
1614 ("wk_weights_proj", "wk", 0),
1615 ("wk_weights_proj", "weights_proj", 1),
1616 ]
1617 stacked_params_mapping.extend(indexer_fused_mapping)
1618
1619 if self.use_mha:
1620 stacked_params_mapping.extend(mha_params_mapping)
1621 else:
1622 stacked_params_mapping.extend(mla_params_mapping)
1623
1624 # Params for weights, fp8 weight scales, fp8 activation scales
1625 # (param_name, weight_name, expert_id, shard_id)
1626 expert_params_mapping = fused_moe_make_expert_params_mapping(
1627 self,
1628 ckpt_gate_proj_name="gate_proj",
1629 ckpt_down_proj_name="down_proj",
1630 ckpt_up_proj_name="up_proj",
1631 num_experts=self.config.n_routed_experts
1632 + (
1633 self.config.n_shared_experts
1634 if self.is_fused_shared_expert_enabled
1635 else 0
1636 ),
1637 num_redundant_experts=self.num_redundant_experts,
1638 )
1639
1640 pp_missing_layer_names = get_pp_missing_layer_names(self)
1641 params_dict = dict(self.named_parameters())
1642 loaded_params: set[str] = set()
1643 # With index_topk_freq>1 only some layers build an indexer, yet the
1644 # checkpoint ships indexer weights for all of them; track the built ones.
1645 indexer_present_prefixes = {
1646 n.rsplit(".indexer.", 1)[0] for n in params_dict if ".indexer." in n
1647 }
1648 for name, loaded_weight in weights:
1649 if "rotary_emb.inv_freq" in name:
1650 continue
1651
1652 spec_layer = get_spec_layer_idx_from_weight_name(self.config, name)
1653 if spec_layer is not None:
1654 continue # skip spec decode layers for main model
1655
1656 if ".indexer." in name and (
1657 name.rsplit(".indexer.", 1)[0] not in indexer_present_prefixes
1658 ):
1659 continue # this layer has no indexer; drop its checkpoint weights
1660
1661 is_fusion_moe_shared_experts_layer = (
1662 self.is_fused_shared_expert_enabled and ("mlp.shared_experts" in name)
1663 )
1664
1665 if _try_load_fp8_indexer_wk(
1666 name,
1667 loaded_weight,
1668 _pending_wk_fp8,
1669 params_dict,
1670 loaded_params,
1671 pp_missing_layer_names,
1672 ):
1673 continue
1674
1675 for param_name, weight_name, shard_id in stacked_params_mapping:
1676 # Skip non-stacked layers and experts (experts handled below).
1677 if weight_name not in name:
1678 continue
1679 # We have mlp.experts[0].gate_proj in the checkpoint.
1680 # Since we handle the experts below in expert_params_mapping,
1681 # we need to skip here BEFORE we update the name, otherwise
1682 # name will be updated to mlp.experts[0].gate_up_proj, which
1683 # will then be updated below in expert_params_mapping
1684 # for mlp.experts[0].gate_gate_up_proj, which breaks load.
1685 if ("mlp.experts." in name) and name not in params_dict:
1686 continue
1687 if is_fusion_moe_shared_experts_layer:
1688 continue
1689 name_mapped = name.replace(weight_name, param_name)
1690
1691 # QKV fusion is optional, fall back to normal
1692 # weight loading if it's not enabled
1693 # if go with fusion option, then update name
1694 if (
1695 param_name == "fused_qkv_a_proj"
1696 ) and name_mapped not in params_dict:
1697 continue
1698 else:
1699 name = name_mapped
1700 # Skip loading extra bias for GPTQ models.
1701 if name.endswith(".bias") and name not in params_dict:
1702 continue
1703
1704 if is_pp_missing_parameter(name, self):
1705 continue
1706
1707 param = params_dict[name]
1708 weight_loader = param.weight_loader
1709 weight_loader(param, loaded_weight, shard_id)
1710 break
1711 else:
1712 is_expert_weight = False
1713
1714 # Special handling: when AITER fusion_shared_experts is enabled,
1715 # checkpoints may provide a single widened shared_experts tensor
1716 # without explicit expert indices
1717 # (e.g. ...mlp.shared_experts.gate_proj.weight).
1718 # For models with multiple shared experts, split that tensor
1719 # evenly into per-shared-expert slices and load them into
1720 # appended expert slots mlp.experts.{n_routed_experts + j}.*
1721 # accordingly.
1722 num_chunks = 1
1723 if is_fusion_moe_shared_experts_layer:
1724 num_chunks = getattr(self.config, "n_shared_experts", 1) or 1
1725 # Determine split axis based on op type
1726 # gate/up: ColumnParallel → split along dim 0
1727 # down: RowParallel → split along dim 1
1728 split_dim = (
1729 1
1730 if ("down_proj.weight" in name and loaded_weight.ndim > 1)
1731 else 0
1732 )
1733 total = loaded_weight.shape[split_dim]
1734 assert total % num_chunks == 0, (
1735 f"Shared expert weight dim {total} "
1736 f"not divisible by num_chunks {num_chunks}"
1737 )
1738 chunk_size = total // num_chunks
1739
1740 for j in range(num_chunks):
1741 chunk_name = name
1742 weight_to_load = loaded_weight
1743
1744 if is_fusion_moe_shared_experts_layer:
1745 chunk_slice = slice(j * chunk_size, (j + 1) * chunk_size)
1746 if loaded_weight.ndim == 1:
1747 weight_to_load = loaded_weight[chunk_slice]
1748 elif split_dim == 0:
1749 weight_to_load = loaded_weight[chunk_slice, :]
1750 else:
1751 weight_to_load = loaded_weight[:, chunk_slice]
1752 # Synthesize an expert-style name so expert mapping
1753 # can route it
1754 chunk_name = name.replace(
1755 "mlp.shared_experts",
1756 f"mlp.experts.{self.config.n_routed_experts + j}",
1757 )
1758
1759 # Use expert_params_mapping to locate the destination
1760 # param and delegate to its expert-aware weight_loader
1761 # with expert_id.
1762 for mapping in expert_params_mapping:
1763 param_name, weight_name, expert_id, expert_shard_id = mapping
1764 if weight_name not in chunk_name:
1765 continue
1766
1767 # Anyway, this is an expert weight and should not be
1768 # attempted to load as other weights later
1769 is_expert_weight = True
1770
1771 # Do not modify `name` since the loop may continue here
1772 # Instead, create a new variable
1773 name_mapped = chunk_name.replace(weight_name, param_name)
1774
1775 if is_pp_missing_parameter(name_mapped, self):
1776 continue
1777
1778 param = params_dict[name_mapped]
1779 # We should ask the weight loader to return success or
1780 # not here since otherwise we may skip experts with
1781 # other available replicas.
1782 weight_loader = typing.cast(
1783 Callable[..., bool], param.weight_loader
1784 )
1785 success = weight_loader(
1786 param,
1787 weight_to_load,
1788 name_mapped,
1789 shard_id=expert_shard_id,
1790 expert_id=expert_id,
1791 return_success=True,
1792 )
1793 if success:
1794 if not is_fusion_moe_shared_experts_layer:
1795 name = name_mapped
1796 else:
1797 loaded_params.add(name_mapped)
1798 break
1799 else:
1800 if is_expert_weight:
1801 # We've checked that this is an expert weight
1802 # However it's not mapped locally to this rank
1803 # So we simply skip it
1804 continue
1805
1806 # Skip loading extra bias for GPTQ models.
1807 if name.endswith(".bias") and name not in params_dict:
1808 continue
1809
1810 # Remapping the name of FP8 kv-scale.
1811 remapped_name = maybe_remap_kv_scale_name(name, params_dict)
1812 if remapped_name is None:
1813 continue
1814 name = remapped_name
1815
1816 if is_pp_missing_parameter(name, self):
1817 continue
1818
1819 param = params_dict[name]
1820 weight_loader = getattr(
1821 param, "weight_loader", default_weight_loader
1822 )
1823 weight_loader(param, loaded_weight)
1824 if name is not None and not is_fusion_moe_shared_experts_layer:
1825 loaded_params.add(name)
1826
1827 return loaded_params
1828
1829
1830 class DeepseekV2MixtureOfExperts(MixtureOfExperts):
1831 moe_mlp_layers: list[DeepseekV2MoE]
1832 """
1833 List of MoE MLP layers in the model.
1834 """
1835
1836 def extract_moe_parameters(self, example_moe: DeepseekV2MoE | None):
1837 if example_moe is None:
1838 self.num_moe_layers = 0
1839 self.num_expert_groups = 0
1840 self.num_logical_experts = 0
1841 self.num_physical_experts = 0
1842 self.num_local_physical_experts = 0
1843 self.num_routed_experts = 0
1844 self.num_shared_experts = 0
1845 self.num_redundant_experts = 0
1846 logger.warning("DeepSeekV2: No DeepseekV2MoE layer found in model.layers.")
1847 else:
1848 self.num_logical_experts = example_moe.n_logical_experts
1849 self.num_physical_experts = example_moe.n_physical_experts
1850 self.num_local_physical_experts = example_moe.n_local_physical_experts
1851 self.num_routed_experts = example_moe.n_routed_experts
1852 self.num_shared_experts = example_moe.n_shared_experts
1853 self.num_redundant_experts = example_moe.n_redundant_experts
1854
1855 def update_physical_experts_metadata(
1856 self,
1857 num_physical_experts: int,
1858 num_local_physical_experts: int,
1859 ) -> None:
1860 assert self.num_local_physical_experts == num_local_physical_experts
1861 self.num_physical_experts = num_physical_experts
1862 self.num_local_physical_experts = num_local_physical_experts
1863 self.num_redundant_experts = num_physical_experts - self.num_logical_experts
1864 for moe in self.moe_mlp_layers:
1865 moe.n_local_physical_experts = num_local_physical_experts
1866 moe.n_physical_experts = num_physical_experts
1867 moe.n_redundant_experts = self.num_redundant_experts
1868 moe.experts.update_expert_map()
1869
1870
1871 class DeepseekV2ForCausalLM(
1872 nn.Module,
1873 SupportsPP,
1874 DeepseekV2MixtureOfExperts,
1875 SupportsLoRA,
1876 SupportsEagle,
1877 SupportsEagle3,
1878 ):
1879 packed_modules_mapping = {
1880 "gate_up_proj": ["gate_proj", "up_proj"],
1881 }
1882 model_cls = DeepseekV2Model
1883
1884 def __init__(self, *, vllm_config: VllmConfig, prefix: str = ""):
1885 super().__init__()
1886 config = vllm_config.model_config.hf_config
1887 quant_config = vllm_config.quant_config
1888 self.config = config
1889 self.quant_config = quant_config
1890
1891 qk_nope_head_dim = getattr(config, "qk_nope_head_dim", 0)
1892 qk_rope_head_dim = getattr(config, "qk_rope_head_dim", 0)
1893 self.use_mha = config.model_type == "deepseek" or all(
1894 dim == 0 for dim in (qk_nope_head_dim, qk_rope_head_dim)
1895 )
1896
1897 if self.use_mha:
1898 self.packed_modules_mapping["qkv_proj"] = ["q_proj", "k_proj", "v_proj"]
1899
1900 # `packed_modules_mapping` needs to be modified before
1901 # initializing DeepseekV2Model, as it is passed inplace to
1902 # quantization config init and may be used to select the
1903 # quant_method for relevant layers during initialization.
1904 self.fuse_qkv_a_proj = (
1905 hasattr(config, "q_lora_rank") and config.q_lora_rank is not None
1906 )
1907 if self.fuse_qkv_a_proj:
1908 self.packed_modules_mapping["fused_qkv_a_proj"] = [
1909 "q_a_proj",
1910 "kv_a_proj_with_mqa",
1911 ]
1912
1913 self.model = self.model_cls(
1914 vllm_config=vllm_config, prefix=maybe_prefix(prefix, "model")
1915 )
1916 if get_pp_group().is_last_rank:
1917 self.lm_head = ParallelLMHead(
1918 config.vocab_size,
1919 config.hidden_size,
1920 quant_config=quant_config,
1921 prefix=maybe_prefix(prefix, "lm_head"),
1922 )
1923 else:
1924 self.lm_head = PPMissingLayer()
1925 self.logits_processor = LogitsProcessor(config.vocab_size)
1926 self.make_empty_intermediate_tensors = (
1927 self.model.make_empty_intermediate_tensors
1928 )
1929 # Set MoE hyperparameters
1930 self.num_moe_layers = (
1931 self.config.num_hidden_layers - self.config.first_k_dense_replace
1932 )
1933 self.set_moe_parameters()
1934
1935 def set_moe_parameters(self):
1936 self.num_expert_groups = getattr(self.config, "n_group", 1)
1937
1938 self.moe_layers = []
1939 self.moe_mlp_layers = []
1940 example_moe = None
1941 for layer in self.model.layers:
1942 if isinstance(layer, PPMissingLayer):
1943 continue
1944
1945 assert isinstance(layer, DeepseekV2DecoderLayer)
1946 if isinstance(layer.mlp, DeepseekV2MoE):
1947 # Pick last one layer since the first ones may be dense layers.
1948 example_moe = layer.mlp
1949 self.moe_mlp_layers.append(layer.mlp)
1950 self.moe_layers.append(layer.mlp.experts)
1951
1952 self.extract_moe_parameters(example_moe)
1953
1954 def set_aux_hidden_state_layers(self, layers: tuple[int, ...]) -> None:
1955 self.model.aux_hidden_state_layers = layers
1956
1957 def get_eagle3_aux_hidden_state_layers(self) -> tuple[int, ...]:
1958 num_layers = len(self.model.layers)
1959 return (2, num_layers // 2, num_layers - 3)
1960
1961 def embed_input_ids(self, input_ids: torch.Tensor) -> torch.Tensor:
1962 return self.model.embed_input_ids(input_ids)
1963
1964 def forward(
1965 self,
1966 input_ids: torch.Tensor | None,
1967 positions: torch.Tensor,
1968 intermediate_tensors: IntermediateTensors | None = None,
1969 inputs_embeds: torch.Tensor | None = None,
1970 ) -> torch.Tensor | IntermediateTensors:
1971 hidden_states = self.model(
1972 input_ids, positions, intermediate_tensors, inputs_embeds
1973 )
1974 return hidden_states
1975
1976 def compute_logits(
1977 self,
1978 hidden_states: torch.Tensor,
1979 ) -> torch.Tensor | None:
1980 logits = self.logits_processor(self.lm_head, hidden_states)
1981 return logits
1982
1983 def get_expert_mapping(self) -> list[tuple[str, str, int, str]]:
1984 # Params for weights, fp8 weight scales, fp8 activation scales
1985 # (param_name, weight_name, expert_id, shard_id)
1986 return fused_moe_make_expert_params_mapping(
1987 self,
1988 ckpt_gate_proj_name="gate_proj",
1989 ckpt_down_proj_name="down_proj",
1990 ckpt_up_proj_name="up_proj",
1991 num_experts=self.config.n_routed_experts,
1992 num_redundant_experts=0,
1993 )
1994
1995 def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
1996 loader = AutoWeightsLoader(self)
1997 return loader.load_weights(weights)
1998
1999
2000 class DeepseekForCausalLM(DeepseekV2ForCausalLM):
2001 pass
2002
2003
2004 class DeepseekV3ForCausalLM(DeepseekV2ForCausalLM):
2005 pass
2006
2007
2008 class GlmMoeDsaForCausalLM(DeepseekV2ForCausalLM):
2009 pass