rag-observability-evals

RAG Observability and Evaluations

Run retrieval-augmented generation like a measurable production system, not a black box.

When to Use This Skill

• Deploying a RAG system to production and need quality monitoring
• Setting up automated evaluation pipelines for retrieval and generation
• Debugging hallucination or relevance regressions
• Building dashboards for RAG-specific golden signals
• Establishing quality gates for RAG pipeline changes

Prerequisites

• RAG pipeline with instrumented retrieval and generation stages
• Python 3.10+ with evaluation libraries (ragas, langchain, openai)
• Prometheus endpoint for custom metrics export
• Benchmark dataset with gold-standard question/answer/source triples
• OpenTelemetry SDK integrated into the RAG service

What to Measure

Retrieval Quality

• Recall@k and MRR for top-k chunks
• Citation coverage and source freshness
• Embedding drift and index staleness

Generation Quality

• Groundedness score (answer supported by retrieved context)
• Hallucination rate by route/use case
• Instruction adherence and format validity

Reliability and Cost

• p50/p95 latency split by retrieval vs generation
• Token usage per stage
• Cache hit rate and cost per successful answer

RAGAS Evaluation Script

# rag_eval.py
"""Evaluate RAG pipeline quality using RAGAS metrics."""
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    context_entity_recall,
    answer_similarity,
)
from datasets import Dataset
import json
import sys

def load_eval_dataset(path: str) -> Dataset:
    """Load evaluation dataset with required columns."""
    with open(path) as f:
        data = json.load(f)

    return Dataset.from_dict({
        "question": [d["question"] for d in data],
        "answer": [d["generated_answer"] for d in data],
        "contexts": [d["retrieved_contexts"] for d in data],
        "ground_truth": [d["reference_answer"] for d in data],
    })

def run_evaluation(dataset_path: str, output_path: str):
    """Run full RAGAS evaluation suite."""
    dataset = load_eval_dataset(dataset_path)

    metrics = [
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
        context_entity_recall,
        answer_similarity,
    ]

    results = evaluate(dataset, metrics=metrics)

    # Print summary
    print("=== RAG Evaluation Results ===")
    for metric_name, score in results.items():
        print(f"  {metric_name}: {score:.4f}")

    # Save detailed results
    with open(output_path, "w") as f:
        json.dump({
            "summary": {k: float(v) for k, v in results.items()},
            "dataset_size": len(dataset),
        }, f, indent=2)

    return results

if __name__ == "__main__":
    run_evaluation(sys.argv[1], sys.argv[2])