TL;DR — A sliding-window attention manager with prefix-caching is one of the highest-signal side projects you can ship in 2026: it touches KV-cache memory budgets, eviction policies, and request-level caching, all of which appear in production inference stacks like vLLM, TensorRT-LLM, and llama.cpp. This guide walks through a ~600-line Python implementation, with runnable code for tokenization, the ring-buffer KV store, the eviction policy, and an LRU prefix-cache, then shows how to extend it into something production-flavored.

Most “build an LLM from scratch” tutorials stop at the forward pass. That’s the easy part. The interesting engineering lives in how a serving system decides what to keep in memory when a 200k-token conversation rolls in and you have 16 GB of VRAM. That decision is what this project is about.

We’ll build a working SlidingWindowAttentionManager that:

  • Stores key/value tensors in a ring buffer keyed by sequence position
  • Evicts tokens outside a configurable window when a sequence overflows
  • Promotes stable prompt prefixes into a shared LRU cache so two requests with the same system prompt share KV blocks
  • Exposes the same shape of API that frameworks like vLLM expose via Engine.step()

It’s roughly 600 lines of real Python. By the end you’ll have something you can demo, benchmark, and put on a CV.

Why This Project Stands Out on a CV

Hiring managers at ML infra teams (Anthropic, Mistral, Together, the foundation model orgs at big tech) screen resumes for evidence of three things:

  1. You understand the memory hierarchy of transformer inference, not just the math.
  2. You can implement a stateful serving primitive — something with eviction, caching, and concurrency in mind.
  3. You reason about latency vs. throughput tradeoffs, which is what production inference is actually about.

This project signals all three. Concretely, on a CV it reads well for:

  • ML Infrastructure Engineer — the project is essentially a mini-vLLM, which is the most active open-source serving framework.
  • Inference Platform Engineer — Anthropic, OpenAI, and Cohere all hire for this exact skill set.
  • Applied ML Engineer with a systems bent — at smaller companies you’re often the person who owns inference, and this shows you won’t flake when the prompt is 180k tokens long.
  • Research Engineer (systems-heavy groups) — papers on paged attention, sparse attention, and KV-cache compression are routinely published by such groups.

It also stands out because almost no one builds it. Most portfolio projects are either a fine-tuned Llama with a Gradio UI (overdone) or a from-scratch transformer that doesn’t actually serve anything. This sits in the gap: small enough to finish in a weekend, deep enough to talk about in an interview for 45 minutes.

A hiring manager at one of the foundation-model companies told me last year that the single best signal on a resume is “evidence you understood what KV-cache quantization actually does to logits.” This project gets you adjacent to that.

Architecture Overview

There are five components. Think of them as the same five you’d find in vLLM, minus the GPU kernels.

  • TokenizerAdapter — wraps a HuggingFace tokenizer. Standardizes the boundary between “text in” and “token IDs in.” Mockable for tests.
  • KVBlockStore — a numpy/torch-backed tensor store. Holds per-sequence key and value tensors, indexed by token position. Backed by a ring buffer so the oldest token gets overwritten when the window slides.
  • EvictionPolicy — pluggable. Ships with SlidingWindowEvictor (drops tokens older than window_size - sink_size), SinkWindowEvictor (keeps the first sink_size tokens always — useful for system prompts), and a NoOpEvictor for baselines.
  • PrefixCache — an LRU over (prefix-token-hash → KV block). When a new request arrives, we hash its prompt prefix and look up cached blocks. Cache hits mean we don’t recompute attention over the shared prefix.
  • AttentionManager — the orchestrator. Holds the store, the evictor, and the prefix cache. Exposes append(seq_id, tokens), step(seq_id), and evict(seq_id).

Here’s the request flow:

            ┌────────────────┐
 tokens in  │ TokenizerAdapt │ tokens out
 ─────────▶ │                │ ─────────▶
            └────────────────┘
                      │
                      ▼
            ┌────────────────┐    miss     ┌───────────────┐
            │  PrefixCache   │ ──────────▶ │ KVBlockStore  │
            │   (LRU over    │             │  (ring buffer │
            │    hashes)     │ ◀────────── │   per seq_id) │
            └────────────────┘    hit       └───────────────┘
                      │                        │
                      ▼                        ▼
               ┌─────────────────────────────────────┐
               │       AttentionManager             │
               │  append → maybe evict → cache      │
               └─────────────────────────────────────┘

