Chunking Strategy

Chunking Strategy for RAG Systems

Overview

Provides chunking strategies for RAG systems, vector databases, and document processing. Recommends chunk sizes, overlap percentages, and boundary detection methods; validates semantic coherence; evaluates retrieval metrics.

When to Use

Use when building or optimizing RAG systems, vector search pipelines, document chunking workflows, or performance-tuning existing systems with poor retrieval quality.

Instructions

Choose Chunking Strategy

Select based on document type and use case:

1. Fixed-Size Chunking (Level 1)

• Use for simple documents without clear structure
• Start with 512 tokens and 10-20% overlap
• Adjust: 256 for factoid queries, 1024 for analytical

2. Recursive Character Chunking (Level 2)

• Use for documents with structural boundaries
• Hierarchical separators: paragraphs → sentences → words
• Customize for document types (HTML, Markdown, JSON)

3. Structure-Aware Chunking (Level 3)

• Use for structured content (Markdown, code, tables, PDFs)
• Preserve semantic units: functions, sections, table blocks
• Validate structure preservation post-split

4. Semantic Chunking (Level 4)

• Use for complex documents with thematic shifts
• Embedding-based boundary detection with 0.8 similarity threshold
• Buffer size: 3-5 sentences

5. Advanced Methods (Level 5)

• Late Chunking for long-context models
• Contextual Retrieval for high-precision requirements
• Monitor computational cost vs. retrieval gain

Reference: references/strategies.md.

Implement Chunking Pipeline

1. Pre-process documents

• Analyze structure, content types, information density
• Identify multi-modal content (tables, images, code)

2. Select parameters

• Chunk size: embedding model context window / 4
• Overlap: 10-20% for most cases
• Strategy-specific settings

3. Process and validate

• Apply chunking strategy
• Validate coherence: run evaluate_chunks.py --coherence (see below)
• Test with representative documents

4. Evaluate and iterate

• Measure precision and recall
• If precision < 0.7: reduce chunk_size by 25% and re-evaluate
• If recall < 0.6: increase overlap by 10% and re-evaluate
• Monitor latency and memory usage

Reference: references/implementation.md.

Validate Chunk Quality

Run validation commands to assess chunk quality:

# Check semantic coherence (requires sentence-transformers)
python -c "
from sentence_transformers import SentenceTransformer
model = SentenceTransformer('all-MiniLM-L6-v2')
chunks = [...]  # your chunks
embeddings = model.encode(chunks)
similarity = (embeddings @ embeddings.T).mean()
print(f'Cohesion: {similarity:.3f}')  # target: 0.3-0.7
"

# Measure retrieval precision
python -c "
relevant = sum(1 for c in retrieved if c in relevant_chunks)
precision = relevant / len(retrieved)
print(f'Precision: {precision:.2f}')  # target: >= 0.7
"

# Check chunk size distribution
python -c "
import numpy as np
sizes = [len(c.split()) for c in chunks]
print(f'Mean: {np.mean(sizes):.0f}, Std: {np.std(sizes):.0f}')
print(f'Min: {min(sizes)}, Max: {max(sizes)}')
"

Reference: references/evaluation.md.

Examples

Fixed-Size Chunking

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=256,
    chunk_overlap=25,
    length_function=len
)
chunks = splitter.split_documents(documents)

Structure-Aware Code Chunking

import ast

def chunk_python_code(code):
    tree = ast.parse(code)
    chunks = []
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.ClassDef)):
            chunks.append(ast.get_source_segment(code, node))
    return chunks

Semantic Chunking

def semantic_chunk(text, similarity_threshold=0.8):
    sentences = split_into_sentences(text)
    embeddings = generate_embeddings(sentences)
    chunks, current = [], [sentences[0]]
    for i in range(1, len(sentences)):
        sim = cosine_similarity(embeddings[i-1], embeddings[i])
        if sim < similarity_threshold:
            chunks.append(" ".join(current))
            current = [sentences[i]]
        else:
            current.append(sentences[i])
    chunks.append(" ".join(current))
    return chunks

Best Practices

Core Principles

• Balance context preservation with retrieval precision
• Maintain semantic coherence within chunks
• Optimize for embedding model context window constraints

Implementation

• Start with fixed-size (512 tokens, 15% overlap)
• Iterate based on document characteristics
• Test with domain-specific documents before deployment

Pitfalls to Avoid

• Over-chunking: context-poor small chunks
• Under-chunking: missing information in oversized chunks
• Ignoring semantic boundaries and document structure
• One-size-fits-all for diverse content types

Constraints and Warnings

Resource Considerations

• Semantic methods require significant compute resources
• Late chunking needs long-context embedding models
• Complex strategies increase processing latency
• Monitor memory for large document batches

Quality Requirements

• Validate semantic coherence post-processing
• Test with representative documents before deployment
• Ensure chunks maintain standalone meaning
• Implement error handling for malformed content

References

• strategies.md - Detailed strategies
• implementation.md - Implementation guidelines
• evaluation.md - Performance metrics
• tools.md - Libraries and frameworks
• research.md - Research papers
• advanced-strategies.md - 11 advanced methods
• semantic-methods.md - Semantic approaches
• visualization-tools.md - Visualization tools

Advanced Chunking Strategies

This document provides detailed implementations of 11 advanced chunking strategies for comprehensive RAG systems.

Strategy Overview

1. Fixed-Length Chunking

Overview

Divide documents into chunks of fixed character/token count regardless of content structure.

Implementation

from langchain.text_splitter import CharacterTextSplitter
import tiktoken

class FixedLengthChunker:
    def __init__(self, chunk_size=1000, chunk_overlap=200, encoding_name="cl100k_base"):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.encoding = tiktoken.get_encoding(encoding_name)

    def chunk_by_characters(self, text):
        """Chunk by character count"""
        splitter = CharacterTextSplitter(
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap,
            separator="\n\n"
        )
        return splitter.split_text(text)

    def chunk_by_tokens(self, text):
        """Chunk by token count using tiktoken"""
        tokens = self.encoding.encode(text)
        chunks = []

        start = 0
        while start < len(tokens):
            end = min(start + self.chunk_size, len(tokens))
            chunk_tokens = tokens[start:end]
            chunk_text = self.encoding.decode(chunk_tokens)
            chunks.append(chunk_text)

            # Calculate next start position with overlap
            start = max(0, end - self.chunk_overlap)

            # Prevent infinite loop
            if end >= len(tokens):
                break

        return chunks

    def chunk_optimized(self, text, strategy="balanced"):
        """Optimized chunking based on strategy"""
        strategies = {
            "conservative": {"chunk_size": 500, "overlap": 100},
            "balanced": {"chunk_size": 1000, "overlap": 200},
            "aggressive": {"chunk_size": 2000, "overlap": 400}
        }

        config = strategies.get(strategy, strategies["balanced"])
        self.chunk_size = config["chunk_size"]
        self.chunk_overlap = config["overlap"]

        return self.chunk_by_tokens(text)

Best Practices

• Start with 1000 tokens for general use
• Use 10-20% overlap for context preservation
• Adjust based on embedding model context window
• Consider document type for optimal sizing

2. Sentence-Based Chunking

Overview

Split documents at sentence boundaries while maintaining target chunk sizes.

Implementation

import nltk
import spacy
from typing import List

class SentenceChunker:
    def __init__(self, max_sentences=10, overlap_sentences=2, library="spacy"):
        self.max_sentences = max_sentences
        self.overlap_sentences = overlap_sentences
        self.library = library

        if library == "spacy":
            self.nlp = spacy.load("en_core_web_sm")
        elif library == "nltk":
            nltk.download('punkt')

    def extract_sentences_spacy(self, text):
        """Extract sentences using spaCy"""
        doc = self.nlp(text)
        return [sent.text.strip() for sent in doc.sents]

    def extract_sentences_nltk(self, text):
        """Extract sentences using NLTK"""
        sentences = nltk.sent_tokenize(text)
        return [sent.strip() for sent in sentences]

    def chunk_sentences(self, text):
        """Chunk text by sentences"""
        if self.library == "spacy":
            sentences = self.extract_sentences_spacy(text)
        else:
            sentences = self.extract_sentences_nltk(text)

        chunks = []
        for i in range(0, len(sentences), self.max_sentences - self.overlap_sentences):
            end_idx = min(i + self.max_sentences, len(sentences))
            chunk_sentences = sentences[i:end_idx]

            if chunk_sentences:
                chunk = " ".join(chunk_sentences)
                chunks.append(chunk)

        return chunks

    def chunk_with_metadata(self, text):
        """Chunk with sentence count metadata"""
        sentences = self.extract_sentences_spacy(text)
        chunks = []

        for i in range(0, len(sentences), self.max_sentences - self.overlap_sentences):
            end_idx = min(i + self.max_sentences, len(sentences))
            chunk_sentences = sentences[i:end_idx]

            if chunk_sentences:
                chunk = {
                    "text": " ".join(chunk_sentences),
                    "sentence_count": len(chunk_sentences),
                    "start_sentence": i,
                    "end_sentence": end_idx - 1,
                    "overlap": self.overlap_sentences > 0 and i > 0
                }
                chunks.append(chunk)

        return chunks

3. Paragraph-Based Chunking

Overview

Split documents at paragraph boundaries while maintaining semantic coherence.

Implementation

import re
from typing import List, Dict

class ParagraphChunker:
    def __init__(self, max_paragraphs=5, min_length=100, merge_short=True):
        self.max_paragraphs = max_paragraphs
        self.min_length = min_length
        self.merge_short = merge_short

    def extract_paragraphs(self, text):
        """Extract paragraphs from text"""
        # Split on various paragraph separators
        paragraphs = re.split(r'\n\s*\n|\r\n\s*\r\n', text)

        # Clean and filter paragraphs
        cleaned_paragraphs = []
        for para in paragraphs:
            para = para.strip()
            if para and len(para) > self.min_length // 4:  # Allow short paragraphs
                cleaned_paragraphs.append(para)

        return cleaned_paragraphs

    def chunk_paragraphs(self, text):
        """Chunk text by paragraphs"""
        paragraphs = self.extract_paragraphs(text)
        chunks = []

        current_chunk = []
        current_length = 0

        for i, paragraph in enumerate(paragraphs):
            paragraph_length = len(paragraph)

            # If adding this paragraph exceeds reasonable limits, start new chunk
            if (current_chunk and
                (len(current_chunk) >= self.max_paragraphs or
                 current_length + paragraph_length > 3000)):

                # Save current chunk
                if current_chunk:
                    chunks.append("\n\n".join(current_chunk))

                # Start new chunk with overlap
                overlap_count = min(2, len(current_chunk))
                current_chunk = current_chunk[-overlap_count:] if overlap_count > 0 else []
                current_length = sum(len(p) for p in current_chunk)

            current_chunk.append(paragraph)
            current_length += paragraph_length

        # Add final chunk
        if current_chunk:
            chunks.append("\n\n".join(current_chunk))

        return chunks

    def chunk_with_structure(self, text):
        """Chunk while preserving structure information"""
        paragraphs = self.extract_paragraphs(text)
        chunks = []

        current_chunk = []
        current_start = 0

        for i, paragraph in enumerate(paragraphs):
            current_chunk.append(paragraph)

            # Check if we should end the current chunk
            should_end = (
                len(current_chunk) >= self.max_paragraphs or
                (i < len(paragraphs) - 1 and
                 self._is_boundary_paragraph(paragraph, paragraphs[i + 1]))
            )

            if should_end or i == len(paragraphs) - 1:
                chunk_data = {
                    "text": "\n\n".join(current_chunk),
                    "paragraph_count": len(current_chunk),
                    "start_paragraph": current_start,
                    "end_paragraph": i,
                    "structure_type": self._detect_structure_type(current_chunk)
                }
                chunks.append(chunk_data)

                # Prepare for next chunk
                current_start = i + 1
                overlap_count = min(1, len(current_chunk))
                current_chunk = current_chunk[-overlap_count:] if overlap_count > 0 else []

        return chunks

    def _is_boundary_paragraph(self, current, next_para):
        """Check if there's a natural boundary between paragraphs"""
        boundary_indicators = [
            lambda c, n: c.strip().endswith(':'),  # Ends with colon
            lambda c, n: n.strip().startswith(('•', '-', '*')),  # List starts
            lambda c, n: bool(re.match(r'^\d+\.', n.strip())),  # Numbered list
            lambda c, n: len(n.strip()) < 50,  # Very short paragraph
        ]

        return any(indicator(current, next_para) for indicator in boundary_indicators)

    def _detect_structure_type(self, paragraphs):
        """Detect the type of structure in the chunk"""
        text = " ".join(paragraphs)

        if re.search(r'^#+\s', text, re.MULTILINE):
            return "markdown_headings"
        elif re.search(r'^\s*[-*+]\s', text, re.MULTILINE):
            return "bullet_points"
        elif re.search(r'^\s*\d+\.\s', text, re.MULTILINE):
            return "numbered_list"
        elif any(char.isdigit() for char in text) and ('%' in text or '$' in text):
            return "data_heavy"
        else:
            return "prose"

