Chunking Strategies

sample_doc = """
# Introduction to Machine Learning

Machine learning is a subset of artificial intelligence that enables
systems to learn from data. It has transformed industries from healthcare
to finance.

## Supervised Learning

In supervised learning, models learn from labeled examples. Common
algorithms include linear regression, decision trees, and neural networks.
The key challenge is overfitting — when a model memorizes training data
rather than learning general patterns.

## Unsupervised Learning

Unsupervised learning finds hidden patterns in unlabeled data. Clustering
algorithms like K-means group similar data points. Dimensionality reduction
techniques like PCA compress high-dimensional data.

## Deep Learning

Deep learning uses neural networks with many layers. Transformers
revolutionized NLP with attention mechanisms. GPT and BERT are
transformer-based architectures that achieve state-of-the-art results.
"""

from langchain.text_splitter import CharacterTextSplitter

# Fixed-size by character count
splitter = CharacterTextSplitter(
    separator="",          # Split anywhere
    chunk_size=300,        # Characters per chunk
    chunk_overlap=50,      # Overlap between chunks
    length_function=len
)

chunks = splitter.split_text(sample_doc)
print(f"Fixed chunks: {len(chunks)}")
for i, chunk in enumerate(chunks):
    print(f"  Chunk {i}: {len(chunk)} chars — '{chunk[:60]}...'")

from langchain.text_splitter import RecursiveCharacterTextSplitter

# Recursive splitting with hierarchy of separators
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=[
        "\n\n",   # First try: split on double newlines (paragraphs)
        "\n",     # Then: single newlines
        ". ",     # Then: sentences
        ", ",     # Then: clauses
        " ",      # Then: words
        ""        # Last resort: characters
    ],
    length_function=len
)

chunks = splitter.split_text(sample_doc)
print(f"Recursive chunks: {len(chunks)}")
for i, chunk in enumerate(chunks):
    print(f"  Chunk {i} ({len(chunk)} chars): '{chunk[:80]}...'")

# --- Token-based splitting (more accurate for LLMs) ---
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Use tiktoken for accurate token counting
splitter_tokens = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    encoding_name="cl100k_base",  # GPT-4 tokenizer
    chunk_size=256,               # Tokens, not characters
    chunk_overlap=30
)

chunks_tok = splitter_tokens.split_text(sample_doc)
print(f"\nToken-based chunks: {len(chunks_tok)}")

import numpy as np
from openai import OpenAI
import re

client = OpenAI()

def semantic_chunk(text: str, threshold: float = 0.75, min_chunk: int = 2):
    """Split text into semantically coherent chunks.
    
    Algorithm:
    1. Split into sentences
    2. Embed each sentence
    3. Compute cosine similarity between consecutive sentences
    4. Split where similarity drops below threshold
    """
    # Split into sentences
    sentences = [s.strip() for s in re.split(r'(?<=[.!?])\s+', text) if s.strip()]
    
    if len(sentences) <= min_chunk:
        return [text]
    
    # Embed all sentences
    response = client.embeddings.create(
        input=sentences, model="text-embedding-3-small"
    )
    embeddings = [d.embedding for d in response.data]
    
    # Calculate consecutive similarities
    similarities = []
    for i in range(len(embeddings) - 1):
        a, b = np.array(embeddings[i]), np.array(embeddings[i + 1])
        sim = np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
        similarities.append(sim)
    
    # Find split points (where similarity drops)
    chunks = []
    current_chunk = [sentences[0]]
    
    for i, sim in enumerate(similarities):
        if sim < threshold and len(current_chunk) >= min_chunk:
            chunks.append(" ".join(current_chunk))
            current_chunk = [sentences[i + 1]]
        else:
            current_chunk.append(sentences[i + 1])
    
    if current_chunk:
        chunks.append(" ".join(current_chunk))
    
    return chunks

# Usage
chunks = semantic_chunk(sample_doc, threshold=0.72)
print(f"Semantic chunks: {len(chunks)}")
for i, chunk in enumerate(chunks):
    print(f"\n--- Chunk {i} ({len(chunk)} chars) ---")
    print(chunk[:150] + "...")

from langchain.text_splitter import (
    MarkdownHeaderTextSplitter,
    RecursiveCharacterTextSplitter,
    Language
)

# --- Markdown-aware splitting ---
headers_to_split_on = [
    ("#", "h1"),
    ("##", "h2"),
    ("###", "h3"),
]

md_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=headers_to_split_on
)

md_chunks = md_splitter.split_text(sample_doc)
for chunk in md_chunks:
    print(f"Headers: {chunk.metadata}")
    print(f"Content: {chunk.page_content[:80]}...\n")

# Each chunk knows its heading hierarchy!
# Headers: {'h1': 'Introduction to Machine Learning', 'h2': 'Supervised Learning'}
# Content: In supervised learning, models learn from labeled examples...

# --- Two-stage: structural split then size limit ---
# First: split by headers
md_chunks = md_splitter.split_text(sample_doc)

# Second: if any chunk is too long, split it further
char_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500, chunk_overlap=50
)

final_chunks = []
for chunk in md_chunks:
    if len(chunk.page_content) > 500:
        sub_chunks = char_splitter.split_text(chunk.page_content)
        for sc in sub_chunks:
            final_chunks.append({"content": sc, "metadata": chunk.metadata})
    else:
        final_chunks.append({
            "content": chunk.page_content, 
            "metadata": chunk.metadata
        })

# --- Code-aware splitting ---
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=500,
    chunk_overlap=50
)

code = '''
class DataProcessor:
    def __init__(self, config):
        self.config = config
    
    def process(self, data):
        """Process raw data into features."""
        cleaned = self.clean(data)
        features = self.extract_features(cleaned)
        return features
    
    def clean(self, data):
        """Remove nulls and normalize."""
        return [d for d in data if d is not None]
'''

code_chunks = python_splitter.split_text(code)
# Splits at class/function boundaries, not mid-function!