latestaiagents RAG Evaluation
Measure, monitor, and improve RAG system performance with comprehensive metrics.
When to Use
- Setting up RAG evaluation pipelines
- Comparing retrieval strategies
- Measuring generation quality
- Building regression tests for RAG
- Debugging poor RAG performance
Evaluation Framework
┌─────────────────────────────────────────────────────────┐
│ RAG Evaluation │
├─────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ Retrieval │ │ Generation │ │ End-to-End │ │
│ │ Metrics │ │ Metrics │ │ Metrics │ │
│ ├─────────────┤ ├─────────────┤ ├─────────────┤ │
│ │ • MRR │ │ • Faithful- │ │ • Answer │ │
│ │ • Recall@k │ │ ness │ │ Correct- │ │
│ │ • Precision │ │ • Relevance │ │ ness │ │
│ │ • NDCG │ │ • Coherence │ │ • Latency │ │
│ │ • Hit Rate │ │ • Toxicity │ │ • Cost │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
│ │
└─────────────────────────────────────────────────────────┘
Retrieval Metrics
Implementation
import numpy as np
from typing import List, Dict
def mean_reciprocal_rank(results: List[List[str]], relevant: List[List[str]]) -> float:
"""
Calculate MRR across queries.
results: List of ranked document IDs per query
relevant: List of relevant document IDs per query
"""
mrr_sum = 0.0
for res, rel in zip(results, relevant):
rel_set = set(rel)
for rank, doc_id in enumerate(res, 1):
if doc_id in rel_set:
mrr_sum += 1.0 / rank
break
return mrr_sum / len(results)
def recall_at_k(results: List[List[str]], relevant: List[List[str]], k: int) -> float:
"""Calculate Recall@k."""
recall_sum = 0.0
for res, rel in zip(results, relevant):
retrieved_k = set(res[:k])
relevant_set = set(rel)
if relevant_set:
recall_sum += len(retrieved_k & relevant_set) / len(relevant_set)
return recall_sum / len(results)
def precision_at_k(results: List[List[str]], relevant: List[List[str]], k: int) -> float:
"""Calculate Precision@k."""
precision_sum = 0.0
for res, rel in zip(results, relevant):
retrieved_k = set(res[:k])
relevant_set = set(rel)
precision_sum += len(retrieved_k & relevant_set) / k
return precision_sum / len(results)
def ndcg_at_k(results: List[List[str]], relevant: List[List[str]], k: int) -> float:
"""Calculate NDCG@k."""
def dcg(scores):
return sum(s / np.log2(i + 2) for i, s in enumerate(scores))
ndcg_sum = 0.0
for res, rel in zip(results, relevant):
rel_set = set(rel)
gains = [1 if doc in rel_set else 0 for doc in res[:k]]
ideal_gains = sorted(gains, reverse=True)
dcg_val = dcg(gains)
idcg_val = dcg(ideal_gains)
ndcg_sum += dcg_val / idcg_val if idcg_val > 0 else 0
return ndcg_sum / len(results)
def hit_rate(results: List[List[str]], relevant: List[List[str]], k: int) -> float:
"""Calculate Hit Rate (any relevant doc in top-k)."""
hits = 0
for res, rel in zip(results, relevant):
if set(res[:k]) & set(rel):
hits += 1
return hits / len(results)
Evaluation Runner
class RetrievalEvaluator:
def __init__(self, retriever, test_dataset: List[Dict]):
"""
test_dataset: [{"query": str, "relevant_docs": [str]}]
"""
self.retriever = retriever
self.dataset = test_dataset
def evaluate(self, k_values: List[int] = [1, 3, 5, 10]) -> Dict:
# Run retrieval for all queries
results = []
relevant = []
for item in self.dataset:
docs = self.retriever.invoke(item["query"])
doc_ids = [d.metadata["id"] for d in docs]
results.append(doc_ids)
relevant.append(item["relevant_docs"])
# Calculate metrics
metrics = {"mrr": mean_reciprocal_rank(results, relevant)}
for k in k_values:
metrics[f"recall@{k}"] = recall_at_k(results, relevant, k)
metrics[f"precision@{k}"] = precision_at_k(results, relevant, k)
metrics[f"ndcg@{k}"] = ndcg_at_k(results, relevant, k)
metrics[f"hit_rate@{k}"] = hit_rate(results, relevant, k)
return metrics
Generation Metrics with RAGAS
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall,
answer_correctness
)
from datasets import Dataset
def evaluate_with_ragas(test_data: List[Dict]) -> Dict:
"""
test_data: [{
"question": str,
"answer": str, # Generated answer
"contexts": [str], # Retrieved contexts
"ground_truth": str # Expected answer (optional)
}]
"""
dataset = Dataset.from_list(test_data)
metrics = [
faithfulness, # Is answer grounded in context?
answer_relevancy, # Is answer relevant to question?
context_precision, # Are retrieved contexts precise?
context_recall, # Do contexts cover ground truth?
]
# Add answer_correctness if ground truth available
if "ground_truth" in test_data[0]:
metrics.append(answer_correctness)
result = evaluate(dataset, metrics=metrics)
return result.to_pandas().mean().to_dict()
Custom LLM-as-Judge Evaluation
from langchain_openai import ChatOpenAI
FAITHFULNESS_PROMPT = """Rate if the answer is faithful to the context (uses only information from context).
Context: {context}
Answer: {answer}
Score 1-5:
1: Completely unfaithful, makes up information
2: Mostly unfaithful, significant hallucinations
3: Partially faithful, some unsupported claims
4: Mostly faithful, minor extrapolations
5: Completely faithful, all claims supported
Return JSON: {{"score": N, "reasoning": "..."}}"""
RELEVANCE_PROMPT = """Rate if the answer is relevant to the question.
Question: {question}
Answer: {answer}
Score 1-5:
1: Completely irrelevant
2: Tangentially related
3: Partially answers the question
4: Mostly answers the question
5: Fully and directly answers the question
Return JSON: {{"score": N, "reasoning": "..."}}"""
class LLMJudge:
def __init__(self, model: str = "gpt-4"):
self.llm = ChatOpenAI(model=model, temperature=0)
def judge_faithfulness(self, context: str, answer: str) -> Dict:
prompt = FAITHFULNESS_PROMPT.format(context=context, answer=answer)
result = self.llm.invoke(prompt).content
return json.loads(result)
def judge_relevance(self, question: str, answer: str) -> Dict:
prompt = RELEVANCE_PROMPT.format(question=question, answer=answer)
result = self.llm.invoke(prompt).content
return json.loads(result)
def evaluate_batch(self, test_data: List[Dict]) -> Dict:
faithfulness_scores = []
relevance_scores = []
for item in test_data:
context = "\n".join(item["contexts"])
faith = self.judge_faithfulness(context, item["answer"])
faithfulness_scores.append(faith["score"])
rel = self.judge_relevance(item["question"], item["answer"])
relevance_scores.append(rel["score"])
return {
"avg_faithfulness": np.mean(faithfulness_scores),
"avg_relevance": np.mean(relevance_scores),
"faithfulness_scores": faithfulness_scores,
"relevance_scores": relevance_scores
}
End-to-End RAG Evaluation
import time
from dataclasses import dataclass
@dataclass
class RAGEvalResult:
query: str
retrieved_docs: List
generated_answer: str
latency_ms: float
retrieval_metrics: Dict
generation_metrics: Dict
cost_usd: float
class RAGEvaluator:
def __init__(self, rag_pipeline, judge: LLMJudge):
self.pipeline = rag_pipeline
self.judge = judge
def evaluate_single(self, query: str, ground_truth: Dict = None) -> RAGEvalResult:
# Measure latency
start = time.time()
result = self.pipeline.invoke(query)
latency_ms = (time.time() - start) * 1000
# Retrieval metrics (if ground truth available)
retrieval_metrics = {}
if ground_truth and "relevant_docs" in ground_truth:
retrieved_ids = [d.metadata["id"] for d in result["documents"]]
relevant_ids = ground_truth["relevant_docs"]
retrieval_metrics = {
"recall@5": len(set(retrieved_ids[:5]) & set(relevant_ids)) / len(relevant_ids),
"precision@5": len(set(retrieved_ids[:5]) & set(relevant_ids)) / 5
}
# Generation metrics
context = "\n".join([d.page_content for d in result["documents"]])
generation_metrics = {
"faithfulness": self.judge.judge_faithfulness(context, result["answer"]),
"relevance": self.judge.judge_relevance(query, result["answer"])
}
return RAGEvalResult(
query=query,
retrieved_docs=result["documents"],
generated_answer=result["answer"],
latency_ms=latency_ms,
retrieval_metrics=retrieval_metrics,
generation_metrics=generation_metrics,
cost_usd=estimate_cost(result)
)
def evaluate_dataset(self, test_dataset: List[Dict]) -> Dict:
results = []
for item in test_dataset:
result = self.evaluate_single(
item["query"],
item.get("ground_truth")
)
results.append(result)
# Aggregate metrics
return {
"avg_latency_ms": np.mean([r.latency_ms for r in results]),
"avg_faithfulness": np.mean([r.generation_metrics["faithfulness"]["score"] for r in results]),
"avg_relevance": np.mean([r.generation_metrics["relevance"]["score"] for r in results]),
"total_cost_usd": sum([r.cost_usd for r in results]),
"detailed_results": results
}
Continuous Evaluation Pipeline
def setup_eval_pipeline(rag_pipeline, test_dataset_path: str):
"""Set up automated evaluation."""
# Load test dataset
with open(test_dataset_path) as f:
test_data = json.load(f)
evaluator = RAGEvaluator(rag_pipeline, LLMJudge())
def run_evaluation() -> Dict:
results = evaluator.evaluate_dataset(test_data)
# Log to monitoring
log_metrics({
"rag_latency_p50": np.percentile([r.latency_ms for r in results["detailed_results"]], 50),
"rag_latency_p99": np.percentile([r.latency_ms for r in results["detailed_results"]], 99),
"rag_faithfulness": results["avg_faithfulness"],
"rag_relevance": results["avg_relevance"]
})
# Alert on regression
if results["avg_faithfulness"] < 3.5:
alert("RAG faithfulness dropped below threshold!")
return results
return run_evaluation
Best Practices
- Build golden dataset - manually curated query-answer pairs
- Test edge cases - ambiguous queries, no-answer scenarios
- Track over time - metrics should improve, not regress
- Separate retrieval vs generation - diagnose issues precisely
- Use multiple judges - LLM judges can disagree
- Monitor production - sample and evaluate live traffic