Capsule 05: Batch Operations — the hidden cost of naive ingestion

Capsule description

The first instinct when you have 1 million documents to index is to write a loop:

for doc in documents:
    collection.add(documents=[doc.text], ids=[doc.id], metadatas=[doc.metadata])

That works perfectly for 100 documents. It fails catastrophically for 100,000. Not because of a bug — because of architecture. Each add() call with a single document triggers overhead that is independent of how much data you're inserting: serialization, validation, writing to storage, updating the HNSW index, fsync. If you make 1 million calls, you multiply that overhead 1 million times.

Batch operations are the difference between your ingestion finishing in 2 minutes or in 3 hours. It's not premature optimization — it's the difference between an operable pipeline and one that breaks the team's workflow every time new documents arrive.

This capsule gives you the mental model to understand why single inserts fail at scale, the criteria for choosing the right batch size for your case, and the most expensive error patterns (the one that silently destroys data when a batch fails halfway through).

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

• ✅ Calculate the real cost (in time and money) of ingesting N documents with single inserts vs batches
• ✅ Choose a batch_size with judgment based on available RAM, embedding cost, and acceptable latency
• ✅ Identify the three failure points in batch ingestion pipelines: rate limits, OOM, mid-batch crash
• ✅ Design idempotency so that retrying batches doesn't duplicate documents
• ✅ Anticipate the most expensive failure mode: a batch that looks successful but silently loses documents

Estimated time: 30-40 minutes

Why single inserts don't scale

Let's do the calculation that almost nobody does before discovering it in production.

Anatomy of a single insert

When you call collection.add(documents=[doc], ids=[id]), ChromaDB does these operations:

1. Validation of the input (~0.5ms): types, unique IDs, valid metadata.
2. Embedding generation via the configured embedding_function:
   • Default ChromaDB local: ~5-15ms per document.
   • OpenAI API: ~80-150ms per call (includes network latency + API processing).
