vllm.v1.kv_cache_interface

Classes:

• AttentionSpec –
• ChunkedLocalAttentionSpec –
• CircularBufferSpec –

  One block per request holding the raw keys of the token group that

• CrossAttentionSpec –

  KV cache spec for cross-attention layers in encoder-decoder models.

• FullAttentionSpec –

  When hybrid allocator is disabled and the model contains both full

• HiSparseHotSpec –

  Ephemeral per-request HiSparse hot-cache allocation.

• HiSparseResidentSpec –

  Reclaimable GPU-resident pages for host-backed HiSparse KV.

• HiddenStateCacheSpec –

  Marker for hidden-state cache layers used by extract_hidden_states.

• KVCacheConfig –

  The KV cache configuration of a model.

• KVCacheGroupSpec –

  Represents a group of model layers that share the same KV cache block table.

• KVCacheSpec –

  A base class for specifying the KV cache format of one layer.

• KVCacheTensor –

  A class for specifying how the workers should initialize the KV cache.

• KVQuantMode –

  KV cache quantization mode.

• KpoolTailSpec –

  One-block circular scratch cache for a kpool indexer's raw tail.

• MLAAttentionSpec –
• RSWASpec –

  KV cache spec for Reference Sliding Window Attention (R-SWA).

• SinkFullAttentionSpec –
• SlidingWindowMLASpec –

  Sliding window attention with MLA cache format.

• SlidingWindowSpec –
• UniformTypeKVCacheSpecs –

  A KV cache spec for multiple layers with the same type of attention. Here,

Functions:

• compute_layer_kv_cache_shape_bytes –

  Return the 4D logical shape (B, H, N, C) where C is in bytes.

• compute_layout_strides –

  Byte strides in logical [L, B, H, N, C] axis order.

• create_kv_cache_views –

  View a flat int8 buffer as one 4D [B, H, N, C] view per layer.

• get_kv_quant_mode –

  Map a kv_cache_dtype string to a :class:KVQuantMode.

• get_mamba_prefill_checkpoint_position –

  Return the reusable Mamba checkpoint boundary for a prefill.

• group_kernel_blocks –

  View a kernel-block-granular layer cache with manager blocks as dim 0.

• is_full_attention_spec –

  Whether a KV cache group spec is (or wraps) full attention.

• is_mamba_prefill_checkpoint_valid –

  Whether a backend can export the checkpoint in this query.

• iter_layer_specs –

  The per-layer specs a KV cache group spec covers.

• kv_cache_uses_per_token_head_scales –

  Return True if kv_cache_dtype needs per-token-head scales.

• replace_as –

  dataclasses.replace, but rebuilding spec as target_cls

AttentionSpec dataclass

Bases: KVCacheSpec

Attributes:

• max_tp_shards (int | None) –

  Distinct shards this cache splits into across TP; TP ranks beyond this

• num_head_slots (int | None) –

  H of the logical [B, H, N, C] page when packing diverges from one

• real_page_size_bytes (int) –

  Alias of unpadded_page_size_bytes

• state_content_bytes (int | None) –

  C in bytes when packed; None means dense K/V content.

• state_content_size_bytes (int) –

  Bytes per (head slot, stored state) cell of the page.

• tokens_per_state (int | Fraction) –

  Tokens covered by one stored state. Ints > 1 compress multiple tokens

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True, kw_only=True)
class AttentionSpec(KVCacheSpec):
    dcp_sharded: bool = True
    max_tp_shards: int | None = None
    """Distinct shards this cache splits into across TP; TP ranks beyond this
    hold replicas. None: unknown, treated as not replicated."""
    num_kv_heads: int
    head_size: int
    dtype: torch.dtype
    head_size_v: int = None  # type: ignore[assignment]
    kv_quant_mode: KVQuantMode = KVQuantMode.NONE
    page_size_padded: int | None = None
    num_head_slots: int | None = None
    """H of the logical ``[B, H, N, C]`` page when packing diverges from one
    slot per KV head. None means one slot per KV head. Published by the backend.
    """
    state_content_bytes: int | None = None
    """C in bytes when packed; None means dense K/V content."""
    tokens_per_state: int | Fraction = 1
    """Tokens covered by one stored state. Ints > 1 compress multiple tokens
    into one state (DSv4 sparse MLA); fractions < 1 store multiple states per
    token (Whisper block pooling: ``Fraction(1, block_pool_size)``)."""

    def __post_init__(self):
        super().__post_init__()
        if self.head_size_v is None:
            object.__setattr__(self, "head_size_v", self.head_size)

    @property
    def num_heads(self) -> int:
        if self.num_head_slots is not None:
            return self.num_head_slots
        return self.num_kv_heads

    @property
    def state_content_size_bytes(self) -> int:
        """Bytes per (head slot, stored state) cell of the page."""
        if self.state_content_bytes is not None:
            return self.state_content_bytes
        return (self.head_size + self.head_size_v) * get_dtype_size(self.dtype)

    @property
    def unpadded_page_size_bytes(self) -> int:
        return self.num_heads * self.num_states * self.state_content_size_bytes

    @property
    def page_size_bytes(self) -> int:
        if self.page_size_padded is not None:
            assert self.page_size_padded >= self.unpadded_page_size_bytes
            return self.page_size_padded
        return self.unpadded_page_size_bytes

    @property
    def real_page_size_bytes(self) -> int:
        """Alias of ``unpadded_page_size_bytes``
        TODO(lucas): follow up with TPU backend to see if we can remove this property.
        """
        return self.unpadded_page_size_bytes

    def max_num_blocks_per_req(self, vllm_config: VllmConfig, max_len: int) -> int:
        parallel_config = vllm_config.parallel_config
        kv_shard_count = (
            parallel_config.decode_context_parallel_size if self.dcp_sharded else 1
        )
        return cdiv(max_len, self.block_size * kv_shard_count)

max_tp_shards = None class-attribute instance-attribute

Distinct shards this cache splits into across TP; TP ranks beyond this hold replicas. None: unknown, treated as not replicated.

num_head_slots = None class-attribute instance-attribute

H of the logical [B, H, N, C] page when packing diverges from one slot per KV head. None means one slot per KV head. Published by the backend.

real_page_size_bytes property

Alias of unpadded_page_size_bytes TODO(lucas): follow up with TPU backend to see if we can remove this property.

state_content_bytes = None class-attribute instance-attribute

C in bytes when packed; None means dense K/V content.

state_content_size_bytes property

Bytes per (head slot, stored state) cell of the page.

tokens_per_state = 1 class-attribute instance-attribute

Tokens covered by one stored state. Ints > 1 compress multiple tokens into one state (DSv4 sparse MLA); fractions < 1 store multiple states per token (Whisper block pooling: Fraction(1, block_pool_size)).

ChunkedLocalAttentionSpec dataclass

Bases: AttentionSpec

Methods:

• max_admission_blocks_per_request –

  Per-request admission cap, in blocks.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True, kw_only=True)
class ChunkedLocalAttentionSpec(AttentionSpec):
    attention_chunk_size: int

    def max_admission_blocks_per_request(
        self, max_in_flight_tokens: int, max_model_len: int
    ) -> int:
        """Per-request admission cap, in blocks.

        Single source of truth for both startup pool sizing
        (`max_memory_usage_bytes`) and the runtime admission gate, so requests
        admitted by startup can also be admitted at runtime.

        `max_in_flight_tokens` is the max tokens scheduled but not yet settled
        (one batch per concurrent step); see `VllmConfig.max_in_flight_tokens`.
        """
        # During chunked prefill, we hold KV for at most one chunk window plus
        # the in-flight tokens, since frees happen on the processed-token basis.
        num_tokens = min(
            self.attention_chunk_size + max_in_flight_tokens, max_model_len
        )
        return cdiv(num_tokens, self.block_size)

    def max_memory_usage_bytes(self, vllm_config: VllmConfig) -> int:
        max_blocks = self.max_admission_blocks_per_request(
            max_in_flight_tokens=vllm_config.max_in_flight_tokens,
            max_model_len=vllm_config.model_config.max_model_len,
        )
        return max_blocks * self.page_size_bytes

    def is_uniform_with_collection(
        self, kv_cache_specs: dict[str, KVCacheSpec]
    ) -> bool:
        return all(
            isinstance(spec, ChunkedLocalAttentionSpec)
            and spec.attention_chunk_size == self.attention_chunk_size
            for spec in kv_cache_specs.values()
        )

max_admission_blocks_per_request(max_in_flight_tokens, max_model_len)

Per-request admission cap, in blocks.

Single source of truth for both startup pool sizing (max_memory_usage_bytes) and the runtime admission gate, so requests admitted by startup can also be admitted at runtime.

max_in_flight_tokens is the max tokens scheduled but not yet settled (one batch per concurrent step); see VllmConfig.max_in_flight_tokens.

Source code in vllm/v1/kv_cache_interface.py

def max_admission_blocks_per_request(
    self, max_in_flight_tokens: int, max_model_len: int
) -> int:
    """Per-request admission cap, in blocks.

    Single source of truth for both startup pool sizing
    (`max_memory_usage_bytes`) and the runtime admission gate, so requests
    admitted by startup can also be admitted at runtime.

    `max_in_flight_tokens` is the max tokens scheduled but not yet settled
    (one batch per concurrent step); see `VllmConfig.max_in_flight_tokens`.
    """
    # During chunked prefill, we hold KV for at most one chunk window plus
    # the in-flight tokens, since frees happen on the processed-token basis.
    num_tokens = min(
        self.attention_chunk_size + max_in_flight_tokens, max_model_len
    )
    return cdiv(num_tokens, self.block_size)

CircularBufferSpec dataclass

Bases: AttentionSpec

One block per request holding the raw keys of the token group that is still being compressed.

block_size is the ring capacity. It must exceed the compression ratio by the speculative lookahead: a speculative step stores all of its rows, drafts included, before acceptance is known, while the next step still reads the open group's committed keys from the ring.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True, kw_only=True)
class CircularBufferSpec(AttentionSpec):
    """One block per request holding the raw keys of the token group that
    is still being compressed.

    ``block_size`` is the ring capacity. It must exceed the compression ratio
    by the speculative lookahead: a speculative step stores all of its rows,
    drafts included, before acceptance is known, while the next step still
    reads the open group's committed keys from the ring.
    """

    @property
    def block_table_token_alignment(self) -> int | None:
        return None

    def max_memory_usage_bytes(self, vllm_config: VllmConfig) -> int:
        # The ring occupies one block per request for its whole lifetime.
        del vllm_config
        return self.page_size_bytes

    def max_num_blocks_per_req(self, vllm_config: VllmConfig, max_len: int) -> int:
        del vllm_config, max_len
        return 1

    def is_uniform_with_collection(
        self, kv_cache_specs: dict[str, KVCacheSpec]
    ) -> bool:
        return all(
            isinstance(spec, CircularBufferSpec) for spec in kv_cache_specs.values()
        )

    @property
    def prefix_cacheable(self) -> bool:
        return False

    @property
    def uses_slot_mapping(self) -> bool:
        return False

