Module 8: Capstone RAG Project with ChromaDB

Capsule 03: Ingestion Pipeline for 1,000+ documents

Next →

Capsule description

You're going to scale the minimal RAG pipeline you built in M4/11 to a real production volume (1,000+ documents) with a focus on stability and traceability: batches, consistent metadata, result validation, and error handling. The pipeline must be reproducible, idempotent when possible, and report basic throughput metrics.

Prerequisites from previous capsules you'll use here:

  • M4/09 — Embeddings with OpenAI: you already know why we use text-embedding-3-small instead of the ChromaDB default, and how to handle the API key correctly.
  • M4/10 — Document chunking: you already understand why chunk_size=512 with overlap=50 is the reasonable range for technical documents, and how RecursiveCharacterTextSplitter decides where to split.
  • M4/11 — End-to-end RAG pipeline: you already built the minimal version of the pipeline (~150 lines). This capsule scales it to 1000+ docs and adds robustness (error handling, batches, validation, metrics).

If you didn't complete those capsules, go back to M4/09 before continuing — this capsule assumes the embedding and chunking decisions are already justified.

Implementation flow

  1. Load source documents — PDF, TXT, MD by type, normalize encoding.
  2. Chunking — RecursiveCharacterTextSplitter with chunk_size=512, overlap=50 (justified in M4/10).
  3. Embeddings — OpenAI text-embedding-3-small in batches of 100-200 (justified in M4/09).
  4. Insertion into ChromaDB — batch of 1K-2K with stable IDs and metadata.
  5. Verification — total count, mandatory metadata present, no duplicate IDs.

Required dependencies

# requirements.txt (extract for the project)
chromadb>=0.4.22
openai>=1.12.0
langchain-text-splitters>=0.2.0
langchain-community>=0.2.0  # optional: loaders
tqdm>=4.66.0
python-dotenv>=1.0.0
pypdf>=4.0.0  # for PDF

Complete pipeline code

1. Configuration and utilities

# ingestion/config.py
import os
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()


@dataclass
class IngestionConfig:
    """Centralized configuration for the ingestion pipeline."""
    chunk_size: int = int(os.getenv("CHUNK_SIZE", "512"))
    chunk_overlap: int = int(os.getenv("CHUNK_OVERLAP", "50"))
    batch_size_chromadb: int = int(os.getenv("BATCH_SIZE_CHROMADB", "1000"))
    batch_size_embeddings: int = int(os.getenv("BATCH_SIZE_EMBEDDINGS", "100"))
    chroma_path: str = os.getenv("CHROMA_PATH", "./chroma_data")
    collection_name: str = os.getenv("COLLECTION_NAME", "rag_docs")
    openai_model: str = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small")

2. Document loading

# ingestion/loaders.py
import os
from pathlib import Path
from typing import Iterator

def load_text_file(path: Path) -> str:
    """Load a text file with robust encoding."""
    encodings = ["utf-8", "latin-1", "cp1252"]
    for enc in encodings:
        try:
            return path.read_text(encoding=enc)
        except UnicodeDecodeError:
            continue
    raise ValueError(f"Could not decode: {path}")


def load_documents_from_dir(
    dir_path: str,
    extensions: tuple[str, ...] = (".txt", ".md", ".rst"),
) -> Iterator[tuple[str, str, str]]:
    """
    Iterates over documents in a directory.
    Yields: (doc_id, content, source_path)
    """
    path = Path(dir_path)
    if not path.exists():
        raise FileNotFoundError(f"Directory not found: {dir_path}")

    for file_path in path.rglob("*"):
        if file_path.suffix.lower() in extensions and file_path.is_file():
            try:
                content = load_text_file(file_path)
                # doc_id: stable and unique
                doc_id = file_path.relative_to(path).as_posix().replace("/", "_")
                source = str(file_path)
                yield doc_id, content, source
            except Exception as e:
                print(f"[WARN] Error loading {file_path}: {e}")
                continue

3. Chunking with RecursiveCharacterTextSplitter

# ingestion/chunking.py
from langchain_text_splitters import RecursiveCharacterTextSplitter
from pathlib import Path
from typing import List
from dataclasses import dataclass


@dataclass
class ChunkWithMeta:
    """Chunk with metadata for ChromaDB."""
    text: str
    doc_id: str
    chunk_index: int
    source: str
    metadata: dict


