Advanced RAG: Query Expansion, Reranking, and Fusion

Basic RAG retrieves the top-k most similar chunks to a query and feeds them to the LLM. This works for straightforward questions but fails when queries are ambiguous, use different vocabulary than the source documents, or require information scattered across multiple sections. Advanced retrieval techniques address these failures by transforming queries, combining multiple search strategies, and reranking results for precision.

Why Basic Retrieval Falls Short

Consider a user asking “What are the performance implications of using microservices?” A basic vector search matches this against chunks about microservices. But the best answer might be in a chunk titled “Latency overhead in distributed architectures” that never mentions the word “microservices.” The vocabulary mismatch means the most relevant chunk scores lower than a less relevant one that happens to contain the query terms.

Advanced retrieval fixes this through three approaches: making queries match documents better (query transformation), searching in multiple ways and combining results (fusion), and reordering results by true relevance (reranking).

Query Expansion: Multi-Query Retrieval

The simplest improvement is to generate multiple versions of the user’s query, search with each one, and combine the results. Different phrasings catch different relevant documents.

from openai import OpenAI
import json

client = OpenAI()

def generate_multi_queries(original_query: str, num_queries: int = 3) -> list[str]:
    """Generate multiple search queries from a single user question."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a search query generator. Given a user question, generate "
                    f"{num_queries} different search queries that would help find relevant information. "
                    "Each query should approach the topic from a different angle or use different terminology. "
                    "Return as a JSON object with a 'queries' array."
                )
            },
            {"role": "user", "content": original_query}
        ],
        response_format={"type": "json_object"},
        temperature=0.7
    )
    
    result = json.loads(response.choices[0].message.content)
    queries = result.get("queries", [original_query])
    
    # Always include the original query
    return [original_query] + queries

# Example
original = "What are the performance implications of using microservices?"
expanded = generate_multi_queries(original)
for i, q in enumerate(expanded):
    print(f"  {i}: {q}")
# Output:
# 0: What are the performance implications of using microservices?
# 1: Latency and throughput overhead in microservice architectures
# 2: How does breaking a monolith into services affect system performance?
# 3: Network latency and resource consumption in distributed microservices

Now search with each query and merge the results:

def multi_query_retrieve(
    query: str,
    vector_store,
    top_k: int = 5,
    num_queries: int = 3
) -> list[dict]:
    """Retrieve using multiple query variations."""
    queries = generate_multi_queries(query, num_queries)
    
    all_results = {}  # Use dict to deduplicate by chunk ID
    
    for q in queries:
        results = vector_store.search(query=q, top_k=top_k)
        for r in results:
            chunk_id = r["id"]
            if chunk_id not in all_results or r["score"] > all_results[chunk_id]["score"]:
                all_results[chunk_id] = r
                all_results[chunk_id]["matched_query"] = q
    
    # Sort by best score
    merged = sorted(all_results.values(), key=lambda x: x["score"], reverse=True)
    return merged[:top_k]

HyDE: Hypothetical Document Embeddings

HyDE takes query transformation further. Instead of searching with the question, you ask the LLM to generate a hypothetical answer, then search with that answer’s embedding. Since the hypothetical answer resembles a document more than a question does, it often matches better against the actual document embeddings.

def hyde_retrieve(
    query: str,
    vector_store,
    top_k: int = 5
) -> list[dict]:
    """Retrieve using Hypothetical Document Embeddings."""
    
    # Generate a hypothetical answer
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "Write a detailed paragraph that would be a perfect answer "
                    "to the following question. Write it as if it were an excerpt "
                    "from a technical document. Be specific and factual."
                )
            },
            {"role": "user", "content": query}
        ],
        temperature=0.5,
        max_tokens=300
    )
    
    hypothetical_doc = response.choices[0].message.content
    
    # Search using the hypothetical document instead of the query
    results = vector_store.search(
        query=hypothetical_doc,
        top_k=top_k
    )
    
    return results

# HyDE works best when:
# - User queries are short or vague
# - There is vocabulary mismatch between queries and documents
# - Documents are technical and use domain-specific terminology

Step-Back Prompting for Better Retrieval

Step-back prompting generates a more general version of the question, which often matches broader context documents that contain the specific answer.

def step_back_query(original_query: str) -> str:
    """Generate a broader, more general version of the query."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "Given a specific question, generate a more general 'step-back' question "
                    "that covers the broader topic. This helps find background information "
                    "needed to answer the specific question.\n\n"
                    "Example:\n"
                    "Specific: 'Why does my Python asyncio task get cancelled silently?'\n"
                    "Step-back: 'How does task cancellation and error handling work in Python asyncio?'"
                )
            },
            {"role": "user", "content": original_query}
        ],
        temperature=0.3,
        max_tokens=100
    )
    
    return response.choices[0].message.content

def step_back_retrieve(
    query: str,
    vector_store,
    top_k: int = 5
) -> list[dict]:
    """Retrieve using both the original and step-back queries."""
    general_query = step_back_query(query)
    
    # Search with both queries
    specific_results = vector_store.search(query=query, top_k=top_k)
    general_results = vector_store.search(query=general_query, top_k=top_k)
    
    # Combine with deduplication
    seen_ids = set()
    combined = []
    
    for r in specific_results + general_results:
        if r["id"] not in seen_ids:
            seen_ids.add(r["id"])
            combined.append(r)
    
    return combined[:top_k]

Cross-Encoder Reranking

Vector search finds candidates quickly but uses bi-encoder similarity, which compares query and document embeddings independently. A cross-encoder processes the query and document together, capturing interactions between them for much more accurate relevance scoring.

from sentence_transformers import CrossEncoder

class CrossEncoderReranker:
    """Rerank retrieval results using a cross-encoder model."""
    
    def __init__(self, model_name: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"):
        self.model = CrossEncoder(model_name)
    
    def rerank(
        self,
        query: str,
        documents: list[dict],
        top_k: int = 5
    ) -> list[dict]:
        """Rerank documents by cross-encoder relevance score."""
        if not documents:
            return []
        
        # Create query-document pairs
        pairs = [(query, doc["text"]) for doc in documents]
        
        # Score all pairs
        scores = self.model.predict(pairs)
        
        # Attach scores and sort
        for doc, score in zip(documents, scores):
            doc["rerank_score"] = float(score)
            doc["original_score"] = doc.get("score", 0)
        
        reranked = sorted(documents, key=lambda x: x["rerank_score"], reverse=True)
        return reranked[:top_k]

# Usage
reranker = CrossEncoderReranker()

# First pass: get candidates with fast vector search
candidates = vector_store.search(query="microservice performance", top_k=20)

# Second pass: rerank with cross-encoder
reranked = reranker.rerank("What are the performance implications of microservices?", candidates, top_k=5)

for r in reranked:
    print(f"  [{r['rerank_score']:.3f}] (was {r['original_score']:.3f}) {r['text'][:80]}...")

The two-stage pattern (fast retrieval then precise reranking) is the standard approach in production search systems. Retrieve 3-5x more candidates than you need, then let the reranker select the best ones.

LLM-Based Reranking

When you do not want to deploy a cross-encoder model, use an LLM to rerank. This is slower and more expensive but requires no additional infrastructure.

def llm_rerank(
    query: str,
    documents: list[dict],
    top_k: int = 5
) -> list[dict]:
    """Rerank using an LLM to assess relevance."""
    
    # Format documents for the LLM
    doc_list = "\n\n".join(
        f"[{i}] {doc['text'][:500]}"
        for i, doc in enumerate(documents)
    )
    
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are a relevance assessor. Given a query and a list of documents, "
                    "rank the documents by their relevance to answering the query. "
                    "Return a JSON object with 'ranking': a list of document indices "
                    "ordered from most to least relevant, and 'scores': a dict mapping "
                    "each index to a relevance score from 0.0 to 1.0."
                )
            },
            {
                "role": "user",
                "content": f"Query: {query}\n\nDocuments:\n{doc_list}"
            }
        ],
        response_format={"type": "json_object"},
        temperature=0
    )
    
    result = json.loads(response.choices[0].message.content)
    ranking = result.get("ranking", list(range(len(documents))))
    scores = result.get("scores", {})
    
    reranked = []
    for idx in ranking[:top_k]:
        if isinstance(idx, int) and idx < len(documents):
            doc = documents[idx].copy()
            doc["rerank_score"] = scores.get(str(idx), scores.get(idx, 0.5))
            reranked.append(doc)
    
    return reranked

Reciprocal Rank Fusion (RRF)

When you have results from multiple retrieval methods (vector search, keyword search, different query expansions), Reciprocal Rank Fusion combines them into a single ranking. RRF is simple, effective, and does not require score normalization.

def reciprocal_rank_fusion(
    result_lists: list[list[dict]],
    k: int = 60,
    top_n: int = 10
) -> list[dict]:
    """Combine multiple ranked lists using Reciprocal Rank Fusion.
    
    RRF score = sum(1 / (k + rank)) across all lists where the document appears.
    The constant k (default 60) controls how much weight is given to top-ranked items.
    """
    rrf_scores = {}  # doc_id -> cumulative RRF score
    doc_map = {}     # doc_id -> document data
    
    for result_list in result_lists:
        for rank, doc in enumerate(result_list, start=1):
            doc_id = doc["id"]
            rrf_score = 1.0 / (k + rank)
            
            if doc_id not in rrf_scores:
                rrf_scores[doc_id] = 0.0
                doc_map[doc_id] = doc
            
            rrf_scores[doc_id] += rrf_score
    
    # Sort by RRF score
    sorted_ids = sorted(rrf_scores.keys(), key=lambda x: rrf_scores[x], reverse=True)
    
    results = []
    for doc_id in sorted_ids[:top_n]:
        doc = doc_map[doc_id].copy()
        doc["rrf_score"] = rrf_scores[doc_id]
        results.append(doc)
    
    return results

# Example: fuse vector search + BM25 keyword search
vector_results = vector_store.search(query="microservice performance", top_k=10)
keyword_results = keyword_store.search(query="microservice performance", top_k=10)

fused = reciprocal_rank_fusion(
    [vector_results, keyword_results],
    top_n=5
)

for r in fused:
    print(f"  [RRF: {r['rrf_score']:.4f}] {r['text'][:80]}...")

RRF works particularly well when combining semantic (vector) and lexical (BM25) search because each method catches what the other misses.

Complete Advanced Retrieval Pipeline

Combining these techniques into a cohesive pipeline:

class AdvancedRetriever:
    """Advanced retrieval combining multiple strategies."""
    
    def __init__(self, vector_store, keyword_store=None):
        self.vector_store = vector_store
        self.keyword_store = keyword_store
        self.reranker = CrossEncoderReranker()
    
    def retrieve(
        self,
        query: str,
        top_k: int = 5,
        strategy: str = "full"
    ) -> list[dict]:
        """Retrieve with configurable strategy.
        
        Strategies:
        - "basic": Simple vector search
        - "multi_query": Query expansion + fusion
        - "hybrid": Vector + keyword + fusion
        - "full": Multi-query + hybrid + reranking
        """
        
        if strategy == "basic":
            return self.vector_store.search(query=query, top_k=top_k)
        
        result_lists = []
        
        if strategy in ("multi_query", "full"):
            # Generate query variations
            queries = generate_multi_queries(query, num_queries=3)
            for q in queries:
                results = self.vector_store.search(query=q, top_k=top_k * 2)
                result_lists.append(results)
        else:
            results = self.vector_store.search(query=query, top_k=top_k * 2)
            result_lists.append(results)
        
        if strategy in ("hybrid", "full") and self.keyword_store:
            keyword_results = self.keyword_store.search(query=query, top_k=top_k * 2)
            result_lists.append(keyword_results)
        
        # Fuse all result lists
        fused = reciprocal_rank_fusion(result_lists, top_n=top_k * 3)
        
        # Rerank if using full strategy
        if strategy == "full" and fused:
            return self.reranker.rerank(query, fused, top_k=top_k)
        
        return fused[:top_k]

Measuring Retrieval Quality

You need to measure whether these techniques actually improve your results. Track retrieval metrics independently from generation quality.

def evaluate_retrieval(
    queries: list[str],
    ground_truth: dict[str, list[str]],  # query -> list of relevant doc IDs
    retriever,
    top_k: int = 5
) -> dict:
    """Evaluate retrieval quality with standard IR metrics."""
    
    metrics = {
        "recall_at_k": [],
        "precision_at_k": [],
        "mrr": [],  # Mean Reciprocal Rank
        "ndcg": [],  # Normalized Discounted Cumulative Gain
    }
    
    for query in queries:
        relevant_ids = set(ground_truth.get(query, []))
        if not relevant_ids:
            continue
        
        results = retriever.retrieve(query, top_k=top_k)
        retrieved_ids = [r["id"] for r in results]
        
        # Recall@K: fraction of relevant docs found
        found = len(set(retrieved_ids) & relevant_ids)
        metrics["recall_at_k"].append(found / len(relevant_ids))
        
        # Precision@K: fraction of retrieved docs that are relevant
        metrics["precision_at_k"].append(found / len(retrieved_ids) if retrieved_ids else 0)
        
        # MRR: reciprocal of rank of first relevant result
        mrr = 0
        for rank, doc_id in enumerate(retrieved_ids, 1):
            if doc_id in relevant_ids:
                mrr = 1.0 / rank
                break
        metrics["mrr"].append(mrr)
    
    # Average all metrics
    return {
        metric: sum(values) / len(values) if values else 0
        for metric, values in metrics.items()
    }

# Compare strategies
for strategy in ["basic", "multi_query", "hybrid", "full"]:
    scores = evaluate_retrieval(
        test_queries, ground_truth, retriever, strategy=strategy
    )
    print(f"\n{strategy}:")
    for metric, value in scores.items():
        print(f"  {metric}: {value:.3f}")

When to Use Which Technique

Not every technique is appropriate for every situation:

Technique	When to Use	Cost
Multi-query	Ambiguous or broad queries	1 LLM call for expansion
HyDE	Vocabulary mismatch, short queries	1 LLM call + 1 embedding
Step-back	Complex questions needing background	1 LLM call + extra search
Cross-encoder reranking	Precision matters, latency budget allows	Model inference per candidate
LLM reranking	No cross-encoder infra, small candidate sets	1 LLM call
RRF	Multiple retrieval signals available	No extra cost (just fusion)
Hybrid (vector + BM25)	Known entity names, exact terms matter	Requires keyword index

Start with basic vector search. Add hybrid search if users search for specific names or codes. Add reranking when you notice that the best result is often in positions 3-10. Add query expansion when users ask vague questions. Measure each addition to verify it actually helps.

Wrapping Up

Advanced retrieval is about closing the gap between what the user asks and what the documents contain. Query expansion bridges vocabulary differences by searching with multiple phrasings. Hybrid search catches exact terms that vector search misses. Reranking promotes the truly relevant results above the merely similar ones. Reciprocal rank fusion combines these signals without needing to normalize scores. The key principle is the retrieve-then-rerank pattern: cast a wide net with fast, approximate methods, then use slower, more precise methods to pick the best results. Always measure with retrieval-specific metrics like recall and MRR, because better retrieval is the single most impactful improvement you can make to a RAG system.