Back to Blog
Feb 11, 20264 min readOnuzulike Anthony

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.

InfrastructureChromaDBVector SearchEmbeddingsRAG

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

bash
pip install chromadb
python
import 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:

python
# 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

python
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

python
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:

python
# 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:

python
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:

python
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.