RAG PIPELINE: Query → Retrieval → Rerank → Context → Generation → Answer EVALUATION LAYERS: 1. Retrieval Quality (precision@k, recall@k, MRR, nDCG) 2. Generation Quality (faithfulness, relevance, hallucination) 3. End-to-End (answer correctness, user satisfaction) 4. System Metrics (latency, cost, throughput, errors) 

# rag_eval/retrieval_metrics.py from typing import List, Dict, Set import numpy as np from rank_bm25 import BM25Okapi def precision_at_k(retrieved: List[str], relevant: Set[str], k: int) -> float: """Precision@K: fraction of top-K retrieved that are relevant.""" if k == 0: return 0.0 top_k = retrieved[:k] relevant_retrieved = sum(1 for doc in top_k if doc in relevant) return relevant_retrieved / k def recall_at_k(retrieved: List[str], relevant: Set[str], k: int) -> float: """Recall@K: fraction of relevant docs retrieved in top-K.""" if len(relevant) == 0: return 0.0 top_k = retrieved[:k] relevant_retrieved = sum(1 for doc in top_k if doc in relevant) return relevant_retrieved / len(relevant) def mrr(retrieved: List[str], relevant: Set[str]) -> float: """Mean Reciprocal Rank: 1/rank of first relevant doc.""" for i, doc in enumerate(retrieved, 1): if doc in relevant: return 1.0 / i return 0.0 def ndcg_at_k(retrieved: List[str], relevant: Dict[str, float], k: int) -> float: """nDCG@K: normalized discounted cumulative gain with graded relevance.""" def dcg(retrieved_docs: List[str], rel_scores: Dict[str, float], k: int) -> float: score = 0.0 for i, doc in enumerate(retrieved_docs[:k], 1): rel = rel_scores.get(doc, 0.0) score += (2**rel - 1) / np.log2(i + 1) return score ideal_dcg = dcg(sorted(relevant.keys(), key=relevant.get, reverse=True), relevant, k) if ideal_dcg == 0: return 0.0 return dcg(retrieved, relevant, k) / ideal_dcg def hit_rate(retrieved: List[str], relevant: Set[str], k: int) -> float: """Hit Rate@K: whether at least one relevant doc in top-K.""" top_k = set(retrieved[:k]) return 1.0 if top_k & relevant else 0.0 # Batch evaluation def evaluate_retrieval( queries: List[str], retrieval_fn, # query -> List[doc_ids]  ground_truth: Dict[str, Set[str]], # query -> relevant doc_ids  k_values: List[int] = [1, 3, 5, 10, 20] ) -> Dict: """Evaluate retrieval across multiple queries.""" results = {k: {"precision": [], "recall": [], "mrr": [], "hit_rate": []} for k in k_values} for query in queries: retrieved = retrieval_fn(query) relevant = ground_truth.get(query, set()) for k in k_values: results[k]["precision"].append(precision_at_k(retrieved, relevant, k)) results[k]["recall"].append(recall_at_k(retrieved, relevant, k)) results[k]["hit_rate"].append(hit_rate(retrieved, relevant, k)) results[k_values[0]]["mrr"].append(mrr(retrieved, relevant)) # MRR uses full list  # Aggregate  aggregated = {} for k in k_values: aggregated[f"precision@{k}"] = np.mean(results[k]["precision"]) aggregated[f"recall@{k}"] = np.mean(results[k]["recall"]) aggregated[f"hit_rate@{k}"] = np.mean(results[k]["hit_rate"]) aggregated["mrr"] = np.mean(results[k_values[0]]["mrr"]) return aggregated 

# rag_eval/retrieval_ablation.py def ablation_study( base_retriever, query_set: List[str], ground_truth: Dict[str, Set[str]], variants: Dict[str, dict] ) -> Dict: """Compare retrieval variants: chunk size, embedding model, reranker, etc.""" results = {} for name, config in variants.items(): retriever = base_retriever.clone_with_config(config) metrics = evaluate_retrieval(query_set, retriever.retrieve, ground_truth) results[name] = metrics print(f"{name}: {metrics}") return results # Example variants ABLATION_VARIANTS = { "baseline": {"chunk_size": 512, "overlap": 50, "embed_model": "bge-small", "rerank": False}, "larger_chunks": {"chunk_size": 1024, "overlap": 100, "embed_model": "bge-small", "rerank": False}, "better_embedder": {"chunk_size": 512, "overlap": 50, "embed_model": "bge-large", "rerank": False}, "with_reranker": {"chunk_size": 512, "overlap": 50, "embed_model": "bge-small", "rerank": "cross-encoder"}, "full_upgrade": {"chunk_size": 1024, "overlap": 100, "embed_model": "bge-large", "rerank": "cross-encoder"}, } 

