RAG Chunking Strategies — How We Lost $4k/Month on Token Waste and Fixed It with One Config Change

import tiktoken
from langchain.text_splitter import RecursiveCharacterTextSplitter

def inspect_chunking(doc_path: str, model_name: str = "text-embedding-3-small"):
    """Show what's really happening during chunking."""
    enc = tiktoken.encoding_for_model(model_name)
    
    with open(doc_path) as f:
        text = f.read()
    
    # Tokenize once to see actual token count
    tokens = enc.encode(text)
    print(f"Document: {len(tokens)} tokens, {len(text)} chars")
    
    # LangChain's splitter hides the fallback logic
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=512,  # target, not exact
        chunk_overlap=64,  # adds ~12.5% more embeddings
        separators=["\n\n", "\n", ".", " "]  # order matters!
    )
    chunks = splitter.split_text(text)
    
    for i, chunk in enumerate(chunks[:5]):
        chunk_tokens = len(enc.encode(chunk))
        print(f"Chunk {i}: {chunk_tokens} tokens, ends with: ...{chunk[-50:]}")
        # Notice: chunk_tokens is rarely exactly 512
        # The splitter stops at the last separator before 512
        # If no separator found, it cuts at 512 (mid-token!)
    
    # The overlap creates duplicate embeddings
    total_embeddings = len(chunks)
    estimated_cost = total_embeddings * 0.0001  # $0.0001 per embedding
    print(f"Total embeddings: {total_embeddings}, estimated cost: ${estimated_cost:.2f}")

if __name__ == "__main__":
    inspect_chunking("legal_doc.txt")

import tiktoken
from langchain.text_splitter import (
    RecursiveCharacterTextSplitter,
    TokenTextSplitter
)
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
from sklearn.cluster import KMeans
import numpy as np

# Strategy 1: Recursive Character (production default)
def recursive_chunk(text: str, chunk_size: int = 1024, overlap: int = 128):
    splitter = RecursiveCharacterTextSplitter(
        separators=["\n\n", "\n", ".", " "],
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        length_function=len  # character-based; use tiktoken for token-based
    )
    return splitter.split_text(text)

# Strategy 2: Token-based (recommended for OpenAI models)
def token_chunk(text: str, model: str = "text-embedding-3-small"):
    enc = tiktoken.encoding_for_model(model)
    splitter = TokenTextSplitter(
        encoding_name=model,
        chunk_size=512,  # in tokens now, not characters
        chunk_overlap=64
    )
    return splitter.split_text(text)

# Strategy 3: Semantic chunking (adds latency but better precision)
def semantic_chunk(text: str):
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    splitter = SemanticChunker(
        embeddings=embeddings,
        breakpoint_threshold_type="percentile",  # or 'standard_deviation'
        breakpoint_threshold_amount=0.3  # lower = more chunks
    )
    return splitter.split_text(text)

# Strategy 4: Simple cluster-based (experimental)
def cluster_chunk(text: str, n_clusters: int = 5):
    sentences = text.split(". ")
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    embeds = embeddings.embed_documents(sentences)
    kmeans = KMeans(n_clusters=n_clusters, random_state=42)
    labels = kmeans.fit_predict(embeds)
    
    chunks = [""] * n_clusters
    for sent, label in zip(sentences, labels):
        chunks[label] += sent + ". "
    return [c for c in chunks if c]

# Production note: always log chunk sizes
if __name__ == "__main__":
    sample = "Your long document here..."
    for name, chunks in [
        ("recursive", recursive_chunk(sample)),
        ("token", token_chunk(sample)),
        ("semantic", semantic_chunk(sample)),
        ("cluster", cluster_chunk(sample))
    ]:
        print(f"{name}: {len(chunks)} chunks, avg size: {np.mean([len(c) for c in chunks]):.0f} chars")

from langchain.text_splitter import (
    MarkdownHeaderTextSplitter,
    RecursiveCharacterTextSplitter
)

# For markdown documents, preserve header hierarchy
def markdown_chunk(markdown_text: str):
    headers_to_split_on = [
        ("#", "Header 1"),
        ("##", "Header 2"),
        ("###", "Header 3"),
    ]
    splitter = MarkdownHeaderTextSplitter(
        headers_to_split_on=headers_to_split_on
    )
    # Returns list of Document objects with metadata
    return splitter.split_text(markdown_text)

