Indexing at Scale

Embedding a handful of documents in a Jupyter notebook is trivial. Embedding 10,000 documents in production without blocking your API, blowing your rate limits, or silently dropping half the corpus is a pipeline problem. This lesson covers the patterns that turn document ingestion from a script into a reliable background process: batch embedding, rate limiting, incremental indexing, and worker isolation.

What you'll learn

• Batch embedding sends 50-100 documents per API call instead of one at a time: 20x faster and 20x cheaper in API overhead
• Background workers handle indexing in a separate process so your API never blocks on ingestion
• Incremental indexing only processes new or changed documents, not the entire corpus every time

The problem

Your knowledge base has 10,000 documents. You write a script that loops through each one, calls the embedding API, and inserts the vector into pgvector. It works on your laptop with 100 documents. In production with 10,000, three things go wrong. First, 10,000 individual API calls take 5-10 minutes and cost significantly more than batched calls because each call has overhead. Second, your API that serves user queries is running in the same process and blocks while indexing runs. Third, you hit the embedding API's rate limit after 500 calls and the remaining 9,500 documents silently fail to index.

Build it

Step 1: Batch embedding with rate limit handling

Processing documents individually is too slow for production volumes. Implement batching logic that aggregates requests and gracefully handles API rate limits via exponential backoff.

import time
from openai import OpenAI

client = OpenAI()

def batch_embed(texts, batch_size=50, model="text-embedding-3-small"):
    """Embed texts in batches with exponential backoff on rate limits."""
    all_embeddings = []

    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        max_retries = 3

        for attempt in range(max_retries):
            try:
                response = client.embeddings.create(model=model, input=batch)
                all_embeddings.extend([e.embedding for e in response.data])
                break
            except RateLimitError:
                if attempt == max_retries - 1:
                    raise
                wait = 2 ** attempt  # 1s, 2s, 4s
                time.sleep(wait)

        # Small delay between batches to stay under sustained rate limits
        if i + batch_size < len(texts):
            time.sleep(0.2)

    return all_embeddings

Step 2: Background worker with a queue

To prevent ingestion from blocking user-facing requests, offload the work. Use a background thread and a concurrent queue to process document batches asynchronously.

import threading
import queue

indexing_queue = queue.Queue()

def indexing_worker():
    """Background worker that processes the indexing queue."""
    while True:
        batch = indexing_queue.get()
        if batch is None:  # Shutdown signal
            break
        try:
            embeddings = batch_embed([doc["content"] for doc in batch])
            for doc, emb in zip(batch, embeddings):
                cursor.execute(
                    """INSERT INTO documents (id, content, embedding, metadata)
                    VALUES (%s, %s, %s, %s)
                    ON CONFLICT (id) DO NOTHING""",
                    (doc["id"], doc["content"], emb, json.dumps(doc.get("metadata", {})))
                )
        except Exception as e:
            logger.error("indexing_batch_failed", error=str(e), batch_size=len(batch))
        finally:
            indexing_queue.task_done()

# Start the worker
worker = threading.Thread(target=indexing_worker, daemon=True)
worker.start()

def enqueue_documents(documents):
    """Non-blocking document ingestion. Returns immediately."""
    indexing_queue.put(documents)

Step 3: Incremental indexing

Re-indexing the entire corpus repeatedly wastes time and money. Query the database first to identify missing records, and only enqueue new documents for embedding.

def get_existing_ids():
    cursor.execute("SELECT id FROM documents")
    return {row[0] for row in cursor.fetchall()}

def incremental_index(documents):
    """Only index documents not already in the store."""
    existing = get_existing_ids()
    new_docs = [d for d in documents if d["id"] not in existing]

    if not new_docs:
        return 0

    enqueue_documents(new_docs)
    return len(new_docs)

What goes wrong

Mistake	How you notice it	The fix
Rate limit errors on large batches	429 errors, partial index	Add exponential backoff. Reduce batch size if errors persist. Check your tier's rate limits
Indexing blocks API responses	API latency spikes during re-index	Always run indexing in a background worker. The API process should never wait on embedding calls
Duplicate embeddings from retries	Same document indexed twice after a partial failure	Use ON CONFLICT (id) DO NOTHING. Track which batches completed vs failed
Worker crashes silently	Documents never appear in search results	Monitor the queue size. Alert if the queue grows without shrinking

Confirm it worked

Simulate a bulk ingestion scenario to verify the pipeline's resilience. The background worker should incrementally process the queue without blocking the main execution thread.

# Index 1000 documents
docs = [{"id": f"doc-{i}", "content": f"Document {i} content"} for i in range(1000)]
indexed = incremental_index(docs)

# Verify count
time.sleep(5)  # Allow background worker to process
cursor.execute("SELECT COUNT(*) FROM documents WHERE id LIKE 'doc-%'")
count = cursor.fetchone()[0]
assert count == indexed

Next: Handling Updates