Reranking search results with a cross-encoder LLM

The typical vector search pipeline retrieves the top-K most similar documents using embedding similarity. It is fast — milliseconds for millions of documents — but similarity in embedding space does not always mean relevance to the specific query. A search for "Python async performance tips" might retrieve an article about Python async syntax (high embedding similarity) that does not actually contain performance tips.

Reranking fixes this. After retrieving candidates with your fast vector search, a reranker scores each candidate against the full query and reorders them by actual relevance. The result is dramatically better precision at the top positions, which is exactly where it matters for search applications.

Why reranking works

Bi-encoders (the embedding models used for vector search) encode the query and documents independently and compare them in vector space. This is fast but loses the fine-grained interaction between query and document.

Cross-encoders process the query and document together, allowing full attention between every token in the query and every token in the document. This captures nuances — the presence of specific keywords, the way terms relate to each other — that bi-encoders miss. The cost is that you cannot pre-compute cross-encoder scores; they must be computed per query per candidate.

LLMs make particularly effective rerankers because they understand language deeply and can be prompted to explain their relevance judgments, which aids debugging.

The two-stage pipeline

Query → Bi-encoder retrieval (fast, recall-oriented) → Top 50-100 candidates
     → Cross-encoder reranking (slower, precision-oriented) → Top 5-10 results

Stage one maximises recall: retrieve more than you need, accepting some irrelevant results. Stage two maximises precision: reorder the candidates to surface the most genuinely relevant ones first.

Implementation

from anthropic import Anthropic
import json
from dataclasses import dataclass
from typing import Optional

client = Anthropic()

@dataclass
class SearchResult:
    id: str
    content: str
    metadata: dict
    initial_score: float       # From bi-encoder retrieval
    rerank_score: Optional[float] = None  # From cross-encoder

def rerank_with_llm(
    query: str,
    candidates: list[SearchResult],
    top_k: int = 5,
) -> list[SearchResult]:
    """
    Rerank search results using an LLM as a cross-encoder.
    Returns the top_k most relevant results, sorted by relevance.
    """
    if not candidates:
        return []

    # Format candidates for the prompt
    candidates_text = "\n\n".join(
        f"[{i+1}] ID: {c.id}\n{c.content[:500]}"  # Truncate for token budget
        for i, c in enumerate(candidates)
    )

    response = client.messages.create(
        model="claude-haiku-4-5",  # Haiku for speed and cost
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"""Score each document for relevance to the query.

QUERY: {query}

DOCUMENTS:
{candidates_text}

For each document, provide a relevance score from 0.0 (completely irrelevant) 
to 1.0 (perfectly answers the query). Consider:
- Does the document directly answer the query?
- Does it contain the specific information asked for?
- Is the content accurate and detailed for this topic?

Return JSON array with objects: {{"id": str, "score": float, "reason": str}}
Order by score descending. Include all {len(candidates)} documents."""
        }]
    )

    text = response.content[0].text
    if "```" in text:
        text = text.split("```")[1].replace("json", "").strip()

    scores = json.loads(text)

    # Map scores back to candidates
    score_map = {item["id"]: item["score"] for item in scores}

    for candidate in candidates:
        candidate.rerank_score = score_map.get(candidate.id, 0.0)

    # Sort by rerank score and return top_k
    ranked = sorted(candidates, key=lambda c: c.rerank_score or 0, reverse=True)
    return ranked[:top_k]

Batching for cost efficiency

For large candidate sets, processing all candidates in a single prompt can exceed context limits and cost more. Batch candidates and aggregate scores:

def rerank_batched(
    query: str,
    candidates: list[SearchResult],
    batch_size: int = 10,
    top_k: int = 5,
) -> list[SearchResult]:
    """
    Rerank large candidate sets by processing in batches.
    For each batch, get relevance scores, then combine and sort.
    """
    all_scored: list[SearchResult] = []

    for i in range(0, len(candidates), batch_size):
        batch = candidates[i:i + batch_size]
        scored_batch = rerank_with_llm(query, batch, top_k=len(batch))
        all_scored.extend(scored_batch)

    # Sort all results by rerank score
    all_scored.sort(key=lambda c: c.rerank_score or 0, reverse=True)
    return all_scored[:top_k]

Note that batch scores are not comparable across batches if the model uses relative scoring (e.g. "best in this group gets 0.9"). Fix this by using absolute scoring instructions ("how well does this document answer the query, on its own merits") rather than comparative ranking.

Integrating with vector search

A complete pipeline connecting vector retrieval to LLM reranking:

