Chunking Strategies — What Nobody Tells You

Too small (50 chars):   "The refund policy for enterprise"
                        → Embedding is vague, matches too many queries

Too large (5000 chars): "Our refund policy... [followed by shipping info, 
                         return windows, tax implications, contact details]"
                        → Embedding is diluted, retrieval misses specifics

Right size (400 chars): "Enterprise customers can request a full refund within 
                         30 days of purchase. After 30 days, a prorated refund 
                         is available for annual plans. Contact your account 
                         manager to initiate the process."
                        → Embedding captures one coherent idea

def fixed_size_chunk(text: str, chunk_size: int = 500, overlap: int = 50) -> list[str]:
    """Split text into fixed-size chunks with overlap."""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunk = text[start:end]
        chunks.append(chunk)
        start = end - overlap  # Step back by overlap amount
    return chunks


# Example
text = "A" * 1200  # 1200 characters of text
chunks = fixed_size_chunk(text, chunk_size=500, overlap=50)
print(f"Number of chunks: {len(chunks)}")  # 3
print(f"Chunk sizes: {[len(c) for c in chunks]}")  # [500, 500, 250]

def recursive_split(
    text: str,
    chunk_size: int = 500,
    chunk_overlap: int = 50,
    separators: list[str] | None = None,
) -> list[str]:
    """Recursively split text using a hierarchy of separators."""
    if separators is None:
        separators = ["\n\n", "\n", ". ", " ", ""]

    chunks = []
    separator = separators[0]
    remaining_separators = separators[1:]

    # Split on the current separator
    splits = text.split(separator) if separator else list(text)

    current_chunk = ""
    for split in splits:
        piece = split if not separator else split + separator
        if len(current_chunk) + len(piece) <= chunk_size:
            current_chunk += piece
        else:
            if current_chunk:
                chunks.append(current_chunk.strip())
            # If this single piece is too large, recurse with finer separator
            if len(piece) > chunk_size and remaining_separators:
                sub_chunks = recursive_split(
                    piece, chunk_size, chunk_overlap, remaining_separators
                )
                chunks.extend(sub_chunks)
                current_chunk = ""
            else:
                current_chunk = piece

    if current_chunk.strip():
        chunks.append(current_chunk.strip())

    # Apply overlap
    if chunk_overlap > 0 and len(chunks) > 1:
        chunks = _apply_overlap(chunks, chunk_overlap)

    return chunks


def _apply_overlap(chunks: list[str], overlap: int) -> list[str]:
    """Add overlap between consecutive chunks."""
    result = [chunks[0]]
    for i in range(1, len(chunks)):
        prev_tail = chunks[i - 1][-overlap:]
        result.append(prev_tail + " " + chunks[i])
    return result

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", ", ", " ", ""],
    length_function=len,
)

text = """
Introduction to Machine Learning

Machine learning is a subset of artificial intelligence that focuses 
on building systems that learn from data. Unlike traditional programming, 
where you write explicit rules, ML systems discover patterns automatically.

There are three main categories of machine learning:

Supervised Learning: The model learns from labeled examples. You provide 
input-output pairs and the model learns the mapping function. Common 
algorithms include linear regression, decision trees, and neural networks.

Unsupervised Learning: The model finds structure in unlabeled data. 
Clustering and dimensionality reduction are typical tasks. K-means and 
PCA are widely used algorithms.

Reinforcement Learning: The model learns by interacting with an 
environment and receiving rewards or penalties. This approach powers 
game-playing AIs and robotic control systems.
"""

chunks = splitter.split_text(text)
for i, chunk in enumerate(chunks):
    print(f"--- Chunk {i} ({len(chunk)} chars) ---")
    print(chunk[:100] + "...")
    print()

import spacy

nlp = spacy.load("en_core_web_sm")


def sentence_chunk(
    text: str,
    max_sentences: int = 5,
    overlap_sentences: int = 1,
) -> list[str]:
    """Chunk text by grouping sentences together."""
    doc = nlp(text)
    sentences = [sent.text.strip() for sent in doc.sents]

    chunks = []
    for i in range(0, len(sentences), max_sentences - overlap_sentences):
        group = sentences[i : i + max_sentences]
        chunks.append(" ".join(group))

    return chunks


