Reranking in RAG Pipelines: When and How to Use Cross-Encoders vs Bi-Encoders

Retrieval-Augmented Generation (RAG) performance is bounded by the quality of the context supplied to the LLM. If your retrieval step introduces irrelevant documents (false positives), the language model either hallucinates or produces diluted, unfocused answers.

In 2026, the industry standard pattern for high-accuracy retrieval is the Two-Stage RAG Pipeline:
1. Stage 1 (Bi-Encoder / Dense Vector Retrieval): Rapidly scans millions of vectors to extract the top-50 candidates in under 15ms.
2. Stage 2 (Cross-Encoder Reranker): Performs deep, full self-attention between the query and candidate documents to rerank candidates, feeding only the top-5 highest-scoring passages to the generation model.

In this tutorial, we implement a production Two-Stage Reranking Pipeline in Python using open-weights Cross-Encoders (bge-reranker-large).

1. Architecture: Bi-Encoder vs Cross-Encoder Self-Attention

┌────────────────────────────────────────────────────────┐
│                     Inbound Query                      │
└───────────────────────────┬────────────────────────────┘
                            │
                            ▼ Stage 1: Fast Bi-Encoder
┌────────────────────────────────────────────────────────┐
│      pgvector / Vector Search (Candidate Recall)       │
│      Cosine Similarity: DotProduct(Emb(Q), Emb(D))     │
│      Latency: ~12ms | Filters 1,000,000 docs ➔ Top-50  │
└───────────────────────────┬────────────────────────────┘
                            │ Top-50 Candidates
                            ▼ Stage 2: Deep Cross-Encoder
┌────────────────────────────────────────────────────────┐
│           Cross-Encoder Reranker (bge-reranker)        │
│       Joint Attention: Transformer([Query, Doc])       │
│       Latency: ~35ms | Selects true Top-5 relevant     │
└───────────────────────────┬────────────────────────────┘
                            │ Pure Top-5 Grounded Context
                            ▼
┌────────────────────────────────────────────────────────┐
│                   LLM Generation Node                  │
└────────────────────────────────────────────────────────┘

2. Environment Setup

pip install sentence-transformers>=3.0.0 torch>=2.2.0 pydantic>=2.7.0

3. Implementation: Two-Stage Reranking Engine

Create two_stage_reranker.py:

from typing import List, Dict, Any
from sentence_transformers import SentenceTransformer, CrossEncoder
from pydantic import BaseModel

class RankedDocument(BaseModel):
    doc_id: int
    content: str
    initial_vector_score: float
    rerank_score: float

class TwoStageRetriever:
    def __init__(
        self,
        bi_encoder_model: str = "all-MiniLM-L6-v2",
        cross_encoder_model: str = "BAAI/bge-reranker-base"
    ):
        print(f"Loading Bi-Encoder ({bi_encoder_model})...")
        self.bi_encoder = SentenceTransformer(bi_encoder_model)
        print(f"Loading Cross-Encoder Reranker ({cross_encoder_model})...")
        self.reranker = CrossEncoder(cross_encoder_model)
        self.corpus: List[str] = []

    def index(self, documents: List[str]):
        self.corpus = documents
        self.embeddings = self.bi_encoder.encode(documents, normalize_embeddings=True)

    def retrieve_and_rerank(self, query: str, top_candidates: int = 15, final_top_k: int = 3) -> List[RankedDocument]:
        # Stage 1: Fast Vector Search (Cosine Similarity)
        query_vec = self.bi_encoder.encode([query], normalize_embeddings=True)[0]
        import numpy as np
        sims = np.dot(self.embeddings, query_vec)
        candidate_indices = np.argsort(sims)[::-1][:top_candidates]

        # Prepare (Query, Candidate) pairs for Cross-Encoder
        candidates = [self.corpus[idx] for idx in candidate_indices]
        pairs = [[query, doc] for doc in candidates]

        # Stage 2: Deep Cross-Encoder Scoring
        rerank_scores = self.reranker.predict(pairs)
        ranked_order = np.argsort(rerank_scores)[::-1][:final_top_k]

        results = []
        for rank_idx in ranked_order:
            orig_doc_idx = candidate_indices[rank_idx]
            results.append(RankedDocument(
                doc_id=int(orig_doc_idx),
                content=self.corpus[orig_doc_idx],
                initial_vector_score=round(float(sims[orig_doc_idx]), 4),
                rerank_score=round(float(rerank_scores[rank_idx]), 4)
            ))
        return results

if __name__ == "__main__":
    docs = [
        "Python 3.12 introduces improved error messages and isolated sub-interpreters.",
        "PostgreSQL pgvector extension enables high-performance vector search in relational tables.",
        "To terminate idle PostgreSQL transactions, execute SELECT pg_terminate_backend(pid).",
        "FastAPI StreamingResponse uses Server-Sent Events to stream LLM responses in real-time.",
        "Model Context Protocol standardizes tool definitions for AI agents."
    ]

    retriever = TwoStageRetriever()
    retriever.index(docs)

    query = "How to kill hung database queries in Postgres?"
    results = retriever.retrieve_and_rerank(query, top_candidates=4, final_top_k=2)

    print(f"\nQuery: '{query}'")
    for r in results:
        print(f"  [Rerank Score: {r.rerank_score} | Vector Score: {r.initial_vector_score}]")
        print(f"  Content: {r.content}\n")

4. Benchmark: Accuracy Impact of Cross-Encoder Reranking

5. Frequently Asked Questions (FAQ)

Why not use Cross-Encoders for the entire million-document corpus?

Cross-encoders jointly process the query and document through all transformer attention layers simultaneously. Scoring 1,000,000 documents with a cross-encoder would take 30+ seconds per query. The two-stage pattern gives you the speed of vector search with the precision of cross-encoders.