Chunking Strategies

Before writing a single line of chunking code, it is worth understanding the fundamental tension you are navigating.

The retrieval granularity problem:

A RAG system embeds each chunk independently and retrieves the top-k chunks most similar to the query embedding. This means the chunk is the atomic unit of retrieval. If a chunk is too large, one query can only surface it when a majority of its sentences are relevant — meaning many relevant passages are buried inside chunks that score poorly. If a chunk is too small, retrieved text may lack the surrounding context the LLM needs to produce a coherent answer.

This tension does not resolve neatly. It shifts depending on the query pattern:

Most real query workloads are mixed, which is why advanced strategies (parent-child retrieval, covered in Lesson 6) often outperform any single fixed strategy.

Two concrete failure modes to recognise:

DOCUMENT EXCERPT:
  "The retry limit defaults to 3. When this limit is exceeded,
   the system raises a MaxRetriesExceeded exception and logs
   the event to the audit trail."

QUERY: "What exception is raised on too many retries?"

--- FAILURE: chunks split at sentence boundary ---
Chunk A: "The retry limit defaults to 3."
Chunk B: "When this limit is exceeded, the system raises a
          MaxRetriesExceeded exception and logs the event
          to the audit trail."

  Chunk A scores high (contains "retry" and "limit") but
  does NOT contain the exception name.
  Chunk B scores lower (no "retry limit" context) — it may
  not even appear in top-k retrieval.

--- SUCCESS: boundary respects sentence grouping ---
Chunk: "The retry limit defaults to 3. When this limit is
        exceeded, the system raises a MaxRetriesExceeded
        exception and logs the event to the audit trail."
  A single retrieval surfaces both the condition and the answer.

Fixed-size chunking splits text every N tokens or characters, with an optional overlap. It is the simplest strategy and a useful baseline, but it is blind to document structure.

Implementing fixed-size chunking with LangChain:

from langchain.text_splitter import CharacterTextSplitter, TokenTextSplitter
from pathlib import Path

raw_text = Path("docs/manual.md").read_text()

# Character-based: split every 1000 chars, 200-char overlap
char_splitter = CharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separator="\n",           # prefer splitting on newlines
    length_function=len,
)
char_chunks = char_splitter.split_text(raw_text)

# Token-based: split every 512 tokens (matches most embedding models)
token_splitter = TokenTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    encoding_name="cl100k_base",   # tiktoken encoding for OpenAI/many models
)
token_chunks = token_splitter.split_text(raw_text)

print(f"Character-based: {len(char_chunks)} chunks")
print(f"Token-based:     {len(token_chunks)} chunks")
print(f"\nFirst token chunk ({len(token_chunks[0].split())} words):")
print(token_chunks[0][:300])

Understanding overlap:

Overlap re-includes the last N tokens of the previous chunk at the start of the next one. This reduces the chance of slicing an answer across a boundary — but it increases storage and embedding costs proportionally.

# Measuring overlap cost
base_chunks = len(token_chunks)
overlap_tokens = 64 * (base_chunks - 1)           # overlap added per boundary
total_tokens_stored = 512 * base_chunks            # approximate
print(f"Overlap adds ~{overlap_tokens} extra tokens ({overlap_tokens / total_tokens_stored:.0%} overhead)")

When fixed-size is appropriate:

LangChain's RecursiveCharacterTextSplitter is the most widely used general-purpose splitter. It attempts to split on a priority-ordered list of separators, falling back to harder splits only when softer ones would exceed the chunk size.

Default separator hierarchy:

["\n\n", "\n", " ", ""]

This means: try to split on paragraph breaks first; if a paragraph is too long, split on line breaks; if a line is too long, split on spaces; as a last resort, split mid-word.

Implementation:

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,           # in tokens when using from_tiktoken_encoder
    chunk_overlap=64,
    length_function=len,      # swap for token counter below
)

# Preferred: use tiktoken for accurate token counting
from langchain.text_splitter import RecursiveCharacterTextSplitter
import tiktoken

enc = tiktoken.get_encoding("cl100k_base")

def token_len(text: str) -> int:
    return len(enc.encode(text))

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    length_function=token_len,
    separators=["\n\n", "\n", ". ", " ", ""],
)

chunks = splitter.split_text(raw_text)

