Capsule 07: Integration with hybrid search — the final "state of the art" pattern

Capsule description

We covered each component separately: chunking (M02), query optimization (M03), re-ranking (M04), hybrid search (M05), metadata filtering (this module). Now comes the operational question: how do they all connect into a single pipeline?

This capsule teaches you the "state of the art" architecture for production-ready RAG as of May 2026: query optimization → metadata filter → hybrid retrieval → reranking → generation. Each component reduces a different problem. Combined, they take precision@5 from ~70% (a RAG MVP) up to 92-95% in real production.

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

• ✅ Implement the complete filter → hybrid → rerank pipeline end-to-end
• ✅ Apply metadata filtering to both the vector search and BM25 (both components of the hybrid)
• ✅ Design a staged fallback that preserves isolation while relaxing the optional filters
• ✅ Benchmark the four architectures incrementally: semantic-only, +hybrid, +rerank, +filter
• ✅ Anticipate the failure points when the components interact
• ✅ Decide which components are mandatory vs optional for your context

Estimated time: 30 minutes

The complete architecture

                                The user's query
                                       │
                                       ▼
                   ┌──────────────────────────────────────┐
                   │ Query Optimization (M03)             │
                   │  ├─ Detect the query type            │
                   │  ├─ Rewriting/expansion if it applies│
                   │  └─ Output: the processed query(s)   │
                   └──────────────────┬───────────────────┘
                                      │
                                      ▼
                   ┌──────────────────────────────────────┐
                   │ Metadata Filter (M06 - this module)  │
                   │  ├─ workspace_id, mandatory          │
                   │  ├─ visibility based on role         │
                   │  ├─ time filters (recency)           │
                   │  └─ the relevant tags                 │
                   └──────────────────┬───────────────────┘
                                      │ filter applied
                                      ▼
                   ┌──────────────────────────────────────┐
                   │ Hybrid Retrieval (M05)               │
                   │                                       │
                   │  ┌──────────────────┐  ┌────────────┐│
                   │  │ Semantic search  │  │ BM25       ││
                   │  │ (with the filter)│  │ (w/ filter)│
                   │  │ Top-30           │  │ Top-30     │
                   │  └────────┬─────────┘  └─────┬──────┘│
                   │           │                  │       │
                   │           └──────┬───────────┘       │
                   │                  ▼                    │
                   │           Reciprocal Rank Fusion      │
                   │                  │                    │
                   │                  ▼                    │
                   │             Top-30 fused              │
                   └──────────────────┬───────────────────┘
                                      │
                                      ▼
                   ┌──────────────────────────────────────┐
                   │ Re-ranking (M04)                     │
                   │  Cross-encoder or LLM rerank         │
                   │  → Final top-5                        │
                   └──────────────────┬───────────────────┘
                                      │
                                      ▼
                   ┌──────────────────────────────────────┐
                   │ LLM Generation                       │
                   │  Prompt: the query + the top-5 docs  │
                   │  → The answer                         │
                   └──────────────────────────────────────┘

Each step reduces a different problem:

Implementing the complete pipeline

# pipeline.py
from concurrent.futures import ThreadPoolExecutor
from typing import Optional

