Runs in your browser — no install. (Works once the repo is public.)

In [ ]:

# --- Colab setup (skipped when running locally) ---
import os, sys
if 'google.colab' in sys.modules and not os.path.exists('data/sample_corpus.json'):
    !git clone -q https://github.com/mdhabibi/llm-search-handbook.git
    %cd llm-search-handbook
    !pip -q install -r requirements.txt

# --- Colab setup (skipped when running locally) --- import os, sys if 'google.colab' in sys.modules and not os.path.exists('data/sample_corpus.json'): !git clone -q https://github.com/mdhabibi/llm-search-handbook.git %cd llm-search-handbook !pip -q install -r requirements.txt

Chapter 11 — Chunking & Ingestion Pipelines¶

Real documents are long and messy. Before search, they must be loaded, cleaned, chunked, embedded, and indexed. Chunking is the highest-leverage knob.

Setup¶

pip install sentence-transformers chromadb

In [ ]:

# Bootstrap: locate repo root and import shared helpers
import sys, os, json
d = os.getcwd()
while d != os.path.dirname(d) and not os.path.exists(os.path.join(d, 'data', 'sample_corpus.json')):
    d = os.path.dirname(d)
ROOT = d; sys.path.insert(0, os.path.join(ROOT, 'src'))
import numpy as np
from corpus import load_corpus, tokenize

# Bootstrap: locate repo root and import shared helpers import sys, os, json d = os.getcwd() while d != os.path.dirname(d) and not os.path.exists(os.path.join(d, 'data', 'sample_corpus.json')): d = os.path.dirname(d) ROOT = d; sys.path.insert(0, os.path.join(ROOT, 'src')) import numpy as np from corpus import load_corpus, tokenize

1. Chunking strategies¶

From src/chunking.py: fixed-size (with overlap) and sentence-based.

In [ ]:

from chunking import fixed_word_chunks, sentence_chunks

long_text = (
    'The blue whale is the largest animal known to have ever lived. '
    'Despite its enormous size it feeds almost entirely on tiny krill. '
    'Whales are mammals: they breathe air and nurse their young with milk. '
    'They migrate thousands of miles between feeding and breeding grounds each year. '
    'Their deep calls travel across entire ocean basins.'
)

print('FIXED-SIZE (size=15, overlap=5):')
for c in fixed_word_chunks(long_text, size=15, overlap=5): print('  |', c)
print('\nSENTENCE-BASED (2 sentences, overlap 1):')
for c in sentence_chunks(long_text, max_sentences=2, overlap_sentences=1): print('  |', c)

from chunking import fixed_word_chunks, sentence_chunks long_text = ( 'The blue whale is the largest animal known to have ever lived. ' 'Despite its enormous size it feeds almost entirely on tiny krill. ' 'Whales are mammals: they breathe air and nurse their young with milk. ' 'They migrate thousands of miles between feeding and breeding grounds each year. ' 'Their deep calls travel across entire ocean basins.' ) print('FIXED-SIZE (size=15, overlap=5):') for c in fixed_word_chunks(long_text, size=15, overlap=5): print(' |', c) print('\nSENTENCE-BASED (2 sentences, overlap 1):') for c in sentence_chunks(long_text, max_sentences=2, overlap_sentences=1): print(' |', c)

2. Why overlap matters¶

Without overlap, a fact that straddles a boundary is split and neither chunk holds the whole thought.

In [ ]:

no_ov  = fixed_word_chunks(long_text, size=15, overlap=0)
with_ov= fixed_word_chunks(long_text, size=15, overlap=5)
print('no overlap  -> boundary between chunk 0 and 1:')
print('   ...', ' '.join(no_ov[0].split()[-4:]), '||', ' '.join(no_ov[1].split()[:4]), '...')
print('with overlap-> the seam text is repeated so context survives:')
print('   ...', ' '.join(with_ov[0].split()[-4:]), '||', ' '.join(with_ov[1].split()[:4]), '...')

no_ov = fixed_word_chunks(long_text, size=15, overlap=0) with_ov= fixed_word_chunks(long_text, size=15, overlap=5) print('no overlap -> boundary between chunk 0 and 1:') print(' ...', ' '.join(no_ov[0].split()[-4:]), '||', ' '.join(no_ov[1].split()[:4]), '...') print('with overlap-> the seam text is repeated so context survives:') print(' ...', ' '.join(with_ov[0].split()[-4:]), '||', ' '.join(with_ov[1].split()[:4]), '...')

3. A configurable ingestion pipeline¶