Customising separators for your document type:

# Markdown — respect heading hierarchy before paragraphs
md_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    length_function=token_len,
    separators=["## ", "### ", "\n\n", "\n", " "],
)

# Python source code — respect function/class boundaries
code_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=32,
    length_function=token_len,
    separators=["\nclass ", "\ndef ", "\n\n", "\n", " "],
)

Comparing fixed vs. recursive on the same document:

fixed = TokenTextSplitter(chunk_size=512, chunk_overlap=64)
recursive = RecursiveCharacterTextSplitter(
    chunk_size=512, chunk_overlap=64, length_function=token_len
)

fixed_chunks = fixed.split_text(raw_text)
recursive_chunks = recursive.split_text(raw_text)

# Measure how often chunks end mid-sentence
def ends_mid_sentence(chunk: str) -> bool:
    return not chunk.rstrip().endswith((".", "!", "?", "```", ":"))

fixed_broken = sum(1 for c in fixed_chunks if ends_mid_sentence(c))
recursive_broken = sum(1 for c in recursive_chunks if ends_mid_sentence(c))

print(f"Fixed — mid-sentence breaks: {fixed_broken}/{len(fixed_chunks)} "
      f"({fixed_broken/len(fixed_chunks):.0%})")
print(f"Recursive — mid-sentence breaks: {recursive_broken}/{len(recursive_chunks)} "
      f"({recursive_broken/len(recursive_chunks):.0%})")

On a typical technical document, recursive splitting reduces mid-sentence breaks by 60–80% versus fixed splitting with the same token budget.

Recursive splitting respects surface syntax (whitespace, punctuation) but is blind to meaning. Two sentences on completely different topics may end up in the same chunk simply because they appear adjacent. Semantic chunking solves this by measuring the embedding distance between consecutive sentences and inserting a boundary whenever the semantic similarity drops sharply — signalling a topic shift.

The algorithm:

1. Split the document into individual sentences (using a sentence tokeniser).
2. Embed each sentence (or a sliding window of sentences for stability).
3. Compute cosine similarity between each adjacent pair.
4. Insert a chunk boundary wherever similarity drops below a threshold — a local minimum in the similarity curve.
5. Group sentences between boundaries into chunks; if a group still exceeds the max token budget, apply a secondary recursive split.

Implementation using LangChain's SemanticChunker:

from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.embeddings import HuggingFaceEmbeddings

embed_model = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-en-v1.5",
    model_kwargs={"device": "cpu"},
    encode_kwargs={"normalize_embeddings": True},
)

# percentile: insert boundary when similarity is in the lowest N-th percentile
semantic_splitter = SemanticChunker(
    embeddings=embed_model,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=80,   # 80th percentile drop = boundary
)

semantic_chunks = semantic_splitter.split_text(raw_text)

print(f"Semantic chunks: {len(semantic_chunks)}")
for i, chunk in enumerate(semantic_chunks[:3]):
    toks = token_len(chunk)
    print(f"  Chunk {i+1}: {toks} tokens — {chunk[:80].strip()!r}...")

Threshold strategies:

When semantic chunking wins and when it struggles:

WINS:
  Long mixed-topic documents (wikis, incident reports, research PDFs)
  Documents where paragraph breaks do not align with topic shifts

STRUGGLES:
  Very short documents (not enough signal for similarity curves)
  Highly technical text where adjacent sentences look dissimilar
  even when they're part of the same argument (low baseline similarity)
  Documents with dense cross-references (similarity stays high throughout)

Many real document corpora are not plain prose — they are Markdown documentation, HTML pages, or source code. These formats encode structure explicitly. Ignoring that structure and falling back to character/token splitting throws away information that makes it trivial to find natural chunk boundaries.

Markdown-aware chunking:

LangChain provides MarkdownHeaderTextSplitter, which splits first at heading boundaries and then applies recursive splitting within each section. Heading levels become chunk metadata automatically.

from langchain.text_splitter import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

headers_to_split_on = [
    ("#", "h1"),
    ("##", "h2"),
    ("###", "h3"),
]

md_header_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on,
    strip_headers=False,    # keep header text in the chunk body
)

header_splits = md_header_splitter.split_text(raw_text)

