rag-observability-evals

name	rag-observability-evals
description	Monitor and evaluate RAG systems with retrieval quality metrics, groundedness checks, hallucination detection, and continuous regression testing.
license	MIT
metadata	{"author":"devops-skills","version":"1.0"}

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
    ()
     metric_name, score  results.items():
        ()

    
     (output_path, )  f:
        json.dump({
            : {k: (v)  k, v  results.items()},
            : (dataset),
        }, f, indent=)

     results

 __name__ == :
    run_evaluation(sys.argv[], sys.argv[])

Groundedness Scoring

# groundedness.py
"""Score whether generated answers are grounded in retrieved context."""
from openai import OpenAI
import json
from typing import List

client = OpenAI()

GROUNDEDNESS_PROMPT = """You are evaluating whether an AI answer is fully grounded
in the provided context documents. Score each claim in the answer.

Context documents:
{contexts}

Answer to evaluate:
{answer}

For each distinct claim in the answer, determine:
1. SUPPORTED - the claim is directly supported by the context
2. PARTIALLY_SUPPORTED - the claim is partially supported
3. NOT_SUPPORTED - the claim has no support in the context

Return JSON:
{{
  "claims": [
    {{"claim": "...", "verdict": "SUPPORTED|PARTIALLY_SUPPORTED|NOT_SUPPORTED", "evidence": "..."}}
  ],
  "groundedness_score": <float 0-1>,
  "unsupported_claims": ["..."]
}}
"""

def score_groundedness(answer: str, contexts: List[str]) -> dict:
    """Score groundedness of a single answer against its contexts."""
    context_text = "\n---\n".join(
        f"[Document {i+1}]: {c}" for i, c in enumerate(contexts)
    )

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": GROUNDEDNESS_PROMPT.format(
                contexts=context_text, answer=answer
            ),
        }],
        response_format={"type": "json_object"},
        temperature=0,
    )

    return json.loads(response.choices[0].message.content)

def () -> :
    
    scores = []
    unsupported_count = 
    total_claims = 

     item  eval_data:
        result = score_groundedness(
            item[],
            item[],
        )
        scores.append(result[])
        unsupported_count += (result[])
        total_claims += (result[])

    avg_score = (scores) / (scores)  scores  
     {
        : avg_score,
        : total_claims,
        : unsupported_count,
        : unsupported_count / total_claims  total_claims  ,
        : (eval_data),
    }

Retrieval Quality Metrics

# retrieval_metrics.py
"""Compute retrieval quality metrics for RAG evaluation."""
from typing import List, Set
import numpy as np

def recall_at_k(
    retrieved_ids: List[str],
    relevant_ids: Set[str],
    k: int
) -> float:
    """Compute Recall@K for a single query."""
    top_k = set(retrieved_ids[:k])
    if not relevant_ids:
        return 0.0
    return len(top_k & relevant_ids) / len(relevant_ids)

def mrr(
    retrieved_ids: List[str],
    relevant_ids: Set[str]
) -> float:
    """Compute Mean Reciprocal Rank for a single query."""
    for i, doc_id in enumerate(retrieved_ids):
        if doc_id in relevant_ids:
            return 1.0 / (i + 1)
    return 0.0

def ndcg_at_k(
    retrieved_ids: List[str],
    relevant_ids: Set[],
    k: 
) -> :
    
    dcg = 
     i, doc_id  (retrieved_ids[:k]):
         doc_id  relevant_ids:
            dcg +=  / np.log2(i + )

    ideal_dcg = ( / np.log2(i + )  i  (((relevant_ids), k)))
     dcg / ideal_dcg  ideal_dcg >   

 () -> :
    
    results = {}
     k  k_values:
        recalls = [
            recall_at_k(q[], (q[]), k)
             q  queries
        ]
        mrrs = [mrr(q[], (q[]))  q  queries]
        ndcgs = [
            ndcg_at_k(q[], (q[]), k)
             q  queries
        ]
        results[] = np.mean(recalls)
        results[] = np.mean(ndcgs)

    results[] = np.mean(mrrs)
     results

Prometheus Metrics Export

# rag_metrics_exporter.py
"""Export RAG quality metrics to Prometheus."""
from prometheus_client import Histogram, Counter, Gauge, start_http_server
import time

# Latency histograms by stage
RETRIEVAL_LATENCY = Histogram(
    "rag_retrieval_duration_seconds",
    "Time spent in retrieval stage",
    ["index_name", "retriever_type"],
    buckets=[0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0],
)

GENERATION_LATENCY = Histogram(
    "rag_generation_duration_seconds",
    "Time spent in generation stage",
    ["model", "route"],
    buckets=[0.5, 1.0, 2.0, 5.0, 10.0, 30.0],
)