import chromadb
from rank_bm25 import BM25Okapi
from sentence_transformers import CrossEncoder


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

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

    def search(
        self,
        query: str,
        tenant: TenantContext,
        additional_filters: Optional[dict] = None,
        n_candidates: int = 30,
        n_final: int = 5,
    ) -> list[dict]:
        """
        End-to-end pipeline: metadata filter → hybrid (sem + BM25) → RRF → rerank.
        """
        # Step 1: Build the secure filter
        where = self._build_filter(tenant, additional_filters)

        # Step 2: Hybrid retrieval in parallel (with the filter applied to both)
        with ThreadPoolExecutor(max_workers=2) as executor:
            future_sem = executor.submit(
                self._semantic_search, query, where, n_candidates
            )
            future_bm25 = executor.submit(
                self._bm25_search, query, where, n_candidates
            )
            sem_ids = future_sem.result()
            bm25_ids = future_bm25.result()

        # Step 3: Fusion with RRF
        fused_ids = self._reciprocal_rank_fusion([sem_ids, bm25_ids])
        top_candidates = fused_ids[:n_candidates]

        # Step 4: Retrieve the content for the rerank
        candidates_data = self.collection.get(
            ids=top_candidates,
            include=["documents", "metadatas"],
        )

        # Step 5: Re-ranking with the cross-encoder
        reranked = self._cross_encoder_rerank(
            query=query,
            candidates=candidates_data,
            top_k=n_final,
        )

        return reranked

    def _build_filter(
        self, tenant: TenantContext, additional: Optional[dict]
    ) -> dict:
        """Builds the filter with a mandatory workspace_id."""
        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")

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

    def _semantic_search(self, query: str, where: dict, n: int) -> list[str]:
        """Vector search with the filter applied."""
        results = self.collection.query(
            query_texts=[query],
            where=where,
            n_results=n,
        )
        return results["ids"][0]

    def _bm25_search(self, query: str, where: dict, n: int) -> list[str]:
        """BM25 search with the filter applied manually over the IDs."""
        query_tokens = query.lower().split()
        all_scores = self.bm25_index.get_scores(query_tokens)

        # The IDs that pass the filter (pre-compute a set for fast lookup)
        allowed_ids = self._get_filtered_ids(where)

        # Take the top-N only among the allowed ones
        scored = [
            (self.all_doc_ids[i], all_scores[i])
            for i in range(len(all_scores))
            if self.all_doc_ids[i] in allowed_ids and all_scores[i] > 0
        ]
        scored.sort(key=lambda x: -x[1])

        return [doc_id for doc_id, _ in scored[:n]]

    def _get_filtered_ids(self, where: dict) -> set[str]:
        """Apply the filter manually over the metadata, for BM25."""
        # Use collection.get with `where` to get the allowed IDs
        result = self.collection.get(where=where, include=[])
        return set(result["ids"])

    def _reciprocal_rank_fusion(
        self, rankings: list[list[str]], k: int = 60
    ) -> list[str]:
        """RRF: combine the rankings while ignoring absolute scores."""
        from collections import defaultdict

        scores = defaultdict(float)
        for ranking in rankings:
            for rank, doc_id in enumerate(ranking, start=1):
                scores[doc_id] += 1.0 / (k + rank)

        sorted_ids = sorted(scores.items(), key=lambda x: -x[1])
        return [doc_id for doc_id, _ in sorted_ids]

    def _cross_encoder_rerank(
        self, query: str, candidates: dict, top_k: int
    ) -> list[dict]:
        """Cross-encoder rerank over the fused candidates."""
        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_indices = 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_indices
        ]

Usage

# Assuming the collection and bm25 are already configured
pipeline = FilteredHybridRagPipeline(
    collection=chroma_collection,
    bm25_index=bm25,
    all_doc_ids=all_ids,
    all_metadatas=all_metas,
)

tenant = TenantContext(
    workspace_id="acme",
    user_id="user_1",
    user_role="member",
)

results = pipeline.search(
    query="how do I configure OAuth2?",
    tenant=tenant,
    additional_filters={
        "$and": [
            {"language": "en"},
            {"created_at": {"$gte": six_months_ago_ts}},
        ]
    },
    n_candidates=30,
    n_final=5,
)

# Pass it to the LLM
context = "\n\n".join(r["document"] for r in results)
answer = generate_answer(query, context)

The staged fallback in the integrated pipeline

If the filter is very restrictive and returns few results, you can escalate the fallback at the pipeline level:

def search_with_pipeline_fallback(
    pipeline,
    query: str,
    tenant: TenantContext,
    optional_filters: dict,
    n_final: int = 5,
):
    """The pipeline with a staged fallback over the optional filters."""
    fallback_steps = [
        optional_filters,                                                # full
        {k: v for k, v in optional_filters.items() if k != "tags"},      # without tags
        {k: v for k, v in optional_filters.items() if k != "created_at"},# without time
        {},                                                               # tenant only
    ]

    for step_filters in fallback_steps:
        try:
            results = pipeline.search(
                query=query,
                tenant=tenant,
                additional_filters=step_filters or None,
                n_final=n_final,
            )
            if len(results) >= 3:
                return {"results": results, "filters_used": step_filters}
        except Exception as e:
            print(f"Step failed: {e}")
            continue

    return {"results": [], "filters_used": None}

workspace_id is ALWAYS preserved — it lives in pipeline._build_filter().

Incremental benchmarking

Demonstrate each component's impact with benchmarks:

def benchmark_incremental(eval_set):
    """Measures each incremental component over the same eval set."""
    results = {}

    # Architecture A: the semantic baseline only
    print("Architecture A (baseline)...")
    results["A_semantic_only"] = evaluate(
        eval_set,
        retriever_fn=lambda q, t: collection.query(query_texts=[q], n_results=5),
    )

    # Architecture B: + hybrid
    print("Architecture B (+ hybrid)...")
    results["B_hybrid"] = evaluate(
        eval_set,
        retriever_fn=lambda q, t: hybrid_search(q, n_results=5),
    )

    # Architecture C: + rerank
    print("Architecture C (+ rerank)...")
    results["C_hybrid_rerank"] = evaluate(
        eval_set,
        retriever_fn=lambda q, t: hybrid_search_with_rerank(q, n_final=5),
    )

    # Architecture D: + metadata filter (the final one)
    print("Architecture D (+ metadata filter — the full pipeline)...")
    pipeline = FilteredHybridRagPipeline(...)
    results["D_full_pipeline"] = evaluate(
        eval_set,
        retriever_fn=lambda q, t: pipeline.search(q, t, n_final=5),
    )

    return results


# Typical expected output:
# A_semantic_only:    Recall@5 = 65%, Precision@5 = 72%, Latency p95 = 220ms
# B_hybrid:           Recall@5 = 78%, Precision@5 = 82%, Latency p95 = 340ms
# C_hybrid_rerank:    Recall@5 = 85%, Precision@5 = 91%, Latency p95 = 480ms
# D_full_pipeline:    Recall@5 = 87%, Precision@5 = 92%, Latency p95 = 290ms  ← the filter improves latency!

The key reading: the metadata filter doesn't only add security, it also improves latency (it searches fewer vectors) and sometimes precision (less noise).

Common traps in the integration

Trap 1: BM25 doesn't respect the filter

The mistake: the vector search applies where, but BM25 searches the whole corpus.

The symptom: RRF fuses tenant A's results (vector) with results from several tenants (BM25). A partial data leak.

How to prevent it: the approach used in the pipeline above — _get_filtered_ids applies the filter manually over the IDs before the BM25 scoring.

Trap 2: the filter skews the RRF ranking

If the vector retrieval with the filter returns 10 docs and BM25 with the filter returns 25, the RRF ranking is biased toward BM25 (more candidates).

How to prevent it: make sure both retrievers return a consistent n_candidates. If a retriever can't find enough docs in the filtered subset, log it as a warning but carry on.

Trap 3: rerank scoring over too little context

After an aggressive filter, you can end up with only 5-8 candidates. A reranker over that few doesn't add much.

How to prevent it: if len(candidates) < min_for_rerank, skip the rerank — the few that exist already passed the filter + RRF, and additional ranking adds no value.

Trap 4: latency rising despite the filter

A correctly applied filter should bring latency down. If it goes up:

• Is tenant_id indexed in ChromaDB? Without an index, the filter is linear.
• Is BM25 applying the filter efficiently, or re-scanning everything?
• Is the rerank batch_size right?

How to prevent it: profile every step of the pipeline. Identify the slow component.

Applied exercise

Scenario: you're an AI Engineer at a DevOps SaaS company. Your current pipeline:

• 50 tenants, ~1M docs total
• Current system: semantic + cross-encoder rerank only
• Metrics: Precision@5 = 80%, Recall@5 = 68%, Latency p95 = 320ms
• Tickets reporting: "I saw another company's doc", "the bot can't find specific error codes"

Your job:

1. Design the final pipeline that attacks both problems (data leak + recall on queries with identifiers).
2. Justify which components you add and in what order.
3. Estimate the expected impact.

class FullPipeline:
    def search(self, query, tenant: TenantContext, n_final=5):
        # Step 1: filter by tenant_id + visibility (CRITICAL for the data leak)
        where = build_secure_filter(tenant)

        # Step 2: hybrid retrieval in parallel
        with ThreadPoolExecutor(max_workers=2) as ex:
            sem_future = ex.submit(semantic_search, query, where, n=30)
            bm25_future = ex.submit(bm25_search_filtered, query, where, n=30)

        sem_ids = sem_future.result()
        bm25_ids = bm25_future.result()

        # Step 3: RRF
        fused = reciprocal_rank_fusion([sem_ids, bm25_ids])[:30]

        # Step 4: retrieve the content + rerank
        candidates = collection.get(ids=fused)
        return cross_encoder_rerank(query, candidates, top_k=n_final)

Metric               Before       After            Change
─────────────────────────────────────────────────────────
Precision@5          80%          92% (+12 pts)    Better
Recall@5             68%          88% (+20 pts)    Dramatically better
Latency p95          320ms        220ms (-100ms)   Better (the filter shrinks the search space)
Data leak risk       HIGH         ZERO             Eliminated
Cost $/month         (unchanged)  +$30 (BM25 infra) Acceptable

Week 1: Implement the metadata filter + secure_query
- Automated isolation tests
- An internal A/B test
- A compliance review

Week 2: Integrate BM25 (rank_bm25 if <1M docs, ES if more)
- Build the index
- Implement bm25_search_filtered with _get_filtered_ids
- Performance tests

Week 3: The complete pipeline + RRF
- Wire it all together
- Benchmark vs the baseline
- Validate the gain on your own eval set

Week 4: Gradual rollout
- Feature flag at 10%
- Monitor data_leak_count (it should be 0)
- Scale to 50% → 100%

Summary and next step

What you learned:

• The complete pipeline: query opt → filter → hybrid (semantic + BM25 with RRF) → rerank → LLM.
• Each component reduces a different problem. Combined, they take precision@5 from 70% to 92-95%.
• BM25 must respect the filter too — not just the vector search.
• The staged fallback preserves tenant_id while relaxing the optional filters.
• The metadata filter improves security AND latency (it searches fewer docs).
• Incremental benchmarking: measure each component to justify its value.

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

• Implement the complete filter → hybrid → rerank pipeline.
• Ensure BM25 respects the filter (not just the vector search).
• Design a staged fallback at the pipeline level.

Next capsule: 08 — The capstone metadata-filtered RAG project.

The close of the module: you'll build this module's complete pipeline — filter + hybrid + rerank + isolation tests + benchmark — as a portfolio project. It's the culmination of M01-M06.

Resources

1. Anthropic — Contextual Retrieval — A complementary technique
2. Pinecone — Hybrid Search — A visual tutorial
3. LangChain — EnsembleRetriever — A reference implementation
4. LlamaIndex — Multi-Step Query Engine — Advanced patterns
5. BEIR Benchmark — Empirical comparisons
6. Cross-Encoders Guide — Reranking

Estimated time: 30 minutes Next: 08-project-metadata-filtered-rag.md