Module 6: Metadata Filtering — the component almost nobody implements first but everyone ends up needing

Capsule 08: Capstone project — an end-to-end Metadata-Filtered RAG

Next →

Project description

This is the close of module 6 — and the culmination of the "state of the art" production RAG architecture. You're going to build a system that integrates it all: metadata filtering + hybrid search + re-ranking + isolation tests + auditing. It's the project that goes into your portfolio or your real codebase.

The system implements the complete pipeline (filter → hybrid → rerank), automated security tests (zero data leak between tenants), a benchmark against the baseline to justify the investment, and documentation a Tech Lead can review to approve a production deploy.

By the end of this project you'll have:

  • ✅ A complete RAG pipeline with metadata filtering integrated
  • ✅ A metadata schema validated with Pydantic
  • ✅ A mandatory secure_query wrapper with TenantContext
  • ✅ Automated isolation tests that run in CI
  • ✅ Your own eval set + a comparison benchmark
  • ✅ Audit logging for compliance
  • ✅ A README that justifies the architecture for deploy approval

Estimated time: 4-6 hours + 1 hour of analysis.

Project architecture

metadata_rag_project/
├── src/
│   ├── schema/
│   │   ├── __init__.py
│   │   ├── document_metadata.py     # Pydantic schemas
│   │   └── tenant_context.py
│   ├── retrievers/
│   │   ├── base.py
│   │   ├── semantic.py
│   │   ├── bm25.py
│   │   └── hybrid_filtered.py        # The complete pipeline
│   ├── security/
│   │   ├── secure_query.py           # The mandatory wrapper
│   │   └── audit_log.py
│   ├── ingestion/
│   │   └── ingest_pipeline.py        # Validation + indexing
│   ├── evaluation/
│   │   ├── eval_set.py
│   │   ├── metrics.py
│   │   └── ab_test.py
│   └── pipeline.py                   # The top-level pipeline
├── tests/
│   ├── test_isolation.py             # Cross-tenant
│   ├── test_security.py              # Permission escalation
│   └── test_pipeline.py
├── data/
│   ├── corpus/
│   └── golden_set.json
├── benchmarks/
│   ├── run_benchmark.py
│   └── reports/
├── scripts/
│   └── pre_commit_check.py           # The anti-bypass linter
├── .env
├── requirements.txt
└── README.md

Step 1: the metadata schema with Pydantic

# src/schema/document_metadata.py
from pydantic import BaseModel, Field
from typing import Literal


class DocumentMetadata(BaseModel):
    """The mandatory schema for every document."""

    # Universals
    workspace_id: str = Field(..., description="Multi-tenant isolation.")
    doc_id: str
    chunk_index: int = Field(..., ge=0)
    source: str
    created_at: int  # Unix timestamp

    # Categorization
    type: Literal["tutorial", "reference", "faq", "changelog", "alert"]
    visibility: Literal["public", "team", "private", "admin_only"] = "public"

    # Domain-specific metadata (adjust it to your case)
    language: Literal["en", "es", "pt"] = "en"
    tags: list[str] = Field(default_factory=list)

    class Config:
        extra = "forbid"

Step 2: TenantContext and secure_query

# src/schema/tenant_context.py
from dataclasses import dataclass


class TenantIsolationError(Exception):
    pass


@dataclass(frozen=True)
class TenantContext:
    workspace_id: str
    user_id: str
    user_role: str = "member"
    project_ids: list[str] = None

    def __post_init__(self):
        if not self.workspace_id or not self.user_id:
            raise TenantIsolationError("workspace_id and user_id are mandatory")


# src/security/secure_query.py
def secure_query(
    collection,
    query_text: str,
    tenant: TenantContext,
    additional_filters: dict = None,
    n_results: int = 5,
):
    if not tenant or not tenant.workspace_id:
        raise TenantIsolationError("A TenantContext is mandatory")

    # Validate that additional_filters doesn't overwrite workspace_id
    if additional_filters and "workspace_id" in additional_filters:
        raise TenantIsolationError("Do not overwrite workspace_id")

    where = {"workspace_id": tenant.workspace_id}
    if additional_filters:
        where = {"$and": [where, additional_filters]}

    audit_log_query(tenant, query_text, n_results)

    return collection.query(
        query_texts=[query_text],
        where=where,
        n_results=n_results,
    )

Step 3: the hybrid filtered retriever

# src/retrievers/hybrid_filtered.py
from concurrent.futures import ThreadPoolExecutor
from sentence_transformers import CrossEncoder
from rank_bm25 import BM25Okapi


