Implement comprehensive evaluation strategies for LLM applications using automated metrics, human feedback, and benchmarking...
Master comprehensive evaluation strategies for LLM applications, from automated metrics to human evaluation and A/B testing.
Fast, repeatable, scalable evaluation using computed scores.
Text Generation:
Classification:
Retrieval (RAG):
Manual assessment for quality aspects difficult to automate.
Dimensions:
Use stronger LLMs to evaluate weaker model outputs.
Approaches:
from llm_eval import EvaluationSuite, Metric
# Define evaluation suite
suite = EvaluationSuite([
Metric.accuracy(),
Metric.bleu(),
Metric.bertscore(),
Metric.custom(name="groundedness", fn=check_groundedness)
])
# Prepare test cases
test_cases = [
{
"input": "What is the capital of France?",
"expected": "Paris",
"context": "France is a country in Europe. Paris is its capital."
},
# ... more test cases
]
# Run evaluation
results = suite.evaluate(
model=your_model,
test_cases=test_cases
)
print(f"Overall Accuracy: {results.metrics['accuracy']}")
print(f"BLEU Score: {results.metrics['bleu']}")
from nltk.translate.bleu_score import sentence_bleu, SmoothingFunction
def calculate_bleu(reference, hypothesis):
"""Calculate BLEU score between reference and hypothesis."""
smoothie = SmoothingFunction().method4
return sentence_bleu(
[reference.split()],
hypothesis.split(),
smoothing_function=smoothie
)
# Usage
bleu = calculate_bleu(
reference="The cat sat on the mat",
hypothesis="A cat is sitting on the mat"
)
from rouge_score import rouge_scorer
def calculate_rouge(reference, hypothesis):
"""Calculate ROUGE scores."""
scorer = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)
scores = scorer.score(reference, hypothesis)
return {
'rouge1': scores['rouge1'].fmeasure,
'rouge2': scores['rouge2'].fmeasure,
'rougeL': scores['rougeL'].fmeasure
}
from bert_score import score
def calculate_bertscore(references, hypotheses):
"""Calculate BERTScore using pre-trained BERT."""
P, R, F1 = score(
hypotheses,
references,
lang='en',
model_type='microsoft/deberta-xlarge-mnli'
)
return {
'precision': P.mean().item(),
'recall': R.mean().item(),
'f1': F1.mean().item()
}
def calculate_groundedness(response, context):
"""Check if response is grounded in provided context."""
# Use NLI model to check entailment
from transformers import pipeline
nli = pipeline("text-classification", model="microsoft/deberta-large-mnli")
result = nli(f"{context} [SEP] {response}")[0]
# Return confidence that response is entailed by context
return result['score'] if result['label'] == 'ENTAILMENT' else 0.0
def calculate_toxicity(text):
"""Measure toxicity in generated text."""
from detoxify import Detoxify
results = Detoxify('original').predict(text)
return max(results.values()) # Return highest toxicity score
def calculate_factuality(claim, knowledge_base):
"""Verify factual claims against knowledge base."""
# Implementation depends on your knowledge base
# Could use retrieval + NLI, or fact-checking API
raise NotImplementedError("A verified fact-checking adapter is required")
def llm_judge_quality(response, question):
"""Use GPT-5 to judge response quality."""
prompt = f"""Rate the following response on a scale of 1-10 for:
1. Accuracy (factually correct)
2. Helpfulness (answers the question)
3. Clarity (well-written and understandable)
Question: {question}
Response: {response}
Provide ratings in JSON format:
{{
"accuracy": <1-10>,
"helpfulness": <1-10>,
"clarity": <1-10>,
"reasoning": "<brief explanation>"
}}
"""
result = openai.ChatCompletion.create(
model="gpt-5",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
return json.loads(result.choices[0].message.content)
def compare_responses(question, response_a, response_b):
"""Compare two responses using LLM judge."""
prompt = f"""Compare these two responses to the question and determine which is better.
Question: {question}
Response A: {response_a}
Response B: {response_b}
Which response is better and why? Consider accuracy, helpfulness, and clarity.
Answer with JSON:
{{
"winner": "A" or "B" or "tie",
"reasoning": "<explanation>",
"confidence": <1-10>
}}
"""
result = openai.ChatCompletion.create(
model="gpt-5",
messages=[{"role": "user", "content": prompt}],
temperature=0
)
return json.loads(result.choices[0].message.content)
class AnnotationTask:
"""Structure for human annotation task."""
def __init__(self, response, question, context=None):
self.response = response
self.question = question
self.context = context
def get_annotation_form(self):
return {
"question": self.question,
"context": self.context,
"response": self.response,
"ratings": {
"accuracy": {
"scale": "1-5",
"description": "Is the response factually correct?"
},
"relevance": {
"scale": "1-5",
"description": "Does it answer the question?"
},
"coherence": {
"scale": "1-5",
"description": "Is it logically consistent?"
}
},
"issues": {
"factual_error": False,
"hallucination": False,
"off_topic": False,
"unsafe_content": False
},
"feedback": ""
}
from sklearn.metrics import cohen_kappa_score
def calculate_agreement(rater1_scores, rater2_scores):
"""Calculate inter-rater agreement."""
kappa = cohen_kappa_score(rater1_scores, rater2_scores)
# Ordered ranges: Boolean dictionary keys overwrite each other.
interpretation = "Undefined"
for upper, label in [(0, "Poor"), (0.2, "Slight"), (0.4, "Fair"),
(0.6, "Moderate"), (0.8, "Substantial"),
(float("inf"), "Almost Perfect")]:
if kappa < upper:
interpretation = label
break
return {
"kappa": kappa,
"interpretation": interpretation
}
from scipy import stats
import numpy as np
class ABTest:
def __init__(self, variant_a_name="A", variant_b_name="B"):
self.variant_a = {"name": variant_a_name, "scores": []}
self.variant_b = {"name": variant_b_name, "scores": []}
def add_result(self, variant, score):
"""Add evaluation result for a variant."""
if variant == "A":
self.variant_a["scores"].append(score)
else:
self.variant_b["scores"].append(score)
def analyze(self, alpha=0.05):
"""Perform statistical analysis."""
a_scores = self.variant_a["scores"]
b_scores = self.variant_b["scores"]
if len(a_scores) < 2 or len(b_scores) < 2:
raise ValueError("At least two observations per independent group are required")
if not np.all(np.isfinite(a_scores)) or not np.all(np.isfinite(b_scores)):
raise ValueError("Require finite observations")
# Illustrative independent-group analysis; use a paired design for matched tasks.
# T-test
t_stat, p_value = stats.ttest_ind(a_scores, b_scores)
# Effect size (Cohen's d)
pooled_std = np.sqrt((np.std(a_scores)**2 + np.std(b_scores)**2) / 2)
cohens_d = (np.mean(b_scores) - np.mean(a_scores)) / pooled_std if pooled_std > 0 else None
return {
"variant_a_mean": np.mean(a_scores),
"variant_b_mean": np.mean(b_scores),
"difference": np.mean(b_scores) - np.mean(a_scores),
"relative_improvement": ((np.mean(b_scores) - np.mean(a_scores)) / np.mean(a_scores)
if np.mean(a_scores) != 0 else None),
"p_value": p_value,
"statistically_significant": p_value < alpha,
"cohens_d": cohens_d,
"effect_size": self.interpret_cohens_d(cohens_d) if cohens_d is not None else "undefined",
"winner": (("B" if np.mean(b_scores) > np.mean(a_scores) else "A")
if np.isfinite(p_value) and p_value < alpha and np.mean(b_scores) != np.mean(a_scores)
else "inconclusive")
}
@staticmethod
def interpret_cohens_d(d):
"""Interpret Cohen's d effect size."""
abs_d = abs(d)
if abs_d < 0.2:
return "negligible"
elif abs_d < 0.5:
return "small"
elif abs_d < 0.8:
return "medium"
else:
return "large"
class RegressionDetector:
def __init__(self, baseline_results, threshold=0.05, directions=None):
import math
if not baseline_results or not math.isfinite(threshold) or threshold < 0:
raise ValueError("Require baseline metrics and a finite nonnegative threshold")
self.baseline = baseline_results
self.threshold = threshold
self.directions = directions or {}
def check_for_regression(self, new_results):
"""Detect if new results show regression."""
import math
regressions = []
missing_metrics = []
for metric in self.baseline.keys():
baseline_score = self.baseline[metric]
new_score = new_results.get(metric)
if new_score is None:
missing_metrics.append(metric)
continue
if not math.isfinite(baseline_score) or not math.isfinite(new_score):
raise ValueError("Metrics must be finite")
# Calculate relative change
if baseline_score <= 0:
raise ValueError("Relative regression requires a positive nonzero baseline; use an absolute threshold instead.")
relative_change = (new_score - baseline_score) / baseline_score
# Default is higher-is-better; explicitly mark latency/cost as lower.
direction = self.directions.get(metric, "higher")
if direction not in {"higher", "lower"}:
raise ValueError("Unknown metric direction")
is_regression = (relative_change < -self.threshold if direction == "higher"
else relative_change > self.threshold)
if is_regression:
regressions.append({
"metric": metric,
"baseline": baseline_score,
"current": new_score,
"change": relative_change
})
return {
"has_regression": True if regressions else (None if missing_metrics else False),
"analysis_complete": not missing_metrics,
"missing_metrics": missing_metrics,
"regressions": regressions
}
Read the local benchmark example for a bounded runner, adapter prerequisites and incomplete-data handling.
The local benchmark reference contains an illustrative runner, not an installed evaluation framework or dataset. Code below illustrates patterns that must be adapted to the project's actual dependencies and validated before use; the llm_eval interface is pseudocode, not a promised installed package. Inspect existing SDK versions instead of copying legacy provider calls unchanged.
Use sanitized, representative data and an explicit evaluation budget. Model downloads, hosted inference, paid judge calls and telemetry exports require authorization for the data and cost. Prefer already cached models; do not treat a prompt as a security boundary against instructions embedded in evaluated outputs. Calibrate judge scores against a human-labeled sample and inspect errors, bias and disagreement. Overlap scores do not prove correctness, safety or production readiness.