Beyond Embeddings: Production Reranking with Vertex AI Ranking and Gemini-as-Judge

Implementation 1: Reranking with Vertex AI Ranking API

Google Cloud's Discovery Engine ships a standalone Ranking API you can use as a drop-in reranker. Send up to 200 documents per call, get back semantic relevance scores, sorted top-N. Here's the minimum integration using the google-cloud-discoveryengine client:

# rerank_with_vertex.py
# Purpose: rerank embedding-retrieved candidates with Vertex AI Ranking API
# Returns: top_k documents with rerank_score attached
import os
from google.cloud import discoveryengine_v1
 
PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
LOCATION = "global"
RANKING_CONFIG = "default_ranking_config"
 
def rerank(query: str, candidates: list[dict], top_k: int = 10) -> list[dict]:
    """
    candidates: [{"id": "doc-1", "title": "...", "content": "..."}, ...]
    """
    client = discoveryengine_v1.RankServiceClient()
    parent = (
        f"projects/{PROJECT_ID}/locations/{LOCATION}/"
        f"rankingConfigs/{RANKING_CONFIG}"
    )
 
    records = [
        discoveryengine_v1.RankingRecord(
            id=c["id"],
            title=c.get("title", ""),
            content=c["content"][:4000],  # respect per-record size limits
        )
        for c in candidates
    ]
 
    request = discoveryengine_v1.RankRequest(
        ranking_config=parent,
        model="semantic-ranker-default@latest",
        top_n=top_k,
        query=query,
        records=records,
    )
    try:
        response = client.rank(request=request)
    except Exception as e:
        # If the reranker fails, fall back to the original ordering.
        print(f"[rerank] fallback to embeddings order: {e}")
        return candidates[:top_k]
 
    by_id = {c["id"]: c for c in candidates}
    ranked = []
    for r in response.records:
        doc = by_id[r.id]
        doc["rerank_score"] = r.score
        ranked.append(doc)
    return ranked

The try/except block matters. The reranker is a precision booster, not a hard dependency. If it goes down, you want your search to keep returning results in the original embedding order rather than failing the request entirely. Building Stage 2 as optional polish over a working Stage 1 is what makes the pipeline robust.

A typical caller looks like this:

# pipeline.py
from rerank_with_vertex import rerank
from your_embeddings_search import search_with_gemini_embeddings
 
def retrieve(query: str) -> list[dict]:
    # Stage 1: pull 50 candidates for recall
    candidates = search_with_gemini_embeddings(query, top_k=50)
    # Stage 2: shrink to the precise top 10
    top10 = rerank(query, candidates, top_k=10)
    return top10

What I like about Vertex AI Ranking is that you don't have to train, host, or evaluate a custom model. Pricing runs a few cents per 1,000 records, latency stays around 100ms, and it handles dozens of languages out of the box. Where it falls short is domain-specific phrasing — medical, legal, or company-internal jargon that the off-the-shelf model wasn't trained heavily on. When that's your situation, the Gemini-as-judge route below adapts better.

Implementation 2: Gemini-as-judge as a reranker

You can also build a reranker by asking Gemini to score candidates directly. The idea is simple: hand it the query and the candidate documents, ask for a relevance score between 0 and 1 for each, sort by score. There are two flavors:

• Pointwise: score each candidate independently. Cheaper and parallelizable.
• Pairwise: compare two candidates at a time, run a tournament. More accurate but O(N log N) LLM calls — usually too expensive for production.

For real systems, batched pointwise scoring in a single Gemini call is the sweet spot. You send all 20 candidates in one request and get back a structured JSON array of scores.

# rerank_with_gemini.py
# Purpose: use Gemini 2.5 Flash as a pointwise reranker, batched in one call
# Returns: top_k documents with rerank_score attached
import os
import google.generativeai as genai
from pydantic import BaseModel
 
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
 
RERANK_PROMPT = """You are a search relevance grader.
Score how directly each document answers the query, on a scale from 0.0 to 1.0.
 
Scoring guide:
- 1.0: directly and clearly answers the query
- 0.7: partial answer or relevant context
- 0.4: tangentially related but not an answer
- 0.0: irrelevant
 
Query:
{query}
 
Documents:
{documents}
 
Respond with JSON in the form `{{"scores": [{{"id": "doc-1", "score": 0.92, "reason": "short reason"}}]}}`.
"""
 
class RankItem(BaseModel):
    id: str
    score: float
    reason: str
 
class RankResult(BaseModel):
    scores: list[RankItem]
 
def rerank_with_gemini(
    query: str, candidates: list[dict], top_k: int = 10
) -> list[dict]:
    docs_text = "\n\n".join(
        f"[{c['id']}]\n{c['content'][:1500]}"  # cap each doc to keep tokens bounded
        for c in candidates
    )
    prompt = RERANK_PROMPT.format(query=query, documents=docs_text)
 
    model = genai.GenerativeModel(
        "gemini-2.5-flash",
        generation_config={
            "response_mime_type": "application/json",
            "response_schema": RankResult,
            "temperature": 0.0,  # deterministic scoring
        },
    )
 
    try:
        response = model.generate_content(prompt)
        result = RankResult.model_validate_json(response.text)
    except Exception as e:
        print(f"[rerank_gemini] fallback: {e}")
        return candidates[:top_k]
 
    score_by_id = {item.id: item.score for item in result.scores}
    ranked = sorted(
        candidates,
        key=lambda c: score_by_id.get(c["id"], 0.0),
        reverse=True,
    )[:top_k]
 
    for c in ranked:
        c["rerank_score"] = score_by_id.get(c["id"], 0.0)
    return ranked

Three things in this code earn their keep:

First, the response_schema with a Pydantic model forces Gemini to return parseable JSON instead of free-form text. Combined with temperature=0.0, you get reproducible scores — which matters because if every run produces a different ordering, you can't measure whether the reranker is helping.

Second, each document is truncated to 1,500 characters. Pushing 50 untruncated documents into the prompt blows past 100k tokens fast. Reranker input doesn't need full documents — it needs the relevant excerpt. Stage 1 should already be returning chunks at this granularity.

Third, the model is gemini-2.5-flash, not Pro. Reranking has a tight latency budget (you want under 500ms) and Flash gets you there. In my benchmarks, Flash reranks 20 candidates in around 400ms, while Pro takes 1.2 seconds for the same payload.

Implementation 3: hybrid retrieval plus reranker

In serious production setups, Stage 1 itself goes hybrid. Run BM25 keyword search and embeddings search in parallel, merge their candidates, then hand the union to the reranker. This is the configuration I use in production.

# hybrid_pipeline.py
# Purpose: BM25 + embeddings union, then Vertex AI Ranking on the merged set
from rank_bm25 import BM25Okapi
from your_embeddings_search import search_with_gemini_embeddings
from rerank_with_vertex import rerank
 
def hybrid_retrieve(
    query: str, corpus: list[dict], bm25: BM25Okapi, top_k: int = 10
) -> list[dict]:
    # 1. BM25 top 30
    bm25_scores = bm25.get_scores(query.split())
    bm25_top = sorted(
        zip(corpus, bm25_scores), key=lambda x: x[1], reverse=True
    )[:30]
    bm25_ids = {doc["id"] for doc, _ in bm25_top}
 
    # 2. Embeddings top 30
    emb_top = search_with_gemini_embeddings(query, top_k=30)
    emb_ids = {doc["id"] for doc in emb_top}
 
    # 3. Union (deduped, up to 60)
    candidates_map = {doc["id"]: doc for doc, _ in bm25_top}
    for doc in emb_top:
        candidates_map.setdefault(doc["id"], doc)
    candidates = list(candidates_map.values())
 
    # 4. Boost candidates that appeared in both retrievers (lightweight RRF)
    overlap = bm25_ids & emb_ids
    for c in candidates:
        c["recall_signal"] = 2 if c["id"] in overlap else 1
 
    # 5. Reranker for final precision
    return rerank(query, candidates, top_k=top_k)

The recall_signal is a poor man's Reciprocal Rank Fusion. A more rigorous version would compute 1/(k + rank_bm25) + 1/(k + rank_emb), but since the reranker makes the final call, a binary "appeared in both" signal is enough to nudge the right candidates upward.

In my own data, embedding-only top-1 accuracy on a 100-query eval set was 52%. Adding hybrid retrieval and Vertex Ranking pushed it to 71%. Swapping in Gemini Flash as the reranker (with the same hybrid Stage 1) reached 78%. The cases where embeddings put the answer at rank 4 mostly migrated to ranks 1–3 after reranking — exactly the failure mode I wanted to fix.

Measuring the reranker — NDCG and MRR in your CI

Subjective improvement is meaningless if you can't track it. Once you ship a reranker, you need numbers. Two metrics cover the ground:

• MRR (Mean Reciprocal Rank): average of 1/rank of the correct document across an eval set. Rank 1 is 1.0, rank 2 is 0.5, missing is 0.0.
• NDCG@5: discounted gain over the top 5, weighted by position. This is the metric most sensitive to reranker tuning.

Build an eval set of 100–300 query/correct-document pairs from real queries you've seen, then bake the metrics into CI:

# eval_rerank.py
# Purpose: compute MRR and NDCG@5 against an eval set
import math
from typing import Callable
 
def mrr(ranked_lists: list[list[str]], golds: list[str]) -> float:
    total = 0.0
    for ranked, gold in zip(ranked_lists, golds):
        try:
            rank = ranked.index(gold) + 1
            total += 1.0 / rank
        except ValueError:
            total += 0.0
    return total / len(ranked_lists)
 
def ndcg_at_k(ranked_lists: list[list[str]], golds: list[str], k: int = 5) -> float:
    total = 0.0
    for ranked, gold in zip(ranked_lists, golds):
        ideal = 1.0  # single correct answer assumption
        dcg = 0.0
        for i, doc_id in enumerate(ranked[:k]):
            if doc_id == gold:
                dcg = 1.0 / math.log2(i + 2)
                break
        total += dcg / ideal
    return total / len(ranked_lists)
 
def evaluate(retrieve_fn: Callable, eval_set: list[dict]) -> dict:
    queries = [item["query"] for item in eval_set]
    golds = [item["gold_id"] for item in eval_set]
    ranked_lists = [
        [doc["id"] for doc in retrieve_fn(q)] for q in queries
    ]
    return {
        "mrr": mrr(ranked_lists, golds),
        "ndcg@5": ndcg_at_k(ranked_lists, golds, k=5),
        "n_queries": len(queries),
    }

I run this in GitHub Actions nightly. If MRR drops more than 5% versus the previous day, a Slack alert fires. That's how I caught a Gemini Flash model update that quietly shifted the score distribution last quarter — without the eval harness, I'd have missed it for weeks.

Common mistakes and gotchas

The traps I hit, in the order I hit them:

Trap 1: starving Stage 1. If you only feed 5 candidates into the reranker, you've capped the ceiling at whatever the embeddings could find in their top 5. Stage 1 is the recall layer — it should be wide. Top 50 to 100 is where I run mine.

Trap 2: cross-language mismatch. Querying English documents with Japanese queries (or vice versa) sometimes works fine on Vertex AI Ranking's default model and sometimes doesn't. When relevance looks off, switch to a cross-lingual ranker variant. Gemini-as-judge sidesteps this entirely because it's natively multilingual.

Trap 3: trusting absolute scores. "Drop anything below 0.5" sounds reasonable but doesn't survive contact with reality — the score distribution shifts per query and per model. Trust relative ordering, and if you want to filter, gate on the gap between top score and second score, not on the raw value.

Trap 4: shipping without an eval set. Ship-then-evaluate is the path to a system that nobody can fix later. The day you suspect "things have gotten worse" without an eval set is the day you start guessing. Three to four hours of curating 100 queries pays back forever.

Trap 5: expecting the reranker to fix bad chunks. A reranker can only reorder what it's given. If your chunks are 5,000 characters of mixed topics, no reranker will save you. Chunking and reranking belong on the same review loop.

How chunking affects reranker quality

Speaking of chunks — when teams report a reranker "isn't helping," chunking is usually the culprit. The reranker matches queries against candidate chunks, so if the chunks aren't sized to "contain a single answer," reranking can't elevate the right one to position one.

The chunking rules I follow:

First, one chunk = one claim or one procedure. Markdown H2/H3 splits give you this naturally; if a section is too long, split by paragraph. I aim for 300–800 characters per chunk. Larger chunks mix the answer paragraph with unrelated paragraphs in the same embedding, dragging precision down.

Second, 50–100 character overlap between adjacent chunks. Queries whose subject is named in the previous chunk ("This feature works when...") need that linking context, or the reranker reads them as orphan fragments and scores them low.

Third, prepend headings as metadata. Send the reranker not just the body text but [Support Policy > Refunds] Within 30 days of delivery... style structured context. Vertex AI Ranking's RankingRecord.title field exists for exactly this. Stable scoring depends on the reranker knowing where each chunk sits in the document.

# chunker.py
# Purpose: split markdown into reranker-friendly chunks with heading metadata
def split_with_headings(markdown: str, max_chars: int = 600) -> list[dict]:
    chunks: list[dict] = []
    current_h1, current_h2, current_h3 = "", "", ""
    buffer: list[str] = []
    buffer_len = 0
 
    def flush():
        nonlocal buffer, buffer_len
        if buffer_len > 0:
            chunks.append({
                "title": " > ".join(filter(None, [current_h1, current_h2, current_h3])),
                "content": "\n".join(buffer).strip(),
            })
            buffer = []
            buffer_len = 0
 
    for line in markdown.split("\n"):
        if line.startswith("# "):
            flush(); current_h1 = line[2:].strip(); current_h2 = current_h3 = ""
            continue
        if line.startswith("## "):
            flush(); current_h2 = line[3:].strip(); current_h3 = ""
            continue
        if line.startswith("### "):
            flush(); current_h3 = line[4:].strip()
            continue
        buffer.append(line)
        buffer_len += len(line)
        if buffer_len >= max_chars:
            flush()
    flush()
    return chunks

When I feed chunks produced by this splitter (with the title field populated) into Vertex AI Ranking, my eval scores move 5–10% upward versus naive fixed-window splitting. Chunking and reranking are the same investment, just at different layers.

Caching reranker results — the cost layer

Rerankers are pricier than retrieval per call, so running them on every request leaves money on the table. A two-layer cache wins back most of it.

Layer 1: normalized-query exact-match cache. Trim, lowercase, full-width/half-width normalize the query, hash it, and store the reranked top-K IDs in Redis with a 30–60 minute TTL. This catches repeated questions from the same user and high-frequency queries like "office hours" across users.

Layer 2: pair-level (query × doc_id) score cache. Rerank scores carry over even when Stage 1's candidates change, so caching md5(query + doc_id) → score for 24 hours means tomorrow's reranker call only scores the new candidates instead of the full set.

# rerank_cache.py
# Purpose: pair-level score cache that cuts Gemini reranker calls
import hashlib
import redis
 
r = redis.Redis(host="localhost", port=6379, decode_responses=True)
SCORE_TTL = 24 * 60 * 60  # 24 hours
 
def _key(query: str, doc_id: str) -> str:
    h = hashlib.md5(f"{query}::{doc_id}".encode()).hexdigest()
    return f"rerank:score:{h}"
 
def get_cached_scores(query: str, doc_ids: list[str]) -> dict[str, float]:
    keys = [_key(query, d) for d in doc_ids]
    values = r.mget(keys)
    return {d: float(v) for d, v in zip(doc_ids, values) if v is not None}
 
def save_scores(query: str, scored: list[dict]) -> None:
    pipe = r.pipeline()
    for item in scored:
        pipe.set(_key(query, item["id"]), str(item["rerank_score"]), ex=SCORE_TTL)
    pipe.execute()
 
def rerank_with_cache(query: str, candidates: list[dict], rerank_fn) -> list[dict]:
    cached = get_cached_scores(query, [c["id"] for c in candidates])
    uncached = [c for c in candidates if c["id"] not in cached]
 
    if uncached:
        new_scored = rerank_fn(query, uncached, top_k=len(uncached))
        save_scores(query, new_scored)
        for c in new_scored:
            cached[c["id"]] = c["rerank_score"]
 
    for c in candidates:
        c["rerank_score"] = cached.get(c["id"], 0.0)
    return sorted(candidates, key=lambda x: x["rerank_score"], reverse=True)

This cache layer cuts reranker invocations by roughly 60% on my workload, which translates directly into Gemini Flash bill savings. The one operational caveat: when you update a document, you have to invalidate every cached score that mentions its ID. The cleanest way is to embed a document-version number into the cache key (rerank:score:v{n}:{hash}), so bumping the version on document edit invalidates the whole tier in one stroke.

What to do tomorrow

Rerankers are the highest-leverage upgrade you can apply to a "search is almost there" RAG system. If your top-1 hit rate is sitting in the 30–50% range with embeddings alone, dropping in Vertex AI Ranking will commonly push it past 70% within an afternoon.

The single concrete next step: build a 100-query eval set and measure your current MRR and NDCG@5. Once you have those numbers, deciding between Vertex AI Ranking and a Gemini reranker becomes an empirical question instead of a religious one — pick whichever scores higher on your data and ship it.

I spent weeks early on swapping embedding models, tuning dimensions, and tweaking similarity thresholds, convinced the answer was upstream. Looking back, plugging in a reranker on day one would have gotten me to my target accuracy a month sooner. If this saves anyone the same detour, the article paid for itself.