Chunking & Ingestion

from langchain_text_splitters import RecursiveCharacterTextSplitter
 
splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
chunks = splitter.split_documents(documents)

"""chunking_ingestion.py - clean, chunk, embed, and index documents."""
from __future__ import annotations
 
import re
import chromadb
from openai import OpenAI
 
client = OpenAI()
 
RAW_DOCS = [
    {"id": "doc1", "source": "handbook", "text": "# Python Testing\n\nUse pytest for unit tests. " * 20},
    {"id": "doc2", "source": "handbook", "text": "# FastAPI\n\nBuild APIs with type hints. " * 20},
]
 
def clean_text(text: str) -> str:
    text = re.sub(r"\s+", " ", text).strip()
    return text
 
def chunk_text(text: str, size: int = 400, overlap: int = 80) -> list[str]:
    chunks = []
    start = 0
    while start < len(text):
        chunks.append(text[start:start + size])
        start += size - overlap
    return chunks
 
chroma = chromadb.Client()
collection = chroma.get_or_create_collection("handbook")
 
for doc in RAW_DOCS:
    cleaned = clean_text(doc["text"])
    for i, chunk in enumerate(chunk_text(cleaned)):
        chunk_id = f"{doc['id']}_chunk_{i}"
        emb = client.embeddings.create(model="text-embedding-3-small", input=[chunk])
        collection.add(
            ids=[chunk_id],
            documents=[chunk],
            embeddings=[emb.data[0].embedding],
            metadatas=[{"source": doc["source"], "doc_id": doc["id"], "chunk_index": i}],
        )
print(f"indexed {collection.count()} chunks")