load → clean → chunk → embed → index. Swap the chunker or embedder freely.

In [ ]:

import re
class IngestionPipeline:
    def __init__(self, encode_fn, chunker=lambda t: sentence_chunks(t, 3, 1)):
        self.encode_fn, self.chunker = encode_fn, chunker
        self.chunks, self.meta, self.matrix = [], [], None
    @staticmethod
    def clean(text): return re.sub(r'\s+', ' ', text).strip()
    def run(self, docs):
        for d in docs:
            for pos, ch in enumerate(self.chunker(self.clean(d['text']))):
                self.chunks.append(ch)
                self.meta.append({'doc_id': d['id'], 'title': d['title'], 'pos': pos})
        M = np.asarray(self.encode_fn(self.chunks), float)
        self.matrix = M / np.clip(np.linalg.norm(M, axis=1, keepdims=True), 1e-12, None)
        return self
    def search(self, query, k=3):
        q = np.asarray(self.encode_fn([query]), float)[0]; q /= np.linalg.norm(q)+1e-12
        order = np.argsort(-(self.matrix @ q))[:k]
        return [(self.chunks[i], self.meta[i]) for i in order]

print('IngestionPipeline defined. Run it with a real encoder:')
print('  from sentence_transformers import SentenceTransformer')
print("  m = SentenceTransformer('all-MiniLM-L6-v2')")
print('  pipe = IngestionPipeline(m.encode).run(load_corpus())')
print('  pipe.search("what do whales eat?")')

import re class IngestionPipeline: def __init__(self, encode_fn, chunker=lambda t: sentence_chunks(t, 3, 1)): self.encode_fn, self.chunker = encode_fn, chunker self.chunks, self.meta, self.matrix = [], [], None @staticmethod def clean(text): return re.sub(r'\s+', ' ', text).strip() def run(self, docs): for d in docs: for pos, ch in enumerate(self.chunker(self.clean(d['text']))): self.chunks.append(ch) self.meta.append({'doc_id': d['id'], 'title': d['title'], 'pos': pos}) M = np.asarray(self.encode_fn(self.chunks), float) self.matrix = M / np.clip(np.linalg.norm(M, axis=1, keepdims=True), 1e-12, None) return self def search(self, query, k=3): q = np.asarray(self.encode_fn([query]), float)[0]; q /= np.linalg.norm(q)+1e-12 order = np.argsort(-(self.matrix @ q))[:k] return [(self.chunks[i], self.meta[i]) for i in order] print('IngestionPipeline defined. Run it with a real encoder:') print(' from sentence_transformers import SentenceTransformer') print(" m = SentenceTransformer('all-MiniLM-L6-v2')") print(' pipe = IngestionPipeline(m.encode).run(load_corpus())') print(' pipe.search("what do whales eat?")')

4. Run it (needs the embedding model)¶

In [ ]:

from sentence_transformers import SentenceTransformer
m = SentenceTransformer('all-MiniLM-L6-v2')
pipe = IngestionPipeline(m.encode).run(load_corpus())
print(f'Indexed {len(pipe.chunks)} chunks from {len(load_corpus())} documents.\n')
for chunk, meta in pipe.search('what do whales eat?'):
    print(f"  [{meta['title']} #{meta['pos']}] {chunk[:70]}...")

from sentence_transformers import SentenceTransformer m = SentenceTransformer('all-MiniLM-L6-v2') pipe = IngestionPipeline(m.encode).run(load_corpus()) print(f'Indexed {len(pipe.chunks)} chunks from {len(load_corpus())} documents.\n') for chunk, meta in pipe.search('what do whales eat?'): print(f" [{meta['title']} #{meta['pos']}] {chunk[:70]}...")

5. Debugging RAG: retrieval or generation?¶

When an answer is wrong, first check what was retrieved. Most failures are retrieval failures — often fixed by chunking.

Bad answer? -> Was the right chunk retrieved?
   no  -> RETRIEVAL: chunk size/overlap, embed model, k, hybrid/rerank
   yes -> GENERATION: prompt grounding/refusal, context budget, model

Takeaway & exercises¶

Chunking + a clean ingestion pipeline is what makes RAG work on real corpora.

Exercises

1. Compare retrieval when a long doc is embedded whole vs. chunked. Which is more precise?
2. Sweep chunk size {10, 30, 60} words; evaluate with Chapter 9 metrics.
3. Add token-based chunking with tiktoken and compare to word-based.
4. Add a content hash cache so unchanged docs aren't re-embedded.