Document Chunking & Embedding

Project Overview

Step 1: Document Chunking

from langchain_text_splitters import RecursiveCharacterTextSplitter
import tiktoken

def chunk_document(text: str, chunk_size: int = 500, overlap: int = 80) -> list[str]:
    """Split a document into overlapping chunks optimized for embedding.
    
    Uses tiktoken to count tokens accurately for OpenAI models.
    RecursiveCharacterTextSplitter tries to split on paragraphs first,
    then sentences, then words — preserving semantic coherence.
    """
    enc = tiktoken.encoding_for_model("text-embedding-3-small")
    
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=overlap,
        length_function=lambda t: len(enc.encode(t)),
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    
    chunks = splitter.split_text(text)
    print(f"Split into {len(chunks)} chunks (target: {chunk_size} tokens, overlap: {overlap})")
    return chunks


# Example usage
with open("company_handbook.txt", "r") as f:
    raw_text = f.read()

chunks = chunk_document(raw_text)
print(f"First chunk preview: {chunks[0][:200]}...")

• RecursiveCharacterTextSplitter cascades through separator types, keeping paragraphs intact when possible.
• Overlap prevents information loss at chunk boundaries — critical for questions that span two paragraphs.
• Use tiktoken (not len()) to count tokens, since character count does not map linearly to token count.

Step 2: Generate Embeddings

from openai import OpenAI

client = OpenAI()  # uses OPENAI_API_KEY env var

def embed_chunks(chunks: list[str], model: str = "text-embedding-3-small") -> list[list[float]]:
    """Generate embeddings for a list of text chunks.
    
    Batches requests for efficiency. OpenAI allows up to 2048 inputs
    per request for the small model.
    """
    batch_size = 512
    all_embeddings = []
    
    for i in range(0, len(chunks), batch_size):
        batch = chunks[i : i + batch_size]
        response = client.embeddings.create(input=batch, model=model)
        batch_embeddings = [item.embedding for item in response.data]
        all_embeddings.extend(batch_embeddings)
        print(f"Embedded batch {i // batch_size + 1}: {len(batch)} chunks")
    
    print(f"Total embeddings: {len(all_embeddings)}, dimensions: {len(all_embeddings[0])}")
    return all_embeddings


embeddings = embed_chunks(chunks)

Step 3: Store in a Vector Database

import chromadb
import uuid

def create_vector_store(
    chunks: list[str],
    embeddings: list[list[float]],
    collection_name: str = "documents",
    metadata_source: str = "company_handbook.txt",
) -> chromadb.Collection:
    """Store chunks and embeddings in ChromaDB with metadata."""
    client = chromadb.PersistentClient(path="./chroma_db")
    
    # Delete existing collection if re-indexing
    try:
        client.delete_collection(collection_name)
    except ValueError:
        pass
    
    collection = client.create_collection(
        name=collection_name,
        metadata={"hnsw:space": "cosine"},  # cosine similarity
    )
    
    # Add chunks with metadata
    ids = [str(uuid.uuid4()) for _ in chunks]
    metadatas = [
        {"source": metadata_source, "chunk_index": i, "token_count": len(chunk.split())}
        for i, chunk in enumerate(chunks)
    ]
    
    collection.add(
        ids=ids,
        documents=chunks,
        embeddings=embeddings,
        metadatas=metadatas,
    )
    
    print(f"Stored {len(chunks)} chunks in collection '{collection_name}'")
    return collection


collection = create_vector_store(chunks, embeddings)