Internals of Llama 3.2's New Attention Mechanism: Faster Inference for 2026 Code Models

import torch
import torch.nn as nn
import torch.nn.functional as F
from dataclasses import dataclass
from typing import Optional, Tuple

@dataclass
class SGLAConfig:
    hidden_size: int = 4096
    num_query_heads: int = 64
    num_latent_kv_heads: int = 8  # Compressed K/V heads per group
    num_query_groups: int = 8  # Q groups sharing latent K/V
    max_position_embeddings: int = 16384
    rope_theta: float = 10000.0
    dropout: float = 0.0
    compress_kv_cache: bool = True  # Enable latent compression

class SGLAAttention(nn.Module):
    def __init__(self, config: SGLAConfig):
        super().__init__()
        self.config = config
        self.hidden_size = config.hidden_size
        self.num_query_heads = config.num_query_heads
        self.num_latent_kv_heads = config.num_latent_kv_heads
        self.num_query_groups = config.num_query_groups
        self.head_dim = self.hidden_size // self.num_query_heads
        self.group_size = self.num_query_heads // self.num_query_groups

        # Validate config consistency
        if self.hidden_size % self.num_query_heads != 0:
            raise ValueError(f"hidden_size {self.hidden_size} must be divisible by num_query_heads {self.num_query_heads}")
        if self.num_query_heads % self.num_query_groups != 0:
            raise ValueError(f"num_query_heads {self.num_query_heads} must be divisible by num_query_groups {self.num_query_groups}")
        if self.num_latent_kv_heads > self.num_query_groups:
            raise ValueError(f"num_latent_kv_heads {self.num_latent_kv_heads} cannot exceed num_query_groups {self.num_query_groups}")

        # Q projection: full query heads
        self.q_proj = nn.Linear(self.hidden_size, self.num_query_heads * self.head_dim, bias=False)
        # Latent K/V compression: 2-layer MLP to compress original K/V heads to latent heads
        self.k_compress = nn.Sequential(
            nn.Linear(self.hidden_size, self.num_latent_kv_heads * self.head_dim),
            nn.GELU(),
            nn.Linear(self.num_latent_kv_heads * self.head_dim, self.num_latent_kv_heads * self.head_dim, bias=False)
        )
        self.v_compress = nn.Sequential(
            nn.Linear(self.hidden_size, self.num_latent_kv_heads * self.head_dim),
            nn.GELU(),
            nn.Linear(self.num_latent_kv_heads * self.head_dim, self.num_latent_kv_heads * self.head_dim, bias=False)
        )
        # Aligned head projection: map latent K/V heads to query groups
        self.k_align = nn.Linear(self.num_latent_kv_heads * self.head_dim, self.num_query_groups * self.head_dim, bias=False)
        self.v_align = nn.Linear(self.num_latent_kv_heads * self.head_dim, self.num_query_groups * self.head_dim, bias=False)
        # Output projection
        self.o_proj = nn.Linear(self.num_query_heads * self.head_dim, self.hidden_size, bias=False)
        self.dropout = nn.Dropout(config.dropout)

    def forward(
        self,
        hidden_states: torch.Tensor,
        past_key_value: Optional[Tuple[torch.Tensor, torch.Tensor]] = None,
        attention_mask: Optional[torch.Tensor] = None,
        position_ids: Optional[torch.Tensor] = None,
        output_attentions: bool = False,
    ) -> Tuple[torch.Tensor, Optional[Tuple[torch.Tensor, torch.Tensor]], Optional[torch.Tensor]]:
        # Validate input shapes
        batch_size, seq_len, _ = hidden_states.shape
        if hidden_states.shape[-1] != self.hidden_size:
            raise ValueError(f"Hidden states last dim {hidden_states.shape[-1]} must match hidden_size {self.hidden_size}")

        # Project queries: [batch, seq_len, num_q_heads * head_dim]
        query_states = self.q_proj(hidden_states)
        # Reshape to [batch, num_q_heads, seq_len, head_dim]
        query_states = query_states.view(batch_size, seq_len, self.num_query_heads, self.head_dim).transpose(1, 2)

        # Compress K/V to latent heads
        key_states_compressed = self.k_compress(hidden_states)  # [batch, seq_len, num_latent_kv * head_dim]
        value_states_compressed = self.v_compress(hidden_states)

        # Reshape latent K/V to [batch, num_latent_kv, seq_len, head_dim]
        key_states_compressed = key_states_compressed.view(batch_size, seq_len, self.num_latent_kv_heads, self.head_dim).transpose(1, 2)
        value_states_compressed = value_states_compressed.view(batch_size, seq_len, self.num_latent_kv_heads, self.head_dim).transpose(1, 2)

        # Align latent K/V to query groups: [batch, num_q_groups, seq_len, head_dim]
        # First reshape to [batch, seq_len, num_q_groups, head_dim] then transpose
        key_aligned = self.k_align(key_states_compressed.transpose(1, 2).contiguous().view(batch_size, seq_len, -1))  # [batch, seq_len, num_q_groups * head_dim]
        key_aligned = key_aligned.view(batch_size, seq_len, self.num_query_groups, self.head_dim).transpose(1, 2)
        value_aligned = self.v_align(value_states_compressed.transpose(1, 2).contiguous().view(batch_size, seq_len, -1))
        value_aligned = value_aligned.view(batch_size, seq_len, self.num_query_groups, self.head_dim).transpose(1, 2)

        # Handle past KV cache
        if past_key_value is not None:
            past_key, past_value = past_key_value
            # Past key shape: [batch, num_q_groups, past_seq_len, head_dim]
            key_aligned = torch.cat([past_key, key_aligned], dim=2)
            value_aligned = torch.cat([past_value, value_aligned], dim=2)

        past_key_value = (key_aligned, value_aligned) if self.config.compress_kv_cache else None

        # Expand aligned K/V to match query heads: [batch, num_q_heads, seq_len, head_dim]
        # Each query group has group_size heads, so repeat K/V group_size times
        key_states = key_aligned.repeat_interleave(self.group_size, dim=1)
        value_states = value_aligned.repeat_interleave(self.group_size, dim=1)

        # Apply RoPE (simplified for brevity; full implementation at https://github.com/meta-llama/llama-models)
        if position_ids is not None:
            query_states = self._apply_rope(query_states, position_ids)
            key_states = self._apply_rope(key_states, position_ids)

        # Compute attention scores
        attn_weights = torch.matmul(query_states, key_states.transpose(2, 3)) / (self.head_dim ** 0.5)

        if attention_mask is not None:
            attn_weights = attn_weights + attention_mask

        attn_weights = F.softmax(attn_weights, dim=-1, dtype=torch.float32).to(query_states.dtype)
        attn_weights = self.dropout(attn_weights)

        # Compute attention output
        attn_output = torch.matmul(attn_weights, value_states)  # [batch, num_q_heads, seq_len, head_dim]
        # Reshape to [batch, seq_len, num_q_heads * head_dim]
        attn_output = attn_output.transpose(1, 2).contiguous().view(batch_size, seq_len, -1)
        # Project to hidden size
        attn_output = self.o_proj(attn_output)

        if output_attentions:
            return attn_output, past_key_value, attn_weights
        return attn_output, past_key_value, None

    def _apply_rope(self, tensor: torch.Tensor, position_ids: torch.Tensor) -> torch.Tensor:
        # Simplified RoPE implementation; full code at https://github.com/meta-llama/llama-models
        batch_size, num_heads, seq_len, head_dim = tensor.shape
        position_embeddings = self._get_rope(position_ids, head_dim, self.config.max_position_embeddings, self.config.rope_theta)
        cos, sin = position_embeddings
        # Split tensor into even and odd dimensions
        even_dims = tensor[..., 0::2]
        odd_dims = tensor[..., 1::2]
        # Apply RoPE rotation
        rotated_even = even_dims * cos - odd_dims * sin
        rotated_odd = even_dims * sin + odd_dims * cos
        return torch.stack([rotated_even, rotated_odd], dim=-1).view(batch_size, num_heads, seq_len, head_dim)

    def _get_rope(self, position_ids: torch.Tensor, head_dim: int, max_pos: int, theta: float) -> Tuple[torch.Tensor, torch.Tensor]:
        # Generate RoPE embeddings for given position IDs
        inv_freq = 1.0 / (theta ** (torch.arange(0, head_dim, 2).float() / head_dim))
        position_ids_expanded = position_ids[:, None, :].float()
        freqs = torch.matmul(position_ids_expanded, inv_freq[None, :, None]).transpose(1, 2)
        emb = torch.cat([freqs, freqs], dim=-1)
        return emb.cos(), emb.sin()

