RAG Evaluation: как измерить качество системы
opensourceaillmragevaluationmetricsit
Введение: RAG Evaluation
Проблема
RAG systems are hard to evaluate:
- No single metric captures quality
- Retrieval and generation are coupled
- Human evaluation is expensive
- Automated metrics don't correlate well
Solution: Multi-metric Evaluation
- Evaluate retrieval and generation separately
- Use automated metrics for speed
- Use human evaluation for accuracy
- Track metrics over time
Факт: Good RAG systems should achieve >0.7 recall@10 and >0.6 faithfulness score.
Что такое RAG Evaluation?
Основная идея
RAG Evaluation Pipeline:
Query → [Retriever] → Documents → [Generator] → Answer
↓ ↓
Retrieve Metrics Generation Metrics
↓ ↓
Recall@K Faithfulness
Precision@K Answer Correctness
MRR Semantic Similarity
Two-Stage Evaluation
1. Retrieval Evaluation:
- Are the right documents retrieved?
- How relevant are the retrieved documents?
- How fast is retrieval?
2. Generation Evaluation:
- Is the answer faithful to retrieved docs?
- Is the answer correct?
- Is the answer fluent and coherent?
3. End-to-End Evaluation:
- Does the full system work well?
- User satisfaction
- Task completion rate
Retrieval Metrics
Recall@K
def recall_at_k(retrieved_docs, relevant_docs, k):
"""
What fraction of relevant documents are retrieved?
retrieved_docs: list of retrieved doc IDs
relevant_docs: list of ground truth relevant doc IDs
k: number of retrieved documents
"""
retrieved_k = retrieved_docs[:k]
hits = sum(1 for doc in retrieved_k if doc in relevant_docs)
return hits / len(relevant_docs)
# Example:
# retrieved: [doc1, doc3, doc5, doc7, doc9]
# relevant: [doc1, doc3, doc5, doc7, doc9, doc11]
# recall@5 = 5/6 = 0.83
# Target: recall@10 > 0.7
Precision@K
def precision_at_k(retrieved_docs, relevant_docs, k):
"""
What fraction of retrieved documents are relevant?
"""
retrieved_k = retrieved_docs[:k]
hits = sum(1 for doc in retrieved_k if doc in relevant_docs)
return hits / k
# Example:
# retrieved: [doc1, doc3, doc5, doc7, doc9]
# relevant: [doc1, doc3, doc5, doc7, doc9, doc11]
# precision@5 = 5/5 = 1.0
# Target: precision@10 > 0.4
Mean Reciprocal Rank (MRR)
def reciprocal_rank(retrieved_docs, relevant_docs):
"""
1 / rank of first relevant document
"""
for rank, doc in enumerate(retrieved_docs, 1):
if doc in relevant_docs:
return 1 / rank
return 0
def mrr(all_queries):
"""Average reciprocal rank across all queries"""
return sum(reciprocal_rank(q["retrieved"], q["relevant"])
for q in all_queries) / len(all_queries)
# Example:
# retrieved: [doc5, doc1, doc3, doc7, doc9]
# relevant: [doc1, doc3, doc5, doc7, doc9, doc11]
# rr = 1/2 = 0.5 (doc1 is at rank 2)
# Target: MRR > 0.6
NDCG@K (Normalized Discounted Cumulative Gain)
def ndcg_at_k(retrieved_docs, relevance_scores, k):
"""
Considers graded relevance, not just binary
"""
# DCG: Discounted Cumulative Gain
dcg = 0
for i, doc in enumerate(retrieved_docs[:k]):
rel = relevance_scores.get(doc, 0)
dcg += rel / np.log2(i + 2) # i+2 because i is 0-indexed
# Ideal DCG
ideal_relevance = sorted(relevance_scores.values(), reverse=True)[:k]
idcg = sum(r / np.log2(i + 2) for i, r in enumerate(ideal_relevance))
return dcg / idcg if idcg > 0 else 0
# Relevance scores:
# 0 = not relevant
# 1 = somewhat relevant
# 2 = relevant
# 3 = highly relevant
# Target: NDCG@10 > 0.6
Generation Metrics
Faithfulness
from transformers import pipeline
def check_faithfulness(answer, context):
"""
Is the answer faithful to the context?
Uses NLI model to verify claims.
"""
# Split answer into claims
claims = split_into_claims(answer)
# Check each claim against context
faithfulness_scores = []
for claim in claims:
# NLI: context + claim → entailment/contradiction
nli_result = nli_pipeline({
"premise": context,
"hypothesis": claim
})
is_entailed = nli_result["label"] == "ENTAILMENT"
faithfulness_scores.append(1 if is_entailed else 0)
return sum(faithfulness_scores) / len(faithfulness_scores)
# Example:
# Context: "The cat sat on the mat"
# Answer: "The dog sat on the mat"
# Claim: "The dog sat on the mat"
# Faithfulness: 0.0 (contradiction)
# Target: faithfulness > 0.8
Answer Correctness
def answer_correctness(answer, ground_truth):
"""
How correct is the answer compared to ground truth?
"""
# Method 1: Exact match
exact = answer.strip().lower() == ground_truth.strip().lower()
# Method 2: Semantic similarity
similarity = semantic_similarity(answer, ground_truth)
# Method 3: LLM-based evaluation
llm_score = llm_evaluate(answer, ground_truth)
# Method 4: ROUGE/BLEU for text generation
rouge_score = rouge_score(answer, ground_truth)
return {
"exact": exact,
"semantic_similarity": similarity,
"llm_score": llm_score,
"rouge": rouge_score,
}
# Method comparison:
# Exact match: strict, but too strict for open-ended
# Semantic similarity: captures meaning, but noisy
# LLM-based: flexible, but expensive
# ROUGE/BLEU: standard for generation, but limited
Semantic Similarity
from sentence_transformers import SentenceTransformer, util
model = SentenceTransformer("all-MiniLM-L6-v2")
def semantic_similarity(text1, text2):
"""Cosine similarity between text embeddings"""
embedding1 = model.encode(text1, convert_to_tensor=True)
embedding2 = model.encode(text2, convert_to_tensor=True)
similarity = util.cos_sim(embedding1, embedding2)
return similarity.item()
# Example:
# text1: "The cat sat on the mat"
# text2: "A feline was sitting on a rug"
# similarity: 0.85
# Target: semantic similarity > 0.7
LLM-as-a-Judge
def llm_judge(answer, question, context, ground_truth=None):
"""
Use LLM to evaluate the answer
"""
prompt = f"""
Question: {question}
Context: {context}
Answer: {answer}
{'Ground Truth: ' + ground_truth if ground_truth else ''}
Evaluate the answer on a scale of 1-5:
1. Completely wrong
2. Mostly wrong
3. Partially correct
4. Mostly correct
5. Completely correct
Score:
"""
response = llm.generate(prompt, max_tokens=100)
score = int(response.strip())
return score
# Advantages:
# - Captures nuanced correctness
# - Can evaluate open-ended answers
# - Correlates well with human judgment
# Disadvantages:
# - Expensive
# - Slow
# - LLM bias
End-to-End Metrics
Task Completion Rate
def task_completion_rate(answers, task_outcomes):
"""
What fraction of tasks were completed successfully?
"""
correct = sum(1 for a, o in zip(answers, task_outcomes)
if o == "success")
return correct / len(answers)
# Example:
# Tasks: ["What is the capital of France?", "Who wrote Hamlet?"]
# Answers: ["Paris", "William Shakespeare"]
# Outcomes: ["success", "success"]
# Task completion rate: 1.0
# Target: task completion rate > 0.8
Response Latency
import time
def measure_latency(retriever, generator, query):
"""Measure end-to-end latency"""
start = time.time()
# Retrieval
retrieved = retriever.search(query)
retrieval_time = time.time() - start
# Generation
answer = generator.generate(query, retrieved)
generation_time = time.time() - start - retrieval_time
return {
"total_latency": time.time() - start,
"retrieval_time": retrieval_time,
"generation_time": generation_time,
"tokens_per_second": len(answer.split()) / generation_time,
}
# Example:
# total_latency: 2.5s
# retrieval_time: 0.3s
# generation_time: 2.2s
# tokens_per_second: 50
# Target: total latency < 5s, TPS > 30
Evaluation Framework
Creating Evaluation Dataset
import json
# Create evaluation dataset
eval_dataset = {
"queries": [
{
"id": "query_001",
"question": "What is the population of Paris?",
"expected_answer": "Paris has a population of approximately 2.1 million.",
"relevant_docs": ["paris_wikipedia.txt", "paris_census.txt"],
"category": "factoid",
},
{
"id": "query_002",
"question": "How do I install Ollama on Linux?",
"expected_answer": "Run curl -fsSL https://ollama.com/install.sh | sh",
"relevant_docs": ["ollama_docs.txt", "linux_install.txt"],
"category": "procedural",
},
# ... more queries
]
}
# Save dataset
with open("eval_dataset.json", "w") as f:
json.dump(eval_dataset, f, indent=2)
# Dataset size:
# - Minimum: 50 queries
# - Recommended: 100-200 queries
# - Categories: factoid, procedural, comparative, open-ended
Running Evaluation
def run_evaluation(retriever, generator, dataset):
"""Run full evaluation pipeline"""
results = []
for query in dataset["queries"]:
# Retrieve
retrieved = retriever.search(query["question"])
# Generate
answer = generator.generate(query["question"], retrieved)
# Evaluate retrieval
retrieval_metrics = {
"recall@10": recall_at_k(retrieved, query["relevant_docs"], 10),
"precision@10": precision_at_k(retrieved, query["relevant_docs"], 10),
"ndcg@10": ndcg_at_k(retrieved, query["relevance_scores"], 10),
}
# Evaluate generation
generation_metrics = {
"faithfulness": check_faithfulness(answer, retrieved),
"semantic_similarity": semantic_similarity(
answer, query["expected_answer"]
),
"llm_score": llm_judge(answer, query["question"], retrieved),
}
results.append({
"query_id": query["id"],
"question": query["question"],
"answer": answer,
"retrieval_metrics": retrieval_metrics,
"generation_metrics": generation_metrics,
})
return results
# Aggregate results
def aggregate_results(results):
"""Aggregate metrics across all queries"""
avg_recall = np.mean([r["retrieval_metrics"]["recall@10"]
for r in results])
avg_faithfulness = np.mean([r["generation_metrics"]["faithfulness"]
for r in results])
avg_similarity = np.mean([r["generation_metrics"]["semantic_similarity"]
for r in results])
return {
"avg_recall@10": avg_recall,
"avg_precision@10": np.mean([r["retrieval_metrics"]["precision@10"]
for r in results]),
"avg_ndcg@10": np.mean([r["retrieval_metrics"]["ndcg@10"]
for r in results]),
"avg_faithfulness": avg_faithfulness,
"avg_semantic_similarity": avg_similarity,
}
Automated Evaluation Tools
RAGAS
# RAGAS: Retrieval Augmented Generation Assessment
from ragas import evaluate
from ragas.metrics import faithfulness, answer_correctness, context_precision
# Load dataset
dataset = load_dataset("my_eval_dataset")
# Evaluate
result = evaluate(
dataset=dataset,
metrics=[faithfulness, answer_correctness, context_precision],
llm=llm,
embeddings=embedding_model,
)
# Get scores
scores = result.to_pandas()[["faithfulness", "answer_correctness", "context_precision"]]
print(scores.mean())
# RAGAS scores:
# - Faithfulness: 0-1 (higher is better)
# - Answer correctness: 0-1 (higher is better)
# - Context precision: 0-1 (higher is better)
DeepEval
# DeepEval: Evaluation framework for RAG
from deepeval import evaluate
from deepeval.metrics import FaithfulnessMetric, AnswerRelevancyMetric
def evaluate_with_deepeval(question, answer, context):
"""Evaluate using DeepEval"""
# Faithfulness
faithfulness_metric = FaithfulnessMetric(
threshold=0.7,
model="GPT-4",
include_reason=True,
)
faithfulness_metric.measure(question, answer, context)
# Answer relevancy
answer_metric = AnswerRelevancyMetric(
threshold=0.7,
model="GPT-4",
include_reason=True,
)
answer_metric.measure(question, answer)
return {
"faithfulness": faithfulness_metric.score,
"answer_relevancy": answer_metric.score,
"faithfulness_reason": faithfulness_metric.reason,
}
LangSmith
# LangSmith: Evaluation for LangChain apps
from langsmith import Client
from langsmith.evaluation import evaluate
client = Client()
@evaluate(
name="RAG Evaluation",
evaluator=[
"qa", # Question-Answer correctness
"context_precision",
"context_recall",
],
)
def predict(question: str) -> str:
"""RAG pipeline"""
context = retriever.search(question)
return generator.generate(question, context)
# Run evaluation
dataset_id = client.upload_csv("eval_dataset.csv")
results = client.evaluate(
predict,
data=dataset_id,
)
A/B Testing
Comparing Configurations
def ab_test(retriever_a, retriever_b, generator, dataset):
"""A/B test two retrieval configurations"""
results_a = []
results_b = []
for query in dataset:
# Version A
docs_a = retriever_a.search(query["question"])
answer_a = generator.generate(query["question"], docs_a)
score_a = llm_judge(answer_a, query["question"], docs_a)
results_a.append(score_a)
# Version B
docs_b = retriever_b.search(query["question"])
answer_b = generator.generate(query["question"], docs_b)
score_b = llm_judge(answer_b, query["question"], docs_b)
results_b.append(score_b)
# Statistical test
from scipy import stats
t_stat, p_value = stats.ttest_ind(results_a, results_b)
return {
"version_a_avg": np.mean(results_a),
"version_b_avg": np.mean(results_b),
"p_value": p_value,
"significant": p_value < 0.05,
}
# Use case:
# - Compare different embedding models
# - Compare different chunk sizes
# - Compare different retrieval strategies
Заключение
RAG evaluation requires multiple metrics:
Retrieval metrics:
- Recall@K, Precision@K
- MRR, NDCG@K
- Target: recall@10 > 0.7, MRR > 0.6
Generation metrics:
- Faithfulness, Answer correctness
- Semantic similarity, LLM-as-a-judge
- Target: faithfulness > 0.8, similarity > 0.7
Tools:
- RAGAS: automated evaluation
- DeepEval: faithfulness + relevancy
- LangSmith: A/B testing
Best practices:
- Create evaluation dataset (100+ queries)
- Evaluate retrieval and generation separately
- Run A/B tests for configuration changes
- Track metrics over time