class HybridFilteredRetriever:
    """The complete pipeline: filter → hybrid → rerank."""

    def __init__(
        self,
        collection,
        bm25_index: BM25Okapi,
        all_doc_ids: list[str],
        all_metadatas: list[dict],
    ):
        self.collection = collection
        self.bm25_index = bm25_index
        self.all_doc_ids = all_doc_ids
        self.all_metadatas = all_metadatas
        self.reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-12-v2")

    def search(
        self,
        query: str,
        tenant: TenantContext,
        additional_filters: dict = None,
        n_candidates: int = 30,
        n_final: int = 5,
    ) -> list[dict]:

        # 1. Build the secure filter
        where = self._build_secure_filter(tenant, additional_filters)

        # 2. Hybrid retrieval in parallel
        with ThreadPoolExecutor(max_workers=2) as executor:
            sem_future = executor.submit(self._semantic_search, query, where, n_candidates)
            bm25_future = executor.submit(self._bm25_search, query, where, n_candidates)
            sem_ids = sem_future.result()
            bm25_ids = bm25_future.result()

        # 3. RRF fusion
        fused_ids = self._reciprocal_rank_fusion([sem_ids, bm25_ids])

        # 4. Retrieve the content + rerank
        candidates = self.collection.get(
            ids=fused_ids[:n_candidates],
            include=["documents", "metadatas"],
        )

        # 5. Cross-encoder rerank
        return self._rerank(query, candidates, top_k=n_final)

    def _build_secure_filter(self, tenant, additional):
        if not tenant or not tenant.workspace_id:
            raise TenantIsolationError("workspace_id is mandatory")

        base = {"workspace_id": tenant.workspace_id}
        if additional and "workspace_id" in additional:
            raise TenantIsolationError("Do not overwrite workspace_id")

        return {"$and": [base, additional]} if additional else base

    def _semantic_search(self, query, where, n):
        results = self.collection.query(
            query_texts=[query], where=where, n_results=n
        )
        return results["ids"][0]

    def _bm25_search(self, query, where, n):
        # Get the IDs allowed by the filter
        allowed = set(self.collection.get(where=where, include=[])["ids"])

        query_tokens = query.lower().split()
        scores = self.bm25_index.get_scores(query_tokens)

        scored = [
            (self.all_doc_ids[i], scores[i])
            for i in range(len(scores))
            if self.all_doc_ids[i] in allowed and scores[i] > 0
        ]
        scored.sort(key=lambda x: -x[1])
        return [doc_id for doc_id, _ in scored[:n]]

    def _reciprocal_rank_fusion(self, rankings, k=60):
        from collections import defaultdict
        scores = defaultdict(float)
        for ranking in rankings:
            for rank, doc_id in enumerate(ranking, 1):
                scores[doc_id] += 1.0 / (k + rank)
        return [doc_id for doc_id, _ in sorted(scores.items(), key=lambda x: -x[1])]

    def _rerank(self, query, candidates, top_k):
        if not candidates["documents"]:
            return []

        pairs = [(query, doc) for doc in candidates["documents"]]
        scores = self.reranker.predict(pairs, batch_size=32, show_progress_bar=False)

        ranked = sorted(
            range(len(scores)),
            key=lambda i: -scores[i],
        )[:top_k]

        return [
            {
                "doc_id": candidates["ids"][i],
                "document": candidates["documents"][i],
                "metadata": candidates["metadatas"][i],
                "score": float(scores[i]),
            }
            for i in ranked
        ]

Step 4: the isolation tests

# tests/test_isolation.py
import pytest
from src.schema.tenant_context import TenantContext, TenantIsolationError
from src.retrievers.hybrid_filtered import HybridFilteredRetriever


@pytest.fixture
def retriever_with_two_tenants():
    """Setup with docs from tenant_a and tenant_b."""
    # ... set up ChromaDB + BM25 with 100 docs per tenant ...
    return retriever


def test_tenant_a_cannot_see_tenant_b_docs(retriever_with_two_tenants):
    ctx = TenantContext(workspace_id="tenant_a", user_id="user_1")
    results = retriever_with_two_tenants.search("test", ctx, n_final=20)

    for r in results:
        assert r["metadata"]["workspace_id"] == "tenant_a", (
            f"LEAK: tenant_a saw {r['metadata']['workspace_id']}"
        )


def test_workspace_id_is_required():
    with pytest.raises(TenantIsolationError):
        TenantContext(workspace_id="", user_id="user_1")