text = """
Dr. Smith published the results on Jan. 15. The study involved 3.5 million 
data points collected across the U.S.A. Results showed a 23% improvement in 
accuracy. The p-value was less than 0.001, which is statistically significant. 
Follow-up studies are planned for Q3 2026. The team expects to expand the 
dataset to 10 million records. Funding has been secured through 2028.
"""

chunks = sentence_chunk(text, max_sentences=3, overlap_sentences=1)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i}: {chunk}\n")

import nltk
nltk.download("punkt_tab", quiet=True)
from nltk.tokenize import sent_tokenize

sentences = sent_tokenize(text)

import numpy as np
from openai import OpenAI

client = OpenAI()


def get_embeddings(texts: list[str]) -> list[list[float]]:
    """Get embeddings for a batch of texts."""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
    )
    return [item.embedding for item in response.data]


def cosine_similarity(a: list[float], b: list[float]) -> float:
    """Compute cosine similarity between two vectors."""
    a, b = np.array(a), np.array(b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))


def semantic_chunk(
    text: str,
    similarity_threshold: float = 0.75,
    min_chunk_size: int = 100,
) -> list[str]:
    """Split text into semantically coherent chunks using embeddings."""
    # Step 1: Split into sentences
    import nltk
    nltk.download("punkt_tab", quiet=True)
    from nltk.tokenize import sent_tokenize

    sentences = sent_tokenize(text)
    if len(sentences) <= 1:
        return [text]

    # Step 2: Embed each sentence
    embeddings = get_embeddings(sentences)

    # Step 3: Find breakpoints where similarity drops
    similarities = []
    for i in range(len(embeddings) - 1):
        sim = cosine_similarity(embeddings[i], embeddings[i + 1])
        similarities.append(sim)

    # Step 4: Split at low-similarity points
    chunks = []
    current_chunk = [sentences[0]]

    for i, sim in enumerate(similarities):
        if sim < similarity_threshold and len(" ".join(current_chunk)) >= min_chunk_size:
            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
text = """
Python is a high-level programming language known for its readability. 
It was created by Guido van Rossum and first released in 1991. Python 
supports multiple programming paradigms including procedural and 
object-oriented programming.

The stock market experienced significant volatility last quarter. 
The S&P 500 dropped 8% in March before recovering in April. Analysts 
attribute the decline to rising interest rates and geopolitical tensions.

Machine learning models require large datasets for training. The quality 
of training data directly impacts model performance. Data augmentation 
techniques can help when labeled data is scarce.
"""

chunks = semantic_chunk(text, similarity_threshold=0.78)
for i, chunk in enumerate(chunks):
    print(f"--- Semantic Chunk {i} ---")
    print(chunk[:150])
    print()

import re


def markdown_chunk(text: str, max_chunk_size: int = 1000) -> list[dict]:
    """Chunk markdown by headings, preserving section hierarchy."""
    # Split on headings (##, ###, etc.)
    sections = re.split(r"(^#{1,4}\s+.+$)", text, flags=re.MULTILINE)

    chunks = []
    current_heading = "Introduction"
    current_content = ""

    for section in sections:
        if re.match(r"^#{1,4}\s+", section):
            # Save previous section
            if current_content.strip():
                chunks.append({
                    "content": current_content.strip(),
                    "heading": current_heading,
                    "char_count": len(current_content.strip()),
                })
            current_heading = section.strip("# \n")
            current_content = ""
        else:
            current_content += section

    # Don't forget the last section
    if current_content.strip():
        chunks.append({
            "content": current_content.strip(),
            "heading": current_heading,
            "char_count": len(current_content.strip()),
        })

    # Split oversized sections with recursive splitting
    final_chunks = []
    for chunk in chunks:
        if chunk["char_count"] > max_chunk_size:
            from langchain.text_splitter import RecursiveCharacterTextSplitter

            splitter = RecursiveCharacterTextSplitter(
                chunk_size=max_chunk_size, chunk_overlap=50
            )
            sub_texts = splitter.split_text(chunk["content"])
            for j, sub in enumerate(sub_texts):
                final_chunks.append({
                    "content": sub,
                    "heading": chunk["heading"],
                    "sub_section": j,
                })
        else:
            final_chunks.append(chunk)

    return final_chunks


