RAG Evaluation & Debugging

"""Retrieval Metrics Implementation"""
import numpy as np
from typing import List, Set

def precision_at_k(retrieved_ids: List[str], relevant_ids: Set[str], k: int) -> float:
    """What fraction of top-K results are relevant?"""
    top_k = retrieved_ids[:k]
    relevant_in_k = sum(1 for doc_id in top_k if doc_id in relevant_ids)
    return relevant_in_k / k

def recall_at_k(retrieved_ids: List[str], relevant_ids: Set[str], k: int) -> float:
    """What fraction of ALL relevant docs appear in top-K?"""
    top_k = retrieved_ids[:k]
    relevant_in_k = sum(1 for doc_id in top_k if doc_id in relevant_ids)
    return relevant_in_k / len(relevant_ids) if relevant_ids else 0.0

def mrr(retrieved_ids: List[str], relevant_ids: Set[str]) -> float:
    """Reciprocal rank of the FIRST relevant result."""
    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[str], k: int) -> float:
    """Normalized Discounted Cumulative Gain."""
    def dcg(ids):
        score = 0.0
        for i, doc_id in enumerate(ids[:k]):
            rel = 1.0 if doc_id in relevant_ids else 0.0
            score += rel / np.log2(i + 2)  # +2 because log2(1) = 0
        return score
    
    actual_dcg = dcg(retrieved_ids)
    # Ideal: all relevant docs at top
    ideal_ids = [d for d in retrieved_ids if d in relevant_ids] + \
                [d for d in retrieved_ids if d not in relevant_ids]
    ideal_dcg = dcg(ideal_ids)
    
    return actual_dcg / ideal_dcg if ideal_dcg > 0 else 0.0

# --- Evaluate a retrieval system ---
def evaluate_retrieval(test_set: list[dict], retriever_fn) -> dict:
    """
    test_set: [{"query": str, "relevant_doc_ids": set}, ...]
    retriever_fn: query → list of doc_ids (ranked)
    """
    metrics = {"precision@5": [], "recall@5": [], "mrr": [], "ndcg@5": []}
    
    for case in test_set:
        retrieved = retriever_fn(case["query"])
        relevant = case["relevant_doc_ids"]
        
        metrics["precision@5"].append(precision_at_k(retrieved, relevant, 5))
        metrics["recall@5"].append(recall_at_k(retrieved, relevant, 5))
        metrics["mrr"].append(mrr(retrieved, relevant))
        metrics["ndcg@5"].append(ndcg_at_k(retrieved, relevant, 5))
    
    return {k: np.mean(v) for k, v in metrics.items()}

"""LLM-as-Judge Generation Metrics"""
from openai import OpenAI

client = OpenAI()

def evaluate_faithfulness(answer: str, context: str) -> dict:
    """Check if answer is grounded in context (no hallucination)."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"""Evaluate if the answer is faithful to the context.
            
Context: {context}

Answer: {answer}

For each claim in the answer, determine if it's:
- SUPPORTED: directly stated or clearly implied by context
- NOT_SUPPORTED: not found in context (hallucination)

Output JSON:
{{"claims": [{{"claim": "...", "verdict": "SUPPORTED|NOT_SUPPORTED"}}], "faithfulness_score": 0.0-1.0}}"""
        }],
        temperature=0,
        response_format={"type": "json_object"}
    )
    
    return json.loads(response.choices[0].message.content)


def evaluate_relevance(question: str, answer: str) -> dict:
    """Check if answer addresses the question."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"""Rate how well this answer addresses the question.

Question: {question}
Answer: {answer}

Score 0-1 where:
- 1.0: Directly and completely answers the question
- 0.5: Partially answers or is tangentially related
- 0.0: Doesn't address the question at all

