If you've ever watched a RAG pipeline confidently return the wrong answer because it retrieved a chunk that said "the policy applies" without also knowing which policy, to whom, and under what conditions — you've already felt the problem this lesson solves. Standard chunking splits documents before embedding them, which means every chunk loses the context of everything around it. A sentence that's perfectly clear in its original position becomes ambiguous noise in isolation.
Late chunking flips this process. Instead of splitting first and embedding second, you embed the full document first — taking advantage of models with long context windows — and then chunk after the attention mechanism has already let every token attend to every other token. The result is chunk embeddings that semantically carry the weight of their surrounding context, without requiring you to store or query massive full-document vectors at retrieval time.
By the end of this lesson, you'll be able to implement late chunking from scratch, understand exactly why it outperforms naive chunking on context-dependent queries, and integrate it into a production RAG pipeline. You'll also understand when it's worth the extra compute and when it isn't.
What you'll learn:
You should be comfortable with:
transformers and sentence-transformers)Let's be specific about what goes wrong with standard chunking. Consider a legal contract that begins with a definitions section:
"The Company" means Meridian Financial Holdings, Inc., a Delaware corporation...
Fifty pages later, buried in a liability clause:
"The Company shall not be liable for indirect damages arising from..."
If you chunk this document into 512-token windows, that liability clause chunk has no reference to what "The Company" means. Its embedding vector encodes the words but not the semantic entity — when a user asks "Is Meridian Financial Holdings liable for data breaches?", the retriever may score this chunk poorly because the entity name doesn't appear in it, even though the clause directly answers the question.
Standard approaches to fix this include:
Late chunking takes a different approach that's both cheaper and more semantically principled.
To understand late chunking, you need to think about what a transformer embedding model actually produces internally.
When you pass a 4096-token document through a model like jina-embeddings-v2-base-en (which supports 8192-token contexts), the model produces a matrix of shape [4096, hidden_dim] — one embedding vector for every input token. The standard "document embedding" you get from calling model.encode("some text") is just the mean-pool of those token vectors (or the [CLS] token, depending on architecture).
Here's the key insight: those per-token vectors already encode context from the entire document, because self-attention lets every token attend to every other token during forward pass. The token embedding for "liable" in your liability clause has already been influenced by the definitions 50 pages earlier.
Late chunking exploits this by:
The resulting chunk embedding represents "what this chunk means given the full document context" rather than "what this chunk means in isolation."
Here's a simple diagram of the flow:
Standard Chunking:
Document → Split into Chunks → Embed each Chunk independently → Vector Store
Late Chunking:
Document → Embed Full Document (get token matrix) → Apply Chunk Boundaries → Pool token vectors per chunk → Vector Store
The retrieval and generation steps are identical after this point. The only difference is in how the chunk vectors are produced.
We'll work with Jina AI's jina-embeddings-v2-base-en model because it was specifically designed to support late chunking and has an 8192-token context window. The implementation pattern works with any long-context model that exposes token-level embeddings.
pip install transformers torch numpy faiss-cpu sentence-transformers tiktoken
import torch
import numpy as np
from transformers import AutoTokenizer, AutoModel
import faiss
from typing import List, Tuple, Dict
# Load the model and tokenizer
MODEL_NAME = "jinaai/jina-embeddings-v2-base-en"
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, trust_remote_code=True)
model = AutoModel.from_pretrained(MODEL_NAME, trust_remote_code=True)
model.eval()
# Move to GPU if available
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = model.to(device)
print(f"Model loaded on {device}")
print(f"Max position embeddings: {model.config.max_position_embeddings}")
The first piece we need is a function that returns the full token embedding matrix rather than a pooled single vector. Most embedding libraries hide this from you — we need to go one level deeper.
def get_token_embeddings(
text: str,
max_length: int = 8192
) -> Tuple[np.ndarray, List[int]]:
"""
Returns token-level embeddings for an entire document.
Args:
text: Full document text
max_length: Maximum token length (model dependent)
Returns:
token_embeddings: np.ndarray of shape [num_tokens, hidden_dim]
token_ids: List of token ids (useful for debugging alignment)
"""
# Tokenize without truncation first so we can warn if needed
tokens = tokenizer(
text,
return_tensors="pt",
max_length=max_length,
truncation=True,
padding=False
)
num_tokens = tokens["input_ids"].shape[1]
if num_tokens == max_length:
print(f"WARNING: Document was truncated to {max_length} tokens. "
f"Consider splitting very long documents into sections first.")
tokens = {k: v.to(device) for k, v in tokens.items()}
with torch.no_grad():
outputs = model(**tokens, output_hidden_states=False)
# outputs.last_hidden_state shape: [1, num_tokens, hidden_dim]
token_embeddings = outputs.last_hidden_state[0] # Remove batch dim
return (
token_embeddings.cpu().numpy(),
tokens["input_ids"][0].cpu().tolist()
)
Let's verify this is working correctly before going further:
# Quick sanity check
test_text = "The Company means Meridian Financial Holdings Inc. The Company shall not be liable."
embeddings, token_ids = get_token_embeddings(test_text)
tokens_decoded = tokenizer.convert_ids_to_tokens(token_ids)
print(f"Token embedding matrix shape: {embeddings.shape}")
print(f"Number of tokens: {len(token_ids)}")
print(f"Tokens: {tokens_decoded}")
You should see something like:
Token embedding matrix shape: (22, 768)
Number of tokens: 22
Tokens: ['[CLS]', 'The', 'Company', 'means', 'Meridian', ...]
Important: Notice the
[CLS]and[SEP]special tokens. When you apply chunk boundaries later, you need to handle these carefully — you don't want special tokens inflating or distorting your chunk embeddings. We'll address this in the next section.
Now we need a function that maps from character-level or sentence-level chunk boundaries to token-level boundaries. This is the most finicky part of the implementation because tokenization isn't character-length-preserving.
def get_chunk_token_spans(
text: str,
chunks: List[str],
token_ids: List[int]
) -> List[Tuple[int, int]]:
"""
Maps text chunks to their corresponding token index spans.
This works by re-tokenizing each chunk independently and finding
where those tokens appear in the full document token sequence.
Uses a sliding window approach to handle subword tokens correctly.
Args:
text: Full document text
chunks: List of text chunks (must be contiguous substrings of text)
token_ids: Token ids from the full document tokenization
Returns:
List of (start_idx, end_idx) token spans, exclusive of special tokens
"""
spans = []
search_start = 1 # Skip [CLS] token at position 0
for chunk in chunks:
# Tokenize just this chunk (without special tokens)
chunk_tokens = tokenizer(
chunk,
add_special_tokens=False,
return_tensors="pt"
)["input_ids"][0].tolist()
chunk_len = len(chunk_tokens)
# Slide through the full document token sequence to find this chunk
found = False
for i in range(search_start, len(token_ids) - chunk_len + 1):
if token_ids[i:i + chunk_len] == chunk_tokens:
spans.append((i, i + chunk_len))
search_start = i + chunk_len # Ensure contiguous, non-overlapping
found = True
break
if not found:
# Fallback: if exact match fails (can happen with whitespace differences),
# use a length-based approximation
print(f"WARNING: Could not find exact token span for chunk: '{chunk[:50]}...'")
end = min(search_start + chunk_len, len(token_ids) - 1)
spans.append((search_start, end))
search_start = end
return spans
Why not just split by character count? Tokenizers don't preserve character-to-token ratios uniformly. "Meridian" might tokenize to 3 tokens while "and" tokenizes to 1. If you naively say "the first 512 characters map to tokens 0-100," you'll misalign your pooling windows and corrupt your embeddings.
With token spans in hand, the actual pooling step is straightforward:
def pool_token_embeddings_to_chunks(
token_embeddings: np.ndarray,
chunk_spans: List[Tuple[int, int]]
) -> np.ndarray:
"""
Pools token embeddings within each chunk span using mean pooling.
Args:
token_embeddings: Full token embedding matrix [num_tokens, hidden_dim]
chunk_spans: List of (start, end) token index tuples
Returns:
chunk_embeddings: np.ndarray of shape [num_chunks, hidden_dim]
"""
chunk_embeddings = []
for start, end in chunk_spans:
if start >= end:
print(f"WARNING: Empty or invalid span ({start}, {end}), skipping.")
continue
# Extract the token embeddings for this span
span_embeddings = token_embeddings[start:end] # [span_len, hidden_dim]
# Mean pool across the token dimension
chunk_vector = span_embeddings.mean(axis=0) # [hidden_dim]
# L2 normalize for cosine similarity retrieval
norm = np.linalg.norm(chunk_vector)
if norm > 0:
chunk_vector = chunk_vector / norm
chunk_embeddings.append(chunk_vector)
return np.array(chunk_embeddings)
On pooling strategies: Mean pooling is the standard for good reason — it gives every token in the chunk equal weight in the final representation. You could experiment with max pooling (captures salient features more aggressively) or weighted pooling (upweighting tokens near the center of the chunk), but mean pooling is the baseline that works reliably.
Let's put this all together into a cohesive class that you can drop into an existing RAG pipeline:
class LateChunkingIndexer:
"""
Builds a FAISS index using late chunking — token-level embeddings
from full-document forward passes, pooled at chunk boundaries.
"""
def __init__(self, model_name: str = "jinaai/jina-embeddings-v2-base-en",
chunk_size_chars: int = 1000, chunk_overlap_chars: int = 100):
self.tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
self.model = AutoModel.from_pretrained(model_name, trust_remote_code=True)
self.model.eval()
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
self.model = self.model.to(self.device)
self.chunk_size = chunk_size_chars
self.chunk_overlap = chunk_overlap_chars
self.index = None
self.chunk_metadata: List[Dict] = []
def _split_text(self, text: str) -> List[str]:
"""Simple character-based chunking with overlap."""
chunks = []
start = 0
text_len = len(text)
while start < text_len:
end = start + self.chunk_size
# Try to break at a sentence boundary
if end < text_len:
# Look for a period or newline near the end of this chunk
boundary = text.rfind('. ', start, end)
if boundary != -1 and boundary > start + self.chunk_size // 2:
end = boundary + 1 # Include the period
chunk = text[start:end].strip()
if chunk:
chunks.append(chunk)
start = end - self.chunk_overlap
return chunks
def _get_token_embeddings(self, text: str) -> Tuple[np.ndarray, List[int]]:
"""Full document forward pass returning token embedding matrix."""
tokens = self.tokenizer(
text,
return_tensors="pt",
max_length=8192,
truncation=True,
padding=False
)
tokens = {k: v.to(self.device) for k, v in tokens.items()}
with torch.no_grad():
outputs = self.model(**tokens)
token_embeddings = outputs.last_hidden_state[0]
return (
token_embeddings.cpu().numpy(),
tokens["input_ids"][0].cpu().tolist()
)
def _get_chunk_spans(self, chunks: List[str],
token_ids: List[int]) -> List[Tuple[int, int]]:
spans = []
search_start = 1 # skip [CLS]
for chunk in chunks:
chunk_token_ids = self.tokenizer(
chunk,
add_special_tokens=False,
return_tensors="pt"
)["input_ids"][0].tolist()
chunk_len = len(chunk_token_ids)
found = False
for i in range(search_start, len(token_ids) - chunk_len + 1):
if token_ids[i:i + chunk_len] == chunk_token_ids:
spans.append((i, i + chunk_len))
search_start = i + chunk_len
found = True
break
if not found:
end = min(search_start + chunk_len, len(token_ids) - 1)
spans.append((search_start, end))
search_start = end
return spans
def _pool_chunks(self, token_embeddings: np.ndarray,
spans: List[Tuple[int, int]]) -> np.ndarray:
pooled = []
for start, end in spans:
vec = token_embeddings[start:end].mean(axis=0)
vec = vec / (np.linalg.norm(vec) + 1e-8)
pooled.append(vec)
return np.array(pooled)
def index_document(self, text: str, doc_id: str, metadata: Dict = None):
"""Process a single document and add its chunks to the index."""
chunks = self._split_text(text)
token_embeddings, token_ids = self._get_token_embeddings(text)
spans = self._get_chunk_spans(chunks, token_ids)
chunk_vectors = self._pool_chunks(token_embeddings, spans)
if self.index is None:
hidden_dim = chunk_vectors.shape[1]
self.index = faiss.IndexFlatIP(hidden_dim) # Inner product = cosine on normalized vecs
self.index.add(chunk_vectors.astype(np.float32))
for i, chunk in enumerate(chunks):
self.chunk_metadata.append({
"doc_id": doc_id,
"chunk_idx": i,
"text": chunk,
"span": spans[i],
**(metadata or {})
})
print(f"Indexed {len(chunks)} chunks from '{doc_id}'")
def search(self, query: str, top_k: int = 5) -> List[Dict]:
"""Embed query and retrieve top-k chunks."""
# Query is short — standard embedding is fine here
query_tokens = self.tokenizer(
query, return_tensors="pt", max_length=512, truncation=True
)
query_tokens = {k: v.to(self.device) for k, v in query_tokens.items()}
with torch.no_grad():
query_output = self.model(**query_tokens)
# Mean pool the query too
query_vec = query_output.last_hidden_state[0][1:-1].mean(axis=0)
query_vec = query_vec / (query_vec.norm() + 1e-8)
query_vec = query_vec.cpu().numpy().astype(np.float32).reshape(1, -1)
distances, indices = self.index.search(query_vec, top_k)
results = []
for score, idx in zip(distances[0], indices[0]):
if idx >= 0:
result = self.chunk_metadata[idx].copy()
result["score"] = float(score)
results.append(result)
return results
Let's create a realistic test that demonstrates the advantage of late chunking. We'll simulate a technical specification document where critical context is established early and referenced repeatedly throughout.
# Simulate a realistic technical specification document
sample_document = """
SYSTEM ARCHITECTURE SPECIFICATION v2.4
Meridian Data Platform — Internal Reference
1. DEFINITIONS
The Primary Cluster refers to the three-node PostgreSQL ensemble hosted in us-east-1,
designated as the authoritative source of record for all transactional data.
The Replica Set refers to the five read-only PostgreSQL instances distributed across
us-west-2, eu-west-1, and ap-southeast-1.
The Cache Layer refers to the Redis cluster maintained at 99.9% availability SLA.
The Orchestrator refers to the Apache Airflow deployment managing all data pipeline DAGs.
2. PERFORMANCE REQUIREMENTS
All write operations MUST target the Primary Cluster exclusively.
Read operations may be distributed across the Replica Set at the application layer's
discretion, subject to a maximum replication lag tolerance of 500 milliseconds.
3. FAILURE HANDLING
If the Primary Cluster becomes unavailable, the Orchestrator must immediately:
- Halt all write-dependent DAG tasks
- Route all incoming writes to the disaster recovery endpoint at dr.meridian-internal.net
- Alert on-call engineering via PagerDuty integration configured in the Orchestrator
The Cache Layer must be flushed entirely if the replication lag on any node in the
Replica Set exceeds 30 seconds, to prevent stale cache hits from corrupting user queries.
4. MAINTENANCE WINDOWS
The Primary Cluster maintenance window is Sundays 02:00-04:00 UTC.
During this window, the Replica Set assumes temporary read-write capability via
a leader election process managed by the Orchestrator.
The Cache Layer is exempt from maintenance windows but should be restarted
in a rolling fashion across availability zones after any Primary Cluster maintenance.
"""
# Also create a short version of just the failure handling section
# to simulate what standard chunking might produce
failure_handling_chunk = """
If the Primary Cluster becomes unavailable, the Orchestrator must immediately:
- Halt all write-dependent DAG tasks
- Route all incoming writes to the disaster recovery endpoint at dr.meridian-internal.net
- Alert on-call engineering via PagerDuty integration configured in the Orchestrator
The Cache Layer must be flushed entirely if the replication lag on any node in the
Replica Set exceeds 30 seconds, to prevent stale cache hits from corrupting user queries.
"""
# Initialize the indexer
indexer = LateChunkingIndexer(chunk_size_chars=600, chunk_overlap_chars=50)
# Index the full document using late chunking
indexer.index_document(sample_document, doc_id="meridian-arch-spec-v2.4")
# Test queries
queries = [
"What happens to the PostgreSQL cluster in us-east-1 if it goes down?",
"When should the Redis cache be flushed?",
"What manages the DAG tasks during failure?",
]
for query in queries:
print(f"\nQuery: {query}")
print("-" * 60)
results = indexer.search(query, top_k=2)
for r in results:
print(f"Score: {r['score']:.4f}")
print(f"Chunk: {r['text'][:200]}...")
print()
Running this, you'll find that even though the query mentions "PostgreSQL cluster in us-east-1" — terminology that only appears in the definitions section — the failure handling chunk retrieves with high confidence. The chunk embedding has absorbed the knowledge that "Primary Cluster" refers to that specific system.
To make the comparison concrete, let's implement a standard chunking baseline and score them head-to-head:
class StandardChunkingIndexer:
"""Baseline: embed each chunk independently, no document-level context."""
def __init__(self, model_name: str = "jinaai/jina-embeddings-v2-base-en",
chunk_size_chars: int = 600, chunk_overlap_chars: int = 50):
self.tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
self.model = AutoModel.from_pretrained(model_name, trust_remote_code=True)
self.model.eval()
self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
self.model = self.model.to(self.device)
self.chunk_size = chunk_size_chars
self.chunk_overlap = chunk_overlap_chars
self.index = None
self.chunk_metadata: List[Dict] = []
def _split_text(self, text: str) -> List[str]:
chunks, start = [], 0
while start < len(text):
end = start + self.chunk_size
if end < len(text):
boundary = text.rfind('. ', start, end)
if boundary != -1 and boundary > start + self.chunk_size // 2:
end = boundary + 1
chunk = text[start:end].strip()
if chunk:
chunks.append(chunk)
start = end - self.chunk_overlap
return chunks
def _embed_chunk(self, text: str) -> np.ndarray:
tokens = self.tokenizer(
text, return_tensors="pt", max_length=512,
truncation=True, padding=False
)
tokens = {k: v.to(self.device) for k, v in tokens.items()}
with torch.no_grad():
outputs = self.model(**tokens)
# Standard mean pool, no document context
vec = outputs.last_hidden_state[0][1:-1].mean(axis=0)
vec = vec / (vec.norm() + 1e-8)
return vec.cpu().numpy().astype(np.float32)
def index_document(self, text: str, doc_id: str, metadata: Dict = None):
chunks = self._split_text(text)
if self.index is None:
sample_vec = self._embed_chunk(chunks[0])
self.index = faiss.IndexFlatIP(sample_vec.shape[0])
for i, chunk in enumerate(chunks):
vec = self._embed_chunk(chunk)
self.index.add(vec.reshape(1, -1))
self.chunk_metadata.append({
"doc_id": doc_id, "chunk_idx": i,
"text": chunk, **(metadata or {})
})
print(f"Indexed {len(chunks)} chunks from '{doc_id}'")
def search(self, query: str, top_k: int = 5) -> List[Dict]:
query_vec = self._embed_chunk(query).reshape(1, -1)
distances, indices = self.index.search(query_vec, top_k)
results = []
for score, idx in zip(distances[0], indices[0]):
if idx >= 0:
result = self.chunk_metadata[idx].copy()
result["score"] = float(score)
results.append(result)
return results
# Build both indexes on the same document
standard_indexer = StandardChunkingIndexer(chunk_size_chars=600, chunk_overlap_chars=50)
standard_indexer.index_document(sample_document, doc_id="meridian-arch-spec-v2.4")
# Compare retrieval on a context-dependent query
context_query = "What should happen to the cache if the us-east-1 PostgreSQL nodes have high replication lag?"
print("=== LATE CHUNKING RESULTS ===")
late_results = indexer.search(context_query, top_k=3)
for r in late_results:
print(f"Score: {r['score']:.4f} | {r['text'][:150]}...\n")
print("\n=== STANDARD CHUNKING RESULTS ===")
standard_results = standard_indexer.search(context_query, top_k=3)
for r in standard_results:
print(f"Score: {r['score']:.4f} | {r['text'][:150]}...\n")
What you'll typically observe: late chunking retrieves the cache-flushing instruction with a higher confidence score on queries that bridge the definitions section and the failure handling section. Standard chunking often retrieves the definitions chunk instead — it matches on "us-east-1" and "PostgreSQL" but misses the semantic connection to the consequence.
Late chunking isn't free. Here's an honest accounting of the trade-offs:
Compute cost at index time: You're running one full-document forward pass instead of N independent chunk forward passes. For a 4000-token document chunked into 10 pieces, you're doing roughly the same FLOPs — but with long-context models, the attention complexity is O(n²) in sequence length, so a single 4000-token pass is more expensive than ten 400-token passes. In practice, for documents under ~4000 tokens, expect 2-4x longer indexing time.
Memory: Storing the full token embedding matrix during indexing requires more memory. A 8192-token document with 768-dim hidden states needs about 48MB in float32. This matters if you're indexing many documents concurrently.
Query time: Identical to standard chunking. Queries are short and don't benefit from late chunking — embed them normally.
When late chunking wins significantly:
When standard chunking is fine:
When to use alternatives instead:
Build a late chunking pipeline for a multi-document corpus and evaluate retrieval quality. Use the following structure:
Step 1: Download 3-5 Wikipedia articles on related topics (e.g., "PostgreSQL," "Database replication," "Apache Airflow"). You can fetch them programmatically:
import urllib.request
import json
def fetch_wikipedia(title: str) -> str:
url = (f"https://en.wikipedia.org/w/api.php?action=query&titles={title}"
f"&prop=extracts&explaintext=true&format=json")
with urllib.request.urlopen(url) as response:
data = json.loads(response.read())
pages = data["query"]["pages"]
page = next(iter(pages.values()))
return page.get("extract", "")
articles = {
"PostgreSQL": fetch_wikipedia("PostgreSQL"),
"Database_replication": fetch_wikipedia("Database_replication"),
"Apache_Airflow": fetch_wikipedia("Apache_Airflow"),
}
Step 2: Index all three articles using both LateChunkingIndexer and StandardChunkingIndexer.
Step 3: Design 5 cross-article queries that require understanding terminology defined in one article to correctly answer a question answered in another. Example:
Step 4: For each query, record the top-3 retrieved chunks from both methods. Score them manually (0/1/2) for relevance. Compute mean reciprocal rank (MRR) across all queries for both methods.
Step 5: Write a one-paragraph analysis: under what conditions did late chunking outperform standard chunking, and when did it not make a difference?
Mistake 1: Applying late chunking to overlapping chunks
If your chunks overlap (a common technique to reduce context loss at boundaries), the span detection logic will fail because the same tokens appear in two chunks. Either use non-overlapping chunks for late chunking, or implement a weighted pooling strategy that handles overlap. The easiest fix: disable overlap when using late chunking — the context benefit from the attention mechanism makes boundary overlap less necessary anyway.
Mistake 2: Forgetting to handle documents that exceed the model's context window
If your document is 20,000 tokens and your model supports 8192, the tokenizer will silently truncate it. The chunks from the second half of the document will be indexed against token embeddings that don't exist. Always check num_tokens == max_length after tokenization and implement a document sectioning strategy for very long inputs — split by chapter or major heading before applying late chunking within each section.
Mistake 3: Using a short-context model
The entire benefit of late chunking depends on the model's context window being large enough to encompass the document (or at least the relevant context). Running late chunking on all-MiniLM-L6-v2 (512-token limit) on 2000-token documents gives you exactly zero benefit — you're just doing more complex work for the same result as standard chunking.
Mistake 4: Not normalizing vectors before FAISS IndexFlatIP
If you use inner product similarity (IndexFlatIP) without normalizing your vectors to unit norm, you're measuring raw dot product, which conflates magnitude with direction. Always L2-normalize chunk vectors before indexing and query vectors before searching when using IP similarity. Use faiss.normalize_L2(vectors) as an alternative to manual normalization.
# The safe way to normalize in-place for FAISS
vectors = chunk_embeddings.astype(np.float32)
faiss.normalize_L2(vectors) # Modifies in-place
self.index.add(vectors)
Mistake 5: Ignoring special tokens in span detection
The [CLS] token at position 0 and [SEP] at the final position don't correspond to any text. If your span detection accidentally includes them in a chunk's pooling window, you're adding noise to the embedding. Always start your search from index 1 and stop before the final special token.
Debugging span alignment issues:
def debug_span(token_ids, span, tokenizer):
"""Print the tokens within a span to verify alignment."""
start, end = span
span_tokens = tokenizer.convert_ids_to_tokens(token_ids[start:end])
print(f"Span [{start}:{end}] contains {end-start} tokens:")
print(" ".join(span_tokens))
Run this on your first few chunks whenever you're testing on a new document type. If you see [SEP] tokens or tokens from adjacent chunks, your alignment is off.
Late chunking is an elegant solution to a real problem in RAG systems. By letting the attention mechanism do its work across the full document before you split, you get chunk embeddings that know where they came from — which definitions surround them, which entities they refer back to, which conditions apply to the claims they contain.
The key mechanics to remember:
For your next steps, consider these paths:
Combine late chunking with reranking: Late chunking improves first-stage retrieval, but a cross-encoder reranker applied to the top-20 results will further boost precision. The two techniques are complementary and don't interfere with each other.
Benchmark against contextual chunking (Anthropic's approach): Contextual chunking uses an LLM to prepend context to each chunk before embedding. It's more expensive but potentially higher quality. Understanding both lets you make principled trade-offs.
Explore hierarchical late chunking: For very long documents, first apply late chunking within sections, then embed section summaries separately. Use a routing step to find the right section, then retrieve within it.
Implement streaming late chunking for real-time indexing: The current approach requires processing full documents in one pass. For real-time document ingestion, investigate how to batch documents efficiently without blowing memory budgets.
The reference implementation we built here is production-capable for corpora where documents fit within 8192 tokens. For anything beyond that, you'll need to add the document-sectioning layer — but the core pattern remains the same.
Learning Path: RAG & AI Agents