LLM Evaluation Guide - From Benchmarks to Building Your Own Evaluation System

import anthropic
 
client = anthropic.Anthropic()
 
def llm_judge(question: str, response: str, criteria: list) -> dict:
    """Use LLM as evaluator"""
    eval_prompt = f"""Evaluate the following response.
 
## Question
{question}
 
## Response
{response}
 
## Criteria (1-5 points each)
{chr(10).join(f"- {c}" for c in criteria)}
 
## Scoring Rules
- 1: Very poor, 2: Poor, 3: Average, 4: Good, 5: Excellent
 
Return JSON: {{"scores": {{"criterion": score}}, "average": avg, "reasoning": "rationale", "improvements": "suggestions"}}
"""
    result = client.messages.create(
        model="claude-sonnet-4-6", max_tokens=1024, temperature=0.0,
        messages=[{"role": "user", "content": eval_prompt}]
    )
    return json.loads(result.content[0].text)

def pairwise_compare(question: str, response_a: str, response_b: str) -> str:
    """Select the better of two responses"""
    prompt = f"""Compare two responses and select the better one.
 
Question: {question}
Response A: {response_a}
Response B: {response_b}
 
Return: {{"winner": "A" or "B" or "tie", "reason": "rationale"}}"""
    result = client.messages.create(
        model="claude-sonnet-4-6", max_tokens=512, temperature=0.0,
        messages=[{"role": "user", "content": prompt}]
    )
    return json.loads(result.content[0].text)

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_precision, context_recall
from datasets import Dataset
 
eval_data = {
    "question": ["How to adjust Spark shuffle partitions?"],
    "answer": ["Change the spark.sql.shuffle.partitions setting..."],
    "contexts": [["Spark shuffle partitions are set via spark.sql.shuffle.partitions..."]],
    "ground_truth": ["Adjust shuffle partition count via spark.sql.shuffle.partitions."]
}
 
dataset = Dataset.from_dict(eval_data)
results = evaluate(dataset, metrics=[faithfulness, answer_relevancy, context_precision, context_recall])
print(results)

eval_dataset = [
    {
        "id": "eval-001",
        "category": "troubleshooting",
        "difficulty": "medium",
        "question": "Spark executor OOM error. What are the causes and solutions?",
        "reference_answer": "Increase executor.memory, increase partitions, or...",
        "required_elements": ["executor.memory", "partitions", "data skew"],
        "evaluation_criteria": {
            "accuracy": "Technically accurate root cause analysis",
            "completeness": "Includes at least 3 major causes",
            "actionability": "Includes specific config values or code",
            "format": "Structured format (numbered list, etc.)"
        }
    }
]

class EvaluationPipeline:
    def __init__(self, judge_model="claude-sonnet-4-6"):
        self.judge = anthropic.Anthropic()
        self.judge_model = judge_model
        self.results = []
 
    def evaluate_batch(self, model_fn, eval_dataset):
        for item in eval_dataset:
            response = model_fn(item["question"])
            auto_scores = self.auto_evaluate(item, response)
            judge_scores = self.judge_evaluate(item, response)
            self.results.append({
                "id": item["id"], "category": item["category"],
                "auto_scores": auto_scores, "judge_scores": judge_scores
            })
        return self.aggregate_results()
 
    def auto_evaluate(self, item, response):
        required = item.get("required_elements", [])
        found = sum(1 for e in required if e.lower() in response.lower())
        return {"element_coverage": found / max(len(required), 1)}
 
    def aggregate_results(self):
        return {
            "overall_average": sum(r["judge_scores"]["average"] for r in self.results) / len(self.results),
            "total_evaluated": len(self.results)
        }

class ABTestManager:
    def __init__(self):
        self.experiments = {}
 
    def create_experiment(self, name, variants, traffic_split):
        self.experiments[name] = {
            "variants": variants, "traffic_split": traffic_split,
            "results": {k: [] for k in variants}
        }
 
    def get_variant(self, experiment_name, user_id):
        hash_val = hash(f"{experiment_name}:{user_id}") % 100
        cumulative = 0
        for variant, ratio in self.experiments[experiment_name]["traffic_split"].items():
            cumulative += ratio * 100
            if hash_val < cumulative:
                return variant
        return list(self.experiments[experiment_name]["variants"].keys())[-1]
 
    def record_feedback(self, experiment_name, variant, score):
        self.experiments[experiment_name]["results"][variant].append(score)