RAG Evaluation & Monitoring: Metrics, Frameworks & Production Observability
Comprehensive guide to evaluating and monitoring Retrieval-Augmented Generation systems in production — from offline benchmarks to real-time observability.
Introduction
RAG systems combine retrieval + generation. Evaluation must cover both components plus the end-to-end pipeline. This guide covers metrics, frameworks, and production monitoring.
RAG Evaluation Dimensions
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)
Part 1: Retrieval Metrics
Offline Retrieval Evaluation
# 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
Retrieval Ablation Studies
# 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"},
}
Part 2: Generation Metrics
Faithfulness (Groundedness)
# 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
}
Relevance & Answer Quality
# 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)
Part 3: End-to-End Frameworks
Ragas — Reference-Free Evaluation
# 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)",
}
DeepEval — Unit Testing for LLMs
# rag_eval/deepeval_integration.py
from deepeval import evaluate
from deepeval.metrics import (
AnswerRelevancyMetric,
FaithfulnessMetric,
ContextualPrecisionMetric,
ContextualRecallMetric,
ContextualRelevancyMetric,
HallucinationMetric,
ToxicityMetric,
BiasMetric,
)
from deepeval.test_case import LLMTestCase
def evaluate_rag_with_deepeval(
test_cases: List[LLMTestCase],
threshold: float = 0.7
) -> Dict:
"""Evaluate with DeepEval's comprehensive metrics."""
metrics = [
AnswerRelevancyMetric(threshold=threshold),
FaithfulnessMetric(threshold=threshold),
ContextualPrecisionMetric(threshold=threshold),
ContextualRecallMetric(threshold=threshold),
ContextualRelevancyMetric(threshold=threshold),
HallucinationMetric(threshold=threshold),
ToxicityMetric(threshold=0.5),
]
results = evaluate(test_cases, metrics)
return {
"passed": all(r.success for r in results),
"scores": {m.__class__.__name__: r.score for r in results for m in r.metrics},
"details": results
}
# Test case creation
def create_test_case(
query: str,
answer: str,
retrieval_context: List[str],
expected_output: str = None
) -> LLMTestCase:
return LLMTestCase(
input=query,
actual_output=answer,
retrieval_context=retrieval_context,
expected_output=expected_output,
)
TruLens — Instrumentation & Feedback
# rag_eval/trulens_integration.py
from trulens_eval import Tru, Feedback, OpenAI as fOpenAI
from trulens_eval.feedback import Groundedness
import numpy as np
def setup_trulens():
"""Configure TruLens feedback functions."""
provider = fOpenAI()
# Groundedness: answer supported by context
grounded = Groundedness(groundedness_provider=provider)
f_groundedness = Feedback(
grounded.groundedness_measure_with_cot_reasons,
name="Groundedness"
).on_input_output()
# Answer relevance
f_answer_relevance = Feedback(
provider.relevance,
name="Answer Relevance"
).on_input_output()
# Context relevance
f_context_relevance = Feedback(
provider.context_relevance,
name="Context Relevance"
).on_input()
return [f_groundedness, f_answer_relevance, f_context_relevance]
def instrument_rag_chain(rag_chain, feedback_funcs):
"""Wrap RAG chain for TruLens logging."""
from trulens_eval import TruChain
tru_chain = TruChain(
rag_chain,
app_id="rag_qa",
feedbacks=feedback_funcs
)
return tru_chain
Part 4: Production Monitoring
Real-Time Metrics Pipeline
# rag_monitoring/production_metrics.py
from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime
import time
import asyncio
from prometheus_client import Counter, Histogram, Gauge, Summary
# Prometheus metrics
RAG_REQUESTS_TOTAL = Counter(
"rag_requests_total",
"Total RAG requests",
["status"] # success, error, timeout
)
RAG_LATENCY = Histogram(
"rag_latency_seconds",
"RAG request latency",
buckets=[0.5, 1, 2, 5, 10, 30, 60]
)
RETRIEVAL_LATENCY = Histogram(
"rag_retrieval_latency_seconds",
"Retrieval step latency"
)
GENERATION_LATENCY = Histogram(
"rag_generation_latency_seconds",
"Generation step latency"
)
RAG_TOKENS = Counter(
"rag_tokens_total",
"Total tokens used",
["type"] # prompt, completion
)
RAG_COST = Counter(
"rag_cost_usd_total",
"Total cost in USD"
)
ACTIVE_REQUESTS = Gauge(
"rag_active_requests",
"Currently processing requests"
)
QUALITY_SCORE = Histogram(
"rag_quality_score",
"Quality scores from evaluators",
["metric"], # faithfulness, relevance, etc.
buckets=[0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
)
@dataclass
class RAGRequestLog:
request_id: str
timestamp: datetime
query: str
retrieved_docs: List[str]
answer: str
latency_ms: float
retrieval_latency_ms: float
generation_latency_ms: float
prompt_tokens: int
completion_tokens: int
cost_usd: float
quality_scores: dict = field(default_factory=dict)
error: Optional[str] = None
class ProductionMonitor:
"""Production RAG monitoring with alerting."""
def __init__(
self,
alert_webhook: str = None,
quality_threshold: float = 0.6,
latency_p99_threshold: float = 30.0,
error_rate_threshold: float = 0.05
):
self.alert_webhook = alert_webhook
self.quality_threshold = quality_threshold
self.latency_threshold = latency_p99_threshold
self.error_rate_threshold = error_rate_threshold
self.request_logs: List[RAGRequestLog] = []
self.window_logs: List[RAGRequestLog] = []
async def log_request(self, log: RAGRequestLog):
"""Log request and update metrics."""
self.request_logs.append(log)
self.window_logs.append(log)
# Update Prometheus
RAG_REQUESTS_TOTAL.labels(
status="error" if log.error else "success"
).inc()
RAG_LATENCY.observe(log.latency_ms / 1000)
RETRIEVAL_LATENCY.observe(log.retrieval_latency_ms / 1000)
GENERATION_LATENCY.observe(log.generation_latency_ms / 1000)
RAG_TOKENS.labels(type="prompt").inc(log.prompt_tokens)
RAG_TOKENS.labels(type="completion").inc(log.completion_tokens)
RAG_COST.inc(log.cost_usd)
for metric, score in log.quality_scores.items():
QUALITY_SCORE.labels(metric=metric).observe(score)
# Check alerts
await self._check_alerts(log)
async def _check_alerts(self, log: RAGRequestLog):
"""Check and fire alerts."""
alerts = []
# Quality alert
for metric, score in log.quality_scores.items():
if score < self.quality_threshold:
alerts.append(f"Quality alert: {metric}={score:.2f} < {self.quality_threshold}")
# Latency alert
if log.latency_ms > self.latency_threshold * 1000:
alerts.append(f"Latency alert: {log.latency_ms}ms > {self.latency_threshold}s")
# Error rate alert (rolling window)
await self._check_error_rate()
for alert in alerts:
await self._send_alert(alert)
async def _check_error_rate(self):
"""Check error rate in last 100 requests."""
if len(self.window_logs) >= 100:
recent = self.window_logs[-100:]
errors = sum(1 for r in recent if r.error)
rate = errors / 100
if rate > self.error_rate_threshold:
await self._send_alert(f"Error rate alert: {rate:.1%} > {self.error_rate_threshold:.1%}")
async def _send_alert(self, message: str):
"""Send alert to webhook."""
if self.alert_webhook:
# Implement webhook call
print(f"ALERT: {message}")
def get_dashboard_data(self, hours: int = 24) -> Dict:
"""Get data for Grafana dashboard."""
cutoff = datetime.now() - timedelta(hours=hours)
recent = [r for r in self.request_logs if r.timestamp > cutoff]
if not recent:
return {}
return {
"total_requests": len(recent),
"success_rate": sum(1 for r in recent if not r.error) / len(recent),
"avg_latency_ms": np.mean([r.latency_ms for r in recent]),
"p99_latency_ms": np.percentile([r.latency_ms for r in recent], 99),
"avg_cost_per_request": np.mean([r.cost_usd for r in recent]),
"total_cost": sum(r.cost_usd for r in recent),
"avg_quality": {
metric: np.mean([r.quality_scores.get(metric, 0) for r in recent])
for metric in ["faithfulness", "relevance", "correctness"]
},
}
Distributed Tracing
# rag_monitoring/tracing.py
from opentelemetry import trace
from opentelemetry.exporter.jaeger import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
def setup_tracing(service_name: str = "rag-service"):
"""Setup OpenTelemetry tracing."""
trace.set_tracer_provider(TracerProvider())
jaeger_exporter = JaegerExporter(
agent_host_name="jaeger",
agent_port=6831,
)
trace.get_tracer_provider().add_span_processor(
BatchSpanProcessor(jaeger_exporter)
)
return trace.get_tracer(__name__)
class TracedRAGChain:
"""RAG chain with full distributed tracing."""
def __init__(self, retriever, generator, tracer):
self.retriever = retriever
self.generator = generator
self.tracer = tracer
async def answer(self, query: str) -> dict:
with self.tracer.start_as_current_span("rag_query") as span:
span.set_attribute("query", query)
# Retrieval span
with self.tracer.start_as_current_span("retrieval") as retr_span:
start = time.time()
docs = await self.retriever.retrieve(query)
retr_span.set_attribute("num_docs", len(docs))
retr_span.set_attribute("latency_ms", (time.time() - start) * 1000)
# Generation span
with self.tracer.start_as_current_span("generation") as gen_span:
start = time.time()
answer = await self.generator.generate(query, docs)
gen_span.set_attribute("answer_length", len(answer))
gen_span.set_attribute("latency_ms", (time.time() - start) * 1000)
span.set_attribute("total_latency_ms", (time.time() - start) * 1000)
return {"answer": answer, "sources": docs}
Part 5: Automated Evaluation Pipeline
CI/CD Integration
# rag_eval/ci_pipeline.py
import pytest
from ragas import evaluate
from datasets import Dataset
class RAGEvalTestSuite:
"""Pytest-style RAG evaluation tests."""
@pytest.fixture(scope="class")
def test_data(self):
return [
{
"question": "What is LoRA?",
"ground_truth": "LoRA is Low-Rank Adaptation for efficient fine-tuning...",
"contexts": ["LoRA freezes base weights and adds low-rank adapters..."],
},
# ... more test cases
]
def test_faithfulness_threshold(self, rag_chain, test_data):
"""Faithfulness must exceed threshold."""
results = []
for case in test_data:
answer = rag_chain.answer(case["question"])
score = self._compute_faithfulness(case["contexts"][0], answer)
results.append(score)
assert score >= 0.7, f"Faithfulness {score:.2f} below threshold for: {case['question']}"
def test_retrieval_precision(self, rag_chain, test_data):
"""Retrieval precision@5 must exceed threshold."""
for case in test_data:
retrieved = rag_chain.retrieve(case["question"])
relevant = self._get_relevant_docs(case["question"])
p5 = precision_at_k(retrieved, relevant, 5)
assert p5 >= 0.6, f"Precision@5 {p5:.2f} below threshold for: {case['question']}"
def test_latency_sla(self, rag_chain, test_data):
"""P99 latency must be under 10 seconds."""
latencies = []
for case in test_data:
start = time.time()
rag_chain.answer(case["question"])
latencies.append(time.time() - start)
p99 = np.percentile(latencies, 99)
assert p99 <= 10.0, f"P99 latency {p99:.1f}s exceeds 10s SLA"
def test_cost_per_query(self, rag_chain, test_data):
"""Average cost per query under $0.01."""
costs = [rag_chain.get_last_cost() for _ in test_data]
avg_cost = np.mean(costs)
assert avg_cost <= 0.01, f"Avg cost ${avg_cost:.4f} exceeds $0.01"
# Regression detection
def detect_regression(
current_scores: Dict,
baseline_scores: Dict,
threshold: float = 0.05
) -> List[str]:
"""Detect metric regressions vs baseline."""
regressions = []
for metric, current in current_scores.items():
baseline = baseline_scores.get(metric)
if baseline and current < baseline - threshold:
regressions.append(
f"REGRESSION: {metric} dropped from {baseline:.3f} to {current:.3f} "
f"(delta: {current-baseline:.3f})"
)
return regressions
A/B Testing Framework
# rag_eval/ab_testing.py
import random
from dataclasses import dataclass
from typing: Callable, Dict, List
import numpy as np
from scipy import stats
@dataclass
class ABTestResult:
variant_a: str
variant_b: str
metric: str
a_mean: float
b_mean: float
p_value: float
significant: bool
winner: str
class RAGABTester:
"""A/B test RAG variants in production."""
def __init__(self, traffic_split: float = 0.5):
self.traffic_split = traffic_split
self.variants: Dict[str, Callable] = {}
self.results: Dict[str, List[float]] = {}
def register_variant(self, name: str, chain_factory: Callable):
self.variants[name] = chain_factory
self.results[name] = []
def route_request(self, query: str) -> tuple:
"""Route to variant based on traffic split."""
variant_name = "A" if random.random() < self.traffic_split else "B"
chain = self.variants[variant_name]()
return variant_name, chain
def record_result(self, variant: str, score: float):
self.results[variant].append(score)
def analyze(self, metric: str = "quality") -> ABTestResult:
"""Statistical significance test."""
a_scores = self.results["A"]
b_scores = self.results["B"]
if len(a_scores) < 30 or len(b_scores) < 30:
return ABTestResult(
"A", "B", metric,
np.mean(a_scores), np.mean(b_scores),
1.0, False, "insufficient_data"
)
# Welch's t-test (unequal variance)
t_stat, p_value = stats.ttest_ind(a_scores, b_scores, equal_var=False)
a_mean, b_mean = np.mean(a_scores), np.mean(b_scores)
winner = "A" if a_mean > b_mean else "B"
return ABTestResult(
"A", "B", metric,
a_mean, b_mean,
p_value, p_value < 0.05, winner
)
Part 6: Data Quality & Drift Detection
Embedding Drift Monitoring
# rag_monitoring/drift_detection.py
from sklearn.decomposition import PCA
from sklearn.manifold import TSNE
import numpy as np
from scipy.spatial.distance import jensenshannon
class EmbeddingDriftDetector:
"""Detect drift in query/document embeddings."""
def __init__(self, reference_embeddings: np.ndarray, window_size: int = 1000):
self.reference = reference_embeddings
self.window_size = window_size
self.recent_embeddings: List[np.ndarray] = []
# Fit PCA on reference for visualization
self.pca = PCA(n_components=2)
self.pca.fit(reference_embeddings)
def add_embeddings(self, embeddings: np.ndarray):
self.recent_embeddings.append(embeddings)
if len(self.recent_embeddings) > self.window_size:
self.recent_embeddings.pop(0)
def compute_drift_score(self) -> Dict:
"""Compute multiple drift metrics."""
if not self.recent_embeddings:
return {"drift_detected": False}
recent = np.vstack(self.recent_embeddings)
# 1. Mean shift (L2 distance between centroids)
ref_centroid = self.reference.mean(axis=0)
recent_centroid = recent.mean(axis=0)
mean_shift = np.linalg.norm(ref_centroid - recent_centroid)
# 2. Covariance shift (Frobenius norm)
ref_cov = np.cov(self.reference.T)
recent_cov = np.cov(recent.T)
cov_shift = np.linalg.norm(ref_cov - recent_cov, 'fro')
# 3. JS divergence on PCA projections
ref_pca = self.pca.transform(self.reference)
recent_pca = self.pca.transform(recent)
# 2D histogram
hist_range = [[ref_pca[:,0].min(), ref_pca[:,0].max()],
[ref_pca[:,1].min(), ref_pca[:,1].max()]]
ref_hist, _, _ = np.histogram2d(ref_pca[:,0], ref_pca[:,1], bins=20, range=hist_range, density=True)
recent_hist, _, _ = np.histogram2d(recent_pca[:,0], recent_pca[:,1], bins=20, range=hist_range, density=True)
# Flatten and add epsilon
ref_flat = ref_hist.flatten() + 1e-10
recent_flat = recent_hist.flatten() + 1e-10
js_div = jensenshannon(ref_flat, recent_flat)
return {
"drift_detected": js_div > 0.1 or mean_shift > 0.5,
"mean_shift": float(mean_shift),
"cov_shift": float(cov_shift),
"js_divergence": float(js_div),
"ref_centroid": ref_centroid.tolist(),
"recent_centroid": recent_centroid.tolist(),
}
Data Quality Checks
# rag_monitoring/data_quality.py
def validate_rag_inputs(
query: str,
retrieved_docs: List[str],
answer: str
) -> Dict:
"""Validate inputs for quality issues."""
issues = []
# Query checks
if len(query.strip()) < 3:
issues.append("Query too short")
if len(query) > 2000:
issues.append("Query extremely long")
# Retrieval checks
if not retrieved_docs:
issues.append("No documents retrieved")
else:
avg_doc_len = np.mean([len(d) for d in retrieved_docs])
if avg_doc_len < 50:
issues.append("Retrieved docs very short")
if avg_doc_len > 5000:
issues.append("Retrieved docs very long")
# Check for duplicate docs
unique_docs = set(retrieved_docs)
if len(unique_docs) < len(retrieved_docs) * 0.8:
issues.append("High duplicate rate in retrieval")
# Answer checks
if not answer.strip():
issues.append("Empty answer")
if len(answer) < 10:
issues.append("Answer too short")
if answer.count("I don't know") > 0:
issues.append("Model abstained")
# Repetition check
words = answer.split()
if len(words) > 10:
unique_ratio = len(set(words)) / len(words)
if unique_ratio < 0.3:
issues.append("High repetition in answer")
return {
"valid": len(issues) == 0,
"issues": issues,
"severity": "high" if any("Empty" in i or "No documents" in i for i in issues) else "low"
}
Part 7: Dashboards & Alerting
Grafana Dashboard JSON
# rag_monitoring/grafana_dashboard.json
{
"dashboard": {
"title": "RAG Production Monitoring",
"panels": [
{
"title": "Request Rate",
"targets": [{"expr": "rate(rag_requests_total[5m])"}]
},
{
"title": "Error Rate",
"targets": [{"expr": "rate(rag_requests_total{status=\"error\"}[5m]) / rate(rag_requests_total[5m])"}]
},
{
"title": "Latency P50/P95/P99",
"targets": [
{"expr": "histogram_quantile(0.50, rag_latency_seconds)"},
{"expr": "histogram_quantile(0.95, rag_latency_seconds)"},
{"expr": "histogram_quantile(0.99, rag_latency_seconds)"}
]
},
{
"title": "Quality Scores",
"targets": [
{"expr": "avg by (metric) (rag_quality_score)"}
]
},
{
"title": "Cost per Request",
"targets": [{"expr": "rate(rag_cost_usd_total[1h]) / rate(rag_requests_total[1h])"}]
},
{
"title": "Tokens per Request",
"targets": [
{"expr": "rate(rag_tokens_total{type=\"prompt\"}[5m]) / rate(rag_requests_total[5m])"},
{"expr": "rate(rag_tokens_total{type=\"completion\"}[5m]) / rate(rag_requests_total[5m])"}
]
}
]
}
}
Alert Rules
# rag_monitoring/alerts.yaml
groups:
- name: rag_alerts
rules:
- alert: RAGHighErrorRate
expr: rate(rag_requests_total{status="error"}[5m]) / rate(rag_requests_total[5m]) > 0.05
for: 5m
labels:
severity: critical
annotations:
summary: "RAG error rate > 5%"
- alert: RAGHighLatency
expr: histogram_quantile(0.99, rag_latency_seconds) > 30
for: 5m
labels:
severity: warning
annotations:
summary: "RAG P99 latency > 30s"
- alert: RAGLowFaithfulness
expr: avg(rag_quality_score{metric="faithfulness"}) < 0.6
for: 10m
labels:
severity: warning
annotations:
summary: "RAG faithfulness dropped below 0.6"
- alert: RAGHighCost
expr: rate(rag_cost_usd_total[1h]) > 10
for: 15m
labels:
severity: warning
annotations:
summary: "RAG cost > $10/hour"
Part 8: Continuous Improvement Loop
# rag_eval/improvement_loop.py
class RAGImprovementLoop:
"""Automated RAG improvement via evaluation feedback."""
def __init__(self, rag_system, evaluator, optimizer):
self.rag = rag_system
self.evaluator = evaluator
self.optimizer = optimizer
self.history = []
async def run_iteration(self, eval_queries: List[str]):
"""One improvement cycle."""
# 1. Evaluate current system
current_metrics = await self.evaluator.evaluate(self.rag, eval_queries)
# 2. Identify weak spots
weak_spots = self._identify_weaknesses(current_metrics)
# 3. Generate optimization hypotheses
hypotheses = self.optimizer.generate_hypotheses(weak_spots)
# 4. Test hypotheses (A/B or shadow)
best_variant = await self._test_hypotheses(hypotheses, eval_queries)
# 5. Deploy if significant improvement
if best_variant and best_variant.improvement > 0.02:
await self._deploy_variant(best_variant)
self.history.append({
"iteration": len(self.history),
"metrics": current_metrics,
"deployed": best_variant.name,
"improvement": best_variant.improvement
})
return current_metrics
def _identify_weaknesses(self, metrics: Dict) -> List[str]:
"""Identify which metrics need improvement."""
thresholds = {
"faithfulness": 0.8,
"answer_relevancy": 0.75,
"context_precision": 0.7,
"context_recall": 0.7,
}
return [
metric for metric, threshold in thresholds.items()
if metrics.get(metric, 0) < threshold
]
Part 9: Cost Optimization
Token & Cost Tracking
# rag_monitoring/cost_optimization.py
class CostOptimizer:
"""Optimize RAG costs while maintaining quality."""
def __init__(self, pricing: Dict):
self.pricing = pricing # {"gpt-4o": {"input": 5, "output": 15}, ...}
def estimate_cost(self, prompt_tokens: int, completion_tokens: int, model: str) -> float:
"""Estimate cost in USD."""
rates = self.pricing.get(model, {"input": 0, "output": 0})
return (prompt_tokens * rates["input"] + completion_tokens * rates["output"]) / 1_000_000
def optimize_model_selection(
self,
query: str,
complexity_score: float, # 0-1
quality_requirement: float # 0-1
) -> str:
"""Route to cheapest model meeting quality."""
models = [
("gpt-4o-mini", 0.7, 0.15), # quality, cost_factor
("gpt-4o", 0.9, 1.0),
("gpt-4-turbo", 0.95, 2.0),
]
for model, quality, cost_factor in models:
if quality >= quality_requirement:
# Check if we can use cheaper for simple queries
if complexity_score < 0.3 and model != "gpt-4o-mini":
continue
return model
return "gpt-4o" # fallback
def optimize_retrieval(self, retrieval_config: Dict) -> Dict:
"""Find optimal retrieval config for cost/quality."""
# Fewer docs = lower prompt tokens = lower cost
# But too few = lower quality
# Use Bayesian optimization
pass
Resources
- Ragas Documentation
- DeepEval Documentation
- TruLens Documentation
- LangSmith Evaluation
- RAG Evaluation Best Practices
- ARES: Automated RAG Evaluation
RAG evaluation is not one-time — it's continuous. Automate metrics, alert on regressions, and close the loop with A/B testing. The best RAG system is the one you can measure and improve.
Top comments (0)