CrossAttentionSpec dataclass

Bases: AttentionSpec

KV cache spec for cross-attention layers in encoder-decoder models.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True)
class CrossAttentionSpec(AttentionSpec):
    """KV cache spec for cross-attention layers in encoder-decoder models."""

    def max_memory_usage_bytes(self, vllm_config: VllmConfig) -> int:
        # For cross-attention, we need to cache encoder states
        # Get encoder length (e.g., 1500 for Whisper).
        max_encoder_len = vllm_config.scheduler_config.max_num_encoder_input_tokens
        return cdiv(max_encoder_len, self.block_size) * self.page_size_bytes

FullAttentionSpec dataclass

Bases: AttentionSpec

When hybrid allocator is disabled and the model contains both full attention layers and sliding window attention layers, sliding window attention are regarded as full attention in KV cache manager (blocks are allocated for all tokens), while computed as sliding window attention in model runner. In this case, we use FullAttentionSpec and record the sliding window size.

Methods:

• merge –

  Merge a list of FullAttentionSpec objects into a single

Attributes:

• non_causal (bool) –

  Whether the layer attends non-causally (e.g. Prefix LM). Carried on the

• sliding_window (int | None) –

  Default to None for not using sliding window attention.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True, kw_only=True)
class FullAttentionSpec(AttentionSpec):
    """When hybrid allocator is disabled and the model contains both full
    attention layers and sliding window attention layers, sliding
    window attention are regarded as full attention in KV cache manager
    (blocks are allocated for all tokens), while computed as sliding window
    attention in model runner.
    In this case, we use FullAttentionSpec and record the sliding window size.
    """

    sliding_window: int | None = None
    """
    Default to None for not using sliding window attention.
    """
    attention_chunk_size: int | None = None

    non_causal: bool = False
    """
    Whether the layer attends non-causally (e.g. Prefix LM). Carried on the
    spec so the engine core, which collects specs from all workers before the
    scheduler is built, can adjust scheduling policy (chunked prefill / prefix
    caching) regardless of tensor-parallel layout. It does not affect the KV
    cache layout itself.
    """

    def max_memory_usage_bytes(self, vllm_config: VllmConfig) -> int:
        max_model_len = vllm_config.model_config.max_model_len
        dcp_world_size = vllm_config.parallel_config.decode_context_parallel_size
        if self.dcp_sharded and dcp_world_size > 1:
            max_model_len = cdiv(max_model_len, dcp_world_size)
        return cdiv(max_model_len, self.block_size) * self.page_size_bytes

    @classmethod
    def merge_window_sizes(cls, window_sizes: set[int]) -> int | None:
        if len(window_sizes) == 0:
            return None
        elif len(window_sizes) == 1:
            return window_sizes.pop()
        else:
            raise ValueError(
                "All attention layers in the same KV cache group must have the "
                "same window size."
            )

    @classmethod
    def merge(cls, specs: list[Self]) -> Self:
        """Merge a list of FullAttentionSpec objects into a single
        FullAttentionSpec object.
        """
        assert all(isinstance(spec, FullAttentionSpec) for spec in specs), (
            "All attention layers in the same KV cache group must be FullAttentionSpec."
        )

        sliding_window = set(
            spec.sliding_window for spec in specs if spec.sliding_window is not None
        )
        attention_chunk_size = set(
            spec.attention_chunk_size
            for spec in specs
            if spec.attention_chunk_size is not None
        )
        assert not any(isinstance(spec, MLAAttentionSpec) for spec in specs), (
            "MLAAttentionSpec should be merged in MLAAttentionSpec.merge"
        )
        merged_spec = cls(
            block_size=specs[0].block_size,
            block_stride_alignment=specs[0].block_stride_alignment,
            num_kv_heads=specs[0].num_kv_heads,
            head_size=specs[0].head_size,
            head_size_v=specs[0].head_size_v,
            dtype=specs[0].dtype,
            kv_quant_mode=specs[0].kv_quant_mode,
            dcp_sharded=specs[0].dcp_sharded,
            max_tp_shards=specs[0].max_tp_shards,
            page_size_padded=specs[0].page_size_padded,
            num_head_slots=specs[0].num_head_slots,
            state_content_bytes=specs[0].state_content_bytes,
            tokens_per_state=specs[0].tokens_per_state,
            sliding_window=cls.merge_window_sizes(sliding_window),
            attention_chunk_size=cls.merge_window_sizes(attention_chunk_size),
            # If any layer in the group is non-causal, treat the group as
            # non-causal so the engine core disables incompatible scheduling.
            non_causal=any(spec.non_causal for spec in specs),
        )
        for spec in specs:
            for f in fields(AttentionSpec):
                assert getattr(spec, f.name) == getattr(merged_spec, f.name), (
                    "All attention layers in the same KV cache group must have "
                    "the same attention spec."
                )
        assert (merged_spec.sliding_window is not None) + (
            merged_spec.attention_chunk_size is not None
        ) <= 1, (
            "Model with both sliding window layers and chunked local attention "
            "layers is not supported."
        )
        return merged_spec

non_causal = False class-attribute instance-attribute

Whether the layer attends non-causally (e.g. Prefix LM). Carried on the spec so the engine core, which collects specs from all workers before the scheduler is built, can adjust scheduling policy (chunked prefill / prefix caching) regardless of tensor-parallel layout. It does not affect the KV cache layout itself.

sliding_window = None class-attribute instance-attribute

Default to None for not using sliding window attention.

merge(specs) classmethod

Merge a list of FullAttentionSpec objects into a single FullAttentionSpec object.

Source code in vllm/v1/kv_cache_interface.py

@classmethod
def merge(cls, specs: list[Self]) -> Self:
    """Merge a list of FullAttentionSpec objects into a single
    FullAttentionSpec object.
    """
    assert all(isinstance(spec, FullAttentionSpec) for spec in specs), (
        "All attention layers in the same KV cache group must be FullAttentionSpec."
    )

    sliding_window = set(
        spec.sliding_window for spec in specs if spec.sliding_window is not None
    )
    attention_chunk_size = set(
        spec.attention_chunk_size
        for spec in specs
        if spec.attention_chunk_size is not None
    )
    assert not any(isinstance(spec, MLAAttentionSpec) for spec in specs), (
        "MLAAttentionSpec should be merged in MLAAttentionSpec.merge"
    )
    merged_spec = cls(
        block_size=specs[0].block_size,
        block_stride_alignment=specs[0].block_stride_alignment,
        num_kv_heads=specs[0].num_kv_heads,
        head_size=specs[0].head_size,
        head_size_v=specs[0].head_size_v,
        dtype=specs[0].dtype,
        kv_quant_mode=specs[0].kv_quant_mode,
        dcp_sharded=specs[0].dcp_sharded,
        max_tp_shards=specs[0].max_tp_shards,
        page_size_padded=specs[0].page_size_padded,
        num_head_slots=specs[0].num_head_slots,
        state_content_bytes=specs[0].state_content_bytes,
        tokens_per_state=specs[0].tokens_per_state,
        sliding_window=cls.merge_window_sizes(sliding_window),
        attention_chunk_size=cls.merge_window_sizes(attention_chunk_size),
        # If any layer in the group is non-causal, treat the group as
        # non-causal so the engine core disables incompatible scheduling.
        non_causal=any(spec.non_causal for spec in specs),
    )
    for spec in specs:
        for f in fields(AttentionSpec):
            assert getattr(spec, f.name) == getattr(merged_spec, f.name), (
                "All attention layers in the same KV cache group must have "
                "the same attention spec."
            )
    assert (merged_spec.sliding_window is not None) + (
        merged_spec.attention_chunk_size is not None
    ) <= 1, (
        "Model with both sliding window layers and chunked local attention "
        "layers is not supported."
    )
    return merged_spec

HiSparseHotSpec dataclass

Bases: KVCacheSpec

Ephemeral per-request HiSparse hot-cache allocation.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True, kw_only=True)
class HiSparseHotSpec(KVCacheSpec):
    """Ephemeral per-request HiSparse hot-cache allocation."""

    page_size: int
    blocks_per_request: int

    @property
    def prefix_cacheable(self) -> bool:
        return False

    @property
    def page_size_bytes(self) -> int:
        return self.page_size

    @property
    def block_table_token_alignment(self) -> int | None:
        return None

    def max_memory_usage_bytes(self, vllm_config: VllmConfig) -> int:
        return self.blocks_per_request * self.page_size

    def max_num_blocks_per_req(self, vllm_config: VllmConfig, max_len: int) -> int:
        return self.blocks_per_request

    @property
    def has_layer_views(self) -> bool:
        return False

    @property
    def uses_slot_mapping(self) -> bool:
        return False

HiSparseResidentSpec dataclass

Bases: KVCacheSpec

Reclaimable GPU-resident pages for host-backed HiSparse KV.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True, kw_only=True)
class HiSparseResidentSpec(KVCacheSpec):
    """Reclaimable GPU-resident pages for host-backed HiSparse KV."""

    page_size: int

    @property
    def prefix_cacheable(self) -> bool:
        return False

    @property
    def page_size_bytes(self) -> int:
        return self.page_size

    def max_admission_blocks_per_request(
        self, max_in_flight_tokens: int, max_model_len: int
    ) -> int:
        num_tokens = min(max_in_flight_tokens, max_model_len)
        return cdiv(num_tokens, self.block_size)

    def max_memory_usage_bytes(self, vllm_config: VllmConfig) -> int:
        max_blocks = self.max_admission_blocks_per_request(
            max_in_flight_tokens=vllm_config.max_in_flight_tokens,
            max_model_len=vllm_config.model_config.max_model_len,
        )
        return max_blocks * self.page_size

    @property
    def has_layer_views(self) -> bool:
        return False

HiddenStateCacheSpec dataclass

Bases: MLAAttentionSpec

Marker for hidden-state cache layers used by extract_hidden_states.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True, kw_only=True)
class HiddenStateCacheSpec(MLAAttentionSpec):
    """Marker for hidden-state cache layers used by extract_hidden_states."""

    pass

