Capsule 05: Batch Ingestion in ChromaDB — from concept to an operable pipeline

Capsule description

In M03/05 you learned why single inserts don't scale and the principles of a robust batch pipeline: idempotency, retry with backoff, observability. This capsule is the concrete implementation on ChromaDB. You'll build and benchmark an operable pipeline that ingests 10K-100K documents efficiently, handling the backend-specific details: when it's safe to parallelize (and when it isn't), how to measure the batch_size sweet spot empirically, what typical errors ChromaDB throws, and how to recover without losing progress.

When you finish, you'll have a reusable pipeline you can copy into your project and adapt — and you'll understand why each component is where it is.

By the end of this capsule, you'll be able to:

• ✅ Find the optimal batch_size for your hardware with a 5-minute benchmark
• ✅ Decide when to parallelize with ThreadPoolExecutor and when not to
• ✅ Implement the canonical pipeline with tqdm + retry + structured logging
• ✅ Handle ChromaDB-specific errors: duplicate ID, dimension mismatch, full disk
• ✅ Differentiate between EphemeralClient (testing) and PersistentClient (production) in the context of massive ingestion
• ✅ Anticipate the real bottleneck: many times it's not ChromaDB, it's the embeddings API

Estimated time: 30-35 minutes

Single insert vs batch — the quick demonstration

Before moving to the pipeline, a minimal demonstration of the impact. This is what happens with 1,000 documents:

import chromadb
import time

client = chromadb.Client()  # ephemeral, in-memory
collection = client.create_collection("demo_single")

docs = [f"Document {i}" for i in range(1000)]
ids = [f"doc_{i}" for i in range(1000)]

# Single inserts (one by one)
start = time.perf_counter()
for doc, doc_id in zip(docs, ids):
    collection.add(documents=[doc], ids=[doc_id])
single_time = time.perf_counter() - start

print(f"Single inserts (1000 docs): {single_time:.2f}s")
# Tens of seconds: with the default local embedding, each add() computes its embedding


# Batch insert (all together)
collection2 = client.create_collection("demo_batch")
start = time.perf_counter()
collection2.add(documents=docs, ids=ids)
batch_time = time.perf_counter() - start

print(f"Batch insert (1000 docs):    {batch_time:.2f}s")
print(f"Speedup: {single_time / batch_time:.0f}x")
# A few seconds; speedup ~7-10x with local embeddings (embedding compute dominates).
# The ~80x from M03/05 applies to pure insert overhead (with embeddings already precomputed).

Why it improves so much: we already covered it in M03/05 — each call to add() has constant overhead (validation, writing, index update) and, with local embeddings, computes each document's embedding. Making 1,000 calls pays that overhead 1,000 times and computes embeddings one at a time; a single call with 1,000 docs amortizes the overhead and batches the compute. With local embeddings the bottleneck is the model itself, so the speedup stays in single digits (~7-10x); the ~80x we saw in M03/05 shows up when the embedding is already precomputed and you only measure the insert overhead.

But there's an important detail for ChromaDB: the maximum reasonable batch depends on whether you use local embeddings (default) or remote ones (OpenAI via embedding_function).

Finding your batch_size sweet spot

The general recommendations (covered in M03/05) say 100-200 with OpenAI, 500-1000 local. But the sweet spot depends on your hardware, your model, and your dataset. Here's the script to find it:

# benchmark_batch_size.py
import chromadb
from chromadb.utils import embedding_functions
import os
import time

# Setup with local embeddings (faster for benchmarking)
client = chromadb.PersistentClient(path="./chroma_bench_batch")

# Generate 10K synthetic docs
total_docs = 10_000
docs = [
    f"This is document number {i} discussing technical topics like vector databases, "
    f"embedding models, and retrieval systems. Each document has approximately 30 words "
    f"to simulate typical chunk sizes in a real RAG application." for i in range(total_docs)
]
ids = [f"doc_{i:05d}" for i in range(total_docs)]


def benchmark_batch_size(batch_size: int) -> dict:
    """Measure throughput for a specific batch_size."""
    name = f"bench_bs_{batch_size}"
    try:
        client.delete_collection(name)
    except Exception:
        pass
    collection = client.create_collection(name)

    start = time.perf_counter()
    for i in range(0, total_docs, batch_size):
        end = min(i + batch_size, total_docs)
        collection.add(documents=docs[i:end], ids=ids[i:end])
    elapsed = time.perf_counter() - start

    throughput = total_docs / elapsed
    n_batches = (total_docs + batch_size - 1) // batch_size

    return {
        "batch_size": batch_size,
        "total_time_s": elapsed,
        "throughput_docs_per_s": throughput,
        "n_batches": n_batches,
    }


