1 min read
Document Chunking Strategies for RAG Systems
Introduction
Document chunking is a critical step in RAG pipelines that directly impacts retrieval quality. This post covers various chunking strategies, their trade-offs, and how to choose the right approach for your use case.
Chunking Fundamentals
from dataclasses import dataclass
from typing import List, Dict, Optional, Callable
from abc import ABC, abstractmethod
@dataclass
class Chunk:
id: str
content: str
metadata: Dict
start_index: int
end_index: int
parent_id: Optional[str] = None
class ChunkingStrategy(ABC):
"""Abstract base class for chunking strategies"""
@abstractmethod
def chunk(self, text: str, metadata: Dict = None) -> List[Chunk]:
pass
class ChunkingConfig:
"""Configuration for chunking"""
def __init__(
self,
chunk_size: int = 512,
chunk_overlap: int = 50,
min_chunk_size: int = 100,
separator: str = "\n\n"
):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.min_chunk_size = min_chunk_size
self.separator = separator
Fixed-Size Chunking
import uuid
class FixedSizeChunker(ChunkingStrategy):
"""Simple fixed-size chunking with overlap"""
def __init__(self, config: ChunkingConfig = None):
self.config = config or ChunkingConfig()
def chunk(self, text: str, metadata: Dict = None) -> List[Chunk]:
"""Split text into fixed-size chunks"""
chunks = []
start = 0
chunk_num = 0
while start < len(text):
end = start + self.config.chunk_size
# Adjust end to word boundary
if end < len(text):
end = self._find_word_boundary(text, end)
chunk_text = text[start:end].strip()
if len(chunk_text) >= self.config.min_chunk_size:
chunks.append(Chunk(
id=f"chunk_{uuid.uuid4().hex[:8]}",
content=chunk_text,
metadata={
**(metadata or {}),
"chunk_index": chunk_num,
"chunking_strategy": "fixed_size"
},
start_index=start,
end_index=end
))
chunk_num += 1
# Move start with overlap
start = end - self.config.chunk_overlap
if start >= len(text) - self.config.min_chunk_size:
break
return chunks
def _find_word_boundary(self, text: str, position: int) -> int:
"""Find nearest word boundary"""
# Look forward for space
forward = position
while forward < len(text) and text[forward] not in ' \n\t':
forward += 1
# Look backward for space
backward = position
while backward > 0 and text[backward] not in ' \n\t':
backward -= 1
# Choose closer boundary
if forward - position < position - backward:
return forward
return backward
# Usage
chunker = FixedSizeChunker(ChunkingConfig(chunk_size=500, chunk_overlap=50))
text = "Your long document text here..." * 100
chunks = chunker.chunk(text, metadata={"source": "document.pdf"})
print(f"Created {len(chunks)} chunks")
Semantic Chunking
class SemanticChunker(ChunkingStrategy):
"""Chunk based on semantic boundaries"""
def __init__(
self,
embedding_model=None,
similarity_threshold: float = 0.5,
max_chunk_size: int = 1000
):
self.embedding_model = embedding_model
self.threshold = similarity_threshold
self.max_size = max_chunk_size
def chunk(self, text: str, metadata: Dict = None) -> List[Chunk]:
"""Split text at semantic boundaries"""
# Split into sentences first
sentences = self._split_sentences(text)
if not sentences:
return []
# Get embeddings for each sentence
embeddings = self._get_embeddings(sentences)
# Find semantic break points
break_points = self._find_break_points(embeddings)
# Create chunks from break points
chunks = self._create_chunks(sentences, break_points, metadata)
return chunks
def _split_sentences(self, text: str) -> List[Dict]:
"""Split text into sentences with positions"""
import re
sentences = []
start = 0
for match in re.finditer(r'[^.!?]+[.!?]+', text):
sentences.append({
"text": match.group().strip(),
"start": match.start(),
"end": match.end()
})
return sentences
def _get_embeddings(self, sentences: List[Dict]) -> List[List[float]]:
"""Get embeddings for sentences"""
if self.embedding_model:
texts = [s["text"] for s in sentences]
return self.embedding_model.encode(texts)
# Fallback: simple word-based representation
embeddings = []
for sentence in sentences:
words = set(sentence["text"].lower().split())
# Simple bag-of-words embedding (placeholder)
embeddings.append(list(words)[:100])
return embeddings
def _find_break_points(self, embeddings: List) -> List[int]:
"""Find semantic break points"""
break_points = []
for i in range(1, len(embeddings)):
similarity = self._cosine_similarity(
embeddings[i-1], embeddings[i]
)
if similarity < self.threshold:
break_points.append(i)
return break_points
def _cosine_similarity(self, a, b) -> float:
"""Calculate cosine similarity"""
if isinstance(a, list) and isinstance(a[0], str):
# Bag-of-words fallback
set_a, set_b = set(a), set(b)
intersection = len(set_a & set_b)
union = len(set_a | set_b)
return intersection / union if union > 0 else 0
# Vector similarity
import math
dot = sum(x*y for x, y in zip(a, b))
norm_a = math.sqrt(sum(x*x for x in a))
norm_b = math.sqrt(sum(x*x for x in b))
return dot / (norm_a * norm_b) if norm_a * norm_b > 0 else 0
def _create_chunks(
self,
sentences: List[Dict],
break_points: List[int],
metadata: Dict
) -> List[Chunk]:
"""Create chunks from sentences and break points"""
chunks = []
start_idx = 0
for i, break_point in enumerate(break_points + [len(sentences)]):
chunk_sentences = sentences[start_idx:break_point]
if chunk_sentences:
content = ' '.join(s["text"] for s in chunk_sentences)
# Check max size
if len(content) > self.max_size:
# Split further if too large
sub_chunks = self._split_large_chunk(
chunk_sentences, metadata, len(chunks)
)
chunks.extend(sub_chunks)
else:
chunks.append(Chunk(
id=f"semantic_{uuid.uuid4().hex[:8]}",
content=content,
metadata={
**(metadata or {}),
"chunk_index": len(chunks),
"chunking_strategy": "semantic"
},
start_index=chunk_sentences[0]["start"],
end_index=chunk_sentences[-1]["end"]
))
start_idx = break_point
return chunks
def _split_large_chunk(
self,
sentences: List[Dict],
metadata: Dict,
base_index: int
) -> List[Chunk]:
"""Split a large chunk into smaller pieces"""
chunks = []
current_content = ""
current_start = sentences[0]["start"] if sentences else 0
for sentence in sentences:
if len(current_content) + len(sentence["text"]) > self.max_size:
if current_content:
chunks.append(Chunk(
id=f"semantic_{uuid.uuid4().hex[:8]}",
content=current_content.strip(),
metadata={
**(metadata or {}),
"chunk_index": base_index + len(chunks),
"chunking_strategy": "semantic"
},
start_index=current_start,
end_index=sentence["start"]
))
current_content = sentence["text"]
current_start = sentence["start"]
else:
current_content += " " + sentence["text"]
if current_content:
chunks.append(Chunk(
id=f"semantic_{uuid.uuid4().hex[:8]}",
content=current_content.strip(),
metadata={
**(metadata or {}),
"chunk_index": base_index + len(chunks),
"chunking_strategy": "semantic"
},
start_index=current_start,
end_index=sentences[-1]["end"] if sentences else current_start
))
return chunks
Recursive Character Text Splitter
class RecursiveChunker(ChunkingStrategy):
"""Recursively split text using multiple separators"""
def __init__(
self,
chunk_size: int = 512,
chunk_overlap: int = 50,
separators: List[str] = None
):
self.chunk_size = chunk_size
self.chunk_overlap = chunk_overlap
self.separators = separators or [
"\n\n", # Paragraphs
"\n", # Lines
". ", # Sentences
", ", # Clauses
" ", # Words
"" # Characters
]
def chunk(self, text: str, metadata: Dict = None) -> List[Chunk]:
"""Recursively split text"""
return self._split_recursive(
text,
self.separators,
metadata or {},
0
)
def _split_recursive(
self,
text: str,
separators: List[str],
metadata: Dict,
start_offset: int
) -> List[Chunk]:
"""Recursively split using separators"""
if not separators:
# No more separators, split by size
return self._split_by_size(text, metadata, start_offset)
separator = separators[0]
remaining_separators = separators[1:]
if separator:
splits = text.split(separator)
else:
splits = list(text)
chunks = []
current_chunk = ""
current_start = start_offset
for i, split in enumerate(splits):
# Calculate potential new chunk
potential = current_chunk + (separator if current_chunk else "") + split
if len(potential) <= self.chunk_size:
current_chunk = potential
else:
# Current chunk is full
if current_chunk:
if len(current_chunk) > self.chunk_size:
# Still too large, recurse
sub_chunks = self._split_recursive(
current_chunk,
remaining_separators,
metadata,
current_start
)
chunks.extend(sub_chunks)
else:
chunks.append(Chunk(
id=f"recursive_{uuid.uuid4().hex[:8]}",
content=current_chunk,
metadata={
**metadata,
"chunk_index": len(chunks),
"chunking_strategy": "recursive"
},
start_index=current_start,
end_index=current_start + len(current_chunk)
))
# Start new chunk with overlap
if chunks and self.chunk_overlap > 0:
overlap = chunks[-1].content[-self.chunk_overlap:]
current_chunk = overlap + split
else:
current_chunk = split
current_start = start_offset + text.find(split)
# Handle remaining content
if current_chunk:
if len(current_chunk) > self.chunk_size:
sub_chunks = self._split_recursive(
current_chunk,
remaining_separators,
metadata,
current_start
)
chunks.extend(sub_chunks)
else:
chunks.append(Chunk(
id=f"recursive_{uuid.uuid4().hex[:8]}",
content=current_chunk,
metadata={
**metadata,
"chunk_index": len(chunks),
"chunking_strategy": "recursive"
},
start_index=current_start,
end_index=current_start + len(current_chunk)
))
return chunks
def _split_by_size(
self,
text: str,
metadata: Dict,
start_offset: int
) -> List[Chunk]:
"""Split text by character size"""
chunks = []
start = 0
while start < len(text):
end = min(start + self.chunk_size, len(text))
chunk_text = text[start:end]
chunks.append(Chunk(
id=f"recursive_{uuid.uuid4().hex[:8]}",
content=chunk_text,
metadata={
**metadata,
"chunk_index": len(chunks),
"chunking_strategy": "recursive"
},
start_index=start_offset + start,
end_index=start_offset + end
))
start = end - self.chunk_overlap
return chunks
Document-Aware Chunking
class DocumentAwareChunker(ChunkingStrategy):
"""Chunk based on document structure"""
def __init__(self, chunk_size: int = 512):
self.chunk_size = chunk_size
def chunk(self, text: str, metadata: Dict = None) -> List[Chunk]:
"""Chunk respecting document structure"""
# Detect document type
doc_type = self._detect_document_type(text)
if doc_type == "markdown":
return self._chunk_markdown(text, metadata)
elif doc_type == "code":
return self._chunk_code(text, metadata)
else:
return self._chunk_generic(text, metadata)
def _detect_document_type(self, text: str) -> str:
"""Detect document type"""
# Check for markdown headers
if text.count('#') > 2 and '\n#' in text:
return "markdown"
# Check for code patterns
code_indicators = ['def ', 'class ', 'function ', 'import ', 'const ']
if sum(1 for ind in code_indicators if ind in text) > 2:
return "code"
return "generic"
def _chunk_markdown(self, text: str, metadata: Dict) -> List[Chunk]:
"""Chunk markdown by headers"""
import re
chunks = []
# Split by headers
header_pattern = r'^(#{1,6})\s+(.+)$'
sections = re.split(r'(?=^#{1,6}\s)', text, flags=re.MULTILINE)
for section in sections:
if not section.strip():
continue
# Extract header
header_match = re.match(header_pattern, section, re.MULTILINE)
header = header_match.group(2) if header_match else ""
chunks.append(Chunk(
id=f"md_{uuid.uuid4().hex[:8]}",
content=section.strip(),
metadata={
**(metadata or {}),
"section_header": header,
"chunking_strategy": "markdown"
},
start_index=text.find(section),
end_index=text.find(section) + len(section)
))
return chunks
def _chunk_code(self, text: str, metadata: Dict) -> List[Chunk]:
"""Chunk code by functions/classes"""
import re
chunks = []
# Split by function/class definitions
patterns = [
r'(^(?:def|async def)\s+\w+.*?(?=^(?:def|async def|class)|\Z))',
r'(^class\s+\w+.*?(?=^(?:def|async def|class)|\Z))',
]
remaining = text
for pattern in patterns:
matches = list(re.finditer(pattern, remaining, re.MULTILINE | re.DOTALL))
for match in matches:
chunks.append(Chunk(
id=f"code_{uuid.uuid4().hex[:8]}",
content=match.group().strip(),
metadata={
**(metadata or {}),
"chunking_strategy": "code"
},
start_index=match.start(),
end_index=match.end()
))
return chunks if chunks else self._chunk_generic(text, metadata)
def _chunk_generic(self, text: str, metadata: Dict) -> List[Chunk]:
"""Fallback to recursive chunking"""
recursive = RecursiveChunker(chunk_size=self.chunk_size)
return recursive.chunk(text, metadata)
Chunking Evaluation
class ChunkingEvaluator:
"""Evaluate chunking quality"""
def evaluate(
self,
chunks: List[Chunk],
original_text: str
) -> Dict:
"""Evaluate chunk quality"""
metrics = {}
# Coverage
metrics["coverage"] = self._calculate_coverage(chunks, original_text)
# Size distribution
metrics["size_stats"] = self._calculate_size_stats(chunks)
# Overlap analysis
metrics["overlap_stats"] = self._analyze_overlap(chunks)
# Completeness
metrics["completeness"] = self._check_completeness(chunks)
return metrics
def _calculate_coverage(
self,
chunks: List[Chunk],
original: str
) -> float:
"""Calculate text coverage"""
covered_chars = set()
for chunk in chunks:
for i in range(chunk.start_index, chunk.end_index):
covered_chars.add(i)
return len(covered_chars) / len(original) if original else 0
def _calculate_size_stats(self, chunks: List[Chunk]) -> Dict:
"""Calculate size statistics"""
sizes = [len(c.content) for c in chunks]
if not sizes:
return {"count": 0}
return {
"count": len(sizes),
"min": min(sizes),
"max": max(sizes),
"avg": sum(sizes) / len(sizes),
"std": self._std(sizes)
}
def _std(self, values: List[float]) -> float:
"""Calculate standard deviation"""
if len(values) < 2:
return 0
mean = sum(values) / len(values)
variance = sum((x - mean) ** 2 for x in values) / len(values)
return variance ** 0.5
def _analyze_overlap(self, chunks: List[Chunk]) -> Dict:
"""Analyze chunk overlap"""
overlaps = []
for i in range(len(chunks) - 1):
current = chunks[i]
next_chunk = chunks[i + 1]
if current.end_index > next_chunk.start_index:
overlap = current.end_index - next_chunk.start_index
overlaps.append(overlap)
return {
"chunks_with_overlap": len(overlaps),
"avg_overlap": sum(overlaps) / len(overlaps) if overlaps else 0,
"max_overlap": max(overlaps) if overlaps else 0
}
def _check_completeness(self, chunks: List[Chunk]) -> Dict:
"""Check for incomplete sentences"""
incomplete = 0
for chunk in chunks:
content = chunk.content.strip()
# Check if ends with sentence terminator
if content and content[-1] not in '.!?"\')':
incomplete += 1
return {
"total_chunks": len(chunks),
"incomplete_endings": incomplete,
"completeness_rate": 1 - (incomplete / len(chunks)) if chunks else 1
}
# Usage
evaluator = ChunkingEvaluator()
text = "Sample document text..." * 50
chunker = RecursiveChunker(chunk_size=200, chunk_overlap=20)
chunks = chunker.chunk(text)
metrics = evaluator.evaluate(chunks, text)
print(f"Chunks: {metrics['size_stats']['count']}")
print(f"Coverage: {metrics['coverage']:.1%}")
print(f"Completeness: {metrics['completeness']['completeness_rate']:.1%}")
Conclusion
Effective chunking is essential for RAG performance. Fixed-size chunking provides simplicity, semantic chunking preserves meaning, recursive chunking respects text hierarchy, and document-aware chunking leverages structure. Evaluate chunking quality using coverage, size distribution, and completeness metrics to optimize for your specific use case.