KVCacheConfig dataclass

The KV cache configuration of a model.

Methods:

• num_blocks_of –

  Number of blocks addressable by the pool backing tensor.

• select_transfer_block_ids –

  Select block IDs for externally transferable cache groups.

Attributes:

• has_mixed_precision_kv_cache (bool) –

  Whether device attention caches use more than one precision.

• hisparse_host_block_stride (int | None) –

  Physical bytes between consecutive HiSparse host blocks.

• hisparse_host_num_blocks (int | None) –

  Capacity of the dedicated HiSparse host-block manager, when enabled.

• hisparse_shared_host_pool (bool) –

  Whether local TP ranks share one physical HiSparse host pool.

• kv_cache_groups (list[KVCacheGroupSpec]) –

  The kv cache groups of the model.

• kv_cache_layout (str | None) –

  The KV cache layout resolved by the engine core, adopted by all workers.

• kv_cache_tensors (list[KVCacheTensor]) –

  How should model runner initialize the KV cache tensors for each layer

• kv_tp_replicas (int) –

  Consecutive TP ranks holding identical KV for every layer (1: none).

• needs_kv_cache_zeroing (bool) –

  Whether newly allocated KV cache blocks must be zeroed before use.

• num_blocks (int) –

  The number of KV cache blocks

• prefix_cache_retention_interval (int | None) –

  Resolved retention policy for local prefix-cache checkpoints.

• prefix_cacheable_group_ids (tuple[int, ...]) –

  IDs of transferable groups eligible for hash-addressed stores.

• prefix_cacheable_groups (tuple[KVCacheGroupSpec, ...]) –

  Transferable groups eligible for hash-addressed stores.

• transfer_group_ids (tuple[int, ...]) –

  IDs of cache groups that participate in external KV transfer.

• transfer_group_index_by_layer (dict[str, int]) –

  Transfer-group tuple index for each participating layer.

• transfer_groups (tuple[KVCacheGroupSpec, ...]) –

  Cache groups that participate in external KV transfer.

Source code in vllm/v1/kv_cache_interface.py

@dataclass
class KVCacheConfig:
    """The KV cache configuration of a model."""

    num_blocks: int
    """The number of KV cache blocks"""
    kv_cache_tensors: list[KVCacheTensor]
    """How should model runner initialize the KV cache tensors for each layer"""
    kv_cache_groups: list[KVCacheGroupSpec]
    """
    The kv cache groups of the model.
    For models with only one type of attention, there is only one group that
    contains all layers.
    For models with multiple types of attention, there will be multiple groups,
    see `_get_kv_cache_config_uniform_page_size` for more details.
    """
    prefix_cache_retention_interval: int | None = None
    """Resolved retention policy for local prefix-cache checkpoints."""
    kv_cache_layout: str | None = None
    """The KV cache layout resolved by the engine core, adopted by all workers."""
    hisparse_host_num_blocks: int | None = None
    """Capacity of the dedicated HiSparse host-block manager, when enabled."""

    hisparse_host_block_stride: int | None = None
    """Physical bytes between consecutive HiSparse host blocks."""

    hisparse_shared_host_pool: bool = False
    """Whether local TP ranks share one physical HiSparse host pool."""

    kv_tp_replicas: int = 1
    """Consecutive TP ranks holding identical KV for every layer (1: none)."""

    @cached_property
    def transfer_group_ids(self) -> tuple[int, ...]:
        """IDs of cache groups that participate in external KV transfer."""
        return tuple(
            group_id
            for group_id, group in enumerate(self.kv_cache_groups)
            if group.enable_kv_transfer
        )

    @cached_property
    def prefix_cacheable_group_ids(self) -> tuple[int, ...]:
        """IDs of transferable groups eligible for hash-addressed stores."""
        return tuple(
            group_id
            for group_id in self.transfer_group_ids
            if self.kv_cache_groups[group_id].kv_cache_spec.prefix_cacheable
        )

    @cached_property
    def prefix_cacheable_groups(self) -> tuple[KVCacheGroupSpec, ...]:
        """Transferable groups eligible for hash-addressed stores."""
        return tuple(
            self.kv_cache_groups[group_id]
            for group_id in self.prefix_cacheable_group_ids
        )

    @cached_property
    def transfer_groups(self) -> tuple[KVCacheGroupSpec, ...]:
        """Cache groups that participate in external KV transfer."""
        return tuple(
            self.kv_cache_groups[group_id] for group_id in self.transfer_group_ids
        )

    @cached_property
    def transfer_group_index_by_layer(self) -> dict[str, int]:
        """Transfer-group tuple index for each participating layer."""
        return {
            layer_name: group_index
            for group_index, group in enumerate(self.transfer_groups)
            for layer_name in group.layer_names
        }

    def select_transfer_block_ids(
        self, block_ids: Sequence[list[int]]
    ) -> tuple[list[int], ...]:
        """Select block IDs for externally transferable cache groups."""
        if len(block_ids) != len(self.kv_cache_groups):
            raise ValueError(
                f"Expected {len(self.kv_cache_groups)} KV cache groups, "
                f"got {len(block_ids)}."
            )
        return tuple(block_ids[group_id] for group_id in self.transfer_group_ids)

    @property
    def host_group_ids(self) -> tuple[int, ...]:
        return tuple(
            group_id
            for group_id, group in enumerate(self.kv_cache_groups)
            if group.host_resident
        )

    def num_blocks_of(self, tensor: KVCacheTensor) -> int:
        """Number of blocks addressable by the pool backing ``tensor``."""
        if not tensor.host_resident:
            return self.num_blocks
        assert self.hisparse_host_num_blocks is not None
        return self.hisparse_host_num_blocks

    @property
    def has_mamba_layers(self) -> bool:
        return any(
            isinstance(spec, MambaSpec)
            for group in self.kv_cache_groups
            for spec in iter_layer_specs(group.kv_cache_spec)
        )

    @property
    def has_mixed_precision_kv_cache(self) -> bool:
        """Whether device attention caches use more than one precision."""
        kv_cache_precisions: set[tuple[torch.dtype, KVQuantMode]] = set()
        for group in self.kv_cache_groups:
            if group.host_resident:
                continue
            kv_cache_precisions.update(
                (spec.dtype, spec.kv_quant_mode)
                for spec in iter_layer_specs(group.kv_cache_spec)
                if isinstance(spec, AttentionSpec)
            )
        return len(kv_cache_precisions) > 1

    @property
    def needs_kv_cache_zeroing(self) -> bool:
        """Whether newly allocated KV cache blocks must be zeroed before use.

        Required for Mamba layers, whose state is read before it is fully written
        (#35219), and for mixed-precision caches, where a block reused across
        groups can be reinterpreted under a different precision and decode stale
        bytes to NaN/Inf. Uniform-precision caches skip zeroing.
        """
        return self.has_mamba_layers or self.has_mixed_precision_kv_cache

has_mixed_precision_kv_cache property

Whether device attention caches use more than one precision.

hisparse_host_block_stride = None class-attribute instance-attribute

Physical bytes between consecutive HiSparse host blocks.

hisparse_host_num_blocks = None class-attribute instance-attribute

Capacity of the dedicated HiSparse host-block manager, when enabled.

hisparse_shared_host_pool = False class-attribute instance-attribute

Whether local TP ranks share one physical HiSparse host pool.

kv_cache_groups instance-attribute

The kv cache groups of the model. For models with only one type of attention, there is only one group that contains all layers. For models with multiple types of attention, there will be multiple groups, see _get_kv_cache_config_uniform_page_size for more details.

kv_cache_layout = None class-attribute instance-attribute

The KV cache layout resolved by the engine core, adopted by all workers.

kv_cache_tensors instance-attribute

How should model runner initialize the KV cache tensors for each layer

kv_tp_replicas = 1 class-attribute instance-attribute

Consecutive TP ranks holding identical KV for every layer (1: none).

needs_kv_cache_zeroing property

Whether newly allocated KV cache blocks must be zeroed before use.

Required for Mamba layers, whose state is read before it is fully written (#35219), and for mixed-precision caches, where a block reused across groups can be reinterpreted under a different precision and decode stale bytes to NaN/Inf. Uniform-precision caches skip zeroing.

num_blocks instance-attribute

The number of KV cache blocks

prefix_cache_retention_interval = None class-attribute instance-attribute

Resolved retention policy for local prefix-cache checkpoints.

prefix_cacheable_group_ids cached property

IDs of transferable groups eligible for hash-addressed stores.

prefix_cacheable_groups cached property

Transferable groups eligible for hash-addressed stores.

transfer_group_ids cached property

IDs of cache groups that participate in external KV transfer.

transfer_group_index_by_layer cached property

Transfer-group tuple index for each participating layer.

transfer_groups cached property

Cache groups that participate in external KV transfer.

num_blocks_of(tensor)

Number of blocks addressable by the pool backing tensor.

Source code in vllm/v1/kv_cache_interface.py

def num_blocks_of(self, tensor: KVCacheTensor) -> int:
    """Number of blocks addressable by the pool backing ``tensor``."""
    if not tensor.host_resident:
        return self.num_blocks
    assert self.hisparse_host_num_blocks is not None
    return self.hisparse_host_num_blocks

select_transfer_block_ids(block_ids)

Select block IDs for externally transferable cache groups.

Source code in vllm/v1/kv_cache_interface.py

def select_transfer_block_ids(
    self, block_ids: Sequence[list[int]]
) -> tuple[list[int], ...]:
    """Select block IDs for externally transferable cache groups."""
    if len(block_ids) != len(self.kv_cache_groups):
        raise ValueError(
            f"Expected {len(self.kv_cache_groups)} KV cache groups, "
            f"got {len(block_ids)}."
        )
    return tuple(block_ids[group_id] for group_id in self.transfer_group_ids)

KVCacheGroupSpec dataclass

Represents a group of model layers that share the same KV cache block table. These layers are regarded as one layer in the KV cache manager.

Source code in vllm/v1/kv_cache_interface.py

@dataclass
class KVCacheGroupSpec:
    """Represents a group of model layers that share the same KV cache block table.
    These layers are regarded as one layer in the KV cache manager.
    """

    # The names of model layers in this group
    layer_names: list[str]
    # The KV cache spec of this manager layer
    kv_cache_spec: KVCacheSpec
    # Whether this group contains EAGLE/MTP draft attention layers.
    is_eagle_group: bool = False
    # Host groups use the dedicated HiSparse pool; others share the device pool.
    host_resident: bool = False
    # Whether this group is part of the externally transferable KV state.
    # Ephemeral accelerator-side replicas are rebuilt from their transferable
    # source group after a connector load.
    enable_kv_transfer: bool = True
    role: KVCacheGroupRole = KVCacheGroupRole.DEFAULT

KVCacheSpec dataclass

A base class for specifying the KV cache format of one layer.

Methods:

• copy_with_new_block_size –

  Create a new KVCacheSpec from self but replacing the block size.

• is_uniform_with_collection –

  Whether this KVCacheSpec is uniform with all specs of all layers.

• max_memory_usage_bytes –

  The maximum possible memory usage of this KV cache in bytes.

• max_num_blocks_per_req –

  The number of block table entries needed per request, i.e. the row

• merge –

  Merge a list of KVCacheSpec objects into a single KVCacheSpec object.

Attributes:

• block_stride_alignment (int | None) –

  Required byte alignment between physical blocks, including packed layers.

• dcp_sharded (bool) –

  Whether DCP shards this cache's token positions across ranks.

• has_layer_views (bool) –

  Whether generic allocation creates per-layer cache views.

• page_size_bytes (int) –

  The size of a page with block_size tokens in bytes.

• prefix_cacheable (bool) –

  Whether this spec's group participates in prefix caching.

• prefix_replay_tokens (int) –

  DeepSeek-V4.1 only: bounded replay. Currently only for DSV41 SWA.

• uses_slot_mapping (bool) –

  Whether the worker computes a per-token slot mapping for this spec.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True)
