Turn clean documents into retrieval units that preserve answers, citations, and measurable search quality.
A clean incident-runbook record gives retrieval a concrete starting point:
incident-runbook-v3.pdf, page 7: Rollback failure: page on-call within 15 minutes with deploy ID.
That sentence is trustworthy evidence only while its pieces stay together. Split it into one chunk containing "Rollback failure" and another containing "15 minutes with deploy ID," and a search result can retrieve the condition without the deadline or the deadline without its required identifier.
Chunking turns normalized records into searchable evidence units. In a retrieval-augmented generation (RAG) system, a retriever selects passages from an external index and the generator answers from those passages.[1] Your chunk boundaries decide what a retrieved passage can prove.
Suppose an engineer asks, "A rollback failed after deploy pay-742. When do I page, and what ID do I include?"
| Candidate retrieval unit | Searchable? | Answerable? | Problem |
|---|---|---|---|
Rollback failure: page on-call within | yes | no | deadline and required ID are missing |
15 minutes with deploy ID. Routine deploy notes: archive within 14 days. | yes | no | condition is missing and a competing rule is present |
Rollback failure: page on-call within 15 minutes with deploy ID. | yes | yes | preserves condition, deadline, and required ID |
The first engineering requirement isn't "make chunks small." It's "make each retrieved chunk a defensible piece of evidence."
1from dataclasses import dataclass
2
3@dataclass(frozen=True)
4class Chunk:
5 chunk_id: str
6 text: str
7 source_id: str
8 locator: str
9
10def is_answerable(chunk: Chunk) -> bool:
11 required = ["rollback failure", "15 minutes", "deploy id"]
12 lowered = chunk.text.lower()
13 return all(phrase in lowered for phrase in required)
14
15chunks = [
16 Chunk("broken-condition", "Rollback failure: page on-call within", "incident-runbook-v3.pdf", "page=7"),
17 Chunk("broken-window", "15 minutes with deploy ID. Routine deploy notes archive within 14 days.", "incident-runbook-v3.pdf", "page=7"),
18 Chunk("complete", "Rollback failure: page on-call within 15 minutes with deploy ID.", "incident-runbook-v3.pdf", "page=7"),
19]
20
21for chunk in chunks:
22 print(f"{chunk.chunk_id}: answerable={is_answerable(chunk)}")1broken-condition: answerable=False
2broken-window: answerable=False
3complete: answerable=TrueThe simplest splitter takes windows of tokens. If a window has size and overlap , the next window starts after tokens. Overlap repeats boundary text, which can help continuity, but it can't guarantee that a complete policy rule survives.
The lab uses whitespace-separated words as visible token stand-ins. A production pipeline should measure with the tokenizer used by its embedding model.
1def fixed_windows(text: str, size: int, overlap: int) -> list[str]:
2 if size <= 0 or overlap < 0 or overlap >= size:
3 raise ValueError("require size > overlap >= 0")
4 words = text.split()
5 step = size - overlap
6 return [
7 " ".join(words[start : start + size])
8 for start in range(0, len(words), step)
9 if words[start : start + size]
10 ]
11
12policy = (
13 "Rollback failure page on-call within 15 minutes with deploy ID. "
14 "Routine deploy notes archive within 14 days."
15)
16
17for index, chunk in enumerate(fixed_windows(policy, size=6, overlap=2)):
18 print(f"{index}: {chunk}")10: Rollback failure page on-call within 15
21: within 15 minutes with deploy ID.
32: deploy ID. Routine deploy notes archive
43: notes archive within 14 days.
54: days.The late notes archive within 14 days. window and final days. fragment expose a second failure: an orphan tail can lose the condition that gives a value meaning. A production fallback should merge an undersized tail into the previous window or send it for review instead of indexing it blindly.
Now measure what overlap bought you. More repeated words increase indexed text, but the rule becomes useful only when a window carries all three required pieces.
1def fixed_windows(text: str, size: int, overlap: int) -> list[str]:
2 words = text.split()
3 step = size - overlap
4 return [
5 " ".join(words[start : start + size])
6 for start in range(0, len(words), step)
7 if words[start : start + size]
8 ]
9
10def carries_rollback_rule(text: str) -> bool:
11 lowered = text.lower()
12 return all(term in lowered for term in ["rollback failure", "15 minutes", "deploy id"])
13
14policy = "Rollback failure page on-call within 15 minutes with deploy ID. Routine deploy notes archive within 14 days."
15for overlap in [0, 2, 4]:
16 chunks = fixed_windows(policy, size=7, overlap=overlap)
17 complete = sum(carries_rollback_rule(chunk) for chunk in chunks)
18 indexed_words = sum(len(chunk.split()) for chunk in chunks)
19 print(f"overlap={overlap}: chunks={len(chunks)} indexed_words={indexed_words} complete={complete}")1overlap=0: chunks=3 indexed_words=17 complete=0
2overlap=2: chunks=4 indexed_words=23 complete=0
3overlap=4: chunks=6 indexed_words=35 complete=0This is the right posture for overlap: a configuration to evaluate, not a ritual to apply to every document.
File ingestion preserved headings and source locations. Use them. A heading boundary is more meaningful than an arbitrary word count when the source already says which rule belongs together.
LangChain's RecursiveCharacterTextSplitter is a common generic-text baseline. Its documentation describes trying separators in order, with the default sequence ["\n\n", "\n", " ", ""], so paragraphs are preferred before smaller cuts.[2] Start with explicit heading boundaries when the source provides them. Add the smaller-cut fallback only for sections that still exceed your size limit.
1from dataclasses import dataclass
2
3@dataclass(frozen=True)
4class PolicyChunk:
5 heading: str
6 body: str
7 indexed_text: str
8 locator: str
9
10def headed_chunks(markdown: str, source_locator: str) -> list[PolicyChunk]:
11 chunks: list[PolicyChunk] = []
12 heading = "Document"
13 body: list[str] = []
14
15 def flush() -> None:
16 if body:
17 body_text = " ".join(body)
18 chunks.append(
19 PolicyChunk(
20 heading,
21 body_text,
22 f"{heading}\n{body_text}",
23 f"{source_locator}#{heading.lower().replace(' ', '-')}",
24 )
25 )
26 body.clear()
27
28 for line in markdown.strip().splitlines():
29 if line.startswith("## "):
30 flush()
31 heading = line[3:]
32 elif line.strip():
33 body.append(line.strip())
34 flush()
35 return chunks
36
37policy = """## Rollback failure
38Page on-call within 15 minutes with deploy ID.
39## Routine deploy notes
40Archive within 14 days after release."""
41
42for chunk in headed_chunks(policy, "page=7"):
43 print(chunk.indexed_text.replace("\n", " | "), "|", chunk.locator)1Rollback failure | Page on-call within 15 minutes with deploy ID. | page=7#rollback-failure
2Routine deploy notes | Archive within 14 days after release. | page=7#routine-deploy-notesEach chunk's indexed_text includes its heading, so the searchable text keeps the condition attached to the deadline. The locator separately preserves the path back to original evidence.
A policy table needs an equally deliberate rule. Splitting a row away from its column names turns exact information into ambiguous fragments.
1table = [
2 "| Condition | Window | Required evidence |",
3 "| --- | --- | --- |",
4 "| Rollback failure | 15 minutes | Deploy ID |",
5 "| Routine deploy notes | 14 days | Release summary |",
6]
7
8def table_chunk(lines: list[str], heading: str) -> dict[str, str]:
9 return {
10 "heading": heading,
11 "text": "\n".join(lines),
12 "quality_check": "header_present" if lines[0].startswith("| Condition |") else "review",
13 }
14
15chunk = table_chunk(table, "Runbook windows")
16print(chunk["heading"], chunk["quality_check"], f"rows={len(table) - 2}")
17print("15 minutes" in chunk["text"] and "Deploy ID" in chunk["text"])1Runbook windows header_present rows=2
2TrueFor a long section, split inside it while carrying its heading and original locator into every child chunk. Merge a tiny final window back into the preceding child so the fallback doesn't emit an orphan fragment.
1def section_windows(
2 text: str, heading: str, source: str, size: int, min_tail_words: int
3) -> list[dict[str, str]]:
4 words = text.split()
5 windows = [words[start : start + size] for start in range(0, len(words), size)]
6 if len(windows) > 1 and len(windows[-1]) < min_tail_words:
7 windows[-2].extend(windows.pop())
8
9 return [
10 {
11 "text": " ".join(window),
12 "heading": heading,
13 "source": source,
14 "chunk_id": f"{source}#{heading.lower().replace(' ', '-')}-{index}",
15 }
16 for index, window in enumerate(windows)
17 ]
18
19children = section_windows(
20 "Page on-call within 15 minutes with deploy ID. Include the service name and rollback attempt.",
21 heading="Rollback failure",
22 source="incident-runbook-v3.pdf:page=7",
23 size=7,
24 min_tail_words=4,
25)
26
27for child in children:
28 print(child["chunk_id"], "|", child["heading"], "|", child["text"])1incident-runbook-v3.pdf:page=7#rollback-failure-0 | Rollback failure | Page on-call within 15 minutes with deploy
2incident-runbook-v3.pdf:page=7#rollback-failure-1 | Rollback failure | ID. Include the service name and rollback attempt.
A chunking strategy isn't good because it sounds fancy. It works when labeled questions retrieve complete supporting evidence.
Start with a tiny, transparent score before involving a vector database. For the query terms {rollback, failure, deploy, minutes}, the complete rollback-runbook chunk matches all four. A fragment that contains only {rollback, failure} may rank, but it can't answer the question.
| Chunk | Matching query terms | Contains answer? |
|---|---|---|
| complete rollback rule | 4 / 4 | yes |
| rollback-condition fragment | 2 / 4 | no |
| routine archive rule | 1 / 4 | no |
This lexical scorer is deliberately simple. It isolates the effect of boundaries; replace its score with real embeddings after the fixture and expected evidence are stable.
1import re
2
3def terms(text: str) -> set[str]:
4 return set(re.findall(r"[a-z0-9]+", text.lower()))
5
6def retrieve(query: str, chunks: list[dict[str, str]]) -> dict[str, str]:
7 query_terms = terms(query)
8 return max(chunks, key=lambda chunk: len(query_terms & terms(chunk["text"])))
9
10query = "rollback failure deploy minutes"
11chunks = [
12 {"id": "rollback-rule", "text": "Rollback failure: page on-call within 15 minutes with deploy ID."},
13 {"id": "routine-rule", "text": "Routine deploy notes: archive within 14 days."},
14]
15
16hit = retrieve(query, chunks)
17print(hit["id"], hit["text"])1rollback-rule Rollback failure: page on-call within 15 minutes with deploy ID.Now compare a broken fixed-window configuration against a section-aware configuration using the same question and the same expected evidence phrase.
1import re
2
3def terms(text: str) -> set[str]:
4 return set(re.findall(r"[a-z0-9]+", text.lower()))
5
6def top_chunk(query: str, chunks: list[str]) -> str:
7 query_terms = terms(query)
8 return max(chunks, key=lambda text: len(query_terms & terms(text)))
9
10query = "rollback failure deploy id"
11expected = "Rollback failure: page on-call within 15 minutes with deploy ID."
12configs = {
13 "broken-fixed": [
14 "Rollback failure: page on-call within",
15 "15 minutes with deploy ID. Routine deploy notes: archive within 14 days.",
16 ],
17 "section-aware": [
18 "Rollback failure: page on-call within 15 minutes with deploy ID.",
19 "Routine deploy notes: archive within 14 days.",
20 ],
21}
22
23for name, chunks in configs.items():
24 hit = top_chunk(query, chunks)
25 print(f"{name}: complete={expected in hit}")1broken-fixed: complete=False
2section-aware: complete=TrueOne question proves little. Ship a small labeled set containing policy exceptions, tables, and boundary failures, then measure each candidate splitter on exactly that set.
1import re
2
3def terms(text: str) -> set[str]:
4 return set(re.findall(r"[a-z0-9]+", text.lower()))
5
6def retrieve(query: str, chunks: list[dict[str, str]]) -> dict[str, str]:
7 query_terms = terms(query)
8 return max(chunks, key=lambda chunk: len(query_terms & terms(chunk["text"])))
9
10chunks = [
11 {"id": "rollback", "text": "Rollback failure: page on-call within 15 minutes with deploy ID."},
12 {"id": "routine", "text": "Routine deploy notes: archive within 14 days."},
13 {"id": "commander", "text": "Commander handoff requires the active incident channel."},
14]
15cases = [
16 ("rollback failure deploy id", "rollback", "15 minutes"),
17 ("routine deploy archive", "routine", "14 days"),
18 ("commander incident channel", "commander", "active incident"),
19]
20
21passed = 0
22for query, expected_id, evidence in cases:
23 hit = retrieve(query, chunks)
24 ok = hit["id"] == expected_id and evidence in hit["text"]
25 passed += int(ok)
26 print(query, "PASS" if ok else "FAIL")
27print(f"summary={passed}/{len(cases)}")1rollback failure deploy id PASS
2routine deploy archive PASS
3commander incident channel PASS
4summary=3/3When you replace this lexical baseline with embeddings, the assertions stay useful: retrieve the correct source and retain text sufficient to answer.
Some questions match a narrow sentence, while a faithful answer needs its surrounding section. A parent-child design indexes small children for search and stores a pointer to the larger source section returned for generation.[3]
1parents = {
2 "rollback": "Rollback failure: page on-call within 15 minutes with deploy ID. Include the service name.",
3 "routine": "Routine deploy notes: archive within 14 days after release.",
4}
5children = [
6 {"text": "15 minutes with deploy ID", "parent_id": "rollback"},
7 {"text": "14 days after release", "parent_id": "routine"},
8]
9
10match = next(child for child in children if "deploy ID" in child["text"])
11print(match["text"])
12print(parents[match["parent_id"]])115 minutes with deploy ID
2Rollback failure: page on-call within 15 minutes with deploy ID. Include the service name.Child windows inside a longer section also benefit from their section label. Embedding a child with a contextual header is a cheap hypothesis to test against your labeled set, not a promise of improvement.
1def indexed_text(heading: str, text: str, source: str) -> str:
2 return f"Source: {source}\nSection: {heading}\n{text}"
3
4child = indexed_text(
5 heading="Rollback failure",
6 text="Page on-call within 15 minutes with deploy ID.",
7 source="Incident Runbook",
8)
9print(child)1Source: Incident Runbook
2Section: Rollback failure
3Page on-call within 15 minutes with deploy ID.Structure-aware chunks handle many handbooks and policy pages. Some corpora force different choices:
| Failure after measuring baseline | Candidate experiment | What must still be checked |
|---|---|---|
| one paragraph shifts between multiple topics | semantic boundary detection | hard size cap and labeled-query score |
| tiny match lacks surrounding explanation | parent-child expansion | deduplication and generation budget |
| short child is ambiguous without its section | contextual header | retrieval comparison against no-header baseline |
| meaning depends on far-away document context | late chunking | model support, latency, and measured retrieval gain |
Semantic chunking places candidate boundaries where neighboring sentence representations change sharply. The next small lab uses transparent topic vectors, so you can see the boundary without trusting an external embedding service.
1import math
2
3def vector(sentence: str) -> list[float]:
4 lowered = sentence.lower()
5 return [
6 float(sum(word in lowered for word in ["rollback", "deploy", "minutes", "id"])),
7 float(sum(word in lowered for word in ["commander", "handoff", "channel"])),
8 ]
9
10def cosine(left: list[float], right: list[float]) -> float:
11 dot = sum(a * b for a, b in zip(left, right))
12 left_norm = math.sqrt(sum(a * a for a in left))
13 right_norm = math.sqrt(sum(b * b for b in right))
14 return dot / (left_norm * right_norm) if left_norm and right_norm else 0.0
15
16sentences = [
17 "Rollback failures require deploy IDs.",
18 "Page on-call within 15 minutes.",
19 "Commander handoff uses the active incident channel.",
20]
21
22for left, right in zip(sentences, sentences[1:]):
23 similarity = cosine(vector(left), vector(right))
24 print(f"{similarity:.2f}", "boundary" if similarity < 0.50 else "keep together")11.00 keep together
20.32 boundaryLate chunking is a different escalation. Günther et al. encode the longer text first and apply chunk pooling after the transformer's contextual token representations have been computed.[4] That means a child representation can carry information from surrounding document text, but it requires a compatible long-context embedding stack and must earn its additional cost in evaluation.
For the incident-runbook corpus, a credible first release is straightforward:
| Design choice | Initial decision | Evidence to collect |
|---|---|---|
| Default boundary | heading-aware sections, recursive fallback | answerable-chunk rate and retrieval regression suite |
| Tables | keep header plus rows together | exact-value questions preserve correct row meaning |
| Overlap | off for complete policy sections; test for fallback text | index size, duplicate hits, and labeled-query results |
| Metadata | source, locator, heading, checksum | cited answer can return to original record |
| Escalation | parent expansion before semantic or late chunking | failure examples that justify extra complexity |
Your release gate can be encoded as an executable manifest check.
1from hashlib import sha256
2
3chunks = [
4 {
5 "id": "rollback",
6 "text": "Rollback failure: page on-call within 15 minutes with deploy ID.",
7 "source": "incident-runbook-v3.pdf",
8 "locator": "page=7#rollback-failure",
9 "heading": "Rollback failure",
10 "quality": "ready",
11 },
12 {
13 "id": "broken",
14 "text": "15 minutes with deploy ID.",
15 "source": "incident-runbook-v3.pdf",
16 "locator": "page=7#fragment",
17 "heading": "Fragment",
18 "quality": "review",
19 },
20]
21
22for chunk in chunks:
23 chunk["checksum"] = sha256(chunk["text"].encode()).hexdigest()
24
25required_metadata = ("source", "locator", "heading", "checksum")
26indexable = [
27 chunk for chunk in chunks
28 if chunk["quality"] == "ready"
29 and all(chunk[field] for field in required_metadata)
30]
31print(f"indexable={[chunk['id'] for chunk in indexable]}")
32print(f"blocked={len(chunks) - len(indexable)}")1indexable=['rollback']
2blocked=1Clean evidence records from ingestion can now become retrieval units that can be evaluated.
15 minutes without Rollback failure. Fix: split at section boundaries and assert answerability.14 days. without Routine deploy notes. Fix: merge undersized tails into the preceding child or block them for review.Answer every question, then check your score. Score above 75% to mark this lesson complete.
8 questions remaining.
Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks.
Lewis, P., et al. · 2020 · NeurIPS 2020
RecursiveCharacterTextSplitter
LangChain · 2023
ParentDocumentRetriever
LangChain · 2024
Late Chunking: Contextual Chunk Embeddings Using Long-Context Embedding Models.
Günther, M., et al. · 2024 · arXiv preprint
Questions and insights from fellow learners.