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

@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

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

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)

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

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

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

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

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

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

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

class ErrorReportGenerator:
    def generate_html_report(self, analysis: ErrorAnalysis) -> str:
        html = f"""
        <html>
        <head><title>Error Analysis Report</title></head>
        <body>
            <h1>Error Analysis Report</h1>
            <h2>Summary</h2>
            <p>Total Errors: {analysis.total_errors}</p>

            <h2>Errors by Category</h2>
            <table>
                <tr><th>Category</th><th>Count</th><th>%</th></tr>
                {''.join(f'<tr><td>{cat}</td><td>{len(errs)}</td><td>{len(errs)/analysis.total_errors*100:.1f}%</td></tr>' 
                         for cat, errs in analysis.by_category.items())}
            </table>

            <h2>Example Errors</h2>
            {''.join(f'<h3>{cat}</h3><ul>{"".join(f"<li>{e.message}</li>" for e in errs[:5])}</ul>' 
                     for cat, errs in analysis.examples.items())}
        </body>
        </html>
        """
        return html

Agent Benchmarks — Standard benchmarks
Evaluation Frameworks — Frameworks
Metrics — Detailed metrics
Online Eval — Online evaluation
Red Teaming — Red teaming
Agent Benchmarks — Benchmarks
Observability — Production monitoring
Reliability — Reliability
Research Directions — 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).

From the Agentic Engineering research vault · CC BY 4.0 · all notes · raw markdown

Offline Evaluation