class KVCacheSpec:
    """A base class for specifying the KV cache format of one layer."""

    # number of tokens in a block
    block_size: int
    dcp_sharded: bool = field(default=False, kw_only=True)
    """Whether DCP shards this cache's token positions across ranks."""

    block_stride_alignment: int | None = field(default=None, kw_only=True)
    """Required byte alignment between physical blocks, including packed layers."""

    def __post_init__(self):
        if self.block_stride_alignment is not None and self.block_stride_alignment <= 0:
            raise ValueError("block_stride_alignment must be positive")

    @property
    def prefix_cacheable(self) -> bool:
        """Whether this spec's group participates in prefix caching."""
        return True

    @property
    def prefix_replay_tokens(self) -> int:
        """DeepSeek-V4.1 only: bounded replay. Currently only for DSV41 SWA."""
        return 0

    @property
    def num_heads(self) -> int:
        raise NotImplementedError

    @property
    def tokens_per_state(self) -> int | Fraction:
        raise NotImplementedError

    @property
    def state_content_size_bytes(self) -> int:
        raise NotImplementedError

    @property
    def page_size_bytes(self) -> int:
        """The size of a page with `block_size` tokens in bytes.

        Returns:
            The page size

        """
        raise NotImplementedError

    @property
    def num_states(self) -> int:
        return self.get_num_kernel_states(self.block_size)

    def get_num_kernel_states(self, kernel_block_size: int) -> int:
        if self.tokens_per_state > 0:
            return kernel_block_size // self.tokens_per_state
        return 1

    @property
    def block_table_token_alignment(self) -> int | None:
        return 128

    def max_memory_usage_bytes(self, vllm_config: VllmConfig) -> int:
        """The maximum possible memory usage of this KV cache in bytes.

        Returns:
            The KV cache size in bytes

        """
        raise NotImplementedError

    def max_num_blocks_per_req(self, vllm_config: VllmConfig, max_len: int) -> int:
        """The number of block table entries needed per request, i.e. the row
        length of the worker-side block table for this cache group.

        Args:
            vllm_config: The vllm config.
            max_len: The maximum sequence length to size for, including the
                encoder length for encoder-decoder models.

        """
        return cdiv(max_len, self.block_size)

    def copy_with_new_block_size(self, block_size: int) -> Self:
        """Create a new KVCacheSpec from self but replacing the block size."""
        return replace(self, block_size=block_size)

    @classmethod
    def merge(cls, specs: list[Self]) -> Self:
        """Merge a list of KVCacheSpec objects into a single KVCacheSpec object."""
        if not all(spec == specs[0] for spec in specs[1:]):
            raise AssertionError(
                "All layers in the same KV cache group must be the same."
            )
        return copy.deepcopy(specs[0])

    def is_uniform_with_collection(
        self, kv_cache_specs: dict[str, KVCacheSpec]
    ) -> bool:
        """Whether this KVCacheSpec is uniform with all specs of all layers."""
        uniform_type_base_spec = KVCacheSpecRegistry.get_uniform_type_base_spec(self)
        assert uniform_type_base_spec is not None, (
            f"Unsupported KV cache spec type: {type(self)}. "
            "Please register it using @register_kv_cache_spec decorator."
        )
        return all(
            isinstance(spec, uniform_type_base_spec) for spec in kv_cache_specs.values()
        )

    @property
    def has_layer_views(self) -> bool:
        """Whether generic allocation creates per-layer cache views.

        Specs without a per-layer shape keep the raw backing tensor and lay it
        out themselves. They also have no attention module of their own, so
        they take no part in attention-backend or kernel-block selection.
        """
        return True

    @property
    def uses_slot_mapping(self) -> bool:
        """Whether the worker computes a per-token slot mapping for this spec.

        Specs that address their pages themselves (raw storage, ring buffers)
        take no slot mapping row.
        """
        return True

block_stride_alignment = field(default=None, kw_only=True) class-attribute instance-attribute

Required byte alignment between physical blocks, including packed layers.

dcp_sharded = field(default=False, kw_only=True) class-attribute instance-attribute

Whether DCP shards this cache's token positions across ranks.

has_layer_views property

Whether generic allocation creates per-layer cache views.

Specs without a per-layer shape keep the raw backing tensor and lay it out themselves. They also have no attention module of their own, so they take no part in attention-backend or kernel-block selection.

page_size_bytes property

The size of a page with block_size tokens in bytes.

Returns:

• int –

  The page size

prefix_cacheable property

Whether this spec's group participates in prefix caching.

prefix_replay_tokens property

DeepSeek-V4.1 only: bounded replay. Currently only for DSV41 SWA.

uses_slot_mapping property

Whether the worker computes a per-token slot mapping for this spec.

Specs that address their pages themselves (raw storage, ring buffers) take no slot mapping row.

copy_with_new_block_size(block_size)

Create a new KVCacheSpec from self but replacing the block size.

Source code in vllm/v1/kv_cache_interface.py

def copy_with_new_block_size(self, block_size: int) -> Self:
    """Create a new KVCacheSpec from self but replacing the block size."""
    return replace(self, block_size=block_size)

is_uniform_with_collection(kv_cache_specs)

Whether this KVCacheSpec is uniform with all specs of all layers.

Source code in vllm/v1/kv_cache_interface.py

def is_uniform_with_collection(
    self, kv_cache_specs: dict[str, KVCacheSpec]
) -> bool:
    """Whether this KVCacheSpec is uniform with all specs of all layers."""
    uniform_type_base_spec = KVCacheSpecRegistry.get_uniform_type_base_spec(self)
    assert uniform_type_base_spec is not None, (
        f"Unsupported KV cache spec type: {type(self)}. "
        "Please register it using @register_kv_cache_spec decorator."
    )
    return all(
        isinstance(spec, uniform_type_base_spec) for spec in kv_cache_specs.values()
    )

max_memory_usage_bytes(vllm_config)

The maximum possible memory usage of this KV cache in bytes.

Returns:

• int –

  The KV cache size in bytes

Source code in vllm/v1/kv_cache_interface.py

def max_memory_usage_bytes(self, vllm_config: VllmConfig) -> int:
    """The maximum possible memory usage of this KV cache in bytes.

    Returns:
        The KV cache size in bytes

    """
    raise NotImplementedError

max_num_blocks_per_req(vllm_config, max_len)

The number of block table entries needed per request, i.e. the row length of the worker-side block table for this cache group.

Parameters:

• vllm_config

  (VllmConfig) –

  The vllm config.

• max_len

  (int) –

  The maximum sequence length to size for, including the encoder length for encoder-decoder models.

Source code in vllm/v1/kv_cache_interface.py

def max_num_blocks_per_req(self, vllm_config: VllmConfig, max_len: int) -> int:
    """The number of block table entries needed per request, i.e. the row
    length of the worker-side block table for this cache group.

    Args:
        vllm_config: The vllm config.
        max_len: The maximum sequence length to size for, including the
            encoder length for encoder-decoder models.

    """
    return cdiv(max_len, self.block_size)

merge(specs) classmethod

Merge a list of KVCacheSpec objects into a single KVCacheSpec object.

Source code in vllm/v1/kv_cache_interface.py

@classmethod
def merge(cls, specs: list[Self]) -> Self:
    """Merge a list of KVCacheSpec objects into a single KVCacheSpec object."""
    if not all(spec == specs[0] for spec in specs[1:]):
        raise AssertionError(
            "All layers in the same KV cache group must be the same."
        )
    return copy.deepcopy(specs[0])

KVCacheTensor dataclass

A class for specifying how the workers should initialize the KV cache.

Placement of a set of same-shaped layers in the KV cache allocation. Layer layers[l]'s page for block b starts at offset + l * layer_stride + b * block_stride bytes into the backing allocation of size bytes. Layer-outermost layouts give each layer a contiguous region (layer_stride = page * num_blocks, block_stride = page); block-outermost layouts make each block a block of all layers' pages (layer_stride = page, block_stride = the packed block). Tensors whose address ranges overlap alias the same bytes: cache groups overlay each other, which is sound because a block ID is owned by one group at a time.

Source code in vllm/v1/kv_cache_interface.py

@dataclass
class KVCacheTensor:
    """A class for specifying how the workers should initialize the KV cache.

    Placement of a set of same-shaped layers in the KV cache allocation.
    Layer ``layers[l]``'s page for block ``b`` starts at
    ``offset + l * layer_stride + b * block_stride`` bytes into the backing
    allocation of ``size`` bytes. Layer-outermost layouts give each layer a
    contiguous region (``layer_stride = page * num_blocks``,
    ``block_stride = page``); block-outermost layouts make each block a
    block of all layers' pages (``layer_stride = page``, ``block_stride`` =
    the packed block). Tensors whose address ranges overlap
    alias the same bytes: cache groups overlay each other, which is sound
    because a block ID is owned by one group at a time.
    """

    size: int  # total size of the backing allocation in bytes
    layers: list[str]  # layer names in L order
    layer_stride: int
    block_stride: int
    offset: int = 0  # byte offset of layers[0]'s block 0
    host_resident: bool = False

