Lesson 16.13: Chunking Strategies for RAG

Table: Product Inventory & Price List (Hardware Dept)
| Product Code | Item Name | Unit Price (USD) | Stock |
| :--- | :--- | :--- | :--- |
| P-101 | Mechanical Keyboard | $120.00 | 25 |
| P-102 | Wireless Mouse | $45.00 | 40 |
| P-103 | 24-inch Monitor | $250.00 | 8 |

from langchain_text_splitters import RecursiveCharacterTextSplitter
from typing import List, Dict, Any

def get_bilingual_text_splitter(
    chunk_size: int = 600,
    chunk_overlap: int = 90
) -> RecursiveCharacterTextSplitter:
    """
    Optimized recursive character splitter for bilingual corpora.
    Includes Bengali sentence terminal (। ) and standard Western punctuation.
    """
    return RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=[
            "\n\n",       # 1. Paragraph boundary
            "\n",         # 2. Line break
            "। ",         # 3. Bengali sentence terminal
            ". ",         # 4. English sentence terminal
            "? ",         # 5. Question mark
            "! ",         # 6. Exclamation mark
            "; ",         # 7. Semicolon
            " ",          # 8. Word boundary
            ""            # 9. Character fallback
        ],
        add_start_index=True,
        strip_whitespace=True
    )

def get_token_aware_splitter(
    chunk_size_tokens: int = 400,
    chunk_overlap_tokens: int = 60,
    model_encoding: str = "cl100k_base"
) -> RecursiveCharacterTextSplitter:
    """
    Constructs a token-aware splitter utilizing the OpenAI BPE tokenizer.
    """
    return RecursiveCharacterTextSplitter.from_tiktoken_encoder(
        encoding_name=model_encoding,
        chunk_size=chunk_size_tokens,
        chunk_overlap=chunk_overlap_tokens,
        separators=["\n\n", "\n", "। ", ". ", " ", ""],
        add_start_index=True
    )

import hashlib

def generate_chunk_id(
    document_id: str,
    unit_index: int,
    chunk_index: int,
    chunking_version: str,
    text: str
) -> str:
    """
    Emits a deterministic SHA-256 chunk identifier.
    Identical document content and version produce an identical hash,
    enabling idempotent re-indexing.
    """
    seed_str = f"{document_id}:{unit_index}:{chunk_index}:{chunking_version}:{text.strip()}"
    digest = hashlib.sha256(seed_str.encode("utf-8")).hexdigest()
    return f"chk_{digest[:16]}"

from typing import List, Dict, Any
from app.chunking import get_bilingual_text_splitter, generate_chunk_id

class DocumentChunkingService:
    def __init__(self, chunk_size: int = 600, chunk_overlap: int = 90, version: str = "v1"):
        self.splitter = get_bilingual_text_splitter(chunk_size, chunk_overlap)
        self.version = version

    def process_document(self, document_record: Dict[str, Any]) -> List[Dict[str, Any]]:
        """
        Consumes ingested page/block units and generates metadata-rich chunks.
        """
        document_id = document_record["document_id"]
        tenant_id = document_record["tenant_id"]
        source_name = document_record["metadata"]["original_filename"]
        units = document_record["units"]
        
        all_chunks = []
        global_chunk_counter = 0

        for unit in units:
            unit_type = unit["unit_type"]
            unit_index = unit["unit_index"]

            # Preserve structured tables without row bisecting
            if unit_type == "table":
                rows = unit.get("rows", [])
                if not rows:
                    continue
                table_md = "\n".join([" | ".join(row) for row in rows])
                global_chunk_counter += 1
                chunk_id = generate_chunk_id(
                    document_id, unit_index, global_chunk_counter, self.version, table_md
                )
                all_chunks.append({
                    "chunk_id": chunk_id,
                    "document_id": document_id,
                    "tenant_id": tenant_id,
                    "chunk_index": global_chunk_counter,
                    "chunking_version": self.version,
                    "text": table_md,
                    "metadata": {
                        "source": source_name,
                        "unit_type": "table",
                        "unit_index": unit_index,
                        "is_table": True
                    }
                })
                continue

            # Recursive splitting for prose units
            raw_text = unit.get("text", "").strip()
            if not raw_text:
                continue

            split_docs = self.splitter.create_documents(texts=[raw_text])
            for s_doc in split_docs:
                text_content = s_doc.page_content.strip()
                if not text_content:
                    continue
                global_chunk_counter += 1
                chunk_id = generate_chunk_id(
                    document_id, unit_index, global_chunk_counter, self.version, text_content
                )
                all_chunks.append({
                    "chunk_id": chunk_id,
                    "document_id": document_id,
                    "tenant_id": tenant_id,
                    "chunk_index": global_chunk_counter,
                    "chunking_version": self.version,
                    "text": text_content,
                    "metadata": {
                        "source": source_name,
                        "unit_type": unit_type,
                        "unit_index": unit_index,
                        "start_index": s_doc.metadata.get("start_index", 0),
                        "is_table": False
                    }
                })

        return all_chunks

import pytest
from app.services.chunking_service import DocumentChunkingService

@pytest.fixture
def chunk_service():
    return DocumentChunkingService(chunk_size=150, chunk_overlap=30, version="v1.0")

def test_bilingual_sentence_splitting(chunk_service):
    sample_doc = {
        "document_id": "doc_test_01",
        "tenant_id": "tenant_byteion",
        "metadata": {"original_filename": "policy.pdf"},
        "units": [
            {
                "unit_type": "page",
                "unit_index": 1,
                "text": "Annual leave quota is 20 days. Staff must follow guidelines. Unused days will expire."
            }
        ]
    }
    chunks = chunk_service.process_document(sample_doc)
    assert len(chunks) >= 1
    assert "20 days." in chunks[0]["text"]
    assert chunks[0]["metadata"]["unit_index"] == 1

def test_table_preservation_without_row_fragmentation(chunk_service):
    sample_doc = {
        "document_id": "doc_test_02",
        "tenant_id": "tenant_byteion",
        "metadata": {"original_filename": "catalog.docx"},
        "units": [
            {
                "unit_type": "table",
                "unit_index": 2,
                "rows": [
                    ["Item", "Price"],
                    ["Keyboard", "$120.00"],
                    ["Mouse", "$45.00"]
                ]
            }
        ]
    }
    chunks = chunk_service.process_document(sample_doc)
    assert len(chunks) == 1
    assert chunks[0]["metadata"]["is_table"] is True
    assert "Keyboard | $120.00" in chunks[0]["text"]

def test_deterministic_chunk_ids(chunk_service):
    doc = {
        "document_id": "doc_fixed",
        "tenant_id": "t1",
        "metadata": {"original_filename": "test.txt"},
        "units": [{"unit_type": "page", "unit_index": 1, "text": "Deterministic test string."}]
    }
    res1 = chunk_service.process_document(doc)
    res2 = chunk_service.process_document(doc)
    assert res1[0]["chunk_id"] == res2[0]["chunk_id"]