Capsule 04: Metadata Filtering Implementation — from concept to code in ChromaDB

Capsule description

In Module 3 (capsule 02) you learned what metadata filtering is and why it's the most critical feature of a production RAG system: it reduces the search space 10x, improves accuracy and, in multi-tenant systems, is the line of defense against data leakage between clients.

This capsule closes the loop with real code. You'll learn ChromaDB's where clause syntax (which is based on MongoDB query operators), the six operators you need to know, how to combine filters with AND/OR/NOT, and how to benchmark the impact on your own data — because "filtering speeds things up 10x" is the general rule, but the real speedup depends on what percentage of your dataset remains after the filter, and you only discover that by measuring.

When you finish, you'll know how to write complex filters without consulting the documentation every time, anticipate the most common syntax errors (especially with numeric vs string types), and design the metadata schema from the start so the filters you'll need are easy to express.

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

• ✅ Write where clauses with the six main operators: equality, $ne, $gt/$gte/$lt/$lte, $in/$nin, $and/$or
• ✅ Combine multiple filters with correct boolean logic
• ✅ Differentiate when a filter goes in where (metadata) vs where_document (text)
• ✅ Benchmark the real impact of filters on your dataset (don't assume "10x speedup")
• ✅ Design a metadata schema considering the filters you'll need
• ✅ Anticipate the three most expensive syntax errors: mixed types, nonexistent keys, misunderstood implicit AND

Estimated time: 30-40 minutes

Inserting metadata correctly

Before filtering, you have to insert structured metadata. ChromaDB accepts a dict per document with values of type str, int, float, bool. It does not accept lists, nested dicts, or None — that's the first limitation to know.

import chromadb
from chromadb.utils import embedding_functions
import os

openai_ef = embedding_functions.OpenAIEmbeddingFunction(
    api_key=os.getenv("OPENAI_API_KEY"),
    model_name="text-embedding-3-small"
)

client = chromadb.PersistentClient(path="./chroma_filtering")
collection = client.get_or_create_collection(
    name="docs_with_metadata",
    embedding_function=openai_ef,
    metadata={"hnsw:space": "cosine"}
)

collection.add(
    documents=[
        "Password reset instructions for Pro plan users",
        "Q1 2026 marketing campaign report",
        "API documentation for v2.0 endpoints",
        "Product roadmap 2026 — confidential",
        "Customer support FAQ — billing section",
        "Engineering blog post about HNSW configuration",
    ],
    metadatas=[
        {"category": "support", "language": "en", "priority": 1, "plan": "pro"},
        {"category": "marketing", "language": "en", "priority": 3, "year": 2026},
        {"category": "docs", "language": "en", "priority": 2, "version": "2.0"},
        {"category": "product", "language": "en", "priority": 1, "confidential": True},
        {"category": "support", "language": "es", "priority": 2, "topic": "billing"},
        {"category": "engineering", "language": "en", "priority": 3, "topic": "hnsw"},
    ],
    ids=["1", "2", "3", "4", "5", "6"],
)

print(f"Total docs: {collection.count()}")  # 6

Important schema conventions:

1. Keep types consistent per key. If a key is int in one doc and str in another, range filters fail silently.
2. Naming in lowercase snake_case. priority, not Priority or priorityLevel.
3. Booleans for flags. confidential: True instead of confidential: "yes".
4. Short strings for categorical values. category: "support" (not long sentences).

The six operators you have to know

1. Equality (default)

# Only docs in category="support"
results = collection.query(
    query_texts=["how to reset password"],
    where={"category": "support"},
    n_results=5,
)

print(f"Results: {len(results['documents'][0])}")
# 2 docs: "Password reset..." and "Customer support FAQ..."

Without an explicit operator, ChromaDB assumes equality. It's the syntax you'll use in 70% of cases.

2. $ne (not equal)

# Everything except category="marketing"
results = collection.query(
    query_texts=["product information"],
    where={"category": {"$ne": "marketing"}},
    n_results=5,
)

Useful for excluding specific categories (e.g., "don't show me marketing docs on support queries").

3. $gt, $gte, $lt, $lte (numeric ranges)

# Docs with priority <= 2 (high priority)
results = collection.query(
    query_texts=["documentation"],
    where={"priority": {"$lte": 2}},
    n_results=5,
)
# Match: docs with priority=1 (Pro support, Product roadmap) and priority=2 (API docs, Spanish support)

# Docs with priority > 1 (non-critical)
results = collection.query(
    query_texts=["..."],
    where={"priority": {"$gt": 1}},
    n_results=5,
)

Important: range operators only work with numeric types (int, float). If the key is a string, ChromaDB throws an error or returns incorrect results without warning (depending on the version).

4. $in, $nin (membership)

# Docs in categories support OR documentation
results = collection.query(
    query_texts=["help with the API"],
    where={"category": {"$in": ["support", "docs"]}},
    n_results=5,
)
# Match: support docs + docs category

# Docs that are NOT support nor marketing
results = collection.query(
    query_texts=["..."],
    where={"category": {"$nin": ["support", "marketing"]}},
    n_results=5,
)

$in is preferable to multiple $or with equality. More readable and faster.

5. $and, $or (boolean logic)

Implicit AND when you pass multiple keys:

# Implicit AND: category="support" AND language="en"
results = collection.query(
    query_texts=["password help"],
    where={
        "category": "support",
        "language": "en",
    },
    n_results=5,
)

Explicit AND (needed for more complex combinations):

results = collection.query(
    query_texts=["..."],
    where={
        "$and": [
            {"category": "support"},
            {"priority": {"$lte": 2}},
            {"language": "en"},
        ]
    },
    n_results=5,
)

Explicit OR:

# (category="support") OR (priority=1)
results = collection.query(
    query_texts=["urgent issue"],
    where={
        "$or": [
            {"category": "support"},
            {"priority": 1},
        ]
    },
    n_results=5,
)

Complex combinations:

# (category in [support, docs]) AND (priority <= 2 OR confidential = true)
results = collection.query(
    query_texts=["..."],
    where={
        "$and": [
            {"category": {"$in": ["support", "docs"]}},
            {
                "$or": [
                    {"priority": {"$lte": 2}},
                    {"confidential": True},
                ]
            }
        ]
    },
    n_results=5,
)

6. where_document (filter by text content)

Besides filtering by metadata, you can filter by a substring of the document itself:

results = collection.query(
    query_texts=["..."],
    where={"category": "engineering"},
    where_document={"$contains": "HNSW"},
    n_results=5,
)
# Only engineering docs that mention "HNSW" in the text

where_document only supports $contains and $not_contains. It's useful but it isn't full-text search — it's simple substring matching. For hybrid search (BM25 + semantic), you need another approach (covered in M03/03 hybrid search and in guide #8 Advanced RAG).

Benchmarking the real impact

The general rule says "metadata filtering speeds things up 10x." That's true when the filter reduces the search space to 10%. If your filter only reduces it to 50%, the speedup is ~2x. If your filter leaves fewer candidates than n_results, it can be slower (we saw that trap in M04/06).

Let's measure over 10K synthetic docs.

# benchmark_filtering.py
import chromadb
from chromadb.utils import embedding_functions
import os
import time
import statistics

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

# If the collection is empty, generate 10K synthetic docs
if collection.count() == 0:
    docs = [f"Document {i} discusses topic {i % 100} in category {['support','marketing','docs','product'][i % 4]}." for i in range(10000)]
    metas = [
        {
            "category": ["support", "marketing", "docs", "product"][i % 4],
            "priority": (i % 3) + 1,
            "topic_id": i % 100,
        }
        for i in range(10000)
    ]
    ids = [f"doc_{i:05d}" for i in range(10000)]
    # Insert in batches (capsule 05)
    for batch_start in range(0, 10000, 200):
        end = min(batch_start + 200, 10000)
        collection.add(
            documents=docs[batch_start:end],
            metadatas=metas[batch_start:end],
            ids=ids[batch_start:end],
        )

# Pre-compute the query embedding (don't count OpenAI latency)
import openai
client_oai = openai.OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
query_emb = client_oai.embeddings.create(
    input="general topic discussion",
    model="text-embedding-3-small"
).data[0].embedding


def benchmark_filter(label, where_clause, n_runs=50):
    """Measure latency with a specific filter."""
    # Warm-up
    for _ in range(5):
        collection.query(query_embeddings=[query_emb], n_results=10, where=where_clause)

    latencies = []
    for _ in range(n_runs):
        start = time.perf_counter()
        collection.query(query_embeddings=[query_emb], n_results=10, where=where_clause)
        latencies.append((time.perf_counter() - start) * 1000)

    latencies.sort()
    p50 = latencies[len(latencies) // 2]
    p95 = latencies[int(len(latencies) * 0.95)]
    return p50, p95


# Filters with different selectivity levels
benchmarks = [
    ("No filter (10K candidates)", None),
    ("category=support (~25% = 2500 candidates)", {"category": "support"}),
    ("category=support AND priority<=2 (~17% = 1700)", {"category": "support", "priority": {"$lte": 2}}),
    ("category=support AND priority=1 (~8% = 800)", {"category": "support", "priority": 1}),
    ("topic_id=42 (~1% = 100)", {"topic_id": 42}),  # Ultra-specific filter
]

print(f"{'Filter':<55} {'p50':>8} {'p95':>8}  Speedup vs no-filter")
baseline_p95 = None
for label, where in benchmarks:
    p50, p95 = benchmark_filter(label, where)
    if baseline_p95 is None:
        baseline_p95 = p95
        speedup = "1.0x (baseline)"
    else:
        speedup = f"{baseline_p95 / p95:.1f}x"
    print(f"{label:<55} {p50:>6.1f}ms {p95:>6.1f}ms  {speedup}")

Typical output:

Filter                                                   p50      p95  Speedup vs no-filter
No filter (10K candidates)                              8.2ms  12.4ms  1.0x (baseline)
category=support (~25% = 2500 candidates)               4.1ms   6.3ms  2.0x
category=support AND priority<=2 (~17% = 1700)          3.2ms   5.1ms  2.4x
category=support AND priority=1 (~8% = 800)             2.4ms   4.0ms  3.1x
topic_id=42 (~1% = 100)                                 5.8ms   9.2ms  1.3x  ← watch out

Lessons:

1. The "10x" is not universal. In this dataset the maximum observed speedup is ~3x because HNSW is already very fast on 10K docs. The 10x shows up more in datasets of 100K+ where reducing the search space really matters.

2. The ultra-specific filter is counterintuitively slower. Reducing the space from 10K to 100 candidates drops the speedup to 1.3x — because HNSW has to scan more nodes looking for candidates that satisfy the filter. When the filter leaves fewer candidates than n_results × 10, consider get() with a direct filter instead of query().

3. The benchmark over your real dataset is the only thing that matters. The numbers above are illustrative; yours will be different depending on the metadata distribution, dataset size, and ef_search.

Designing the metadata schema from the start

The most expensive mistake with metadata filtering isn't syntax — it's not having added the right fields at ingestion time. If you discover in production that you want to filter by language, but the chunks don't have that metadata, you have to re-ingest everything.

Before starting the massive ingestion, list the filters you'll need:

# Example: metadata schema for a corporate RAG
WE_EXPECT_TO_FILTER_BY = [
    "category",       # e.g.: "support", "engineering", "legal"
    "language",       # e.g.: "en", "es", "pt"
    "department",     # e.g.: "sales", "engineering", "hr"
    "access_level",   # e.g.: "public", "internal", "confidential"
    "doc_date",       # e.g.: 1714521600 (timestamp for range)
    "doc_id",         # for direct retrieval of chunks of a specific doc
    "chunk_index",    # to order chunks of the same doc
]

# This translates to metadata per chunk:
def build_metadata(doc, chunk_index, total_chunks):
    return {
        "doc_id": doc.id,
        "chunk_index": chunk_index,
        "total_chunks": total_chunks,
        "category": doc.category,
        "language": doc.language,
        "department": doc.department,
        "access_level": doc.access_level,
        "doc_date": int(doc.created_at.timestamp()),  # int for range filters
        "source": doc.source_path,
    }

Heuristics:

• Over-including is cheap, omitting is expensive. Adding 2-3 extra fields costs milliseconds in ingestion. Missing 1 critical field costs hours of re-ingestion.
• Timestamps as int (Unix epoch), not str. Enables date-range filters.
• For multi-tenant data: tenant_id always. It's the line of defense against leaks (capsule 04 of Module 3 covered why).
• Stable identifiers. doc_id must survive re-ingests. Don't use random UUIDs.

Traps and common mistakes

Trap 1: inconsistent types in a key

The mistake:

# Doc 1
{"priority": 1}  # int

# Doc 2
{"priority": "1"}  # string

Symptom: range filters ({"priority": {"$lte": 2}}) return only some docs with no explicit error. Other docs "disappear" from the result even though their priority numerically satisfies the condition.

Why it happens: ChromaDB compares the operator with the actual stored value. 1 <= 2 is True, but "1" <= 2 isn't evaluated correctly in SQLite.

How to prevent it: validate types in the ingestion pipeline:

from typing import Any

EXPECTED_TYPES = {
    "category": str,
    "language": str,
    "priority": int,
    "doc_date": int,
    "confidential": bool,
}

def validate_metadata(meta: dict[str, Any]) -> dict[str, Any]:
    cleaned = {}
    for key, value in meta.items():
        if key in EXPECTED_TYPES:
            expected = EXPECTED_TYPES[key]
            if not isinstance(value, expected):
                raise TypeError(f"Metadata key '{key}' expected {expected.__name__}, got {type(value).__name__}: {value}")
        cleaned[key] = value
    return cleaned

# Before collection.add():
metadatas = [validate_metadata(m) for m in raw_metadatas]

Trap 2: filtering by a key that doesn't exist

The mistake:

# Some docs have "language", others don't
collection.query(
    query_texts=["..."],
    where={"language": "en"},
    n_results=10,
)

Symptom: docs without the language key are excluded from the result, even though they should be included per the product logic.

How to prevent it: ensure that all filtered keys exist in all docs. If a field is optional, use a default value ("unknown", a null sentinel) — not the absence of the key.

Trap 3: $and confused with implicit AND

The mistake:

# This is NOT what you think
where = {
    "$and": [
        {"category": "support", "priority": 1},  # ← implicit AND inside the $and
        {"language": "en"}
    ]
}

Symptom: the filter works but covers fewer cases than expected.

How to write it: inside $and, each element is an independent filter. The above is equivalent to:

# (category=support AND priority=1) AND (language=en)
where = {
    "$and": [
        {"category": "support"},
        {"priority": 1},
        {"language": "en"},
    ]
}

More explicit and less prone to misunderstandings.

Trap 4: confusing where with where_document

The mistake:

# You want to filter docs whose TEXT contains "HNSW"
collection.query(
    query_texts=["..."],
    where={"text": {"$contains": "HNSW"}},  # ❌
)

Symptom: error or empty results. The $contains operator does not apply to metadata.

How to fix it:

collection.query(
    query_texts=["..."],
    where_document={"$contains": "HNSW"},  # ✅
)

where is for metadata. where_document is for the document text. They're different parameters.

Trap 5: an ultra-specific filter that makes the query slow

The mistake:

# Filter that leaves only 5 candidates
collection.query(
    query_embeddings=[emb],
    n_results=10,
    where={"unique_session_id": "abc123def456"},
)

Symptom: the query is slower than without a filter.

How to fix it: if the filter leaves fewer candidates than n_results × 5, use a direct get():

# Faster: direct get when the filter is very restrictive
results = collection.get(
    where={"unique_session_id": "abc123def456"},
    include=['documents', 'metadatas']
)
# It's not semantic search, but the filter already selects the relevant docs

Trap 6: metadata schema without thinking about future filters

The mistake: you insert 1 million docs with metadata {"source": path, "chunk_index": i} without adding category, language, or other fields. In week 4, marketing asks "let's filter by language to personalize responses."

Symptom: "we have to re-ingest 1M docs because we don't have language."

How to prevent it: before the massive ingestion, run a workshop with stakeholders ("what filters will you need over the next 12 months?") and add ALL the fields to the schema. The extra cost is trivial. The cost of re-ingesting is significant (time, re-embedding money, downtime).

Applied exercise

Scenario: you're building a RAG for a multi-tenant SaaS company where each client has their own documents. Requirements:

• Each chunk belongs to a specific tenant (unique ID)
• Some chunks are private (only accessible to that tenant)
• Other chunks are public (visible to all tenants — e.g.: general support docs)
• Queries must respect permissions: tenant A never sees tenant B's private chunks
• Chunks have a language (en, es, pt)
• There's a type classification (product_doc, support_faq, legal, marketing)
• Filtering by date is frequent (show only docs from the last year)

Task:

1. Design the metadata schema for the chunks
2. Write the filter you would apply to a query from tenant acme_corp looking for Spanish docs from the last year
3. Identify a security risk if the filter is built wrong

def build_chunk_metadata(chunk, doc, tenant_id):
    return {
        # Identity
        "tenant_id": tenant_id,             # str — always present
        "doc_id": doc.id,                   # str — stable
        "chunk_index": chunk.index,         # int
        "total_chunks": len(doc.chunks),    # int

        # Permissions
        "visibility": doc.visibility,       # str: "private" | "public"

        # Classification
        "doc_type": doc.type,               # str: product_doc, support_faq, legal, marketing
        "language": doc.language,           # str: en, es, pt
        "category": doc.category,           # str: optional fine granularity

        # Temporal (Unix epoch timestamps for range filters)
        "created_at": int(doc.created_at.timestamp()),
        "updated_at": int(doc.updated_at.timestamp()),

        # Traceability
        "source": doc.source_path,          # str — for citations
    }

import time

now = int(time.time())
one_year_ago = now - (365 * 24 * 60 * 60)

# The security filter: tenant_id="acme_corp" OR (visibility="public" AND another tenant)
filter_query = {
    "$and": [
        {
            "$or": [
                {"tenant_id": "acme_corp"},
                {"visibility": "public"},
            ]
        },
        {"language": "es"},
        {"created_at": {"$gte": one_year_ago}},
    ]
}

results = collection.query(
    query_texts=["how do I set up billing?"],
    where=filter_query,
    n_results=10,
)

# ❌ BROKEN FILTER: leak between tenants
filter_query_BROKEN = {
    "$or": [                              # ← OR at the root level
        {"tenant_id": "acme_corp"},
        {"visibility": "public"},
        {"language": "es"},                # ← this OR also applies to other tenants
    ]
}
# This returns:
# - acme_corp docs (correct)
# - public docs from any tenant (correct)
# - Spanish docs from ANY tenant (LEAK!)

def build_secure_filter(tenant_id: str, additional_filters: dict) -> dict:
    """
    Builds a filter that ALWAYS respects per-tenant isolation.
    additional_filters is AND-ed on top of the security clause,
    it can never bypass the tenant.
    """
    security_clause = {
        "$or": [
            {"tenant_id": tenant_id},
            {"visibility": "public"},
        ]
    }
    return {
        "$and": [security_clause, additional_filters],
    }

# Safe use
filter_query = build_secure_filter(
    tenant_id="acme_corp",
    additional_filters={
        "$and": [
            {"language": "es"},
            {"created_at": {"$gte": one_year_ago}},
        ]
    },
)

def test_tenant_isolation():
    # Insert private docs from tenant_a and tenant_b
    collection.add(
        documents=["secret_a", "secret_b", "public_doc"],
        metadatas=[
            {"tenant_id": "tenant_a", "visibility": "private"},
            {"tenant_id": "tenant_b", "visibility": "private"},
            {"tenant_id": "tenant_a", "visibility": "public"},
        ],
        ids=["1", "2", "3"],
    )

    # A query as tenant_a must NOT find "secret_b"
    filter = build_secure_filter("tenant_a", {})
    results = collection.query(
        query_texts=["secret"],
        where=filter,
        n_results=10,
    )

    assert "2" not in results['ids'][0], "LEAK: tenant_a saw tenant_b's private doc"

Summary and next step

What you learned:

• ChromaDB accepts metadata with types str/int/float/bool (not lists, dicts, or None).
• Six operators cover 99% of cases: equality (default), $ne, $gt/$gte/$lt/$lte, $in/$nin, $and, $or.
• AND is implicit when you pass multiple keys; explicit with $and when you combine with $or.
• where filters by metadata. where_document filters by a substring of the text.
• The real speedup depends on the % of the dataset that remains after the filter — benchmark your case, don't assume "10x".
• An ultra-specific filter (leaves <10 candidates) can be slower than no filter — use a direct get() in those cases.
• Design the metadata schema from the start considering future filters — adding fields later means re-ingesting.
• In multi-tenant systems, encapsulate the security clause (tenant_id + visibility) in a reusable function and test it with isolation tests.

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

• Write a complex filter combining $and, $or, $in, and range operators.
• Identify the classic security bug in multi-tenant filters (root-level $or).
• Design a metadata schema for a RAG considering future filters (not just current ones).

Next capsule: 05 — Batch Ingestion Pipeline.

You just mastered how to filter when running queries. But before you can filter, you have to insert the data — and doing it efficiently when it's thousands or millions of docs requires knowing batch ingestion. Capsule 05 covers how to amortize the overhead of inserts, handle rate limits of the embeddings API, and build an idempotent pipeline that can be re-run without duplicating.

Resources

1. ChromaDB — Metadata Filtering — Official operator reference
2. ChromaDB — Where Document Filtering — Filter by text content
3. MongoDB Query Operators — ChromaDB inherits the syntax from MongoDB; useful for advanced cases
4. Multi-tenancy Patterns in Vector DBs — Comparison of strategies (collection per tenant vs metadata filtering vs namespaces)
5. Schema Design Best Practices — General patterns applicable to a metadata schema
6. Indexed Property Filters in HNSW (paper) — How filters affect HNSW performance

Estimated time: 30-40 minutes Next: 05-batch-ingestion.md