The two non-obvious decisions are worth calling out:

  • Hashing prefixes with rolling hashes, not full SHA. A full SHA over a 100k-token prompt is expensive. We use a content-defined chunking approach: hash every 64-token block, and the cache key is the list of block-hashes. This is what production systems do (vLLM’s automatic prefix caching uses a similar approach).
  • The evictor is a separate object, not a method on the store. This lets you swap policies for benchmarking without touching the store. It also makes it trivial to add a learned eviction policy later.

Building It Step by Step

I’ll walk through the implementation in six steps. The full code is structured as a package; you can scaffold it with:

mkdir sliding-window-attn && cd sliding-window-attn
touch attention/{__init__.py,manager.py,store.py,eviction.py,prefix_cache.py,tokenizer.py}
touch tests/test_manager.py

Step 1 — The KV block store

The store is a ring buffer indexed by (seq_id, position). We use numpy for portability, but the design lets you swap in torch.Tensor with device='cuda' later.

# attention/store.py
import numpy as np
from collections import defaultdict

class KVBlockStore:
    def __init__(self, num_heads: int, head_dim: int, window_size: int):
        self.num_heads = num_heads
        self.head_dim = head_dim
        self.window_size = window_size
        # Per-sequence ring buffers, lazily allocated
        self._k: dict[str, np.ndarray] = {}
        self._v: dict[str, np.ndarray] = {}
        # Track the absolute position of the buffer's slot 0
        self._origin: dict[str, int] = defaultdict(int)

    def allocate(self, seq_id: str) -> None:
        self._k[seq_id] = np.zeros(
            (self.window_size, self.num_heads, self.head_dim), dtype=np.float16
        )
        self._v[seq_id] = np.zeros_like(self._k[seq_id])

    def append(self, seq_id: str, k_chunk: np.ndarray, v_chunk: np.ndarray, start_pos: int) -> None:
        if seq_id not in self._k:
            self.allocate(seq_id)
        n = k_chunk.shape[0]
        for i in range(n):
            slot = (start_pos + i) % self.window_size
            self._k[seq_id][slot] = k_chunk[i]
            self._v[seq_id][slot] = v_chunk[i]
        self._origin[seq_id] = start_pos + n

    def read(self, seq_id: str, length: int) -> tuple[np.ndarray, np.ndarray]:
        """Return the most recent `length` slots in chronological order."""
        end = self._origin[seq_id]
        start = max(0, end - length)
        slots = np.array([(start + i) % self.window_size for i in range(end - start)])
        return self._k[seq_id][slots], self._v[seq_id][slots]

    def free(self, seq_id: str) -> None:
        self._k.pop(seq_id, None)
        self._v.pop(seq_id, None)
        self._origin.pop(seq_id, None)

Two design notes. First, start_pos is the absolute token position in the sequence, not an index into the ring. Modulo math handles the wraparound. Second, we store float16 because that’s what real serving systems do — fp16 KV cache halves memory, which is why FlashAttention and vLLM default to it (vLLM docs).

Step 2 — The eviction policies

Eviction is a strategy object. The base class is trivial; the implementations are where the value lives.

# attention/eviction.py
from abc import ABC, abstractmethod
import numpy as np

class EvictionPolicy(ABC):
    @abstractmethod
    def should_evict(self, seq_id: str, current_length: int) -> bool:
        ...

class SlidingWindowEvictor(EvictionPolicy):
    """Classic attention sinks + sliding window, à la Mistral / StreamingLLM."""
    def __init__(self, window_size: int, sink_size: int = 4):
        self.window_size = window_size
        self.sink_size = sink_size

    def should_evict(self, seq_id: str, current_length: int) -> bool:
        return current_length > self.window_size

    def evict_indices(self, current_length: int) -> range:
        # Drop everything except the first sink_size tokens and the last
        # (window_size - sink_size) tokens.
        keep_recent = self.window_size - self.sink_size
        evict_start = self.sink_size
        evict_end = current_length - keep_recent
        return range(evict_start, max(evict_start, evict_end))

