main
py 537 lines 21.1 KB
Raw
1 # SPDX-License-Identifier: Apache-2.0
2 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3 """K3 dense MLA draft model for DSpark speculative decoding."""
4
5 from collections.abc import Iterable
6
7 import torch
8 import torch.nn as nn
9
10 import vllm._custom_ops as ops
11 from vllm.config import VllmConfig
12 from vllm.model_executor.layers.layernorm import RMSNorm
13 from vllm.model_executor.layers.linear import (
14 MergedColumnParallelLinear,
15 ReplicatedLinear,
16 )
17 from vllm.model_executor.layers.logits_processor import LogitsProcessor
18 from vllm.model_executor.models.qwen3_dspark import (
19 DSparkConfidenceHead,
20 DSparkMarkovHead,
21 )
22 from vllm.model_executor.models.utils import (
23 AutoWeightsLoader,
24 WeightsMapper,
25 get_draft_quant_config,
26 maybe_prefix,
27 )
28 from vllm.models.common.ops.fused_allreduce_rms_norm import fused_allreduce_rms_norm
29 from vllm.models.kimi_k3.nvidia.mla import MultiHeadLatentAttention
30 from vllm.models.kimi_k3.nvidia.model import KimiMLP
31 from vllm.utils.torch_utils import is_quantized_kv_cache
32 from vllm.v1.worker.workspace import current_workspace_manager
33
34 _GROUPED_KV_CACHE_DTYPES = frozenset(
35 {"auto", "bfloat16", "fp8", "fp8_e4m3", "fp8_e5m2"}
36 )
37
38
39 def _duplicate_context_kv_weights(
40 weights: Iterable[tuple[str, torch.Tensor]], num_layers: int
41 ) -> Iterable[tuple[str, torch.Tensor]]:
42 """Load each layer's KV projection into the cross-layer linear."""
43 for name, weight in weights:
44 yield name, weight
45 layer_prefix, marker, param_name = name.partition(
46 ".self_attn.kv_a_proj_with_mqa."
47 )
48 if not marker:
49 continue
50 layer_idx_str = layer_prefix.rsplit(".", 1)[-1]
51 if not layer_idx_str.isdecimal():
52 continue
53 layer_idx = int(layer_idx_str)
54 if layer_idx >= num_layers:
55 continue
56 fused_weight = weight.detach()
57 fused_weight.shard_id = layer_idx
58 yield f"context_kv_proj.{param_name}", fused_weight
59
60
61 class K3DSparkDecoderLayer(nn.Module):
62 def __init__(
63 self,
64 *,
65 vllm_config: VllmConfig,
66 config,
67 layer_idx: int,
68 start_layer_id: int,
69 prefix: str,
70 ) -> None:
71 super().__init__()
72 quant_config = get_draft_quant_config(vllm_config)
73 self.self_attn = MultiHeadLatentAttention(
74 config=config,
75 hidden_size=config.hidden_size,
76 num_heads=config.num_attention_heads,
77 qk_nope_head_dim=config.qk_nope_head_dim,
78 qk_rope_head_dim=config.qk_rope_head_dim,
79 v_head_dim=config.v_head_dim,
80 q_lora_rank=config.q_lora_rank,
81 kv_lora_rank=config.kv_lora_rank,
82 cache_config=vllm_config.cache_config,
83 quant_config=quant_config,
84 prefix=maybe_prefix(
85 prefix, f"layers.{start_layer_id + layer_idx}.self_attn"
86 ),
87 use_rope=True,
88 non_causal_multi_token_decode=True,
89 )
90 # Both row-parallel outputs stay un-reduced; their all-reduces are fused
91 # into the RMSNorm that follows via fused_allreduce_rms_norm.
92 self.self_attn.o_proj.reduce_results = False
93 self.mlp = KimiMLP(
94 hidden_size=config.hidden_size,
95 intermediate_size=config.intermediate_size,
96 hidden_act=config.hidden_act,
97 quant_config=quant_config,
98 reduce_results=False,
99 prefix=maybe_prefix(prefix, f"layers.{start_layer_id + layer_idx}.mlp"),
100 )
101 self.input_layernorm = RMSNorm(config.hidden_size, eps=config.rms_norm_eps)
102 self.post_attention_layernorm = RMSNorm(
103 config.hidden_size, eps=config.rms_norm_eps
104 )
105
106 def forward(
107 self,
108 positions: torch.Tensor,
109 hidden_states: torch.Tensor,
110 residual: torch.Tensor | None,
111 ) -> tuple[torch.Tensor, torch.Tensor]:
112 if residual is None:
113 # First layer: hidden_states is the (already reduced) embedding.
114 residual = hidden_states
115 hidden_states = self.input_layernorm(hidden_states)
116 else:
117 hidden_states, residual = fused_allreduce_rms_norm(
118 hidden_states, residual, self.input_layernorm
119 )
120
121 hidden_states = self.self_attn(
122 positions=positions,
123 hidden_states=hidden_states,
124 )
125 hidden_states, residual = fused_allreduce_rms_norm(
126 hidden_states, residual, self.post_attention_layernorm
127 )
128 # The MLP output is reduced by the next layer's input_layernorm (or by
129 # the model's final_norm).
130 hidden_states = self.mlp(hidden_states)
131 return hidden_states, residual
132
133
134 class K3DSparkModel(nn.Module):
135 def __init__(
136 self,
137 *,
138 vllm_config: VllmConfig,
139 start_layer_id: int,
140 prefix: str,
141 ) -> None:
142 super().__init__()
143 assert vllm_config.speculative_config is not None
144 self.config = vllm_config.speculative_config.draft_model_config.hf_config
145 self.quant_config = get_draft_quant_config(vllm_config)
146
147 # The frozen target embedding is aliased after the draft checkpoint loads.
148 self.embed_tokens: nn.Module | None = None
149
150 self.context_proj = ReplicatedLinear(
151 self.config.target_hidden_size * self.config.num_target_layers,
152 self.config.hidden_size,
153 bias=False,
154 return_bias=False,
155 quant_config=self.quant_config,
156 prefix=maybe_prefix(prefix, "context_proj"),
157 )
158 self.context_norm = RMSNorm(
159 self.config.hidden_size, eps=self.config.rms_norm_eps
160 )
161
162 self.layers = nn.ModuleList(
163 [
164 K3DSparkDecoderLayer(
165 vllm_config=vllm_config,
166 config=self.config,
167 layer_idx=layer_idx,
168 start_layer_id=start_layer_id,
169 prefix=prefix,
170 )
171 for layer_idx in range(self.config.num_hidden_layers)
172 ]
173 )
174 kv_width = self.config.kv_lora_rank + self.config.qk_rope_head_dim
175 self.context_kv_proj = MergedColumnParallelLinear(
176 self.config.hidden_size,
177 [kv_width] * self.config.num_hidden_layers,
178 bias=False,
179 return_bias=False,
180 quant_config=self.quant_config,
181 prefix=maybe_prefix(
182 prefix,
183 f"layers.{start_layer_id}.self_attn.fused_qkv_a_proj",
184 ),
185 disable_tp=True,
186 )
187 self.final_norm = RMSNorm(self.config.hidden_size, eps=self.config.rms_norm_eps)
188 self.markov_head = DSparkMarkovHead(
189 self.config.vocab_size,
190 self.config.draft_vocab_size,
191 self.config.markov_rank,
192 prefix=maybe_prefix(prefix, "markov_head"),
193 )
194 self.confidence_head: DSparkConfidenceHead | None = None
195 if getattr(self.config, "enable_confidence_head", False):
196 with_markov = getattr(self.config, "confidence_head_with_markov", False)
197 input_dim = self.config.hidden_size + (
198 self.config.markov_rank if with_markov else 0
199 )
200 self.confidence_head = DSparkConfidenceHead(
201 input_dim,
202 prefix=maybe_prefix(prefix, "confidence_head"),
203 bias=True,
204 with_markov=with_markov,
205 )
206 self._max_num_context_tokens = (
207 vllm_config.scheduler_config.max_num_batched_tokens
208 )
209
210 def embed_input_ids(self, input_ids: torch.Tensor) -> torch.Tensor:
211 assert self.embed_tokens is not None
212 return self.embed_tokens(input_ids)
213
214 def combine_hidden_states(self, hidden_states: torch.Tensor) -> torch.Tensor:
215 return self.context_norm(self.context_proj(hidden_states))
216
217 @torch.inference_mode()
218 def precompute_and_store_context_kv(
219 self,
220 context_states: torch.Tensor,
221 context_positions: torch.Tensor,
222 context_slot_mapping: torch.Tensor | list[torch.Tensor | None] | None = None,
223 ) -> None:
224 """Project target-derived context into each draft layer's latent cache."""
225 if not hasattr(self, "_num_context_layers"):
226 self._build_fused_context_kv_metadata()
227 self._precompute_fused_context_kv(
228 context_states, context_positions, context_slot_mapping
229 )
230
231 def _build_fused_context_kv_metadata(self) -> None:
232 """Build cross-layer metadata after checkpoint loading."""
233 attentions = [layer.self_attn for layer in self.layers]
234 assert attentions
235 attn0 = attentions[0]
236 assert attn0.q_lora_rank is not None
237 kv_width = attn0.kv_lora_rank + attn0.qk_rope_head_dim
238 for attn in attentions:
239 assert attn.q_lora_rank is not None
240 assert (
241 attn.q_lora_rank == attn0.q_lora_rank
242 and attn.kv_lora_rank == attn0.kv_lora_rank
243 and attn.qk_rope_head_dim == attn0.qk_rope_head_dim
244 and attn.kv_a_layernorm.variance_epsilon
245 == attn0.kv_a_layernorm.variance_epsilon
246 ), "All MLA DSpark layers must share their latent KV geometry."
247 self._context_kv_norm_weights = torch.stack(
248 [attn.kv_a_layernorm.weight.detach() for attn in attentions], dim=0
249 ).contiguous()
250 self._num_context_layers = len(attentions)
251 self._context_kv_width = kv_width
252 self._context_kv_lora_rank = attn0.kv_lora_rank
253 self._context_rope_dim = attn0.qk_rope_head_dim
254 self._context_rms_norm_eps = attn0.kv_a_layernorm.variance_epsilon
255 self._context_kv_scales: torch.Tensor | None = None
256 if attn0.kv_cache_dtype in _GROUPED_KV_CACHE_DTYPES and is_quantized_kv_cache(
257 attn0.kv_cache_dtype
258 ):
259 self._context_kv_scales = torch.stack(
260 [attn._k_scale.reshape(()) for attn in attentions]
261 )
262
263 def _precompute_fused_context_kv(
264 self,
265 context_states: torch.Tensor,
266 context_positions: torch.Tensor,
267 context_slot_mapping: torch.Tensor | list[torch.Tensor | None] | None,
268 ) -> None:
269 num_ctx = context_states.shape[0]
270 num_layers = self._num_context_layers
271
272 # One KV-only GEMM replaces five full Q+KV GEMMs. For K3 this projects
273 # 5*576 rows rather than 5*2112 rows (72.7% fewer A-projection FLOPs).
274 all_kv = self.context_kv_proj(context_states)
275 all_kv = all_kv.view(num_ctx, num_layers, self._context_kv_width)
276 all_kv_c = all_kv[..., : self._context_kv_lora_rank]
277 all_k_pe = all_kv[..., self._context_kv_lora_rank :]
278
279 # Layer-major layout lets the 2-D RMSNorm weights select a distinct row
280 # for each draft layer in one grouped kernel.
281 all_kv_c = all_kv_c.permute(1, 0, 2).contiguous()
282 all_kv_c_normed = torch.empty_like(all_kv_c)
283 ops.rms_norm(
284 all_kv_c_normed,
285 all_kv_c,
286 self._context_kv_norm_weights,
287 self._context_rms_norm_eps,
288 )
289
290 all_k_pe = all_k_pe.permute(1, 0, 2).contiguous()
291 all_k_pe_flat = all_k_pe.view(num_layers * num_ctx, 1, self._context_rope_dim)
292 (repeated_positions,) = current_workspace_manager().get_simultaneous(
293 ((num_layers * self._max_num_context_tokens,), torch.int64),
294 )
295 repeated_positions = repeated_positions[: num_layers * num_ctx]
296 repeated_positions.view(num_layers, num_ctx).copy_(context_positions)
297 # Keep the single-tensor context RoPE on vLLM's optimized CUDA op;
298 # DeepSeek YaRN's FlashInfer wrapper assumes a non-null key tensor.
299 rotary_emb = self.layers[0].self_attn.rotary_emb
300 assert rotary_emb is not None
301 ops.rotary_embedding(
302 repeated_positions,
303 all_k_pe_flat,
304 None,
305 rotary_emb.head_size,
306 rotary_emb.cos_sin_cache,
307 rotary_emb.is_neox_style,
308 )
309 all_k_pe = all_k_pe_flat.view(num_layers, num_ctx, 1, self._context_rope_dim)
310
311 if context_slot_mapping is None:
312 return
313
314 cache_layers = [layer.self_attn for layer in self.layers]
315 cache_dtype = cache_layers[0].kv_cache_dtype
316 if (
317 cache_dtype in _GROUPED_KV_CACHE_DTYPES
318 and all(cl.kv_cache_dtype == cache_dtype for cl in cache_layers)
319 and self._has_uniform_block_layout(cache_layers)
320 and (
321 isinstance(context_slot_mapping, torch.Tensor)
322 or all(s is not None for s in context_slot_mapping)
323 )
324 ):
325 # Grouped context KV insert assumes that all layers share the same
326 # cache dtype and block layout.
327
328 if isinstance(context_slot_mapping, (list, tuple)):
329 per_layer_slot_mappings = [
330 s for s in context_slot_mapping if s is not None
331 ]
332 if len({s.data_ptr() for s in per_layer_slot_mappings}) == 1:
333 # All rows alias to the same slot mapping.
334 slot_mapping = (
335 per_layer_slot_mappings[0].unsqueeze(0).expand(num_layers, -1)
336 )
337 else:
338 slot_mapping = torch.stack(per_layer_slot_mappings, dim=0)
339 else:
340 # Broadcast the single shared context_slot_mapping tensor.
341 slot_mapping = context_slot_mapping.unsqueeze(0).expand(num_layers, -1)
342
343 ref_cache = cache_layers[0].kv_cache
344 ops.concat_and_cache_mla_grouped(
345 all_kv_c_normed,
346 all_k_pe.squeeze(2),
347 self._get_context_kv_cache_ptrs(cache_layers),
348 slot_mapping,
349 ref_cache.size(1),
350 ref_cache.stride(0),
351 ref_cache.stride(1),
352 self._context_kv_scales,
353 cache_dtype,
354 )
355 return
356
357 for layer_idx, layer in enumerate(self.layers):
358 slot_mapping = (
359 context_slot_mapping[layer_idx]
360 if isinstance(context_slot_mapping, (list, tuple))
361 else context_slot_mapping
362 )
363 if slot_mapping is None:
364 continue
365 attn = layer.self_attn
366 attn.impl.do_kv_cache_update(
367 all_kv_c_normed[layer_idx],
368 all_k_pe[layer_idx],
369 attn.kv_cache,
370 slot_mapping,
371 attn.kv_cache_dtype,
372 attn._k_scale,
373 )
374
375 def _has_uniform_block_layout(
376 self,
377 cache_layers: list[MultiHeadLatentAttention],
378 ) -> bool:
379 key = tuple(cl.kv_cache.data_ptr() for cl in cache_layers)
380 if getattr(self, "_kv_cache_ptrs_key", None) != key:
381 assert not torch.cuda.is_current_stream_capturing()
382 self._kv_cache_ptrs_key = key
383 if hasattr(self, "_context_cache_ptrs"):
384 del self._context_cache_ptrs
385 ref_cache = cache_layers[0].kv_cache
386 self._layers_share_kv_block_layout = all(
387 cl.kv_cache.size(1) == ref_cache.size(1)
388 and cl.kv_cache.stride(0) == ref_cache.stride(0)
389 and cl.kv_cache.stride(1) == ref_cache.stride(1)
390 for cl in cache_layers
391 )
392 return self._layers_share_kv_block_layout
393
394 def _get_context_kv_cache_ptrs(
395 self,
396 cache_layers: list[MultiHeadLatentAttention],
397 ) -> torch.Tensor:
398 # The per-layer KV cache base pointers are stable after allocation, so
399 # build the pointer array once and return it on every call.
400 if not hasattr(self, "_context_cache_ptrs"):
401 ref_cache = cache_layers[0].kv_cache
402 cache_ptrs = torch.tensor(
403 [cl.kv_cache.data_ptr() for cl in cache_layers],
404 dtype=torch.int64,
405 device=ref_cache.device,
406 )
407 self._context_cache_ptrs = cache_ptrs
408 return self._context_cache_ptrs
409
410 def forward(
411 self,
412 input_ids: torch.Tensor,
413 positions: torch.Tensor,
414 inputs_embeds: torch.Tensor | None = None,
415 ) -> torch.Tensor:
416 if inputs_embeds is None:
417 inputs_embeds = self.embed_input_ids(input_ids)
418
419 hidden_states = inputs_embeds
420 residual = None
421 for layer in self.layers:
422 hidden_states, residual = layer(
423 positions=positions,
424 hidden_states=hidden_states,
425 residual=residual,
426 )
427 hidden_states, _ = fused_allreduce_rms_norm(
428 hidden_states, residual, self.final_norm
429 )
430 return hidden_states
431
432
433 class K3DSparkForCausalLM(nn.Module):
434 has_own_embed_tokens = False
435 has_own_lm_head = False
436 draft_id_to_target_id = None
437 hf_to_vllm_mapper = WeightsMapper(
438 # confidence_head is training-only. The frozen target embedding and LM
439 # head are shared after this draft-specific checkpoint is loaded.
440 orig_to_new_substr={
441 "confidence_head": None,
442 "embed_tokens": None,
443 "lm_head": None,
444 },
445 orig_to_new_prefix={"": "model."},
446 orig_to_new_stacked={
447 ".gate_proj": (".gate_up_proj", 0),
448 ".up_proj": (".gate_up_proj", 1),
449 ".q_a_proj": (".fused_qkv_a_proj", 0),
450 ".kv_a_proj_with_mqa": (".fused_qkv_a_proj", 1),
451 },
452 )
453
454 def __init__(self, *, vllm_config: VllmConfig, prefix: str = "") -> None:
455 super().__init__()
456 assert vllm_config.speculative_config is not None
457 self.draft_model_config = vllm_config.speculative_config.draft_model_config
458 self.config = self.draft_model_config.hf_config
459 if getattr(self.config, "enable_confidence_head", False):
460 # The class mapper drops the head as training-only; keep it for drafts
461 # that actually built one, since adaptive verification reads it.
462 self.hf_to_vllm_mapper = self.hf_to_vllm_mapper | WeightsMapper(
463 orig_to_new_substr={"confidence_head": "confidence_head"}
464 )
465 target_layer_num = vllm_config.model_config.get_total_num_hidden_layers()
466 self.model = K3DSparkModel(
467 vllm_config=vllm_config,
468 start_layer_id=target_layer_num,
469 prefix=maybe_prefix(prefix, "model"),
470 )
471
472 # Assigned by load_dspark_model from the target. Keeping no placeholder
473 # avoids a transient full-vocabulary allocation for this 163k-vocab model.
474 self.lm_head: nn.Module | None = None
475 logit_scale = getattr(self.config, "logit_scale", 1.0)
476 self.logits_processor = LogitsProcessor(
477 self.config.draft_vocab_size, scale=logit_scale
478 )
479
480 def embed_input_ids(self, input_ids: torch.Tensor) -> torch.Tensor:
481 return self.model.embed_input_ids(input_ids)
482
483 def combine_hidden_states(self, hidden_states: torch.Tensor) -> torch.Tensor:
484 return self.model.combine_hidden_states(hidden_states)
485
486 def get_draft_kv_cache_layer_names(self) -> list[str]:
487 return [layer.self_attn.layer_name for layer in self.model.layers]
488
489 def precompute_and_store_context_kv(
490 self,
491 context_states: torch.Tensor,
492 context_positions: torch.Tensor,
493 context_slot_mapping: torch.Tensor | list[torch.Tensor | None] | None = None,
494 ) -> None:
495 self.model.precompute_and_store_context_kv(
496 context_states, context_positions, context_slot_mapping
497 )
498
499 def forward(
500 self,
501 input_ids: torch.Tensor,
502 positions: torch.Tensor,
503 inputs_embeds: torch.Tensor | None = None,
504 ) -> torch.Tensor:
505 return self.model(input_ids, positions, inputs_embeds)
506
507 def compute_logits(self, hidden_states: torch.Tensor) -> torch.Tensor:
508 assert self.lm_head is not None
509 return self.logits_processor(self.lm_head, hidden_states)
510
511 def compute_draft_logits(self, hidden_states: torch.Tensor) -> torch.Tensor:
512 return self.compute_logits(hidden_states)
513
514 def map_draft_to_target(self, draft_ids: torch.Tensor) -> torch.Tensor:
515 return draft_ids
516
517 def markov_embed(self, token_ids: torch.Tensor) -> torch.Tensor:
518 return self.model.markov_head.embed(token_ids)
519
520 def markov_bias(self, markov_embed: torch.Tensor) -> torch.Tensor:
521 return self.model.markov_head.bias(markov_embed, self.logits_processor)
522
523 def compute_confidence(
524 self, head_hidden: torch.Tensor, markov_embed: torch.Tensor
525 ) -> torch.Tensor:
526 """Per-position acceptance probability for each drafted token."""
527 assert self.model.confidence_head is not None
528 return torch.sigmoid(self.model.confidence_head(head_hidden, markov_embed))
529
530 def load_weights(self, weights: Iterable[tuple[str, torch.Tensor]]) -> set[str]:
531 loader = AutoWeightsLoader(self)
532 # read: 1. all weights. 2. context kv weights
533 weights = _duplicate_context_kv_weights(weights, len(self.model.layers))
534 return loader.load_weights(weights, mapper=self.hf_to_vllm_mapper)
535
536 def process_weights_after_loading(self) -> None:
537 self.model._build_fused_context_kv_metadata()