RERANKING_LATENCY = Histogram(
    "rag_reranking_duration_seconds",
    "Time spent in reranking stage",
    ["reranker_model"],
    buckets=[0.05, 0.1, 0.25, 0.5, 1.0],
)

# Quality gauges (updated from offline evals)
GROUNDEDNESS_SCORE = Gauge(
    "rag_groundedness_score",
    "Latest groundedness evaluation score",
    ["route", "model"],
)

FAITHFULNESS_SCORE = Gauge(
    "rag_faithfulness_score",
    "Latest faithfulness evaluation score",
    ["route", "model"],
)

CONTEXT_PRECISION = Gauge(
    ,
    ,
    [, ],
)

RECALL_AT_K = Gauge(
    ,
    ,
    [, ],
)


REQUESTS_TOTAL = Counter(
    ,
    ,
    [, ],
)

HALLUCINATION_DETECTED = Counter(
    ,
    ,
    [, ],
)

FALLBACK_TRIGGERED = Counter(
    ,
    ,
    [, ],
)

TOKENS_USED = Counter(
    ,
    ,
    [, ],
)

CACHE_HITS = Counter(
    ,
    ,
    [],
)


INDEX_STALENESS_SECONDS = Gauge(
    ,
    ,
    [],
)

INDEX_DOCUMENT_COUNT = Gauge(
    ,
    ,
    [],
)

 ():
    
    start_http_server(port)
    ()

Evaluation Pipeline

1. Curate a benchmark set with gold answers and source docs.
2. Run nightly offline evals for every retriever/model configuration.
3. Execute online shadow evals on sampled production traffic.
4. Gate releases on minimum quality + safety + latency thresholds.

# eval-pipeline-cron.yaml
apiVersion: batch/v1
kind: CronJob
metadata:
  name: rag-nightly-eval
  namespace: ai-evals
spec:
  schedule: "0 2 * * *"
  jobTemplate:
    spec:
      template:
        spec:
          containers:
            - name: eval-runner
              image: registry.internal/rag-eval:latest
              command:
                - python
                - -m
                - rag_eval
                - --dataset=/data/benchmark_v3.json
                - --output=/results/nightly-$(date +%Y%m%d).json
                - --push-metrics
                - --fail-on-regression
              env:
                - name: PROMETHEUS_PUSHGATEWAY
                  value: "http://pushgateway:9091"
                - name: MLFLOW_TRACKING_URI
                  value: "http://mlflow:5000"
              volumeMounts:
                  
                   
                  
                   
          
              
              
                 
              
              
                 
           

Alerting Strategy

# rag-alerts.yaml
groups:
  - name: rag-quality-alerts
    rules:
      - alert: GroundednessDropped
        expr: rag_groundedness_score < 0.75
        for: 10m
        labels:
          severity: sev2
        annotations:
          summary: "Groundedness score dropped below 0.75 for {{ $labels.route }}"

      - alert: HallucinationSpike
        expr: |
          rate(rag_hallucination_detected_total[15m])
          / rate(rag_requests_total[15m]) > 0.10
        for: 5m
        labels:
          severity: sev1

      - alert: IndexStale
        expr: rag_index_staleness_seconds > 86400
        for: 5m
        labels:
          severity: sev3
        annotations:
          summary: "Index {{ $labels.index_name }} not updated in 24h"

      - alert: HighFallbackRate
                  
        
           

        
                  
        
           

Practical Guardrails

• Force citations for high-risk domains.
• Return abstain/fallback when confidence is below threshold.
• Re-rank retrieved chunks before final generation.
• Use query rewriting only with strict regression tests.

Incident Triage Checklist

Symptom	Check First	Check Second
Groundedness dropped	Embedding model change?	Chunking/indexing logic change?
Retrieval returning irrelevant docs	Index freshness and document count	Embedding model version mismatch
Latency spike in retrieval	Vector DB connection pool and load	Index size growth beyond threshold
Cost per answer increasing	Token usage per stage breakdown	Cache hit rate decline
Hallucination spike	Model version or temperature change	Context window overflow (truncated docs)

Troubleshooting

Issue	Diagnosis	Resolution
RAGAS eval returns 0 for all metrics	Check dataset format matches expected schema	Ensure contexts are lists, not strings
Groundedness score unreliable	LLM judge inconsistency	Increase judge sample size, set temperature=0
Index staleness alert firing	Ingestion pipeline failure	Check data source connectivity and ingestion logs
Retrieval recall dropping	Embedding drift after model update	Re-index corpus with current embedding model
High latency in generation	Context too large for model	Reduce top-k or add summarization step

Related Skills

• rag-infrastructure - Deploy robust RAG backends
• agent-observability - Instrument requests, traces, and costs
• agent-evals - Build repeatable eval suites
• ai-sre-incident-response - Incident response for quality regressions
• opentelemetry - Distributed tracing for RAG pipelines