RAG Evaluation That Actually Correlates with Users

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset
 
data = {
    "question": ["What is the refund window?"],
    "answer": ["You have 30 days to request a refund."],
    "contexts": [["Customers may request refunds within 30 days of purchase."]],
    "ground_truth": ["The refund window is 30 days."],
}
 
dataset = Dataset.from_dict(data)
results = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_precision, context_recall])
print(results)

# Structured eval record
from dataclasses import dataclass
from typing import Optional
 
@dataclass
class EvalRecord:
    query: str
    expected_answer: str
    relevant_doc_ids: list[str]       # for retrieval recall
    query_type: str                   # "canonical" | "adversarial" | "failure"
    failure_mode: Optional[str]       # "no_answer" | "multi_chunk" | "ambiguous"
    created_at: str
    annotator: str
 
# Save as JSONL for versioning
import json
 
def save_eval_set(records: list[EvalRecord], path: str):
    with open(path, "w") as f:
        for r in records:
            f.write(json.dumps(r.__dict__) + "\n")

from openai import OpenAI
 
client = OpenAI()
 
G_EVAL_PROMPT = """
You are evaluating a RAG system's answer. Score the answer on this criterion:
 
CRITERION: Factual Grounding
Score 1–5 where:
5 = Every factual claim in the answer is directly supported by a verbatim or close paraphrase of the context.
4 = All key claims supported; minor stylistic paraphrasing.
3 = Most claims supported; one claim is implied but not explicit.
2 = Some claims unsupported or require inference beyond the context.
1 = Answer contains claims that contradict or are absent from the context.
 
Context: {context}
Answer: {answer}
 
Output: {{"score": <1-5>, "reasoning": "<one sentence>"}}
"""
 
def g_eval_grounding(context: str, answer: str) -> dict:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "user", "content": G_EVAL_PROMPT.format(
                context=context, answer=answer
            )}
        ],
        response_format={"type": "json_object"},
        temperature=0,
    )
    return json.loads(response.choices[0].message.content)

def check_citation_accuracy(answer: str, cited_chunks: list[str]) -> float:
    """Fraction of cited chunks that actually support the answer."""
    supported = 0
    for chunk in cited_chunks:
        result = g_eval_grounding(context=chunk, answer=answer)
        if result["score"] >= 4:
            supported += 1
    return supported / len(cited_chunks) if cited_chunks else 0.0

UNANSWERABLE_QUERIES = [
    "What is the CEO's home address?",
    "What will the stock price be next quarter?",
    # ... domain-specific questions outside your corpus
]
 
def measure_refusal_rate(rag_pipeline, unanswerable: list[str]) -> float:
    refused = 0
    for query in unanswerable:
        answer = rag_pipeline.answer(query)
        # Check for refusal signals
        refusal_phrases = ["I don't have information", "not in my knowledge base", "I cannot find"]
        if any(p.lower() in answer.lower() for p in refusal_phrases):
            refused += 1
    return refused / len(unanswerable)

from sklearn.cluster import KMeans
import numpy as np
 
def find_uncovered_query_clusters(
    production_queries: list[str],
    eval_queries: list[str],
    embedder,
    n_clusters: int = 20,
) -> list[int]:
    """Return cluster IDs that have production queries but no eval coverage."""
    all_embeddings = embedder.encode(production_queries + eval_queries)
    prod_embs = all_embeddings[:len(production_queries)]
    eval_embs = all_embeddings[len(production_queries):]
 
    kmeans = KMeans(n_clusters=n_clusters, random_state=42)
    kmeans.fit(all_embeddings)
 
    prod_clusters = set(kmeans.predict(prod_embs))
    eval_clusters = set(kmeans.predict(eval_embs))
 
    uncovered = prod_clusters - eval_clusters
    return list(uncovered)

REGRESSION_THRESHOLDS = {
    "faithfulness": 0.05,    # max allowed drop
    "recall@10": 0.03,
    "citation_accuracy": 0.04,
    "refusal_rate_unanswerable": 0.10,
}
 
def regression_check(baseline_scores: dict, new_scores: dict) -> list[str]:
    failures = []
    for metric, max_drop in REGRESSION_THRESHOLDS.items():
        drop = baseline_scores[metric] - new_scores[metric]
        if drop > max_drop:
            failures.append(
                f"{metric}: dropped {drop:.3f} (threshold {max_drop})"
            )
    return failures

def semantic_equivalence(answer_a: str, answer_b: str, embedder) -> float:
    emb_a = embedder.encode([answer_a])[0]
    emb_b = embedder.encode([answer_b])[0]
    return float(np.dot(emb_a, emb_b) / (np.linalg.norm(emb_a) * np.linalg.norm(emb_b)))
 
# Flag if new answer diverges too much from baseline
EQUIVALENCE_THRESHOLD = 0.85

# Log every RAG response with context for post-hoc analysis
import uuid
from datetime import datetime
 
def log_rag_response(query, contexts, answer, user_id):
    return {
        "id": str(uuid.uuid4()),
        "timestamp": datetime.utcnow().isoformat(),
        "user_id": user_id,
        "query": query,
        "retrieved_chunk_ids": [c.id for c in contexts],
        "answer": answer,
        "feedback": None,  # filled when user signals good/bad
    }