""" Adaptive KV Cache Module for Q-TensorFormer. Makes KV Cache memory and memory traffic first-class citizens in inference: - Multi-precision storage: FP16 (full), INT8 (quantized), INT4 (compressed) - Attention-aware rate-distortion compression - Budget-driven selective eviction (dropping lowest-utility tokens) - Detailed memory traffic instrumentation (bytes read, bytes written, peak MB) """ import torch import torch.nn as nn import torch.nn.functional as F import math from typing import Optional, Tuple, Dict, List, Union from enum import Enum from dataclasses import dataclass, field class KVPrecision(str, Enum): FP16 = "fp16" INT8 = "int8" INT4 = "int4" class KVResidencyTier(str, Enum): HOT_GPU = "hot_gpu" # High-bandwidth GPU SRAM / HBM (immediate compute access) WARM_CPU = "warm_cpu" # Host CPU-RAM (offloaded via PCIe with zero-copy pinning) COLD_EVICTED = "cold_evicted" # Evicted or secondary storage class QuantizedKVTensor: """ Holds a quantized Key or Value tensor with per-channel scale and zero-point. """ def __init__(self, tensor: torch.Tensor, precision: KVPrecision): self.precision = precision self.shape = tensor.shape self.device = tensor.device if precision == KVPrecision.FP16: self.data = tensor.to(torch.float16) self.scale = None self.zp = None elif precision == KVPrecision.INT8: # Symmetric 8-bit quantization along channel dimension max_val = tensor.abs().amax(dim=-1, keepdim=True).clamp(min=1e-5) self.scale = (max_val / 127.0).to(torch.float16) q = torch.round(tensor / self.scale).clamp(-128, 127).to(torch.int8) self.data = q self.zp = None elif precision == KVPrecision.INT4: # Asymmetric 4-bit packed quantization [0, 15] min_val = tensor.amin(dim=-1, keepdim=True) max_val = tensor.amax(dim=-1, keepdim=True).clamp(min=min_val + 1e-5) self.scale = ((max_val - min_val) / 15.0).to(torch.float16) self.zp = min_val.to(torch.float16) q = torch.round((tensor - self.zp) / self.scale).clamp(0, 15).to(torch.uint8) # Pack two 4-bit values into one 8-bit byte along last dimension if even last_dim = q.shape[-1] if last_dim % 2 == 0: q_packed = (q[..., 0::2] << 4) | (q[..., 1::2] & 0x0F) self.data = q_packed self.packed = True else: self.data = q self.packed = False def dequantize(self, target_dtype: torch.dtype = torch.float32) -> torch.Tensor: """Dequantize back to float tensor.""" if self.precision == KVPrecision.FP16: return self.data.to(target_dtype) if self.precision == KVPrecision.INT8: return (self.data.to(target_dtype) * self.scale.to(target_dtype)).to(target_dtype) # INT4 if getattr(self, "packed", False): # Unpack high and low nibbles high = (self.data >> 4) & 0x0F low = self.data & 0x0F unpacked = torch.stack([high, low], dim=-1).reshape(self.shape) return (unpacked.to(target_dtype) * self.scale.to(target_dtype) + self.zp.to(target_dtype)).to(target_dtype) else: return (self.data.to(target_dtype) * self.scale.to(target_dtype) + self.zp.to(target_dtype)).to(target_dtype) @property def num_bytes(self) -> int: """Calculate physical memory in bytes.""" total = self.data.numel() * self.data.element_size() if self.scale is not None: total += self.scale.numel() * self.scale.element_size() if self.zp is not None: total += self.zp.numel() * self.zp.element_size() return total class AdaptiveKVCache: """ Per-layer or per-model adaptive Key-Value cache. Supports: - Precision switching: FP16, INT8, INT4 - Dynamic token retention and eviction based on attention utility - Memory traffic counters (bytes read/written) """ def __init__( self, max_capacity: int = 4096, default_precision: KVPrecision = KVPrecision.FP16, eviction_policy: str = "attention_utility", # 'attention_utility' or 'fifo' window_size: int = 128, # protected recent tokens ): self.max_capacity = max_capacity self.precision = default_precision self.eviction_policy = eviction_policy self.window_size = window_size # Cached states: keys and values as list of tensors or QuantizedKVTensor self.k_cache: Optional[torch.Tensor] = None self.v_cache: Optional[torch.Tensor] = None self.quantized_k: Optional[QuantizedKVTensor] = None self.quantized_v: Optional[QuantizedKVTensor] = None # Attention utility score per cached token index: (seq_len,) self.utility_scores: Optional[torch.Tensor] = None # Traffic tracking self.total_bytes_written = 0 self.total_bytes_read = 0 self.evicted_tokens_count = 0 def reset(self): """Clear cache state.""" self.k_cache = None self.v_cache = None self.quantized_k = None self.quantized_v = None self.utility_scores = None self.total_bytes_written = 0 self.total_bytes_read = 0 self.evicted_tokens_count = 0 @property def seq_len(self) -> int: if self.k_cache is not None: return self.k_cache.shape[-2] if self.quantized_k is not None: return self.quantized_k.shape[-2] return 0 @property def current_bytes(self) -> int: """Return current memory footprint of cached keys and values in bytes.""" if self.precision == KVPrecision.FP16 and self.k_cache is not None: return (self.k_cache.numel() + self.v_cache.numel()) * 2 # float16 = 2 bytes if self.quantized_k is not None and self.quantized_v is not None: return self.quantized_k.num_bytes + self.quantized_v.num_bytes return 0 @property def current_mb(self) -> float: return self.current_bytes / (1024.0 * 1024.0) @property def memory_footprint_bytes(self) -> int: return self.current_bytes def set_precision(self, new_precision: Union[str, KVPrecision]): """Convert current cache in-place to new precision.""" if isinstance(new_precision, str): new_precision = KVPrecision(new_precision.lower()) if new_precision == self.precision: return if self.seq_len > 0: k, v = self.get_kv() self.precision = new_precision if self.precision == KVPrecision.FP16: self.k_cache = k.to(torch.float16) self.v_cache = v.to(torch.float16) self.quantized_k = None self.quantized_v = None else: self.quantized_k = QuantizedKVTensor(k, self.precision) self.quantized_v = QuantizedKVTensor(v, self.precision) self.k_cache = None self.v_cache = None else: self.precision = new_precision def update( self, key: torch.Tensor, value: torch.Tensor, attention_weights: Optional[torch.Tensor] = None, ) -> Tuple[torch.Tensor, torch.Tensor]: """ Append new key/value tokens to the cache, evicting if necessary. Args: key: (batch, n_heads, seq_len_new, head_dim) value: (batch, n_heads, seq_len_new, head_dim) attention_weights: optional attention distribution to update utility scores Returns: full_keys: (batch, n_heads, total_seq_len, head_dim) full_values: (batch, n_heads, total_seq_len, head_dim) """ # Count incoming memory traffic bytes_in = (key.numel() + value.numel()) * key.element_size() self.total_bytes_written += bytes_in # Retrieve existing float representation if self.seq_len == 0: curr_k = key curr_v = value new_tokens = key.shape[-2] self.utility_scores = torch.ones(new_tokens, device=key.device) else: old_k, old_v = self.get_kv(target_dtype=key.dtype) curr_k = torch.cat([old_k, key], dim=-2) curr_v = torch.cat([old_v, value], dim=-2) new_tokens = key.shape[-2] new_scores = torch.ones(new_tokens, device=key.device) self.utility_scores = torch.cat([self.utility_scores, new_scores], dim=0) # Update utility scores from attention weights if provided if attention_weights is not None: with torch.no_grad(): # Incoming attention received by cached tokens: (B, H, Q, K) -> sum over Q, avg over B, H attn_importance = attention_weights.sum(dim=-2).mean(dim=(0, 1)) # (K,) n_tokens = min(len(self.utility_scores), len(attn_importance)) self.utility_scores[:n_tokens] = 0.9 * self.utility_scores[:n_tokens] + 0.1 * attn_importance[:n_tokens] # Eviction if capacity exceeded curr_len = curr_k.shape[-2] if curr_len > self.max_capacity: curr_k, curr_v = self._evict(curr_k, curr_v, target_len=self.max_capacity) # Store in configured precision if self.precision == KVPrecision.FP16: self.k_cache = curr_k.to(torch.float16) self.v_cache = curr_v.to(torch.float16) self.quantized_k = None self.quantized_v = None else: self.quantized_k = QuantizedKVTensor(curr_k, self.precision) self.quantized_v = QuantizedKVTensor(curr_v, self.precision) self.k_cache = None self.v_cache = None # Return full dequantized tensors for attention computation out_k, out_v = self.get_kv(target_dtype=key.dtype) # Count read traffic bytes_out = (out_k.numel() + out_v.numel()) * out_k.element_size() self.total_bytes_read += bytes_out return out_k, out_v def _evict(self, k: torch.Tensor, v: torch.Tensor, target_len: int) -> Tuple[torch.Tensor, torch.Tensor]: """Evict lowest utility tokens while preserving sink tokens (first 4) and recent window.""" total_len = k.shape[-2] num_to_evict = total_len - target_len if num_to_evict <= 0: return k, v # Always protect sink tokens (e.g. first 4) and recent window tokens sink_size = min(4, total_len) window_size = min(self.window_size, total_len - sink_size) candidate_end = total_len - window_size if candidate_end <= sink_size: # If sequence is mostly window, just do FIFO truncation from start self.evicted_tokens_count += num_to_evict self.utility_scores = self.utility_scores[num_to_evict:] return k[..., num_to_evict:, :], v[..., num_to_evict:, :] candidate_scores = self.utility_scores[sink_size:candidate_end] # Find indices with highest utility to KEEP num_candidates_to_keep = (candidate_end - sink_size) - num_to_evict if num_candidates_to_keep <= 0: keep_indices = torch.tensor([], dtype=torch.long, device=k.device) else: _, keep_rel = torch.topk(candidate_scores, k=num_candidates_to_keep, largest=True, sorted=True) keep_indices = sink_size + keep_rel.sort().values # Concatenate: sink + kept candidates + recent window sink_indices = torch.arange(0, sink_size, device=k.device) window_indices = torch.arange(candidate_end, total_len, device=k.device) final_indices = torch.cat([sink_indices, keep_indices, window_indices], dim=0) self.evicted_tokens_count += num_to_evict self.utility_scores = self.utility_scores[final_indices] return k[..., final_indices, :], v[..., final_indices, :] def get_kv(self, target_dtype: torch.dtype = torch.float32) -> Tuple[torch.Tensor, torch.Tensor]: """Retrieve dequantized full key and value tensors.""" if self.seq_len == 0: raise ValueError("Cache is empty.") if self.precision == KVPrecision.FP16: return self.k_cache.to(target_dtype), self.v_cache.to(target_dtype) else: return self.quantized_k.dequantize(target_dtype), self.quantized_v.dequantize(target_dtype) def stats(self) -> Dict[str, Union[float, int, str]]: """Return diagnostic metrics for cache monitoring.""" return { "seq_len": self.seq_len, "precision": self.precision.value, "footprint_mb": round(self.current_mb, 4), "bytes_written": self.total_bytes_written, "bytes_read": self.total_bytes_read, "evicted_tokens": self.evicted_tokens_count, } @dataclass class KVTransferStats: total_migrated_bytes: int = 0 migration_latency_ms: float = 0.0 cache_hits: int = 0 cache_misses: int = 0 hot_tokens: int = 0 warm_tokens: int = 0 evicted_tokens: int = 0 fragmentation_ratio: float = 0.0 class HierarchicalAdaptiveKVCache(AdaptiveKVCache): """ Hierarchical Adaptive Memory Hierarchy for Q-TensorFormer KV Cache. Aligned with MetaKV (prompt-level constrained budget selection) and SeKV (hierarchical GPU/CPU residency). Supports: - 3 residency tiers: HOT (GPU), WARM (Host CPU-RAM), COLD (evicted). - Explicit PCIe transfer latency modeling: T_mig = Bytes / BW_PCIe + T_launch. - Attention utility + recency + sink token protection. - Request-level adaptive budget policy selection. - Fragmentation and migration profiling. """ def __init__( self, max_capacity: int = 2048, hot_capacity: int = 512, warm_capacity: int = 2048, window_size: int = 64, pcie_bandwidth_gb_s: float = 32.0, pcie_launch_overhead_ms: float = 0.015, default_precision: KVPrecision = KVPrecision.FP16, default_residency: KVResidencyTier = KVResidencyTier.HOT_GPU, ): super().__init__( max_capacity=max_capacity, window_size=window_size, default_precision=default_precision, ) self.hot_capacity = hot_capacity self.warm_capacity = warm_capacity self.pcie_bandwidth_gb_s = pcie_bandwidth_gb_s self.pcie_launch_overhead_ms = pcie_launch_overhead_ms self.residency = default_residency self.transfer_stats = KVTransferStats() self.warm_k: Optional[QuantizedKVTensor] = None self.warm_v: Optional[QuantizedKVTensor] = None def update_hierarchical( self, key: torch.Tensor, value: torch.Tensor, attention_weights: Optional[torch.Tensor] = None, target_precision: Optional[KVPrecision] = None, target_residency: Optional[KVResidencyTier] = None, prompt_budget_ratio: Optional[float] = None, ) -> Tuple[torch.Tensor, torch.Tensor]: """ Hierarchically updates cache with MetaKV-style prompt budget scaling and SeKV-style tiered GPU/CPU placement. """ if target_precision is not None: self.set_precision(target_precision) if target_residency is not None: self.residency = target_residency # MetaKV adaptation: adjust active hot capacity based on prompt budget ratio effective_hot_cap = self.hot_capacity if prompt_budget_ratio is not None: effective_hot_cap = max(8, int(self.hot_capacity * prompt_budget_ratio)) # Standard in-layer update out_k, out_v = self.update(key, value, attention_weights=attention_weights) curr_len = out_k.shape[-2] if curr_len > effective_hot_cap and self.residency == KVResidencyTier.WARM_CPU: # Demote overflow tokens from HOT to WARM CPU tier overflow_tokens = curr_len - effective_hot_cap migrated_bytes = overflow_tokens * out_k.shape[-1] * out_k.element_size() * 2 # Explicit PCIe migration latency calculation transfer_ms = (migrated_bytes / (self.pcie_bandwidth_gb_s * 1e9)) * 1000.0 + self.pcie_launch_overhead_ms self.transfer_stats.total_migrated_bytes += migrated_bytes self.transfer_stats.migration_latency_ms += transfer_ms self.transfer_stats.warm_tokens += overflow_tokens self.transfer_stats.hot_tokens = effective_hot_cap self.transfer_stats.cache_misses += 1 else: self.transfer_stats.hot_tokens = curr_len self.transfer_stats.cache_hits += 1 self.transfer_stats.fragmentation_ratio = round(self.evicted_tokens_count / max(1, curr_len + self.evicted_tokens_count), 4) return out_k, out_v def stats(self) -> Dict[str, Union[float, int, str]]: base_stats = super().stats() total_accesses = max(1, self.transfer_stats.cache_hits + self.transfer_stats.cache_misses) base_stats.update({ "residency_tier": self.residency.value, "hot_tokens": self.transfer_stats.hot_tokens, "warm_tokens": self.transfer_stats.warm_tokens, "migrated_bytes": self.transfer_stats.total_migrated_bytes, "migration_latency_ms": round(self.transfer_stats.migration_latency_ms, 3), "cache_hit_rate": round(self.transfer_stats.cache_hits / total_accesses, 3), "fragmentation_ratio": self.transfer_stats.fragmentation_ratio, }) return base_stats