KVQuantMode

Bases: IntEnum

KV cache quantization mode.

Used by attention backends and kernels to dispatch quantization logic without string matching on kv_cache_dtype.

Attributes:

• is_nvfp4 (bool) –

  True for NVFP4 packed quantization mode.

• is_per_token_head (bool) –

  True for any per-token-head quantization mode.

• is_turboquant (bool) –

  True for any turboquant quantization mode.

Source code in vllm/v1/kv_cache_interface.py

class KVQuantMode(IntEnum):
    """KV cache quantization mode.

    Used by attention backends and kernels to dispatch quantization logic
    without string matching on ``kv_cache_dtype``.
    """

    NONE = 0
    FP8_PER_TENSOR = 1  # per-tensor scales (current fp8 path)
    INT8_PER_TOKEN_HEAD = 2  # per-token-head dynamic scales for int8
    FP8_PER_TOKEN_HEAD = 3  # per-token-head dynamic scales for fp8
    INT4_PER_TOKEN_HEAD = 4  # packed 2×int4/byte, RHT + asymmetric zp
    NVFP4 = 5  # packed fp4 data + fp8 block scales
    # Hadamard-rotated Lloyd-Max quant, packed K+V per slot.
    TURBOQUANT_K8V4 = 6
    TURBOQUANT_4BIT_NC = 7
    TURBOQUANT_K3V4_NC = 8
    TURBOQUANT_3BIT_NC = 9
    NVFP4_DS_MLA = 10  # opaque-bytes NVFP4 DS-MLA layouts (FlashMLA sparse)

    @property
    def is_per_token_head(self) -> bool:
        """True for any per-token-head quantization mode."""
        return self in (
            KVQuantMode.INT8_PER_TOKEN_HEAD,
            KVQuantMode.FP8_PER_TOKEN_HEAD,
            KVQuantMode.INT4_PER_TOKEN_HEAD,
        )

    @property
    def is_nvfp4(self) -> bool:
        """True for NVFP4 packed quantization mode."""
        return self == KVQuantMode.NVFP4

    @property
    def is_turboquant(self) -> bool:
        """True for any turboquant quantization mode."""
        return self in (
            KVQuantMode.TURBOQUANT_K8V4,
            KVQuantMode.TURBOQUANT_4BIT_NC,
            KVQuantMode.TURBOQUANT_K3V4_NC,
            KVQuantMode.TURBOQUANT_3BIT_NC,
        )

is_nvfp4 property

True for NVFP4 packed quantization mode.

is_per_token_head property

True for any per-token-head quantization mode.

is_turboquant property

True for any turboquant quantization mode.

KpoolTailSpec dataclass

Bases: SlidingWindowSpec

One-block circular scratch cache for a kpool indexer's raw tail.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True, kw_only=True)
class KpoolTailSpec(SlidingWindowSpec):
    """One-block circular scratch cache for a kpool indexer's raw tail."""

    def max_admission_blocks_per_request(
        self, max_in_flight_tokens: int, max_model_len: int
    ) -> int:
        return 1

    def max_num_blocks_per_req(self, vllm_config: VllmConfig, max_len: int) -> int:
        return 1

    def is_uniform_with_collection(
        self, kv_cache_specs: dict[str, KVCacheSpec]
    ) -> bool:
        return all(isinstance(spec, KpoolTailSpec) for spec in kv_cache_specs.values())

    @property
    def prefix_cacheable(self) -> bool:
        return False

    @property
    def uses_slot_mapping(self) -> bool:
        return False

MLAAttentionSpec dataclass

Bases: FullAttentionSpec

Attributes:

• storage_block_size (int | None) –

  Token width used to view storage when it differs from the kernel block.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True, kw_only=True)
class MLAAttentionSpec(FullAttentionSpec):
    # TODO(Lucas/Chen): less hacky way to do this
    cache_dtype_str: str | None = None
    # DeepseekV4 only fields. Non-DeepseekV4 MLA models leave these at defaults.
    alignment: int | None = None  # Default to None for no padding.
    model_version: str | None = None
    cache_role: SparseCacheRole = SparseCacheRole.SPARSE
    is_index_group_leader: bool = False
    storage_block_size: int | None = None
    """Token width used to view storage when it differs from the kernel block."""
    # Group capability enabled when any member flattens a non-causal query block
    # into decode rows. Runtime metadata still selects causal vs. non-causal mode.
    non_causal_multi_token_decode: bool = False
    # MLA stores a single latent vector per state; there is no separate V.
    head_size_v: int = 0

    def __post_init__(self):
        super().__post_init__()
        _apply_alignment_padding(self)

    @classmethod
    def merge(cls, specs: list[Self]) -> Self:
        assert all(isinstance(spec, MLAAttentionSpec) for spec in specs), (
            "All attention layers in the same KV cache group must be MLAAttentionSpec."
        )
        cache_dtype_str_set = set(spec.cache_dtype_str for spec in specs)
        tokens_per_state_set = set(spec.tokens_per_state for spec in specs)
        model_version_set = set(spec.model_version for spec in specs)
        cache_role_set = {spec.cache_role for spec in specs}
        index_group_leader_set = {spec.is_index_group_leader for spec in specs}
        storage_block_size_set = set(spec.storage_block_size for spec in specs)
        block_stride_alignment_set = {spec.block_stride_alignment for spec in specs}
        assert (
            len(cache_dtype_str_set) == 1
            and len(tokens_per_state_set) == 1
            and len(model_version_set) == 1
            and len(cache_role_set) == 1
            and len(index_group_leader_set) == 1
            and len(storage_block_size_set) == 1
            and len(block_stride_alignment_set) == 1
        ), (
            "All attention layers in the same KV cache group must use the same "
            "quantization method, tokens per state, model version, cache role, "
            "index-sharing role, storage block size and block stride alignment."
        )
        merged_spec = cls(
            block_size=specs[0].block_size,
            num_kv_heads=specs[0].num_kv_heads,
            head_size=specs[0].head_size,
            dtype=specs[0].dtype,
            kv_quant_mode=specs[0].kv_quant_mode,
            dcp_sharded=specs[0].dcp_sharded,
            max_tp_shards=specs[0].max_tp_shards,
            page_size_padded=specs[0].page_size_padded,
            num_head_slots=specs[0].num_head_slots,
            state_content_bytes=specs[0].state_content_bytes,
            cache_dtype_str=cache_dtype_str_set.pop(),
            tokens_per_state=tokens_per_state_set.pop(),
            model_version=model_version_set.pop(),
            cache_role=cache_role_set.pop(),
            is_index_group_leader=index_group_leader_set.pop(),
            storage_block_size=storage_block_size_set.pop(),
            block_stride_alignment=block_stride_alignment_set.pop(),
            non_causal_multi_token_decode=any(
                spec.non_causal_multi_token_decode for spec in specs
            ),
        )
        for spec in specs:
            for f in fields(AttentionSpec):
                assert getattr(spec, f.name) == getattr(merged_spec, f.name), (
                    "All attention layers in the same KV cache group must have "
                    "the same attention spec."
                )
        return merged_spec

storage_block_size = None class-attribute instance-attribute

Token width used to view storage when it differs from the kernel block.

RSWASpec dataclass

Bases: FullAttentionSpec

KV cache spec for Reference Sliding Window Attention (R-SWA).

Prefill (image + text prompt) tokens are always globally visible. Only the last rswa_window generated tokens are kept in the KV cache; gap blocks (between the prefill tail and the current decode window) are evicted during each decode step to bound memory at O(prefix_blocks + window_blocks).

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True, kw_only=True)
class RSWASpec(FullAttentionSpec):
    """KV cache spec for Reference Sliding Window Attention (R-SWA).

    Prefill (image + text prompt) tokens are always globally visible.
    Only the last ``rswa_window`` generated tokens are kept in the KV cache;
    gap blocks (between the prefill tail and the current decode window) are
    evicted during each decode step to bound memory at
    O(prefix_blocks + window_blocks).
    """

    rswa_window: int

    @classmethod
    def merge(cls, specs: list[RSWASpec]) -> RSWASpec:
        assert all(isinstance(spec, RSWASpec) for spec in specs), (
            "All attention layers in the same KV cache group must be RSWASpec."
        )
        rswa_windows = {spec.rswa_window for spec in specs}
        assert len(rswa_windows) == 1, (
            f"All R-SWA layers must share the same rswa_window, got {rswa_windows}"
        )
        # Delegate common field merging to the parent, then reattach rswa_window.
        base = FullAttentionSpec.merge(specs)  # type: ignore[arg-type]
        return cls(
            block_size=base.block_size,
            block_stride_alignment=base.block_stride_alignment,
            num_kv_heads=base.num_kv_heads,
            head_size=base.head_size,
            head_size_v=base.head_size_v,
            dtype=base.dtype,
            kv_quant_mode=base.kv_quant_mode,
            dcp_sharded=base.dcp_sharded,
            max_tp_shards=base.max_tp_shards,
            page_size_padded=base.page_size_padded,
            num_head_slots=base.num_head_slots,
            state_content_bytes=base.state_content_bytes,
            tokens_per_state=base.tokens_per_state,
            sliding_window=base.sliding_window,
            attention_chunk_size=base.attention_chunk_size,
            non_causal=base.non_causal,
            rswa_window=rswa_windows.pop(),
        )

SinkFullAttentionSpec dataclass

Bases: FullAttentionSpec

Methods:

• merge –

  Merge a list of FullAttentionSpec objects into a single

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True)
class SinkFullAttentionSpec(FullAttentionSpec):
    sink_len: int | None = None

    @classmethod
    def merge(cls, specs: list[Self]) -> Self:
        """Merge a list of FullAttentionSpec objects into a single
        FullAttentionSpec object.
        """
        assert all(isinstance(spec, FullAttentionSpec) for spec in specs), (
            "All attention layers in the same KV cache group must be FullAttentionSpec."
        )

        sliding_window = set(
            spec.sliding_window for spec in specs if spec.sliding_window is not None
        )
        attention_chunk_size = set(
            spec.attention_chunk_size
            for spec in specs
            if spec.attention_chunk_size is not None
        )
        assert not any(isinstance(spec, MLAAttentionSpec) for spec in specs), (
            "MLAAttentionSpec should be merged in MLAAttentionSpec.merge"
        )
        merged_spec = cls(
            block_size=specs[0].block_size,
            num_kv_heads=specs[0].num_kv_heads,
            head_size=specs[0].head_size,
            head_size_v=specs[0].head_size_v,
            sink_len=specs[0].sink_len,
            block_stride_alignment=specs[0].block_stride_alignment,
            dtype=specs[0].dtype,
            kv_quant_mode=specs[0].kv_quant_mode,
            dcp_sharded=specs[0].dcp_sharded,
            max_tp_shards=specs[0].max_tp_shards,
            page_size_padded=specs[0].page_size_padded,
            num_head_slots=specs[0].num_head_slots,
            state_content_bytes=specs[0].state_content_bytes,
            sliding_window=cls.merge_window_sizes(sliding_window),
            attention_chunk_size=cls.merge_window_sizes(attention_chunk_size),
            non_causal=any(spec.non_causal for spec in specs),
        )
        for spec in specs:
            for f in fields(AttentionSpec):
                assert getattr(spec, f.name) == getattr(merged_spec, f.name), (
                    "All attention layers in the same KV cache group must have "
                    "the same attention spec."
                )
        assert (merged_spec.sliding_window is not None) + (
            merged_spec.attention_chunk_size is not None
        ) <= 1, (
            "Model with both sliding window layers and chunked local attention "
            "layers is not supported."
        )
        return merged_spec

merge(specs) classmethod

Merge a list of FullAttentionSpec objects into a single FullAttentionSpec object.

Source code in vllm/v1/kv_cache_interface.py

@classmethod
def merge(cls, specs: list[Self]) -> Self:
    """Merge a list of FullAttentionSpec objects into a single
    FullAttentionSpec object.
    """
    assert all(isinstance(spec, FullAttentionSpec) for spec in specs), (
        "All attention layers in the same KV cache group must be FullAttentionSpec."
    )

    sliding_window = set(
        spec.sliding_window for spec in specs if spec.sliding_window is not None
    )
    attention_chunk_size = set(
        spec.attention_chunk_size
        for spec in specs
        if spec.attention_chunk_size is not None
    )
    assert not any(isinstance(spec, MLAAttentionSpec) for spec in specs), (
        "MLAAttentionSpec should be merged in MLAAttentionSpec.merge"
    )
    merged_spec = cls(
        block_size=specs[0].block_size,
        num_kv_heads=specs[0].num_kv_heads,
        head_size=specs[0].head_size,
        head_size_v=specs[0].head_size_v,
        sink_len=specs[0].sink_len,
        block_stride_alignment=specs[0].block_stride_alignment,
        dtype=specs[0].dtype,
        kv_quant_mode=specs[0].kv_quant_mode,
        dcp_sharded=specs[0].dcp_sharded,
        max_tp_shards=specs[0].max_tp_shards,
        page_size_padded=specs[0].page_size_padded,
        num_head_slots=specs[0].num_head_slots,
        state_content_bytes=specs[0].state_content_bytes,
        sliding_window=cls.merge_window_sizes(sliding_window),
        attention_chunk_size=cls.merge_window_sizes(attention_chunk_size),
        non_causal=any(spec.non_causal for spec in specs),
    )
    for spec in specs:
        for f in fields(AttentionSpec):
            assert getattr(spec, f.name) == getattr(merged_spec, f.name), (
                "All attention layers in the same KV cache group must have "
                "the same attention spec."
            )
    assert (merged_spec.sliding_window is not None) + (
        merged_spec.attention_chunk_size is not None
    ) <= 1, (
        "Model with both sliding window layers and chunked local attention "
        "layers is not supported."
    )
    return merged_spec

SlidingWindowMLASpec dataclass

Bases: SlidingWindowSpec

Sliding window attention with MLA cache format.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True, kw_only=True)
class SlidingWindowMLASpec(SlidingWindowSpec):
    """Sliding window attention with MLA cache format."""

    cache_dtype_str: str | None = None
    # DeepseekV4-only: see MLAAttentionSpec.model_version.
    alignment: int | None = None  # Default to None for no padding.
    model_version: str | None = None
    bounded_replay: bool = False

    # MLA stores a single latent vector per state; there is no separate V.
    head_size_v: int = 0

    def __post_init__(self):
        assert self.model_version in (None, "deepseek_v4"), (
            f"Unsupported model version: {self.model_version}"
        )
        super().__post_init__()
        _apply_alignment_padding(self)

    @property
    def prefix_cacheable(self) -> bool:
        return not self.bounded_replay

    @property
    def prefix_replay_tokens(self) -> int:
        return self.sliding_window if self.bounded_replay else 0

    @classmethod
    def merge(cls, specs: list[Self]) -> Self:
        assert all(isinstance(spec, SlidingWindowMLASpec) for spec in specs), (
            "All attention layers in the same KV cache group must be "
            "SlidingWindowMLASpec."
        )
        cache_dtype_str_set = set(spec.cache_dtype_str for spec in specs)
        tokens_per_state_set = set(spec.tokens_per_state for spec in specs)
        model_version_set = set(spec.model_version for spec in specs)
        sliding_window_set = set(spec.sliding_window for spec in specs)
        extra_retained_set = set(spec.extra_retained_tokens for spec in specs)
        bounded_replay_set = set(spec.bounded_replay for spec in specs)
        block_stride_alignment_set = {spec.block_stride_alignment for spec in specs}
        assert len({spec.dcp_sharded for spec in specs}) == 1
        assert len({spec.max_tp_shards for spec in specs}) == 1
        assert (
            len(cache_dtype_str_set) == 1
            and len(tokens_per_state_set) == 1
            and len(model_version_set) == 1
            and len(sliding_window_set) == 1
            and len(extra_retained_set) == 1
            and len(bounded_replay_set) == 1
            and len(block_stride_alignment_set) == 1
        ), (
            "All attention layers in the same KV cache group must use the same "
            "quantization method, tokens per state, model version, sliding "
            "window size, retained token count, and replay policy."
        )
        return cls(
            block_size=specs[0].block_size,
            block_stride_alignment=block_stride_alignment_set.pop(),
            num_kv_heads=specs[0].num_kv_heads,
            head_size=specs[0].head_size,
            dtype=specs[0].dtype,
            dcp_sharded=specs[0].dcp_sharded,
            max_tp_shards=specs[0].max_tp_shards,
            page_size_padded=specs[0].page_size_padded,
            num_head_slots=specs[0].num_head_slots,
            state_content_bytes=specs[0].state_content_bytes,
            sliding_window=sliding_window_set.pop(),
            extra_retained_tokens=extra_retained_set.pop(),
            cache_dtype_str=cache_dtype_str_set.pop(),
            tokens_per_state=tokens_per_state_set.pop(),
            model_version=model_version_set.pop(),
            bounded_replay=bounded_replay_set.pop(),
        )

    def is_uniform_with_collection(
        self, kv_cache_specs: dict[str, KVCacheSpec]
    ) -> bool:
        return all(
            isinstance(spec, SlidingWindowMLASpec)
            and spec.sliding_window == self.sliding_window
            and spec.bounded_replay == self.bounded_replay
            for spec in kv_cache_specs.values()
        )

SlidingWindowSpec dataclass

Bases: AttentionSpec

Methods:

• max_admission_blocks_per_request –

  Per-request admission cap, in blocks.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True, kw_only=True)
class SlidingWindowSpec(AttentionSpec):
    sliding_window: int
    # The trailing edge of the window is extended by ``extra_retained_tokens``
    # so that those extra trailing tokens' blocks are retained (but not
    # attended). This is needed for multi-module spec decoding which can
    # re-prefill the last num_spec_prefill_tokens - 1 tokens from the end
    # of the sequence, and thus needs to delay freeing/caching of blocks.
    extra_retained_tokens: int = 0

    def max_admission_blocks_per_request(
        self, max_in_flight_tokens: int, max_model_len: int
    ) -> int:
        """Per-request admission cap, in blocks.

        Single source of truth for both startup pool sizing
        (`max_memory_usage_bytes`) and the runtime admission gate. Per-request
        real-held blocks plateau at this bound because
        `SlidingWindowManager.remove_skipped_blocks` runs from `allocate_slots`
        before each chunk's `get_num_blocks_to_allocate`.

        `max_in_flight_tokens` is the max tokens scheduled but not yet settled
        (one batch per concurrent step); see `VllmConfig.max_in_flight_tokens`.
        """
        # During chunked prefill, we hold KV for the last `sliding_window-1`
        # computed tokens plus the in-flight tokens (frees happen on the
        # processed-token basis); never more than `max_model_len`. An additional
        # `extra_retained_tokens` trailing tokens are kept alive below the
        # window for multi-module spec decoding, and must be accounted here too.
        num_tokens = min(
            self.sliding_window - 1 + self.extra_retained_tokens + max_in_flight_tokens,
            max_model_len,
        )
        # +1 because the sliding window may not start from the beginning of
        # the block. E.g. block size 4 and num_token 4 needs two blocks
        # [XXCD][EF] to store the 6-token window [CDEF].
        return cdiv(num_tokens, self.block_size) + 1

    def max_memory_usage_bytes(self, vllm_config: VllmConfig) -> int:
        assert (
            not self.dcp_sharded
            or vllm_config.parallel_config.decode_context_parallel_size == 1
        ), "DCP not support sliding window."
        max_blocks = self.max_admission_blocks_per_request(
            max_in_flight_tokens=vllm_config.max_in_flight_tokens,
            max_model_len=vllm_config.model_config.max_model_len,
        )
        return max_blocks * self.page_size_bytes

    def is_uniform_with_collection(
        self, kv_cache_specs: dict[str, KVCacheSpec]
    ) -> bool:
        return all(
            isinstance(spec, SlidingWindowSpec)
            and spec.sliding_window == self.sliding_window
            for spec in kv_cache_specs.values()
        )

max_admission_blocks_per_request(max_in_flight_tokens, max_model_len)

Per-request admission cap, in blocks.

Single source of truth for both startup pool sizing (max_memory_usage_bytes) and the runtime admission gate. Per-request real-held blocks plateau at this bound because SlidingWindowManager.remove_skipped_blocks runs from allocate_slots before each chunk's get_num_blocks_to_allocate.

max_in_flight_tokens is the max tokens scheduled but not yet settled (one batch per concurrent step); see VllmConfig.max_in_flight_tokens.

Source code in vllm/v1/kv_cache_interface.py