# Example
markdown_doc = """
# API Reference

## Authentication

All API calls require a Bearer token in the Authorization header. 
Tokens expire after 24 hours. Use the /auth/refresh endpoint to 
get a new token without re-authenticating.

## Rate Limiting

The API enforces rate limits of 100 requests per minute per API key. 
Exceeding the limit returns a 429 status code with a Retry-After 
header indicating how many seconds to wait.

### Burst Limits

Short bursts of up to 20 requests per second are allowed. Sustained 
traffic above 100 RPM will trigger rate limiting.

## Pagination

All list endpoints support cursor-based pagination. Pass the cursor 
parameter from the previous response to get the next page. The default 
page size is 20 items, configurable up to 100.
"""

chunks = markdown_chunk(markdown_doc)
for chunk in chunks:
    print(f"[{chunk['heading']}] {chunk['content'][:80]}...")
    print()

Without overlap:
  Chunk 1: "...the refund policy requires customers to"
  Chunk 2: "submit a written request within 30 days."

With 50-char overlap:
  Chunk 1: "...the refund policy requires customers to"
  Chunk 2: "policy requires customers to submit a written request within 30 days."

from openai import OpenAI
from langchain.text_splitter import RecursiveCharacterTextSplitter
import numpy as np
import json

client = OpenAI()


def embed(texts: list[str]) -> list[list[float]]:
    """Batch embed texts."""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts,
    )
    return [d.embedding for d in response.data]


def cosine_sim(a, b):
    a, b = np.array(a), np.array(b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))


def evaluate_chunk_size(
    documents: list[str],
    queries: list[dict],  # [{"query": "...", "expected_content": "..."}]
    chunk_sizes: list[int],
    top_k: int = 5,
) -> dict:
    """Compare retrieval quality across different chunk sizes."""
    results = {}

    for size in chunk_sizes:
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=size, chunk_overlap=int(size * 0.1)
        )

        # Chunk all documents
        all_chunks = []
        for doc in documents:
            all_chunks.extend(splitter.split_text(doc))

        # Embed all chunks
        chunk_embeddings = embed(all_chunks)

        # Test each query
        hits = 0
        for q in queries:
            query_emb = embed([q["query"]])[0]

            # Find top-k most similar chunks
            sims = [cosine_sim(query_emb, ce) for ce in chunk_embeddings]
            top_indices = np.argsort(sims)[-top_k:][::-1]
            retrieved_text = " ".join(all_chunks[i] for i in top_indices)

            # Check if expected content appears in retrieved chunks
            if q["expected_content"].lower() in retrieved_text.lower():
                hits += 1

        results[size] = {
            "num_chunks": len(all_chunks),
            "recall": hits / len(queries),
            "avg_chunk_chars": np.mean([len(c) for c in all_chunks]),
        }

    return results


# Example usage
documents = [open(f).read() for f in ["doc1.txt", "doc2.txt", "doc3.txt"]]
queries = [
    {
        "query": "What is the refund policy for enterprise?",
        "expected_content": "30 days",
    },
    {
        "query": "How do I reset my API key?",
        "expected_content": "dashboard settings",
    },
    # ... 50-100 queries for a real evaluation
]

results = evaluate_chunk_size(
    documents, queries, chunk_sizes=[200, 500, 1000, 2000]
)

for size, metrics in sorted(results.items()):
    print(f"Chunk size {size:>5}: "
          f"{metrics['num_chunks']:>4} chunks, "
          f"recall={metrics['recall']:.2%}")

from dataclasses import dataclass, field
from datetime import datetime


@dataclass
class EnrichedChunk:
    content: str
    metadata: dict = field(default_factory=dict)