Output JSON: {{"score": 0.0-1.0, "reasoning": "..."}}"""
        }],
        temperature=0,
        response_format={"type": "json_object"}
    )
    
    return json.loads(response.choices[0].message.content)

"""RAGAS Evaluation Pipeline"""
from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# Prepare evaluation dataset
eval_data = {
    "question": [
        "How do I deploy with Docker?",
        "What is the rate limit for the API?",
        "How do I handle authentication errors?",
    ],
    "answer": [
        # Answers generated by your RAG system
        "To deploy with Docker, create a Dockerfile...",
        "The API rate limit is 100 requests per minute...",
        "Authentication errors return 401 status codes...",
    ],
    "contexts": [
        # Retrieved context chunks (list of lists)
        ["Docker deployment guide: Create a Dockerfile with...", "Use docker-compose for..."],
        ["Rate limits: Free tier allows 100 req/min...", "Enterprise tier has no limits"],
        ["Error handling: 401 errors indicate invalid tokens...", "Retry with refresh token"],
    ],
    "ground_truth": [
        # Human-written reference answers
        "Deploy using Docker by creating a Dockerfile and running docker-compose up",
        "100 requests per minute on free tier",
        "401 errors mean invalid token; refresh and retry",
    ]
}

dataset = Dataset.from_dict(eval_data)

# Run evaluation
results = evaluate(
    dataset,
    metrics=[
        faithfulness,          # Is answer grounded in context?
        answer_relevancy,      # Does answer address the question?
        context_precision,     # Are retrieved docs relevant?
        context_recall,        # Did we find all relevant docs?
    ]
)

print(results)
# {'faithfulness': 0.92, 'answer_relevancy': 0.88, 
#  'context_precision': 0.85, 'context_recall': 0.78}

# Per-question breakdown
df = results.to_pandas()
print(df[['question', 'faithfulness', 'answer_relevancy']])

"""RAG Debugging Toolkit"""
import json
from dataclasses import dataclass

@dataclass
class DebugTrace:
    query: str
    retrieved_chunks: list[dict]  # content, score, source
    generated_answer: str
    ground_truth: str = None
    
    def diagnose(self) -> str:
        """Identify the most likely failure mode."""
        # Check retrieval quality
        if not self.retrieved_chunks:
            return "RETRIEVAL_MISS: No documents retrieved"
        
        top_score = self.retrieved_chunks[0]["score"]
        if top_score < 0.6:
            return f"RETRIEVAL_MISS: Low relevance (top score: {top_score:.3f})"
        
        # Check if ground truth info is in context
        if self.ground_truth:
            context_text = " ".join(c["content"] for c in self.retrieved_chunks)
            # Simple keyword overlap check
            gt_words = set(self.ground_truth.lower().split())
            ctx_words = set(context_text.lower().split())
            overlap = len(gt_words & ctx_words) / len(gt_words)
            
            if overlap < 0.3:
                return f"RETRIEVAL_MISS: Ground truth not in context (overlap: {overlap:.0%})"
            
            # Context has the info — check generation
            if self.ground_truth.lower() not in self.generated_answer.lower():
                return "GENERATION_MISS: Context has answer but LLM missed it"
        
        return "LIKELY_OK: Retrieval and generation appear functional"


def debug_pipeline(rag, test_queries: list[dict]):
    """Run diagnostics on a set of test queries."""
    issues = {"RETRIEVAL_MISS": [], "RETRIEVAL_NOISE": [], 
              "GENERATION_MISS": [], "HALLUCINATION": [], "OK": []}
    
    for case in test_queries:
        result = rag.query(case["query"])
        
        trace = DebugTrace(
            query=case["query"],
            retrieved_chunks=result.sources,
            generated_answer=result.answer,
            ground_truth=case.get("expected_answer")
        )
        
        diagnosis = trace.diagnose()
        category = diagnosis.split(":")[0]
        
        if category in issues:
            issues[category].append({
                "query": case["query"],
                "diagnosis": diagnosis,
                "top_score": trace.retrieved_chunks[0]["score"] if trace.retrieved_chunks else 0
            })
        else:
            issues["OK"].append(case["query"])
    
    # Summary report
    print("\n=== RAG Health Report ===")
    total = len(test_queries)
    for category, items in issues.items():
        pct = len(items) / total * 100
        print(f"  {category}: {len(items)}/{total} ({pct:.0f}%)")
        if items and category != "OK":
            print(f"    Example: {items[0]['query']}")
    
    return issues

"""Production Monitoring Hooks"""
import time
from datetime import datetime

class RAGMonitor:
    def __init__(self):
        self.logs = []
    
    def log_query(self, query, result, latency_ms):
        self.logs.append({
            "timestamp": datetime.now().isoformat(),
            "query": query,
            "answer_length": len(result.answer),
            "top_retrieval_score": max(result.retrieval_scores) if result.retrieval_scores else 0,
            "avg_retrieval_score": sum(result.retrieval_scores) / len(result.retrieval_scores) if result.retrieval_scores else 0,
            "num_sources": len(result.sources),
            "latency_ms": latency_ms,
            "is_abstention": "don't have" in result.answer.lower() or "not sure" in result.answer.lower(),
        })
    
    def get_health_summary(self, last_n: int = 100):
        recent = self.logs[-last_n:]
        return {
            "avg_latency_ms": np.mean([l["latency_ms"] for l in recent]),
            "p95_latency_ms": np.percentile([l["latency_ms"] for l in recent], 95),
            "avg_retrieval_score": np.mean([l["avg_retrieval_score"] for l in recent]),
            "abstention_rate": np.mean([l["is_abstention"] for l in recent]),
            "low_confidence_rate": np.mean([l["top_retrieval_score"] < 0.6 for l in recent]),
        }
    
    def alert_if_degraded(self):
        summary = self.get_health_summary()
        alerts = []
        if summary["abstention_rate"] > 0.3:
            alerts.append(f"⚠️ High abstention rate: {summary['abstention_rate']:.0%}")
        if summary["avg_retrieval_score"] < 0.65:
            alerts.append(f"⚠️ Low retrieval quality: {summary['avg_retrieval_score']:.3f}")
        if summary["p95_latency_ms"] > 5000:
            alerts.append(f"⚠️ High latency: P95={summary['p95_latency_ms']:.0f}ms")
        return alerts