import logging from dataclasses import dataclass from typing import List, Optional, Tuple import torch from sglang.srt.constrained.base_grammar_backend import BaseGrammarObject from sglang.srt.environ import envs from sglang.srt.layers.attention.utils import create_flashinfer_kv_indices_triton from sglang.srt.model_executor.forward_batch_info import CaptureHiddenMode from sglang.srt.runtime_context import get_server_args from sglang.srt.speculative.spec_info import SpecInput, SpecInputType logger = logging.getLogger(__name__) @dataclass class EagleVerifyInput(SpecInput): draft_token: torch.Tensor custom_mask: torch.Tensor positions: torch.Tensor retrieve_index: torch.Tensor retrieve_next_token: torch.Tensor retrieve_next_sibling: torch.Tensor retrieve_cum_len: torch.Tensor spec_steps: int topk: int draft_token_num: int capture_hidden_mode: CaptureHiddenMode seq_lens_sum: int seq_lens_cpu: torch.Tensor grammar: BaseGrammarObject = None # Stacked per-step draft proposal distribution q, shape (bs, num_steps, # vocab); only set under rejection sampling. Consumed by the verify kernel. draft_probs: torch.Tensor = None # Shape info for padding num_tokens_per_req: int = -1 # -1 auto-fills from draft_token_num. def __post_init__(self): super().__init__(SpecInputType.EAGLE_VERIFY) if self.num_tokens_per_req < 0: self.num_tokens_per_req = self.draft_token_num @property def max_tree_depth(self) -> int: """Longest root-to-leaf chain of the verify tree, incl. the root; bounds the accept_index row width. EAGLE trees are depth-bounded by the draft loop. Algorithms with other tree shapes override this.""" return self.spec_steps + 1 @property def tree_topk(self) -> int: """Branching factor passed to the tree-verify kernels; -1 means an irregular tree (no fixed per-level branching).""" return self.topk def get_spec_adjust_token_coefficient(self) -> Tuple[int, int]: return self.draft_token_num, self.draft_token_num @classmethod def create_idle_input( cls, topk: int, spec_steps: int, num_verify_tokens: int, device: str ): return cls( draft_token=torch.empty((0,), dtype=torch.long, device=device), custom_mask=torch.full((0,), True, dtype=torch.bool, device=device), positions=torch.empty((0,), dtype=torch.int64, device=device), retrieve_index=torch.full( (0, num_verify_tokens), -1, dtype=torch.long, device=device ), retrieve_next_token=torch.full( (0, num_verify_tokens), -1, dtype=torch.long, device=device ), retrieve_next_sibling=torch.full( (0, num_verify_tokens), -1, dtype=torch.long, device=device ), retrieve_cum_len=None, topk=topk, draft_token_num=num_verify_tokens, spec_steps=spec_steps, capture_hidden_mode=CaptureHiddenMode.FULL, seq_lens_sum=0, seq_lens_cpu=torch.empty((0,), dtype=torch.int64), ) def generate_attn_arg_prefill( self, req_pool_indices: torch.Tensor, paged_kernel_lens: torch.Tensor, paged_kernel_lens_sum: int, req_to_token: torch.Tensor, ): device = req_pool_indices.device batch_size = len(req_pool_indices) qo_indptr = torch.arange( 0, (1 + batch_size) * self.draft_token_num, step=self.draft_token_num, dtype=torch.int32, device=device, ) cum_kv_seq_len = torch.zeros( (batch_size + 1,), dtype=torch.int32, device=device ) paged_kernel_lens = paged_kernel_lens + self.draft_token_num cum_kv_seq_len[1:] = torch.cumsum(paged_kernel_lens, dim=0) kv_indices = torch.empty( paged_kernel_lens_sum + self.draft_token_num * batch_size, dtype=torch.int32, device=device, ) create_flashinfer_kv_indices_triton[(batch_size,)]( req_to_token, req_pool_indices, paged_kernel_lens, cum_kv_seq_len, None, kv_indices, req_to_token.size(1), ) mask_numel = ( paged_kernel_lens_sum * self.draft_token_num + (self.draft_token_num**2) * batch_size ) if self.custom_mask.numel() < mask_numel: # FIXME(attn): temporary fix for custom mask padding with cuda graph self.custom_mask = torch.cat( [ self.custom_mask, torch.full( (mask_numel - self.custom_mask.numel(),), True, dtype=torch.bool, device=device, ), ], dim=0, ) return kv_indices, cum_kv_seq_len, qo_indptr, self.custom_mask @dataclass class EagleDraftInput(SpecInput): # For idle stubs use `create_idle_input`, not the bare ctor: `filter_batch` # / `merge_batch` slice / cat `topk_p` / `topk_index` / `hidden_states` / # `bonus_tokens` unconditionally. # shape: (b, topk) topk_p: torch.Tensor = None topk_index: torch.Tensor = None # Draft proposal q from draft-extend, only set under rejection sampling: # (b, vocab) single-layer; (b, num_steps, vocab) multi-layer chain. draft_probs: torch.Tensor = None # shape: (b, hidden_size) - one hidden per req, consumed by `draft` forward. # None when the spec algorithm's draft doesn't read hidden_states # (e.g., STANDALONE — vanilla LLM draft). hidden_states: Optional[torch.Tensor] = None capture_hidden_mode: CaptureHiddenMode = CaptureHiddenMode.FULL # Survives across draft steps: spec_info is shared by reference across the # per-step forwards (each runs on a copied ForwardBatch, dropping writebacks). dsa_topk_indices: Optional[torch.Tensor] = None # Per-req bonus token (the "+1" target prediction at end of each accept # chain); the worker copies it here post-extend for next iter's draft. bonus_tokens: torch.Tensor = None # shape: (b + 1,) kv_indptr: torch.Tensor = None kv_indices: torch.Tensor = None num_tokens_per_req: int = -1 num_tokens_for_logprob_per_req: int = -1 # V2 overlap worker only: req_pool_indices used as buf slot keys. future_indices: Optional[torch.Tensor] = None def __post_init__(self): super().__init__(SpecInputType.EAGLE_DRAFT) def get_spec_adjust_token_coefficient(self) -> Tuple[int, int]: return self.num_tokens_per_req, self.num_tokens_for_logprob_per_req @classmethod def create_idle_input( cls, device: torch.device, hidden_size: Optional[int], dtype: Optional[torch.dtype], topk: int, capture_hidden_mode: CaptureHiddenMode, vocab_size: int = 0, ): return cls( bonus_tokens=torch.empty((0,), device=device, dtype=torch.int32), hidden_states=( torch.empty((0, hidden_size), device=device, dtype=dtype) if hidden_size is not None else None ), topk_p=torch.empty((0, topk), device=device, dtype=torch.float32), topk_index=torch.empty((0, topk), device=device, dtype=torch.int64), draft_probs=( torch.empty((0, vocab_size), device=device, dtype=torch.float32) if get_server_args().speculative_use_rejection_sampling else None ), capture_hidden_mode=capture_hidden_mode, ) def filter_batch(self, new_indices: torch.Tensor, has_been_filtered: bool = True): if self.future_indices is not None: self.future_indices = self.future_indices[new_indices] return strict_check = envs.SGLANG_SPEC_ENABLE_STRICT_FILTER_CHECK.get() if has_been_filtered: # in eagle_utils.py:verify, we have already filtered the batch by `unfinished_index` # therefore, we don't need to filter the batch again in scheduler error_msg = f"length of new_indices: {len(new_indices)} != length of topk_p: {len(self.topk_p)}, this should not happen" if len(new_indices) != len(self.topk_p): if strict_check: raise ValueError(error_msg) else: logger.warning(error_msg) self.topk_p = self.topk_p[: len(new_indices)] self.topk_index = self.topk_index[: len(new_indices)] if self.draft_probs is not None: self.draft_probs = self.draft_probs[: len(new_indices)] if self.hidden_states is not None: self.hidden_states = self.hidden_states[: len(new_indices)] self.bonus_tokens = self.bonus_tokens[: len(new_indices)] if self.dsa_topk_indices is not None: self.dsa_topk_indices = self.dsa_topk_indices[: len(new_indices)] else: # in some cases(e.g draft_extend), we have not filtered the batch by `unfinished_index` self.topk_p = self.topk_p[new_indices] self.topk_index = self.topk_index[new_indices] if self.draft_probs is not None: self.draft_probs = self.draft_probs[new_indices] if self.hidden_states is not None: self.hidden_states = self.hidden_states[new_indices] self.bonus_tokens = self.bonus_tokens[new_indices] if self.dsa_topk_indices is not None: self.dsa_topk_indices = self.dsa_topk_indices[new_indices] def merge_batch(self, spec_info: "EagleDraftInput"): if self.future_indices is not None: assert spec_info.future_indices is not None self.future_indices = torch.cat( [self.future_indices, spec_info.future_indices] ) return # Detect idle stub by `topk_index` length (idle inputs have # shape[0] == 0 across all fields). Don't use `hidden_states is None`: # for STANDALONE all non-idle inputs also have None hidden_states. if len(self.topk_index) == 0: self.hidden_states = spec_info.hidden_states self.bonus_tokens = spec_info.bonus_tokens self.topk_p = spec_info.topk_p self.topk_index = spec_info.topk_index self.draft_probs = spec_info.draft_probs self.dsa_topk_indices = spec_info.dsa_topk_indices return if len(spec_info.topk_index) == 0: return if self.hidden_states is not None and spec_info.hidden_states is not None: self.hidden_states = torch.cat( [self.hidden_states, spec_info.hidden_states], axis=0 ) self.bonus_tokens = torch.cat( [self.bonus_tokens, spec_info.bonus_tokens], axis=0 ) self.topk_p = torch.cat([self.topk_p, spec_info.topk_p]) self.topk_index = torch.cat([self.topk_index, spec_info.topk_index]) if self.dsa_topk_indices is not None and spec_info.dsa_topk_indices is not None: self.dsa_topk_indices = torch.cat( [self.dsa_topk_indices, spec_info.dsa_topk_indices] ) else: self.dsa_topk_indices = None if self.draft_probs is not None and spec_info.draft_probs is not None: self.draft_probs = torch.cat([self.draft_probs, spec_info.draft_probs]) @dataclass class EagleDraftExtendInput(SpecInput): """Inputs to the draft-extend forward (the fill-draft-kvcache pass after target prefill / verify). Installed on `batch.spec_info` by the worker's `_draft_extend_for_*` (and synthetically by draft-extend cuda-graph capture), then replaced with a fresh `EagleDraftInput` for the next iter's draft. """ # Target-model hidden states for the draft-extend forward; None when the # draft doesn't read hidden_states (e.g., STANDALONE). Shape: decode # (bs * num_draft_tokens, hidden), prefill (extend_num_tokens, hidden). hidden_states: Optional[torch.Tensor] = None # Per-req accept counts. `num_accept_tokens = num_correct_drafts + 1`. # Both kept for cuda-graph buffer indexing. num_correct_drafts: torch.Tensor = None num_accept_tokens: torch.Tensor = None # CPU view, read by attention backends during the extend forward. num_accept_tokens_cpu: List[int] = None # Per-req batch-state slices for the draft-extend forward: # - input_ids: accept tokens flat over surviving reqs # - seq_lens / _cpu: per-req sequence length (post-accept) # - req_pool_indices: per-req kv-pool slot input_ids: torch.Tensor = None seq_lens: torch.Tensor = None seq_lens_cpu: torch.Tensor = None req_pool_indices: torch.Tensor = None # - positions: shape `[total_accepted]`. # - bonus_tokens: shape `[bs]`; read post-extend to populate next iter's # `EagleDraftInput.bonus_tokens`. positions: Optional[torch.Tensor] = None bonus_tokens: Optional[torch.Tensor] = None capture_hidden_mode: CaptureHiddenMode = CaptureHiddenMode.LAST num_tokens_per_req: int = -1 num_tokens_for_logprob_per_req: int = 1 dsa_seed_topk_capture: Optional[torch.Tensor] = None dsa_seed_topk_select: Optional[torch.Tensor] = None # None for draft-extend's idle batch; attention backends fall back to # rebuilding plain metadata from seq_lens when this is None. kv_indptr: torch.Tensor = None def __post_init__(self): super().__init__(SpecInputType.EAGLE_DRAFT_EXTEND) def get_spec_adjust_token_coefficient(self) -> Tuple[int, int]: return self.num_tokens_per_req, self.num_tokens_for_logprob_per_req @classmethod def create_idle_input( cls, device: torch.device, hidden_size: Optional[int], dtype: Optional[torch.dtype], capture_hidden_mode: CaptureHiddenMode = CaptureHiddenMode.LAST, ) -> "EagleDraftExtendInput": return cls( hidden_states=( torch.empty((0, hidden_size), device=device, dtype=dtype) if hidden_size is not None else None ), num_correct_drafts=torch.empty((0,), device=device, dtype=torch.int32), num_accept_tokens=torch.empty((0,), device=device, dtype=torch.int32), num_accept_tokens_cpu=[], input_ids=torch.empty((0,), device=device, dtype=torch.long), seq_lens=torch.empty((0,), device=device, dtype=torch.int64), seq_lens_cpu=torch.empty((0,), dtype=torch.int64), req_pool_indices=torch.empty((0,), device=device, dtype=torch.int64), capture_hidden_mode=capture_hidden_mode, ) def generate_attn_arg_prefill( self, req_pool_indices: torch.Tensor, paged_kernel_lens: torch.Tensor, paged_kernel_lens_sum: Optional[int], req_to_token: torch.Tensor, ): device = req_pool_indices.device bs = self.num_correct_drafts.numel() # Constant num_tokens_per_req qo layout (required for cuda-graph capture). qo_indptr = torch.arange( 0, (bs + 1) * self.num_tokens_per_req, step=self.num_tokens_per_req, dtype=torch.int32, device=device, ) cum_kv_seq_len = torch.zeros((bs + 1,), dtype=torch.int32, device=device) cum_kv_seq_len[1:] = torch.cumsum(paged_kernel_lens, dim=0) if paged_kernel_lens_sum is None: paged_kernel_lens_sum = cum_kv_seq_len[-1] kv_indices = torch.empty( paged_kernel_lens_sum, dtype=torch.int32, device=device ) create_flashinfer_kv_indices_triton[(bs,)]( req_to_token, req_pool_indices, paged_kernel_lens, cum_kv_seq_len, None, kv_indices, req_to_token.size(1), ) return kv_indices, cum_kv_seq_len, qo_indptr, None