def enrich_chunks(
    chunks: list[str],
    document_title: str,
    source_url: str,
    section_headings: list[str] | None = None,
) -> list[EnrichedChunk]:
    """Add metadata to chunks for better retrieval context."""
    enriched = []
    for i, chunk in enumerate(chunks):
        meta = {
            "document_title": document_title,
            "source_url": source_url,
            "chunk_index": i,
            "total_chunks": len(chunks),
            "char_count": len(chunk),
            "indexed_at": datetime.utcnow().isoformat(),
        }

        # Add section heading if available
        if section_headings and i < len(section_headings):
            meta["section"] = section_headings[i]

        # Prepend context to the chunk text for better embeddings
        context_prefix = f"Document: {document_title}"
        if "section" in meta:
            context_prefix += f" | Section: {meta['section']}"

        enriched.append(EnrichedChunk(
            content=f"{context_prefix}\n\n{chunk}",
            metadata=meta,
        ))

    return enriched


# Usage
chunks = ["The process takes 3-5 business days...", "Contact support at..."]
enriched = enrich_chunks(
    chunks,
    document_title="Enterprise Onboarding Guide",
    source_url="https://docs.example.com/onboarding",
    section_headings=["Timeline", "Support"],
)

for ec in enriched:
    print(f"Content: {ec.content[:100]}...")
    print(f"Metadata: {json.dumps(ec.metadata, indent=2)}\n")

from dataclasses import dataclass


@dataclass
class ParentChildChunk:
    child_content: str  # Small, embedded for retrieval
    parent_content: str  # Large, sent to LLM
    child_id: str
    parent_id: str


def create_parent_child_chunks(
    text: str,
    parent_size: int = 2000,
    child_size: int = 300,
    child_overlap: int = 50,
) -> list[ParentChildChunk]:
    """Create two-level chunk hierarchy."""
    from langchain.text_splitter import RecursiveCharacterTextSplitter
    import hashlib

    # Level 1: Large parent chunks
    parent_splitter = RecursiveCharacterTextSplitter(
        chunk_size=parent_size, chunk_overlap=200
    )
    parent_texts = parent_splitter.split_text(text)

    # Level 2: Small child chunks within each parent
    child_splitter = RecursiveCharacterTextSplitter(
        chunk_size=child_size, chunk_overlap=child_overlap
    )

    all_chunks = []
    for p_idx, parent in enumerate(parent_texts):
        parent_id = hashlib.md5(parent[:100].encode()).hexdigest()[:12]
        children = child_splitter.split_text(parent)

        for c_idx, child in enumerate(children):
            child_id = f"{parent_id}_c{c_idx}"
            all_chunks.append(ParentChildChunk(
                child_content=child,
                parent_content=parent,
                child_id=child_id,
                parent_id=parent_id,
            ))

    return all_chunks


# Retrieval flow:
# 1. Embed all child_content and store in vector DB
# 2. On query, search against child embeddings
# 3. Return the parent_content of the matched children to the LLM
#
# This gives you the best of both worlds:
# - Precise retrieval (small chunks match specific queries)
# - Rich context (LLM sees the full surrounding text)

chunks = create_parent_child_chunks(open("large_doc.txt").read())
print(f"Parents: {len(set(c.parent_id for c in chunks))}")
print(f"Children: {len(chunks)}")
print(f"Child sizes: {[len(c.child_content) for c in chunks[:5]]}")
print(f"Parent sizes: {[len(c.parent_content) for c in chunks[:5]]}")

import ast


def chunk_python_code(source_code: str) -> list[dict]:
    """Chunk Python source code by functions and classes."""
    tree = ast.parse(source_code)
    lines = source_code.split("\n")
    chunks = []

    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            start = node.lineno - 1
            end = node.end_lineno
            func_code = "\n".join(lines[start:end])

            # Get docstring if present
            docstring = ast.get_docstring(node) or ""

            chunks.append({
                "content": func_code,
                "type": "function",
                "name": node.name,
                "docstring": docstring,
                "line_start": node.lineno,
                "line_end": node.end_lineno,
            })

        elif isinstance(node, ast.ClassDef):
            start = node.lineno - 1
            end = node.end_lineno
            class_code = "\n".join(lines[start:end])

            chunks.append({
                "content": class_code,
                "type": "class",
                "name": node.name,
                "docstring": ast.get_docstring(node) or "",
                "line_start": node.lineno,
                "line_end": node.end_lineno,
            })

    # Also capture module-level code that's not in any function/class
    covered_lines = set()
    for chunk in chunks:
        covered_lines.update(range(chunk["line_start"], chunk["line_end"] + 1))

    module_lines = []
    for i, line in enumerate(lines, 1):
        if i not in covered_lines and line.strip():
            module_lines.append(line)

    if module_lines:
        chunks.insert(0, {
            "content": "\n".join(module_lines),
            "type": "module_level",
            "name": "module",
            "docstring": "",
        })

    return chunks


