Learn how to turn RAG evaluation into concrete pipeline improvements with hybrid search, metadata filters, chunking, reranking, query expansion, compression, and latency-aware optimization.
The trap most teams fall into
A team spent weeks optimizing their RAG pipeline. Every metric was green. Precision@K: 0.92. MRR: 0.88. Groundedness: 0.94. Their dashboard was a thing of beauty. Then users started complaining. ‘The answer is wrong.’ ‘It contradicts what I just read.’ The metrics said ‘success.’ The users said ‘broken.’ That’s when they learned that metrics without action are just beautiful lies.
RAG optimization is the part where metrics become engineering decisions.
What optimization really means ?
Optimization in RAG means changing the pipeline so the metrics move in the right direction.
If recall is low, fix retrieval.
If precision is low, reduce noise.
If groundedness is low, improve evidence quality or prompting.
If latency is high, compress context or reduce unnecessary retrieval.
If the answer is relevant but unsupported, tighten verification.
If a topic keeps failing, refresh the index and improve metadata.
That is the practical loop: measure, diagnose, change, re-evaluate, monitor.
The optimization loop:
Measure ───> Diagnose ───> Change ───> Re-evaluate ───> Monitor
▲ │
└───────────────────────────────────────────────────────────┘
If you skip diagnosis, you tune randomly. If you skip re-evaluation, you do not know whether the change helped.
The multi-stage retrieval framework
The strongest production systems do not rely on a single retriever. They split the workflow into specialized stages.
User Query ───> [ LLM Query Rewrite ] ───> [ Hybrid Search (Sparse + Dense) ]
│
Retrieves Top-50
▼
[ Verified Output ] <─── [ Compressed Context ] <─── [ Reranker (Top-10) ]
This is the right pattern because broad retrieval improves recall, reranking improves precision, and compression controls token cost.
Start with the metric
Every optimization should begin with the metric that is failing.
- 1. Measure (The Baseline) – Optimization starts with visibility. You cannot fix what you do not track, so establish clear baselines for your retrieval and generation layers first.
- 2. Diagnose (The Root Cause) – Look past the surface errors. Use specialized evaluation frameworks to pinpoint exactly why your pipeline is hallucinating or missing key context.
- 3. Change (The Action) – Turn insights into action. Whether it means tweaking your chunk size, upgrading your embedding model, or rewriting your system prompts, make targeted adjustments.
- 4. Re-evaluate (The Verification) – Never deploy blindly. Run regression tests against your golden dataset to prove that your changes actually improved system performance without breaking something else.
- 5. Monitor (The Safeguard) – Production environments change, and user behavior shifts. Set up continuous guardrails to catch performance drift, cost spikes, or latency lag in real-time.
- Recall: Did the system actually find all the relevant context, or did it leave vital documents behind?
- Precision: Is the retrieved data hyper-relevant, or is the LLM choking on noisy, useless information?
- MRR (Mean Reciprocal Rank): How high up in the search results is the absolute best answer hiding?
- NDCG (Normalized Discounted Cumulative Gain): Does the overall ranking quality order documents correctly by true relevance?
- Groundedness: Is the LLM sticking strictly to your data, or is it hallucinating plausible lies?
- Latency: How long are users waiting for the AI to retrieve, process, and type out the final response?
- Token Cost: Are your prompt engineering tactics and heavy context windows silently draining your API budget?
Query rewriting and expansion:
Query rewriting is one of the highest-leverage RAG optimizations. The user’s original query is often vague, underspecified, or too short for good retrieval. In production, rewriting is not just string concatenation. Teams often use an LLM to generate multiple semantic variants, sub-queries, or HyDE-style retrieval prompts.
from openai import AsyncOpenAI
from typing import List
import asyncio
class ProductionQueryRewriter:
def __init__(self, model: str = “gpt-4o-mini”):
self.client = AsyncOpenAI()
self.model = model
async def rewrite(self, original_query: str) -> List[str]:
system_prompt = “””You are a query expansion engine.
Generate 3 semantically distinct search queries for the user’s question.
Each should capture a different aspect or phrasing of the intent.
Return as a list of strings, one per line.”””
response = await self.client.chat.completions.create(
model=self.model,
messages=[
{“role”: “system”, “content”: system_prompt},
{“role”: “user”, “content”: f”Original query: {original_query}”}
],
temperature=0.7,
max_tokens=150
)
lines = response.choices[0].message.content.strip().split(‘\n’)
return [line.strip() for line in lines if line.strip()]
A rewritten query often improves recall before retrieval even runs.
Query expansion with synonyms:
Query expansion can help when the query vocabulary does not match the document vocabulary.
import nltk
from nltk.corpus import wordnet
def expand_query(query: str) -> str:
words = query.lower().split()
expanded = []
for word in words:
expanded.append(word)
synonyms = []
for syn in wordnet.synsets(word):
for lemma in syn.lemmas():
name = lemma.name().lower()
if name != word:
synonyms.append(name)
expanded.extend(synonyms[:3])
return ” “.join(expanded)
print(expand_query(“leave policy employee”))
For beginners, this is a useful baseline. For complex corpora, developers should pivot to HyDE or LLM-driven sub-query generation, because WordNet-style synonym expansion often injects irrelevant terms and dilutes vector meaning.
HyDE retrieval
HyDE is useful when the query is too short or vague for dense retrieval.
class HyDERetriever:
def __init__(self, llm_client, encoder):
self.llm = llm_client
self.encoder = encoder
async def hyde_search(self, query: str, top_k: int = 5) -> List[str]:
prompt = f”””Write a hypothetical document that would answer this question:
{query}
The document should be detailed and factual.”””
response = await self.llm.generate(prompt)
hypothetical_doc = response.text
hyde_embedding = self.encoder.encode([hypothetical_doc])
results = self.vector_db.search(hyde_embedding, top_k=top_k)
return results
HyDE is especially valuable when the user’s wording is thin but the intent is rich.
Hybrid search
Hybrid search combines sparse retrieval, dense retrieval, and metadata awareness.
Sparse search is good for exact terms, codes, names, and policy IDs.
Dense search is good for semantic similarity and paraphrases.
Metadata filters are good for region, department, date, and document type.
from sklearn.feature_extraction.text import TfidfVectorizer
from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List, Dict
class HybridRetriever:
def __init__(self, documents: List[str]):
self.documents = documents
self.vectorizer = TfidfVectorizer(stop_words=’english’)
self.tfidf_matrix = self.vectorizer.fit_transform(documents)
self.encoder = SentenceTransformer(‘all-MiniLM-L6-v2’)
self.dense_embeddings = self.encoder.encode(documents)
def search(self, query: str, top_k: int = 5, sparse_weight: float = 0.3, dense_weight: float = 0.7) -> List[Dict]:
query_vector = self.vectorizer.transform([query])
sparse_scores = np.array(self.tfidf_matrix @ query_vector.T).flatten()
query_embedding = self.encoder.encode([query])
dense_scores = np.array([
np.dot(query_embedding, doc_emb)
for doc_emb in self.dense_embeddings
]).flatten()
sparse_scores = sparse_scores / (sparse_scores.max() + 1e-8)
dense_scores = dense_scores / (dense_scores.max() + 1e-8)
combined_scores = (sparse_weight * sparse_scores) + (dense_weight * dense_scores)
top_indices = np.argsort(combined_scores)[::-1][:top_k]
return [
{
“document”: self.documents[i],
“score”: float(combined_scores[i]),
“sparse_score”: float(sparse_scores[i]),
“dense_score”: float(dense_scores[i])
}
for i in top_indices
]
Metadata filtering:
Metadata filters are essential for enterprise RAG because many failures are really scope failures.
from typing import List, Dict, Optional
class MetadataFilteredRetriever:
def __init__(self, documents: List[Dict]):
self.documents = documents
def search(self, query: str, filters: Optional[Dict] = None, top_k: int = 5) -> List[Dict]:
filtered_docs = self.documents
if filters:
filtered_docs = [
doc for doc in self.documents
if all(doc.get(‘metadata’, {}).get(k) == v for k, v in filters.items())
]
query_terms = set(query.lower().split())
scored = []
for doc in filtered_docs:
text = doc[‘text’].lower()
score = sum(1 for t in query_terms if t in text)
scored.append((score, doc))
scored.sort(key=lambda x: x[0], reverse=True)
return scored[:top_k]
This is especially important for region-specific, department-specific, or year-specific policy search.
Structure-aware chunking:
Bad chunking can ruin a good retriever. Structure-aware chunking preserves meaning while keeping retrieval focused.
from typing import List, Dict
import re
def structure_aware_chunking(document: str, max_chunk_size: int = 500) -> List[Dict]:
chunks = []
sections = re.split(r'(?=^(?:#|Section|CHAPTER|Part)\s+)’, document, flags=re.MULTILINE)
current_chunk = “”
current_heading = “intro”
for section in sections:
if not section.strip():
continue
heading_match = re.match(r’^(#|Section|CHAPTER|Part)\s+(.+)$’, section.strip(), re.MULTILINE)
if heading_match:
if current_chunk:
chunks.append({
“heading”: current_heading,
“text”: current_chunk.strip(),
“length”: len(current_chunk.split())
})
current_chunk = “”
current_heading = heading_match.group(2).strip()
current_chunk = section
else:
if len(current_chunk.split()) + len(section.split()) <= max_chunk_size:
current_chunk += ” ” + section
else:
chunks.append({
“heading”: current_heading,
“text”: current_chunk.strip(),
“length”: len(current_chunk.split())
})
current_chunk = section
if current_chunk:
chunks.append({
“heading”: current_heading,
“text”: current_chunk.strip(),
“length”: len(current_chunk.split())
})
return chunks
Useful chunking rules:
split by headings when possible,
keep related policy clauses together,
avoid cutting tables in the middle,
preserve references and definitions,
align chunk size with answer granularity.
Semantic reranking:
Reranking is one of the strongest ways to improve precision and MRR. But there is a trade-off: cross-encoder rerankers are slower than vector search because they score each query-document pair jointly.
The mechanical difference matters:
Bi-encoder: embeds query and document independently, then compares vectors; fast and scalable.
Cross-encoder: processes query and document together with joint attention; slower, but far better at capturing semantic intersections.
The practical rule is simple: retrieve broadly, then rerank only the top candidates.
from sentence_transformers import CrossEncoder
from typing import List, Dict
class SemanticReranker:
def __init__(self, model_name=”cross-encoder/ms-marco-MiniLM-L-6-v2″):
self.model = CrossEncoder(model_name)
def rerank(self, query: str, documents: List[str], top_k: int = 5) -> List[Dict]:
if not documents:
return []
pairs = [[query, doc] for doc in documents]
scores = self.model.predict(pairs)
ranked = sorted(
[{“document”: doc, “score”: float(score)} for doc, score in zip(documents, scores)],
key=lambda x: x[“score”],
reverse=True
)
return ranked[:top_k]
Reranking latency trade-off
Reranking improves MRR and precision, but it increases latency. The safest production pattern is to retrieve 50 to 100 candidates with vector search, then rerank only the top 10 or top 20.
That keeps the context high quality without turning the reranker into your main bottleneck.
Smart context compression
Truncation works, but summarization is better when the context is long.
from transformers import pipeline
class SmartCompressor:
def __init__(self):
self.summarizer = pipeline(“summarization”, model=”facebook/bart-large-cnn”)
def compress(self, chunks: List[str], max_words: int = 100) -> List[str]:
compressed = []
for chunk in chunks:
words = chunk.split()
if len(words) <= max_words:
compressed.append(chunk)
else:
try:
summary = self.summarizer(chunk, max_length=max_words, min_length=30, do_sample=False)
compressed.append(summary[0][‘summary_text’])
except:
compressed.append(” “.join(words[:max_words]) + “…”)
return compressed
In production, smaller models or API-based summarizers may be more practical than a local BART pipeline.
Lost in the middle mitigation
Long contexts can bury the most important evidence in the center of the prompt. A simple mitigation is to reorder chunks so the most relevant content appears near the edges.
from typing import List, Tuple
def prioritize_chunks(chunks: List[str], important_terms: List[str]) -> List[str]:
scored = []
for chunk in chunks:
score = sum(1 for term in important_terms if term.lower() in chunk.lower())
scored.append((score, chunk))
scored.sort(key=lambda x: x[0], reverse=True)
reordered = [chunk for _, chunk in scored]
if len(reordered) > 4:
result = reordered[:2] + reordered[4:] + reordered[2:4]
return result
return reordered
def contextual_summarization(chunks: List[str], query: str) -> str:
return “\n\n”.join(chunks[:3]) + “\n\n[Additional context follows]”
Latency reduction techniques:
def optimize_retrieval_depth(recall_metric: float, current_top_k: int) -> int:
if recall_metric > 0.95:
return max(5, current_top_k // 2)
elif recall_metric > 0.90:
return current_top_k
else:
return min(20, current_top_k * 2)
def optimize_chunk_size(precision_metric: float, current_size: int) -> int:
if precision_metric > 0.85:
return min(800, current_size + 200)
else:
return max(200, current_size – 100)
These simple rules show how metric movement can guide operational tuning.
Production RAG optimizer:
from typing import List, Dict
import asyncio
class ProductionRAGOptimizer:
def __init__(self, cross_encoder_client=None, llm_client=None):
self.encoder = cross_encoder_client
self.llm = llm_client
async def generate_expanded_queries(self, original_query: str) -> List[str]:
if self.llm:
return [
original_query,
f”Official documentation guidelines for {original_query}”,
f”Standard operating procedures and exceptions regarding {original_query}”
]
return [
original_query,
f”Official documentation guidelines for {original_query}”,
f”Standard operating procedures and exceptions regarding {original_query}”
]
def enforce_context_compression(self, retrieved_chunks: List[str], max_tokens_per_chunk: int = 50) -> List[str]:
compressed_payload = []
for chunk in retrieved_chunks:
tokens = chunk.split()
if len(tokens) > max_tokens_per_chunk:
compressed_payload.append(” “.join(tokens[:max_tokens_per_chunk]) + “…”)
else:
compressed_payload.append(chunk)
return compressed_payload
Cost-aware optimization logic:
def decide_optimization_path(current_cost_per_query: float, target_cost: float = 0.01) -> dict:
decisions = {}
if current_cost_per_query > target_cost:
decisions[“compression”] = True
decisions[“retrieval_depth”] = “reduce_to_20”
decisions[“reranking”] = “use_fast_model”
decisions[“query_rewrites”] = “reduce_to_2”
else:
decisions[“compression”] = False
decisions[“retrieval_depth”] = “keep_50”
decisions[“reranking”] = “use_accurate_model”
decisions[“query_rewrites”] = “use_3”
return decisions
print(decide_optimization_path(0.025))
print(decide_optimization_path(0.005))
Feedback-driven indexing:
Optimization does not stop with retrieval and reranking. The index itself should improve from failure patterns.
class FeedbackDrivenIndexer:
def __init__(self, vector_store, feedback_threshold: float = 0.3):
self.vector_store = vector_store
self.threshold = feedback_threshold
self.failures = []
def log_failure(self, query: str, retrieved_docs: List[str], wrong_answer: str, correct_answer: str):
self.failures.append({
“query”: query,
“retrieved”: retrieved_docs,
“wrong”: wrong_answer,
“correct”: correct_answer
})
def refresh_index(self):
new_documents = self._generate_synthetic_examples()
self.vector_store.add_documents(new_documents)
self.failures.clear()
def _generate_synthetic_examples(self):
examples = []
for failure in self.failures:
examples.append({
“text”: f”Q: {failure[‘query’]}\nA: {failure[‘correct’]}”,
“metadata”: {“type”: “synthetic”, “corrected”: True}
})
return examples
Retrieval-generation correlation analysis:
This is one of the fastest ways to understand whether the problem is retrieval or generation.
from typing import List, Dict
def analyze_pipeline_performance(eval_results: List[Dict]) -> Dict:
good_retrieval_bad_generation = 0
bad_retrieval_good_generation = 0
both_good = 0
both_bad = 0
for result in eval_results:
retrieval_ok = result[“retrieval”][“precision”] > 0.7
generation_ok = result[“response”][“groundedness”] > 0.8
if retrieval_ok and not generation_ok:
good_retrieval_bad_generation += 1
elif not retrieval_ok and generation_ok:
bad_retrieval_good_generation += 1
elif retrieval_ok and generation_ok:
both_good += 1
else:
both_bad += 1
total = len(eval_results) if eval_results else 1
return {
“retrieval_problem”: bad_retrieval_good_generation / total,
“generation_problem”: good_retrieval_bad_generation / total,
“both_good”: both_good / total,
“both_bad”: both_bad / total,
“recommendation”: (
“Fix retrieval” if bad_retrieval_good_generation > 0.3 else
“Fix generation” if good_retrieval_bad_generation > 0.3 else
“Both need work” if both_bad > 0.3 else
“System is healthy”
)
}
Symptom-driven diagnostic matrix:
def estimate_evaluation_cost(eval_runs: int, tokens_per_run: int, cost_per_1k_tokens: float = 0.002) -> dict:
total_tokens = eval_runs * tokens_per_run
total_cost = total_tokens * cost_per_1k_tokens / 1000
return {
“eval_runs”: eval_runs,
“tokens_per_run”: tokens_per_run,
“total_tokens”: total_tokens,
“estimated_cost_usd”: round(total_cost, 3)
}
print(estimate_evaluation_cost(1000, 500))
When to use which optimization ?
This matters a lot in enterprise systems where the knowledge base may contain English policies, regional updates, office-specific exceptions, and code-switched content. If your system relies strictly on dense vector search, it can miss documents with exact legal terms, local codes, or regional policy labels.
Hybrid retrieval plus metadata tagging is the cleanest way to recover context while keeping token usage under control.
Quick-reference card
Low recall → widen retrieval or rewrite the query.
Low precision → rerank and filter more aggressively.
Low groundedness → improve evidence quality and compression.
High latency → shrink context and reduce reranker input.
Repeated topic failures → refresh the index and add feedback loops.
Never tune blindly; always tie a change to a metric.
Glossary
Query Rewriting: Restructuring the user query for better retrieval.
Query Expansion: Adding synonyms or semantic variants.
HyDE: Retrieval using a hypothetical document embedding.
Hybrid Search: Combining sparse, dense, and metadata-based retrieval.
Reranking: Reordering candidate chunks by stronger relevance scoring.
Context Compression: Reducing retrieved text while preserving meaning.
Feedback-driven Indexing: Updating the index based on observed failures.
Groundedness: Whether the answer is supported by retrieved evidence.
Precision: How much of the retrieved content is relevant.
Recall: How much of the relevant content was retrieved.
RAG optimization is not about changing the model first. It is about changing the pipeline around the model so the metrics move in the right direction. If recall is low, fix retrieval. If precision is low, fix ranking. If groundedness is low, fix evidence quality. If latency is high, fix context size. The best RAG teams do not chase clever prompts first; they turn metrics into concrete system changes and then remeasure.