import torch
import pickle
import os
from typing import Optional, Dict, Any
from pathlib import Path

class SGLACacheManager:
    """Manages compressed KV caches for SGLA attention, with disk offloading and corruption checks."""
    def __init__(
        self,
        max_cache_size_gb: float = 16.0,
        offload_dir: Optional[str] = None,
        num_latent_kv_heads: int = 8,
        num_query_groups: int = 8,
        head_dim: int = 64,
        device: str = "cuda"
    ):
        self.max_cache_size_bytes = max_cache_size_gb * 1024 * 1024 * 1024
        self.offload_dir = Path(offload_dir) if offload_dir else None
        self.num_latent_kv_heads = num_latent_kv_heads
        self.num_query_groups = num_query_groups
        self.head_dim = head_dim
        self.device = device
        self.cache: Dict[str, Tuple[torch.Tensor, torch.Tensor]] = {}  # key: (past_key, past_value)
        self.current_cache_size = 0

        if self.offload_dir and not self.offload_dir.exists():
            self.offload_dir.mkdir(parents=True, exist_ok=True)

    def get_cache(self, cache_key: str) -> Optional[Tuple[torch.Tensor, torch.Tensor]]:
        """Retrieve cached K/V pair, loading from disk if offloaded."""
        if cache_key not in self.cache:
            if self.offload_dir:
                offload_path = self.offload_dir / f"{cache_key}.pkl"
                if offload_path.exists():
                    try:
                        with open(offload_path, "rb") as f:
                            cached = pickle.load(f)
                        # Validate cache shape
                        past_key, past_value = cached
                        expected_key_shape = (1, self.num_query_groups, -1, self.head_dim)
                        if past_key.shape[1] != self.num_query_groups or past_key.shape[-1] != self.head_dim:
                            raise ValueError(f"Corrupted cache {cache_key}: invalid key shape {past_key.shape}")
                        if past_value.shape[1] != self.num_query_groups or past_value.shape[-1] != self.head_dim:
                            raise ValueError(f"Corrupted cache {cache_key}: invalid value shape {past_value.shape}")
                        self.cache[cache_key] = (past_key.to(self.device), past_value.to(self.device))
                        self.current_cache_size += past_key.numel() * past_key.element_size() + past_value.numel() * past_value.element_size()
                    except (pickle.PickleError, ValueError) as e:
                        print(f"Failed to load cache {cache_key}: {e}")
                        return None
                else:
                    return None
            else:
                return None
        return self.cache[cache_key]

    def set_cache(self, cache_key: str, past_key: torch.Tensor, past_value: torch.Tensor) -> None:
        """Store K/V pair in cache, offloading to disk if memory limit exceeded."""
        # Validate input tensors
        if past_key.device != torch.device(self.device):
            raise ValueError(f"past_key must be on {self.device}, got {past_key.device}")
        if past_value.device != torch.device(self.device):
            raise ValueError(f"past_value must be on {self.device}, got {past_value.device}")
        if past_key.shape[1] != self.num_query_groups:
            raise ValueError(f"past_key must have {self.num_query_groups} groups, got {past_key.shape[1]}")
        if past_value.shape[1] != self.num_query_groups:
            raise ValueError(f"past_value must have {self.num_query_groups} groups, got {past_value.shape[1]}")

        new_size = past_key.numel() * past_key.element_size() + past_value.numel() * past_value.element_size()
        # Check if adding new cache exceeds limit
        if self.current_cache_size + new_size > self.max_cache_size_bytes:
            self._evict_cache(new_size)

        self.cache[cache_key] = (past_key, past_value)
        self.current_cache_size += new_size

        # Offload to disk if configured
        if self.offload_dir:
            offload_path = self.offload_dir / f"{cache_key}.pkl"
            try:
                with open(offload_path, "wb") as f:
                    pickle.dump((past_key.cpu(), past_value.cpu()), f)
            except OSError as e:
                print(f"Failed to offload cache {cache_key}: {e}")

    def _evict_cache(self, required_size: int) -> None:
        """Evict least recently used caches until required size is available."""
        # Sort caches by key (simplified LRU; production uses access timestamps)
        sorted_keys = sorted(self.cache.keys())
        for key in sorted_keys:
            if self.current_cache_size + required_size <= self.max_cache_size_bytes:
                break
            past_key, past_value = self.cache.pop(key)
            self.current_cache_size -= (past_key.numel() * past_key.element_size() + past_value.numel() * past_value.element_size())
            # Remove offloaded file if exists
            if self.offload_dir:
                offload_path = self.offload_dir / f"{key}.pkl"
                if offload_path.exists():
                    offload_path.unlink()

    def clear(self) -> None:
        """Clear all caches from memory and disk."""
        self.cache.clear()
        self.current_cache_size = 0
        if self.offload_dir and self.offload_dir.exists():
            for f in self.offload_dir.glob("*.pkl"):
                f.unlink()

    def get_cache_stats(self) -> Dict[str, Any]:
        """Return cache usage statistics."""
        return {
            "current_size_gb": self.current_cache_size / (1024 ** 3),
            "max_size_gb": self.max_cache_size_bytes / (1024 ** 3),
            "num_entries": len(self.cache),
            "device": self.device
        }