def test_additional_filters_cannot_override_tenant():
    ctx = TenantContext(workspace_id="tenant_a", user_id="user_1")

    with pytest.raises(TenantIsolationError):
        retriever.search(
            "test",
            ctx,
            additional_filters={"workspace_id": "tenant_b"},  # a bypass attempt
        )


def test_concurrent_tenants_no_leakage(retriever_with_two_tenants):
    """Concurrent queries from different tenants don't cross over."""
    import threading

    leaks = []

    def query_as(tenant_name):
        ctx = TenantContext(workspace_id=tenant_name, user_id=f"u_{tenant_name}")
        results = retriever_with_two_tenants.search("test", ctx, n_final=20)
        for r in results:
            if r["metadata"]["workspace_id"] != tenant_name:
                leaks.append((tenant_name, r["metadata"]["workspace_id"]))

    threads = [
        threading.Thread(target=query_as, args=("tenant_a",))
        for _ in range(10)
    ] + [
        threading.Thread(target=query_as, args=("tenant_b",))
        for _ in range(10)
    ]

    for t in threads:
        t.start()
    for t in threads:
        t.join()

    assert not leaks, f"Concurrent leakage: {leaks}"

Step 5: the eval set + benchmark

# data/golden_set.json
[
  {
    "query": "OAuth2PasswordBearer scopes",
    "tenant_id": "tenant_a",
    "expected_doc_ids": ["a_doc_42"],
    "category": "exact_match"
  },
  {
    "query": "how do I authenticate users",
    "tenant_id": "tenant_a",
    "expected_doc_ids": ["a_doc_103"],
    "category": "conceptual"
  }
]


# benchmarks/run_benchmark.py
def benchmark_pipeline_variants(eval_set):
    """Compares 4 architectures: semantic, +hybrid, +rerank, +filter."""

    metrics = {}

    for arch_name, retriever_fn in [
        ("A_semantic_baseline", baseline_semantic_only),
        ("B_hybrid", hybrid_no_filter),
        ("C_hybrid_rerank", hybrid_with_rerank_no_filter),
        ("D_full_pipeline", pipeline_with_everything),
    ]:
        results = []
        for item in eval_set:
            tenant = TenantContext(workspace_id=item["tenant_id"], user_id="bench")
            res = retriever_fn(item["query"], tenant)
            retrieved_ids = [r["doc_id"] for r in res]

            relevant = set(item["expected_doc_ids"])
            in_top_5 = retrieved_ids[:5]
            precision = sum(1 for d in in_top_5 if d in relevant) / 5
            recall = sum(1 for d in in_top_5 if d in relevant) / max(len(relevant), 1)
            results.append({"precision": precision, "recall": recall})

        metrics[arch_name] = {
            "precision_at_5": sum(r["precision"] for r in results) / len(results),
            "recall_at_5": sum(r["recall"] for r in results) / len(results),
        }

    return metrics

Step 6: the solution's README

# Metadata-Filtered RAG System

## Architecture

A production-ready pipeline: `metadata filter → hybrid retrieval → rerank → LLM`.

Each component attacks a specific problem:
- Metadata filter: multi-tenant isolation + recency + categorization
- Hybrid (semantic + BM25 + RRF): conceptual coverage + exact match
- Cross-encoder rerank: refine the top-K with better scoring
- LLM with citations: generation with anti-hallucination

## Benchmark results (an eval set of 80 queries)

| Architecture | Precision@5 | Recall@5 | Latency p95 | Data Leak Risk |
|--------------|-------------|----------|--------------|----------------|
| A: Semantic baseline | 72% | 65% | 220ms | HIGH |
| B: + Hybrid | 84% | 78% | 320ms | HIGH |
| C: + Rerank | 91% | 84% | 470ms | HIGH |
| D: + Metadata filter (full) | **92%** | **87%** | **310ms** | **ZERO** |

**The highlight:** architecture D (with the filter) is **faster** than C because it searches
fewer vectors, on top of eliminating the data leak risk.

## Isolation tests

An automated suite in `tests/test_isolation.py`:
- ✅ Cross-tenant leakage: 0 detected
- ✅ Bypass via additional_filters: blocked
- ✅ Concurrent queries from different tenants: isolated
- ✅ workspace_id mandatory: enforced

## Compliance

- ✅ Audit logging for every query (workspace_id, user_id, timestamp, query_hash)
- ✅ Structured logs with a 7-year retention
- ✅ Isolation tests in CI before every deploy
- ✅ A pre-commit hook that blocks a direct collection.query()