3. Write to storage (SQLite or persistent backend): ~3-8ms.
4. HNSW index update (recomputing the graph's connections): ~1-3ms.
5. fsync optional to guarantee durability: ~5-10ms.

Total: ~10-25ms with a local embedding, ~85-180ms with OpenAI.

Multiplied by volume:

The insight: what makes the time explode isn't the actual processing of the data — it's the constant overhead per call. Generating the embedding for one document costs 100ms. Generating embeddings for 100 documents in a single API call costs ~150ms total (200ms with network latency, once). In other words, in a batch you get 100 embeddings for almost the same time it costs to get a single one.

The API cost calculation

For OpenAI specifically, there's an additional monetary cost to doing single inserts:

# Calculator of cost per API calls
COST_PER_REQUEST_OVERHEAD_SECONDS = 0.1  # latency + processing per call
COST_PER_TOKEN = 0.02 / 1_000_000  # USD per token (text-embedding-3-small)
TOKENS_PER_DOC_AVG = 500

scenarios = {
    "single_insert": {"requests": 100_000, "tokens_per_request": 500},
    "batch_size_10": {"requests": 10_000, "tokens_per_request": 5_000},
    "batch_size_100": {"requests": 1_000, "tokens_per_request": 50_000},
    "batch_size_500": {"requests": 200, "tokens_per_request": 250_000},
}

for name, s in scenarios.items():
    total_tokens = s["requests"] * s["tokens_per_request"]
    monetary_cost = total_tokens * COST_PER_TOKEN
    time_overhead = s["requests"] * COST_PER_REQUEST_OVERHEAD_SECONDS

    print(f"\n=== {name} ===")
    print(f"  API requests: {s['requests']:,}")
    print(f"  Monetary cost: ${monetary_cost:.2f}")
    print(f"  Overhead time: {time_overhead/60:.1f} minutes")

Output:

=== single_insert ===
  API requests: 100,000
  Monetary cost: $1.00
  Overhead time: 166.7 minutes

=== batch_size_10 ===
  API requests: 10,000
  Monetary cost: $1.00
  Overhead time: 16.7 minutes

=== batch_size_100 ===
  API requests: 1,000
  Monetary cost: $1.00
  Overhead time: 1.7 minutes

=== batch_size_500 ===
  API requests: 200
  Monetary cost: $1.00
  Overhead time: 0.3 minutes

Key takeaway: the monetary cost is the same because it depends on total tokens, not on the number of requests. But the overhead time drops from 2.7 hours (single insert) to 20 seconds (batch=500). And there's still another hidden cost: each request can fail (rate limit, timeout) — more requests = more chances to fail.

How to choose the right batch_size

There's no magic value. There are three constraints that intersect.

Constraint 1: API rate limits

OpenAI allows up to 8191 tokens per request in text-embedding-3-small, and a maximum number of elements in the input array per request (~2048 for text). If your documents average 500 tokens:

• Theoretical max batch by tokens: 8191 / 500 ≈ 16 docs
• Theoretical max batch by elements: 2048 docs

The bottleneck is tokens, not elements. For text-embedding-3-small, batches of 100-200 documents are safe.

For other models (Cohere, local Sentence Transformers), check the specific documentation.

Constraint 2: Available RAM

The entire batch lives in memory while it's being processed:

# Approximation of RAM consumed per batch
batch_size = 1000
embedding_dim = 1536
bytes_per_float = 4  # float32

ram_per_batch = batch_size * embedding_dim * bytes_per_float
# = 1000 * 1536 * 4 = 6.144 MB per batch

# If you process in parallel (4 workers):
total_ram = ram_per_batch * 4  # = 24.6 MB

That's trivial for batches of 1000. But if you do batch_size=100_000 with 4096-dim vectors (large models) and 16 parallel workers, you consume ~26 GB just in embeddings — and that's without counting metadata, IDs, and Python's overhead.

Practical rule: keep the RAM per batch (including parallel workers) under 30% of your machine's available RAM.

Constraint 3: Failure latency

If a batch fails halfway through processing, how much work do you lose? With batch_size=10000, a timeout makes you restart 10000 docs. With batch_size=200, only 200.

Clear trade-off:

• Large batch (1000+): better throughput, worse error recovery.
• Small batch (50-200): better recovery, more overhead.

Recommendations by scenario

Reasonable default if you don't know: batch_size=200 with OpenAI, batch_size=500 local.

The right pipeline: idempotent, observable, recoverable

# pipeline_batch_ingestion.py
import os
import time
from dataclasses import dataclass
from tqdm import tqdm
from openai import RateLimitError, APIError
import chromadb
from chromadb.utils import embedding_functions


@dataclass
class IngestionResult:
    total_docs: int
    inserted: int
    skipped: int
    failed_batches: list[int]
    duration_seconds: float


def ingest_documents_batched(
    collection,
    documents: list[str],
    metadatas: list[dict],
    ids: list[str],
    batch_size: int = 200,
    sleep_between: float = 0.0,
    max_retries: int = 3,
) -> IngestionResult:
    """
    Ingestion pipeline with three guarantees:
    - Idempotent: if an id already exists, it isn't inserted (no error).
    - Observable: progress bar + logging per batch.
    - Recoverable: if a batch fails, it's retried before being abandoned.
    """
    assert len(documents) == len(metadatas) == len(ids), (
        "documents, metadatas, ids must have the same length"
    )

    start = time.time()
    inserted = 0
    skipped = 0
    failed_batches = []

    # Check for already-existing IDs (idempotency)
    existing_response = collection.get(ids=ids, include=[])
    existing_ids = set(existing_response['ids'])
    if existing_ids:
        print(f"  {len(existing_ids)} IDs already exist, they will be skipped")

    new_indices = [i for i, doc_id in enumerate(ids) if doc_id not in existing_ids]
    skipped = len(ids) - len(new_indices)

    # Process in batches with a progress bar
    n_batches = (len(new_indices) + batch_size - 1) // batch_size

    for batch_num in tqdm(range(n_batches), desc="Ingesting batches"):
        batch_start = batch_num * batch_size
        batch_end = min(batch_start + batch_size, len(new_indices))
        batch_indices = new_indices[batch_start:batch_end]

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

        # Retry with exponential backoff
        success = False
        for attempt in range(max_retries):
            try:
                collection.add(
                    documents=batch_docs,
                    metadatas=batch_metas,
                    ids=batch_ids,
                )
                inserted += len(batch_ids)
                success = True
                break

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

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

            except Exception as e:
                print(f"\n  Unexpected error on batch {batch_num+1}: {e}")
                break

        if not success:
            failed_batches.append(batch_num + 1)
            print(f"\n  ❌ Batch {batch_num+1} failed after {max_retries} attempts")

        if sleep_between > 0:
            time.sleep(sleep_between)

    duration = time.time() - start

    return IngestionResult(
        total_docs=len(ids),
        inserted=inserted,
        skipped=skipped,
        failed_batches=failed_batches,
        duration_seconds=duration,
    )

Testing the pipeline

# test_pipeline.py
openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.getenv("OPENAI_API_KEY"),
    model_name="text-embedding-3-small"
)
client = chromadb.PersistentClient(path="./chroma_batch_test")
collection = client.get_or_create_collection(
    name="batch_demo",
    embedding_function=openai_ef
)

# Generate a synthetic dataset
docs = [f"Document number {i} about vector databases and RAG systems." for i in range(1000)]
metas = [{"index": i, "category": "test"} for i in range(1000)]
ids = [f"doc_{i:04d}" for i in range(1000)]

