Document Processing & Chunking

# Document parsing with Unstructured.io
from unstructured.partition.pdf import partition_pdf
# Parse a PDF with layout detection
elements = partition_pdf(
    filename="technical_report.pdf",
    strategy="hi_res", # Use layout detection model
    infer_table_structure=True, # Extract table structure
    include_page_breaks=True, # Track page boundaries
    )
# Inspect extracted elements
for element in elements[:10]:
    print(f"Type: {type(element).__name__:20s} | "
        f"Page: {element.metadata.page_number} | "
        f"Text: {str(element)[:60]}...")
    # Filter by element type
    from unstructured.documents.elements import Title, NarrativeText, Table
    titles = [e for e in elements if isinstance(e, Title)]
    text_blocks = [e for e in elements if isinstance(e, NarrativeText)]
    tables = [e for e in elements if isinstance(e, Table)]
    print(f"\nExtracted: {len(titles)} titles, "
        f"{len(text_blocks)} text blocks, "
        f"{len(tables)} tables")

# Fixed-size chunking with overlap
from typing import List
def fixed_size_chunk(
    text: str,
    chunk_size: int = 500,
    chunk_overlap: int = 50
    ) -> List[str]:
    """
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    Split text into fixed-size chunks with overlap.
    Args:
    text: Input text to chunk
    chunk_size: Maximum characters per chunk
    chunk_overlap: Characters to overlap between consecutive chunks
    """
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        # If not the last chunk, try to break at a sentence boundary
        if end < len(text):
            # Look for sentence boundary near the end
            for boundary in [". ", ".\n", "? ", "! "]:
                last_boundary = text[start:end].rfind(boundary)
                if last_boundary > chunk_size * 0.5:
                    end = start + last_boundary + len(boundary)
                    break
                chunk = text[start:end].strip()
                if chunk:
                    chunks.append(chunk)
                    # Move start position, accounting for overlap
                    start = end - chunk_overlap
                    return chunks
# Example
sample_text = """
Vector databases are specialized systems designed for storing and querying
high-dimensional vectors. They use approximate nearest neighbor algorithms
to find similar vectors efficiently.
The most common algorithm is HNSW, which builds a multi-layer graph structure.
Each layer connects vectors to their nearest neighbors, enabling fast navigation
from any starting point to the target region of the vector space.
Product Quantization reduces memory usage by compressing vectors. Each vector
is split into sub-vectors, and each sub-vector is replaced by its nearest
codebook entry. This can achieve 32x compression with acceptable accuracy loss.
"""
chunks = fixed_size_chunk(sample_text, chunk_size=200, chunk_overlap=30)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i} ({len(chunk)} chars): {chunk[:70]}...")

# Recursive character text splitting (LangChain-style)
from langchain_text_splitters import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
    chunk_size=400,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""],
    length_function=len,
    is_separator_regex=False,
    )
document = """# Introduction to Embeddings
Text embeddings convert natural language into dense vector representations.
These vectors capture semantic meaning, allowing mathematical operations
like cosine similarity to measure how related two pieces of text are.
## Training Approaches
Modern embedding models use contrastive learning. The model is trained to
produce similar vectors for semantically related text pairs and different
vectors for unrelated pairs. Hard negative mining improves training by
providing challenging negative examples that force the model to learn
fine-grained distinctions.
## Applications
Embeddings power semantic search, recommendation systems, clustering,
and retrieval-augmented generation. They serve as the foundation for
virtually every modern NLP application that requires understanding
meaning beyond keyword matching.
"""
chunks = splitter.split_text(document)
for i, chunk in enumerate(chunks):
    print(f"Chunk {i} ({len(chunk)} chars):")
    print(f" {chunk[:80]}...")
    print()

