TL;DR — This post walks you through building a pure‑Python block‑sparse sliding‑window attention module that replicates Longformer’s linear‑time behavior on long sequences. You’ll end up with runnable code, a test harness, and a roadmap to turn the prototype into a production‑grade component.
In a world where transformers are expected to ingest whole books, a naive O(N²) attention quickly becomes a bottleneck. Longformer introduced a block‑sparse pattern that keeps the cost linear in sequence length while preserving local context. By reimplementing that pattern from scratch in pure Python, you demonstrate a rare combination of algorithmic insight, low‑level coding skill, and systems awareness—exactly the profile hiring managers look for in ML infrastructure roles.
Why This Project Stands Out on a CV
- Algorithmic depth – you’ll write the core attention loop, not just call a library.
- Performance engineering – block decomposition and window masking teach cache‑friendly design.
- Systems thinking – you’ll see how memory, compute, and sparsity interact at scale.
- Relevance to production – the same pattern powers real‑world services at companies like Cohere and Anthropic.
- Portfolio signal – a runnable, tested implementation is far more compelling than a theoretical essay.
Architecture Overview
The implementation is composed of three cooperating parts:
- Block‑Sparse Mask Generator – divides the sequence into fixed‑size blocks (e.g., 64 tokens) and decides which blocks are “active” for each query.
- Sliding‑Window Mask – limits attention to a ±W token neighborhood, preserving local context.
- Sparse Attention Engine – combines the two masks, computes scaled dot‑product attention only on the allowed positions, and returns the weighted sum.
A high‑level flow looks like this:
Input embeddings (L, H) ──► BlockMaskGenerator ──► Block mask (L, L)
│
▼
SlidingWindowMask ──► Window mask (L, L)
│
▼
Combine masks (logical AND)
│
▼
SparseAttention (scaled dot‑product)
│
▼
Output (L, H)
Each component is a pure‑Python class, making the pipeline easy to inspect, unit‑test, and extend.
Building It Step by Step
Step 1 – Set up the environment
No external dependencies are required; the standard library is sufficient. Optionally, you may install numpy for numerical comparisons, but the core logic stays pure Python.
python -m venv venv
source venv/bin/activate
# optional: pip install numpy
Step 2 – Define constants and data structures
Create a new file block_sparse_attention.py. Start with the hyper‑parameters that control sparsity.
# block_sparse_attention.py
from typing import Tuple
import math
BLOCK_SIZE = 64 # number of tokens per block
WINDOW_SIZE = 256 # total half‑width of the sliding window
HIDDEN_DIM = 512 # size of the embedding dimension
Step 3 – Implement the mask generators
The block‑sparse mask marks a block as active if its start index lies within the window of the query.
class BlockSparseMask:
def __init__(self, seq_len: int, block_size: int = BLOCK_SIZE):
self.seq_len = seq_len
self.block_size = block_size
self.num_blocks = (seq_len + block_size - 1) // block_size
def block_start(self, block_idx: int) -> int:
return block_idx * self.block_size
def is_block_active(self, query_pos: int, block_idx: int) -> bool:
# A block is active if any of its tokens fall inside the window
start = self.block_start(block_idx)
end = min(start + self.block_size, self.seq_len)
return abs(query_pos - start) <= WINDOW_SIZE or \
abs(query_pos - (end - 1)) <= WINDOW_SIZE
def mask(self) -> list[list[bool]]:
"""Return a boolean matrix M[i][j] where True means token j may attend to token i."""
m = [[False] * self.seq_len for _ in range(self.seq_len)]
for i in range(self.seq_len):
for b in range(self.num_blocks):
if self.is_block_active(i, b):
start = self.block_start(b)
end = min(start + self.block_size, self.seq_len)
for j in range(start, end):
# additionally enforce the sliding window
if abs(i - j) <= WINDOW_SIZE:
m[i][j] = True
return m
The sliding‑window mask is simply a band around the diagonal; the code above already incorporates it inside is_block_active, but you can separate the two concerns if you prefer.
Step 4 – Write the sparse attention forward pass
We’ll compute the scaled dot‑product attention only for positions where the mask is True. To keep the implementation clear, we use Python lists; for production you’d swap in NumPy or a JIT‑compiled kernel.
class SparseAttention:
def __init__(self, hidden_dim: int = HIDDEN_DIM, scale: float = None):
self.hidden_dim = hidden_dim
self.scale = scale or 1.0 / math.sqrt(hidden_dim)
def _softmax(self, row: list[float]) -> list[float]:
max_val = max(row)
exps = [math.exp(v - max_val) for v in row]
sum_exps = sum(exps)
return [e / sum_exps for e in exps]
def forward(self, queries: list[list[float]], keys: list[list[float]],
values: list[list[float]], mask: list[list[bool]]) -> list[list[float]]:
"""
queries, keys, values: shape (seq_len, hidden_dim)
mask: boolean matrix (seq_len, seq_len)
Returns: output of same shape as queries
"""
seq_len = len(queries)
output = [[0.0] * self.hidden_dim for _ in range(seq_len)]
for i in range(seq_len):
# collect scores for allowed positions
scores = []
indices = []
for j in range(seq_len):
if mask[i][j]:
# dot product
dot = sum(q * k for q, k in zip(queries[i], keys[j]))
scores.append(dot * self.scale)
indices.append(j)
if not scores:
continue # no attention possible, output stays zero
probs = self._softmax(scores)
# weighted sum of values
for idx, prob in zip(indices, probs):
for d in range(self.hidden_dim):
output[i][d] += prob * values[idx][d]
return output
Step 5 – Glue everything together and test
Add a small driver that creates random embeddings, builds the mask, runs the forward pass, and compares against a dense baseline (optional, requires NumPy).
def test_sparse_attention():
import random
seq_len = 128
hidden = HIDDEN_DIM
# random embeddings
queries = [[random.random() for _ in range(hidden)] for _ in range(seq_len)]
keys = [[random.random() for _ in range(hidden)] for _ in range(seq_len)]
values = [[random.random() for _ in range(hidden)] for _ in range(seq_len)]
mask_gen = BlockSparseMask(seq_len)
mask = mask_gen.mask()
attn = SparseAttention(hidden)
out = attn.forward(queries, keys, values, mask)
# sanity check: output shape
assert len(out) == seq_len and len(out[0]) == hidden
print("Sparse attention output shape OK")
if __name__ == "__main__":
test_sparse_attention()
Run the module:
python block_sparse_attention.py
You should see Sparse attention output shape OK printed to the console.
Running and Testing It
- Execute the unit test – the script above already verifies shape correctness.
- Compare with dense attention – if NumPy is installed, you can compute the full O(N²) result and assert that the sparse version matches within a tolerance for positions that are allowed by the mask.
- Profile the runtime – use Python’s
timeitto measure how the sparse forward scales with sequence length; you should observe roughly linear growth rather than quadratic.
import timeit
def bench():
seq = 512
# ... setup as above ...
mask = BlockSparseMask(seq).mask()
start = timeit.default_timer()
_ = SparseAttention().forward(queries, keys, values, mask)
end = timeit.default_timer()
print(f"Sparse forward on seq={seq} took {end-start:.4f}s")
Running bench() demonstrates that the block‑sparse design indeed reduces compute cost.
Extending It: Your Roadmap to Senior-Level
- Persistence layer – store the learned block‑sparse weights in an SQLite database, enabling fast reload without re‑compiling the mask logic. This matters for model serving where startup latency is critical.
- Horizontal scaling – shard the sequence dimension across multiple workers using Ray, allowing the attention module to handle million‑token inputs. Scaling out is essential for real‑time inference at high throughput.
- Observability – expose metrics (latency, memory usage, mask sparsity) to a Prometheus endpoint; observability lets you detect regressions before users notice.
- Fault tolerance – wrap the forward pass in retry logic with exponential backoff; in distributed settings, transient node failures are the norm, not the exception.
- Benchmarking suite – build a micro‑benchmark harness that compares sparse vs. dense attention across sequence lengths, block sizes, and hardware (CPU/GPU). Concrete numbers are what convince stakeholders to adopt your solution.
- GPU acceleration – port the inner loops to CuPy or a custom CUDA kernel; this transforms the prototype into a production‑grade component that can be served alongside existing PyTorch models.
Key Takeaways
- You now have a pure‑Python implementation of block‑sparse sliding‑window attention that mirrors Longformer’s efficiency.
- The code is modular, making it easy to swap mask strategies or integrate into larger pipelines.
- By following the extension roadmap, you can evolve this prototype into a scalable, observable, and fault‑tolerant service.
- The project demonstrates algorithmic design, performance optimization, and systems engineering—skills that resonate with hiring managers.
- Use this as a portfolio piece that showcases end‑to‑end ownership from research idea to runnable code.
Further Reading
- Longformer: The Long-Document Transformer – the original paper introducing block‑sparse attention.
- PyTorch Scaled Dot‑Product Attention Documentation – reference for the dense baseline.
- Hugging Face Transformers — Longformer Model – practical usage and configuration details.
- Ray Documentation – for building distributed, fault‑tolerant Python applications.