RAG Evaluation: Beyond Precision/Recall

def compute_faithfulness(answer: str, contexts: List[str]) -> float:
    """
    1. Extract individual claims from answer
    2. Check if each claim is supported by context
    3. Return ratio of supported claims
    """
    claims = extract_claims(answer)
    supported = 0

    for claim in claims:
        if is_supported_by_context(claim, contexts):
            supported += 1

    return supported / len(claims) if claims else 0

def compute_answer_relevance(question: str, answer: str) -> float:
    """
    1. Generate questions from the answer
    2. Measure similarity between original and generated questions
    """
    # Guess the question from just the answer
    generated_questions = generate_questions_from_answer(answer, n=3)

    # Similarity to original question
    similarities = [
        semantic_similarity(question, gen_q)
        for gen_q in generated_questions
    ]

    return np.mean(similarities)

def compute_context_recall(
    ground_truth: str,
    contexts: List[str]
) -> float:
    """
    1. Extract key statements from ground truth
    2. Check if each statement is supported by context
    """
    gt_statements = extract_statements(ground_truth)
    attributed = 0

    for statement in gt_statements:
        if any(supports(ctx, statement) for ctx in contexts):
            attributed += 1

    return attributed / len(gt_statements) if gt_statements else 0

def compute_context_precision(
    question: str,
    answer: str,
    contexts: List[str]
) -> float:
    """
    Check if each retrieved context contributed to the answer
    """
    useful = 0

    for ctx in contexts:
        if contributes_to_answer(ctx, question, answer):
            useful += 1

    return useful / len(contexts) if contexts else 0

# pip install ragas

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

# Prepare evaluation data
eval_data = {
    "question": ["Who is Tesla's CEO?"],
    "answer": ["Elon Musk is Tesla's CEO."],
    "contexts": [["Elon Musk is Tesla's CEO and founder."]],
    "ground_truth": ["Elon Musk"]  # Needed for Context Recall
}

dataset = Dataset.from_dict(eval_data)

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

print(results)

def evaluate_rag_batch(
    questions: List[str],
    rag_system,
    ground_truths: List[str] = None
) -> pd.DataFrame:
    """Evaluate RAG system on multiple questions"""

    results = []
    for i, question in enumerate(questions):
        # Run RAG
        answer, contexts = rag_system.query(question)

        # Evaluate
        result = {
            "question": question,
            "answer": answer,
            "faithfulness": compute_faithfulness(answer, contexts),
            "relevance": compute_answer_relevance(question, answer),
            "context_precision": compute_context_precision(
                question, answer, contexts
            ),
        }

        if ground_truths:
            result["context_recall"] = compute_context_recall(
                ground_truths[i], contexts
            )

        results.append(result)

    return pd.DataFrame(results)

FAITHFULNESS_PROMPT = """
Given the context and answer, determine if each claim in the answer
is supported by the context.

Context:
{context}

Answer:
{answer}

For each claim in the answer, respond with:
- Claim: [the claim]
- Verdict: [Supported/Not Supported]
- Evidence: [quote from context if supported]

Finally, provide the overall faithfulness score (0-1).
"""

def llm_faithfulness(answer: str, context: str, llm) -> float:
    prompt = FAITHFULNESS_PROMPT.format(context=context, answer=answer)
    response = llm.generate(prompt)
    return parse_faithfulness_score(response)

RELEVANCE_PROMPT = """
Given the question and answer, rate how relevant the answer is
to the question on a scale of 0-1.

Question: {question}
Answer: {answer}

Consider:
- Does the answer address the question directly?
- Is the answer complete?
- Is there irrelevant information?

Score (0-1):
Reasoning:
"""

def llm_relevance(question: str, answer: str, llm) -> float:
    prompt = RELEVANCE_PROMPT.format(question=question, answer=answer)
    response = llm.generate(prompt)
    return parse_relevance_score(response)

# Include diverse question types
eval_set = {
    "simple": [  # Single doc sufficient
        "Who is Tesla's CEO?",
        "When was OpenAI founded?",
    ],
    "multi_hop": [  # Multiple docs needed
        "What did Microsoft's CEO say when OpenAI's CEO was fired?",
    ],
    "temporal": [  # Time reasoning required
        "Who was CEO before Sam Altman returned?",
    ],
    "comparison": [  # Comparison questions
        "Which sold more in 2023, Tesla or BYD?",
    ],
    "unanswerable": [  # Cannot be answered
        "What are Tesla's 2025 sales figures?",
    ]
}

class RAGEvaluator:
    def __init__(self, rag_system, llm_judge):
        self.rag = rag_system
        self.judge = llm_judge
        self.metrics_history = []

    def evaluate(self, eval_set: Dict[str, List[str]]) -> Dict:
        results = {}

        for category, questions in eval_set.items():
            category_results = []

            for question in questions:
                answer, contexts = self.rag.query(question)

                metrics = {
                    "faithfulness": self.compute_faithfulness(answer, contexts),
                    "relevance": self.compute_relevance(question, answer),
                    "context_precision": self.compute_precision(
                        question, answer, contexts
                    ),
                }
                category_results.append(metrics)

            results[category] = {
                "avg_faithfulness": np.mean([r["faithfulness"] for r in category_results]),
                "avg_relevance": np.mean([r["relevance"] for r in category_results]),
                "avg_precision": np.mean([r["context_precision"] for r in category_results]),
            }

        self.metrics_history.append({
            "timestamp": datetime.now(),
            "results": results
        })

        return results

    def compare_versions(self, v1_results: Dict, v2_results: Dict) -> Dict:
        """Compare two versions of RAG system"""
        comparison = {}
        for category in v1_results:
            comparison[category] = {
                metric: v2_results[category][metric] - v1_results[category][metric]
                for metric in v1_results[category]
            }
        return comparison