RAG Evaluation in Production: Building Continuous Quality Monitoring

import json
import os
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional

from datasets import Dataset
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    answer_correctness,
)
from ragas.llms import LangchainLLMWrapper
from ragas.embeddings import LangchainEmbeddingsWrapper
from langchain_openai import ChatOpenAI, OpenAIEmbeddings


# --- Configuration ---
@dataclass
class EvalThresholds:
    faithfulness: float = 0.85
    answer_relevancy: float = 0.80
    context_precision: float = 0.75
    context_recall: float = 0.75
    answer_correctness: float = 0.70


@dataclass
class EvalResult:
    passed: bool
    scores: dict[str, float]
    failures: list[str]
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    dataset_size: int = 0


# --- RAG pipeline stub (replace with your actual pipeline) ---
class ProductionRAGPipeline:
    """Stub for your production RAG pipeline."""

    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm

    def run(self, query: str) -> dict:
        """Returns {answer, contexts} for a given query."""
        # In production: call your actual retriever and LLM
        docs = self.retriever.get_relevant_documents(query)
        contexts = [d.page_content for d in docs]
        # LLM call with retrieved context
        context_str = "\n\n".join(contexts)
        prompt = f"Context:\n{context_str}\n\nQuestion: {query}\n\nAnswer:"
        response = self.llm.invoke(prompt)
        return {"answer": response.content, "contexts": contexts}


# --- Golden dataset loader ---
def load_golden_dataset(path: str) -> list[dict]:
    """
    Golden dataset format:
    [{"question": str, "ground_truth": str, "reference_contexts": [str]}]
    """
    with open(path) as f:
        return json.load(f)


# --- Evaluation runner ---
class RAGEvaluator:
    def __init__(
        self,
        rag_pipeline: ProductionRAGPipeline,
        thresholds: Optional[EvalThresholds] = None,
    ):
        self.pipeline = rag_pipeline
        self.thresholds = thresholds or EvalThresholds()

        eval_llm = LangchainLLMWrapper(ChatOpenAI(model="gpt-4o", temperature=0))
        eval_embeddings = LangchainEmbeddingsWrapper(
            OpenAIEmbeddings(model="text-embedding-3-small")
        )
        self.metrics = [
            faithfulness.faithfulness,
            answer_relevancy.answer_relevancy,
            context_precision.context_precision,
            context_recall.context_recall,
            answer_correctness.answer_correctness,
        ]
        for m in self.metrics:
            m.llm = eval_llm
            if hasattr(m, "embeddings"):
                m.embeddings = eval_embeddings

    def _build_eval_dataset(self, golden: list[dict]) -> Dataset:
        questions, answers, contexts, ground_truths = [], [], [], []
        for item in golden:
            result = self.pipeline.run(item["question"])
            questions.append(item["question"])
            answers.append(result["answer"])
            contexts.append(result["contexts"])
            ground_truths.append(item["ground_truth"])
        return Dataset.from_dict({
            "question": questions,
            "answer": answers,
            "contexts": contexts,
            "ground_truth": ground_truths,
        })

    def run_evaluation(self, golden_dataset_path: str) -> EvalResult:
        golden = load_golden_dataset(golden_dataset_path)
        print(f"Running evaluation on {len(golden)} golden samples...")

        eval_dataset = self._build_eval_dataset(golden)
        results = evaluate(eval_dataset, metrics=self.metrics)
        scores = {
            "faithfulness": float(results["faithfulness"]),
            "answer_relevancy": float(results["answer_relevancy"]),
            "context_precision": float(results["context_precision"]),
            "context_recall": float(results["context_recall"]),
            "answer_correctness": float(results["answer_correctness"]),
        }

        failures = []
        thresholds = vars(self.thresholds)
        for metric, score in scores.items():
            threshold = thresholds.get(metric, 0.0)
            if score < threshold:
                failures.append(f"{metric}: {score:.3f} < threshold {threshold:.3f}")

        result = EvalResult(
            passed=len(failures) == 0,
            scores=scores,
            failures=failures,
            dataset_size=len(golden),
        )

        self._write_report(result)
        return result

    def _write_report(self, result: EvalResult) -> None:
        report_path = f"eval_report_{result.timestamp[:10]}.json"
        with open(report_path, "w") as f:
            json.dump({
                "passed": result.passed,
                "timestamp": result.timestamp,
                "dataset_size": result.dataset_size,
                "scores": result.scores,
                "failures": result.failures,
                "thresholds": vars(self.thresholds),
            }, f, indent=2)
        status = "PASSED" if result.passed else "FAILED"
        print(f"\nEvaluation {status}")
        for metric, score in result.scores.items():
            flag = " *** BELOW THRESHOLD" if any(metric in f for f in result.failures) else ""
            print(f"  {metric}: {score:.3f}{flag}")
        if result.failures:
            print(f"\nFailed metrics: {result.failures}")