print(f"{'batch_size':>10} {'time (s)':>10} {'docs/s':>10} {'batches':>10}")
print("-" * 45)
for bs in [50, 100, 200, 500, 1000, 2000, 5000]:
    result = benchmark_batch_size(bs)
    print(
        f"{result['batch_size']:>10} "
        f"{result['total_time_s']:>10.2f} "
        f"{result['throughput_docs_per_s']:>10.0f} "
        f"{result['n_batches']:>10}"
    )

Typical output (Macbook M2, local embeddings):

batch_size   time (s)     docs/s    batches
---------------------------------------------
        50      18.40       543        200
       100      11.20       893        100
       200       6.85      1460         50
       500       4.30      2326         20
      1000       3.85      2597         10
      2000       3.62      2762          5
      5000       4.10      2439          2

Reading:

• Throughput grows fast between 50 and 500 (overhead amortized).
• Sweet spot at batch_size=2000 — ~2,762 docs/second.
• At batch_size=5000 throughput drops slightly — probably memory swap or GC pressure.

Your output will be different. Your hardware may prefer 1000 (tight RAM) or 5000 (ample RAM). That's why the benchmark on your machine is worth more than copying numbers from a tutorial.

With OpenAI embeddings, the pattern changes

If you use OpenAIEmbeddingFunction instead of local embeddings, the bottleneck stops being ChromaDB and becomes the OpenAI API:

openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.getenv("OPENAI_API_KEY"),
    model_name="text-embedding-3-small"
)

OpenAI constraints:

• Maximum ~8191 tokens per request
• Rate limits per tier (tier 1: ~3000 RPM, tier 2: ~5000 RPM)

Typical output with OpenAI:

batch_size   time (s)     docs/s    batches
---------------------------------------------
        50      32.10       312        200
       100      18.50       541        100
       200      12.20       820         50
       500       9.40      1064         20
      1000      11.20       893         10  ← now hits rate limits
      2000     ERROR (token limit)

With OpenAI, the sweet spot drops to ~200-500 docs/batch. Going over generates 429 errors (rate limit) or 400 (max tokens). That's why the general recommendation "200 with OpenAI" — it's calibrated not to break.

The canonical pipeline

Here's the implementation you'll copy into your project. It combines batch + tqdm + retry + idempotency + handling of ChromaDB-specific errors.

# ingestion_pipeline.py
import chromadb
from chromadb.utils import embedding_functions
import os
import time
from dataclasses import dataclass, field
from tqdm import tqdm

# Specific errors we'll handle
try:
    from openai import RateLimitError, APIError, APITimeoutError
except ImportError:
    RateLimitError = APIError = APITimeoutError = Exception


@dataclass
class IngestionStats:
    total_input: int = 0
    inserted: int = 0
    skipped_existing: int = 0
    failed: list[int] = field(default_factory=list)
    duration_seconds: float = 0.0

    @property
    def throughput_per_second(self) -> float:
        return self.inserted / self.duration_seconds if self.duration_seconds > 0 else 0

    def report(self):
        print(f"\nIngestion stats:")
        print(f"  Input total:       {self.total_input}")
        print(f"  Inserted:          {self.inserted}")
        print(f"  Skipped (existing): {self.skipped_existing}")
        print(f"  Failed batches:    {len(self.failed)}")
        print(f"  Duration:          {self.duration_seconds:.1f}s")
        print(f"  Throughput:        {self.throughput_per_second:.0f} docs/s")


