You've built your RAG pipeline. You've got a vector store loaded with chunked documents, an embedding model doing its job, and a retriever pulling back the top-k results. And yet, your answers are still frustrating. The LLM keeps picking up on tangentially relevant chunks while ignoring the one paragraph that actually answers the question. Your retrieval step is returning related content, not relevant content — and that distinction is everything.
This is the gap that a reranking layer fills. Standard vector retrieval is optimized for speed and approximate similarity: it finds documents that live near your query in embedding space. But embedding-based similarity is a coarse instrument. It can't reason about the precise semantic relationship between a question and a candidate passage the way a model can when it reads both simultaneously. Rerankers do exactly that — they score each retrieved candidate against the original query with full attention across both, and return a much more precise relevance ranking.
By the end of this lesson, you'll have built a complete, production-ready reranking layer that integrates seamlessly into a RAG pipeline. You'll understand when to use it, how to make it fast enough to actually deploy, and how to evaluate whether it's actually helping.
What you'll learn:
sentence-transformersYou should be comfortable with:
sentence-transformers and at least one LLM API (OpenAI, Anthropic, or similar)Let's make the problem concrete. Imagine you're building a RAG system for a large enterprise software company's internal knowledge base. A support engineer asks:
"What's the maximum file upload size allowed for the legacy API v2 endpoint?"
Your retriever pulls back the top 5 chunks based on embedding similarity. You get:
The correct answer is in chunk 5 — but it ranked last because the dense embedding for that chunk looked less similar to the query than the architectural overview at the top. Your LLM will likely hallucinate or give an answer based on the v3 limits.
This happens because bi-encoders (the models that generate your embeddings) encode the query and each document independently. They never see both at the same time. The resulting vectors capture general semantic meaning well, but miss fine-grained relevance signals like: "this exact document answers this exact question."
A cross-encoder fixes this by taking both the query and the candidate document as a single concatenated input and running attention across all tokens simultaneously. It can reason about specific term matches, negations, numerical specificity, and context — at the cost of being slower, since you can't pre-compute anything.
This creates a natural two-stage architecture:
Key insight: You're trading a small amount of latency in stage 2 to dramatically improve the quality of what your LLM actually sees. Given that your LLM call is already the most expensive part of the pipeline, this trade-off almost always makes sense.
Let's establish our working environment and base pipeline first. We'll build on this throughout the lesson.
pip install sentence-transformers chromadb openai tiktoken numpy
Here's our base retrieval system using a realistic document corpus — internal HR policy documents:
import chromadb
from sentence_transformers import SentenceTransformer
from dataclasses import dataclass, field
from typing import Optional
import json
@dataclass
class RetrievedChunk:
chunk_id: str
text: str
source: str
initial_score: float # cosine similarity from bi-encoder
rerank_score: Optional[float] = None # populated after reranking
metadata: dict = field(default_factory=dict)
class BaseRetriever:
def __init__(self, collection_name: str = "hr_policies"):
self.client = chromadb.Client()
self.embed_model = SentenceTransformer("all-MiniLM-L6-v2")
self.collection = self.client.get_or_create_collection(collection_name)
self._seed_example_documents()
def _seed_example_documents(self):
"""Load a small but realistic HR policy corpus."""
documents = [
{
"id": "pto-001",
"text": "Full-time employees accrue 15 days of PTO per calendar year during their first three years of employment. PTO accrual increases to 20 days per year after 3 years of continuous service.",
"source": "pto_policy_v4.pdf"
},
{
"id": "pto-002",
"text": "PTO requests must be submitted at least 5 business days in advance for absences of 3 or more consecutive days. For single-day absences, same-day notification to your direct manager is acceptable.",
"source": "pto_policy_v4.pdf"
},
{
"id": "pto-003",
"text": "Unused PTO may be carried over to the following calendar year, subject to a maximum carryover of 10 days. PTO does not carry a cash value and is not paid out upon voluntary resignation.",
"source": "pto_policy_v4.pdf"
},
{
"id": "benefits-001",
"text": "The company offers 12 weeks of paid parental leave for primary caregivers and 4 weeks for secondary caregivers, available after 6 months of employment.",
"source": "benefits_guide_2024.pdf"
},
{
"id": "benefits-002",
"text": "Annual leave requests for the holiday period (December 20 through January 3) must be submitted by October 31st. Approvals are granted based on team operational requirements and seniority.",
"source": "benefits_guide_2024.pdf"
},
{
"id": "remote-001",
"text": "Employees working remotely are eligible for a one-time home office stipend of $800 and a monthly internet reimbursement of $50, subject to manager approval.",
"source": "remote_work_policy.pdf"
},
{
"id": "remote-002",
"text": "Remote employees must maintain core working hours of 10am-3pm in their local time zone and be reachable via Slack during those hours.",
"source": "remote_work_policy.pdf"
},
{
"id": "perf-001",
"text": "Performance reviews occur twice per year: mid-year check-ins in June and formal annual reviews in December. Compensation adjustments resulting from reviews take effect January 1st.",
"source": "performance_review_process.pdf"
}
]
existing = self.collection.get()
if len(existing["ids"]) > 0:
return
embeddings = self.embed_model.encode(
[d["text"] for d in documents]
).tolist()
self.collection.add(
ids=[d["id"] for d in documents],
documents=[d["text"] for d in documents],
embeddings=embeddings,
metadatas=[{"source": d["source"]} for d in documents]
)
print(f"Seeded {len(documents)} documents into collection.")
def retrieve(self, query: str, top_k: int = 5) -> list[RetrievedChunk]:
query_embedding = self.embed_model.encode(query).tolist()
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=top_k
)
chunks = []
for i, doc_id in enumerate(results["ids"][0]):
chunks.append(RetrievedChunk(
chunk_id=doc_id,
text=results["documents"][0][i],
source=results["metadatas"][0][i]["source"],
initial_score=1 - results["distances"][0][i], # convert distance to similarity
))
return chunks
Now let's see what vanilla retrieval gives us on a specific query:
retriever = BaseRetriever()
query = "How many vacation days do I get after working here for 5 years?"
results = retriever.retrieve(query, top_k=5)
print(f"Query: {query}\n")
for i, chunk in enumerate(results, 1):
print(f"Rank {i} | Score: {chunk.initial_score:.4f} | Source: {chunk.source}")
print(f" {chunk.text[:120]}...")
print()
You'll typically see the results in a reasonable but imperfect order. Now let's build the reranking layer that fixes the imperfections.
The sentence-transformers library ships with several pre-trained cross-encoder models. The workhorse for general reranking is cross-encoder/ms-marco-MiniLM-L-6-v2, which was trained on the MS MARCO passage retrieval dataset — a large collection of real Bing search queries and human relevance judgments. It's fast and surprisingly capable.
from sentence_transformers import CrossEncoder
import time
class CrossEncoderReranker:
"""
Reranks retrieved chunks using a cross-encoder model.
Scores query-document pairs jointly, giving much higher precision
than bi-encoder similarity alone.
"""
def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):
print(f"Loading cross-encoder: {model_name}")
self.model = CrossEncoder(model_name)
self.model_name = model_name
def rerank(
self,
query: str,
chunks: list[RetrievedChunk],
top_n: int = 3
) -> list[RetrievedChunk]:
"""
Reranks a list of retrieved chunks and returns the top_n.
The cross-encoder scores each (query, document) pair on a scale
that roughly corresponds to relevance probability — higher is better.
"""
if not chunks:
return chunks
# Build input pairs — this is the key difference from bi-encoders
# Both query and document are seen together
pairs = [(query, chunk.text) for chunk in chunks]
start = time.perf_counter()
scores = self.model.predict(pairs)
elapsed = time.perf_counter() - start
# Attach scores to chunks
for chunk, score in zip(chunks, scores):
chunk.rerank_score = float(score)
# Sort by rerank score descending
reranked = sorted(chunks, key=lambda c: c.rerank_score, reverse=True)
print(f" [CrossEncoder] Scored {len(chunks)} chunks in {elapsed*1000:.1f}ms")
return reranked[:top_n]
Let's run this and compare:
reranker = CrossEncoderReranker()
query = "How many vacation days do I get after working here for 5 years?"
initial_results = retriever.retrieve(query, top_k=8) # fetch more for reranker to work with
print("=== BEFORE RERANKING ===")
for i, chunk in enumerate(initial_results[:5], 1):
print(f" Rank {i} | Embed Score: {chunk.initial_score:.4f}")
print(f" {chunk.text[:100]}...")
print("\n=== AFTER CROSS-ENCODER RERANKING ===")
reranked = reranker.rerank(query, initial_results, top_n=3)
for i, chunk in enumerate(reranked, 1):
print(f" Rank {i} | Rerank Score: {chunk.rerank_score:.4f} | Was initially #{initial_results.index(chunk)+1}")
print(f" {chunk.text[:100]}...")
Notice that pto-001 (the chunk about accrual rates mentioning "first three years" and implying what happens after) and the carryover chunk typically rank much more precisely after reranking. The cross-encoder picks up on the temporal specificity that the embedding model flattened.
Model selection tip: For production, consider
cross-encoder/ms-marco-MiniLM-L-12-v2(more layers, more accurate, ~2x slower) orBAAI/bge-reranker-largefor better multilingual and domain-specific performance. Always benchmark on your actual data — the MS MARCO models are trained on web search queries, which may not match your domain's language patterns.
Cross-encoders are great, but they have a hard ceiling: they're limited by their training data. For highly specialized domains — legal documents, medical literature, proprietary codebases — a well-prompted LLM can outperform a cross-encoder because it brings broader world knowledge and can follow explicit relevance criteria you define in the prompt.
The tradeoff is cost and latency. An LLM-based reranker makes one API call per candidate document (or a carefully batched version), which adds up. The trick is to do this thoughtfully.
from openai import OpenAI
import json
import asyncio
from concurrent.futures import ThreadPoolExecutor
class LLMReranker:
"""
Uses an LLM to score relevance of each retrieved chunk against the query.
More expensive than cross-encoders but more flexible for specialized domains
or when you need to score against complex relevance criteria.
"""
SCORE_PROMPT = """You are evaluating whether a document passage is relevant to a user's question.
Question: {query}
Passage:
{passage}
Score the relevance of this passage for answering the question above.
Return ONLY a JSON object with two fields:
- "score": a number from 0 to 10 (0 = completely irrelevant, 10 = directly and completely answers the question)
- "reason": one sentence explaining your score
Examples of scoring:
- Score 0-2: Passage is on a different topic or shares only surface-level keywords
- Score 3-5: Passage is related but doesn't directly address the specific question
- Score 6-8: Passage partially answers the question or contains relevant supporting information
- Score 9-10: Passage directly and specifically answers the question
JSON response only:"""
def __init__(self, model: str = "gpt-4o-mini", max_workers: int = 5):
self.client = OpenAI()
self.model = model
self.max_workers = max_workers
def _score_single(self, query: str, chunk: RetrievedChunk) -> RetrievedChunk:
"""Score a single chunk. Called in parallel for efficiency."""
prompt = self.SCORE_PROMPT.format(
query=query,
passage=chunk.text
)
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0, # deterministic scoring
response_format={"type": "json_object"},
max_tokens=100
)
result = json.loads(response.choices[0].message.content)
chunk.rerank_score = float(result.get("score", 0)) / 10.0 # normalize to 0-1
chunk.metadata["rerank_reason"] = result.get("reason", "")
except Exception as e:
print(f" Warning: LLM scoring failed for chunk {chunk.chunk_id}: {e}")
chunk.rerank_score = chunk.initial_score # fall back to embedding score
return chunk
def rerank(
self,
query: str,
chunks: list[RetrievedChunk],
top_n: int = 3
) -> list[RetrievedChunk]:
"""
Reranks chunks using parallel LLM scoring.
Uses a thread pool to score multiple chunks concurrently,
which is critical for keeping latency manageable.
"""
if not chunks:
return chunks
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
scored_chunks = list(
executor.map(
lambda chunk: self._score_single(query, chunk),
chunks
)
)
elapsed = time.perf_counter() - start
reranked = sorted(scored_chunks, key=lambda c: c.rerank_score, reverse=True)
print(f" [LLMReranker] Scored {len(chunks)} chunks in {elapsed*1000:.1f}ms "
f"using {self.model}")
# Print reasoning for transparency
for i, chunk in enumerate(reranked[:top_n], 1):
reason = chunk.metadata.get("rerank_reason", "")
print(f" Rank {i} | Score: {chunk.rerank_score:.2f} | {reason}")
return reranked[:top_n]
Cost warning: At current GPT-4o-mini pricing (~$0.15/1M input tokens), scoring 20 chunks of ~200 tokens each runs about $0.0006 per query. That's negligible. But if you're running thousands of queries per hour, add it up. For high-throughput systems, consider caching reranker scores for common queries or falling back to the cross-encoder for bulk traffic.
Now let's wire everything together into a production-quality pipeline that's actually configurable and observable:
from enum import Enum
from typing import Literal
class RerankerStrategy(Enum):
NONE = "none"
CROSS_ENCODER = "cross_encoder"
LLM = "llm"
HYBRID = "hybrid" # cross-encoder first, LLM for top candidates
@dataclass
class RAGPipelineConfig:
initial_retrieval_k: int = 20 # how many to fetch from vector store
rerank_top_n: int = 5 # how many to pass to LLM after reranking
strategy: RerankerStrategy = RerankerStrategy.CROSS_ENCODER
hybrid_llm_top_n: int = 5 # in hybrid mode, how many cross-encoder winners to re-score with LLM
score_threshold: float = 0.0 # filter out chunks below this rerank score (0 = no filtering)
class TwoStageRAGPipeline:
"""
A complete RAG retrieval pipeline with pluggable reranking strategies.
Architecture:
Query → Vector Retrieval (top-k) → Reranker → Top-N → LLM
"""
def __init__(self, config: RAGPipelineConfig):
self.config = config
self.retriever = BaseRetriever()
self._cross_encoder = None
self._llm_reranker = None
@property
def cross_encoder(self) -> CrossEncoderReranker:
if self._cross_encoder is None:
self._cross_encoder = CrossEncoderReranker()
return self._cross_encoder
@property
def llm_reranker(self) -> LLMReranker:
if self._llm_reranker is None:
self._llm_reranker = LLMReranker()
return self._llm_reranker
def retrieve_and_rerank(self, query: str) -> list[RetrievedChunk]:
"""
Full retrieval + reranking pipeline.
Returns chunks sorted by final relevance score.
"""
print(f"\nQuery: '{query}'")
print(f"Strategy: {self.config.strategy.value}")
print("-" * 60)
# Stage 1: Broad retrieval
initial_chunks = self.retriever.retrieve(
query, top_k=self.config.initial_retrieval_k
)
print(f"Stage 1: Retrieved {len(initial_chunks)} candidates from vector store")
# Stage 2: Reranking
if self.config.strategy == RerankerStrategy.NONE:
final_chunks = initial_chunks[:self.config.rerank_top_n]
elif self.config.strategy == RerankerStrategy.CROSS_ENCODER:
final_chunks = self.cross_encoder.rerank(
query, initial_chunks, top_n=self.config.rerank_top_n
)
elif self.config.strategy == RerankerStrategy.LLM:
final_chunks = self.llm_reranker.rerank(
query, initial_chunks, top_n=self.config.rerank_top_n
)
elif self.config.strategy == RerankerStrategy.HYBRID:
# First pass: cross-encoder narrows the field
cross_encoder_top = self.cross_encoder.rerank(
query, initial_chunks, top_n=self.config.hybrid_llm_top_n
)
# Second pass: LLM makes the final call on the finalists
print(f"Stage 2b: LLM rescoring top {len(cross_encoder_top)} from cross-encoder")
final_chunks = self.llm_reranker.rerank(
query, cross_encoder_top, top_n=self.config.rerank_top_n
)
# Optional score threshold filtering
if self.config.score_threshold > 0:
before_filter = len(final_chunks)
final_chunks = [
c for c in final_chunks
if (c.rerank_score or c.initial_score) >= self.config.score_threshold
]
if len(final_chunks) < before_filter:
print(f" Score threshold {self.config.score_threshold} filtered "
f"{before_filter - len(final_chunks)} low-confidence chunks")
return final_chunks
def generate_answer(self, query: str, llm_client: OpenAI) -> dict:
"""
Full end-to-end RAG: retrieve, rerank, generate.
Returns the answer plus retrieval metadata for evaluation.
"""
chunks = self.retrieve_and_rerank(query)
if not chunks:
return {
"answer": "I couldn't find relevant information to answer this question.",
"sources": [],
"chunks_used": 0
}
context = "\n\n---\n\n".join([
f"[Source: {c.source}]\n{c.text}" for c in chunks
])
response = llm_client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": (
"You are a helpful HR assistant. Answer questions based strictly "
"on the provided context. If the context doesn't contain enough "
"information to answer, say so clearly."
)
},
{
"role": "user",
"content": f"Context:\n{context}\n\nQuestion: {query}"
}
],
temperature=0.1
)
return {
"answer": response.choices[0].message.content,
"sources": list(set(c.source for c in chunks)),
"chunks_used": len(chunks),
"top_chunk_score": chunks[0].rerank_score if chunks[0].rerank_score else chunks[0].initial_score
}
Building the reranker is the easy part. Knowing whether it's working is what separates engineers from practitioners. Let's build a lightweight evaluation harness.
For RAG evaluation, we care about two things at the retrieval level:
from dataclasses import dataclass
import statistics
@dataclass
class EvalQuery:
query: str
relevant_chunk_ids: list[str] # ground truth — which chunks should be in top results
description: str = ""
def evaluate_pipeline(
pipeline: TwoStageRAGPipeline,
eval_set: list[EvalQuery],
k: int = 3
) -> dict:
"""
Evaluate retrieval quality on a labeled eval set.
Returns Recall@k and MRR metrics.
"""
recall_scores = []
reciprocal_ranks = []
for eval_query in eval_set:
results = pipeline.retrieve_and_rerank(eval_query.query)
retrieved_ids = [c.chunk_id for c in results[:k]]
# Recall@k: did any relevant chunk appear in top-k?
hits = set(retrieved_ids) & set(eval_query.relevant_chunk_ids)
recall = 1.0 if hits else 0.0
recall_scores.append(recall)
# MRR: where was the first relevant chunk?
rr = 0.0
for rank, chunk_id in enumerate(retrieved_ids, 1):
if chunk_id in eval_query.relevant_chunk_ids:
rr = 1.0 / rank
break
reciprocal_ranks.append(rr)
# Human-readable output
status = "✓ HIT" if recall else "✗ MISS"
print(f"{status} | RR: {rr:.2f} | Query: {eval_query.query[:60]}...")
return {
f"Recall@{k}": statistics.mean(recall_scores),
"MRR": statistics.mean(reciprocal_ranks),
"n_queries": len(eval_set)
}
# Define a ground-truth eval set
eval_queries = [
EvalQuery(
query="How many vacation days do I get after 5 years?",
relevant_chunk_ids=["pto-001"],
description="PTO accrual rate after tenure threshold"
),
EvalQuery(
query="Can I carry unused vacation days into next year?",
relevant_chunk_ids=["pto-003"],
description="PTO carryover policy"
),
EvalQuery(
query="How far in advance do I need to request time off for a week-long vacation?",
relevant_chunk_ids=["pto-002"],
description="PTO advance notice requirement"
),
EvalQuery(
query="What do I get for setting up my home office?",
relevant_chunk_ids=["remote-001"],
description="Remote work equipment stipend"
),
EvalQuery(
query="When do salary increases from performance reviews kick in?",
relevant_chunk_ids=["perf-001"],
description="Performance review compensation timing"
),
]
# Compare strategies
from openai import OpenAI
for strategy in [RerankerStrategy.NONE, RerankerStrategy.CROSS_ENCODER]:
config = RAGPipelineConfig(
initial_retrieval_k=8,
rerank_top_n=3,
strategy=strategy
)
pipeline = TwoStageRAGPipeline(config)
print(f"\n{'='*60}")
print(f"Strategy: {strategy.value}")
print('='*60)
metrics = evaluate_pipeline(pipeline, eval_queries, k=3)
print(f"\nResults: {metrics}")
On building eval sets: This is the highest-leverage thing you can do for any RAG project. Even 20-30 labeled query-document pairs will give you signal about whether your changes are helping or hurting. Treat retrieval quality as a metric you track over time, not a one-time setup decision.
Now let's build something more realistic: a reranker that adapts its scoring criteria based on the type of query being asked. This is a pattern you'd use when your RAG system handles multiple categories of questions with different relevance requirements.
The scenario: Your HR RAG system handles three types of questions with different precision requirements:
class AdaptiveLLMReranker:
"""
An LLM reranker that adjusts its scoring prompt based on detected query intent.
Demonstrates how to encode domain knowledge into your reranking strategy.
"""
PROMPTS = {
"compliance": """You are evaluating whether this passage provides the exact policy rule or number needed to answer a compliance question.
Question: {query}
Passage: {passage}
Score 0-10. A score of 9-10 requires the passage to contain a specific policy rule, number, date, or procedure that directly answers the question. Vague or general passages score 0-3 even if topically related.
Return JSON: {{"score": <number>, "reason": "<one sentence>"}}""",
"eligibility": """You are checking whether this passage defines eligibility criteria relevant to the question.
Question: {query}
Passage: {passage}
Score 0-10. A score of 9-10 requires the passage to explicitly state who qualifies and under what conditions. Look for tenure requirements, employment type, and specific conditions.
Return JSON: {{"score": <number>, "reason": "<one sentence>"}}""",
"general": """You are evaluating relevance of a passage to a general HR question.
Question: {query}
Passage: {passage}
Score 0-10 based on how well the passage helps answer the question.
Return JSON: {{"score": <number>, "reason": "<one sentence>"}}"""
}
def __init__(self, llm_client: OpenAI, model: str = "gpt-4o-mini"):
self.client = llm_client
self.model = model
def _classify_query(self, query: str) -> str:
"""Quick query classification to select the right scoring prompt."""
query_lower = query.lower()
eligibility_signals = ["eligible", "qualify", "can i", "am i", "do i get", "who gets"]
compliance_signals = ["maximum", "minimum", "required", "must", "deadline", "how many days",
"how much", "what is the limit", "policy"]
if any(s in query_lower for s in eligibility_signals):
return "eligibility"
elif any(s in query_lower for s in compliance_signals):
return "compliance"
else:
return "general"
def _score_chunk(self, query: str, chunk: RetrievedChunk, query_type: str) -> float:
prompt_template = self.PROMPTS[query_type]
prompt = prompt_template.format(query=query, passage=chunk.text)
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0,
response_format={"type": "json_object"},
max_tokens=80
)
result = json.loads(response.choices[0].message.content)
score = float(result.get("score", 0)) / 10.0
chunk.metadata["query_type"] = query_type
chunk.metadata["rerank_reason"] = result.get("reason", "")
return score
def rerank(self, query: str, chunks: list[RetrievedChunk], top_n: int = 3) -> list[RetrievedChunk]:
query_type = self._classify_query(query)
print(f" [AdaptiveReranker] Detected query type: '{query_type}'")
with ThreadPoolExecutor(max_workers=5) as executor:
scores = list(executor.map(
lambda c: self._score_chunk(query, c, query_type),
chunks
))
for chunk, score in zip(chunks, scores):
chunk.rerank_score = score
reranked = sorted(chunks, key=lambda c: c.rerank_score, reverse=True)
return reranked[:top_n]
# Test the adaptive reranker
client = OpenAI()
adaptive_reranker = AdaptiveLLMReranker(client)
test_queries = [
"How many vacation days do I get?", # compliance
"Am I eligible for parental leave?", # eligibility
"Tell me about the performance review process" # general
]
base_retriever = BaseRetriever()
for q in test_queries:
chunks = base_retriever.retrieve(q, top_k=5)
print(f"\nQuery: {q}")
reranked = adaptive_reranker.rerank(q, chunks, top_n=2)
for i, c in enumerate(reranked, 1):
print(f" {i}. [{c.rerank_score:.2f}] {c.text[:80]}...")
print(f" Reason: {c.metadata.get('rerank_reason', '')}")
This is the most common setup error. If you fetch top_k=3 from your vector store and then rerank, you haven't given the reranker anything to work with — you've just slowed down retrieval that was already decent. The reranker needs room to maneuver.
Rule of thumb: Fetch 4-10x more candidates from the vector store than you plan to pass to your LLM. If you want 3 context chunks for your LLM, retrieve 15-20 for the reranker.
The ms-marco cross-encoders output raw logits, not probabilities. A score of -3.0 is not "30% relevant." Scores are only meaningful for relative ranking within a single query. Never compare scores across different queries or use them as absolute confidence thresholds without calibration.
If you need absolute thresholds, normalize the scores using softmax across all candidates, or use the LLM reranker which gives you a 0-10 scale that's easier to reason about.
Cross-encoder latency scales linearly with both the number of candidates and the document length. A batch of 20 chunks at ~200 tokens each takes roughly 40-80ms on CPU. On GPU, that drops to 5-15ms. If you're running on CPU in production, this matters.
# Profile your specific setup before committing to a config
import time
reranker = CrossEncoderReranker()
test_chunks = base_retriever.retrieve("vacation policy", top_k=20)
for batch_size in [5, 10, 15, 20]:
subset = test_chunks[:batch_size]
start = time.perf_counter()
for _ in range(10): # average over 10 runs
reranker.rerank("vacation days accrual", subset, top_n=3)
elapsed = (time.perf_counter() - start) / 10
print(f"Batch size {batch_size:2d}: {elapsed*1000:.1f}ms avg")
The ms-marco models were trained on web search data. They're excellent at ranking passages against search-engine-style queries, but they can behave unexpectedly with technical jargon, legal language, or dense scientific text. Always validate on a sample from your actual data before deploying.
For specialized domains, look at:
BAAI/bge-reranker-large — strong multilingual and technical domain performancecross-encoder/nli-deberta-v3-base — trained on natural language inference, better at detecting contradictions and entailmentsentence-transformers library makes this straightforward)Enterprise RAG systems often have high query repetition — the same questions get asked by different users. A simple query-level cache with a reasonable TTL can eliminate most reranker latency for real-world traffic:
from functools import lru_cache
import hashlib
class CachedCrossEncoderReranker(CrossEncoderReranker):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self._cache = {}
def rerank(self, query: str, chunks: list[RetrievedChunk], top_n: int = 3):
# Create a cache key from query + chunk IDs (order-independent)
chunk_ids_hash = hashlib.md5(
"|".join(sorted(c.chunk_id for c in chunks)).encode()
).hexdigest()
cache_key = f"{query}::{chunk_ids_hash}::{top_n}"
if cache_key in self._cache:
print(" [CrossEncoder] Cache hit — skipping reranking")
return self._cache[cache_key]
result = super().rerank(query, chunks, top_n)
self._cache[cache_key] = result
return result
You've built a complete, production-ready reranking layer for RAG. Here's what you now have in your toolkit:
The central lesson is architectural: retrieval and ranking are separate problems that deserve separate solutions. Your vector store is excellent at approximate recall — use it for that. Your reranker is excellent at precise ranking — don't ask your vector store to do that job.
When to use each strategy:
Where to go from here:
rank_bm25 or Elasticsearch) before the reranking step. Dense retrieval handles semantic similarity; sparse retrieval handles exact term matches. Together, they cover more ground for the reranker to work with.The reranking layer is one of those improvements that tends to work immediately and visibly. Once you add it to a production system, you'll wonder how you shipped anything without it.
Learning Path: Building with LLMs