# Example
source = '''
import os
from pathlib import Path

CONFIG_DIR = Path.home() / ".config"


class DatabaseConnection:
    """Manages database connections with pooling."""

    def __init__(self, url: str, pool_size: int = 5):
        self.url = url
        self.pool_size = pool_size

    def connect(self):
        """Establish a connection from the pool."""
        pass

    def close(self):
        """Return connection to pool."""
        pass


def process_file(path: str) -> dict:
    """Read and parse a configuration file."""
    with open(path) as f:
        return json.loads(f.read())
'''

chunks = chunk_python_code(source)
for chunk in chunks:
    print(f"[{chunk['type']}] {chunk['name']}: {len(chunk['content'])} chars")

import re
import hashlib
from dataclasses import dataclass, field
from pathlib import Path
from enum import Enum

from langchain.text_splitter import RecursiveCharacterTextSplitter


class DocType(Enum):
    MARKDOWN = "markdown"
    PLAIN_TEXT = "plain_text"
    CODE = "code"
    HTML = "html"


@dataclass
class ProcessedChunk:
    chunk_id: str
    content: str
    content_for_embedding: str  # May include metadata prefix
    metadata: dict = field(default_factory=dict)


class ChunkingPipeline:
    """Production chunking pipeline with multiple strategies."""

    def __init__(
        self,
        default_chunk_size: int = 500,
        default_overlap: int = 50,
    ):
        self.default_chunk_size = default_chunk_size
        self.default_overlap = default_overlap

    def process(
        self,
        text: str,
        doc_type: DocType,
        source: str = "",
        title: str = "",
    ) -> list[ProcessedChunk]:
        """Process a document into enriched chunks."""
        # Step 1: Detect doc type and apply appropriate strategy
        if doc_type == DocType.MARKDOWN:
            raw_chunks = self._chunk_markdown(text)
        elif doc_type == DocType.CODE:
            raw_chunks = self._chunk_code(text)
        elif doc_type == DocType.HTML:
            raw_chunks = self._chunk_html(text)
        else:
            raw_chunks = self._chunk_plain_text(text)

        # Step 2: Enrich with metadata
        processed = []
        for i, chunk_data in enumerate(raw_chunks):
            content = chunk_data["content"]
            section = chunk_data.get("section", "")

            chunk_id = hashlib.sha256(
                f"{source}:{i}:{content[:50]}".encode()
            ).hexdigest()[:16]

            # Build embedding text with context
            embed_parts = []
            if title:
                embed_parts.append(f"Document: {title}")
            if section:
                embed_parts.append(f"Section: {section}")
            embed_parts.append(content)
            content_for_embedding = "\n".join(embed_parts)

            processed.append(ProcessedChunk(
                chunk_id=chunk_id,
                content=content,
                content_for_embedding=content_for_embedding,
                metadata={
                    "source": source,
                    "title": title,
                    "section": section,
                    "chunk_index": i,
                    "char_count": len(content),
                    "doc_type": doc_type.value,
                },
            ))

        return processed

    def _chunk_markdown(self, text: str) -> list[dict]:
        """Chunk markdown respecting heading structure."""
        sections = re.split(r"(^#{1,4}\s+.+$)", text, flags=re.MULTILINE)
        chunks = []
        current_heading = ""
        current_text = ""

        for part in sections:
            if re.match(r"^#{1,4}\s+", part):
                if current_text.strip():
                    chunks.extend(self._split_section(
                        current_text.strip(), current_heading
                    ))
                current_heading = part.strip("# \n")
                current_text = ""
            else:
                current_text += part

        if current_text.strip():
            chunks.extend(self._split_section(
                current_text.strip(), current_heading
            ))

        return chunks

    def _split_section(self, text: str, heading: str) -> list[dict]:
        """Split a section that might be too large."""
        if len(text) <= self.default_chunk_size:
            return [{"content": text, "section": heading}]

        splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.default_chunk_size,
            chunk_overlap=self.default_overlap,
        )
        parts = splitter.split_text(text)
        return [{"content": p, "section": heading} for p in parts]

    def _chunk_plain_text(self, text: str) -> list[dict]:
        """Chunk plain text with recursive splitting."""
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=self.default_chunk_size,
            chunk_overlap=self.default_overlap,
        )
        parts = splitter.split_text(text)
        return [{"content": p, "section": ""} for p in parts]

    def _chunk_code(self, text: str) -> list[dict]:
        """Chunk source code by functions/classes."""
        try:
            chunks = chunk_python_code(text)  # from earlier example
            return [
                {"content": c["content"], "section": f"{c['type']}: {c['name']}"}
                for c in chunks
            ]
        except SyntaxError:
            return self._chunk_plain_text(text)

    def _chunk_html(self, text: str) -> list[dict]:
        """Chunk HTML by stripping tags and splitting."""
        from html.parser import HTMLParser

        class TextExtractor(HTMLParser):
            def __init__(self):
                super().__init__()
                self.parts = []
                self.current_heading = ""

            def handle_starttag(self, tag, attrs):
                if tag in ("h1", "h2", "h3", "h4"):
                    self._tag = tag

            def handle_data(self, data):
                if hasattr(self, "_tag"):
                    self.current_heading = data.strip()
                    del self._tag
                else:
                    if data.strip():
                        self.parts.append({
                            "content": data.strip(),
                            "section": self.current_heading,
                        })

        extractor = TextExtractor()
        extractor.feed(text)

        # Merge small consecutive parts from the same section
        merged = []
        current = {"content": "", "section": ""}
        for part in extractor.parts:
            if (part["section"] == current["section"]
                    and len(current["content"]) + len(part["content"])
                    < self.default_chunk_size):
                current["content"] += " " + part["content"]
            else:
                if current["content"]:
                    merged.append(current)
                current = dict(part)
        if current["content"]:
            merged.append(current)

        return merged


