Build a document chunking pipeline (recursive, semantic, token-aware)

Introduction

If you've ever fed a long document into an LLM only to hit a context-window error, or watched retrieval surface mangled fragments because a chunk split mid-sentence, you've felt the cost of getting chunking wrong. Teams that approximate token counts with character heuristics ship pipelines that silently overflow context windows in production—wasting inference spend on truncated answers and producing retrieval that drifts off-topic at every chunk boundary. By the end of this lesson, you will be able to implement a configurable document chunking pipeline that converts raw document text—from Unstructured.io partitioning or Crawl4AI markdown output—into uniformly sized DocumentChunk objects suitable for downstream Instructor-based relevance scoring. You will build a ChunkConfig Pydantic model that selects between recursive character splitting, semantic paragraph grouping, and token-aware boundary splitting, and you will use tiktoken's cl100k_base encoding to measure exact token counts so chunks stay within the LLM context budget.

Key Terminology

• Token-aware chunking: Splitting text using exact token counts from a tokenizer (tiktoken's cl100k_base encoding) rather than character length, so chunk sizes align with the LLM's real context budget.
• Chunk overlap: A configurable number of trailing tokens from one chunk that are repeated at the start of the next chunk, preserving cross-boundary context for retrieval; constrained to less than half of max_tokens to prevent runaway recursion.
• Separator hierarchy: The ordered list (\n\n, \n, . , ) that the recursive splitter tries from largest to smallest, falling through to a finer-grained boundary only when the current segment still exceeds max_tokens.

Concepts

Why Character Counts Fail at Scale

Most tutorials approximate token counts using character length divided by four. This heuristic breaks catastrophically for multilingual text, code-heavy documents, and structured data. The word "café" is one token in most tokenizers, but five characters. A Base64-encoded image string might be 10,000 characters but consume far fewer tokens due to tokenizer vocabulary patterns. When you are packing chunks into an LLM's context window for the Instructor-based relevance scoring pipeline that follows this stage, even a 5% token count error compounds across dozens of chunks, leading to either wasted context capacity or—worse—hard truncation errors from the API. Using tiktoken's cl100k_base encoding (the tokenizer behind GPT-4 and used as a reference standard) gives you exact counts that align with real model behavior.

Code Walkthrough

Chunking Strategy Overview

Before writing code, understand the three strategies and when each applies:

• Recursive Character Splitting: Splits text hierarchically using a priority list of separators (\n\n, \n, . , ). Tries the largest separator first; if a resulting chunk still exceeds the limit, recurses with the next separator. Best for general-purpose documents where structural markers like double newlines indicate paragraph boundaries.

• Semantic Paragraph Chunking: Groups consecutive paragraphs into chunks that stay below the token limit, treating each \n\n-delimited block as an atomic unit. Never splits mid-paragraph. Produces more coherent chunks for narrative content—reading material, articles, and web content ingested through Crawl4AI's markdown output.

• Token-Aware Splitting: Operates directly on token sequences from tiktoken, splitting at exact token boundaries with configurable overlap measured in tokens rather than characters. Essential when you need guaranteed maximum token counts, such as when packing chunks for context window injection with the Instructor-based relevance scoring service.

ChunkConfig.strategy drives a three-way branching pipeline that selects between recursive character splitting, semantic paragraph grouping, and token-aware boundary splitting—each critical for preserving context during document ingestion. Recursive mode applies a separator hierarchy and recurses when chunks exceed max_tokens, while semantic mode clusters paragraphs under a token limit. All paths converge through an overlap stage, producing List[DocumentChunk] objects scored by an Instructor-based relevance model to filter noise before downstream retrieval.

Loading diagram...

Configuration Models and the Pipeline Class

The pipeline accepts a ChunkConfig Pydantic model that controls every aspect of chunking behavior. This configuration-driven design means callers—whether the file upload endpoint handling python-multipart validated documents or the Crawl4AI web ingestion service—never embed chunking logic directly. They pass a config object and receive uniform DocumentChunk outputs.

The single module below defines three pieces in dependency order: the ChunkConfig and DocumentChunk Pydantic models that form the input/output contract, and the ChunkingPipeline dataclass that implements all three strategies behind one chunk_document() entry point. ChunkConfig uses a Literal type to restrict strategy to the three supported values, and a model_validator rejects configurations where chunk_overlap is ≥ half of max_tokens (which would cause infinite recursion). ChunkingPipeline lazily initializes the tiktoken encoder on first use and dispatches by config.strategy through a method map. The _recursive_split, _semantic_split, and _token_aware_split methods live side-by-side on the same class so one instance can serve any strategy choice without re-instantiation.

Code snippetpython
1from pydantic import BaseModel, model_validator, Field
2from typing import Literal
3from dataclasses import dataclass, field
4import tiktoken
5
6class ChunkConfig(BaseModel):
7    strategy: Literal["recursive", "semantic", "token_aware"] = "recursive"
8    max_tokens: int = Field(default=512, ge=64, le=8192)
9    chunk_overlap: int = Field(default=50, ge=0)
10    encoding_name: str = "cl100k_base"
11    separators: list[str] = Field(
12        default=["\n\n", "\n", ". ", " "]
13    )
14
15    @model_validator(mode="after")
16    def validate_overlap(self):
17        if self.chunk_overlap >= self.max_tokens // 2:
18            raise ValueError(
19                f"chunk_overlap ({self.chunk_overlap}) must be less than "
20                f"half of max_tokens ({self.max_tokens // 2})"
21            )
22        return self
23
24class DocumentChunk(BaseModel):
25    text: str
26    token_count: int
27    chunk_index: int
28    has_overlap: bool = False
29    source_document_id: str | None = None
30
31@dataclass
32class ChunkingPipeline:
33    config: ChunkConfig
34    _encoder: tiktoken.Encoding | None = field(
35        default=None, init=False, repr=False
36    )
37
38    def _get_encoder(self) -> tiktoken.Encoding:
39        if self._encoder is None:
40            self._encoder = tiktoken.get_encoding(
41                self.config.encoding_name
42            )
43        return self._encoder
44
45    def count_tokens(self, text: str) -> int:
46        return len(self._get_encoder().encode(text))
47
48    def chunk_document(
49        self, text: str, document_id: str | None = None
50    ) -> list[DocumentChunk]:
51        if not text or not text.strip():
52            return []
53        strategy_map = {
54            "recursive": self._recursive_split,
55            "semantic": self._semantic_split,
56            "token_aware": self._token_aware_split,
57        }
58        splitter = strategy_map[self.config.strategy]
59        raw_chunks = splitter(text)
60        return [
61            DocumentChunk(
62                text=chunk,
63                token_count=self.count_tokens(chunk),
64                chunk_index=i,
65                has_overlap=self.config.chunk_overlap > 0,
66                source_document_id=document_id,
67            )
68            for i, chunk in enumerate(raw_chunks)
69            if chunk.strip()
70        ]
71
72    def _recursive_split(
73        self, text: str, depth: int = 0
74    ) -> list[str]:
75        if self.count_tokens(text) <= self.config.max_tokens:
76            return [text]
77        if depth >= len(self.config.separators):
78            encoder = self._get_encoder()
79            tokens = encoder.encode(text)
80            return [
81                encoder.decode(
82                    tokens[: self.config.max_tokens]
83                )
84            ]
85        sep = self.config.separators[depth]
86        segments = text.split(sep)
87        chunks, current = [], ""
88        for segment in segments:
89            candidate = f"{current}{sep}{segment}" if current else segment
90            if self.count_tokens(candidate) <= self.config.max_tokens:
91                current = candidate
92            else:
93                if current:
94                    chunks.append(current)
95                if self.count_tokens(segment) > self.config.max_tokens:
96                    chunks.extend(
97                        self._recursive_split(segment, depth + 1)
98                    )
99                    current = ""
100                else:
101                    current = segment
102        if current:
103            chunks.append(current)
104        return self._apply_overlap(chunks)
105
106    def _semantic_split(self, text: str) -> list[str]:
107        paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
108        chunks, current_parts, current_tokens = [], [], 0
109        for para in paragraphs:
110            para_tokens = self.count_tokens(para)
111            if para_tokens > self.config.max_tokens:
112                if current_parts:
113                    chunks.append("\n\n".join(current_parts))
114                    current_parts, current_tokens = [], 0
115                chunks.extend(self._recursive_split(para))
116                continue
117            if current_tokens + para_tokens > self.config.max_tokens:
118                chunks.append("\n\n".join(current_parts))
119                current_parts, current_tokens = [], 0
120            current_parts.append(para)
121            current_tokens += para_tokens
122        if current_parts:
123            chunks.append("\n\n".join(current_parts))
124        return self._apply_overlap(chunks)
125
126    def _token_aware_split(self, text: str) -> list[str]:
127        encoder = self._get_encoder()
128        tokens = encoder.encode(text)
129        if len(tokens) <= self.config.max_tokens:
130            return [text]
131        chunks = []
132        step = self.config.max_tokens - self.config.chunk_overlap
133        for start in range(0, len(tokens), step):
134            chunk_tokens = tokens[start : start + self.config.max_tokens]
135            chunks.append(encoder.decode(chunk_tokens))
136        return chunks
137
138    def _apply_overlap(self, chunks: list[str]) -> list[str]:
139        if self.config.chunk_overlap == 0 or len(chunks) <= 1:
140            return chunks
141        encoder = self._get_encoder()
142        result = [chunks[0]]
143        for i in range(1, len(chunks)):
144            prev_tokens = encoder.encode(chunks[i - 1])
145            overlap_text = encoder.decode(
146                prev_tokens[-self.config.chunk_overlap :]
147            )
148            result.append(f"{overlap_text} {chunks[i]}")
149        return result

• ChunkConfig restricts strategy to one of "recursive", "semantic", or "token_aware" via a Literal type. max_tokens is bounded between 64 and 8192 — low enough to prevent degenerate single-sentence chunks, high enough to fit a sensible per-chunk slice of typical LLM context windows. The separators list defines the recursive splitter's hierarchy: paragraph break → newline → sentence boundary → word boundary.
• The model_validator rejects configurations where chunk_overlap ≥ max_tokens // 2; without this guard the recursive splitter would produce overlapping regions larger than the non-overlapping content, causing redundant chunks and potential infinite recursion.
• DocumentChunk carries the chunk text, the exact token count (computed via tiktoken at split time), the positional index inside the source document, a has_overlap flag, and an optional source_document_id that links chunks back to the original record in PostgreSQL.
• ChunkingPipeline is a dataclass that wraps a ChunkConfig and a lazily initialized tiktoken encoder. field(default=None, init=False) keeps the encoder out of the constructor signature while still allowing internal caching across calls.
• _get_encoder() initializes the encoder on first call using config.encoding_name. cl100k_base covers GPT-4 and GPT-3.5-turbo; switch to o200k_base for newer models. count_tokens() is the single source of truth for token measurement — never approximate with len(text) // 4.
• chunk_document() is the public entry point. It returns an empty list for blank input, dispatches to the selected strategy via a dictionary lookup, and wraps raw string chunks into typed DocumentChunk records, filtering whitespace-only chunks that can result from splitting on consecutive separators.
• _recursive_split() returns the text as-is if it already fits max_tokens. If every separator is exhausted, it falls back to hard token truncation — guaranteeing termination even on pathological input like a single 10,000-character word. The core loop greedily accumulates segments on the current-depth separator and recurses with depth + 1 whenever a single segment is itself too large.
• _semantic_split() treats paragraph boundaries as inviolable: paragraphs are joined with \n\n and accumulated until adding the next would exceed the budget. A single oversized paragraph (e.g. an Unstructured.io table element with thousands of tokens) is flushed and delegated to _recursive_split() so structural integrity is preserved everywhere else.
• _token_aware_split() encodes the entire text once, then slides a window of max_tokens width with step max_tokens - chunk_overlap over the raw token sequence — producing perfectly sized chunks with exact token-level overlap. Token boundaries may fall mid-word, which is acceptable for embedding-based retrieval but may require post-processing for display.
• _apply_overlap() prepends the last chunk_overlap tokens of the previous chunk to each subsequent chunk. Overlap is measured in tokens, not characters, so behavior stays consistent regardless of language or encoding. The first chunk receives no prefix since there is no predecessor.

Integrating with Upstream Parsers

In practice, the chunking pipeline receives input from two sources: Unstructured.io's element-based partitioning (for uploaded files processed through the python-multipart validation endpoint) and Crawl4AI's markdown output (for web-ingested content). The integration is straightforward—flatten elements to text and pass through chunk_document():

The chunk_elements function bridges Unstructured.io's multi-format parser output with a downstream chunking pipeline. It accepts a list of Element objects—extracted from PDFs, DOCX, or PPTX files—filters out empty entries, and concatenates their .text fields into a single string. A ChunkingPipeline instance, configured via ChunkConfig, then splits that text into DocumentChunk objects tagged with document_id, making parsed content ready for vector storage and retrieval.

1# From Unstructured.io elements (PDF, DOCX, PPTX)
2from unstructured.documents.elements import Element
3
4def chunk_elements(
5    elements: list[Element],
6    config: ChunkConfig,
7    document_id: str,
8) -> list[DocumentChunk]:
9    text = "\n\n".join(
10        el.text for el in elements if el.text and el.text.strip()
11    )
12    pipeline = ChunkingPipeline(config=config)
13    return pipeline.chunk_document(text, document_id=document_id)

Do's and Don'ts

Do's

1. ✓Do use ChunkConfig's model_validator to reject overlaps ≥ half of max_tokens — a chunk_overlap that meets or exceeds max_tokens // 2 causes the recursive splitter to produce chunks that can never shrink below the limit, looping indefinitely; catching this at config construction time prevents silent infinite recursion at runtime.
2. ✓Do use tiktoken's cl100k_base encoding to measure token counts rather than estimating from character length — character heuristics diverge from actual token counts for punctuation-dense text, code, and non-ASCII content, leading chunks to silently overflow the LLM context budget and causing truncated answers or stale retrieval in the Instructor-based relevance scoring stage.
3. ✓Do match ChunkConfig.strategy to the document source — use recursive for structured documents where double newlines mark paragraph boundaries, semantic for narrative Crawl4AI markdown output where splitting mid-paragraph degrades coherence, and token_aware when packing chunks for guaranteed maximum-token context-window injection.

Don'ts

1. ✗Don't embed chunking logic directly in file upload endpoints or Crawl4AI ingestion handlers — hardcoding separator choices or token limits at the call site bypasses the ChunkingPipeline dispatch mechanism and makes it impossible to swap strategies without modifying ingestion code; pass a ChunkConfig object and let chunk_document() handle all three paths.
2. ✗Don't split mid-paragraph when ingesting Crawl4AI markdown with the semantic strategy — the semantic chunker treats each \n\n-delimited block as an atomic unit precisely to preserve coherence; forcing a character split across a paragraph boundary fragments context and degrades Instructor-based relevance scores on the resulting DocumentChunk objects.
3. ✗Don't reuse a single ChunkingPipeline instance configured for one strategy across documents that need a different one — while the class supports all three strategies, mixing token_aware overlap (measured in tokens) with a recursive config that specifies character-based separators produces mismatched chunk_overlap semantics and inconsistent DocumentChunk sizing across the pipeline.