def ingest_to_chromadb(
    collection: chromadb.Collection,
    documents: list[str],
    ids: list[str],
    metadatas: list[dict] | None = None,
    batch_size: int = 200,
    sleep_between_batches: float = 0.0,
    max_retries: int = 3,
    skip_existing: bool = True,
) -> IngestionStats:
    """
    Robust ingestion pipeline for ChromaDB.

    - Idempotent: with `skip_existing=True`, IDs already present are skipped.
    - Recoverable: each batch has retry with exponential backoff.
    - Observable: progress bar + final stats.
    - Tolerates partial failures: if a batch fails all retries, it's recorded and it continues.
    """
    assert len(documents) == len(ids), "documents and ids must have same length"
    if metadatas is not None:
        assert len(metadatas) == len(ids), "metadatas length mismatch"

    stats = IngestionStats(total_input=len(ids))
    start = time.perf_counter()

    # Filter existing IDs (idempotency)
    if skip_existing:
        existing = collection.get(ids=ids, include=[])
        existing_ids = set(existing['ids'])
        if existing_ids:
            print(f"Skipping {len(existing_ids)} existing IDs")
        new_indices = [i for i, doc_id in enumerate(ids) if doc_id not in existing_ids]
        stats.skipped_existing = len(ids) - len(new_indices)
    else:
        new_indices = list(range(len(ids)))

    # Process in batches
    n_batches = (len(new_indices) + batch_size - 1) // batch_size

    for batch_num in tqdm(range(n_batches), desc="Ingesting", unit="batch"):
        b_start = batch_num * batch_size
        b_end = min(b_start + batch_size, len(new_indices))
        batch_indices = new_indices[b_start:b_end]

        batch_docs = [documents[i] for i in batch_indices]
        batch_ids = [ids[i] for i in batch_indices]
        batch_metas = [metadatas[i] for i in batch_indices] if metadatas else None

        # Retry with exponential backoff
        success = False
        for attempt in range(max_retries):
            try:
                kwargs = {"documents": batch_docs, "ids": batch_ids}
                if batch_metas is not None:
                    kwargs["metadatas"] = batch_metas
                collection.add(**kwargs)
                stats.inserted += len(batch_ids)
                success = True
                break

            except RateLimitError:
                wait = (2 ** attempt) * 5  # 5, 10, 20s
                print(f"\n  Rate limit (batch {batch_num+1}), waiting {wait}s...")
                time.sleep(wait)

            except (APIError, APITimeoutError) as e:
                wait = (2 ** attempt) * 2
                print(f"\n  API error (batch {batch_num+1}): {e}, waiting {wait}s...")
                time.sleep(wait)

            except ValueError as e:
                # ChromaDB throws ValueError for schema problems (dimension mismatch, etc.)
                print(f"\n  Schema error (batch {batch_num+1}): {e}")
                # No point retrying schema errors
                break

            except Exception as e:
                wait = 1
                print(f"\n  Unexpected error (batch {batch_num+1}): {e}, waiting {wait}s...")
                time.sleep(wait)

        if not success:
            stats.failed.append(batch_num + 1)

        if sleep_between_batches > 0:
            time.sleep(sleep_between_batches)

    stats.duration_seconds = time.perf_counter() - start
    return stats

Testing it

# test_pipeline.py
import os
import chromadb
from chromadb.utils import embedding_functions

openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.getenv("OPENAI_API_KEY"),
    model_name="text-embedding-3-small"
)

client = chromadb.PersistentClient(path="./chroma_test")
collection = client.get_or_create_collection(
    name="pipeline_test",
    embedding_function=openai_ef,
    metadata={"hnsw:space": "cosine", "hnsw:M": 32, "hnsw:construction_ef": 200},
)

# Generate dataset
docs = [f"Document {i} about technical topics in AI engineering." for i in range(2000)]
ids = [f"doc_{i:05d}" for i in range(2000)]
metas = [{"index": i, "category": "demo"} for i in range(2000)]

stats = ingest_to_chromadb(
    collection,
    documents=docs,
    ids=ids,
    metadatas=metas,
    batch_size=200,
    sleep_between_batches=0.3,  # Margin for OpenAI rate limits
    max_retries=3,
)
stats.report()

Expected output on the first run:

Skipping 0 existing IDs
Ingesting: 100%|██████████| 10/10 [00:32<00:00, 3.27s/batch]

Ingestion stats:
  Input total:       2000
  Inserted:          2000
  Skipped (existing): 0
  Failed batches:    0
  Duration:          32.7s
  Throughput:        61 docs/s

Output on re-run (idempotency):

Skipping 2000 existing IDs
Ingesting: 0it [00:00, ?it/s]

Ingestion stats:
  Input total:       2000
  Inserted:          0
  Skipped (existing): 2000
  Failed batches:    0
  Duration:          0.4s
  Throughput:        0 docs/s

The second execution finishes in 400ms because it checked existing IDs before inserting. That's operational idempotency.

When to parallelize with ThreadPoolExecutor?

Intuition says "more threads = faster". For ChromaDB, it's almost always false. You have to understand why.

The persistence backend limits concurrency

ChromaDB PersistentClient uses SQLite underneath. SQLite internally serializes concurrent writes — only one write transaction can be active at a time. Launching 8 threads that insert concurrently gives the same result as 1 thread, but with extra coordination overhead.

