What are common ChromaDB errors and how do you handle them in production code?

ChromaDB raises specific exception types that should be caught and handled gracefully in production applications. Understanding the error hierarchy helps you write resilient ingestion pipelines and retrieval code.

import chromadb
from chromadb.errors import (
    InvalidCollectionException,
    IDAlreadyExistsError,
    InvalidDimensionException,
)

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

# --- Error 1: Collection not found ---
try:
    col = client.get_collection("does_not_exist")
except InvalidCollectionException as e:
    print(f"Collection missing: {e}")
    col = client.create_collection("does_not_exist")  # create it

# --- Error 2: Duplicate ID ---
col.add(documents=["Original doc"], ids=["doc-1"])
try:
    col.add(documents=["Duplicate doc"], ids=["doc-1"])
except IDAlreadyExistsError:
    print("ID already exists — use upsert() instead")
    col.upsert(documents=["Updated doc"], ids=["doc-1"])  # safe

# --- Error 3: Dimension mismatch ---
# Occurs when pre-computed embeddings don't match collection's embedding dimensions
col2 = client.create_collection("fixed_dim")
col2.add(embeddings=[[0.1, 0.2, 0.3]], documents=["Doc"], ids=["x"])
try:
    col2.add(embeddings=[[0.1, 0.2]], documents=["Wrong dim"], ids=["y"])  # 2-dim
except InvalidDimensionException as e:
    print(f"Dimension mismatch: {e}")

# --- Error 4: Connection error (HttpClient) ---
try:
    remote = chromadb.HttpClient(host="bad-host", port=9999)
    remote.heartbeat()
except Exception as e:
    print(f"Server unreachable: {e}")

# --- Production pattern: retry wrapper ---
import time
from functools import wraps

def with_retry(max_attempts=3, delay=1.0):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return fn(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    print(f"Attempt {attempt+1} failed: {e}. Retrying...")
                    time.sleep(delay * (attempt + 1))
        return wrapper
    return decorator

@with_retry(max_attempts=3)
def safe_add(collection, documents, ids):
    collection.upsert(documents=documents, ids=ids)