Chunking and ChromaDB ingest

Every RAG pipeline starts the same way: split raw text into chunks, embed each chunk, and persist the vectors with enough metadata to rebuild answers later. The quality of this step caps the quality of everything downstream. Chunks that are too small lose context. Chunks that are too large dilute the signal.

rag_utils.py
python
from langchain_text_splitters import RecursiveCharacterTextSplitter

def chunk_text(text: str, chunk_size: int = 500, chunk_overlap: int = 50) -> list[str]:
    """Split text using LangChain's recursive splitter."""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", " ", ""],
    )
    return splitter.split_text(text)

RecursiveCharacterTextSplitter tries each separator in order. Paragraphs first, then lines, then words, then characters. Overlap keeps a sliding window of context across chunk boundaries.

rag_utils.py
python
class VectorStore:
    """Persistent ChromaDB collection wrapper."""

    def __init__(self, collection_name=None, persist_directory=None):
        self.persist_directory = persist_directory or os.getenv("CHROMA_PATH", "./chroma_db")
        self.collection_name = collection_name or os.getenv("CHROMA_COLLECTION", "advanced_rag")
        self.client = chromadb.PersistentClient(path=self.persist_directory)
        self.collection = self.client.get_or_create_collection(name=self.collection_name)

    def add_documents(self, documents, embeddings):
        ids = [f"{doc.source}_{doc.chunk_index}_{i}" for i, doc in enumerate(documents)]
        metadatas = [{"source": doc.source, "chunk_index": doc.chunk_index, **doc.metadata} for doc in documents]
        contents = [doc.content for doc in documents]
        self.collection.add(ids=ids, embeddings=embeddings, metadatas=metadatas, documents=contents)

PersistentClient writes to disk, so your ingested documents survive server restarts. The ids include source + chunk_index so you can trace any retrieved chunk back to its origin.

service.py
python
async def index_documents(request: IndexRequest) -> dict:
    all_chunks: list[DocumentChunk] = []
    for doc in request.documents:
        pieces = chunk_text(doc.content, request.chunk_size, request.chunk_overlap)
        for i, piece in enumerate(pieces):
            all_chunks.append(DocumentChunk(
                content=piece, source=doc.source, chunk_index=i,
                metadata={"source": doc.source, **doc.metadata},
            ))

    if not all_chunks:
        return {"indexed": 0, "total": vector_store.count()}

    embeddings = await embedder.generate_embeddings([c.content for c in all_chunks])
    vector_store.add_documents(all_chunks, embeddings)
    return {"indexed": len(all_chunks), "total": vector_store.count()}

The service batches the embedding call, then writes all chunks to ChromaDB in one go. The embedder is a shared singleton with the retriever so you only load the model once.

Quiz: Quiz

Loading practice…