LLM Evaluation and Benchmarking 2026: How to Measure AI Quality

from ragas import evaluate
from ragas.metrics import (
    faithfulness,           # Answer grounded in context?
    answer_relevancy,       # Answer relevant to question?
    context_precision,      # Retrieved context is relevant?
    context_recall,         # All necessary context retrieved?
)
from datasets import Dataset

# Your RAG pipeline output
data = {
    "question": [
        "What is the capital of France?",
        "How does transformer attention work?",
    ],
    "answer": [
        "The capital of France is Paris.",
        "Attention computes weighted sums of values based on query-key similarity.",
    ],
    "contexts": [
        ["France is a country in Western Europe. Its capital city is Paris."],
        ["Transformers use self-attention mechanisms that compute dot products between queries and keys."],
    ],
    "ground_truth": [
        "Paris is the capital of France.",
        "Transformer attention uses scaled dot-product attention between queries, keys, and values.",
    ]
}

dataset = Dataset.from_dict(data)
results = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_precision, context_recall])

print(results)
# {'faithfulness': 0.96, 'answer_relevancy': 0.94,
#  'context_precision': 0.89, 'context_recall': 0.87}

from openai import OpenAI
import json

client = OpenAI()

JUDGE_PROMPT = """You are an expert evaluator. Score the AI response on these criteria.

Question: {question}
AI Response: {response}
Reference Answer: {reference}

Score each criterion 1-5:
- accuracy: factually correct and complete
- relevance: directly answers the question
- clarity: clear and well-structured
- safety: no harmful or misleading content

Return JSON: {{"accuracy": N, "relevance": N, "clarity": N, "safety": N, "reasoning": "brief explanation"}}"""

def llm_judge(question: str, response: str, reference: str = "") -> dict:
    result = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": JUDGE_PROMPT.format(
                question=question,
                response=response,
                reference=reference or "No reference provided"
            )
        }],
        response_format={"type": "json_object"},
        temperature=0,
    )
    scores = json.loads(result.choices[0].message.content)
    scores["overall"] = round(sum(v for k, v in scores.items() if isinstance(v, (int, float))) / 4, 2)
    return scores

# Evaluate your model's outputs
test_cases = [
    {
        "question": "What is machine learning?",
        "response": "ML is a type of AI that learns from data.",
        "reference": "Machine learning is a subset of artificial intelligence that enables systems to learn and improve from experience without being explicitly programmed."
    }
]

for case in test_cases:
    scores = llm_judge(**case)
    print(f"Overall: {scores['overall']}/5 — {scores['reasoning']}")

import json
from dataclasses import dataclass
from typing import Callable
import statistics

@dataclass
class EvalCase:
    id: str
    input: str | dict
    expected: str
    metadata: dict = None

class EvalSuite:
    def __init__(self, name: str):
        self.name = name
        self.cases: list[EvalCase] = []
        self.results = []

    def add_case(self, id: str, input: str, expected: str, **metadata):
        self.cases.append(EvalCase(id, input, expected, metadata))

    def run(self, model_fn: Callable, judge_fn: Callable = None):
        """Run all eval cases against a model function."""
        print(f"Running {len(self.cases)} eval cases...")
        self.results = []

        for case in self.cases:
            response = model_fn(case.input)
            scores = judge_fn(case.input, response, case.expected) if judge_fn else {}

            self.results.append({
                "id": case.id,
                "input": case.input,
                "expected": case.expected,
                "actual": response,
                "scores": scores,
            })

        return self.report()

    def report(self) -> dict:
        if not self.results:
            return {}

        all_scores = [r["scores"].get("overall", 0) for r in self.results if r["scores"]]
        return {
            "suite": self.name,
            "total": len(self.results),
            "mean_score": round(statistics.mean(all_scores), 3) if all_scores else 0,
            "median_score": round(statistics.median(all_scores), 3) if all_scores else 0,
            "pass_rate": sum(1 for s in all_scores if s >= 4) / len(all_scores) if all_scores else 0,
            "results": self.results,
        }

# Build eval suite
suite = EvalSuite("chatbot-qa-v1")
suite.add_case("q1", "What is RAG?", "RAG stands for Retrieval-Augmented Generation...")
suite.add_case("q2", "Explain transformers", "Transformers are neural network architectures...")
suite.add_case("q3", "What is fine-tuning?", "Fine-tuning is training a pre-trained model...")

# Run against your model
def my_model(question: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": question}],
        max_tokens=200,
    )
    return response.choices[0].message.content

report = suite.run(my_model, llm_judge)
print(f"Pass rate: {report['pass_rate']:.1%}")
print(f"Mean score: {report['mean_score']}/5")

from concurrent.futures import ThreadPoolExecutor

def compare_models(test_cases: list[dict], models: list[str]) -> dict:
    """Compare multiple models on the same test cases."""
    results = {model: [] for model in models}

    def run_case(model: str, case: dict) -> dict:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": case["question"]}],
            max_tokens=500,
            temperature=0,
        )
        answer = response.choices[0].message.content
        scores = llm_judge(case["question"], answer, case.get("reference", ""))
        return {**scores, "response": answer, "tokens": response.usage.total_tokens}

    with ThreadPoolExecutor(max_workers=4) as executor:
        for model in models:
            futures = [executor.submit(run_case, model, case) for case in test_cases]
            results[model] = [f.result() for f in futures]

    # Summarize
    summary = {}
    for model in models:
        model_results = results[model]
        scores = [r["overall"] for r in model_results]
        tokens = [r["tokens"] for r in model_results]
        summary[model] = {
            "mean_score": round(statistics.mean(scores), 3),
            "avg_tokens": round(statistics.mean(tokens)),
        }

    return summary

comparison = compare_models(test_cases, ["gpt-4o-mini", "gpt-4o", "gpt-3.5-turbo"])
for model, stats in comparison.items():
    print(f"{model}: score={stats['mean_score']}/5, tokens={stats['avg_tokens']}")

import time
from datetime import datetime

class ProductionEvaluator:
    def __init__(self, sample_rate: float = 0.1):
        """Evaluate a random sample of production traffic."""
        self.sample_rate = sample_rate
        self.log = []

    def should_evaluate(self) -> bool:
        import random
        return random.random() < self.sample_rate

    def log_and_evaluate(self, question: str, response: str, latency_ms: int):
        entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "question": question,
            "response": response,
            "latency_ms": latency_ms,
        }

        if self.should_evaluate():
            scores = llm_judge(question, response)
            entry["scores"] = scores

            # Alert on low quality
            if scores.get("overall", 5) < 3:
                self.alert_low_quality(entry)

        self.log.append(entry)
        return entry

    def alert_low_quality(self, entry: dict):
        print(f"⚠️  LOW QUALITY RESPONSE DETECTED")
        print(f"Q: {entry['question'][:100]}")
        print(f"Score: {entry['scores']['overall']}/5")

    def get_stats(self, last_n: int = 100) -> dict:
        recent = self.log[-last_n:]
        evaluated = [e for e in recent if "scores" in e]
        if not evaluated:
            return {"message": "No evaluated samples yet"}

        scores = [e["scores"]["overall"] for e in evaluated]
        latencies = [e["latency_ms"] for e in recent]

        return {
            "samples": len(recent),
            "evaluated": len(evaluated),
            "mean_quality": round(statistics.mean(scores), 2),
            "p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)],
        }

evaluator = ProductionEvaluator(sample_rate=0.05)  # Evaluate 5% of requests