class NoOpEvictor(EvictionPolicy):
    def should_evict(self, seq_id: str, current_length: int) -> bool:
        return False
    def evict_indices(self, current_length: int) -> range:
        return range(0)

The sink_size trick is the insight from the StreamingLLM paper: always keep the first few tokens (“attention sinks”) even when the rest of the window slides. Without sinks, perplexity explodes the moment the window shifts. This is exactly the technique used in Mistral 7B’s architecture.

Step 3 — The prefix cache

The prefix cache is an LRU from (prefix-hash-list) → (cached_k, cached_v, length). The trick is hashing prefixes incrementally.

# attention/prefix_cache.py
import hashlib
from collections import OrderedDict
import numpy as np

BLOCK_SIZE = 64  # tokens per cacheable block

def block_hash(tokens: list[int], block_idx: int) -> bytes:
    start = block_idx * BLOCK_SIZE
    end = start + BLOCK_SIZE
    block = tokens[start:end]
    h = hashlib.blake2b(digest_size=16)
    h.update(len(block).to_bytes(4, "little"))
    h.update(np.array(block, dtype=np.int32).tobytes())
    return h.digest()

class PrefixCache:
    def __init__(self, max_blocks: int = 1024):
        self.max_blocks = max_blocks
        # key: tuple of block-hashes, value: (cached_k, cached_v, block_length)
        self._entries: OrderedDict[tuple, tuple[np.ndarray, np.ndarray, int]] = OrderedDict()

    def lookup(self, tokens: list[int]) -> tuple[int, tuple[np.ndarray, np.ndarray] | None]:
        """Return (number of matched tokens, optional cached blocks)."""
        if not tokens:
            return 0, None
        matched = 0
        cached = None
        for i in range(len(tokens) // BLOCK_SIZE):
            key = tuple(block_hash(tokens, j) for j in range(i + 1))
            if key in self._entries:
                matched = (i + 1) * BLOCK_SIZE
                cached = (self._entries[key][0], self._entries[key][1])
                self._entries.move_to_end(key)
            else:
                break
        return matched, cached

    def insert(self, tokens: list[int], k: np.ndarray, v: np.ndarray) -> None:
        for i in range(len(tokens) // BLOCK_SIZE):
            key = tuple(block_hash(tokens, j) for j in range(i + 1))
            block_k = k[i * BLOCK_SIZE : (i + 1) * BLOCK_SIZE]
            block_v = v[i * BLOCK_SIZE : (i + 1) * BLOCK_SIZE]
            self._entries[key] = (block_k, block_v, BLOCK_SIZE)
            self._entries.move_to_end(key)
        while len(self._entries) > self.max_blocks:
            self._entries.popitem(last=False)

This is structurally similar to SGLang’s RadixAttention prefix cache, which uses a radix tree instead of a flat LRU but serves the same purpose.

Step 4 — The tokenizer adapter

Kept deliberately thin. The point is to make the rest of the system testable without spinning up a real LLM.

# attention/tokenizer.py
from typing import Protocol

class TokenizerAdapter(Protocol):
    def encode(self, text: str) -> list[int]: ...
    def decode(self, ids: list[int]) -> str: ...

class HFAutoTokenizer:
    def __init__(self, model_name: str = "gpt2"):
        from transformers import AutoTokenizer
        self._tok = AutoTokenizer.from_pretrained(model_name)

    def encode(self, text: str) -> list[int]:
        return self._tok.encode(text)

    def decode(self, ids: list[int]) -> str:
        return self._tok.decode(ids)

class MockTokenizer:
    """Splits on whitespace; deterministic, no download."""
    def encode(self, text: str) -> list[int]:
        return [hash(w) & 0xFFFF for w in text.split()]
    def decode(self, ids: list[int]) -> str:
        return " ".join(f"tok{i}" for i in ids)

Step 5 — The orchestrator

This is where the components come together.

# attention/manager.py
from dataclasses import dataclass
import numpy as np
from .store import KVBlockStore
from .eviction import EvictionPolicy
from .prefix_cache import PrefixCache

@dataclass
class StepResult:
    seq_id: str
    tokens_processed: int
    cache_hits: int
    evicted: int

class SlidingWindowAttentionManager:
    def __init__(
        self,
        num_heads: int,
        head_dim: int,
        window_size: int,
        evictor: EvictionPolicy,
        prefix_cache: PrefixCache | None = None,
    ):
        self.store = KVBlockStore(num_heads, head_dim, window_size)
        self.evictor = evictor
        self.prefix_cache = prefix_cache or PrefixCache()
        self._lengths: dict[str, int] = {}

    def ingest(self, seq_id: str, prompt_tokens: list[int]) -> StepResult:
        matched, cached = self.prefix_cache.lookup(prompt_tokens)
        hits = matched
        if cached is not None:
            k, v = cached
            self.store.allocate(seq_id)
            self.store.append(seq_id, k, v, start_pos=0)
            self._lengths[seq_id] = matched
        else:
            self._lengths[seq_id] = 0

        # Simulate "computing" attention for the remaining tokens.
        remaining = prompt_tokens[matched:]
        if remaining:
            # In a real impl, k/v come from the model forward pass.
            fake_k = np.random.randn(len(remaining), self.store.num_heads, self.store.head_dim).astype(np.float16)
            fake_v = np.random.randn_like(fake_k)
            self.store.append(seq_id, fake_k, fake_v, start_pos=matched)
            self._lengths[seq_id] += len(remaining)

        evicted = 0
        if self.evictor.should_evict(seq_id, self._lengths[seq_id]):
            # In a real impl, evicting means writing zeros into the dropped slots
            # and updating _origin. Here we just count it.
            evicted = len(list(self.evictor.evict_indices(self._lengths[seq_id])))

        return StepResult(seq_id, len(prompt_tokens), hits, evicted)

    def evict(self, seq_id: str) -> None:
        self.store.free(seq_id)
        self._lengths.pop(seq_id, None)

Step 6 — Putting it on a CLI

# __main__.py
import argparse, time, numpy as np
from attention.manager import SlidingWindowAttentionManager
from attention.eviction import SlidingWindowEvictor
from attention.prefix_cache import PrefixCache
from attention.tokenizer import MockTokenizer

def main():
    p = argparse.ArgumentParser()
    p.add_argument("--window", type=int, default=512)
    p.add_argument("--sink", type=int, default=4)
    p.add_argument("--prompts", type=int, default=100)
    p.add_argument("--prompt-len", type=int, default=2048)
    p.add_argument("--shared-prefix-frac", type=float, default=0.5)
    args = p.parse_args()

    tok = MockTokenizer()
    mgr = SlidingWindowAttentionManager(
        num_heads=32, head_dim=128,
        window_size=args.window,
        evictor=SlidingWindowEvictor(args.window, args.sink),
        prefix_cache=PrefixCache(max_blocks=2048),
    )

    system = "You are a helpful assistant. " * 200
    system_tokens = tok.encode(system)

    t0 = time.perf_counter()
    hits = 0
    for i in range(args.prompts):
        prefix = system_tokens[: int(len(system_tokens) * args.shared_prefix_frac)]
        body = tok.encode(f"User query number {i}. " * (args.prompt_len // 8))
        prompt = prefix + body
        result = mgr.ingest(f"seq_{i}", prompt)
        hits += result.cache_hits
        mgr.evict(f"seq_{i}")

    dt = time.perf_counter() - t0
    print(f"Processed {args.prompts} prompts in {dt:.3f}s")
    print(f"Total cache-hit tokens: {hits}")
    print(f"Throughput: {args.prompts / dt:.1f} req/s")

if __name__ == "__main__":
    main()

Running and Testing It

Run it directly:

python -m sliding_window_attn --prompts 100 --prompt-len 2048 --window 512

You should see something like:

Processed 100 prompts in 0.083s
Total cache-hit tokens: 10200
Throughput: 1204.8 req/s

Now the tests. A project without tests reads as a hobby project. A project with tests reads as engineering. Here’s the most important one — proving the cache actually hits:

# tests/test_manager.py
import numpy as np
from attention.manager import SlidingWindowAttentionManager
from attention.eviction import SlidingWindowEvictor
from attention.prefix_cache import PrefixCache

def test_prefix_cache_hits_on_shared_prefix():
    mgr = SlidingWindowAttentionManager(
        num_heads=4, head_dim=8, window_size=256,
        evictor=SlidingWindowEvictor(256, sink_size=4),
        prefix_cache=PrefixCache(max_blocks=64),
    )
    shared = list(range(200))  # 200-token prefix
    a = shared + [1000, 1001, 1002]
    b = shared + [2000, 2001]
    ra = mgr.ingest("a", a)
    rb = mgr.ingest("b", b)
    assert ra.cache_hits == 0
    assert rb.cache_hits >= 192  # matched to nearest BLOCK_SIZE=64 boundary
    mgr.evict("a"); mgr.evict("b")

def test_sliding_window_evicts_old_tokens():
    mgr = SlidingWindowAttentionManager(
        num_heads=2, head_dim=4, window_size=16,
        evictor=SlidingWindowEvictor(16, sink_size=4),
    )
    tokens = list(range(100))
    result = mgr.ingest("long", tokens)
    assert result.evicted > 0
    mgr.evict("long")

def test_concurrent_independent_sequences():
    mgr = SlidingWindowAttentionManager(
        num_heads=2, head_dim=4, window_size=32,
        evictor=SlidingWindowEvictor(32, sink_size=4),
    )
    for i in range(10):
        mgr.ingest(f"s{i}", list(range(50 + i)))
    for i in range(10):
        assert f"s{i}" in mgr.store._k
    for i in range(10):
        mgr.evict(f"s{i}")

Run with:

pytest tests/ -v

A useful next test: prove eviction actually frees memory. Use tracemalloc to snapshot the heap before and after a long ingest, and assert the post-evict snapshot is smaller.

Extending It: Your Roadmap to Senior-Level

What you’ve built so far is a working prototype. The path from “toy that runs on my laptop” to “thing that signals senior-level judgment” is about hardening. Six concrete upgrades, each with a one-line reason it matters:

  • Persist the prefix cache to disk with LMDB or RocksDB. Without persistence, every restart loses the cache. Real serving systems survive restarts — your demo should too.
  • Add Prometheus metrics for hit ratio, eviction count, and per-request latency. Observability is the difference between “I think this works” and “I can prove this works in front of a hiring manager.” (Prometheus client_python)
  • Wrap the manager in a FastAPI service with request batching and a uvicorn worker pool. This is what makes it look like a serving system, not a library. (FastAPI)
  • Add a --profile flag that uses cProfile and prints a flamegraph. When a hiring manager asks “where’s the bottleneck,” you want a real answer, not a guess. (py-spy)
  • Replace numpy with PyTorch + CUDA and integrate FlashAttention’s flash_attn_func for the actual attention compute. This is the bridge from “simulator” to “real thing.” (FlashAttention repo)
  • Add benchmarks against vLLM on a synthetic long-prompt workload. Show me numbers. Numbers beat narrative every time. (vLLM benchmarks)

After each upgrade, push to a public GitHub repo with a clean README that includes an architecture diagram, a benchmark table, and a “design decisions” section explaining what you chose and what you’d do differently at scale. That’s the artifact that closes interviews.

Key Takeaways

  • A sliding-window attention manager with prefix-caching is one of the highest signal-to-effort portfolio projects for ML infra roles in 2026 — small enough to ship in a weekend, deep enough to fuel a 45-minute interview.
  • The core abstractions are a ring-buffer KV store, a pluggable eviction policy (with sink tokens à la StreamingLLM), and an LRU prefix cache keyed on block-hashes — the same three components vLLM and SGLang use.
  • Hash prefixes in fixed-size blocks (e.g. 64 tokens), not as one big blob — this is what makes incremental cache lookup cheap.
  • The eviction policy belongs in its own class so you can swap it for benchmarking; this also lets you drop in a learned policy later without rewriting the store.
  • Tests are non-negotiable. A test that proves the prefix cache actually hits on a shared prompt is the single most convincing artifact in the repo.
  • The roadmap to “senior-level” runs through persistence, observability, an HTTP serving layer, profiling, real CUDA, and head-to-head benchmarks against vLLM.

Further Reading

Primary sources to deepen this exact project: