Beyond Naive Vector Search
Basic RAG (Retrieval-Augmented Generation) follows a simple pattern: embed the query, find the top-k most similar vectors, and pass them to the LLM as context. But this approach has well-documented limitations: embedding similarity doesn't always correlate with relevance (especially for complex queries), the top-k results may include redundant or contradictory information, and multi-faceted queries (requiring information from different topics) are poorly served by a single embedding.
Advanced RAG techniques—re-ranking, query transformation, and multi-stage retrieval—address these limitations by adding intelligence between the initial retrieval and the LLM call. This article covers production-ready implementations of the most impactful advanced RAG techniques.
Phase 1: Re-ranking Strategies
Cross-Encoder Re-ranking
# rag/reranking.py — Multi-stage retrieval with cross-encoder re-ranking
from sentence_transformers import CrossEncoder, SentenceTransformer
import numpy as np
from typing import List, Dict, Tuple
from dataclasses import dataclass
@dataclass
class RetrievedDocument:
document_id: str
content: str
score: float
metadata: Dict
rerank_score: float = 0.0
class RerankedRetriever:
"""
Two-stage retrieval:
1. Bi-encoder (fast): Retrieve top-100 candidates using vector similarity
2. Cross-encoder (accurate): Re-rank candidates using query-document attention
"""
def __init__(self,
bi_encoder_model: str = "all-MiniLM-L6-v2",
cross_encoder_model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2",
top_k_candidates: int = 100,
top_k_final: int = 10):
self.bi_encoder = SentenceTransformer(bi_encoder_model)
self.cross_encoder = CrossEncoder(cross_encoder_model)
self.top_k_candidates = top_k_candidates
self.top_k_final = top_k_final
def retrieve(self, query: str,
vector_store, # Your vector DB client
filters: Dict = None) -> List[RetrievedDocument]:
"""
Retrieve and re-rank documents for a query.
Stage 1: Fast vector search (bi-encoder) — ~10ms for 100 results
Stage 2: Accurate re-ranking (cross-encoder) — ~50ms for 100 pairs
Total: ~60ms (vs ~10ms for naive retrieval)
"""
# Stage 1: Bi-encoder retrieval (fast, approximate)
query_embedding = self.bi_encoder.encode(query)
candidates = vector_store.search(
vector=query_embedding,
top_k=self.top_k_candidates,
filters=filters,
)
if not candidates:
return []
# Stage 2: Cross-encoder re-ranking (slow, accurate)
# Cross-encoder processes query + document together through attention
pairs = [(query, doc.content) for doc in candidates]
rerank_scores = self.cross_encoder.predict(pairs)
# Assign re-rank scores
for doc, score in zip(candidates, rerank_scores):
doc.rerank_score = float(score)
# Sort by re-rank score and return top-k
reranked = sorted(candidates, key=lambda d: -d.rerank_score)
return reranked[:self.top_k_final]
class RecencyBoostedReranker:
"""
Re-ranker that combines semantic relevance with recency.
Useful for news, documentation, and time-sensitive content.
"""
def __init__(self, cross_encoder: CrossEncoder,
recency_weight: float = 0.2,
recency_halflife_days: float = 30):
self.cross_encoder = cross_encoder
self.recency_weight = recency_weight
self.halflife = recency_halflife_days
def rerank(self, query: str, candidates: List[RetrievedDocument]) -> List[RetrievedDocument]:
"""Re-rank with recency boost."""
from datetime import datetime, timezone
pairs = [(query, doc.content) for doc in candidates]
semantic_scores = self.cross_encoder.predict(pairs)
now = datetime.now(timezone.utc)
for doc, sem_score in zip(candidates, semantic_scores):
# Compute recency score (exponential decay)
doc_date = datetime.fromisoformat(doc.metadata.get("published_at", now.isoformat()))
days_old = (now - doc_date).days
recency_score = 0.5 ** (days_old / self.halflife)
# Combined score
doc.rerank_score = (
(1 - self.recency_weight) * sem_score +
self.recency_weight * recency_score
)
return sorted(candidates, key=lambda d: -d.rerank_score)Phase 2: Query Transformation
Multi-Query Expansion
# rag/query_transformation.py — Transform queries for better retrieval
from openai import OpenAI
from typing import List
import asyncio
class QueryTransformer:
"""
Transform user queries into optimized retrieval queries.
Techniques: expansion, decomposition, HyDE, step-back prompting.
"""
def __init__(self, model: str = "gpt-4o-mini"):
self.client = OpenAI()
self.model = model
def multi_query_expansion(self, query: str, n_variants: int = 3) -> List[str]:
"""
Generate multiple query variants to improve recall.
Different phrasings may match different relevant documents.
"""
prompt = f"""Generate {n_variants} alternative search queries for the following question.
Each variant should approach the question from a different angle or use different terminology.
Original question: {query}
Generate {n_variants} variants, one per line:"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0.7,
)
variants = response.choices[0].message.content.strip().split("\n")
variants = [v.strip().lstrip("0123456789. ") for v in variants if v.strip()]
return [query] + variants[:n_variants]
def query_decomposition(self, query: str) -> List[str]:
"""
Decompose complex queries into simpler sub-queries.
Each sub-query retrieves a specific piece of information.
"""
prompt = f"""Break down the following complex question into 2-4 simpler sub-questions.
Each sub-question should be independently searchable.
Complex question: {query}
Sub-questions (one per line):"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
)
sub_queries = response.choices[0].message.content.strip().split("\n")
return [q.strip().lstrip("0123456789. ") for q in sub_queries if q.strip()]
def hyde(self, query: str) -> str:
"""
Hypothetical Document Embeddings (HyDE).
Generate a hypothetical answer, then use IT as the search query.
The hypothetical answer's embedding is often closer to relevant documents
than the question's embedding.
"""
prompt = f"""Answer the following question concisely in 1-2 sentences.
If you don't know the answer, make your best educated guess.
Question: {query}
Answer:"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
max_tokens=150,
)
hypothetical_answer = response.choices[0].message.content.strip()
return hypothetical_answer
def step_back_prompting(self, query: str) -> str:
"""
Generate a broader, more abstract version of the query.
Useful when the specific query is too narrow to find relevant context.
"""
prompt = f"""Rewrite the following specific question as a broader, more general question
that would help answer the original question.
Specific question: {query}
Broader question:"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0.0,
)
return response.choices[0].message.content.strip()
class MultiQueryRetriever:
"""
Retrieve using multiple query variants and merge results.
Combines multi-query expansion with Reciprocal Rank Fusion (RRF).
"""
def __init__(self, transformer: QueryTransformer, retriever):
self.transformer = transformer
self.retriever = retriever
def retrieve(self, query: str, top_k: int = 10) -> List[RetrievedDocument]:
"""Retrieve using multiple query variants and fuse results."""
# Generate query variants
queries = self.transformer.multi_query_expansion(query, n_variants=3)
# Retrieve for each variant
all_results = []
for q in queries:
results = self.retriever.retrieve(q, top_k=20)
all_results.append(results)
# Fuse results using Reciprocal Rank Fusion
fused = self._reciprocal_rank_fusion(all_results, k=60)
return fused[:top_k]
def _reciprocal_rank_fusion(self, result_lists: List[List[RetrievedDocument]],
k: int = 60) -> List[RetrievedDocument]:
"""
Reciprocal Rank Fusion: score = Σ 1/(k + rank_i)
Documents that appear in multiple result lists score higher.
"""
scores = {}
doc_map = {}
for results in result_lists:
for rank, doc in enumerate(results):
doc_id = doc.document_id
doc_map[doc_id] = doc
if doc_id not in scores:
scores[doc_id] = 0.0
scores[doc_id] += 1.0 / (k + rank + 1)
# Sort by RRF score
sorted_ids = sorted(scores.keys(), key=lambda x: -scores[x])
fused = []
for doc_id in sorted_ids:
doc = doc_map[doc_id]
doc.rerank_score = scores[doc_id]
fused.append(doc)
return fusedPhase 3: Production RAG Pipeline
Complete Advanced RAG System
# rag/production_pipeline.py — Full advanced RAG pipeline
from typing import List, Dict, Optional
import time
class AdvancedRAGPipeline:
"""
Production RAG pipeline with advanced retrieval techniques.
Pipeline:
1. Query analysis (classify query type)
2. Query transformation (expansion, decomposition, or HyDE)
3. Multi-stage retrieval (vector search → re-ranking)
4. Context assembly (deduplication, ordering, compression)
5. LLM generation with structured context
"""
def __init__(self, vector_store, reranker, query_transformer,
llm_client, context_compressor=None):
self.vector_store = vector_store
self.reranker = reranker
self.transformer = query_transformer
self.llm = llm_client
self.compressor = context_compressor
def query(self, question: str, conversation_history: List[Dict] = None,
filters: Dict = None) -> Dict:
"""Execute a full RAG query with advanced retrieval."""
timings = {}
# Step 1: Query analysis
start = time.time()
query_type = self._classify_query(question)
timings["query_analysis"] = time.time() - start
# Step 2: Query transformation (strategy depends on query type)
start = time.time()
search_queries = self._transform_query(question, query_type)
timings["query_transform"] = time.time() - start
# Step 3: Multi-stage retrieval
start = time.time()
all_candidates = []
for q in search_queries:
candidates = self.vector_store.search(
query=q, top_k=50, filters=filters
)
all_candidates.extend(candidates)
# Deduplicate candidates
seen = set()
unique_candidates = []
for doc in all_candidates:
if doc.document_id not in seen:
seen.add(doc.document_id)
unique_candidates.append(doc)
timings["retrieval"] = time.time() - start
# Step 4: Re-ranking
start = time.time()
reranked = self.reranker.rerank(question, unique_candidates[:50])
top_docs = reranked[:10]
timings["reranking"] = time.time() - start
# Step 5: Context assembly
start = time.time()
context = self._assemble_context(question, top_docs, query_type)
timings["context_assembly"] = time.time() - start
# Step 6: LLM generation
start = time.time()
response = self._generate_response(question, context, conversation_history)
timings["generation"] = time.time() - start
return {
"answer": response,
"sources": [
{"id": d.document_id, "content": d.content[:200], "score": d.rerank_score}
for d in top_docs[:5]
],
"timings": timings,
"query_type": query_type,
"search_queries_used": search_queries,
}
def _classify_query(self, question: str) -> str:
"""Classify query type to determine retrieval strategy."""
# Simple keyword-based classification (use LLM for production)
if any(word in question.lower() for word in ["compare", "vs", "difference"]):
return "comparative"
elif any(word in question.lower() for word in ["how", "step", "process"]):
return "procedural"
elif any(word in question.lower() for word in ["what is", "define", "explain"]):
return "definitional"
elif "?" not in question:
return "keyword"
else:
return "general"
def _transform_query(self, question: str, query_type: str) -> List[str]:
"""Apply query transformation based on query type."""
if query_type == "comparative":
# Decompose into sub-queries for each entity
return self.transformer.query_decomposition(question)
elif query_type == "procedural":
# HyDE: generate hypothetical procedure
hyde_query = self.transformer.hyde(question)
return [question, hyde_query]
elif query_type == "definitional":
# Multi-query expansion
return self.transformer.multi_query_expansion(question)
else:
# General: HyDE + original
hyde_query = self.transformer.hyde(question)
return [question, hyde_query]
def _assemble_context(self, question: str, documents: List[RetrievedDocument],
query_type: str) -> str:
"""Assemble retrieved documents into a coherent context string."""
context_parts = []
for i, doc in enumerate(documents):
context_parts.append(
f"[Source {i+1}] (relevance: {doc.rerank_score:.2f})\n{doc.content}"
)
context = "\n\n".join(context_parts)
# Optional: compress context if it exceeds token limit
if self.compressor and len(context) > 8000:
context = self.compressor.compress(question, context, max_tokens=4000)
return context
def _generate_response(self, question: str, context: str,
conversation_history: List[Dict] = None) -> str:
"""Generate response using LLM with retrieved context."""
system_prompt = """You are a helpful assistant that answers questions based on the provided context.
Rules:
- Only use information from the provided sources
- Cite sources as [Source N] when referencing specific information
- If the context doesn't contain enough information, say so clearly
- Be concise and direct"""
messages = [{"role": "system", "content": system_prompt}]
if conversation_history:
messages.extend(conversation_history[-6:]) # Last 3 turns
messages.append({
"role": "user",
"content": f"Context:\n{context}\n\nQuestion: {question}",
})
response = self.llm.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0.0,
)
return response.choices[0].message.contentPhase 4: Evaluation
Retrieval Quality Metrics
# rag/evaluation.py — Evaluate advanced RAG techniques
import numpy as np
from typing import List, Dict
class RAGEvaluator:
"""Evaluate RAG retrieval and generation quality."""
def evaluate_retrieval(self, queries_with_ground_truth: List[Dict],
retriever) -> Dict:
"""
Evaluate retrieval quality using standard IR metrics.
Ground truth format:
{"query": "...", "relevant_doc_ids": ["doc1", "doc2", ...]}
"""
precisions = []
recalls = []
ndcgs = []
mrrs = []
for item in queries_with_ground_truth:
query = item["query"]
relevant = set(item["relevant_doc_ids"])
retrieved = retriever.retrieve(query, top_k=20)
retrieved_ids = [doc.document_id for doc in retrieved]
# Precision@k
for k in [5, 10]:
top_k = retrieved_ids[:k]
precision = len(set(top_k) & relevant) / k
precisions.append(precision)
# Recall@k
for k in [10, 20]:
top_k = retrieved_ids[:k]
recall = len(set(top_k) & relevant) / len(relevant) if relevant else 0
recalls.append(recall)
# MRR (Mean Reciprocal Rank)
for rank, doc_id in enumerate(retrieved_ids):
if doc_id in relevant:
mrrs.append(1.0 / (rank + 1))
break
else:
mrrs.append(0.0)
# NDCG@10
ndcg = self._compute_ndcg(retrieved_ids, relevant, k=10)
ndcgs.append(ndcg)
return {
"precision@5": np.mean(precisions[:len(queries_with_ground_truth)]),
"precision@10": np.mean(precisions[len(queries_with_ground_truth):]),
"recall@10": np.mean(recalls[:len(queries_with_ground_truth)]),
"recall@20": np.mean(recalls[len(queries_with_ground_truth):]),
"MRR": np.mean(mrrs),
"NDCG@10": np.mean(ndcgs),
}
def _compute_ndcg(self, retrieved: List[str], relevant: set, k: int) -> float:
"""Compute Normalized Discounted Cumulative Gain."""
dcg = 0.0
for i, doc_id in enumerate(retrieved[:k]):
relevance = 1.0 if doc_id in relevant else 0.0
dcg += relevance / np.log2(i + 2)
# Ideal DCG (all relevant docs at top)
ideal_hits = min(len(relevant), k)
idcg = sum(1.0 / np.log2(i + 2) for i in range(ideal_hits))
return dcg / idcg if idcg > 0 else 0.0Conclusion
Advanced RAG techniques—re-ranking, query transformation, and multi-stage retrieval—improve retrieval quality by 15-40% over naive vector search. Cross-encoder re-ranking is the single highest-impact technique, improving NDCG@10 by 20-30% at the cost of 50ms additional latency. Query transformation (multi-query expansion, HyDE, decomposition) improves recall for complex queries by 15-25%. The optimal strategy depends on query type: comparative queries benefit from decomposition, procedural queries benefit from HyDE, and definitional queries benefit from multi-query expansion. Production implementations should classify queries first, then apply the appropriate transformation strategy. Teams that layer these techniques achieve retrieval quality that approaches human-curated context, dramatically improving the accuracy and relevance of LLM-generated answers.