4. Sliding Window Chunking

Overview

Create overlapping chunks using a sliding window approach for maximum context preservation.

Implementation

from typing import List, Iterator
import numpy as np

class SlidingWindowChunker:
    def __init__(self, window_size=1000, step_size=500, unit="tokens"):
        self.window_size = window_size
        self.step_size = step_size
        self.unit = unit

    def sliding_chunk_tokens(self, text, encoding_name="cl100k_base"):
        """Create sliding window chunks by tokens"""
        import tiktoken
        encoding = tiktoken.get_encoding(encoding_name)
        tokens = encoding.encode(text)

        chunks = []
        for start in range(0, len(tokens), self.step_size):
            end = min(start + self.window_size, len(tokens))
            window_tokens = tokens[start:end]
            chunk_text = encoding.decode(window_tokens)

            chunks.append({
                "text": chunk_text,
                "start_token": start,
                "end_token": end - 1,
                "token_count": len(window_tokens),
                "overlap": self.window_size - self.step_size
            })

            if end >= len(tokens):
                break

        return chunks

    def sliding_chunk_characters(self, text):
        """Create sliding window chunks by characters"""
        chunks = []
        for start in range(0, len(text), self.step_size):
            end = min(start + self.window_size, len(text))
            chunk_text = text[start:end]

            chunks.append({
                "text": chunk_text,
                "start_char": start,
                "end_char": end - 1,
                "char_count": len(chunk_text),
                "overlap": self.window_size - self.step_size
            })

            if end >= len(text):
                break

        return chunks

    def adaptive_sliding_window(self, text, min_overlap=0.1, max_overlap=0.5):
        """Adaptive sliding window based on content density"""
        if self.unit == "tokens":
            base_chunks = self.sliding_chunk_tokens(text)
        else:
            base_chunks = self.sliding_chunk_characters(text)

        # Analyze content density
        adaptive_chunks = []
        for i, chunk in enumerate(base_chunks):
            text_content = chunk["text"]
            density = self._calculate_content_density(text_content)

            # Adjust overlap based on density
            if density > 0.8:  # High density - more overlap
                adjusted_overlap = int(self.window_size * max_overlap)
            elif density < 0.3:  # Low density - less overlap
                adjusted_overlap = int(self.window_size * min_overlap)
            else:
                adjusted_overlap = self.window_size - self.step_size

            chunk["content_density"] = density
            chunk["adjusted_overlap"] = adjusted_overlap
            adaptive_chunks.append(chunk)

        return adaptive_chunks

    def _calculate_content_density(self, text):
        """Calculate content density (information per unit)"""
        # Simple heuristic: unique words / total words
        words = text.split()
        if not words:
            return 0.0

        unique_words = set(word.lower().strip('.,!?;:()[]{}"\'') for word in words)
        density = len(unique_words) / len(words)

        # Adjust for punctuation and special characters
        special_chars = sum(1 for char in text if not char.isalnum() and not char.isspace())
        density += special_chars / len(text) * 0.1

        return min(density, 1.0)

    def semantic_sliding_window(self, text, embedding_model, similarity_threshold=0.7):
        """Sliding window with semantic boundary detection"""
        from sentence_transformers import SentenceTransformer
        from sklearn.metrics.pairwise import cosine_similarity
        import numpy as np

        # Split into sentences
        sentences = self._split_into_sentences(text)
        if len(sentences) < 2:
            return [{"text": text, "method": "single_sentence"}]

        # Generate sentence embeddings
        sentence_embeddings = embedding_model.encode(sentences)

        chunks = []
        current_window_sentences = []
        current_window_start = 0

        for i, sentence in enumerate(sentences):
            current_window_sentences.append(sentence)

            # Check if we should create a boundary
            should_create_boundary = (
                len(current_window_sentences) >= 10 or  # Max sentences per window
                (i < len(sentences) - 1 and  # Not the last sentence
                 self._should_create_semantic_boundary(
                     sentence_embeddings, i, similarity_threshold
                 ))
            )

            if should_create_boundary:
                chunk_text = " ".join(current_window_sentences)
                chunks.append({
                    "text": chunk_text,
                    "sentence_count": len(current_window_sentences),
                    "start_sentence": current_window_start,
                    "end_sentence": i,
                    "method": "semantic_sliding_window"
                })

                # Start new window with overlap
                overlap_size = min(2, len(current_window_sentences) // 2)
                current_window_sentences = current_window_sentences[-overlap_size:]
                current_window_start = i + 1 - overlap_size

        # Add final chunk
        if current_window_sentences:
            chunk_text = " ".join(current_window_sentences)
            chunks.append({
                "text": chunk_text,
                "sentence_count": len(current_window_sentences),
                "start_sentence": current_window_start,
                "end_sentence": len(sentences) - 1,
                "method": "semantic_sliding_window"
            })

        return chunks

    def _split_into_sentences(self, text):
        """Split text into sentences"""
        import re
        # Simple sentence splitting
        sentences = re.split(r'[.!?]+', text)
        return [s.strip() for s in sentences if s.strip()]

    def _should_create_semantic_boundary(self, embeddings, current_idx, threshold):
        """Determine if semantic boundary should be created"""
        if current_idx >= len(embeddings) - 1:
            return True

        # Calculate similarity with next sentence
        current_embedding = embeddings[current_idx].reshape(1, -1)
        next_embedding = embeddings[current_idx + 1].reshape(1, -1)

        similarity = cosine_similarity(current_embedding, next_embedding)[0][0]

        return similarity < threshold

5. Semantic Chunking

Overview

Use semantic similarity to identify natural boundaries in text.

Implementation

from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np
from typing import List, Dict

class SemanticChunker:
    def __init__(self, model_name="all-MiniLM-L6-v2",
                 similarity_threshold=0.8,
                 min_chunk_size=2,
                 max_chunk_size=10):
        self.model = SentenceTransformer(model_name)
        self.similarity_threshold = similarity_threshold
        self.min_chunk_size = min_chunk_size
        self.max_chunk_size = max_chunk_size

    def semantic_chunk_sentences(self, text):
        """Chunk text based on semantic similarity between sentences"""
        # Split into sentences
        sentences = self._split_into_sentences(text)
        if len(sentences) <= self.min_chunk_size:
            return [{"text": text, "sentence_count": len(sentences), "method": "single_chunk"}]

        # Generate embeddings for all sentences
        sentence_embeddings = self.model.encode(sentences)

        # Find semantic boundaries
        boundaries = self._find_semantic_boundaries(sentence_embeddings)

        # Create chunks based on boundaries
        chunks = []
        start_idx = 0

        for boundary_idx in boundaries:
            if boundary_idx > start_idx:
                chunk_sentences = sentences[start_idx:boundary_idx + 1]
                chunk_text = " ".join(chunk_sentences)

                chunks.append({
                    "text": chunk_text,
                    "sentence_count": len(chunk_sentences),
                    "start_sentence": start_idx,
                    "end_sentence": boundary_idx,
                    "method": "semantic_boundary"
                })

                start_idx = boundary_idx + 1

        # Add remaining sentences
        if start_idx < len(sentences):
            chunk_sentences = sentences[start_idx:]
            chunk_text = " ".join(chunk_sentences)

            chunks.append({
                "text": chunk_text,
                "sentence_count": len(chunk_sentences),
                "start_sentence": start_idx,
                "end_sentence": len(sentences) - 1,
                "method": "semantic_boundary"
            })

        return self._merge_small_chunks(chunks)

    def _find_semantic_boundaries(self, embeddings):
        """Find semantic boundaries based on similarity thresholds"""
        boundaries = []

        for i in range(len(embeddings) - 1):
            # Calculate similarity between consecutive sentences
            similarity = cosine_similarity(
                embeddings[i].reshape(1, -1),
                embeddings[i + 1].reshape(1, -1)
            )[0][0]

            # If similarity is below threshold, create boundary
            if similarity < self.similarity_threshold:
                boundaries.append(i)

        return boundaries

    def _split_into_sentences(self, text):
        """Split text into sentences"""
        import re
        # Enhanced sentence splitting
        sentences = re.split(r'(?<=[.!?])\s+(?=[A-Z])', text)
        return [s.strip() for s in sentences if s.strip()]

    def _merge_small_chunks(self, chunks):
        """Merge chunks that are too small"""
        if not chunks:
            return chunks

        merged_chunks = []
        current_chunk = chunks[0].copy()

        for next_chunk in chunks[1:]:
            if (current_chunk["sentence_count"] < self.min_chunk_size and
                current_chunk["sentence_count"] + next_chunk["sentence_count"] <= self.max_chunk_size):

                # Merge chunks
                current_chunk["text"] += " " + next_chunk["text"]
                current_chunk["sentence_count"] += next_chunk["sentence_count"]
                current_chunk["end_sentence"] = next_chunk["end_sentence"]
            else:
                merged_chunks.append(current_chunk)
                current_chunk = next_chunk.copy()

        merged_chunks.append(current_chunk)
        return merged_chunks

    def adaptive_semantic_chunking(self, text, content_analyzer=None):
        """Semantic chunking with adaptive threshold"""
        sentences = self._split_into_sentences(text)
        if len(sentences) <= 2:
            return [{"text": text, "method": "too_short"}]

        # Generate embeddings
        embeddings = self.model.encode(sentences)

        # Analyze content complexity
        if content_analyzer:
            complexity = content_analyzer.analyze_complexity(text)
            # Adjust threshold based on complexity
            adaptive_threshold = self.similarity_threshold * (1.0 + complexity * 0.2)
        else:
            adaptive_threshold = self.similarity_threshold

        # Find boundaries with adaptive threshold
        boundaries = self._find_adaptive_boundaries(embeddings, adaptive_threshold)

        # Create chunks
        chunks = []
        start_idx = 0

        for boundary_idx in boundaries:
            if boundary_idx > start_idx:
                chunk_sentences = sentences[start_idx:boundary_idx + 1]
                chunk_text = " ".join(chunk_sentences)

                chunks.append({
                    "text": chunk_text,
                    "sentence_count": len(chunk_sentences),
                    "start_sentence": start_idx,
                    "end_sentence": boundary_idx,
                    "method": "adaptive_semantic",
                    "threshold_used": adaptive_threshold
                })

                start_idx = boundary_idx + 1

        # Add remaining sentences
        if start_idx < len(sentences):
            chunk_sentences = sentences[start_idx:]
            chunk_text = " ".join(chunk_sentences)

            chunks.append({
                "text": chunk_text,
                "sentence_count": len(chunk_sentences),
                "start_sentence": start_idx,
                "end_sentence": len(sentences) - 1,
                "method": "adaptive_semantic",
                "threshold_used": adaptive_threshold
            })

        return chunks

    def _find_adaptive_boundaries(self, embeddings, threshold):
        """Find boundaries with adaptive threshold based on local context"""
        boundaries = []

        for i in range(len(embeddings) - 1):
            # Calculate local similarity
            local_similarities = []

            # Look at local window of similarities
            window_size = min(3, i)
            for j in range(max(0, i - window_size), i + 1):
                if j < len(embeddings) - 1:
                    similarity = cosine_similarity(
                        embeddings[j].reshape(1, -1),
                        embeddings[j + 1].reshape(1, -1)
                    )[0][0]
                    local_similarities.append(similarity)

            # Use local average for comparison
            if local_similarities:
                local_avg = np.mean(local_similarities)
                current_similarity = local_similarities[-1]

                # Create boundary if current similarity is significantly lower than local average
                if current_similarity < local_avg * threshold:
                    boundaries.append(i)
            else:
                # Fallback to global threshold
                similarity = cosine_similarity(
                    embeddings[i].reshape(1, -1),
                    embeddings[i + 1].reshape(1, -1)
                )[0][0]

                if similarity < threshold:
                    boundaries.append(i)

        return boundaries

    def hierarchical_semantic_chunking(self, text, max_levels=3):
        """Multi-level semantic chunking"""
        sentences = self._split_into_sentences(text)

        if len(sentences) <= 4:
            return [{
                "text": text,
                "level": 0,
                "sentence_count": len(sentences),
                "method": "hierarchical_semantic"
            }]

        # Level 0: Original text
        chunks = [{
            "text": text,
            "level": 0,
            "sentence_count": len(sentences),
            "method": "hierarchical_semantic"
        }]

        # Generate embeddings once
        embeddings = self.model.encode(sentences)

        # Create hierarchical chunks
        current_level_sentences = sentences
        current_level_embeddings = embeddings

        for level in range(1, max_levels + 1):
            if len(current_level_sentences) <= 2:
                break

            # Find boundaries at this level
            boundaries = self._find_semantic_boundaries(current_level_embeddings)

            # Create chunks at this level
            level_chunks = []
            start_idx = 0

            for boundary_idx in boundaries:
                if boundary_idx > start_idx:
                    chunk_sentences = current_level_sentences[start_idx:boundary_idx + 1]
                    chunk_text = " ".join(chunk_sentences)

                    level_chunks.append({
                        "text": chunk_text,
                        "level": level,
                        "sentence_count": len(chunk_sentences),
                        "start_sentence": start_idx,
                        "end_sentence": boundary_idx,
                        "method": "hierarchical_semantic"
                    })

                    start_idx = boundary_idx + 1

            # Add remaining sentences
            if start_idx < len(current_level_sentences):
                chunk_sentences = current_level_sentences[start_idx:]
                chunk_text = " ".join(chunk_sentences)

                level_chunks.append({
                    "text": chunk_text,
                    "level": level,
                    "sentence_count": len(chunk_sentences),
                    "start_sentence": start_idx,
                    "end_sentence": len(current_level_sentences) - 1,
                    "method": "hierarchical_semantic"
                })

            chunks.extend(level_chunks)

            # Prepare for next level
            if len(level_chunks) > 1:
                current_level_sentences = [chunk["text"] for chunk in level_chunks]
                current_level_embeddings = self.model.encode(current_level_sentences)
            else:
                break

        return chunks

6. Recursive Chunking

Overview

Hierarchical splitting using ordered separators to preserve document structure.

Implementation

from typing import List, Dict, Optional
import re

class RecursiveChunker:
    def __init__(self, chunk_size=1000, chunk_overlap=200,
                 separators=None, length_function=len):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.length_function = length_function

        # Default separators in order of preference
        self.separators = separators or [
            "\n\n\n",  # Triple newlines (section breaks)
            "\n\n",    # Double newlines (paragraph breaks)
            "\n",      # Single newlines (line breaks)
            " ",       # Spaces (word breaks)
            ""         # Character-level (last resort)
        ]

    def recursive_split(self, text, separators=None):
        """Recursively split text using hierarchical separators"""
        separators = separators or self.separators
        final_chunks = []

        # Try each separator in order
        for separator in separators:
            if separator == "":
                # Last resort: split by characters
                return self._split_by_characters(text)

            # Split by current separator
            splits = text.split(separator)

            # Filter out empty splits
            splits = [split for split in splits if split.strip()]

            if len(splits) > 1:
                # Found a good separator
                for split in splits:
                    if self.length_function(split) <= self.chunk_size:
                        final_chunks.append(split)
                    else:
                        # Recursively split this piece
                        sub_chunks = self.recursive_split(split, separators[separators.index(separator) + 1:])
                        final_chunks.extend(sub_chunks)

                return self._merge_chunks(final_chunks)

        # No separator worked, split by characters
        return self._split_by_characters(text)

    def _split_by_characters(self, text):
        """Split text by characters as last resort"""
        chunks = []
        start = 0

        while start < len(text):
            end = min(start + self.chunk_size, len(text))
            chunk = text[start:end]
            chunks.append(chunk)

            # Calculate next start with overlap
            start = max(0, end - self.chunk_overlap)

            if end >= len(text):
                break

        return chunks

    def _merge_chunks(self, chunks):
        """Merge chunks that are too small"""
        if not chunks:
            return chunks

        merged_chunks = []
        current_chunk = chunks[0]

        for next_chunk in chunks[1:]:
            combined_length = self.length_function(current_chunk + next_chunk)

            if combined_length <= self.chunk_size:
                # Merge chunks
                current_chunk += "\n\n" + next_chunk
            else:
                # Add current chunk and start new one
                merged_chunks.append(current_chunk)
                current_chunk = next_chunk

        merged_chunks.append(current_chunk)
        return merged_chunks

    def recursive_split_with_metadata(self, text, separators=None):
        """Recursive split with detailed metadata"""
        separators = separators or self.separators
        chunks = []

        def _recursive_split_with_context(text_chunk, parent_separator=""):
            nonlocal chunks

            for separator in separators:
                if separator == "":
                    sub_chunks = self._split_by_characters(text_chunk)
                    for i, chunk in enumerate(sub_chunks):
                        chunks.append({
                            "text": chunk,
                            "separator": "character",
                            "parent_separator": parent_separator,
                            "level": len(separators) - separators.index(separator),
                            "chunk_index": len(chunks),
                            "size": self.length_function(chunk)
                        })
                    return

                splits = text_chunk.split(separator)
                splits = [split for split in splits if split.strip()]

                if len(splits) > 1:
                    for i, split in enumerate(splits):
                        if self.length_function(split) <= self.chunk_size:
                            chunks.append({
                                "text": split,
                                "separator": separator,
                                "parent_separator": parent_separator,
                                "level": len(separators) - separators.index(separator),
                                "chunk_index": len(chunks),
                                "size": self.length_function(split)
                            })
                        else:
                            # Recursively split this piece
                            _recursive_split_with_context(split, separator)
                    return

            # No separator worked
            sub_chunks = self._split_by_characters(text_chunk)
            for i, chunk in enumerate(sub_chunks):
                chunks.append({
                    "text": chunk,
                    "separator": "character_fallback",
                    "parent_separator": parent_separator,
                    "level": len(separators),
                    "chunk_index": len(chunks),
                    "size": self.length_function(chunk)
                })

        _recursive_split_with_context(text)
        return chunks

    def markdown_aware_recursive_split(self, text):
        """Recursive splitting optimized for Markdown documents"""
        markdown_separators = [
            "\n# ",      # H1 headers
            "\n## ",     # H2 headers
            "\n### ",    # H3 headers
            "\n#### ",   # H4 headers
            "\n##### ",  # H5 headers
            "\n###### ", # H6 headers
            "\n\n",      # Paragraph breaks
            "\n",        # Line breaks
            " ",         # Spaces
            ""           # Characters
        ]

        chunks = []

        def _split_markdown(text_chunk, separator_idx=0):
            if separator_idx >= len(markdown_separators):
                return self._split_by_characters(text_chunk)

            separator = markdown_separators[separator_idx]

            if separator.startswith("\n#"):
                # Markdown headers
                pattern = re.escape(separator)
                splits = re.split(pattern, text_chunk)

                if len(splits) > 1:
                    # Re-add separator to splits (except first)
                    for i in range(1, len(splits)):
                        splits[i] = separator + splits[i]

                    result_chunks = []
                    for split in splits:
                        if self.length_function(split) <= self.chunk_size:
                            result_chunks.append(split)
                        else:
                            # Try next level separator
                            sub_chunks = _split_markdown(split, separator_idx + 1)
                            result_chunks.extend(sub_chunks)

                    return result_chunks
            else:
                # Regular separators
                splits = text_chunk.split(separator)
                splits = [split for split in splits if split.strip()]

                if len(splits) > 1:
                    result_chunks = []
                    for split in splits:
                        if self.length_function(split) <= self.chunk_size:
                            result_chunks.append(split)
                        else:
                            # Try next level separator
                            sub_chunks = _split_markdown(split, separator_idx + 1)
                            result_chunks.extend(sub_chunks)

                    return result_chunks

            # Try next separator
            return _split_markdown(text_chunk, separator_idx + 1)

        raw_chunks = _split_markdown(text)

        # Add metadata
        for i, chunk in enumerate(raw_chunks):
            chunks.append({
                "text": chunk,
                "chunk_index": i,
                "size": self.length_function(chunk),
                "format": "markdown",
                "contains_header": bool(re.search(r'^#+\s', chunk, re.MULTILINE)),
                "contains_code": bool(re.search(r'```', chunk)),
                "contains_list": bool(re.search(r'^\s*[-*+]\s', chunk, re.MULTILINE))
            })

        return chunks

7-11. Additional Advanced Strategies

7. Context-Enriched Chunking

class ContextEnrichedChunker:
    def __init__(self, base_chunker, context_generator=None):
        self.base_chunker = base_chunker
        self.context_generator = context_generator

    def enrich_chunks(self, text, query_context=None):
        """Add contextual information to chunks"""
        base_chunks = self.base_chunker.chunk(text)
        enriched_chunks = []

        for i, chunk in enumerate(base_chunks):
            # Generate context for this chunk
            context = self._generate_context(chunk, text, i, query_context)

            enriched_chunk = {
                "original_text": chunk,
                "context": context,
                "enriched_text": f"Context: {context}\n\nContent: {chunk}",
                "chunk_index": i,
                "method": "context_enriched"
            }
            enriched_chunks.append(enriched_chunk)

        return enriched_chunks

    def _generate_context(self, chunk, full_text, chunk_index, query_context):
        """Generate contextual information for a chunk"""
        # Simple context generation
        sentences = full_text.split('.')

        # Find sentences before and after
        chunk_start = full_text.find(chunk)
        chunk_end = chunk_start + len(chunk)

        # Get preceding and following context
        pre_context = full_text[max(0, chunk_start - 200):chunk_start]
        post_context = full_text[chunk_end:chunk_end + 200]

        context_parts = []
        if pre_context.strip():
            context_parts.append(f"Preceding: {pre_context.strip()}")
        if post_context.strip():
            context_parts.append(f"Following: {post_context.strip()}")

        return " | ".join(context_parts)

8. Modality-Specific Chunking

class ModalitySpecificChunker:
    def __init__(self):
        self.chunkers = {
            "text": RecursiveChunker(),
            "code": CodeChunker(),
            "table": TableChunker(),
            "image": ImageChunker()
        }

    def chunk_mixed_content(self, document):
        """Chunk document with multiple content types"""
        chunks = []

        # Detect content types
        sections = self._detect_content_types(document)

        for section in sections:
            content_type = section["type"]
            content = section["content"]

            if content_type in self.chunkers:
                section_chunks = self.chunkers[content_type].chunk(content)

                for chunk in section_chunks:
                    chunks.append({
                        "content": chunk,
                        "type": content_type,
                        "metadata": section.get("metadata", {}),
                        "method": f"modality_specific_{content_type}"
                    })

        return chunks

    def _detect_content_types(self, document):
        """Detect different content types in document"""
        sections = []

        # Simple detection logic
        if "```" in document:
            # Code blocks detected
            code_blocks = re.findall(r'```(\w+)?\n(.*?)\n```', document, re.DOTALL)
            for lang, code in code_blocks:
                sections.append({
                    "type": "code",
                    "content": code,
                    "metadata": {"language": lang}
                })

        if "|" in document and "\n" in document:
            # Potential table detected
            sections.append({
                "type": "table",
                "content": document,  # Simplified
                "metadata": {}
            })

        # Default to text
        sections.append({
            "type": "text",
            "content": document,
            "metadata": {}
        })

        return sections

9. Agentic Chunking

class AgenticChunker:
    def __init__(self, chunking_agents):
        self.agents = chunking_agents

    def adaptive_chunking(self, text, requirements):
        """Use agents to determine optimal chunking strategy"""
        # Analyze text characteristics
        text_analysis = self._analyze_text(text)

        # Select appropriate agent based on requirements and text
        selected_agent = self._select_agent(text_analysis, requirements)

        # Use selected agent for chunking
        chunks = selected_agent.chunk(text, requirements)

        return {
            "chunks": chunks,
            "selected_agent": selected_agent.name,
            "reasoning": selected_agent.reasoning,
            "text_analysis": text_analysis
        }

    def _analyze_text(self, text):
        """Analyze text characteristics"""
        return {
            "length": len(text),
            "complexity": self._calculate_complexity(text),
            "structure": self._detect_structure(text),
            "content_type": self._detect_content_type(text)
        }

    def _select_agent(self, analysis, requirements):
        """Select best chunking agent"""
        for agent in self.agents:
            if agent.can_handle(analysis, requirements):
                return agent

        # Fallback to first agent
        return self.agents[0]

10. Subdocument Chunking

class SubdocumentChunker:
    def __init__(self, max_size=5000):
        self.max_size = max_size

    def chunk_by_logical_sections(self, document):
        """Chunk document by logical sections"""
        sections = self._identify_logical_sections(document)
        chunks = []

        for section in sections:
            if len(section["content"]) <= self.max_size:
                chunks.append({
                    "content": section["content"],
                    "title": section["title"],
                    "level": section["level"],
                    "method": "subdocument_section"
                })
            else:
                # Further split large sections
                sub_chunks = self._split_large_section(section)
                chunks.extend(sub_chunks)

        return chunks

    def _identify_logical_sections(self, document):
        """Identify logical sections in document"""
        sections = []

        # Simple heading detection
        heading_pattern = r'^(#{1,6})\s+(.+)$'
        lines = document.split('\n')

        current_section = {"title": "Introduction", "content": "", "level": 0}

        for line in lines:
            match = re.match(heading_pattern, line)
            if match:
                # Save current section
                if current_section["content"].strip():
                    sections.append(current_section)

                # Start new section
                level = len(match.group(1))
                title = match.group(2)
                current_section = {
                    "title": title,
                    "content": "",
                    "level": level
                }
            else:
                current_section["content"] += line + "\n"

        # Add final section
        if current_section["content"].strip():
            sections.append(current_section)

        return sections

11. Hybrid Chunking

class HybridChunker:
    def __init__(self, strategies, weights=None):
        self.strategies = strategies
        self.weights = weights or [1.0 / len(strategies)] * len(strategies)

    def hybrid_chunk(self, text, evaluation_criteria=None):
        """Combine multiple chunking strategies"""
        all_chunks = []

        # Apply all strategies
        for i, strategy in enumerate(self.strategies):
            strategy_chunks = strategy.chunk(text)

            for chunk in strategy_chunks:
                all_chunks.append({
                    "content": chunk,
                    "strategy": strategy.name,
                    "strategy_weight": self.weights[i],
                    "method": "hybrid"
                })

        # Evaluate and select best chunks
        if evaluation_criteria:
            evaluated_chunks = self._evaluate_chunks(all_chunks, evaluation_criteria)
        else:
            evaluated_chunks = all_chunks

        # Merge overlapping chunks from different strategies
        merged_chunks = self._merge_overlapping_chunks(evaluated_chunks)

        return merged_chunks

    def _evaluate_chunks(self, chunks, criteria):
        """Evaluate chunks based on criteria"""
        for chunk in chunks:
            score = 0.0
            for criterion, weight in criteria.items():
                criterion_score = self._evaluate_criterion(chunk, criterion)
                score += criterion_score * weight

            chunk["evaluation_score"] = score

        # Sort by evaluation score
        chunks.sort(key=lambda x: x["evaluation_score"], reverse=True)
        return chunks

    def _merge_overlapping_chunks(self, chunks):
        """Merge chunks that overlap significantly"""
        # Simple implementation - could be more sophisticated
        merged = []
        used_indices = set()

        for i, chunk1 in enumerate(chunks):
            if i in used_indices:
                continue

            best_chunk = chunk1.copy()

            for j, chunk2 in enumerate(chunks[i+1:], i+1):
                if j in used_indices:
                    continue

                # Check overlap
                overlap = self._calculate_overlap(chunk1["content"], chunk2["content"])

                if overlap > 0.7:  # High overlap
                    # Merge chunks
                    best_chunk["content"] = max(
                        chunk1["content"],
                        chunk2["content"],
                        key=len
                    )
                    best_chunk["merged_strategies"] = [
                        chunk1["strategy"],
                        chunk2["strategy"]
                    ]
                    used_indices.add(j)

            merged.append(best_chunk)
            used_indices.add(i)

        return merged

    def _calculate_overlap(self, text1, text2):
        """Calculate text overlap ratio"""
        words1 = set(text1.lower().split())
        words2 = set(text2.lower().split())

        intersection = words1 & words2
        union = words1 | words2

        return len(intersection) / len(union) if union else 0

Usage Examples

Basic Usage

# Initialize different chunkers
fixed_chunker = FixedLengthChunker(chunk_size=1000, chunk_overlap=200)
semantic_chunker = SemanticChunker(similarity_threshold=0.8)
hybrid_chunker = HybridChunker([fixed_chunker, semantic_chunker])

# Apply chunking
text = "Your long document text here..."

fixed_chunks = fixed_chunker.chunk_optimized(text, strategy="balanced")
semantic_chunks = semantic_chunker.semantic_chunk_sentences(text)
hybrid_chunks = hybrid_chunker.hybrid_chunk(text)

print(f"Fixed chunks: {len(fixed_chunks)}")
print(f"Semantic chunks: {len(semantic_chunks)}")
print(f"Hybrid chunks: {len(hybrid_chunks)}")

Advanced Usage with Evaluation

# Create evaluation criteria
evaluation_criteria = {
    "coherence": 0.4,
    "size_appropriateness": 0.3,
    "content_completeness": 0.3
}

# Apply hybrid chunking with evaluation
results = hybrid_chunker.hybrid_chunk(text, evaluation_criteria)

# Analyze results
for chunk in results[:5]:
    print(f"Strategy: {chunk['strategy']}")
    print(f"Score: {chunk.get('evaluation_score', 'N/A')}")
    print(f"Content preview: {chunk['content'][:100]}...")
    print("-" * 50)

These 11 advanced chunking strategies provide comprehensive coverage of different approaches for various document types and use cases, from simple fixed-size chunking to sophisticated hybrid methods that combine multiple strategies.

Performance Evaluation Framework

This document provides comprehensive methodologies for evaluating chunking strategy performance and effectiveness.

Evaluation Metrics

Core Retrieval Metrics

Retrieval Precision

Measures the fraction of retrieved chunks that are relevant to the query.

def calculate_precision(retrieved_chunks: List[Dict], relevant_chunks: List[Dict]) -> float:
    """
    Calculate retrieval precision
    Precision = |Relevant ∩ Retrieved| / |Retrieved|
    """
    retrieved_ids = {chunk.get('id') for chunk in retrieved_chunks}
    relevant_ids = {chunk.get('id') for chunk in relevant_chunks}

    intersection = retrieved_ids & relevant_ids

    if not retrieved_ids:
        return 0.0

    return len(intersection) / len(retrieved_ids)

Retrieval Recall

Measures the fraction of relevant chunks that are successfully retrieved.

def calculate_recall(retrieved_chunks: List[Dict], relevant_chunks: List[Dict]) -> float:
    """
    Calculate retrieval recall
    Recall = |Relevant ∩ Retrieved| / |Relevant|
    """
    retrieved_ids = {chunk.get('id') for chunk in retrieved_chunks}
    relevant_ids = {chunk.get('id') for chunk in relevant_chunks}

    intersection = retrieved_ids & relevant_ids

    if not relevant_ids:
        return 0.0

    return len(intersection) / len(relevant_ids)

F1-Score

Harmonic mean of precision and recall.

def calculate_f1_score(precision: float, recall: float) -> float:
    """
    Calculate F1-score
    F1 = 2 * (Precision * Recall) / (Precision + Recall)
    """
    if precision + recall == 0:
        return 0.0

    return 2 * (precision * recall) / (precision + recall)

Mean Reciprocal Rank (MRR)

Measures the rank of the first relevant result.

def calculate_mrr(queries: List[Dict], results: List[List[Dict]]) -> float:
    """
    Calculate Mean Reciprocal Rank
    """
    reciprocal_ranks = []

    for query, query_results in zip(queries, results):
        relevant_found = False

        for rank, result in enumerate(query_results, 1):
            if result.get('is_relevant', False):
                reciprocal_ranks.append(1.0 / rank)
                relevant_found = True
                break

        if not relevant_found:
            reciprocal_ranks.append(0.0)

    return sum(reciprocal_ranks) / len(reciprocal_ranks)

Mean Average Precision (MAP)

Considers both precision and the ranking of relevant documents.

def calculate_average_precision(retrieved_chunks: List[Dict], relevant_chunks: List[Dict]) -> float:
    """
    Calculate Average Precision for a single query
    """
    retrieved_ids = {chunk.get('id') for chunk in retrieved_chunks}
    relevant_ids = {chunk.get('id') for chunk in relevant_chunks}

    if not relevant_ids:
        return 0.0

    precisions = []
    relevant_count = 0

    for rank, chunk in enumerate(retrieved_chunks, 1):
        if chunk.get('id') in relevant_ids:
            relevant_count += 1
            precision_at_rank = relevant_count / rank
            precisions.append(precision_at_rank)

    return sum(precisions) / len(relevant_ids) if relevant_ids else 0.0

def calculate_map(queries: List[Dict], results: List[List[Dict]]) -> float:
    """
    Calculate Mean Average Precision across multiple queries
    """
    average_precisions = []

    for query, query_results in zip(queries, results):
        ap = calculate_average_precision(query_results, query.get('relevant_chunks', []))
        average_precisions.append(ap)

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

Normalized Discounted Cumulative Gain (NDCG)

Measures ranking quality with emphasis on highly relevant results.

def calculate_dcg(retrieved_chunks: List[Dict]) -> float:
    """
    Calculate Discounted Cumulative Gain
    """
    dcg = 0.0

    for rank, chunk in enumerate(retrieved_chunks, 1):
        relevance = chunk.get('relevance_score', 0)
        dcg += relevance / np.log2(rank + 1)

    return dcg

def calculate_ndcg(retrieved_chunks: List[Dict], ideal_chunks: List[Dict]) -> float:
    """
    Calculate Normalized Discounted Cumulative Gain
    """
    dcg = calculate_dcg(retrieved_chunks)
    idcg = calculate_dcg(ideal_chunks)

    if idcg == 0:
        return 0.0

    return dcg / idcg

End-to-End RAG Evaluation

Answer Quality Metrics

Factual Consistency

Measures how well the generated answer aligns with retrieved chunks.

import spacy
from transformers import pipeline

class FactualConsistencyEvaluator:
    def __init__(self):
        self.nlp = spacy.load("en_core_web_sm")
        self.nli_pipeline = pipeline("text-classification",
                                   model="roberta-large-mnli")

    def evaluate_consistency(self, answer: str, retrieved_chunks: List[str]) -> float:
        """
        Evaluate factual consistency between answer and retrieved context
        """
        if not retrieved_chunks:
            return 0.0

        # Combine retrieved chunks as context
        context = " ".join(retrieved_chunks[:3])  # Use top 3 chunks

        # Use Natural Language Inference to check consistency
        result = self.nli_pipeline(f"premise: {context} hypothesis: {answer}")

        # Extract consistency score (entailment probability)
        for item in result:
            if item['label'] == 'ENTAILMENT':
                return item['score']
            elif item['label'] == 'CONTRADICTION':
                return 1.0 - item['score']

        return 0.5  # Neutral if NLI is inconclusive

Answer Completeness

Measures how completely the answer addresses the user's query.

def evaluate_completeness(answer: str, query: str, reference_answer: str = None) -> float:
    """
    Evaluate answer completeness
    """
    # Extract key entities from query
    query_entities = extract_entities(query)
    answer_entities = extract_entities(answer)

    # Calculate entity coverage
    if not query_entities:
        return 0.5  # Neutral if no entities in query

    covered_entities = query_entities & answer_entities
    entity_coverage = len(covered_entities) / len(query_entities)

    # If reference answer is available, compare against it
    if reference_answer:
        reference_entities = extract_entities(reference_answer)
        answer_reference_overlap = len(answer_entities & reference_entities) / max(len(reference_entities), 1)
        return (entity_coverage + answer_reference_overlap) / 2

    return entity_coverage

def extract_entities(text: str) -> set:
    """
    Extract named entities from text (simplified)
    """
    # This would use a proper NER model in practice
    import re

    # Simple noun phrase extraction as placeholder
    noun_phrases = re.findall(r'\b[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*\b', text)
    return set(noun_phrases)

Response Relevance

Measures how relevant the answer is to the original query.

from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity

class RelevanceEvaluator:
    def __init__(self, model_name="all-MiniLM-L6-v2"):
        self.model = SentenceTransformer(model_name)

    def evaluate_relevance(self, query: str, answer: str) -> float:
        """
        Evaluate semantic relevance between query and answer
        """
        # Generate embeddings
        query_embedding = self.model.encode([query])
        answer_embedding = self.model.encode([answer])

        # Calculate cosine similarity
        similarity = cosine_similarity(query_embedding, answer_embedding)[0][0]

        return float(similarity)

Performance Metrics

Processing Time

import time
from dataclasses import dataclass
from typing import List, Dict

@dataclass
class PerformanceMetrics:
    total_time: float
    chunking_time: float
    embedding_time: float
    search_time: float
    generation_time: float
    throughput: float  # documents per second

class PerformanceProfiler:
    def __init__(self):
        self.timings = {}
        self.start_times = {}

    def start_timer(self, operation: str):
        self.start_times[operation] = time.time()

    def end_timer(self, operation: str):
        if operation in self.start_times:
            duration = time.time() - self.start_times[operation]
            if operation not in self.timings:
                self.timings[operation] = []
            self.timings[operation].append(duration)
            return duration
        return 0.0

    def get_performance_metrics(self, document_count: int) -> PerformanceMetrics:
        total_time = sum(sum(times) for times in self.timings.values())

        return PerformanceMetrics(
            total_time=total_time,
            chunking_time=sum(self.timings.get('chunking', [0])),
            embedding_time=sum(self.timings.get('embedding', [0])),
            search_time=sum(self.timings.get('search', [0])),
            generation_time=sum(self.timings.get('generation', [0])),
            throughput=document_count / total_time if total_time > 0 else 0
        )

Memory Usage

import psutil
import os
from typing import Dict, List

class MemoryProfiler:
    def __init__(self):
        self.process = psutil.Process(os.getpid())
        self.memory_snapshots = []

    def take_memory_snapshot(self, label: str):
        """Take a snapshot of current memory usage"""
        memory_info = self.process.memory_info()
        memory_mb = memory_info.rss / 1024 / 1024  # Convert to MB

        self.memory_snapshots.append({
            'label': label,
            'memory_mb': memory_mb,
            'timestamp': time.time()
        })

    def get_peak_memory_usage(self) -> float:
        """Get peak memory usage in MB"""
        if not self.memory_snapshots:
            return 0.0
        return max(snapshot['memory_mb'] for snapshot in self.memory_snapshots)

    def get_memory_usage_by_operation(self) -> Dict[str, float]:
        """Get memory usage breakdown by operation"""
        if not self.memory_snapshots:
            return {}

        memory_by_op = {}
        for i in range(1, len(self.memory_snapshots)):
            prev_snapshot = self.memory_snapshots[i-1]
            curr_snapshot = self.memory_snapshots[i]

            operation = curr_snapshot['label']
            memory_delta = curr_snapshot['memory_mb'] - prev_snapshot['memory_mb']

            if operation not in memory_by_op:
                memory_by_op[operation] = []
            memory_by_op[operation].append(memory_delta)

        return {op: sum(deltas) for op, deltas in memory_by_op.items()}

Evaluation Datasets

Standardized Test Sets

Question-Answer Pairs

from dataclasses import dataclass
from typing import List, Optional
import json

@dataclass
class EvaluationQuery:
    id: str
    question: str
    reference_answer: Optional[str]
    relevant_chunk_ids: List[str]
    query_type: str  # factoid, analytical, comparative
    difficulty: str  # easy, medium, hard
    domain: str  # finance, medical, legal, technical

class EvaluationDataset:
    def __init__(self, name: str):
        self.name = name
        self.queries: List[EvaluationQuery] = []
        self.documents: Dict[str, str] = {}
        self.chunks: Dict[str, Dict] = {}

    def add_query(self, query: EvaluationQuery):
        self.queries.append(query)

    def add_document(self, doc_id: str, content: str):
        self.documents[doc_id] = content

    def add_chunk(self, chunk_id: str, content: str, doc_id: str, metadata: Dict):
        self.chunks[chunk_id] = {
            'id': chunk_id,
            'content': content,
            'doc_id': doc_id,
            'metadata': metadata
        }

    def save_to_file(self, filepath: str):
        data = {
            'name': self.name,
            'queries': [
                {
                    'id': q.id,
                    'question': q.question,
                    'reference_answer': q.reference_answer,
                    'relevant_chunk_ids': q.relevant_chunk_ids,
                    'query_type': q.query_type,
                    'difficulty': q.difficulty,
                    'domain': q.domain
                }
                for q in self.queries
            ],
            'documents': self.documents,
            'chunks': self.chunks
        }

        with open(filepath, 'w') as f:
            json.dump(data, f, indent=2)

    @classmethod
    def load_from_file(cls, filepath: str):
        with open(filepath, 'r') as f:
            data = json.load(f)

        dataset = cls(data['name'])
        dataset.documents = data['documents']
        dataset.chunks = data['chunks']

        for q_data in data['queries']:
            query = EvaluationQuery(
                id=q_data['id'],
                question=q_data['question'],
                reference_answer=q_data.get('reference_answer'),
                relevant_chunk_ids=q_data['relevant_chunk_ids'],
                query_type=q_data['query_type'],
                difficulty=q_data['difficulty'],
                domain=q_data['domain']
            )
            dataset.add_query(query)

        return dataset

Dataset Generation

Synthetic Query Generation

import random
from typing import List, Dict

class SyntheticQueryGenerator:
    def __init__(self):
        self.query_templates = {
            'factoid': [
                "What is {concept}?",
                "When did {event} occur?",
                "Who developed {technology}?",
                "How many {items} are mentioned?",
                "What is the value of {metric}?"
            ],
            'analytical': [
                "Compare and contrast {concept1} and {concept2}.",
                "Analyze the impact of {concept} on {domain}.",
                "What are the advantages and disadvantages of {technology}?",
                "Explain the relationship between {concept1} and {concept2}.",
                "Evaluate the effectiveness of {approach} for {problem}."
            ],
            'comparative': [
                "Which is better: {option1} or {option2}?",
                "How does {method1} differ from {method2}?",
                "Compare the performance of {system1} and {system2}.",
                "What are the key differences between {approach1} and {approach2}?"
            ]
        }

    def generate_queries_from_chunks(self, chunks: List[Dict], num_queries: int = 100) -> List[EvaluationQuery]:
        """Generate synthetic queries from document chunks"""
        queries = []

        # Extract entities and concepts from chunks
        entities = self._extract_entities_from_chunks(chunks)

        for i in range(num_queries):
            query_type = random.choice(['factoid', 'analytical', 'comparative'])
            template = random.choice(self.query_templates[query_type])

            # Fill template with extracted entities
            query_text = self._fill_template(template, entities)

            # Find relevant chunks for this query
            relevant_chunks = self._find_relevant_chunks(query_text, chunks)

            query = EvaluationQuery(
                id=f"synthetic_{i}",
                question=query_text,
                reference_answer=None,  # Would need generation model
                relevant_chunk_ids=[chunk['id'] for chunk in relevant_chunks],
                query_type=query_type,
                difficulty=random.choice(['easy', 'medium', 'hard']),
                domain='synthetic'
            )

            queries.append(query)

        return queries

    def _extract_entities_from_chunks(self, chunks: List[Dict]) -> Dict[str, List[str]]:
        """Extract entities, concepts, and relationships from chunks"""
        # This would use proper NER in practice
        entities = {
            'concepts': [],
            'technologies': [],
            'methods': [],
            'metrics': [],
            'events': []
        }

        for chunk in chunks:
            content = chunk['content']
            # Simplified entity extraction
            words = content.split()
            entities['concepts'].extend([word for word in words if len(word) > 6])
            entities['technologies'].extend([word for word in words if 'technology' in word.lower()])
            entities['methods'].extend([word for word in words if 'method' in word.lower()])
            entities['metrics'].extend([word for word in words if '%' in word or '$' in word])

        # Remove duplicates and limit
        for key in entities:
            entities[key] = list(set(entities[key]))[:50]

        return entities

    def _fill_template(self, template: str, entities: Dict[str, List[str]]) -> str:
        """Fill query template with random entities"""
        import re

        def replace_placeholder(match):
            placeholder = match.group(1)

            # Map placeholders to entity types
            entity_mapping = {
                'concept': 'concepts',
                'concept1': 'concepts',
                'concept2': 'concepts',
                'technology': 'technologies',
                'method': 'methods',
                'method1': 'methods',
                'method2': 'methods',
                'metric': 'metrics',
                'event': 'events',
                'items': 'concepts',
                'option1': 'concepts',
                'option2': 'concepts',
                'approach': 'methods',
                'problem': 'concepts',
                'domain': 'concepts',
                'system1': 'concepts',
                'system2': 'concepts'
            }

            entity_type = entity_mapping.get(placeholder, 'concepts')
            available_entities = entities.get(entity_type, ['something'])

            if available_entities:
                return random.choice(available_entities)
            else:
                return 'something'

        return re.sub(r'\{(\w+)\}', replace_placeholder, template)

    def _find_relevant_chunks(self, query: str, chunks: List[Dict], k: int = 3) -> List[Dict]:
        """Find chunks most relevant to the query"""
        # Simple keyword matching for synthetic generation
        query_words = set(query.lower().split())

        chunk_scores = []
        for chunk in chunks:
            chunk_words = set(chunk['content'].lower().split())
            overlap = len(query_words & chunk_words)
            chunk_scores.append((overlap, chunk))

        # Sort by overlap and return top k
        chunk_scores.sort(key=lambda x: x[0], reverse=True)
        return [chunk for _, chunk in chunk_scores[:k]]

A/B Testing Framework

Statistical Significance Testing

import numpy as np
from scipy import stats
from typing import List, Dict, Tuple

class ABTestAnalyzer:
    def __init__(self):
        self.significance_level = 0.05

    def compare_metrics(self, control_metrics: List[float],
                       treatment_metrics: List[float],
                       metric_name: str) -> Dict:
        """
        Compare metrics between control and treatment groups
        """
        control_mean = np.mean(control_metrics)
        treatment_mean = np.mean(treatment_metrics)

        control_std = np.std(control_metrics)
        treatment_std = np.std(treatment_metrics)

        # Perform t-test
        t_statistic, p_value = stats.ttest_ind(control_metrics, treatment_metrics)

        # Calculate effect size (Cohen's d)
        pooled_std = np.sqrt(((len(control_metrics) - 1) * control_std**2 +
                             (len(treatment_metrics) - 1) * treatment_std**2) /
                            (len(control_metrics) + len(treatment_metrics) - 2))

        cohens_d = (treatment_mean - control_mean) / pooled_std if pooled_std > 0 else 0

        # Determine significance
        is_significant = p_value < self.significance_level

        return {
            'metric_name': metric_name,
            'control_mean': control_mean,
            'treatment_mean': treatment_mean,
            'absolute_difference': treatment_mean - control_mean,
            'relative_difference': ((treatment_mean - control_mean) / control_mean * 100) if control_mean != 0 else 0,
            'control_std': control_std,
            'treatment_std': treatment_std,
            't_statistic': t_statistic,
            'p_value': p_value,
            'is_significant': is_significant,
            'effect_size': cohens_d,
            'significance_level': self.significance_level
        }

    def analyze_ab_test_results(self,
                               control_results: Dict[str, List[float]],
                               treatment_results: Dict[str, List[float]]) -> Dict:
        """
        Analyze A/B test results across multiple metrics
        """
        analysis_results = {}

        # Ensure both dictionaries have the same keys
        all_metrics = set(control_results.keys()) & set(treatment_results.keys())

        for metric in all_metrics:
            if metric in control_results and metric in treatment_results:
                analysis_results[metric] = self.compare_metrics(
                    control_results[metric],
                    treatment_results[metric],
                    metric
                )

        # Calculate overall summary
        significant_improvements = sum(1 for result in analysis_results.values()
                                     if result['is_significant'] and result['relative_difference'] > 0)
        significant_degradations = sum(1 for result in analysis_results.values()
                                      if result['is_significant'] and result['relative_difference'] < 0)

        analysis_results['summary'] = {
            'total_metrics_compared': len(analysis_results),
            'significant_improvements': significant_improvements,
            'significant_degradations': significant_degradations,
            'no_significant_change': len(analysis_results) - significant_improvements - significant_degradations
        }

        return analysis_results

Automated Evaluation Pipeline

End-to-End Evaluation

class ChunkingEvaluationPipeline:
    def __init__(self, strategies: Dict[str, Any], dataset: EvaluationDataset):
        self.strategies = strategies
        self.dataset = dataset
        self.results = {}
        self.profiler = PerformanceProfiler()
        self.memory_profiler = MemoryProfiler()

    def run_evaluation(self) -> Dict:
        """Run comprehensive evaluation of all strategies"""
        evaluation_results = {}

        for strategy_name, strategy in self.strategies.items():
            print(f"Evaluating strategy: {strategy_name}")

            # Reset profilers for each strategy
            self.profiler = PerformanceProfiler()
            self.memory_profiler = MemoryProfiler()

            # Evaluate strategy
            strategy_results = self._evaluate_strategy(strategy, strategy_name)
            evaluation_results[strategy_name] = strategy_results

        # Compare strategies
        comparison_results = self._compare_strategies(evaluation_results)

        return {
            'individual_results': evaluation_results,
            'comparison': comparison_results,
            'recommendations': self._generate_recommendations(comparison_results)
        }

    def _evaluate_strategy(self, strategy: Any, strategy_name: str) -> Dict:
        """Evaluate a single chunking strategy"""
        results = {
            'strategy_name': strategy_name,
            'retrieval_metrics': {},
            'quality_metrics': {},
            'performance_metrics': {}
        }

        # Track memory usage
        self.memory_profiler.take_memory_snapshot(f"{strategy_name}_start")

        # Process all documents
        self.profiler.start_timer('total_processing')

        all_chunks = {}
        for doc_id, content in self.dataset.documents.items():
            self.profiler.start_timer('chunking')
            chunks = strategy.chunk(content)
            self.profiler.end_timer('chunking')

            all_chunks[doc_id] = chunks

        self.memory_profiler.take_memory_snapshot(f"{strategy_name}_after_chunking")

        # Generate embeddings for chunks
        self.profiler.start_timer('embedding')
        chunk_embeddings = self._generate_embeddings(all_chunks)
        self.profiler.end_timer('embedding')

        self.memory_profiler.take_memory_snapshot(f"{strategy_name}_after_embedding")

        # Evaluate retrieval performance
        retrieval_results = self._evaluate_retrieval(all_chunks, chunk_embeddings)
        results['retrieval_metrics'] = retrieval_results

        # Evaluate chunk quality
        quality_results = self._evaluate_chunk_quality(all_chunks)
        results['quality_metrics'] = quality_results

        # Get performance metrics
        self.profiler.end_timer('total_processing')
        performance_metrics = self.profiler.get_performance_metrics(len(self.dataset.documents))
        results['performance_metrics'] = performance_metrics.__dict__

        # Get memory metrics
        self.memory_profiler.take_memory_snapshot(f"{strategy_name}_end")
        results['memory_metrics'] = {
            'peak_memory_mb': self.memory_profiler.get_peak_memory_usage(),
            'memory_by_operation': self.memory_profiler.get_memory_usage_by_operation()
        }

        return results

    def _evaluate_retrieval(self, all_chunks: Dict, chunk_embeddings: Dict) -> Dict:
        """Evaluate retrieval performance"""
        retrieval_metrics = {
            'precision': [],
            'recall': [],
            'f1_score': [],
            'mrr': [],
            'map': []
        }

        for query in self.dataset.queries:
            # Perform retrieval
            self.profiler.start_timer('search')
            retrieved_chunks = self._retrieve_chunks(query.question, chunk_embeddings, k=10)
            self.profiler.end_timer('search')

            # Get relevant chunks for this query
            relevant_chunk_ids = set(query.relevant_chunk_ids)
            relevant_chunks = [chunk for chunk in retrieved_chunks
                             if chunk.get('id') in relevant_chunk_ids]

            # Calculate metrics
            precision = calculate_precision(retrieved_chunks, relevant_chunks)
            recall = calculate_recall(retrieved_chunks, relevant_chunks)
            f1 = calculate_f1_score(precision, recall)

            retrieval_metrics['precision'].append(precision)
            retrieval_metrics['recall'].append(recall)
            retrieval_metrics['f1_score'].append(f1)

        # Calculate averages
        return {metric: np.mean(values) for metric, values in retrieval_metrics.items()}

    def _evaluate_chunk_quality(self, all_chunks: Dict) -> Dict:
        """Evaluate quality of generated chunks"""
        quality_assessor = ChunkQualityAssessor()
        quality_scores = []

        for doc_id, chunks in all_chunks.items():
            # Analyze document
            content = self.dataset.documents[doc_id]
            analyzer = DocumentAnalyzer()
            analysis = analyzer.analyze(content)

            # Assess chunk quality
            scores = quality_assessor.assess_chunks(chunks, analysis)
            quality_scores.append(scores)

        # Aggregate quality scores
        if quality_scores:
            avg_scores = {}
            for metric in quality_scores[0].keys():
                avg_scores[metric] = np.mean([scores[metric] for scores in quality_scores])
            return avg_scores

        return {}

    def _compare_strategies(self, evaluation_results: Dict) -> Dict:
        """Compare performance across strategies"""
        ab_analyzer = ABTestAnalyzer()

        comparison = {}

        # Compare each metric across strategies
        strategy_names = list(evaluation_results.keys())

        for i in range(len(strategy_names)):
            for j in range(i + 1, len(strategy_names)):
                strategy1 = strategy_names[i]
                strategy2 = strategy_names[j]

                comparison_key = f"{strategy1}_vs_{strategy2}"
                comparison[comparison_key] = {}

                # Compare retrieval metrics
                for metric in ['precision', 'recall', 'f1_score']:
                    if (metric in evaluation_results[strategy1]['retrieval_metrics'] and
                        metric in evaluation_results[strategy2]['retrieval_metrics']):

                        comparison[comparison_key][f"retrieval_{metric}"] = ab_analyzer.compare_metrics(
                            [evaluation_results[strategy1]['retrieval_metrics'][metric]],
                            [evaluation_results[strategy2]['retrieval_metrics'][metric]],
                            f"retrieval_{metric}"
                        )

        return comparison

    def _generate_recommendations(self, comparison_results: Dict) -> Dict:
        """Generate recommendations based on evaluation results"""
        recommendations = {
            'best_overall': None,
            'best_for_precision': None,
            'best_for_recall': None,
            'best_for_performance': None,
            'trade_offs': []
        }

        # This would analyze the comparison results and generate specific recommendations
        # Implementation depends on specific use case requirements

        return recommendations

    def _generate_embeddings(self, all_chunks: Dict) -> Dict:
        """Generate embeddings for all chunks"""
        # This would use the actual embedding model
        # Placeholder implementation
        embeddings = {}

        for doc_id, chunks in all_chunks.items():
            embeddings[doc_id] = []
            for chunk in chunks:
                # Generate embedding for chunk content
                embedding = np.random.rand(384)  # Placeholder
                embeddings[doc_id].append({
                    'chunk': chunk,
                    'embedding': embedding
                })

        return embeddings

    def _retrieve_chunks(self, query: str, chunk_embeddings: Dict, k: int = 10) -> List[Dict]:
        """Retrieve most relevant chunks for a query"""
        # This would use actual similarity search
        # Placeholder implementation
        all_chunks = []

        for doc_embeddings in chunk_embeddings.values():
            for chunk_data in doc_embeddings:
                all_chunks.append(chunk_data['chunk'])

        # Simple random selection as placeholder
        selected = random.sample(all_chunks, min(k, len(all_chunks)))

        return selected

This comprehensive evaluation framework provides the tools needed to thoroughly assess chunking strategies across multiple dimensions: retrieval effectiveness, answer quality, system performance, and statistical significance. The modular design allows for easy extension and customization based on specific requirements and use cases.

Complete Implementation Guidelines

This document provides comprehensive implementation guidance for building effective chunking systems.

System Architecture

Core Components

Document Processor
├── Ingestion Layer
│   ├── Document Type Detection
│   ├── Format Parsing (PDF, HTML, Markdown, etc.)
│   └── Content Extraction
├── Analysis Layer
│   ├── Structure Analysis
│   ├── Content Type Identification
│   └── Complexity Assessment
├── Strategy Selection Layer
│   ├── Rule-based Selection
│   ├── ML-based Prediction
│   └── Adaptive Configuration
├── Chunking Layer
│   ├── Strategy Implementation
│   ├── Parameter Optimization
│   └── Quality Validation
└── Output Layer
    ├── Chunk Metadata Generation
    ├── Embedding Integration
    └── Storage Preparation

Pre-processing Pipeline

Document Analysis Framework

from dataclasses import dataclass
from typing import List, Dict, Any
import re

@dataclass
class DocumentAnalysis:
    doc_type: str
    structure_score: float  # 0-1, higher means more structured
    complexity_score: float  # 0-1, higher means more complex
    content_types: List[str]
    language: str
    estimated_tokens: int
    has_multimodal: bool

class DocumentAnalyzer:
    def __init__(self):
        self.structure_patterns = {
            'markdown': [r'^#+\s', r'^\*\*.*\*\*$', r'^\* ', r'^\d+\. '],
            'html': [r'<h[1-6]>', r'<p>', r'<div>', r'<table>'],
            'latex': [r'\\section', r'\\subsection', r'\\begin\{', r'\\end\{'],
            'academic': [r'^\d+\.', r'^\d+\.\d+', r'^[A-Z]\.', r'^Figure \d+']
        }

    def analyze(self, content: str) -> DocumentAnalysis:
        doc_type = self.detect_document_type(content)
        structure_score = self.calculate_structure_score(content, doc_type)
        complexity_score = self.calculate_complexity_score(content)
        content_types = self.identify_content_types(content)
        language = self.detect_language(content)
        estimated_tokens = self.estimate_tokens(content)
        has_multimodal = self.detect_multimodal_content(content)

        return DocumentAnalysis(
            doc_type=doc_type,
            structure_score=structure_score,
            complexity_score=complexity_score,
            content_types=content_types,
            language=language,
            estimated_tokens=estimated_tokens,
            has_multimodal=has_multimodal
        )

    def detect_document_type(self, content: str) -> str:
        content_lower = content.lower()

        if '<html' in content_lower or '<body' in content_lower:
            return 'html'
        elif '#' in content and '##' in content:
            return 'markdown'
        elif '\\documentclass' in content_lower or '\\begin{' in content_lower:
            return 'latex'
        elif any(keyword in content_lower for keyword in ['abstract', 'introduction', 'conclusion', 'references']):
            return 'academic'
        elif 'def ' in content or 'class ' in content or 'function ' in content_lower:
            return 'code'
        else:
            return 'plain'

    def calculate_structure_score(self, content: str, doc_type: str) -> float:
        patterns = self.structure_patterns.get(doc_type, [])
        if not patterns:
            return 0.5  # Default for unstructured content

        line_count = len(content.split('\n'))
        structured_lines = 0

        for line in content.split('\n'):
            for pattern in patterns:
                if re.search(pattern, line.strip()):
                    structured_lines += 1
                    break

        return min(structured_lines / max(line_count, 1), 1.0)

    def calculate_complexity_score(self, content: str) -> float:
        # Factors that increase complexity
        avg_sentence_length = self.calculate_avg_sentence_length(content)
        vocabulary_richness = self.calculate_vocabulary_richness(content)
        nested_structure = self.detect_nested_structure(content)

        # Normalize and combine
        complexity = (
            min(avg_sentence_length / 30, 1.0) * 0.3 +
            vocabulary_richness * 0.4 +
            nested_structure * 0.3
        )

        return min(complexity, 1.0)

    def identify_content_types(self, content: str) -> List[str]:
        types = []

        if '```' in content or 'def ' in content or 'function ' in content.lower():
            types.append('code')
        if '|' in content and '\n' in content:
            types.append('tables')
        if re.search(r'\!\[.*\]\(.*\)', content):
            types.append('images')
        if re.search(r'http[s]?://', content):
            types.append('links')
        if re.search(r'\d+\.\d+', content) or re.search(r'\$\d', content):
            types.append('numbers')

        return types if types else ['text']

    def detect_language(self, content: str) -> str:
        # Simple language detection - can be enhanced with proper language detection libraries
        if re.search(r'[\u4e00-\u9fff]', content):
            return 'chinese'
        elif re.search(r'[u0600-\u06ff]', content):
            return 'arabic'
        elif re.search(r'[u0400-\u04ff]', content):
            return 'russian'
        else:
            return 'english'  # Default assumption

    def estimate_tokens(self, content: str) -> int:
        # Rough estimation - actual tokenization varies by model
        word_count = len(content.split())
        return int(word_count * 1.3)  # Average tokens per word

    def detect_multimodal_content(self, content: str) -> bool:
        multimodal_indicators = [
            r'\!\[.*\]\(.*\)',  # Images
            r'<iframe',        # Embedded content
            r'<object',        # Embedded objects
            r'<embed',         # Embedded media
        ]

        return any(re.search(pattern, content) for pattern in multimodal_indicators)

    def calculate_avg_sentence_length(self, content: str) -> float:
        sentences = re.split(r'[.!?]+', content)
        sentences = [s.strip() for s in sentences if s.strip()]
        if not sentences:
            return 0
        return sum(len(s.split()) for s in sentences) / len(sentences)

    def calculate_vocabulary_richness(self, content: str) -> float:
        words = content.lower().split()
        if not words:
            return 0
        unique_words = set(words)
        return len(unique_words) / len(words)

    def detect_nested_structure(self, content: str) -> float:
        # Detect nested lists, indented content, etc.
        lines = content.split('\n')
        indented_lines = 0

        for line in lines:
            if line.strip() and line.startswith(' '):
                indented_lines += 1

        return indented_lines / max(len(lines), 1)

Strategy Selection Engine

from abc import ABC, abstractmethod
from typing import Dict, Any

class ChunkingStrategy(ABC):
    @abstractmethod
    def chunk(self, content: str, analysis: DocumentAnalysis) -> List[Dict[str, Any]]:
        pass

class StrategySelector:
    def __init__(self):
        self.strategies = {
            'fixed_size': FixedSizeStrategy(),
            'recursive': RecursiveStrategy(),
            'structure_aware': StructureAwareStrategy(),
            'semantic': SemanticStrategy(),
            'adaptive': AdaptiveStrategy()
        }

    def select_strategy(self, analysis: DocumentAnalysis) -> str:
        # Rule-based selection logic
        if analysis.structure_score > 0.8 and analysis.doc_type in ['markdown', 'html', 'latex']:
            return 'structure_aware'
        elif analysis.complexity_score > 0.7 and analysis.estimated_tokens < 10000:
            return 'semantic'
        elif analysis.doc_type == 'code':
            return 'structure_aware'
        elif analysis.structure_score < 0.3:
            return 'fixed_size'
        elif analysis.complexity_score > 0.5:
            return 'recursive'
        else:
            return 'adaptive'

    def get_strategy(self, analysis: DocumentAnalysis) -> ChunkingStrategy:
        strategy_name = self.select_strategy(analysis)
        return self.strategies[strategy_name]

# Example strategy implementations
class FixedSizeStrategy(ChunkingStrategy):
    def __init__(self, default_size=512, default_overlap=50):
        self.default_size = default_size
        self.default_overlap = default_overlap

    def chunk(self, content: str, analysis: DocumentAnalysis) -> List[Dict[str, Any]]:
        # Adjust parameters based on analysis
        if analysis.complexity_score > 0.7:
            chunk_size = 1024
        elif analysis.complexity_score < 0.3:
            chunk_size = 256
        else:
            chunk_size = self.default_size

        overlap = int(chunk_size * 0.1)  # 10% overlap

        # Implementation here...
        return self._fixed_size_chunk(content, chunk_size, overlap)

    def _fixed_size_chunk(self, content: str, chunk_size: int, overlap: int) -> List[Dict[str, Any]]:
        # Implementation using RecursiveCharacterTextSplitter or custom logic
        pass

class AdaptiveStrategy(ChunkingStrategy):
    def chunk(self, content: str, analysis: DocumentAnalysis) -> List[Dict[str, Any]]:
        # Combine multiple strategies based on content characteristics
        if analysis.structure_score > 0.6:
            # Use structure-aware for structured parts
            structured_chunks = self._chunk_structured_parts(content, analysis)
        else:
            # Use fixed-size for unstructured parts
            unstructured_chunks = self._chunk_unstructured_parts(content, analysis)

        # Merge and optimize
        return self._merge_chunks(structured_chunks + unstructured_chunks)

    def _chunk_structured_parts(self, content: str, analysis: DocumentAnalysis) -> List[Dict[str, Any]]:
        # Implementation for structured content
        pass

    def _chunk_unstructured_parts(self, content: str, analysis: DocumentAnalysis) -> List[Dict[str, Any]]:
        # Implementation for unstructured content
        pass

    def _merge_chunks(self, chunks: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
        # Implementation for merging and optimizing chunks
        pass

Quality Assurance Framework

Chunk Quality Metrics

from typing import List, Dict, Any
import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

class ChunkQualityAssessor:
    def __init__(self):
        self.quality_weights = {
            'coherence': 0.3,
            'completeness': 0.25,
            'size_appropriateness': 0.2,
            'semantic_similarity': 0.15,
            'boundary_quality': 0.1
        }

    def assess_chunks(self, chunks: List[Dict[str, Any]], analysis: DocumentAnalysis) -> Dict[str, float]:
        scores = {}

        # Coherence: Do chunks make sense on their own?
        scores['coherence'] = self._assess_coherence(chunks)

        # Completeness: Do chunks preserve important information?
        scores['completeness'] = self._assess_completeness(chunks, analysis)

        # Size appropriateness: Are chunks within optimal size range?
        scores['size_appropriateness'] = self._assess_size(chunks)

        # Semantic similarity: Are chunks thematically consistent?
        scores['semantic_similarity'] = self._assess_semantic_consistency(chunks)

        # Boundary quality: Are chunk boundaries placed well?
        scores['boundary_quality'] = self._assess_boundary_quality(chunks)

        # Calculate overall quality score
        overall_score = sum(
            score * self.quality_weights[metric]
            for metric, score in scores.items()
        )

        scores['overall'] = overall_score
        return scores

    def _assess_coherence(self, chunks: List[Dict[str, Any]]) -> float:
        # Simple heuristic-based coherence assessment
        coherence_scores = []

        for chunk in chunks:
            content = chunk['content']

            # Check for complete sentences
            sentences = re.split(r'[.!?]+', content)
            complete_sentences = sum(1 for s in sentences if s.strip())
            coherence = complete_sentences / max(len(sentences), 1)

            coherence_scores.append(coherence)

        return np.mean(coherence_scores)

    def _assess_completeness(self, chunks: List[Dict[str, Any]], analysis: DocumentAnalysis) -> float:
        # Check if important structural elements are preserved
        if analysis.doc_type in ['markdown', 'html']:
            return self._assess_structure_preservation(chunks, analysis)
        else:
            return self._assess_content_preservation(chunks)

    def _assess_structure_preservation(self, chunks: List[Dict[str, Any]], analysis: DocumentAnalysis) -> float:
        # Check if headings, lists, and other structural elements are preserved
        preserved_elements = 0
        total_elements = 0

        for chunk in chunks:
            content = chunk['content']

            # Count preserved structural elements
            headings = len(re.findall(r'^#+\s', content, re.MULTILINE))
            lists = len(re.findall(r'^\s*[-*+]\s', content, re.MULTILINE))

            preserved_elements += headings + lists
            total_elements += 1  # Simplified count

        return preserved_elements / max(total_elements, 1)

    def _assess_content_preservation(self, chunks: List[Dict[str, Any]]) -> float:
        # Simple check based on content ratio
        total_content = ''.join(chunk['content'] for chunk in chunks)
        # This would need comparison with original content
        return 0.8  # Placeholder

    def _assess_size(self, chunks: List[Dict[str, Any]]) -> float:
        optimal_min = 100  # tokens
        optimal_max = 1000  # tokens

        size_scores = []
        for chunk in chunks:
            token_count = self._estimate_tokens(chunk['content'])
            if optimal_min <= token_count <= optimal_max:
                score = 1.0
            elif token_count < optimal_min:
                score = token_count / optimal_min
            else:
                score = max(0, 1 - (token_count - optimal_max) / optimal_max)

            size_scores.append(score)

        return np.mean(size_scores)

    def _assess_semantic_consistency(self, chunks: List[Dict[str, Any]]) -> float:
        # This would require embedding models for actual implementation
        # Placeholder implementation
        return 0.7

    def _assess_boundary_quality(self, chunks: List[Dict[str, Any]]) -> float:
        # Check if boundaries don't split important content
        boundary_scores = []

        for i, chunk in enumerate(chunks):
            content = chunk['content']

            # Check for incomplete sentences at boundaries
            if not content.strip().endswith(('.', '!', '?', '>', '}')):
                boundary_scores.append(0.5)
            else:
                boundary_scores.append(1.0)

        return np.mean(boundary_scores)

    def _estimate_tokens(self, content: str) -> int:
        # Simple token estimation
        return len(content.split()) * 4 // 3  # Rough approximation

Error Handling and Edge Cases

Robust Error Handling

import logging
from typing import Optional, List
from dataclasses import dataclass

@dataclass
class ChunkingError:
    error_type: str
    message: str
    chunk_index: Optional[int] = None
    recovery_action: Optional[str] = None

class ChunkingErrorHandler:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
        self.error_handlers = {
            'empty_content': self._handle_empty_content,
            'oversized_chunk': self._handle_oversized_chunk,
            'encoding_error': self._handle_encoding_error,
            'memory_error': self._handle_memory_error,
            'structure_parsing_error': self._handle_structure_parsing_error
        }

    def handle_error(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:
        error_type = self._classify_error(error)
        handler = self.error_handlers.get(error_type, self._handle_generic_error)
        return handler(error, context)

    def _classify_error(self, error: Exception) -> str:
        if isinstance(error, ValueError) and 'empty' in str(error).lower():
            return 'empty_content'
        elif isinstance(error, MemoryError):
            return 'memory_error'
        elif isinstance(error, UnicodeError):
            return 'encoding_error'
        elif 'too large' in str(error).lower():
            return 'oversized_chunk'
        elif 'parsing' in str(error).lower():
            return 'structure_parsing_error'
        else:
            return 'generic_error'

    def _handle_empty_content(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:
        self.logger.warning(f"Empty content encountered: {error}")
        return ChunkingError(
            error_type='empty_content',
            message=str(error),
            recovery_action='skip_empty_content'
        )

    def _handle_oversized_chunk(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:
        self.logger.warning(f"Oversized chunk detected: {error}")
        return ChunkingError(
            error_type='oversized_chunk',
            message=str(error),
            chunk_index=context.get('chunk_index'),
            recovery_action='reduce_chunk_size'
        )

    def _handle_encoding_error(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:
        self.logger.error(f"Encoding error: {error}")
        return ChunkingError(
            error_type='encoding_error',
            message=str(error),
            recovery_action='fallback_encoding'
        )

    def _handle_memory_error(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:
        self.logger.error(f"Memory error during chunking: {error}")
        return ChunkingError(
            error_type='memory_error',
            message=str(error),
            recovery_action='process_in_batches'
        )

    def _handle_structure_parsing_error(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:
        self.logger.warning(f"Structure parsing failed: {error}")
        return ChunkingError(
            error_type='structure_parsing_error',
            message=str(error),
            recovery_action='fallback_to_fixed_size'
        )

    def _handle_generic_error(self, error: Exception, context: Dict[str, Any]) -> ChunkingError:
        self.logger.error(f"Unexpected error during chunking: {error}")
        return ChunkingError(
            error_type='generic_error',
            message=str(error),
            recovery_action='skip_and_continue'
        )

Performance Optimization

Caching and Memoization

import hashlib
import pickle
from functools import lru_cache
from typing import Dict, Any, Optional
import redis
import json

class ChunkingCache:
    def __init__(self, redis_url: Optional[str] = None):
        if redis_url:
            self.redis_client = redis.from_url(redis_url)
        else:
            self.redis_client = None
        self.local_cache = {}

    def _generate_cache_key(self, content: str, strategy: str, params: Dict[str, Any]) -> str:
        content_hash = hashlib.md5(content.encode()).hexdigest()
        params_str = json.dumps(params, sort_keys=True)
        params_hash = hashlib.md5(params_str.encode()).hexdigest()
        return f"chunking:{strategy}:{content_hash}:{params_hash}"

    def get(self, content: str, strategy: str, params: Dict[str, Any]) -> Optional[List[Dict[str, Any]]]:
        cache_key = self._generate_cache_key(content, strategy, params)

        # Try local cache first
        if cache_key in self.local_cache:
            return self.local_cache[cache_key]

        # Try Redis cache
        if self.redis_client:
            try:
                cached_data = self.redis_client.get(cache_key)
                if cached_data:
                    chunks = pickle.loads(cached_data)
                    self.local_cache[cache_key] = chunks  # Cache locally too
                    return chunks
            except Exception as e:
                logging.warning(f"Redis cache error: {e}")

        return None

    def set(self, content: str, strategy: str, params: Dict[str, Any], chunks: List[Dict[str, Any]]) -> None:
        cache_key = self._generate_cache_key(content, strategy, params)

        # Store in local cache
        self.local_cache[cache_key] = chunks

        # Store in Redis cache
        if self.redis_client:
            try:
                cached_data = pickle.dumps(chunks)
                self.redis_client.setex(cache_key, 3600, cached_data)  # 1 hour TTL
            except Exception as e:
                logging.warning(f"Redis cache set error: {e}")

    def clear_local_cache(self):
        self.local_cache.clear()

    def clear_redis_cache(self):
        if self.redis_client:
            pattern = "chunking:*"
            keys = self.redis_client.keys(pattern)
            if keys:
                self.redis_client.delete(*keys)

Batch Processing

import asyncio
import concurrent.futures
from typing import List, Callable, Any

class BatchChunkingProcessor:
    def __init__(self, max_workers: int = 4, batch_size: int = 10):
        self.max_workers = max_workers
        self.batch_size = batch_size

    def process_documents_batch(self, documents: List[str],
                               chunking_function: Callable[[str], List[Dict[str, Any]]]) -> List[List[Dict[str, Any]]]:
        """Process multiple documents in parallel"""
        results = []

        # Process in batches to avoid memory issues
        for i in range(0, len(documents), self.batch_size):
            batch = documents[i:i + self.batch_size]

            with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
                future_to_doc = {
                    executor.submit(chunking_function, doc): doc
                    for doc in batch
                }

                batch_results = []
                for future in concurrent.futures.as_completed(future_to_doc):
                    try:
                        chunks = future.result()
                        batch_results.append(chunks)
                    except Exception as e:
                        logging.error(f"Error processing document: {e}")
                        batch_results.append([])  # Empty result for failed processing

                results.extend(batch_results)

        return results

    async def process_documents_async(self, documents: List[str],
                                     chunking_function: Callable[[str], List[Dict[str, Any]]]) -> List[List[Dict[str, Any]]]:
        """Process documents asynchronously"""
        semaphore = asyncio.Semaphore(self.max_workers)

        async def process_single_document(doc: str) -> List[Dict[str, Any]]:
            async with semaphore:
                # Run the synchronous chunking function in an executor
                loop = asyncio.get_event_loop()
                return await loop.run_in_executor(None, chunking_function, doc)

        tasks = [process_single_document(doc) for doc in documents]
        return await asyncio.gather(*tasks, return_exceptions=True)

Monitoring and Observability

Metrics Collection

import time
from dataclasses import dataclass
from typing import Dict, Any, List
from collections import defaultdict

@dataclass
class ChunkingMetrics:
    total_documents: int
    total_chunks: int
    avg_chunk_size: float
    processing_time: float
    memory_usage: float
    error_count: int
    strategy_distribution: Dict[str, int]

class MetricsCollector:
    def __init__(self):
        self.metrics = defaultdict(list)
        self.start_time = None

    def start_timing(self):
        self.start_time = time.time()

    def end_timing(self) -> float:
        if self.start_time:
            duration = time.time() - self.start_time
            self.metrics['processing_time'].append(duration)
            self.start_time = None
            return duration
        return 0.0

    def record_chunk_count(self, count: int):
        self.metrics['chunk_count'].append(count)

    def record_chunk_size(self, size: int):
        self.metrics['chunk_size'].append(size)

    def record_strategy_usage(self, strategy: str):
        self.metrics['strategy'][strategy] = self.metrics['strategy'].get(strategy, 0) + 1

    def record_error(self, error_type: str):
        self.metrics['errors'].append(error_type)

    def record_memory_usage(self, memory_mb: float):
        self.metrics['memory_usage'].append(memory_mb)

    def get_summary(self) -> ChunkingMetrics:
        return ChunkingMetrics(
            total_documents=len(self.metrics['processing_time']),
            total_chunks=sum(self.metrics['chunk_count']),
            avg_chunk_size=sum(self.metrics['chunk_size']) / max(len(self.metrics['chunk_size']), 1),
            processing_time=sum(self.metrics['processing_time']),
            memory_usage=sum(self.metrics['memory_usage']) / max(len(self.metrics['memory_usage']), 1),
            error_count=len(self.metrics['errors']),
            strategy_distribution=dict(self.metrics['strategy'])
        )

    def reset(self):
        self.metrics.clear()
        self.start_time = None

This implementation guide provides a comprehensive foundation for building robust, scalable chunking systems that can handle various document types and use cases while maintaining high quality and performance.

Key Research Papers and Findings

This document summarizes important research papers and findings related to chunking strategies for RAG systems.

Seminal Papers

"Reconstructing Context: Evaluating Advanced Chunking Strategies for RAG" (arXiv:2504.19754)

Key Findings:

• Page-level chunking achieved highest average accuracy (0.648) with lowest variance across different query types
• Optimal chunk size varies significantly by document type and query complexity
• Factoid queries perform better with smaller chunks (256-512 tokens)
• Complex analytical queries benefit from larger chunks (1024+ tokens)

Methodology:

• Evaluated 7 different chunking strategies across multiple document types
• Tested with both factoid and analytical queries
• Measured end-to-end RAG performance

Practical Implications:

• Start with page-level chunking for general-purpose RAG systems
• Adapt chunk size based on expected query patterns
• Consider hybrid approaches for mixed query types

"Lost in the Middle: How Language Models Use Long Contexts"

Key Findings:

• Language models tend to pay more attention to information at the beginning and end of context
• Information in the middle of long contexts is often ignored
• Performance degradation is most severe for centrally located information

Practical Implications:

• Place most important information at chunk boundaries
• Consider chunk overlap to ensure important context appears multiple times
• Use ranking to prioritize relevant chunks for inclusion in context

"Grounded Language Learning in a Simulated 3D World"

Related Concepts:

• Importance of grounding text in visual/contextual information
• Multi-modal learning approaches for better understanding

Relevance to Chunking:

• Supports contextual chunking approaches that preserve visual/contextual relationships
• Validates importance of maintaining document structure and relationships

Industry Research

NVIDIA Research: "Finding the Best Chunking Strategy for Accurate AI Responses"

Key Findings:

• Page-level chunking outperformed sentence and paragraph-level approaches
• Fixed-size chunking showed consistent but suboptimal performance
• Semantic chunking provided improvements for complex documents

Technical Details:

• Tested chunk sizes from 128 to 2048 tokens
• Evaluated across financial, technical, and legal documents
• Measured both retrieval accuracy and generation quality

Recommendations:

• Use 512-1024 token chunks as starting point
• Implement adaptive chunking based on document complexity
• Consider page boundaries as natural chunk separators

Cohere Research: "Effective Chunking Strategies for RAG"

Key Findings:

• Recursive character splitting provides good balance of performance and simplicity
• Document structure awareness improves retrieval by 15-20%
• Overlap of 10-20% provides optimal context preservation

Methodology:

• Compared 12 chunking strategies across 6 document types
• Measured retrieval precision, recall, and F1-score
• Tested with both dense and sparse retrieval

Best Practices Identified:

• Start with recursive character splitting with 10-20% overlap
• Preserve document structure (headings, lists, tables)
• Customize chunk size based on embedding model context window

Anthropic: "Contextual Retrieval"

Key Innovation:

• Enhance each chunk with LLM-generated contextual information before embedding
• Improves retrieval precision by 25-30% for complex documents
• Particularly effective for technical and academic content

Implementation Approach: 1. Split document using traditional methods 2. For each chunk, generate contextual information using LLM 3. Prepend context to chunk before embedding 4. Use hybrid search (dense + sparse) with weighted ranking

Trade-offs:

• Significant computational overhead (2-3x processing time)
• Higher embedding storage requirements
• Improved retrieval precision justifies cost for high-value applications

Algorithmic Advances

Semantic Chunking Algorithms

"Semantic Segmentation of Text Documents"

Core Idea: Use cosine similarity between consecutive sentence embeddings to identify natural boundaries.

Algorithm: 1. Split document into sentences 2. Generate embeddings for each sentence 3. Calculate similarity between consecutive sentences 4. Create boundaries where similarity drops below threshold 5. Merge short segments with neighbors

Performance: 20-30% improvement in retrieval relevance over fixed-size chunking for technical documents.

"Hierarchical Semantic Chunking"

Core Idea: Multi-level semantic segmentation for document organization.

Algorithm: 1. Document-level semantic analysis 2. Section-level boundary detection 3. Paragraph-level segmentation 4. Sentence-level refinement

Benefits: Maintains document hierarchy while adapting to semantic structure.

Advanced Embedding Techniques

"Late Chunking: Contextual Chunk Embeddings"

Core Innovation: Generate embeddings for entire document first, then create chunk embeddings from token-level embeddings.

Advantages:

• Preserves global document context
• Reduces context fragmentation
• Better for documents with complex inter-relationships

Requirements:

• Long-context embedding models (8k+ tokens)
• Significant computational resources
• Specialized implementation

"Hierarchical Embedding Retrieval"

Approach: Create embeddings at multiple granularities (document, section, paragraph, sentence).

Implementation: 1. Generate embeddings at each level 2. Store in hierarchical vector database 3. Query at appropriate granularity based on information needs

Performance: 15-25% improvement in precision for complex queries.

Evaluation Methodologies

Retrieval-Augmented Generation Assessment Frameworks

RAGAS Framework

Metrics:

• Faithfulness: Consistency between generated answer and retrieved context
• Answer Relevancy: Relevance of generated answer to the question
• Context Relevancy: Relevance of retrieved context to the question
• Context Recall: Coverage of relevant information in retrieved context

Evaluation Process: 1. Generate questions from document corpus 2. Retrieve relevant chunks using different strategies 3. Generate answers using retrieved chunks 4. Evaluate using automated metrics and human judgment

ARES Framework

Innovation: Automated evaluation using synthetic questions and LLM-based assessment.

Key Features:

• Generates diverse question types (factoid, analytical, comparative)
• Uses LLMs to evaluate answer quality
• Provides scalable evaluation without human annotation

Benchmark Datasets

Natural Questions (NQ)

Description: Real user questions from Google Search with relevant Wikipedia passages.

Relevance: Natural language queries with authentic relevance judgments.

MS MARCO

Description: Large-scale passage ranking dataset with real search queries.

Relevance: High-quality relevance judgments for passage retrieval.

HotpotQA

Description: Multi-hop question answering requiring information from multiple documents.

Relevance: Tests ability to retrieve and synthesize information from multiple chunks.

Domain-Specific Research

Medical Documents

"Optimal Chunking for Medical Question Answering"

Key Findings:

• Medical terminology requires specialized handling
• Section-based chunking (History, Diagnosis, Treatment) most effective
• Preserving doctor-patient dialogue context crucial

Recommendations:

• Use medical-specific tokenizers
• Preserve section headers and structure
• Maintain temporal relationships in medical histories

Legal Documents

"Chunking Strategies for Legal Document Analysis"

Key Findings:

• Legal citations and cross-references require special handling
• Contract clause boundaries serve as natural chunk separators
• Case law benefits from hierarchical chunking

Best Practices:

• Preserve legal citation structure
• Use clause and section boundaries
• Maintain context for legal definitions and references

Financial Documents

"SEC Filing Chunking for Financial Analysis"

Key Findings:

• Table preservation critical for financial data
• XBRL tagging provides natural segmentation
• Risk factors sections benefit from specialized treatment

Approach:

• Preserve complete tables when possible
• Use XBRL tags for structured data
• Create specialized chunks for risk sections

Emerging Trends

Multi-Modal Chunking

"Integrating Text, Tables, and Images in RAG Systems"

Innovation: Unified chunking approach for mixed-modal content.

Approach:

• Extract and describe images using vision models
• Preserve table structure and relationships
• Create unified embeddings for mixed content

Results: 35% improvement in complex document understanding.

Adaptive Chunking

"Machine Learning-Based Chunk Size Optimization"

Core Idea: Use ML models to predict optimal chunking parameters.

Features:

• Document length and complexity
• Query type distribution
• Embedding model characteristics
• Performance requirements

Benefits: Dynamic optimization based on use case and content.

Real-time Chunking

"Streaming Chunking for Live Document Processing"

Innovation: Process documents as they become available.

Techniques:

• Incremental boundary detection
• Dynamic chunk size adjustment
• Context preservation across chunks

Applications: Live news feeds, social media analysis, meeting transcripts.

Implementation Challenges

Computational Efficiency

"Scalable Chunking for Large Document Collections"

Challenges:

• Processing millions of documents efficiently
• Memory usage optimization
• Distributed processing requirements

Solutions:

• Batch processing with parallel execution
• Streaming approaches for large documents
• Distributed chunking with load balancing

Quality Assurance

"Evaluating Chunk Quality at Scale"

Challenges:

• Automated quality assessment
• Detecting poor chunk boundaries
• Maintaining consistency across document types

Approaches:

• Heuristic-based quality metrics
• LLM-based evaluation
• Human-in-the-loop validation

Future Research Directions

Context-Aware Chunking

Open Questions:

• How to optimally preserve cross-chunk relationships?
• Can we predict chunk quality without human evaluation?
• What is the optimal balance between size and context?

Domain Adaptation

Research Areas:

• Automatic domain detection and adaptation
• Transfer learning across domains
• Zero-shot chunking for new document types

Evaluation Standards

Needs:

• Standardized evaluation benchmarks
• Cross-paper comparison methodologies
• Real-world performance metrics

Practical Recommendations Based on Research

Starting Points

1. For General RAG Systems: Page-level or recursive character chunking with 512-1024 tokens and 10-20% overlap 2. For Technical Documents: Structure-aware chunking with semantic boundary detection 3. For High-Value Applications: Contextual retrieval with LLM-generated context

Evolution Strategy

1. Begin: Simple fixed-size chunking (512 tokens) 2. Improve: Add document structure awareness 3. Optimize: Implement semantic boundaries 4. Advanced: Consider contextual retrieval for critical use cases

Key Success Factors

1. Match strategy to document type and query patterns 2. Preserve document structure when beneficial 3. Use overlap to maintain context across boundaries 4. Monitor both accuracy and computational costs 5. Iterate based on specific use case requirements

This research foundation provides evidence-based guidance for implementing effective chunking strategies across various domains and use cases.