def max_admission_blocks_per_request(
    self, max_in_flight_tokens: int, max_model_len: int
) -> int:
    """Per-request admission cap, in blocks.

    Single source of truth for both startup pool sizing
    (`max_memory_usage_bytes`) and the runtime admission gate. Per-request
    real-held blocks plateau at this bound because
    `SlidingWindowManager.remove_skipped_blocks` runs from `allocate_slots`
    before each chunk's `get_num_blocks_to_allocate`.

    `max_in_flight_tokens` is the max tokens scheduled but not yet settled
    (one batch per concurrent step); see `VllmConfig.max_in_flight_tokens`.
    """
    # During chunked prefill, we hold KV for the last `sliding_window-1`
    # computed tokens plus the in-flight tokens (frees happen on the
    # processed-token basis); never more than `max_model_len`. An additional
    # `extra_retained_tokens` trailing tokens are kept alive below the
    # window for multi-module spec decoding, and must be accounted here too.
    num_tokens = min(
        self.sliding_window - 1 + self.extra_retained_tokens + max_in_flight_tokens,
        max_model_len,
    )
    # +1 because the sliding window may not start from the beginning of
    # the block. E.g. block size 4 and num_token 4 needs two blocks
    # [XXCD][EF] to store the 6-token window [CDEF].
    return cdiv(num_tokens, self.block_size) + 1

UniformTypeKVCacheSpecs dataclass

Bases: KVCacheSpec

A KV cache spec for multiple layers with the same type of attention. Here, same types means always need the same number of token slots. For example, sliding window attentions with different window sizes are not the same type and should not be merged into one UniformTypeKVCacheSpecs.

Methods:

• from_specs –

  Return a SameTypeKVCacheSpecs object if all layers have the same type

• get_max_layers_per_page_size –

  Max number of layers sharing a page size. For a balanced bucket

• is_uniform_type –

  Whether all layers have the same type of KV cache spec.

Attributes:

• first_spec (KVCacheSpec) –

  Return the first spec in the group.

Source code in vllm/v1/kv_cache_interface.py

@dataclass(frozen=True)
class UniformTypeKVCacheSpecs(KVCacheSpec):
    """A KV cache spec for multiple layers with the same type of attention. Here,
    same types means always need the same number of token slots. For example,
    sliding window attentions with different window sizes are not the same type
    and should not be merged into one UniformTypeKVCacheSpecs.
    """

    kv_cache_specs: dict[str, KVCacheSpec]

    def __post_init__(self):
        super().__post_init__()
        if self.kv_cache_specs:
            object.__setattr__(self, "dcp_sharded", self.first_spec.dcp_sharded)

    @property
    def prefix_cacheable(self) -> bool:
        return all(spec.prefix_cacheable for spec in self.kv_cache_specs.values())

    @property
    def prefix_replay_tokens(self) -> int:
        return max(spec.prefix_replay_tokens for spec in self.kv_cache_specs.values())

    @property
    def first_spec(self) -> KVCacheSpec:
        """Return the first spec in the group."""
        return next(iter(self.kv_cache_specs.values()))

    @property
    def page_size_bytes(self) -> int:
        return sum(spec.page_size_bytes for spec in self.kv_cache_specs.values())

    @property
    def block_table_token_alignment(self) -> int | None:
        return self.first_spec.block_table_token_alignment

    def max_memory_usage_bytes(self, vllm_config: VllmConfig) -> int:
        max_num_pages = max(
            cdiv(spec.max_memory_usage_bytes(vllm_config), spec.page_size_bytes)
            for spec in self.kv_cache_specs.values()
        )
        return max_num_pages * self.page_size_bytes

    def max_num_blocks_per_req(self, vllm_config: VllmConfig, max_len: int) -> int:
        # Metadata builders are constructed from the per-layer spec, so the base
        # cdiv(max_len, block_size) would drop its DCP sharding and size the
        # block table wider than those builders expect.
        widths = {
            spec.max_num_blocks_per_req(vllm_config, max_len)
            for spec in self.kv_cache_specs.values()
        }
        assert len(widths) == 1, (
            "All layers in the same KV cache group must need the same number "
            f"of block table entries, got {sorted(widths)}."
        )
        return next(iter(widths))

    @classmethod
    def is_uniform_type(cls, kv_cache_specs: dict[str, KVCacheSpec]) -> bool:
        """Whether all layers have the same type of KV cache spec.

        Uses the registry to determine grouping base classes, so custom specs
        that inherit from FullAttentionSpec are treated as full attention.
        """
        block_sizes = set(spec.block_size for spec in kv_cache_specs.values())
        if len(block_sizes) > 1:
            # Different block sizes, not uniform.
            return False
        if len({spec.dcp_sharded for spec in kv_cache_specs.values()}) > 1:
            return False
        first_spec = next(iter(kv_cache_specs.values()))
        return first_spec.is_uniform_with_collection(kv_cache_specs)

    @classmethod
    def from_specs(cls, kv_cache_specs: dict[str, KVCacheSpec]) -> Self | None:
        """Return a SameTypeKVCacheSpecs object if all layers have the same type
        of KV cache spec. Return None if not.
        """
        if cls.is_uniform_type(kv_cache_specs):
            block_size = next(iter(kv_cache_specs.values())).block_size
            return cls(block_size=block_size, kv_cache_specs=kv_cache_specs)
        else:
            return None

    def get_max_layers_per_page_size(self) -> int:
        """Max number of layers sharing a page size. For a balanced bucket
        this equals the number of repetitions of the layer pattern."""
        return Counter(
            spec.page_size_bytes for spec in self.kv_cache_specs.values()
        ).most_common(1)[0][1]

    def max_memory_usage_pages(self, vllm_config: VllmConfig) -> int:
        return max(
            cdiv(spec.max_memory_usage_bytes(vllm_config), spec.page_size_bytes)
            for spec in self.kv_cache_specs.values()
        )

    @property
    def uses_slot_mapping(self) -> bool:
        return self.first_spec.uses_slot_mapping

first_spec property

Return the first spec in the group.

from_specs(kv_cache_specs) classmethod

Return a SameTypeKVCacheSpecs object if all layers have the same type of KV cache spec. Return None if not.

Source code in vllm/v1/kv_cache_interface.py

@classmethod
def from_specs(cls, kv_cache_specs: dict[str, KVCacheSpec]) -> Self | None:
    """Return a SameTypeKVCacheSpecs object if all layers have the same type
    of KV cache spec. Return None if not.
    """
    if cls.is_uniform_type(kv_cache_specs):
        block_size = next(iter(kv_cache_specs.values())).block_size
        return cls(block_size=block_size, kv_cache_specs=kv_cache_specs)
    else:
        return None

get_max_layers_per_page_size()

Max number of layers sharing a page size. For a balanced bucket this equals the number of repetitions of the layer pattern.

Source code in vllm/v1/kv_cache_interface.py

def get_max_layers_per_page_size(self) -> int:
    """Max number of layers sharing a page size. For a balanced bucket
    this equals the number of repetitions of the layer pattern."""
    return Counter(
        spec.page_size_bytes for spec in self.kv_cache_specs.values()
    ).most_common(1)[0][1]

is_uniform_type(kv_cache_specs) classmethod

Whether all layers have the same type of KV cache spec.

Uses the registry to determine grouping base classes, so custom specs that inherit from FullAttentionSpec are treated as full attention.

Source code in vllm/v1/kv_cache_interface.py

@classmethod
def is_uniform_type(cls, kv_cache_specs: dict[str, KVCacheSpec]) -> bool:
    """Whether all layers have the same type of KV cache spec.

    Uses the registry to determine grouping base classes, so custom specs
    that inherit from FullAttentionSpec are treated as full attention.
    """
    block_sizes = set(spec.block_size for spec in kv_cache_specs.values())
    if len(block_sizes) > 1:
        # Different block sizes, not uniform.
        return False
    if len({spec.dcp_sharded for spec in kv_cache_specs.values()}) > 1:
        return False
    first_spec = next(iter(kv_cache_specs.values()))
    return first_spec.is_uniform_with_collection(kv_cache_specs)

compute_layer_kv_cache_shape_bytes(spec, num_blocks, kernel_block_size=None)

Return the 4D logical shape (B, H, N, C) where C is in bytes.

Source code in vllm/v1/kv_cache_interface.py

def compute_layer_kv_cache_shape_bytes(
    spec: KVCacheSpec,
    num_blocks: int,
    kernel_block_size: int | None = None,
) -> tuple[int, ...]:
    """Return the 4D logical shape ``(B, H, N, C)`` where C is in bytes."""
    bs = kernel_block_size if kernel_block_size is not None else spec.block_size
    assert spec.block_size % bs == 0, (
        f"Kernel block size {bs} must divide KV cache block size {spec.block_size}."
    )
    blocks_per_page = spec.block_size // bs
    return (
        num_blocks * blocks_per_page,
        spec.num_heads,
        spec.get_num_kernel_states(bs),
        spec.state_content_size_bytes,
    )

compute_layout_strides(spec, num_blocks, num_layers, layout, kernel_block_size=None, fixed_strides=(None,) * 5)

Byte strides in logical [L, B, H, N, C] axis order.

Source code in vllm/v1/kv_cache_interface.py

def compute_layout_strides(
    spec: KVCacheSpec,
    num_blocks: int,
    num_layers: int,
    layout: KVCacheLayout,
    kernel_block_size: int | None = None,
    fixed_strides: tuple[int | None, ...] = (None,) * 5,
) -> tuple[int, ...]:
    """Byte strides in logical ``[L, B, H, N, C]`` axis order."""
    assert len(fixed_strides) == 5
    assert all(stride is None or stride > 0 for stride in fixed_strides)
    shape = (
        num_layers,
        *compute_layer_kv_cache_shape_bytes(spec, num_blocks, kernel_block_size),
    )
    order = layout.stride_order
    padded_page_size = getattr(spec, "page_size_padded", None)
    if padded_page_size is not None:
        assert kernel_block_size is None or kernel_block_size == spec.block_size, (
            "Padded KV pages do not support kernel block splitting."
        )
        page_grid_end = max(order.index(_DIM_L), order.index(_DIM_B)) + 1
        page_grid_shape = tuple(shape[dim] for dim in order[:page_grid_end])
        assert prod(page_grid_shape) == num_layers * num_blocks, (
            "Page padding requires dimensions outside the page tail to be L, B, "
            f"or singleton; got {layout.name} with shape {shape}."
        )

    strides = [0] * 5
    current_stride = 1
    for physical_idx, dim in reversed(tuple(enumerate(order))):
        if padded_page_size is not None and physical_idx == page_grid_end - 1:
            current_stride = max(current_stride, padded_page_size)
            assert current_stride % padded_page_size == 0
        strides[dim] = fixed_strides[dim] or current_stride
        current_stride = strides[dim] * shape[dim]
    return tuple(strides)