# Each split now carries heading metadata:
# split.metadata == {"h1": "Installation", "h2": "Quick Start", "h3": "macOS"}

# Apply token-level splitting within each section
secondary_splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    length_function=token_len,
)

final_chunks = secondary_splitter.split_documents(header_splits)

print(f"Sections after header split: {len(header_splits)}")
print(f"Final chunks after token split: {len(final_chunks)}")
print(f"\nSample metadata: {final_chunks[2].metadata}")

HTML-aware chunking:

from langchain_community.document_transformers import Html2TextTransformer
from langchain_community.document_loaders import AsyncHtmlLoader

# Load HTML and convert to clean Markdown-like text
loader = AsyncHtmlLoader(["https://example.com/docs/api"])
html_docs = loader.load()

html2text = Html2TextTransformer()
docs_transformed = html2text.transform_documents(html_docs)

# Now apply recursive splitting to the cleaned text
chunks = secondary_splitter.split_documents(docs_transformed)

Code-aware chunking with tree-sitter:

For source code, respecting function and class boundaries matters far more than token counts. The unstructured library provides language-aware splitting:

from langchain.text_splitter import Language, RecursiveCharacterTextSplitter

python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=512,
    chunk_overlap=32,
    length_function=token_len,
)

# Language.PYTHON uses these separators in order:
# ["\nclass ", "\ndef ", "\n\tdef ", "\n\n", "\n", " ", ""]

code_text = Path("src/retriever.py").read_text()
code_chunks = python_splitter.split_text(code_text)

# Supported languages: PYTHON, JS, TS, RUST, GO, JAVA, C, CPP, MARKDOWN, HTML, SOL, ...

Comparison of structure-aware vs. naive on a Markdown document:

A chunk without metadata is an orphan — the LLM can read it but cannot tell the user where it came from or filter it by provenance. Metadata enrichment attaches contextual information to every chunk at index time, enabling three things: better post-retrieval citations, metadata-filtered similarity search, and hybrid ranking signals.

Core metadata fields to attach to every chunk:

from dataclasses import dataclass, field
from typing import Any

@dataclass
class ChunkMetadata:
    source_path: str          # absolute or relative path to originating file
    source_url: str           # URL if loaded from the web
    doc_title: str            # document-level title
    section_heading: str      # most specific heading above this chunk
    chunk_index: int          # position within the document (0-based)
    total_chunks: int         # total chunks in this document
    char_start: int           # character offset in original document
    char_end: int
    ingested_at: str          # ISO-8601 timestamp
    doc_version: str          # e.g., git SHA or document modified date
    extra: dict[str, Any] = field(default_factory=dict)

Attaching metadata during chunking:

from langchain.schema import Document
from datetime import datetime, timezone
from pathlib import Path

def chunk_document_with_metadata(
    file_path: str,
    splitter,
    doc_title: str = "",
    doc_version: str = "",
) -> list[Document]:
    path = Path(file_path)
    raw_text = path.read_text()

    # Split into LangChain Documents
    raw_chunks = splitter.create_documents([raw_text])

    ingested_at = datetime.now(timezone.utc).isoformat()

    enriched = []
    for i, chunk in enumerate(raw_chunks):
        chunk.metadata.update({
            "source_path": str(path),
            "doc_title": doc_title or path.stem,
            "chunk_index": i,
            "total_chunks": len(raw_chunks),
            "ingested_at": ingested_at,
            "doc_version": doc_version,
        })
        enriched.append(chunk)
    return enriched

Contextual headers — prepending section context to every chunk:

One of the highest-leverage metadata enrichment techniques is prepending a brief context header to the chunk text itself before embedding. This means the embedding captures both the content and its provenance, dramatically improving retrieval for queries that reference section names.

def add_contextual_header(chunk: Document) -> Document:
    """Prepend doc title and section heading to chunk text before embedding."""
    meta = chunk.metadata
    header_parts = []
    if meta.get("doc_title"):
        header_parts.append(f"Document: {meta['doc_title']}")
    if meta.get("h1"):
        header_parts.append(f"Section: {meta['h1']}")
    if meta.get("h2"):
        header_parts.append(f"Subsection: {meta['h2']}")

    if header_parts:
        header = " | ".join(header_parts)
        enriched_text = f"[{header}]\n\n{chunk.page_content}"
        return Document(page_content=enriched_text, metadata=chunk.metadata)
    return chunk

