Why Chunking Matters
How you split documents in a RAG pipeline directly affects retrieval quality. Chunks that are too large introduce noise; chunks that are too small lose context.
Contextifier's Chunking Pipeline
flowchart TD
A[원본 문서] --> B[전처리기]
B --> C[텍스트 + 테이블 + 이미지]
C --> D{PageTagProcessor}
D --> E[페이지/슬라이드/시트 태그 부착]
E --> F{청킹 엔진}
F --> G[텍스트 청크]
F --> H[테이블 청크]
G --> I[ChunkResult]
H --> I
I --> J[메타데이터 포함 최종 결과]
The ChunkResult Class
ChunkResult is the class that holds document-processing output in a standardized structure:
class ChunkResult:
chunks: List[Chunk]
metadata: DocumentMetadata
images: List[ImageData]
def write_to_file(self, path: str):
# surrogate character 안전 처리
...
def to_dict(self) -> dict:
return {
"chunks": [c.to_dict() for c in self.chunks],
"metadata": self.metadata.to_dict(),
"image_count": len(self.images)
}
PageTagProcessor
Introduced in v0.1.5, this module attaches tags at the page, slide, or sheet level, making it possible to trace each chunk back to its location in the source document.
Table Splitting Strategy
Very large tables don't fit cleanly into a single chunk. The force_chunking flag enables row-level splitting for these cases.
# v0.2.2에서 개선된 테이블 분할
def split_table(table: str, max_tokens: int, force_chunking: bool):
if force_chunking:
# 헤더를 보존하면서 행 단위 분할
header = table.split('\n')[0:2]
rows = table.split('\n')[2:]
chunks = []
current = header.copy()
for row in rows:
if token_count('\n'.join(current + [row])) > max_tokens:
chunks.append('\n'.join(current))
current = header.copy()
current.append(row)
if current != header:
chunks.append('\n'.join(current))
return chunks
return [table]
Chunking Improvements by Version
| Version | Change |
|---|---|
| v0.1.0 | Basic fixed-size chunking |
| v0.1.5 | PageTag-based chunking |
| v0.2.1 | Documentation and API cleanup |
| v0.2.2 | force_chunking flag |
There were 12 commits related to chunking in total, with significant effort going into removing deprecated functions and refactoring.