create_kv_cache_views(raw, spec, num_blocks, layout, kv_cache_tensor, kernel_block_size=None)

View a flat int8 buffer as one 4D [B, H, N, C] view per layer.

Block b of layer l starts at the tensor offset plus its layer and block stride contributions.

Source code in vllm/v1/kv_cache_interface.py

def create_kv_cache_views(
    raw: torch.Tensor,
    spec: KVCacheSpec,
    num_blocks: int,
    layout: KVCacheLayout,
    kv_cache_tensor: KVCacheTensor,
    kernel_block_size: int | None = None,
) -> list[torch.Tensor]:
    """View a flat int8 buffer as one 4D ``[B, H, N, C]`` view per layer.

    Block ``b`` of layer ``l`` starts at the tensor offset plus its layer and
    block stride contributions.
    """
    num_layers = len(kv_cache_tensor.layers)
    layer_stride = kv_cache_tensor.layer_stride
    block_stride = kv_cache_tensor.block_stride
    shape_bytes = compute_layer_kv_cache_shape_bytes(
        spec, num_blocks, kernel_block_size
    )
    ratio = shape_bytes[0] // num_blocks
    if ratio > 1:
        # Kernel blocks subdivide a manager block into `ratio` equal pieces, so
        # they sit a constant stride apart only if a block is one dense page: no
        # padding at its end, and no other layer's page before the next block.
        dense_page_size = prod(compute_layer_kv_cache_shape_bytes(spec, 1)[1:])
        if block_stride != dense_page_size:
            raise ValueError(
                f"The resolved KV cache layout ({layout.name}) does not store "
                "blocks as dense, unpadded pages (block stride "
                f"{block_stride} != page {dense_page_size}), so a manager "
                f"block cannot be split into {ratio} kernel blocks of "
                f"{kernel_block_size} tokens. Reduce --block-size to "
                f"{kernel_block_size} or set VLLM_KV_CACHE_LAYOUT to a "
                "layer-compact layout (e.g. LBNHC)."
            )
        assert block_stride % ratio == 0, (
            f"Block stride {block_stride} must divide into {ratio} equal kernel blocks."
        )
        block_stride //= ratio

    logical_shape = (num_layers, *shape_bytes)
    strides = compute_layout_strides(
        spec,
        num_blocks,
        num_layers,
        layout,
        kernel_block_size,
        fixed_strides=(layer_stride, block_stride, None, None, None),
    )
    dtype = getattr(spec, "dtype", None)

    view_5d = torch.as_strided(
        raw,
        size=logical_shape,
        stride=strides,
        storage_offset=raw.storage_offset() + kv_cache_tensor.offset,
    )

    views = []
    for layer_idx in range(num_layers):
        cache_logical = view_5d[layer_idx]
        if dtype is not None:
            cache_logical = cache_logical.view(dtype)
        views.append(cache_logical)
    return views

get_kv_quant_mode(kv_cache_dtype)

Map a kv_cache_dtype string to a :class:KVQuantMode.

Source code in vllm/v1/kv_cache_interface.py

def get_kv_quant_mode(kv_cache_dtype: str) -> KVQuantMode:
    """Map a ``kv_cache_dtype`` string to a :class:`KVQuantMode`."""
    if kv_cache_dtype == "int4_per_token_head":
        return KVQuantMode.INT4_PER_TOKEN_HEAD
    if kv_cache_dtype == "int8_per_token_head":
        return KVQuantMode.INT8_PER_TOKEN_HEAD
    if kv_cache_dtype == "fp8_per_token_head":
        return KVQuantMode.FP8_PER_TOKEN_HEAD
    # Must precede the ``nvfp4`` prefix test below, which would otherwise match.
    if kv_cache_dtype == "nvfp4_ds_mla":
        # Page size is keyed on cache_dtype_str in the MLA specs, not
        # nvfp4_kv_cache_full_dim.
        return KVQuantMode.NVFP4_DS_MLA
    if kv_cache_dtype.startswith("nvfp4"):
        return KVQuantMode.NVFP4
    if isinstance(kv_cache_dtype, str) and kv_cache_dtype.startswith("turboquant_"):
        return KVQuantMode[kv_cache_dtype.upper()]
    if isinstance(kv_cache_dtype, str) and kv_cache_dtype.startswith("fp8"):
        return KVQuantMode.FP8_PER_TENSOR
    return KVQuantMode.NONE

get_mamba_prefill_checkpoint_position(num_tokens, hash_block_size, drop_eagle_block)

Return the reusable Mamba checkpoint boundary for a prefill.

Source code in vllm/v1/kv_cache_interface.py

def get_mamba_prefill_checkpoint_position(
    num_tokens: int,
    hash_block_size: int,
    drop_eagle_block: bool,
) -> int:
    """Return the reusable Mamba checkpoint boundary for a prefill."""
    checkpoint_position = (num_tokens - 1) // hash_block_size * hash_block_size
    if drop_eagle_block:
        checkpoint_position -= hash_block_size
    return max(checkpoint_position, 0)

group_kernel_blocks(cache, num_blocks)

View a kernel-block-granular layer cache with manager blocks as dim 0.

Kernel block splitting subdivides each manager block into uniformly strided kernel blocks, so grouping is a pure view: (num_blocks * ratio, ...)

Source code in vllm/v1/kv_cache_interface.py

def group_kernel_blocks(cache: torch.Tensor, num_blocks: int) -> torch.Tensor:
    """View a kernel-block-granular layer cache with manager blocks as dim 0.

    Kernel block splitting subdivides each manager block into uniformly strided
    kernel blocks, so grouping is a pure view: ``(num_blocks * ratio, ...)``
    """
    if cache.shape[0] == num_blocks:
        return cache
    assert cache.shape[0] % num_blocks == 0
    return cache.unflatten(0, (num_blocks, -1))

is_full_attention_spec(kv_cache_spec)

Whether a KV cache group spec is (or wraps) full attention.

UniformTypeKVCacheSpecs is not itself a FullAttentionSpec, so a bare isinstance check misses groups that carry the wrapper -- DeepSeek-V4's MLA layers, or any model taking the UniformTypeKVCacheSpecs.from_specs path.

Every layer must be full attention: a group holding a recycling (sliding-window) layer has no stable slot layout, so callers that key data by slot cannot use it.

Source code in vllm/v1/kv_cache_interface.py

def is_full_attention_spec(kv_cache_spec: KVCacheSpec) -> bool:
    """Whether a KV cache group spec is (or wraps) full attention.

    ``UniformTypeKVCacheSpecs`` is not itself a ``FullAttentionSpec``, so a bare
    isinstance check misses groups that carry the wrapper -- DeepSeek-V4's MLA
    layers, or any model taking the ``UniformTypeKVCacheSpecs.from_specs`` path.

    Every layer must be full attention: a group holding a recycling
    (sliding-window) layer has no stable slot layout, so callers that key data
    by slot cannot use it.
    """
    layer_specs = iter_layer_specs(kv_cache_spec)
    return len(layer_specs) > 0 and all(
        isinstance(spec, FullAttentionSpec) for spec in layer_specs
    )

is_mamba_prefill_checkpoint_valid(query_start, query_end, checkpoint_position, hash_block_size, mamba_block_size, checkpoint_alignment)

Whether a backend can export the checkpoint in this query.

Source code in vllm/v1/kv_cache_interface.py

def is_mamba_prefill_checkpoint_valid(
    query_start: int,
    query_end: int,
    checkpoint_position: int,
    hash_block_size: int,
    mamba_block_size: int,
    checkpoint_alignment: int | None,
) -> bool:
    """Whether a backend can export the checkpoint in this query."""
    if checkpoint_alignment is None:
        return False
    assert checkpoint_alignment > 0

    initial_state_col = (query_start - 1) // mamba_block_size
    checkpoint_col = cdiv(query_end, mamba_block_size) - 2
    return (
        query_start % hash_block_size == 0
        and checkpoint_col > initial_state_col
        and query_start + hash_block_size <= checkpoint_position
        and query_start < checkpoint_position < query_end
        and (checkpoint_position - query_start) % checkpoint_alignment == 0
    )

iter_layer_specs(kv_cache_spec)

The per-layer specs a KV cache group spec covers.

UniformTypeKVCacheSpecs groups keep one spec per layer; every other spec describes its group on its own. Returns the layer specs either way so callers do not have to special-case the wrapper.

Source code in vllm/v1/kv_cache_interface.py

def iter_layer_specs(kv_cache_spec: KVCacheSpec) -> Collection[KVCacheSpec]:
    """The per-layer specs a KV cache group spec covers.

    ``UniformTypeKVCacheSpecs`` groups keep one spec per layer; every other
    spec describes its group on its own. Returns the layer specs either way so
    callers do not have to special-case the wrapper.
    """
    if isinstance(kv_cache_spec, UniformTypeKVCacheSpecs):
        return kv_cache_spec.kv_cache_specs.values()
    return (kv_cache_spec,)

kv_cache_uses_per_token_head_scales(kv_cache_dtype)

Return True if kv_cache_dtype needs per-token-head scales.

Source code in vllm/v1/kv_cache_interface.py

def kv_cache_uses_per_token_head_scales(kv_cache_dtype: str) -> bool:
    """Return True if *kv_cache_dtype* needs per-token-head scales."""
    return get_kv_quant_mode(kv_cache_dtype).is_per_token_head

replace_as(spec, target_cls, *, drop=(), **changes)

dataclasses.replace, but rebuilding spec as target_cls e.g. SlidingWindowSpec -> FullAttentionSpec

Every field of spec must exist on target_cls unless named in drop; fields only target_cls has keep their default values.

Source code in vllm/v1/kv_cache_interface.py

def replace_as(
    spec: KVCacheSpec,
    target_cls: type[_SpecT],
    *,
    drop: Collection[str] = (),
    **changes,
) -> _SpecT:
    """``dataclasses.replace``, but rebuilding *spec* as *target_cls*
      e.g. ``SlidingWindowSpec`` -> ``FullAttentionSpec``

    Every field of *spec* must exist on *target_cls* unless named in *drop*;
    fields only *target_cls* has keep their default values.
    """
    kwargs = {
        f.name: getattr(spec, f.name)
        for f in fields(spec)
        if f.init and f.name not in drop
    }
    kwargs.update(changes)
    return target_cls(**kwargs)