# Apply before embedding
chunks_for_embedding = [add_contextual_header(c) for c in enriched_chunks]

Metadata-filtered retrieval in Chroma:

import chromadb
from chromadb.utils.embedding_functions import SentenceTransformerEmbeddingFunction

client = chromadb.PersistentClient(path="./chroma_db")
embed_fn = SentenceTransformerEmbeddingFunction(model_name="BAAI/bge-small-en-v1.5")

collection = client.get_or_create_collection(
    name="rag_chunks",
    embedding_function=embed_fn,
)

# Upsert chunks with full metadata
collection.upsert(
    ids=[f"chunk_{i}" for i in range(len(chunks_for_embedding))],
    documents=[c.page_content for c in chunks_for_embedding],
    metadatas=[c.metadata for c in chunks_for_embedding],
)

# Retrieval filtered by source document
results = collection.query(
    query_texts=["What is the retry limit?"],
    n_results=5,
    where={"source_path": {"$eq": "docs/manual.md"}},
)

Choosing a chunking strategy without measuring its effect is guesswork. This step walks through building a lightweight evaluation harness that compares strategies on the same corpus and query set.

Core retrieval metrics (defined from first principles):

• Precision@k — of the top-k chunks retrieved, what fraction are relevant? Measures whether retrieved results are signal or noise.
• Recall@k — of all relevant chunks in the corpus, what fraction appear in top-k? Measures whether relevant chunks are being found at all.
• MRR (Mean Reciprocal Rank) — the mean of 1/rank of the first relevant chunk across queries. Measures how quickly a relevant result appears.

from sentence_transformers import SentenceTransformer, util
import numpy as np
from typing import Callable

# A golden eval set: list of (query, list_of_relevant_source_passages)
EVAL_SET = [
    {
        "query": "What is the maximum retry count?",
        "relevant_passages": [
            "The retry limit defaults to 3. When this limit is exceeded",
        ],
    },
    {
        "query": "How does the authentication token expire?",
        "relevant_passages": [
            "Tokens expire after 24 hours. After expiry, the client must",
            "The token expiry policy is configurable via AUTH_TOKEN_TTL",
        ],
    },
    # ... add 20-50 queries for meaningful signal
]

def precision_at_k(retrieved: list[str], relevant: list[str], k: int) -> float:
    top_k = retrieved[:k]
    hits = sum(
        1 for r in top_k
        if any(rel.lower() in r.lower() for rel in relevant)
    )
    return hits / k

def recall_at_k(retrieved: list[str], relevant: list[str], k: int) -> float:
    top_k = retrieved[:k]
    found = sum(
        1 for rel in relevant
        if any(rel.lower() in r.lower() for r in top_k)
    )
    return found / len(relevant) if relevant else 0.0

def mrr(retrieved: list[str], relevant: list[str]) -> float:
    for rank, r in enumerate(retrieved, start=1):
        if any(rel.lower() in r.lower() for rel in relevant):
            return 1.0 / rank
    return 0.0

def evaluate_chunking_strategy(
    strategy_name: str,
    collection,
    eval_set: list[dict],
    k: int = 5,
) -> dict:
    p_scores, r_scores, mrr_scores = [], [], []

    for item in eval_set:
        results = collection.query(
            query_texts=[item["query"]],
            n_results=k,
        )
        retrieved_docs = results["documents"][0]

        p_scores.append(precision_at_k(retrieved_docs, item["relevant_passages"], k))
        r_scores.append(recall_at_k(retrieved_docs, item["relevant_passages"], k))
        mrr_scores.append(mrr(retrieved_docs, item["relevant_passages"]))

    return {
        "strategy": strategy_name,
        "precision@5": np.mean(p_scores),
        "recall@5": np.mean(r_scores),
        "mrr": np.mean(mrr_scores),
    }

Running a head-to-head comparison:

import chromadb
from chromadb.utils.embedding_functions import SentenceTransformerEmbeddingFunction

client = chromadb.EphemeralClient()
embed_fn = SentenceTransformerEmbeddingFunction(model_name="BAAI/bge-small-en-v1.5")