# For code, use a code-aware splitter
# (LangChain doesn't have one built-in, so we hack it)
def code_chunk(code_text: str, language: str = "python"):
    # Custom separators for Python
    if language == "python":
        separators = [
            "\nclass ",
            "\ndef ",
            "\n    def ",  # nested methods
            "\n\n",
            "\n",
            " "
        ]
    else:
        separators = ["\n\n", "\n", " "]
    
    splitter = RecursiveCharacterTextSplitter(
        separators=separators,
        chunk_size=1024,
        chunk_overlap=128
    )
    return splitter.split_text(code_text)

# Production note: always validate chunk boundaries
# Check if chunks end with a complete statement
import ast
def validate_chunk(chunk: str):
    try:
        ast.parse(chunk)
        return True
    except SyntaxError:
        return False

if __name__ == "__main__":
    code = """
def foo():
    return 1

def bar():
    return 2
"""
    chunks = code_chunk(code)
    for c in chunks:
        print(f"Valid Python: {validate_chunk(c)} -> {c[:50]}...")

import hashlib
from concurrent.futures import ProcessPoolExecutor
from langchain.text_splitter import RecursiveCharacterTextSplitter
import chromadb
from chromadb.utils import embedding_functions

def chunk_document(text: str, chunk_size: int = 1024):
    """Return list of (chunk_text, chunk_hash) tuples."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=128
    )
    chunks = splitter.split_text(text)
    return [(c, hashlib.sha256(c.encode()).hexdigest()) for c in chunks]

def process_batch(docs: list):
    """Process a batch of documents in parallel."""
    with ProcessPoolExecutor(max_workers=8) as executor:
        results = list(executor.map(chunk_document, docs))
    return results

class IncrementalIndexer:
    def __init__(self, collection_name: str):
        self.client = chromadb.PersistentClient(path="./chroma_db")
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=embedding_functions.OpenAIEmbeddingFunction(
                model_name="text-embedding-3-small"
            )
        )
    
    def index_document(self, doc_id: str, text: str):
        # Check if document has changed
        doc_hash = hashlib.sha256(text.encode()).hexdigest()
        existing = self.collection.get(ids=[doc_id])
        if existing and existing['metadatas'][0].get('hash') == doc_hash:
            return  # Skip unchanged documents
        
        # Chunk and index
        chunks = chunk_document(text)
        ids = [f"{doc_id}_{i}" for i in range(len(chunks))]
        metadatas = [{"doc_id": doc_id, "hash": doc_hash} for _ in chunks]
        texts = [c[0] for c in chunks]
        
        # Delete old chunks for this document
        self.collection.delete(where={"doc_id": doc_id})
        # Add new chunks
        self.collection.add(ids=ids, documents=texts, metadatas=metadatas)

if __name__ == "__main__":
    indexer = IncrementalIndexer("docs_v2")
    indexer.index_document("doc_001", "Your document text...")
    print("Indexed successfully")

import logging
from dataclasses import dataclass, asdict

@dataclass
class ChunkMetric:
    doc_id: str
    chunk_index: int
    token_count: int
    char_count: int
    similarity_score: float  # to query
    position_ratio: float  # 0.0 = start, 1.0 = end of doc

def log_chunk_metrics(metrics: list[ChunkMetric]):
    logger = logging.getLogger("rag_monitor")
    for m in metrics:
        logger.info(f"ChunkMetric: {asdict(m)}")
        # Alert if token count is far from expected
        if m.token_count > 800:  # expecting 512
            logger.warning(f"Chunk {m.doc_id}_{m.chunk_index} has {m.token_count} tokens")
        # Alert if similarity is too low
        if m.similarity_score < 0.5:
            logger.warning(f"Low similarity chunk: {m.similarity_score:.2f}")

# In production, call this after retrieval
# metrics = [ChunkMetric(...) for chunk in retrieved_chunks]
# log_chunk_metrics(metrics)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate

# Query rewriting as a complement to chunking
rewrite_prompt = ChatPromptTemplate.from_messages([
    ("system", "Rewrite the user's query to be more specific for document retrieval. Focus on key entities and terms."),
    ("human", "{query}")
])

def rewrite_query(query: str) -> str:
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    chain = rewrite_prompt | llm
    return chain.invoke({"query": query}).content

# Production A/B test: compare chunking alone vs chunking + rewriting
def compare_strategies(queries: list[str], documents: list[str]):
    from langchain.vectorstores import Chroma
    from langchain.embeddings import OpenAIEmbeddings
    
    emb = OpenAIEmbeddings()
    db = Chroma.from_texts(documents, emb)
    
    for q in queries:
        # Baseline: chunking alone
        results_baseline = db.similarity_search(q, k=5)
        
        # With query rewriting
        rewritten = rewrite_query(q)
        results_rewritten = db.similarity_search(rewritten, k=5)
        
        print(f"Query: {q}")
        print(f"Rewritten: {rewritten}")
        print(f"Baseline relevance: {len(results_baseline)} chunks")
        print(f"Rewritten relevance: {len(results_rewritten)} chunks")
        # In production, compute precision@k for both

if __name__ == "__main__":
    compare_strategies(
        ["What's the refund policy?", "Tell me about pricing"],
        ["Our refund policy allows returns within 30 days...", "Pricing starts at $10/month..."]
    )

import json
import logging
from datetime import datetime

class RAGMonitor:
    def __init__(self):
        self.logger = logging.getLogger("rag_monitor")
        handler = logging.FileHandler("rag_monitor.log")
        handler.setFormatter(logging.Formatter("%(asctime)s - %(message)s"))
        self.logger.addHandler(handler)
    
    def log_query(self, query: str, chunks: list, scores: list, answer: str):
        """Log retrieval and generation metrics per query."""
        record = {
            "timestamp": datetime.utcnow().isoformat(),
            "query": query,
            "num_chunks": len(chunks),
            "avg_score": sum(scores) / len(scores) if scores else 0,
            "min_score": min(scores) if scores else 0,
            "max_score": max(scores) if scores else 0,
            "answer_length": len(answer.split()),
        }
        self.logger.info(json.dumps(record))
        
        # Alert conditions
        if record["avg_score"] < 0.5:
            self.logger.warning(f"Low retrieval quality: avg_score={record['avg_score']:.2f}")
        if record["num_chunks"] > 10:
            self.logger.warning(f"Too many chunks retrieved: {record['num_chunks']}")
    
    def get_health_report(self) -> dict:
        """Aggregate metrics for the last hour."""
        # In production, read from log file or database
        return {
            "avg_retrieval_score": 0.72,
            "p95_chunks_per_query": 8,
            "avg_answer_length": 150,
            "chunking_strategy": "recursive_char_1024_128",
            "alert_count_last_hour": 2
        }

if __name__ == "__main__":
    monitor = RAGMonitor()
    monitor.log_query(
        query="What is the return policy?",
        chunks=["Our return policy...", "You can return..."],
        scores=[0.85, 0.72],
        answer="You can return items within 30 days."
    )
    print(monitor.get_health_report())

// io.thecodeforge
import tiktoken

def estimate_chunking_cost(docs, chunk_size, overlap, embedding_cost_per_1k_tokens, vector_store_cost_monthly):
    tokenizer = tiktoken.get_encoding("cl100k_base")
    total_tokens = sum(len(tokenizer.encode(d)) for d in docs)
    num_chunks = (total_tokens - overlap * (total_tokens // chunk_size)) / (chunk_size - overlap)
    embedding_tokens = num_chunks * chunk_size
    embedding_cost = (embedding_tokens / 1000) * embedding_cost_per_1k_tokens
    print(f"Chunks: {int(num_chunks)}, Embedding Cost: ${embedding_cost:.2f}, Storage: ${vector_store_cost_monthly:.2f}/mo")
    return embedding_cost + vector_store_cost_monthly

# Real production numbers
docs = ["long document..."] * 1000
estimate_chunking_cost(docs, chunk_size=512, overlap=32, embedding_cost_per_1k_tokens=0.0001, vector_store_cost_monthly=200)

// io.thecodeforge
from langchain.text_splitter import RecursiveCharacterTextSplitter

def semantic_chunk(document_text: str, chunk_size: int = 1000, chunk_overlap: int = 200):
    separators = ["\n\n", "\n", ". ", " "]  # Priority order: paragraph > newline > sentence > word
    splitter = RecursiveCharacterTextSplitter(
        separators=separators,
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
    )
    return splitter.split_text(document_text)

doc = "# Report\n\nParagraph one...\n\nParagraph two with more details here."
chunks = semantic_chunk(doc)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i}: {chunk[:50]}...")

// io.thecodeforge
from bs4 import BeautifulSoup

def graph_chunk(html_doc: str, max_chunk_size: int = 1500) -> list:
    soup = BeautifulSoup(html_doc, "html.parser")
    chunks = []
    current_chunk = ""
    
    for tag in soup.recursiveChildGenerator():
        if hasattr(tag, 'name'):
            text = tag.get_text(strip=True)
            if len(current_chunk) + len(text) > max_chunk_size:
                chunks.append(current_chunk)
                current_chunk = text
            else:
                current_chunk += " " + text
    
    if current_chunk:
        chunks.append(current_chunk)
    return chunks

html = "<div><h2>Section 1</h2><p>Details...</p></div><div><h2>Section 2</h2><ul><li>Item</li></ul></div>"
print(graph_chunk(html))