# Usage
pipeline = ChunkingPipeline(default_chunk_size=500, default_overlap=50)

# Process a markdown doc
md_chunks = pipeline.process(
    text=open("docs/api-reference.md").read(),
    doc_type=DocType.MARKDOWN,
    source="docs/api-reference.md",
    title="API Reference",
)

# Process source code
code_chunks = pipeline.process(
    text=open("src/auth.py").read(),
    doc_type=DocType.CODE,
    source="src/auth.py",
    title="Authentication Module",
)

for chunk in md_chunks[:3]:
    print(f"ID: {chunk.chunk_id}")
    print(f"Section: {chunk.metadata['section']}")
    print(f"Embed text: {chunk.content_for_embedding[:100]}...")
    print()

def score_chunk_coherence(chunks: list[str], sample_size: int = 50) -> float:
    """Use an LLM to score chunk coherence."""
    import random

    sample = random.sample(chunks, min(sample_size, len(chunks)))
    scores = []

    for chunk in sample:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {
                    "role": "system",
                    "content": (
                        "Rate the following text chunk on coherence from 1-5. "
                        "5 = complete, self-contained thought. "
                        "1 = fragment, starts/ends mid-sentence, or mixes unrelated topics. "
                        "Respond with just the number."
                    ),
                },
                {"role": "user", "content": chunk},
            ],
            max_tokens=5,
        )
        try:
            score = int(response.choices[0].message.content.strip())
            scores.append(score)
        except ValueError:
            continue

    return sum(scores) / len(scores) if scores else 0.0


# Compare strategies
from langchain.text_splitter import RecursiveCharacterTextSplitter

text = open("large_document.txt").read()

fixed_chunks = fixed_size_chunk(text, chunk_size=500)
recursive_chunks = RecursiveCharacterTextSplitter(
    chunk_size=500, chunk_overlap=50
).split_text(text)

print(f"Fixed-size coherence:  {score_chunk_coherence(fixed_chunks):.2f}/5")
print(f"Recursive coherence:   {score_chunk_coherence(recursive_chunks):.2f}/5")