# rag_eval/generation_metrics.py from transformers import AutoModelForSequenceClassification, AutoTokenizer import torch class FaithfulnessEvaluator: """NLI-based faithfulness: does answer entail from context?""" def __init__(self, model_name: str = "microsoft/deberta-v3-large-mnli"): self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForSequenceClassification.from_pretrained(model_name) self.model.eval() def score(self, context: str, answer: str) -> float: """Entailment probability: context → answer.""" # Premise: context, Hypothesis: answer  inputs = self.tokenizer( context, answer, return_tensors="pt", truncation=True, max_length=512, padding=True ) with torch.no_grad(): logits = self.model(**inputs).logits probs = torch.softmax(logits, dim=-1) # MNLI: 0=entailment, 1=neutral, 2=contradiction  entailment_prob = probs[0, 0].item() return entailment_prob class HallucinationDetector: """Detect unsupported claims in generated answer.""" def __init__(self): self.nli = FaithfulnessEvaluator() def detect(self, context: str, answer: str, threshold: float = 0.5) -> Dict: """Return hallucination score and flagged spans.""" sentences = self._split_sentences(answer) results = [] for sent in sentences: score = self.nli.score(context, sent) results.append({ "sentence": sent, "faithfulness": score, "hallucinated": score < threshold }) overall = np.mean([r["faithfulness"] for r in results]) hallucinated_spans = [r["sentence"] for r in results if r["hallucinated"]] return { "overall_faithfulness": overall, "hallucination_rate": len(hallucinated_spans) / len(sentences), "flagged_spans": hallucinated_spans, "per_sentence": results } 

# rag_eval/relevance_metrics.py from sentence_transformers import SentenceTransformer, util class RelevanceEvaluator: """Semantic similarity between query and answer.""" def __init__(self, model_name: str = "sentence-transformers/all-MiniLM-L6-v2"): self.model = SentenceTransformer(model_name) def query_answer_relevance(self, query: str, answer: str) -> float: """Cosine similarity between query and answer embeddings.""" q_emb = self.model.encode(query, convert_to_tensor=True) a_emb = self.model.encode(answer, convert_to_tensor=True) return util.cos_sim(q_emb, a_emb).item() def context_answer_relevance(self, context: str, answer: str) -> float: """How well answer addresses context.""" c_emb = self.model.encode(context, convert_to_tensor=True) a_emb = self.model.encode(answer, convert_to_tensor=True) return util.cos_sim(c_emb, a_emb).item() class AnswerCorrectnessEvaluator: """LLM-as-judge for answer correctness.""" def __init__(self, judge_model: str = "gpt-4o-mini"): self.judge_model = judge_model # Use OpenAI API or local model  def evaluate( self, query: str, ground_truth: str, generated: str, context: str = None ) -> Dict: """Score 1-5 on correctness, completeness, conciseness.""" prompt = f"""Rate the generated answer (1-5) on: 1. Correctness: factually accurate vs ground truth 2. Completeness: covers all key points from ground truth 3. Conciseness: no fluff, direct answer Query: {query} Ground Truth: {ground_truth} Generated: {generated} Context: {context or 'N/A'} Return JSON: {{"correctness": 1-5, "completeness": 1-5, "conciseness": 1-5, "reasoning": "..."}}""" # Call judge model  return self._call_judge(prompt) 

# rag_eval/ragas_integration.py from ragas import evaluate from ragas.metrics import ( faithfulness, answer_relevancy, context_precision, context_recall, context_relevancy, answer_correctness, answer_similarity, ) from datasets import Dataset def evaluate_rag_with_ragas( questions: List[str], answers: List[str], contexts: List[List[str]], ground_truths: List[str] = None ) -> Dict: """Evaluate RAG using Ragas metrics.""" data = { "question": questions, "answer": answers, "contexts": contexts, } if ground_truths: data["ground_truth"] = ground_truths dataset = Dataset.from_dict(data) # Select metrics based on available data  metrics = [ faithfulness, answer_relevancy, context_precision, context_recall, ] if ground_truths: metrics.extend([answer_correctness, answer_similarity]) result = evaluate(dataset, metrics=metrics) return result.to_pandas().to_dict(orient="records")[0] # Ragas metric explanations RAGAS_METRICS = { "faithfulness": "Answer grounded in context? (0-1, higher better)", "answer_relevancy": "Answer relevant to question? (0-1)", "context_precision": "Retrieved contexts ranked by relevance? (0-1)", "context_recall": "All needed info retrieved? (0-1)", "context_relevancy": "Context relevant to question? (0-1)", "answer_correctness": "Answer matches ground truth? (0-1, needs GT)", "answer_similarity": "Semantic similarity to ground truth? (0-1, needs GT)", } 

# rag_eval/deepeval_integration.py from deepeval import evaluate from deepeval.metrics import ( AnswerRelevancyMetric, FaithfulnessMetric, ContextualPrecisionMetric, ContextualRecallMetric, ContextualRelevancyMetricSource：https://dev.to/ipsphil_digital/rag-evaluation-monitoring-metrics-frameworks-production-observability-4f48
My favorites