result = ingest_documents_batched(
    collection=collection,
    documents=docs,
    metadatas=metas,
    ids=ids,
    batch_size=200,
    sleep_between=0.5,
    max_retries=3,
)

print(f"\nResult:")
print(f"  Total: {result.total_docs}")
print(f"  Inserted: {result.inserted}")
print(f"  Skipped (already existed): {result.skipped}")
print(f"  Failed batches: {result.failed_batches}")
print(f"  Duration: {result.duration_seconds:.1f}s ({result.inserted/result.duration_seconds:.0f} docs/s)")

Expected output (first run):

  0 IDs already exist, they will be skipped
Ingesting batches: 100%|█████████| 5/5 [00:14<00:00,  2.96s/batch]

Result:
  Total: 1000
  Inserted: 1000
  Skipped (already existed): 0
  Failed batches: []
  Duration: 14.8s (68 docs/s)

Output when re-run (idempotency):

  1000 IDs already exist, they will be skipped
Ingesting batches: 0it [00:00, ?it/s]

Result:
  Total: 1000
  Inserted: 0
  Skipped (already existed): 1000
  Failed batches: []
  Duration: 0.3s (0 docs/s)

The second run doesn't duplicate anything and finishes in 300ms because it checked for existing IDs before inserting. That's idempotency: running the pipeline twice produces the same final state as running it once.

Traps and common errors

Trap 1: batch_size too large hits rate limits or OOM

The error: you see benchmarks that say "more batch = more throughput" and you crank batch_size=5000 with OpenAI.

Symptom A: RateLimitError or BadRequestError: max tokens exceeded errors. The batch never gets inserted.

Symptom B (more subtle): the batch works locally but in CI with less RAM, OOM kills the process halfway through ingestion.

Why it happens: your pipeline exceeds the limits of the API (8191 tokens per request) or of the execution environment.

How to prevent it:

1. Calculate tokens per batch before running:

import tiktoken
enc = tiktoken.encoding_for_model("text-embedding-3-small")

batch = documents[0:200]
total_tokens = sum(len(enc.encode(d)) for d in batch)
print(f"Batch tokens: {total_tokens}")
# If > 8000, reduce batch_size

2. Measure RAM used locally before deploying to CI.

Trap 2: pipeline without idempotency → massive duplication

The error: a batch fails halfway through ingestion. The operator relaunches the script. The successful documents from the first attempt + all of the second attempt → duplicates.

Symptom: queries return results with almost-consecutive identical scores (the same documents with different IDs). collection.count() is higher than expected.

Why it happens: the pipeline assumes "empty state at the start". If that's not the case, it fails.

How to prevent it: the implementation above covers it — check collection.get(ids=...) before inserting and only insert the new IDs.

Trap 3: not handling partial batch failures

The error: a batch of 200 docs fails with a timeout. The code assumes "the whole batch failed" and marks them for retry. But ChromaDB internally already wrote 150 of 200 before the timeout.

Symptom: the retry inserts all 200 again, generating 150 duplicates.

Why it happens: ChromaDB doesn't have transactions — each add() can be partially successful.

How to prevent it:

• If you have idempotency by ID (recommended), the retry is safe.
• If your pipeline generates new IDs on each run (a bad pattern), fix it: the IDs must be deterministic from the document (hash, stable doc_id).

# ❌ Non-deterministic IDs
import uuid
ids = [str(uuid.uuid4()) for _ in documents]

# ✅ Deterministic IDs from the content
import hashlib
ids = [
    hashlib.md5(doc.encode()).hexdigest() for doc in documents
]
# Or using stable metadata
ids = [f"doc_{meta['source']}_{meta['chunk_index']}" for meta in metadatas]

Trap 4: ignoring progress, no visibility

The error: a loop with 50,000 iterations without a progress bar or logs. The script runs for 30 minutes, looks hung, and someone kills it thinking it's deadlocked.

Symptom: lost work, frustration.

How to prevent it: always tqdm or an equivalent in batch ingestion pipelines. It's 1 line of code and prevents killing healthy processes.

Trap 5: sleep_between=0 with OpenAI in production

The error: processing 100 consecutive batches without a pause. OpenAI has rate limits per minute (RPM, requests per minute). If you make 100 requests in 10 seconds, you're going to hit the rate limit.

Symptom: the first 30% of the pipeline works, then 429 errors start.

How to prevent it: add time.sleep(0.5-1.0) between large batches. You lose <5% of total throughput and avoid the unnecessary retries.

Trap 6: parallel batch ingestion without concurrency control

The error: "faster = more threads". You run 16 workers with batch_size=1000 each.