Benchmark over PersistentClient:

from concurrent.futures import ThreadPoolExecutor

def ingest_one_batch(args):
    collection, batch_docs, batch_ids = args
    collection.add(documents=batch_docs, ids=batch_ids)

# Sequential
collection_seq = client.get_or_create_collection("seq")
batches = [(collection_seq, docs[i:i+200], ids[i:i+200]) for i in range(0, 2000, 200)]
start = time.perf_counter()
for batch in batches:
    ingest_one_batch(batch)
seq_time = time.perf_counter() - start

# Concurrent (4 workers)
collection_conc = client.get_or_create_collection("conc")
batches_conc = [(collection_conc, docs[i:i+200], ids[i:i+200]) for i in range(0, 2000, 200)]
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=4) as executor:
    list(executor.map(ingest_one_batch, batches_conc))
conc_time = time.perf_counter() - start

print(f"Sequential: {seq_time:.2f}s")
print(f"Concurrent (4 workers): {conc_time:.2f}s")
print(f"Speedup: {seq_time/conc_time:.2f}x")

Typical output (PersistentClient):

Sequential: 8.40s
Concurrent (4 workers): 7.80s
Speedup: 1.08x   ← marginal, almost nothing

When DOES it parallelize with benefit?

When the bottleneck is outside ChromaDB, specifically the remote embeddings API. If each batch calls OpenAI for embeddings, that call is I/O-bound (network wait). While one thread waits, others can do work.

# With remote embeddings (OpenAI), parallelization DOES help
# 4 threads can have 4 requests pending simultaneously
# Each one waits ~150ms for OpenAI; with 4 in parallel it effectively becomes 40ms

# Typical output with OpenAI + 4 workers:
# Sequential:  120s
# Concurrent:   45s
# Speedup:    2.7x

Practical rule

Careful with parallelizing wrong:

• Too many workers with OpenAI → rate limit crossed between threads, 429 errors.
• Workers with different EphemeralClients → each has its DB in memory; the data isn't shared.
• Workers writing to the same collection without a lock → ChromaDB serializes it but you lose time on contention.

ChromaDB-specific errors and how to handle them

Error 1: dimension mismatch

ValueError: Embedding dimension 1536 does not match collection dimensionality 384

Cause: you're passing an embedding (or using an embedding_function) that produces vectors of a different size than the collection expects.

When it happens:

• You changed embedding_function (e.g.: from default ChromaDB 384-dim to OpenAI 1536-dim) without re-creating the collection.
• You pass embeddings=... directly with vectors of the wrong size.

How to prevent it: use a single source of truth for the embedding model. If the collection was created with OpenAIEmbeddingFunction, all queries and inserts must use the same embedding_function.

How to recover: create a new collection with the correct dimensionality and re-insert.

Error 2: duplicate ID

chromadb.errors.IDAlreadyExistsError: ID 'doc_42' already exists

Cause: you're inserting an ID that already exists in the collection.

When it happens:

• You re-run the pipeline without idempotency.
• You generate non-deterministic IDs (random UUID) and a retry inserts the same doc twice.

How to prevent it: the canonical pipeline above already covers it with skip_existing=True.

How to recover: either filter the existing IDs (idempotency), or use collection.upsert() instead of collection.add(). upsert updates if the ID exists.

Error 3: full disk

sqlite3.OperationalError: database or disk is full

Cause: PersistentClient writes to SQLite + parquet on disk. When the disk fills up, it fails.

When it happens: large datasets (>1M vectors) on machines with little space.

How to prevent it: monitor disk space. Estimate storage:

storage ≈ N × D × 4 bytes (vectors) + SQLite overhead

For 1M vectors × 1536 dim:
  vectors: 6 GB
  HNSW index: ~3 GB
  metadata + IDs: ~500 MB
  total: ~10 GB on disk

How to recover: free space, or migrate to larger storage, or move to a distributed vector DB.

Error 4: out of memory during the build

Cause: you insert too many vectors with EphemeralClient (everything in RAM) or the HNSW graph exceeds available RAM.

When it happens:

• 5M+ vectors with M=32 on a 16 GB machine.
• Huge batch_size with large embeddings.

How to prevent it: estimate RAM as we saw in M04/03. If it exceeds, lower M or migrate to a vector DB with persistent storage.

Error 5: ChromaDB server unavailable (HTTP mode)

chromadb.errors.ChromaError: HTTP request failed: connection refused