import numpy as np

class TwoStageSearch:
    def __init__(self, embedding_fn, vector_index, documents: dict[str, dict]):
        """
        embedding_fn: callable that takes text, returns embedding vector
        vector_index: FAISS or similar index for fast ANN search
        documents: dict mapping id → {content, metadata}
        """
        self.embed = embedding_fn
        self.index = vector_index
        self.documents = documents

    def search(
        self,
        query: str,
        initial_k: int = 30,    # How many to retrieve in stage 1
        final_k: int = 5,       # How many to return after reranking
        rerank: bool = True,
    ) -> list[SearchResult]:
        # Stage 1: Fast bi-encoder retrieval
        query_vec = np.array(self.embed(query), dtype="float32").reshape(1, -1)
        distances, indices = self.index.search(query_vec, initial_k)

        candidates = []
        for score, idx in zip(distances[0], indices[0]):
            if idx == -1:
                continue
            doc_id = str(idx)
            doc = self.documents.get(doc_id, {})
            candidates.append(SearchResult(
                id=doc_id,
                content=doc.get("content", ""),
                metadata=doc.get("metadata", {}),
                initial_score=float(score),
            ))

        if not rerank:
            return candidates[:final_k]

        # Stage 2: LLM reranking
        return rerank_batched(query, candidates, top_k=final_k)

Caching reranked results

Reranking is the expensive step. Cache aggressively:

import hashlib
from functools import lru_cache

def cache_key(query: str, candidate_ids: list[str]) -> str:
    """Deterministic cache key from query + candidate set."""
    content = query + "|" + ",".join(sorted(candidate_ids))
    return hashlib.sha256(content.encode()).hexdigest()

class CachedReranker:
    def __init__(self, ttl_seconds: int = 3600):
        self._cache: dict[str, tuple[list[SearchResult], float]] = {}
        self.ttl = ttl_seconds

    def rerank(
        self,
        query: str,
        candidates: list[SearchResult],
        top_k: int = 5,
    ) -> list[SearchResult]:
        import time

        key = cache_key(query, [c.id for c in candidates])
        cached, cached_at = self._cache.get(key, (None, 0))

        if cached and (time.time() - cached_at) < self.ttl:
            return cached[:top_k]

        results = rerank_with_llm(query, candidates, top_k=top_k)
        self._cache[key] = (results, time.time())
        return results

Measuring reranking improvement

To justify the latency and cost overhead, measure the precision improvement. The standard metric for this is nDCG@k (normalised discounted cumulative gain at k positions):

def ndcg_at_k(relevant_ids: set[str], results: list[SearchResult], k: int) -> float:
    """
    Measure ranking quality. relevant_ids is the ground truth set of
    relevant document IDs for this query.
    """
    dcg = 0.0
    for i, result in enumerate(results[:k]):
        if result.id in relevant_ids:
            dcg += 1.0 / (i + 1)  # log2(i+2) for standard DCG; simplified here

    ideal_dcg = sum(1.0 / (i + 1) for i in range(min(len(relevant_ids), k)))
    return dcg / ideal_dcg if ideal_dcg > 0 else 0.0

In practice, reranking typically improves nDCG@5 by 15–30% over bi-encoder retrieval alone, with the largest gains on queries that require understanding relationships between query terms rather than simple keyword matching. The latency cost is 200–800ms for a batch of 20–30 candidates — acceptable for most search applications where the user is waiting for results anyway.

Prompt engineering for better relevance judgments

The quality of LLM reranking depends heavily on how you frame the scoring task. A generic "rate this document" prompt underperforms a prompt that is explicit about what the query is asking for. Three techniques make a material difference:

Decompose the query before scoring. Before sending candidates to the reranker, extract the key requirements from the query in a separate call. Pass these requirements explicitly into each scoring prompt so the model evaluates against the right criteria:

def decompose_query(query: str) -> list[str]:
    """Extract the core information needs from a query."""
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"""Break this search query into its core information requirements.
List each distinct requirement on its own line.

QUERY: {query}

Requirements (one per line):"""
        }]
    )
    text = response.content[0].text.strip()
    return [line.strip("- ").strip() for line in text.split("\n") if line.strip()]