# Semantic chunking based on embedding similarity
import numpy as np
from sentence_transformers import SentenceTransformer
from typing import List, Tuple
import re
def semantic_chunk(
    text: str,
    model: SentenceTransformer,
    threshold_percentile: int = 25,
    min_chunk_size: int = 100,
    ) -> List[str]:
    """
    Split text into semantically coherent chunks by detecting
    topic boundaries using embedding similarity.
    """
    # Split into sentences
    sentences = re.split(r'(?<=[.!?])\s+', text.strip())
    sentences = [s for s in sentences if len(s) > 10]
    if len(sentences) <= 1:
        return [text]
        # Embed all sentences
        embeddings = model.encode(sentences, normalize_embeddings=True)
        # Compute cosine similarity between consecutive sentences
        similarities = []
        for i in range(len(embeddings) - 1):
            sim = np.dot(embeddings[i], embeddings[i + 1])
            similarities.append(sim)
            # Find breakpoints where similarity drops below threshold
            threshold = np.percentile(similarities, threshold_percentile)
            breakpoints = [i + 1 for i, sim in enumerate(similarities)
                if sim < threshold]
            # Build chunks from breakpoints
            chunks = []
            start = 0
            for bp in breakpoints:
                chunk = " ".join(sentences[start:bp])
                if len(chunk) >= min_chunk_size:
                    chunks.append(chunk)
                    start = bp
                    # Add remaining sentences
                    final_chunk = " ".join(sentences[start:])
                    if final_chunk:
                        chunks.append(final_chunk)
                        return chunks
                        # Example usage
                        model = SentenceTransformer("all-MiniLM-L6-v2")
                        text = """
                        Machine learning models learn patterns from data. They adjust internal
                        parameters to minimize prediction errors. The training process uses
                        gradient descent to iteratively improve the model.
                        Vector databases store high-dimensional vectors. They use algorithms like
                        HNSW for fast approximate nearest neighbor search. These systems are
                        critical for semantic search applications.
                        Python is the most popular language for data science. It provides libraries
                        like NumPy, pandas, and scikit-learn. The ecosystem continues to grow rapidly.
                        """
                        chunks = semantic_chunk(text, model)
                        for i, chunk in enumerate(chunks):
                            print(f"Semantic Chunk {i}: {chunk[:70]}...")

# Parent-child chunking strategy
from langchain_text_splitters import RecursiveCharacterTextSplitter
from typing import List, Dict
import uuid
def create_parent_child_chunks(
    text: str,
    parent_chunk_size: int = 1000,
    child_chunk_size: int = 200,
    child_overlap: int = 20,
    ) -> List[Dict]:
    """
    Create a two-tier chunking structure for parent-child retrieval.
    Child chunks are used for embedding and retrieval.
    Parent chunks are returned for LLM context.
    """
    # Create parent chunks
    parent_splitter = RecursiveCharacterTextSplitter(
        chunk_size=parent_chunk_size,
        chunk_overlap=0,
        )
    parent_chunks = parent_splitter.split_text(text)
    all_chunks = []
    for parent_idx, parent_text in enumerate(parent_chunks):
        parent_id = str(uuid.uuid4())
        # Store parent chunk
        all_chunks.append({
            "id": parent_id,
            "text": parent_text,
            "type": "parent",
            "parent_id": None,
            })
        # Create child chunks from this parent
        child_splitter = RecursiveCharacterTextSplitter(
            chunk_size=child_chunk_size,
            chunk_overlap=child_overlap,
            )
        child_texts = child_splitter.split_text(parent_text)
        for child_idx, child_text in enumerate(child_texts):
            all_chunks.append({
                "id": str(uuid.uuid4()),
                "text": child_text,
                "type": "child",
                "parent_id": parent_id,
                })
            parents = [c for c in all_chunks if c["type"] == "parent"]
            children = [c for c in all_chunks if c["type"] == "child"]
            print(f"Created {len(parents)} parents, {len(children)} children")
            print(f"Avg parent size: {sum(len(p['text']) for p in parents) / len(parents):.0f} chars")
            print(f"Avg child size: {sum(len(c['text']) for c in children) / len(children):.0f} chars")
            return all_chunks

# Usage: embed children, retrieve parents
# At query time:
    # 1. Search child embeddings for top-k matches
    # 2. For each matching child, look up its parent_id
    # 3. Return deduplicated parent chunks to the LLM