## Setup

\`\`\`bash
pip install -r requirements.txt
cp .env.example .env  # fill in OPENAI_API_KEY
python scripts/build_indexes.py
python benchmarks/run_benchmark.py
\`\`\`

## Next steps

- Migration to Pinecone to scale (M07)
- Continuous RAG evaluation (M08)

Project delivery checklist

  • A Pydantic schema validated in the ingest pipeline
  • TenantContext + secure_query mandatory throughout the code
  • A working end-to-end HybridFilteredRetriever
  • Your own eval set of 50+ queries with tenant_id and expected_doc_ids
  • Isolation tests running in CI (5 tests minimum)
  • A pre-commit hook that blocks a direct collection.query()
  • Structured audit logging
  • A benchmark report comparing the 4 architectures
  • A README with the architecture, the metrics, the compliance story
  • Documentation on how to deploy and monitor it

Optional extensions

Extension 1: a metrics dashboard

Build a Grafana (or similar) dashboard showing:

  • Latency p50/p95/p99 per tenant
  • The recall rate per query category
  • An audit log search interface
  • Alerts if a data leak is detected

Extension 2: A/B testing in production

A feature flag for a gradual rollout:

if user.workspace_id in ROLLOUT_GROUP:
    pipeline = HybridFilteredPipeline(...)
else:
    pipeline = LegacyPipeline(...)

Compare the metrics between the two groups for 2 weeks.

Extension 3: automated tenant onboarding

A script that onboards a new tenant:

  • Validates that the schema still meets compliance
  • Indexes the initial docs
  • Creates a TenantContext template
  • Configures the tenant-specific monitoring

Extension 4: pipeline observability

Every step of the pipeline emits traces (OpenTelemetry):

  • Filter time, hybrid time, rerank time
  • The hit rate of each component
  • Bottleneck analysis

Traps and common mistakes in the project

Trap 1: isolation tests that only cover the happy path

Make sure you test:

  • Concurrency (threads from different tenants)
  • Bypass attempts (additional_filters with a workspace_id)
  • Edge cases (an empty tenant_id, None, invalid values)

Trap 2: ignoring audit log retention

Compliance typically requires 7-year retention. It isn't optional.

Trap 3: BM25 with no filter

If BM25 searches the whole corpus even though the vector search is filtered, RRF fuses in cross-tenant docs.

Trap 4: deploying with no smoke test

After deploying, the first mandatory step: a real query from the app verifying it returns docs from the right tenant.

Trap 5: outdated documentation

The README is for the Tech Lead who's going to approve the deploy. If it documents the old architecture, the audit fails.

Project summary

You built:

  • ✅ A complete pipeline (filter + hybrid + rerank) with security by design
  • ✅ A schema validated with Pydantic
  • ✅ Automated isolation tests
  • ✅ Anti-bypass pre-commit hooks
  • ✅ Audit logging for compliance
  • ✅ A comparison benchmark
  • ✅ A README for approval

The typical improvement expected with the full pipeline:

Metric             Baseline         Full pipeline    Gain
Precision@5        72%              92%              +20 pts
Recall@5           65%              87%              +22 pts
Latency p95        220ms            310ms            +90ms (better than without the filter)
Data leak risk     HIGH             ZERO             eliminated

Next steps

Right away:

  1. Run the full benchmark over your eval set.
  2. Validate the isolation tests at 100%.
  3. Review the README with compliance/security.
  4. A gradual deploy behind a feature flag.

Module 7 (Production with Pinecone):

What you built with ChromaDB translates directly to Pinecone:

  • The workspace_id filter → native namespaces
  • BM25 → Pinecone supports sparse vectors (native BM25 since 2024)
  • Audit logging → Pinecone has native logs on the enterprise plans
  • Isolation tests → you already have them; only the backend changes

M07 covers the migration + how to scale to 10M+ vectors while keeping the isolation.

Resources

  1. Pydantic Documentation — Validation
  2. ChromaDB — Metadata Filtering — Reference
  3. SOC2 Compliance Framework — The standard
  4. OWASP Multi-Tenancy Security
  5. GDPR Article 32 — Security context
  6. Anthropic — Contextual Retrieval

Estimated time: 4-6 hours + 1 hour of analysis Next module: Module 7 — Production with Pinecone

← PreviousCapsule 07: Integration with hybrid search — the final "state of the art" patternNext →Module 7: Production with Pinecone — the migration from "working demo" to "24/7 service"