# Example usage for a code completion request
if __name__ == "__main__":
    manager = SGLACacheManager(
        max_cache_size_gb=4.0,
        offload_dir="./sgla_cache",
        num_latent_kv_heads=8,
        num_query_groups=8,
        head_dim=64,
        device="cuda" if torch.cuda.is_available() else "cpu"
    )
    # Simulate storing a 1k token cache
    past_key = torch.randn(1, 8, 1024, 64, device=manager.device)
    past_value = torch.randn(1, 8, 1024, 64, device=manager.device)
    manager.set_cache("req_12345", past_key, past_value)
    print(f"Cache stats: {manager.get_cache_stats()}")
    retrieved = manager.get_cache("req_12345")
    assert retrieved is not None, "Cache retrieval failed"
    manager.clear()

import torch
import re
from typing import List, Optional
from transformers import AutoTokenizer

class CodeSparseMaskGenerator:
    """Generates block-sparse attention masks for code tokens, skipping non-code regions (comments, whitespace)."""
    def __init__(
        self,
        tokenizer: AutoTokenizer,
        block_size: int = 64,
        code_token_threshold: float = 0.7,  # 70% of block must be code tokens to attend
        supported_languages: List[str] = ["python", "javascript", "go", "rust"]
    ):
        self.tokenizer = tokenizer
        self.block_size = block_size
        self.code_token_threshold = code_token_threshold
        self.supported_languages = supported_languages
        # Precompile regex for comment detection (simplified for common languages)
        self.comment_regex = re.compile(
            r"#.*|//.*|/\*.*?\*/|",
            re.MULTILINE | re.DOTALL
        )
        # Token IDs for whitespace (tab, newline, space) – adjust per tokenizer
        self.whitespace_token_ids = set()
        for ws_char in [" ", "\t", "\n", "\r", "\f"]:
            ws_id = self.tokenizer.encode(ws_char, add_special_tokens=False)
            if ws_id:
                self.whitespace_token_ids.add(ws_id[0])

    def generate_mask(
        self,
        input_ids: torch.Tensor,
        language: str,
        attention_mask: Optional[torch.Tensor] = None
    ) -> torch.Tensor:
        """
        Generate sparse attention mask: 0 for masked (skip attention), 1 for attend.
        Shape: [batch_size, 1, seq_len, seq_len] (broadcastable to attention scores)
        """
        batch_size, seq_len = input_ids.shape
        if seq_len % self.block_size != 0:
            raise ValueError(f"seq_len {seq_len} must be divisible by block_size {self.block_size}")
        if language not in self.supported_languages:
            raise ValueError(f"Unsupported language {language}. Supported: {self.supported_languages}")

        # Initialize full attention mask (all 1s)
        mask = torch.ones(batch_size, 1, seq_len, seq_len, dtype=torch.bool, device=input_ids.device)

        for batch_idx in range(batch_size):
            # Decode tokens to text to detect comments
            try:
                text = self.tokenizer.decode(input_ids[batch_idx], skip_special_tokens=True)
            except IndexError as e:
                raise ValueError(f"Invalid token IDs in batch {batch_idx}: {e}")

            # Find non-code regions (comments + whitespace)
            non_code_spans = self._get_non_code_spans(text, language)
            # Get token positions corresponding to non-code spans
            non_code_positions = self._map_spans_to_tokens(non_code_spans, input_ids[batch_idx], text)

            # Split into blocks
            num_blocks = seq_len // self.block_size
            for block_idx in range(num_blocks):
                start = block_idx * self.block_size
                end = start + self.block_size
                block_tokens = input_ids[batch_idx, start:end].tolist()
                # Count code tokens in block: not whitespace, not in non_code_positions
                code_token_count = 0
                for pos_in_block, token_id in enumerate(block_tokens):
                    global_pos = start + pos_in_block
                    if global_pos not in non_code_positions and token_id not in self.whitespace_token_ids:
                        code_token_count += 1
                # If code token ratio below threshold, mask entire block's attention
                if code_token_count / self.block_size < self.code_token_threshold:
                    # Mask all queries in this block attending to any key
                    mask[batch_idx, 0, start:end, :] = 0
                    # Also mask all keys in this block being attended by any query
                    mask[batch_idx, 0, :, start:end] = 0

        # Apply original attention mask (if provided) to mask padding tokens
        if attention_mask is not None:
            # Expand attention_mask to [batch, 1, seq_len, seq_len]
            expanded_mask = attention_mask[:, None, :, None].bool() & attention_mask[:, None, None, :].bool()
            mask = mask & expanded_mask

        # Convert to additive mask (0 for attend, -inf for mask) for softmax
        additive_mask = torch.zeros_like(mask, dtype=torch.float32)
        additive_mask[~mask] = -torch.inf
        return additive_mask

    def _get_non_code_spans(self, text: str, language: str) -> List[Tuple[int, int]]:
        """Return list of (start_char, end_char) spans for non-code regions (comments)."""
        spans = []
        # Detect comments
        for match in self.comment_regex.finditer(text):
            spans.append((match.start(), match.end()))
        # TODO: Add string literal detection for full non-code coverage
        return sorted(spans, key=lambda x: x[0])

    def _map_spans_to_tokens(self, spans: List[Tuple[int, int]], input_ids: torch.Tensor, text: str) -> set:
        """Map character spans to token positions (simplified; production uses offset mapping)."""
        non_code_positions = set()
        # Get offset mapping from tokenizer
        try:
            encoding = self.tokenizer(
                text,
                return_offsets_mapping=True,
                add_special_tokens=False
            )
        except TypeError:
            # Tokenizer doesn't support offset mapping; fall back to approximate mapping
            return non_code_positions

        offset_mapping = encoding["offset_mapping"]
        for start_char, end_char in spans:
            for token_idx, (token_start, token_end) in enumerate(offset_mapping):
                if token_start >= start_char and token_end <= end_char:
                    # Map to input_ids position (adjust for batch)
                    non_code_positions.add(token_idx)
        return non_code_positions

# Example usage with Python code
if __name__ == "__main__":
    tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-3.2-7B-Instruct")
    generator = CodeSparseMaskGenerator(tokenizer, block_size=64, supported_languages=["python"])
    # Sample Python code with a comment
    code = """def add(a, b):
    # This is a comment
    return a + b
"""
    input_ids = tokenizer.encode(code, return_tensors="pt")
    mask = generator.generate_mask(input_ids, language="python")
    print(f"Mask shape: {mask.shape}, Masked positions: {(mask == -torch.inf).sum().item()}")

from torch.profiler import profile, ProfilerActivity

model = SGLAAttention(SGLAConfig())
input = torch.randn(1, 128, 4096)
with profile(activities=[ProfilerActivity.CPU, ProfilerActivity.CUDA]) as prof:
    model(input)
prof.export_chrome_trace("sgla_trace.json")

generator = CodeSparseMaskGenerator(tokenizer, block_size=32)
mask = generator.generate_mask(input_ids, language="rust")

manager = SGLACacheManager(max_cache_size_gb=16, offload_dir="/nvme/sgla_cache")