# Offline Evaluation
Evaluating agents on static datasets without live execution.
---
## Offline Evaluation Concept
Offline evaluation uses static datasets with ground truth to assess agent capabilities without live API calls or environment interaction.
```
Offline Evaluation Pipeline
├── Dataset Preparation
│ ├── Curation
│ ├── Annotation
│ ├── Split (Train/Val/Test)
│ └── Versioning
├── Evaluation
│ ├── Metric Computation
│ ├── Statistical Testing
│ └── Error Analysis
└── Reporting
├── Metrics Dashboard
├── Error Categorization
└── Regression Detection
```
---
## Dataset Management
### Dataset Structure
```python
@dataclass
class EvaluationDataset:
name: str
version: str
description: str
license: str
splits: dict[str, list[Example]] # train, val, test
metadata: dict
@dataclass
class Example:
id: str
input: str | dict
expected_output: str | dict | None
metadata: dict # difficulty, category, tags, source
ground_truth: Any # For exact matching
```
### Dataset Versioning
```python
class DatasetManager:
def __init__(self, storage: StorageBackend):
self.storage = storage
def create_dataset(self, name: str, examples: list[Example],
version: str = "1.0.0") -> Dataset:
dataset = EvaluationDataset(
name=name,
version=version,
examples=examples,
created_at=datetime.utcnow(),
hash=self.compute_hash(examples)
)
self.storage.save(f"datasets/{name}/{version}", dataset)
return dataset
def get_dataset(self, name: str, version: str = "latest") -> EvaluationDataset:
if version == "latest":
versions = self.storage.list(f"datasets/{name}/")
version = max(versions)
return self.storage.load(f"datasets/{name}/{version}")
def diff_versions(self, name: str, v1: str, v2: str) -> DatasetDiff:
d1 = self.get_dataset(name, v1)
d2 = self.get_dataset(name, v2)
return self.diff(d1, d2)
```
---
## Metric Computation Pipeline
### Batch Evaluation
```python
class OfflineEvaluator:
def __init__(self, metrics: list[Metric], max_workers: int = 10):
self.metrics = metrics
self.executor = ThreadPoolExecutor(max_workers=max_workers)
def evaluate(self,
predictions: list[Prediction],
references: list[Reference]) -> EvaluationResult:
# Parallel metric computation
futures = []
for metric in self.metrics:
future = self.executor.submit(metric.compute, predictions, references)
futures.append((metric.name, future))
results = {}
for name, future in futures:
results[name] = future.result()
return EvaluationResult(
metrics=results,
timestamp=datetime.utcnow(),
num_samples=len(predictions)
)
class Metric:
def compute(self, predictions: list[Prediction], references: list[Reference]) -> MetricResult:
raise NotImplementedError()
def aggregate(self, scores: list[float]) -> AggregatedMetric:
return AggregatedMetric(
mean=statistics.mean(scores),
std=statistics.stdev(scores) if len(scores) > 1 else 0,
min=min(scores),
max=max(scores),
median=statistics.median(scores),
percentiles=np.percentile(scores, [25, 50, 75, 90, 95, 99])
)
```
### Streaming Evaluation (Large Datasets)
```python
class StreamingEvaluator:
def __init__(self, metrics: list[Metric], batch_size: int = 100):
self.metrics = metrics
self.batch_size = batch_size
self.accumulators = {m.name: [] for m in self.metrics}
def add_batch(self, predictions: list[Prediction], references: list[Reference]):
for metric in self.metrics:
scores = metric.compute_batch(predictions, references)
self.accumulators[metric.name].extend(scores)
def finalize(self) -> EvaluationResult:
results = {}
for metric in self.metrics:
scores = self.accumulators[metric.name]
results[metric.name] = metric.aggregate(scores)
return EvaluationResult(metrics=results)
```
---
## Statistical Analysis
### Significance Testing
```python
class StatisticalAnalyzer:
@staticmethod
def paired_ttest(scores_a: list[float], scores_b: list[float]) -> TTestResult:
"""Compare two systems on same examples"""
from scipy import stats
t_stat, p_value = stats.ttest_rel(scores_a, scores_b)
return TTestResult(
t_statistic=t_stat,
p_value=p_value,
significant=p_value < 0.05,
effect_size=statistics.mean(scores_a) - statistics.mean(scores_b)
)
@staticmethod
def bootstrap_ci(metric_fn, data: list, n_bootstrap: int = 10000,
confidence: float = 0.95) -> tuple[float, float]:
"""Bootstrap confidence interval"""
bootstrapped = []
for _ in range(n_bootstrap):
sample = np.random.choice(data, size=len(data), replace=True)
bootstrapped.append(metric_fn(sample))
alpha = (1 - confidence) / 2
lower = np.percentile(bootstrapped, alpha * 100)
upper = np.percentile(bootstrapped, (1 - alpha) * 100)
return (lower, upper)
@staticmethod
def mcnemar_test(outcomes_a: list[bool], outcomes_b: list[bool]) -> McNemarResult:
"""Test for paired binary outcomes"""
from statsmodels.stats.contingency_tables import mcnemar
both_correct = sum(a and b for a, b in zip(outcomes_a, outcomes_b))
a_only = sum(a and not b for a, b in zip(outcomes_a, outcomes_b))
b_only = sum(not a and b for a, b in zip(outcomes_a, outcomes_b))
both_wrong = sum(not a and not b for a, b in zip(outcomes_a, outcomes_b))
table = [[both_correct, a_only], [b_only, both_wrong]]
result = mcnemar(table, exact=False)
return McNemarResult(statistic=result.statistic, p_value=result.pvalue)
```
### Multiple Comparison Correction
```python
def bonferroni_correction(p_values: list[float], alpha: float = 0.05) -> list[bool]:
"""Bonferroni correction for multiple comparisons"""
threshold = alpha / len(p_values)
return [p < threshold for p in p_values]
def benjamini_hochberg(p_values: list[float], alpha: float = 0.05) -> list[bool]:
"""Benjamini-Hochberg FDR control"""
sorted_p = sorted(enumerate(p_values), key=lambda x: x[1])
rejected = [False] * len(p_values)
for i, (orig_idx, p) in enumerate(sorted_p):
if p <= alpha * (i + 1) / len(p_values):
rejected[orig_idx] = True
return rejected
```
---
## Error Analysis
### Error Categorization
```python
class ErrorAnalyzer:
def __init__(self, taxonomy: ErrorTaxonomy):
self.taxonomy = taxonomy
def categorize_errors(self, predictions: list[Prediction],
references: list[Reference]) -> ErrorAnalysis:
errors = []
for pred, ref in zip(predictions, references):
if not pred.matches(ref):
error = self.categorize(pred, reference)
errors.append(error)
# Aggregate by category
by_category = defaultdict(list)
for error in errors:
by_category[error.category].append(error)
return ErrorAnalysis(
total_errors=len(errors),
by_category={cat: len(errors) for cat, errors in by_category.items()},
examples={cat: errors[:5] for cat, errors in by_category.items()},
root_causes=self.identify_root_causes(errors)
)
def categorize(self, prediction: Prediction, reference: Reference) -> Error:
# Rule-based categorization
if prediction.is_refusal:
return Error(category="refusal", message="Incorrect refusal")
if prediction.has_hallucination:
return Error(category="hallucination", message="Unsupported claim")
if prediction.misses_key_info:
return Error(category="incomplete", message="Missing key information")
if prediction.format_error:
return Error(category="format", message="Format violation")
return Error(category="other", message="Other error")
```
### Error Clustering
```python
def cluster_errors(errors: list[Error], n_clusters: int = 10) -> list[ErrorCluster]:
"""Cluster similar errors using embeddings"""
texts = [e.description for e in errors]
embeddings = embed(texts)
from sklearn.cluster import KMeans
kmeans = KMeans(n_clusters=min(n_clusters, len(errors)))
labels = kmeans.fit_predict(embeddings)
clusters = defaultdict(list)
for idx, label in enumerate(labels):
clusters[label].append(errors[idx])
return [
ErrorCluster(
label=f"Cluster {label}",
errors=errors,
representative=errors[0],
size=len(errors)
)
for label, errors in clusters.items()
]
```
---
## Regression Detection
### Golden Set Evaluation
```python
class RegressionDetector:
def __init__(self, golden_set: GoldenSet):
self.golden_set = golden_set
self.baseline_metrics = {}
def establish_baseline(self, model_version: str, metrics: dict):
self.baseline_metrics[model_version] = metrics
def detect_regression(self, current_metrics: dict,
baseline_version: str = "production") -> RegressionReport:
baseline = self.baseline_metrics.get(baseline_version)
if not baseline:
return RegressionReport(regression_detected=False, reason="No baseline")
regressions = []
for metric, current_value in current_metrics.items():
baseline_value = baseline.get(metric)
if baseline_value is None:
continue
change = (current_value - baseline_value) / baseline_value
if change < -0.05: # 5% regression threshold
regressions.append(Regression(
metric=metric,
baseline=baseline_value,
current=current_value,
change_pct=change * 100,
severity="high" if change < -0.2 else "medium"
))
return RegressionReport(
regressions_detected=len(regressions) > 0,
regressions=regressions,
overall_status="regression" if regressions else "stable"
)
```
### Continuous Evaluation
```python
class ContinuousEvaluator:
def __init__(self, evaluator: OfflineEvaluator,
dataset: EvaluationDataset,
schedule: str = "daily"):
self.evaluator = evaluator
self.dataset = dataset
self.schedule = schedule
async def run_continuous(self, model_version: str) -> ContinuousReport:
# Run evaluation
results = self.evaluator.evaluate(
model_version=model_version,
dataset=self.dataset
)
# Check regression
regression = self.regression_detector.detect_regression(
results.metrics, "production"
)
# Alert if regression
if regression.regressions_detected:
await self.alert_on_regression(regression)
return ContinuousReport(
model_version=model_version,
metrics=results.metrics,
regression=regression,
timestamp=datetime.utcnow()
)
```
---
## Visualization & Reporting
### Error Analysis Dashboard
```python
class ErrorReportGenerator:
def generate_html_report(self, analysis: ErrorAnalysis) -> str:
html = f"""
Error Analysis Report
Error Analysis Report
Summary
Total Errors: {analysis.total_errors}
Errors by Category
| Category | Count | % |
|---|
{''.join(f'| {cat} | {len(errs)} | {len(errs)/analysis.total_errors*100:.1f}% |
'
for cat, errs in analysis.by_category.items())}
Example Errors
{''.join(f'{cat}
{"".join(f"- {e.message}
" for e in errs[:5])}
'
for cat, errs in analysis.examples.items())}
"""
return html
```
---
## Related Notes
- [[Evaluation/Agent-Benchmarks.md]] — Standard benchmarks
- [[Evaluation/Evaluation-Frameworks.md]] — Frameworks
- [[Evaluation/Metrics.md]] — Detailed metrics
- [[Evaluation/Online-Eval.md]] — Online evaluation
- [[Evaluation/Red-Teaming.md]] — Red teaming
- [[Evaluation/Agent-Benchmarks.md]] — Benchmarks
- [[Production/Observability.md]] — Production monitoring
- [[Production/Reliability.md]] — Reliability
- [[Frontiers/Research-Directions.md]] — Evaluation research
---
*Sources: "Statistical Significance Testing" (Demšar), "Multiple Comparisons" (Benjamini & Hochberg), "Bootstrap Methods" (Efron & Tibshirani), "McNemar's Test", "Paired t-test", Bootstrap (Efron & Tibshirani), Error Analysis (error analysis papers), Regression Testing (software engineering), A/B Testing (Kohavi), Causal Inference (Imbens & Rubin).*