Symptom: rate limit 429 on ALL workers simultaneously. Some workers enter backoff, others keep going, the system is unpredictable. ChromaDB locks on concurrent writes to SQLite.

How to prevent it:

• For OpenAI: respect your tier's rate limit. Tier 1 allows ~3000 RPM, which with batch_size=200 is ~15 batches/second. With 4 parallel workers you're at the limit.
• For ChromaDB persistent: concurrent writes to SQLite are serialized internally, so more workers doesn't speed things up beyond 2-3.

# Controlled concurrency
from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = [
        executor.submit(ingest_batch, batch)
        for batch in chunks(documents, batch_size=200)
    ]
    for future in futures:
        future.result()  # awaits and re-raises

Applied exercise

Scenario: you work at a news company that needs to ingest 200,000 articles into the RAG system. Characteristics:

• Each article: ~800 tokens average (some up to 4000)
• Embeddings model: OpenAI text-embedding-3-small
• OpenAI tier: Tier 2 (5000 RPM, 5M tokens/minute)
• RAM of the ingestion machine: 16 GB
• Operational constraint: ingestion can't take more than 90 minutes (maintenance window)
• Robustness constraint: if the script crashes, it must be resumable without duplicating anything

Question: design the pipeline configuration (batch_size, parallelism, sleep, idempotency) and justify each decision with numbers.

total_docs = 200_000
avg_tokens_per_doc = 800
total_tokens = total_docs * avg_tokens_per_doc  # 160M tokens

# Monetary cost
cost = (total_tokens / 1_000_000) * 0.02  # = $3.20

# Required throughput
target_minutes = 90
required_docs_per_minute = total_docs / target_minutes  # 2,222 docs/min
required_tokens_per_minute = total_tokens / target_minutes  # 1.78M tokens/min

print(f"Cost: ${cost}")
print(f"Required throughput: {required_docs_per_minute} docs/min")
print(f"Required tokens/min: {required_tokens_per_minute:.0f}")
print(f"Tier 2 allows: 5,000,000 tokens/min → 2.8x margin")

avg_tokens_per_batch = 100 * 800  # 80,000 tokens — OK
worst_tokens_per_batch = 100 * 4000  # 400,000 tokens
# Per individual document: 4000 < 8191 → OK

def article_to_id(article):
    """Stable ID based on the article's ID in the source."""
    return f"news_{article['source_id']}_{article['version']}"

config = {
    "batch_size": 100,
    "max_workers": 3,
    "sleep_between": 0.2,
    "max_retries": 3,
    "retry_backoff": "exponential",  # 5s, 10s, 20s
    "log_failed_batches_to": "./failed_batches.json",
}

ram_per_batch = 100 * 1536 * 4  # 614 KB per batch
ram_total = ram_per_batch * 3   # 1.8 MB with 3 workers
# Trivial for a 16 GB machine

Summary and next step

What you learned:

• Single inserts don't scale. The constant overhead per call multiplies linearly with the number of documents, turning 100 docs (feasible) into 1M docs (days).
• Batch ingestion amortizes the overhead. With batch_size=200 and OpenAI, throughput jumps from ~10 docs/second to ~150 docs/second.
• The right batch_size depends on three constraints: API limits (tokens per request), available RAM, and acceptable failure latency.
• Reasonable defaults: 100-200 with OpenAI, 500 local with CPU, 1000+ with GPU. But always check the limits of your specific model.
• A robust pipeline requires idempotency (deterministic IDs + prior check) + retry with backoff + observability (progress bar + logs per batch).
• Most expensive failure modes: duplication from a re-run without idempotency, OOM from a batch that's too large, rate limits from uncontrolled concurrency.

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

• Calculate the total expected ingestion time for N documents with a given batch_size.
• Justify the choice of batch_size for a specific scenario, citing the three trade-offs (API, RAM, failure).
• Design idempotency so that a pipeline can resume after a crash without duplicating.

Next capsule: 06 — Distance Metrics.

You just learned to feed data into the system efficiently. But there's a decision you made without thinking when creating the collections: the distance metric. Cosine, L2, or dot product — each one assumes a different geometry of the embedding space, and choosing wrong silently degrades retrieval. Capsule 06 gives you the geometric mental model to choose with judgment.

Resources

1. OpenAI — Rate Limits Guide — Limits per tier and handling strategies
2. OpenAI — Batching Embeddings — Official batching recommendations
3. ChromaDB — Adding Documents — Insertion and batch operations
4. tqdm — Progress Bars in Python — Progress bars for pipelines
5. Tenacity — Retry Library for Python — A robust alternative to manual try/except with retry
6. Idempotency Patterns in Distributed Systems — Stripe explains the general concept applicable to APIs

Estimated time: 30-40 minutes Next: 06-distance-metrics.md