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

# Incremental indexing with content hashing
import hashlib
import json
from typing import Dict, List, Optional
from pathlib import Path

class IncrementalIndexer:
 """
 Tracks document versions to enable incremental re-indexing.
 Only processes documents that have changed since the last run.
 """

 def __init__(self, state_file: str = "indexer_state.json"):
 self.state_file = Path(state_file)
 self.state: Dict[str, str] = {}
 if self.state_file.exists():
 self.state = json.loads(self.state_file.read_text())

 def content_hash(self, content: str) -> str:
 return hashlib.sha256(content.encode()).hexdigest()

 def get_changes(
 self, documents: Dict[str, str]
 ) -> Dict[str, List[str]]:
 """
 Compare current documents against stored state.

 Args:
 documents: dict of {doc_id: content}

 Returns:
 {"added": [...], "modified": [...], "deleted": [...]}
 """
 current_ids = set(documents.keys())
 stored_ids = set(self.state.keys())

 added = current_ids - stored_ids
 deleted = stored_ids - current_ids
 modified = set()

 for doc_id in current_ids & stored_ids:
 new_hash = self.content_hash(documents[doc_id])
 if new_hash != self.state[doc_id]:
 modified.add(doc_id)

 return {
 "added": list(added),
 "modified": list(modified),
 "deleted": list(deleted),
 }

 def update_state(self, documents: Dict[str, str]):
 """Update stored hashes after successful indexing."""
 for doc_id, content in documents.items():
 self.state[doc_id] = self.content_hash(content)
 self.state_file.write_text(json.dumps(self.state, indent=2))

 def process_changes(self, documents: Dict[str, str]):
 """Main entry point for incremental processing."""
 changes = self.get_changes(documents)

 print(f"Added: {len(changes['added'])} documents")
 print(f"Modified: {len(changes['modified'])} documents")
 print(f"Deleted: {len(changes['deleted'])} documents")

 # For added/modified: chunk, embed, upsert
 to_process = changes["added"] + changes["modified"]
 if to_process:
 print(f"Processing {len(to_process)} documents...")
 # chunk_and_embed(to_process)
 # vector_db.upsert(chunks)

 # For deleted: remove from vector DB
 if changes["deleted"]:
 print(f"Removing {len(changes['deleted'])} documents...")
 # vector_db.delete(filter={"doc_id": {"$in": changes["deleted"]}})

 # For modified: also remove old chunks before upserting new ones
 if changes["modified"]:
 print(f"Replacing chunks for {len(changes['modified'])} documents...")
 # vector_{db}.delete(filter={"doc_{id}": {"$in": changes["modified"]}})
 # vector_db.upsert(new_chunks)

 self.update_state(documents)

# Usage
indexer = IncrementalIndexer()
docs = {
 "report_2024.pdf": "Full text of the 2024 report...",
 "manual_v3.pdf": "Updated product manual content...",
 "faq.md": "Frequently asked questions...",
}
indexer.process_changes(docs)

# BERTopic: embedding-based topic modeling
# Uses the same sentence transformers as RAG embedding pipelines
from bertopic import BERTopic
from sentence_transformers import SentenceTransformer
from umap import UMAP
from hdbscan import HDBSCAN

# Step 1: Configure each pipeline component
embedding_model = SentenceTransformer("all-MiniLM-L6-v2")
umap_model = UMAP(
 n_neighbors=15, n_components=5,
 min_dist=0.0, metric="cosine",
)
hdbscan_model = HDBSCAN(
 min_cluster_size=15,
 metric="euclidean",
 prediction_data=True,
)

# Step 2: Build the BERTopic model with custom components
topic_model = BERTopic(
 embedding_model=embedding_model,
 umap_model=umap_model,
 hdbscan_model=hdbscan_model,
 verbose=True,
)

# Step 3: Fit on your documents (e.g., customer support tickets)
documents = [
 "My order has not arrived after two weeks",
 "How do I reset my password for the dashboard?",
 "The API returns a 500 error on large payloads",
 "I was charged twice for the same subscription",
 "Can I export my data as a CSV file?",
 # ... thousands more documents
]

topics, probs = topic_model.fit_transform(documents)

# Step 4: Inspect discovered topics
topic_info = topic_model.get_topic_info()
print(topic_info.head(10))

# Step 5: Get the top terms for a specific topic
for topic_id in range(min(5, len(topic_model.get_topics()))):
 terms = topic_model.get_topic(topic_id)
 print(f"\nTopic {topic_id}:")
 for term, score in terms[:5]:
 print(f" {score:.3f} {term}")