Evaluating RAG Systems with RAGAS: Metrics That Matter

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

from evaluation.generate import evaluate_rag_system

# Your evaluation questions and ground truths
questions = [
    "What is the embedding dimension of all-MiniLM-L6-v2?",
    "How does hybrid search improve retrieval quality?",
    "When should you use ColBERT reranking?",
]

ground_truths = [
    "The all-MiniLM-L6-v2 model produces embeddings with 384 dimensions.",
    "Hybrid search combines dense semantic embeddings with sparse keyword matching.",
    "ColBERT reranking is best for precision-critical applications.",
]

# Generate answers and retrieve contexts
answers, contexts = evaluate_rag_system(questions)

# Create RAGAS dataset
eval_dataset = Dataset.from_dict({
    "question": questions,
    "ground_truth": ground_truths,
    "answer": answers,
    "contexts": contexts,
})

# Run evaluation
results = evaluate(
    dataset=eval_dataset,
    metrics=[
        faithfulness,
        answer_relevancy,
        context_precision,
        context_recall,
    ],
)

print(results)

from ragas.metrics import faithfulness

# Example of faithful response
faithful_example = {
    "question": "What is the dimension of MiniLM embeddings?",
    "contexts": [["MiniLM-L6-v2 produces 384-dimensional embeddings."]],
    "answer": "MiniLM embeddings have 384 dimensions.",
}

# Example of unfaithful (hallucinated) response
unfaithful_example = {
    "question": "What is the dimension of MiniLM embeddings?",
    "contexts": [["MiniLM-L6-v2 produces 384-dimensional embeddings."]],
    "answer": "MiniLM embeddings have 768 dimensions and support GPU acceleration.",
}

Faithfulness = (Supported Claims) / (Total Claims)

# Highly relevant answer
relevant = {
    "question": "How do I configure Qdrant for hybrid search?",
    "answer": "To configure Qdrant for hybrid search, create a collection with both dense and sparse vector configs...",
}

# Irrelevant answer (factually correct but doesn't answer the question)
irrelevant = {
    "question": "How do I configure Qdrant for hybrid search?",
    "answer": "Qdrant is a vector database written in Rust that supports HNSW indexing.",
}

def answer_relevancy_score(question: str, answer: str, llm) -> float:
    """Simplified relevancy calculation."""
    # Generate 3 questions that the answer could be responding to
    generated_questions = llm.generate(
        f"Generate 3 questions that this answer responds to:\n{answer}"
    )

    # Compute embeddings
    q_embedding = embed(question)
    gen_embeddings = [embed(q) for q in generated_questions]

    # Average cosine similarity
    similarities = [cosine_sim(q_embedding, ge) for ge in gen_embeddings]
    return sum(similarities) / len(similarities)

# Good precision: relevant docs at top
good_ranking = {
    "question": "What is BM25?",
    "contexts": [
        "BM25 is a ranking function for information retrieval.",  # Relevant ✓
        "BM25 uses term frequency and inverse document frequency.",  # Relevant ✓
        "Vector databases store embeddings.",  # Less relevant
    ],
}

# Poor precision: relevant docs buried
poor_ranking = {
    "question": "What is BM25?",
    "contexts": [
        "Qdrant is a vector database.",  # Not relevant
        "FastEmbed generates embeddings.",  # Not relevant
        "BM25 is a ranking function for information retrieval.",  # Relevant but ranked last!  
    ],
}

# Good recall: all needed info retrieved
good_recall = {
    "question": "Compare BM25 and dense embeddings.",
    "ground_truth": "BM25 uses keyword matching while dense embeddings capture semantic similarity.",
    "contexts": [
        "BM25 is a keyword-based ranking algorithm.",  # ✓ Covers BM25
        "Dense embeddings capture semantic meaning.",  # ✓ Covers dense
    ],
}

# Poor recall: missing information
poor_recall = {
    "question": "Compare BM25 and dense embeddings.",
    "ground_truth": "BM25 uses keyword matching while dense embeddings capture semantic similarity.",
    "contexts": [
        "BM25 is a keyword-based ranking algorithm.",  # ✓ Covers BM25
        # Missing any context about dense embeddings!  
    ],
}

import json
from datetime import datetime
from pathlib import Path

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset


class RAGEvaluator:
    """Automated RAG evaluation pipeline."""

    def __init__(self, output_dir: str = "eval_results"):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(exist_ok=True)
        self.metrics = [
            faithfulness,
            answer_relevancy,
            context_precision,
            context_recall,
        ]

    def run_evaluation(
        self,
        questions: list[str],
        ground_truths: list[str],
        answers: list[str],
        contexts: list[list[str]],
        run_name: str = None,
    ) -> dict:
        """Run full RAGAS evaluation and save results."""
        dataset = Dataset.from_dict({
            "question": questions,
            "ground_truth": ground_truths,
            "answer": answers,
            "contexts": contexts,
        })

        results = evaluate(dataset=dataset, metrics=self.metrics)

        # Save results
        run_name = run_name or datetime.now().strftime("%Y%m%d_%H%M%S")
        self._save_results(results, run_name)

        return results

    def _save_results(self, results: dict, run_name: str) -> None:
        """Persist evaluation results."""
        output_file = self.output_dir / f"{run_name}.json"
        with open(output_file, "w") as f:
            json.dump(
                {
                    "timestamp": datetime.now().isoformat(),
                    "metrics": dict(results),
                },
                f,
                indent=2,
            )

    def compare_runs(self, run1: str, run2: str) -> dict:
        """Compare two evaluation runs."""
        with open(self.output_dir / f"{run1}.json") as f:
            results1 = json.load(f)
        with open(self.output_dir / f"{run2}.json") as f:
            results2 = json.load(f)

        comparison = {}
        for metric in results1["metrics"]:
            diff = results2["metrics"][metric] - results1["metrics"][metric]
            comparison[metric] = {
                "before": results1["metrics"][metric],
                "after": results2["metrics"][metric],
                "change": diff,
                "improved": diff > 0,
            }

        return comparison

from evaluation.pipeline import RAGEvaluator
from evaluation.generate import evaluate_rag_system

evaluator = RAGEvaluator()

# Load your test set
questions = load_test_questions()
ground_truths = load_ground_truths()

# Run your RAG system
answers, contexts = evaluate_rag_system(questions)

# Evaluate
results = evaluator.run_evaluation(
    questions=questions,
    ground_truths=ground_truths,
    answers=answers,
    contexts=contexts,
    run_name="baseline_v1",
)

print(f"Faithfulness: {results['faithfulness']:.3f}")
print(f"Relevancy: {results['answer_relevancy']:.3f}")
print(f"Precision: {results['context_precision']:.3f}")
print(f"Recall: {results['context_recall']:.3f}")

from evaluation.pipeline import RAGEvaluator

evaluator = RAGEvaluator()

# Test current system
current_answers, current_contexts = evaluate_rag_system_v1(questions)
current_results = evaluator.run_evaluation(
    questions, ground_truths, current_answers, current_contexts,
    run_name="current",
)

# Test proposed changes
new_answers, new_contexts = evaluate_rag_system_v2(questions)
new_results = evaluator.run_evaluation(
    questions, ground_truths, new_answers, new_contexts,
    run_name="proposed",
)

# Compare
comparison = evaluator.compare_runs("current", "proposed")
for metric, data in comparison.items():
    status = "📈" if data["improved"] else "📉"
    print(f"{status} {metric}: {data['before']:.3f} → {data['after']:.3f}")

name: RAG Evaluation

on:
  pull_request:
    paths:
      - "src/rag/**"
      - "prompts/**"

jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.12"

      - name: Install dependencies
        run: |
          pip install uv
          uv sync

      - name: Run RAGAS evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          uv run python -m evaluation.run_ragas

      - name: Check quality gates
        run: |
          uv run python -m evaluation.check_gates  # [!code highlight]

from ragas.metrics.base import MetricWithLLM
from dataclasses import dataclass


@dataclass
class TechnicalAccuracy(MetricWithLLM):
    """Custom metric for technical documentation accuracy."""

    name: str = "technical_accuracy"

    def _score(self, row: dict) -> float:
        """Score technical accuracy using LLM-as-judge."""
        prompt = f"""
        Question: {row['question']}
        Answer: {row['answer']}
        Ground Truth: {row['ground_truth']}

        Rate the technical accuracy from 0.0 to 1.0.
        Consider: code correctness, API accuracy, version specificity.

        Return only a number.
        """

        response = self.llm.invoke(prompt)
        return float(response.content.strip())


# Use in evaluation
from ragas import evaluate

custom_metric = TechnicalAccuracy()
results = evaluate(
    dataset=eval_dataset,
    metrics=[faithfulness, answer_relevancy, custom_metric],
)