ChromaDB for Vector Search in Local Apps
Setting up ChromaDB, embedding documents, querying with metadata filters, and integrating with a hybrid BM25 + vector retrieval pipeline.
ChromaDB is an open-source embedding database designed for local-first applications. No cloud account, no billing, no rate limits — just a Python library that stores vectors and metadata on disk. For personal knowledge bases and local RAG applications, it's the simplest starting point.
Installation and Setup
pip install chromadbimport chromadb
from chromadb.config import Settings
# Persistent storage (survives restarts)
client = chromadb.PersistentClient(
path="./chroma_db",
settings=Settings(anonymized_telemetry=False)
)
# In-memory (testing only)
client = chromadb.EphemeralClient()Collections
A collection is a namespace — group documents by type or project:
# Get or create (idempotent)
collection = client.get_or_create_collection(
name="knowledge_base",
metadata={"hnsw:space": "cosine"} # cosine similarity (default is L2)
)Prefer cosine similarity for text embeddings. L2 is for geometric distances; cosine normalizes for document length.
Adding Documents
def add_documents(collection, chunks: list[dict]):
collection.upsert(
ids=[chunk["id"] for chunk in chunks],
embeddings=[chunk["embedding"] for chunk in chunks],
documents=[chunk["text"] for chunk in chunks],
metadatas=[{
"source_path": chunk["source_path"],
"chunk_index": chunk["index"],
"file_type": chunk["file_type"],
"ingested_at": chunk["ingested_at"],
} for chunk in chunks]
)upsert is idempotent — re-ingesting a file updates existing vectors rather than creating duplicates. This is critical for background re-indexing jobs.
Basic Query
def vector_search(collection, query_embedding: list[float], n: int = 10) -> list[dict]:
results = collection.query(
query_embeddings=[query_embedding],
n_results=n,
include=["documents", "metadatas", "distances"]
)
return [
{
"text": results["documents"][0][i],
"metadata": results["metadatas"][0][i],
"score": 1 - results["distances"][0][i], # cosine: distance → similarity
}
for i in range(len(results["ids"][0]))
]Metadata Filtering
Filter results before vector search using where clauses:
# Only search Python files
results = collection.query(
query_embeddings=[embedding],
n_results=10,
where={"file_type": {"$eq": "py"}},
)
# Multiple conditions
results = collection.query(
query_embeddings=[embedding],
n_results=10,
where={
"$and": [
{"file_type": {"$in": ["py", "ts"]}},
{"ingested_at": {"$gte": "2025-01-01"}},
]
},
)Metadata filters run before vector similarity computation, so they don't slow down the search — they reduce the candidate pool.
Hybrid Retrieval
ChromaDB handles the vector half of hybrid retrieval. The BM25 (keyword) half runs separately, then both results are merged with Reciprocal Rank Fusion:
import sqlite3
from rank_bm25 import BM25Okapi
class HybridRetriever:
def __init__(self, chroma_collection, sqlite_conn):
self.collection = chroma_collection
self.conn = sqlite_conn
def search(self, query: str, query_embedding: list[float], k: int = 10) -> list[dict]:
# Vector search
vector_results = vector_search(self.collection, query_embedding, n=k * 2)
# BM25 keyword search from SQLite
bm25_results = self._bm25_search(query, k=k * 2)
# Reciprocal Rank Fusion
return self._rrf_merge(vector_results, bm25_results, k=k)
def _bm25_search(self, query: str, k: int) -> list[dict]:
rows = self.conn.execute(
"SELECT id, text, source_path FROM chunks"
).fetchall()
corpus = [row[1].split() for row in rows]
bm25 = BM25Okapi(corpus)
scores = bm25.get_scores(query.split())
top_k = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:k]
return [{"text": rows[i][1], "metadata": {"source_path": rows[i][2]}, "score": scores[i]}
for i in top_k]
def _rrf_merge(self, results_a: list, results_b: list, k: int = 10) -> list[dict]:
scores = {}
for rank, item in enumerate(results_a):
key = item["text"][:100]
scores[key] = scores.get(key, {"item": item, "score": 0})
scores[key]["score"] += 1.0 / (60 + rank + 1)
for rank, item in enumerate(results_b):
key = item["text"][:100]
scores[key] = scores.get(key, {"item": item, "score": 0})
scores[key]["score"] += 1.0 / (60 + rank + 1)
ranked = sorted(scores.values(), key=lambda x: x["score"], reverse=True)
return [item["item"] for item in ranked[:k]]Managing Stale Documents
When a source file changes, delete its old chunks before re-ingesting:
def remove_file_chunks(collection, source_path: str):
results = collection.get(
where={"source_path": {"$eq": source_path}},
include=[]
)
if results["ids"]:
collection.delete(ids=results["ids"])When to Move Beyond ChromaDB
ChromaDB works well up to ~1M vectors on a modern machine. Beyond that, or when you need:
- Horizontal scaling
- Fine-grained access control
- Production SLAs
Consider Qdrant (self-hosted) or Weaviate. For cloud-hosted, Pinecone or LanceDB on S3. For most personal projects and team tools, ChromaDB on a single machine handles the load without the operational complexity.