def rerank_with_requirements(
    query: str,
    candidates: list[SearchResult],
    top_k: int = 5,
) -> list[SearchResult]:
    requirements = decompose_query(query)
    requirements_text = "\n".join(f"- {r}" for r in requirements)

    candidates_text = "\n\n".join(
        f"[{i+1}] ID: {c.id}\n{c.content[:400]}"
        for i, c in enumerate(candidates)
    )

    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"""Score each document for relevance to this query.

ORIGINAL QUERY: {query}

WHAT THE QUERY IS LOOKING FOR:
{requirements_text}

DOCUMENTS:
{candidates_text}

Score each document 0.0–1.0 based on how many of the requirements it fulfills.
A score of 1.0 means it addresses every requirement fully.
Return JSON: [{{"id": str, "score": float, "fulfilled_requirements": list[str]}}]"""
        }]
    )

    text = response.content[0].text
    if "```" in text:
        text = text.split("```")[1].replace("json", "").strip()

    scores = json.loads(text)
    score_map = {item["id"]: item["score"] for item in scores}
    for c in candidates:
        c.rerank_score = score_map.get(c.id, 0.0)

    return sorted(candidates, key=lambda c: c.rerank_score or 0, reverse=True)[:top_k]

Use the reason field for debugging. The reason string in each scored result is not just documentation — it is a debugging tool. When precision is lower than expected on a test query, read the reasons the model assigned to low-ranked documents. They frequently reveal prompt ambiguities: the model is scoring against a different interpretation of the query than you intended.

Score against document sections, not full documents. Long documents dilute relevance signals. If your documents exceed 1,000 words, split them into sections and score sections independently. The highest-scoring section determines the document's overall rerank score, and you can surface the winning section as a snippet:

def score_document_sections(
    query: str,
    document: SearchResult,
    section_chars: int = 600,
) -> tuple[float, str]:
    """
    Score a long document by its most relevant section.
    Returns (best_score, best_section_text).
    """
    sections = [
        document.content[i:i + section_chars]
        for i in range(0, len(document.content), section_chars - 100)
    ]

    best_score = 0.0
    best_section = sections[0] if sections else ""

    # Score sections in a single call to save round trips
    sections_text = "\n\n".join(
        f"[section-{i+1}]\n{s}" for i, s in enumerate(sections[:6])  # Cap at 6 sections
    )
    response = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"""QUERY: {query}

{sections_text}

Score each section 0.0–1.0 for relevance to the query.
Return JSON: [{{"section": "section-N", "score": float}}]"""
        }]
    )
    text = response.content[0].text
    if "```" in text:
        text = text.split("```")[1].replace("json", "").strip()

    section_scores = json.loads(text)
    for item in section_scores:
        if item["score"] > best_score:
            best_score = item["score"]
            idx = int(item["section"].split("-")[1]) - 1
            best_section = sections[idx] if idx < len(sections) else best_section

    return best_score, best_section

This section-level scoring approach also enables passage-level result snippets, which give users better context for why a result was surfaced — a significant UX improvement over showing the document's beginning regardless of where the relevant content sits.

Latency budgeting and when to skip reranking

Reranking adds latency that may not always be acceptable. Design your system with a latency budget and a fallback:

• Under 100ms required: skip reranking, return bi-encoder results directly
• 100–500ms budget: rerank the top 10 candidates in a single batch
• 500ms+ budget: rerank 20–30 candidates in batches of 10

You can also skip reranking when the top bi-encoder result has a very high similarity score (above 0.95 cosine similarity, for example), since that usually indicates a near-exact match that reranking is unlikely to displace. Build the logic as a configurable bypass so you can tune the threshold on real traffic data.

The two-stage pattern with LLM reranking is now the standard approach for applications where search precision directly affects user outcomes — customer support, legal document retrieval, code search, and knowledge base Q&A all benefit substantially from the additional precision, and the latency cost is typically invisible behind the time the user spends reading results.

Hybrid retrieval before reranking

Vector search retrieves semantically similar documents well but struggles with exact keyword matches — searches for product codes, names, or specific technical terms. BM25 keyword search is the opposite: excellent for exact terms, poor at semantic understanding. Combining both before reranking gets the best of both worlds:

from rank_bm25 import BM25Okapi
import re

