Chunking Strategies for RAG: From Basic to Enriched

from langchain.text_splitter import CharacterTextSplitter

def basic_chunk(text: str, chunk_size: int = 512, chunk_overlap: int = 0) -> list[str]:
    """
    Splits text into fixed-size chunks with no overlap.
    chunk_size: number of characters per chunk
    chunk_overlap: 0 = no overlap (pure basic chunking)
    """
    splitter = CharacterTextSplitter(
        separator="",          # split on any character boundary
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
    )
    return splitter.split_text(text)


# Example
with open("document.txt") as f:
    raw = f.read()

chunks = basic_chunk(raw, chunk_size=512)
for i, c in enumerate(chunks[:3]):
    print(f"--- Chunk {i} ({len(c)} chars) ---")
    print(c[:120], "...\n")

from langchain.text_splitter import RecursiveCharacterTextSplitter

def overlap_chunk(
    text: str,
    chunk_size: int = 512,
    chunk_overlap: int = 128,
) -> list[str]:
    """
    RecursiveCharacterTextSplitter tries to split on paragraphs, then
    sentences, then words, falling back to characters only if needed.
    chunk_overlap: how many chars to repeat from the previous chunk.
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    return splitter.split_text(text)


chunks = overlap_chunk(raw, chunk_size=512, chunk_overlap=128)
print(f"Total chunks: {len(chunks)}")
# Verify overlap between consecutive chunks
overlap_sample = set(chunks[0].split()).intersection(set(chunks[1].split()))
print(f"Shared words between chunk 0 and 1: {len(overlap_sample)}")

from langchain.text_splitter import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter

# ── Step 1: split on structural boundaries ──────────────────────────────────
headers_to_split_on = [
    ("#",  "H1"),
    ("##", "H2"),
    ("###","H3"),
]
md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on,
    strip_headers=False,          # keep heading text inside the chunk
)
md_chunks = md_splitter.split_text(raw_markdown)

# ── Step 2: further split sections that are still too large ─────────────────
char_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1024,
    chunk_overlap=128,
)
final_chunks = char_splitter.split_documents(md_chunks)

# ── Step 3: inspect metadata ─────────────────────────────────────────────────
for doc in final_chunks[:3]:
    print("METADATA:", doc.metadata)   # {"H1": "Introduction", "H2": "Background"}
    print("CONTENT: ", doc.page_content[:120])
    print()

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# Any embedding model works, swap for HuggingFaceEmbeddings, VoyageAI, etc.
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# ── Three breakpoint strategies ───────────────────────────────────────────────
# "percentile"     → split at distances above the p-th percentile
# "standard_deviation" → split when distance > mean + z*std
# "interquartile"  → split when distance > Q3 + 1.5*IQR

chunker = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=85,   # split at the top 15% of distance jumps
)

docs = chunker.create_documents([raw_text])

for i, doc in enumerate(docs[:4]):
    print(f"── Chunk {i} ({len(doc.page_content)} chars) ──")
    print(doc.page_content[:200])
    print()

from langchain.schema import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter

def enrich_deterministic(
    doc_text: str,
    metadata: dict,         # e.g. {"title": "Q2 Report", "section": "Revenue", "date": "2023-06"}
    chunk_size: int = 512,
    chunk_overlap: int = 64,
) -> list[Document]:
    """
    Prepends a breadcrumb derived from metadata to every chunk so that
    the embedding encodes document provenance alongside content.
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
    )
    raw_chunks = splitter.split_text(doc_text)

    # Build a human-readable breadcrumb from available metadata fields
    breadcrumb_parts = filter(None, [
        metadata.get("title"),
        metadata.get("section"),
        metadata.get("date"),
    ])
    breadcrumb = " > ".join(breadcrumb_parts)

    enriched = []
    for chunk in raw_chunks:
        enriched_text = f"[{breadcrumb}]\n\n{chunk}" if breadcrumb else chunk
        enriched.append(Document(page_content=enriched_text, metadata=metadata))

    return enriched


# Usage
docs = enrich_deterministic(
    doc_text=raw_text,
    metadata={"title": "ACME Corp 10-K", "section": "Revenue", "date": "2023-06"},
)
print(docs[0].page_content[:300])

<document> 
{{WHOLE_DOCUMENT}} 
</document> 

Here is the chunk we want to situate within the whole document:
<chunk> 
{{CHUNK_CONTENT}} 
</chunk> 

Please give a short succinct context to situate this chunk within the overall 
document for the purposes of improving search retrieval of the chunk. 
Answer only with the succinct context and nothing else.

import anthropic
from langchain.schema import Document
from langchain.text_splitter import RecursiveCharacterTextSplitter

client = anthropic.Anthropic()

CONTEXT_PROMPT = """<document>
{document}
</document>
Here is the chunk we want to situate within the whole document:
<chunk>
{chunk}
</chunk>
Please give a short succinct context to situate this chunk within the overall
document for the purposes of improving search retrieval of the chunk.
Answer only with the succinct context and nothing else."""


def generate_context(document: str, chunk: str) -> str:
    """
    Calls Claude to generate a 50-100 token situating context for the chunk.
    Uses prompt caching so the full document is only processed once, dramatically
    reducing cost when a document has many chunks.
    """
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",  # fast + cheap for batch enrichment
        max_tokens=200,
        system="You are a retrieval optimisation assistant. Be concise.",
        messages=[
            {
                "role": "user",
                "content": [
                    # Cache the whole document, only paid once per document
                    {
                        "type": "text",
                        "text": f"<document>\n{document}\n</document>",
                        "cache_control": {"type": "ephemeral"},
                    },
                    {
                        "type": "text",
                        "text": (
                            f"Here is the chunk we want to situate:\n<chunk>\n{chunk}\n</chunk>\n"
                            "Give a short succinct context to situate this chunk within the "
                            "document for improving search retrieval. Answer only with the context."
                        ),
                    },
                ],
            }
        ],
    )
    return response.content[0].text.strip()


def enrich_with_llm(
    doc_text: str,
    metadata: dict,
    chunk_size: int = 800,
    chunk_overlap: int = 100,
) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size, chunk_overlap=chunk_overlap
    )
    raw_chunks = splitter.split_text(doc_text)

    enriched = []
    for chunk in raw_chunks:
        context = generate_context(doc_text, chunk)
        # Prepend context to the chunk before embedding
        enriched_text = f"{context}\n\n{chunk}"
        enriched.append(Document(page_content=enriched_text, metadata=metadata))

    return enriched

def generate_hypothetical_questions(chunk: str, n: int = 3) -> list[str]:
    """
    Ask the LLM to generate n questions that this chunk answers.
    These questions are embedded alongside the chunk, bridging the
    query–document lexical gap.
    """
    response = client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=300,
        messages=[
            {
                "role": "user",
                "content": (
                    f"Given this text:\n<chunk>\n{chunk}\n</chunk>\n\n"
                    f"Generate {n} concise questions that this text directly answers. "
                    "Return one question per line, no bullets or numbering."
                ),
            }
        ],
    )
    return [q.strip() for q in response.content[0].text.strip().split("\n") if q.strip()]


# Example: embed questions + chunk together
def enrich_with_questions(doc_text: str, metadata: dict) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(chunk_size=800, chunk_overlap=100)
    raw_chunks = splitter.split_text(doc_text)

    enriched = []
    for chunk in raw_chunks:
        questions = generate_hypothetical_questions(chunk, n=3)
        questions_block = "\n".join(f"Q: {q}" for q in questions)
        enriched_text = f"{questions_block}\n\n{chunk}"
        enriched.append(Document(page_content=enriched_text, metadata=metadata))

    return enriched