Cause: you're using chromadb.HttpClient(host=..., port=...) and the server is down or not accepting connections.

How to handle it: retry with backoff, healthcheck before starting the pipeline, alert if the server doesn't respond after N retries.

Traps and common mistakes

Trap 1: using EphemeralClient for massive ingestion

The mistake:

client = chromadb.Client()  # ephemeral, all in RAM
collection = client.create_collection("docs")
collection.add(documents=[...] * 1_000_000, ids=[...])  # OOM!

Symptom: OOM kill when the dataset grows.

How to prevent it: EphemeralClient is only for tests and small prototypes. For any real ingestion, PersistentClient with a path on disk.

client = chromadb.PersistentClient(path="./chroma_db")

Trap 2: ignoring the cost of re-running the pipeline

The mistake: the pipeline runs 30 minutes. It fails at 95%. You re-run from scratch. You pay 30 more minutes + the cost of duplicated embeddings.

How to prevent it: idempotency with skip_existing=True. If you re-run, the 95% already inserted are skipped in seconds.

Trap 3: parallelizing with PersistentClient without measuring

The mistake: you copy ThreadPoolExecutor(max_workers=8) from a tutorial. You see the "elegant" code and deploy it.

Symptom: throughput equal to or worse than sequential. Logs show SQLite contention.

How to prevent it: benchmark sequential vs concurrent with your specific setup before deciding. The practical rule from the table above.

Trap 4: forgetting the embedding_function when retrieving the collection

The mistake:

# Day 1: create with OpenAI ef
collection = client.create_collection("docs", embedding_function=openai_ef)
collection.add(documents=docs, ids=ids)

# Day 2: retrieve WITHOUT the embedding_function
collection_v2 = client.get_collection("docs")  # ❌
collection_v2.add(documents=more_docs, ids=more_ids)
# Uses ChromaDB's default embedding → dimension mismatch

How to prevent it: always pass the same embedding_function when retrieving the collection:

collection_v2 = client.get_collection("docs", embedding_function=openai_ef)

Trap 5: ingestion without progress tracking → killing healthy processes

The mistake: the script runs without tqdm or logs. 25 minutes with no output. The operator thinks it hung. kill -9.

Symptom: lost work.

How to prevent it: always tqdm or per-batch logging. It's 1 line of code and prevents this scenario.

Trap 6: not separating the embeddings pipeline from the insertion one

The mistake: a single function does the document download, chunking, embedding, and insertion. When something fails, you don't know where.

How to prevent it: separate the phases in the pipeline:

[download docs] → [chunk] → [embed] → [insert into ChromaDB]
   pipeline 1     pipeline 2  pipeline 3     pipeline 4

Each phase has its own error tolerance and can be run/retried independently. If embedding fails, you don't redo the chunking.

Applied exercise

Scenario: you're given a dataset of 50,000 PDFs. Specifications:

• Each PDF has 5-30 pages. They're already chunked (we saw chunking in M4/10) → ~150,000 total chunks.
• Embedding model: OpenAI text-embedding-3-small.
• Hardware: laptop with 16 GB RAM, MacOS, stable connection.
• OpenAI tier 2 (5000 RPM, 5M tokens/min).
• Budget: $20 USD for embeddings.
• Operational SLA: the ingestion must finish in ≤2 hours (they gave you one morning of work).
• Robustness: if the script crashes (network drop, laptop sleep), it must be able to resume without duplicating.

Your task: implement the pipeline (or adapt the canonical one above) with a justified configuration. Calculate the estimated cost and time, and adjust if it doesn't fit the budget.

total_chunks = 150_000
avg_tokens_per_chunk = 500  # estimate for ~400-char chunks
total_tokens = total_chunks * avg_tokens_per_chunk  # 75M tokens

# Cost
cost_per_million_tokens = 0.02
cost = total_tokens / 1_000_000 * cost_per_million_tokens
# = $1.50 — fits comfortably within $20

# Required throughput
target_minutes = 120
required_chunks_per_minute = total_chunks / target_minutes  # 1,250 chunks/min
required_tokens_per_minute = total_tokens / target_minutes  # 625K tokens/min

print(f"Estimated cost: ${cost}")
print(f"Required throughput: {required_chunks_per_minute} chunks/min")
print(f"Required tokens/min: {required_tokens_per_minute:.0f}")
print(f"Tier 2 allows 5,000,000 tokens/min → 8x margin")

import chromadb
from chromadb.utils import embedding_functions
import os
from concurrent.futures import ThreadPoolExecutor
import time

openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.getenv("OPENAI_API_KEY"),
    model_name="text-embedding-3-small"
)

client = chromadb.PersistentClient(path="./chroma_pdfs")
collection = client.get_or_create_collection(
    name="pdf_chunks",
    embedding_function=openai_ef,
    metadata={
        "hnsw:space": "cosine",
        "hnsw:M": 32,
        "hnsw:construction_ef": 200,
        "hnsw:search_ef": 50,
    },
)

# Pipeline with controlled parallelism (OpenAI is I/O bound → parallelizes well)
def ingest_chunks_parallel(chunks, ids, metadatas, batch_size=200, workers=3):
    """Pipeline with 3 parallel workers to amortize OpenAI latency."""

    # Idempotency: filter existing IDs
    existing = collection.get(ids=ids, include=[])
    existing_set = set(existing['ids'])
    new_indices = [i for i, doc_id in enumerate(ids) if doc_id not in existing_set]

    print(f"Total: {len(ids)}, already exist: {len(existing_set)}, to insert: {len(new_indices)}")

    # Create batches
    batches = []
    for batch_start in range(0, len(new_indices), batch_size):
        batch_idx = new_indices[batch_start:batch_start + batch_size]
        batches.append({
            "documents": [chunks[i] for i in batch_idx],
            "ids": [ids[i] for i in batch_idx],
            "metadatas": [metadatas[i] for i in batch_idx],
        })

    def process_batch(batch_data):
        for attempt in range(3):
            try:
                collection.add(**batch_data)
                return True
            except Exception as e:
                if attempt < 2:
                    time.sleep((2 ** attempt) * 5)  # 5, 10s
                else:
                    print(f"FAILED batch after retries: {e}")
                    return False
        return False

    start = time.perf_counter()
    success = 0
    failed = 0

    with ThreadPoolExecutor(max_workers=workers) as executor:
        results = list(executor.map(process_batch, batches))

    success = sum(1 for r in results if r)
    failed = sum(1 for r in results if not r)
    elapsed = time.perf_counter() - start

    print(f"\nCompleted in {elapsed/60:.1f} minutes")
    print(f"Successful batches: {success}/{len(batches)}")
    print(f"Failed batches: {failed}")
    print(f"Throughput: {len(new_indices)/elapsed:.0f} docs/s")

# Sanity check
print(f"Total docs in collection: {collection.count()}")
# Expected: 150,000

# Test query
result = collection.query(
    query_texts=["how do I configure HNSW?"],
    n_results=5,
)
for doc, meta in zip(result['documents'][0], result['metadatas'][0]):
    print(f"  [{meta['source']}] {doc[:80]}...")

Summary and next step

What you learned:

• Single inserts in ChromaDB are dramatically slower than batch — the sweet spot is 1000-2000 docs with local embeddings, 100-500 with OpenAI.
• Finding the optimal batch_size for your hardware is worth 5 minutes of benchmarking — don't copy numbers from tutorials.
• Parallelizing with ThreadPoolExecutor only helps when the bottleneck is remote I/O (OpenAI). With PersistentClient + local embeddings, it doesn't help.
• The canonical pipeline combines: idempotency (skip existing IDs), retry with backoff, handling of ChromaDB-specific errors, observability with tqdm.
• Typical ChromaDB errors and how to handle them: dimension mismatch, duplicate ID, full disk, OOM.
• EphemeralClient is for tests; production uses PersistentClient.
• Separating the pipeline into phases (download → chunking → embedding → insertion) improves recovery from failures.

Checkpoint: before moving on, you should be able to:

• Find the optimal batch_size with a 5-minute benchmark on your hardware.
• Decide how many workers to use based on local vs remote embeddings.
• Implement the canonical pipeline with idempotency and error handling.

Next capsule: 06 — Query Optimization.

You just put data into the system efficiently. Now you'll optimize the other side: the queries. You'll learn to measure latency with percentiles (not averages), to understand which parameters move the needle (n_results, ef_search, metadata filtering), and to benchmark changes before deploying them.

Resources

1. ChromaDB — Adding Data — Insertion operations
2. ChromaDB — Persistent vs Ephemeral Client — Differences and when to use each
3. OpenAI — Rate Limits Guide — Limits per tier
4. tqdm Documentation — Progress bars
5. Tenacity — Retry Library — A more robust alternative to manual retry
6. SQLite Concurrency Internals — Why parallelizing PersistentClient gives little speedup

Estimated time: 30-35 minutes Next: 06-query-optimization.md