strategies = {
    "fixed_512": TokenTextSplitter(chunk_size=512, chunk_overlap=0),
    "fixed_512_overlap": TokenTextSplitter(chunk_size=512, chunk_overlap=64),
    "recursive_512": RecursiveCharacterTextSplitter(
        chunk_size=512, chunk_overlap=64, length_function=token_len
    ),
    "recursive_256": RecursiveCharacterTextSplitter(
        chunk_size=256, chunk_overlap=32, length_function=token_len
    ),
    "md_header_aware": md_header_splitter,  # from Step 5
}

results_table = []
for name, splitter in strategies.items():
    # Build a fresh collection for each strategy
    coll = client.get_or_create_collection(name=name, embedding_function=embed_fn)

    chunks = splitter.split_text(raw_text)
    coll.upsert(
        ids=[f"{name}_{i}" for i in range(len(chunks))],
        documents=chunks,
    )

    metrics = evaluate_chunking_strategy(name, coll, EVAL_SET, k=5)
    results_table.append(metrics)
    print(f"{name:30s}  P@5={metrics['precision@5']:.3f}  "
          f"R@5={metrics['recall@5']:.3f}  MRR={metrics['mrr']:.3f}")

What typical results show:

The precise numbers depend on your corpus and queries, but patterns that appear consistently:

The four strategies covered in this lesson are not mutually exclusive — a production pipeline often applies different strategies to different document types and then stores everything in the same vector collection (keyed by doc_type metadata).

Decision tree:

Is your corpus structured (Markdown, HTML, code with headers)?
├── YES → Use structure-aware chunking (Step 5) as the primary split,
│          then token-level secondary split within each section.
│          Add heading metadata to every chunk.
└── NO → Is it long-form prose with clear topic shifts?
         ├── YES → Try semantic chunking (Step 4).
         │          Tune threshold type and amount with your eval set.
         └── NO → Use RecursiveCharacterTextSplitter (Step 3).
                   Start at 512 tokens / 64 overlap and adjust
                   based on eval metrics.

Are your queries mostly fact-lookup (one-sentence answers)?
→ Skew smaller: 128–256 tokens

Are your queries mostly conceptual / multi-step?
→ Skew larger: 512–1024 tokens, or use parent-child retrieval (Lesson 6)

Does your LLM context window constrain you?
→ top_k × avg_chunk_tokens must fit comfortably; reduce chunk size or top_k

Quick-reference summary table:

Putting it all together — a multi-strategy ingestion pipeline:

from pathlib import Path

STRATEGY_BY_EXT = {
    ".md": "markdown_header",
    ".py": "code_python",
    ".html": "html",
    ".txt": "recursive",
    ".pdf": "recursive",   # after PDF text extraction
}

def get_splitter(ext: str):
    if ext == ".md":
        return md_header_splitter        # from Step 5, secondary token split applied after
    elif ext == ".py":
        return RecursiveCharacterTextSplitter.from_language(
            Language.PYTHON, chunk_size=512, chunk_overlap=32, length_function=token_len
        )
    elif ext == ".html":
        return secondary_splitter        # after Html2TextTransformer
    else:
        return RecursiveCharacterTextSplitter(
            chunk_size=512, chunk_overlap=64, length_function=token_len
        )

def ingest_corpus(corpus_dir: str, collection) -> int:
    corpus = Path(corpus_dir)
    total = 0
    for doc_path in corpus.rglob("*"):
        if not doc_path.is_file():
            continue
        ext = doc_path.suffix.lower()
        if ext not in STRATEGY_BY_EXT:
            continue

        splitter = get_splitter(ext)
        chunks = chunk_document_with_metadata(
            str(doc_path),
            splitter,
            doc_version="v1.0",
        )
        chunks = [add_contextual_header(c) for c in chunks]

        collection.upsert(
            ids=[f"{doc_path.stem}_{c.metadata['chunk_index']}" for c in chunks],
            documents=[c.page_content for c in chunks],
            metadatas=[c.metadata for c in chunks],
        )
        total += len(chunks)
        print(f"  {doc_path.name}: {len(chunks)} chunks ({ext})")

    return total

total_chunks = ingest_corpus("./corpus", collection)
print(f"\nTotal chunks indexed: {total_chunks}")