def chunk_document(
    content: str,
    doc_id: str,
    source: str,
    chunk_size: int = 512,
    chunk_overlap: int = 50,
) -> List[ChunkWithMeta]:
    """
    Split a document into chunks with overlap.
    chunk_size and overlap are in characters (approx ~4 chars = 1 token).
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        length_function=len,
        separators=["\n\n", "\n", ". ", " ", ""],
    )
    chunks_raw = splitter.split_text(content)

    result = []
    for i, text in enumerate(chunks_raw):
        meta = {
            "doc_id": doc_id,
            "source": source,
            "chunk_index": i,
            "title": Path(source).stem if source else doc_id,
        }
        result.append(ChunkWithMeta(
            text=text,
            doc_id=doc_id,
            chunk_index=i,
            source=source,
            metadata=meta,
        ))
    return result

Note: If you use Path in chunking.py, add from pathlib import Path at the top.

4. Embedding generation (OpenAI)

# ingestion/embeddings.py
from openai import OpenAI
from typing import List
import time

client = OpenAI()


def get_embeddings_batch(texts: List[str], model: str = "text-embedding-3-small") -> List[List[float]]:
    """
    Generate embeddings for a batch of texts.
    Rate limit: ~3000 RPM for text-embedding-3-small; use batch 100-200.
    """
    if not texts:
        return []

    response = client.embeddings.create(
        model=model,
        input=texts,
    )
    # Order guaranteed the same as input
    return [item.embedding for item in sorted(response.data, key=lambda x: x.index)]

5. Complete ingestion pipeline

# ingestion/pipeline.py
import chromadb
from chromadb.config import Settings
from tqdm import tqdm
import hashlib
import time
from pathlib import Path

from ingestion.config import IngestionConfig
from ingestion.loaders import load_documents_from_dir
from ingestion.chunking import chunk_document
from ingestion.embeddings import get_embeddings_batch


def make_chunk_id(doc_id: str, chunk_index: int) -> str:
    """Deterministic ID for idempotency."""
    return f"{doc_id}_chunk_{chunk_index}"


def ingest_directory(
    dir_path: str,
    config: IngestionConfig | None = None,
    extensions: tuple[str, ...] = (".txt", ".md", ".rst"),
) -> dict:
    """
    Complete pipeline: load docs, chunking, embeddings, ChromaDB.
    Returns metrics: total_docs, total_chunks, time_seconds, errors.
    """
    config = config or IngestionConfig()
    client = chromadb.PersistentClient(path=config.chroma_path)
    collection = client.get_or_create_collection(
        name=config.collection_name,
        metadata={"description": "RAG collection"},
    )

    # Phase 1: Load and chunk
    all_chunks = []
    all_metadatas = []
    all_ids = []
    errors = []

    for doc_id, content, source in load_documents_from_dir(dir_path, extensions):
        try:
            chunks = chunk_document(
                content=content,
                doc_id=doc_id,
                source=source,
                chunk_size=config.chunk_size,
                chunk_overlap=config.chunk_overlap,
            )
            for c in chunks:
                all_chunks.append(c.text)
                all_metadatas.append(c.metadata)
                all_ids.append(make_chunk_id(doc_id, c.chunk_index))
        except Exception as e:
            errors.append({"doc_id": doc_id, "error": str(e)})
            continue

    total_chunks = len(all_chunks)
    if total_chunks == 0:
        return {
            "total_docs": 0,
            "total_chunks": 0,
            "time_seconds": 0,
            "errors": errors,
        }

    # Phase 2: Embeddings in batch
    embeddings_list = []
    batch_size = config.batch_size_embeddings
    for i in tqdm(range(0, total_chunks, batch_size), desc="Embeddings"):
        batch_texts = all_chunks[i : i + batch_size]
        try:
            emb = get_embeddings_batch(batch_texts, model=config.openai_model)
            embeddings_list.extend(emb)
        except Exception as e:
            errors.append({"batch": i, "error": str(e)})
            # Fallback: zeros so as not to break the add (better: skip the batch)
            embeddings_list.extend([[0.0] * 1536 for _ in batch_texts])  # dim of 3-small

    # Phase 3: Add to ChromaDB in batches
    chroma_batch = config.batch_size_chromadb
    start = time.time()
    for i in tqdm(range(0, total_chunks, chroma_batch), desc="ChromaDB"):
        batch_ids = all_ids[i : i + chroma_batch]
        batch_docs = all_chunks[i : i + chroma_batch]
        batch_emb = embeddings_list[i : i + chroma_batch]
        batch_meta = all_metadatas[i : i + chroma_batch]
        try:
            collection.add(
                ids=batch_ids,
                documents=batch_docs,
                embeddings=batch_emb,
                metadatas=batch_meta,
            )
        except Exception as e:
            errors.append({"chroma_batch": i, "error": str(e)})

    elapsed = time.time() - start

    return {
        "total_docs": len(set(m.get("doc_id") for m in all_metadatas)),
        "total_chunks": total_chunks,
        "time_seconds": round(elapsed, 2),
        "docs_per_second": round(total_chunks / elapsed, 1) if elapsed > 0 else 0,
        "errors": errors,
    }

6. Execution script

# run_ingestion.py
import argparse
from ingestion.config import IngestionConfig
from ingestion.pipeline import ingest_directory


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("dir", help="Directory with documents (.txt, .md, .rst)")
    parser.add_argument("--batch-size", type=int, default=1000)
    parser.add_argument("--chunk-size", type=int, default=512)
    parser.add_argument("--chunk-overlap", type=int, default=50)
    args = parser.parse_args()

    config = IngestionConfig(
        batch_size_chromadb=args.batch_size,
        chunk_size=args.chunk_size,
        chunk_overlap=args.chunk_overlap,
    )
    result = ingest_directory(args.dir, config=config)

    print("\n=== Ingestion result ===")
    print(f"Documents processed: {result['total_docs']}")
    print(f"Total chunks: {result['total_chunks']}")
    print(f"Time: {result['time_seconds']} s")
    print(f"Throughput: {result.get('docs_per_second', 0)} chunks/s")
    print(f"Errors: {len(result['errors'])}")
    if result["errors"]:
        for e in result["errors"][:5]:
            print(f"  - {e}")


if __name__ == "__main__":
    main()

Usage:

python run_ingestion.py ./my_documents --batch-size 1000

Minimal post-ingestion validations

After running the pipeline, run:

# validation/check_ingestion.py
import chromadb

client = chromadb.PersistentClient(path="./chroma_data")
coll = client.get_collection("rag_docs")

count = coll.count()
print(f"Total vectors: {count}")

# Random sample of metadata
sample = coll.peek(limit=5)
for i, meta in enumerate(sample["metadatas"]):
    required = ["doc_id", "source", "chunk_index"]
    ok = all(k in meta for k in required)
    print(f"Chunk {i}: metadata ok={ok}, keys={list(meta.keys())}")

# Verify unique IDs
ids = coll.get()["ids"]
assert len(ids) == len(set(ids)), "Duplicate IDs detected"
print("Unique IDs: OK")

Recommended parameters

ParameterValueJustification
chunk_size512Context/granularity balance, compatible with the embedding model
chunk_overlap50Semantic continuity without duplicating much
batch_size_chromadb1000-2000Optimal throughput for ChromaDB
batch_size_embeddings100-200Respect OpenAI rate limits

Document these values in the project's README and expose them via environment variables.

Idempotency strategy

To avoid duplicates on re-ingestion:

  1. Deterministic IDs: {doc_id}_chunk_{index} — the same doc always produces the same IDs.
  2. Delete the collection before re-ingesting (simple): client.delete_collection(name) and create it again.
  3. Upsert (if ChromaDB supports it): Some versions allow add with existing IDs that update them; check your version's documentation.

Applied exercises

Exercise 1: Implement ingestion of 1,000 docs

Create a directory with 1,000 .txt files (you can generate them or use a public corpus). Run the pipeline and report:

  • Total time
  • Chunks/second
  • Error percentage
  • Percentage of docs without complete metadata

Solution: Use a script to generate 1,000 files:

from pathlib import Path
Path("test_docs").mkdir(exist_ok=True)
for i in range(1000):
    (Path("test_docs") / f"doc_{i:04d}.txt").write_text(f"Content of document {i}. " * 50)

Then: python run_ingestion.py test_docs. Check the result output for metrics. If there are 0 errors and complete metadata in all, the percentage is 0%.

Exercise 2: PDF handling

Extend the loader to support PDF using pypdf. Add .pdf to the extensions and a load_pdf(path) function that extracts text from each page.

Solution:

from pypdf import PdfReader

def load_pdf(path: Path) -> str:
    reader = PdfReader(path)
    return "\n".join(page.extract_text() or "" for page in reader.pages)

In load_documents_from_dir, add ".pdf" and in the PDF branch use load_pdf instead of load_text_file.

Exercise 3: Progress bar per phase

Add tqdm to show progress in: (1) document loading, (2) chunking, (3) embeddings, (4) ChromaDB add. Each phase should have its own bar.

Solution: In load_documents_from_dir, convert the iterator into a list and wrap it with tqdm. For chunking, if you process doc by doc, tqdm(docs). For embeddings and ChromaDB, there is already tqdm in the batch loops.

Exercise 4: Retry with backoff for embeddings

If the OpenAI API fails due to a rate limit, implement retry with exponential backoff (1s, 2s, 4s) up to 3 attempts.

Solution:

import time

def get_embeddings_batch_with_retry(texts, model="text-embedding-3-small", max_retries=3):
    for attempt in range(max_retries):
        try:
            return get_embeddings_batch(texts, model)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt
            time.sleep(wait)

Exercise 5: Metadata validation before add

Before calling collection.add, validate that each element of batch_meta has the keys doc_id, source, chunk_index. If any is missing, log a warning and add a default value or discard the chunk.

Solution:

REQUIRED_KEYS = {"doc_id", "source", "chunk_index"}

def validate_metadata(meta: dict) -> bool:
    return REQUIRED_KEYS.issubset(meta.keys())

# In the ChromaDB add loop:
valid = [(i, m, d, e) for i, m, d, e in zip(...) if validate_metadata(m)]
invalid_count = len(batch_meta) - len(valid)
if invalid_count:
    print(f"[WARN] {invalid_count} chunks with incomplete metadata discarded")

Exercise 6: Batch size benchmark

Test batch_size_chromadb at [500, 1000, 2000, 5000] with 5,000 chunks. Measure time and throughput. What is the sweet spot on your machine?

Solution: Benchmark script:

for bs in [500, 1000, 2000, 5000]:
    config = IngestionConfig(batch_size_chromadb=bs)
    # Delete the collection first
    r = ingest_directory("test_docs", config=config)
    print(f"batch={bs}: {r['time_seconds']}s, {r.get('docs_per_second')} docs/s")

Typical: 1000-2000 gives the best throughput; 5000 sometimes doesn't improve due to memory overhead.

Ingestion troubleshooting

"Ingestion takes too long"

Cause: Small batch size, or a bottleneck in embeddings (API).

Solution: Increase batch_size_chromadb to 1500-2000. For embeddings, use batch_size_embeddings of 100-200 and verify you're not limited by OpenAI rate limits. Measure time per phase to locate the bottleneck.

"Duplicates appear"

Cause: Non-deterministic IDs or re-ingestion without deleting the collection.

Solution: Use make_chunk_id(doc_id, chunk_index). If you re-ingest, delete the collection first or implement upsert if your ChromaDB version allows it.

"Metadata is missing in part of the dataset"

Cause: Some loaders don't extract source or doc_id, or there are docs without a title.

Solution: Validate metadata before add. Add defaults: source=doc_id if missing, title=doc_id if there is no title. Log which docs have incomplete metadata.

"OpenAI 429 error (rate limit)"

Cause: Too many requests per minute.

Solution: Reduce batch_size_embeddings to 50-100. Implement retry with backoff. Consider time.sleep(1) between batches if necessary.

"ChromaDB out of memory"

Cause: Batch too large or too many vectors in memory.

Solution: Reduce batch_size_chromadb to 500. Process in several runs if the dataset is huge. Verify you're not accumulating all embeddings in memory before add (process in streams).

Summary

  • The pipeline has 5 phases: load, chunking, embeddings, add to ChromaDB, verification.
  • Use RecursiveCharacterTextSplitter with chunk_size=512, overlap=50.
  • Embeddings via OpenAI text-embedding-3-small in batches of 100-200.
  • ChromaDB add in batches of 1K-2K with deterministic IDs and mandatory metadata.
  • Progress tracking with tqdm, error handling per batch, post-ingestion validation.
  • Document parameters in the README; use environment variables for configuration.
  • The resulting index is ready to connect to retrieval/generation in the next capsule.

Additional resources

Post-implementation checklist

Before moving to capsule 04 (API), verify:

  • The pipeline processes 100+ documents without errors.
  • All chunks have metadata with doc_id, source, chunk_index.
  • The IDs are deterministic (re-running doesn't duplicate if you delete the collection).
  • You have a progress bar or logs that indicate progress.
  • You documented batch_size, chunk_size, and chunk_overlap in the README.

Estimated time: 30-35 minutes
Next: 04-retrieval-generation-api.md

← PreviousCapsule 02: Project ArchitectureNext →Capsule 04: Retrieval, Generation, and API