class HybridRetriever:
    """
    Combine BM25 keyword search and vector similarity for candidate retrieval.
    Merge the two ranked lists using Reciprocal Rank Fusion (RRF) before reranking.
    """

    def __init__(self, documents: dict[str, dict], embedding_fn, vector_index):
        self.documents = documents
        self.embed = embedding_fn
        self.index = vector_index

        # Build BM25 index over tokenised document text
        doc_ids = sorted(documents.keys())
        self._doc_ids = doc_ids
        tokenised = [
            re.findall(r"\w+", documents[d].get("content", "").lower())
            for d in doc_ids
        ]
        self.bm25 = BM25Okapi(tokenised)

    def _rrf_score(self, rank: int, k: int = 60) -> float:
        """Reciprocal Rank Fusion score. k=60 is a standard default."""
        return 1.0 / (k + rank + 1)

    def retrieve(
        self,
        query: str,
        initial_k: int = 30,
        bm25_weight: float = 0.4,
        vector_weight: float = 0.6,
    ) -> list[SearchResult]:
        import numpy as np

        # BM25 retrieval
        query_tokens = re.findall(r"\w+", query.lower())
        bm25_scores = self.bm25.get_scores(query_tokens)
        bm25_ranked = sorted(
            range(len(self._doc_ids)), key=lambda i: bm25_scores[i], reverse=True
        )[:initial_k]

        # Vector retrieval
        query_vec = np.array(self.embed(query), dtype="float32").reshape(1, -1)
        _, vector_indices = self.index.search(query_vec, initial_k)
        vector_ranked = [i for i in vector_indices[0] if i != -1]

        # RRF fusion
        rrf: dict[int, float] = {}
        for rank, idx in enumerate(bm25_ranked):
            rrf[idx] = rrf.get(idx, 0) + bm25_weight * self._rrf_score(rank)
        for rank, idx in enumerate(vector_ranked):
            rrf[idx] = rrf.get(idx, 0) + vector_weight * self._rrf_score(rank)

        top_indices = sorted(rrf, key=lambda i: rrf[i], reverse=True)[:initial_k]

        candidates = []
        for idx in top_indices:
            doc_id = self._doc_ids[idx]
            doc = self.documents[doc_id]
            candidates.append(SearchResult(
                id=doc_id,
                content=doc.get("content", ""),
                metadata=doc.get("metadata", {}),
                initial_score=rrf[idx],
            ))

        return candidates


class HybridTwoStageSearch:
    def __init__(self, documents: dict, embedding_fn, vector_index):
        self.retriever = HybridRetriever(documents, embedding_fn, vector_index)
        self.reranker = CachedReranker(ttl_seconds=1800)

    def search(self, query: str, initial_k: int = 40, final_k: int = 5) -> list[SearchResult]:
        candidates = self.retriever.retrieve(query, initial_k=initial_k)
        return self.reranker.rerank(query, candidates, top_k=final_k)

Reciprocal Rank Fusion is preferred over a simple score average because it is robust to score scale differences between BM25 and cosine similarity — they live on incomparable scales, but their rank positions are directly comparable.

A/B testing reranking in production

Before committing to reranking for all traffic, validate the improvement on your users' actual queries using a controlled experiment. Route a fraction of traffic to the reranked path and measure click-through rate on the top result, which is the most reliable proxy for perceived relevance:

import random
from dataclasses import dataclass
from datetime import datetime

@dataclass
class SearchEvent:
    query: str
    variant: str           # "control" (no reranking) or "treatment" (reranking)
    result_ids: list[str]
    clicked_id: str | None
    timestamp: datetime

def search_with_experiment(
    searcher: TwoStageSearch,
    query: str,
    experiment_fraction: float = 0.2,
    event_logger=None,
) -> list[SearchResult]:
    """
    Run search with optional reranking based on experiment assignment.
    Log events for click-through rate analysis.
    """
    in_treatment = random.random() < experiment_fraction
    results = searcher.search(query, rerank=in_treatment)

    if event_logger:
        event_logger.log(SearchEvent(
            query=query,
            variant="treatment" if in_treatment else "control",
            result_ids=[r.id for r in results],
            clicked_id=None,  # Filled in later when the click is recorded
            timestamp=datetime.utcnow(),
        ))

    return results


def analyse_experiment(events: list[SearchEvent]) -> dict:
    """
    Compute CTR@1 (click-through rate on the top result) per variant.
    """
    by_variant: dict[str, list[SearchEvent]] = {}
    for e in events:
        by_variant.setdefault(e.variant, []).append(e)

    summary = {}
    for variant, variant_events in by_variant.items():
        clicked_top = sum(
            1 for e in variant_events
            if e.clicked_id and e.result_ids and e.clicked_id == e.result_ids[0]
        )
        summary[variant] = {
            "total_queries": len(variant_events),
            "ctr_at_1": clicked_top / len(variant_events) if variant_events else 0,
        }

    return summary

Run the experiment for at least one week to capture variation across query types and user segments. A consistent CTR@1 improvement of 5% or more in the treatment group is a strong signal that the reranking overhead is justified for that traffic slice. Scale up the treatment fraction once you have statistical confidence.