LLM evaluation frameworks, benchmarks, and quality metrics for production systems.
sasmp_version
1.3.0
bonded_agent
05-evaluation-monitoring
bond_type
PRIMARY_BOND
Evaluation Metrics
Measure and improve LLM quality systematically.
Quick Start
Basic Evaluation with RAGAS
from ragas import evaluate
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall
)
from datasets import Dataset
# Prepare evaluation data
eval_data = {
"question": ["What is machine learning?"],
"answer": ["ML is a subset of AI that learns from data."],
"contexts": [["Machine learning is a field of AI..."]],
: []
}
dataset = Dataset.from_dict(eval_data)
results = evaluate(
dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall
]
)
(results)
"ground_truth"
"Machine learning is AI that learns patterns."
# Run evaluation
print
LangChain Evaluation
from langchain.evaluation import load_evaluator
# Criteria-based evaluation
evaluator = load_evaluator("criteria", criteria="helpfulness")
result = evaluator.evaluate_strings(
prediction="Paris is the capital of France.",
input="What is the capital of France?"
)
print(f"Score: {result['score']}, Reasoning: {result['reasoning']}")
classRAGMetrics:
def__init__(self, llm):
self.llm = llm
deffaithfulness(self, answer: str, contexts: list) -> float:
"""Check if answer is supported by retrieved contexts."""
prompt = f"""Given the following context and answer, determine if the answer
is fully supported by the context.
Context: {' '.join(contexts)}
Answer: {answer}
Score from 0 (not supported) to 1 (fully supported):"""
response = self.llm.generate(prompt)
returnfloat(response.strip())
defrelevance(self, question: str, answer: str) -> float:
"""Check if answer is relevant to the question."""
prompt = f"""Rate how relevant this answer is to the question.
Question: {question}
Answer: {answer}
Score from 0 (irrelevant) to 1 (highly relevant):"""
response = self.llm.generate(prompt)
returnfloat(response.strip())
defcontext_precision(self, question: str, contexts: list) -> float:
"""Check if retrieved contexts are relevant to question."""
relevant_count = 0for ctx in contexts:
prompt = f"""Is this context relevant to answering the question?
Question: {question}
Context: {ctx}
Answer Yes or No:"""if"yes"inself.llm.generate(prompt).lower():
relevant_count += 1return relevant_count / len(contexts)
Hallucination Detection
classHallucinationDetector:
def__init__(self, llm, knowledge_base=None):
self.llm = llm
self.knowledge_base = knowledge_base
defdetect(self, claim: str, source: str = None) -> dict:
"""Detect potential hallucinations in a claim."""
results = {
'claim': claim,
'is_hallucination': False,
'confidence': 0.0,
'reason': ''
}
# Check against source if providedif source:
prompt = f"""Determine if this claim is supported by the source.
Source: {source}
Claim: {claim}
Is the claim fully supported? Answer with:
SUPPORTED, PARTIALLY_SUPPORTED, or NOT_SUPPORTED
Reason:"""
response = self.llm.generate(prompt)
if"NOT_SUPPORTED"in response:
results['is_hallucination'] = True
results['confidence'] = 0.9elif"PARTIALLY"in response:
results['confidence'] = 0.5# Check for self-consistency
regenerations = [
self.llm.generate(f"Verify: {claim}")
for _ inrange(3)
]
consistency = self._check_consistency(regenerations)
if consistency < 0.7:
results['is_hallucination'] = True
results['reason'] = 'Inconsistent across regenerations'return results
Benchmark Suites
MMLU (Massive Multitask Language Understanding)
from datasets import load_dataset
defevaluate_mmlu(model, tokenizer, subjects=None):
dataset = load_dataset("cais/mmlu", "all")
results = {}
for subject in subjects or dataset.keys():
correct = 0
total = 0for example in dataset[subject]:
question = example['question']
choices = example['choices']
answer = example['answer']
# Format prompt
prompt = f"{question}\n"for i, choice inenumerate(choices):
prompt += f"{chr(65+i)}. {choice}\n"
prompt += "Answer:"# Get model prediction
response = model.generate(prompt)
predicted = response[0].upper()
if predicted == chr(65 + answer):
correct += 1
total += 1
results[subject] = correct / total
return results
HumanEval (Code Generation)
defevaluate_humaneval(model):
from human_eval.data import read_problems
from human_eval.execution import check_correctness
problems = read_problems()
results = []
for task_id, problem in problems.items():
prompt = problem['prompt']
# Generate completions
completions = [model.generate(prompt) for _ inrange(10)]
# Check correctnessfor completion in completions:
result = check_correctness(problem, completion, timeout=10.0)
results.append(result['passed'])
pass_at_1 = sum(results[:len(problems)]) / len(problems)
return {'pass@1': pass_at_1}
Evaluation Framework
from dataclasses import dataclass
from typing importList, Dict, Callable@dataclassclassEvaluationConfig:
metrics: List[str]
sample_size: int = 100
confidence_level: float = 0.95classLLMEvaluator:
def__init__(self, model, config: EvaluationConfig):
self.model = model
self.config = config
self.metrics_registry: Dict[str, Callable] = {}
defregister_metric(self, name: str, func: Callable):
self.metrics_registry[name] = func
defevaluate(self, test_data: List[dict]) -> dict:
results = {metric: [] for metric inself.config.metrics}
for sample in test_data[:self.config.sample_size]:
prediction = self.model.generate(sample['input'])
for metric_name inself.config.metrics:
metric_func = self.metrics_registry[metric_name]
score = metric_func(
prediction=prediction,
reference=sample.get('expected'),
context=sample.get('context')
)
results[metric_name].append(score)
# Aggregate results
aggregated = {}
for metric, scores in results.items():
aggregated[metric] = {
'mean': np.mean(scores),
'std': np.std(scores),
'min': np.min(scores),
'max': np.max(scores)
}
return aggregated
Use multiple metrics: No single metric captures everything
Human evaluation: Ground truth for subjective quality
Domain-specific metrics: Customize for your use case
Regular benchmarking: Track quality over time
Statistical significance: Use proper sample sizes
Version everything: Models, prompts, and test data
Error Handling & Retry
from tenacity import retry, stop_after_attempt
@retry(stop=stop_after_attempt(3))defevaluate_with_retry(model_output, reference):
return evaluator.evaluate(model_output, reference)
defbatch_evaluate(samples, batch_size=50):
results = []
for i inrange(0, len(samples), batch_size):
batch = samples[i:i+batch_size]
results.extend([evaluate_with_retry(s) for s in batch])
return results