System Design 04 — Distributed Offline Evaluation Platform

============================================================

Design a scalable platform for running large-scale model evaluations

across hundreds of model checkpoints and dozens of benchmarks.

Problem Statement

Build an offline evaluation platform that:

• Evaluates 200+ model checkpoints per week against 60+ benchmarks
• Supports custom task registration (beyond lm_eval built-ins)
• Provides reproducible, versioned evaluation results
• Detects accuracy regressions automatically
• Scales from a single laptop to a 100-GPU cluster

Scale target:

• 200 models × 60 benchmarks × 1000 examples = 12M eval calls/week
• At 100ms/call = 1,200,000 GPU-seconds = 333 GPU-hours/week
• Need: 50-100 GPUs running continuously (with ~70% utilization)

Architecture

┌─────────────────────────────────────────────────────────────────────────────┐
│                     Distributed Eval Platform                               │
└─────────────────────────────────────────────────────────────────────────────┘

[Model Registry] ──────────────────────────────────────────────────────────────
  - Stores: model checkpoint (S3), metadata.json, sha256 fingerprint
  - Triggers: "new model event" on every push to registry
  
[Benchmark Registry] ─────────────────────────────────────────────────────────
  - MMLU (57 subjects), HellaSwag, ARC, WinoGrande, GSM8K, HumanEval
  - Custom tasks (registered via Python @task_registry.register decorator)
  - Versioned: each benchmark has a version hash (dataset + prompt format)

[Job Scheduler] ─────────────────────────────────────────────────────────────
  Airflow DAG (weekly trigger + event trigger):
  1. Fetch all new/changed model checkpoints
  2. Cross with all benchmarks → (model, benchmark) task matrix
  3. Filter: skip already-cached evaluations (check model_sha + benchmark_version)
  4. Submit tasks to Ray cluster

[Ray Eval Workers] ──────────────────────────────────────────────────────────
  Head node: Airflow + job management
  Worker nodes: 50-100 A10G GPUs
  
  Per worker task:
    1. Download model from S3 (cached on worker)
    2. Load benchmark examples (from NFS cache)
    3. Run evaluation (batched inference)
    4. Write results to Results DB

[Results Database] ──────────────────────────────────────────────────────────
  PostgreSQL + TimescaleDB (for time-series accuracy tracking)
  
[Regression Detector] ────────────────────────────────────────────────────────
  - Compares new results vs. 7-day rolling baseline per (model_family, benchmark)
  - Triggers: accuracy drop > 1% with p < 0.05 → Slack alert + Jira ticket

[Dashboard] ─────────────────────────────────────────────────────────────────
  Grafana + custom frontend
  - Accuracy over time per model family
  - Pareto frontier: accuracy vs. latency vs. model size
  - Regression heatmap

Database Schema

-- Model registry
CREATE TABLE models (
    id          BIGSERIAL PRIMARY KEY,
    model_id    TEXT UNIQUE NOT NULL,
    model_sha   CHAR(64) NOT NULL,        -- sha256 of checkpoint
    s3_path     TEXT NOT NULL,
    family      TEXT,                      -- e.g., "llama-3", "gemma-2"
    n_params_b  FLOAT,
    registered_at TIMESTAMPTZ DEFAULT NOW()
);

-- Benchmark registry
CREATE TABLE benchmarks (
    id              BIGSERIAL PRIMARY KEY,
    benchmark_name  TEXT NOT NULL,
    benchmark_version CHAR(16) NOT NULL,  -- version hash
    dataset_path    TEXT,
    n_examples      INT,
    primary_metric  TEXT DEFAULT 'accuracy',
    UNIQUE(benchmark_name, benchmark_version)
);

-- Evaluation results (hot path: partitioned by date)
CREATE TABLE eval_results (
    id              BIGSERIAL,
    model_sha       CHAR(64) NOT NULL,
    benchmark_name  TEXT NOT NULL,
    benchmark_version CHAR(16) NOT NULL,
    n_correct       INT NOT NULL,
    n_total         INT NOT NULL,
    accuracy        FLOAT NOT NULL GENERATED ALWAYS AS (n_correct::float / n_total) STORED,
    eval_time_s     FLOAT,
    gpu_type        TEXT,
    evaluated_at    TIMESTAMPTZ DEFAULT NOW(),
    PRIMARY KEY (id, evaluated_at)
) PARTITION BY RANGE (evaluated_at);

CREATE INDEX ON eval_results (model_sha, benchmark_name);
CREATE UNIQUE INDEX ON eval_results (model_sha, benchmark_name, benchmark_version);  -- dedup

-- Baseline cache (updated weekly)
CREATE TABLE accuracy_baselines (
    model_family    TEXT NOT NULL,
    benchmark_name  TEXT NOT NULL,
    baseline_accuracy FLOAT NOT NULL,
    computed_at     TIMESTAMPTZ NOT NULL,
    PRIMARY KEY (model_family, benchmark_name)
);

Eval Worker Implementation

import ray
import torch
from pathlib import Path

@ray.remote(num_gpus=1)
class EvalWorker:
    def __init__(self, model_cache_dir="/tmp/model_cache"):
        self.model_cache_dir = Path(model_cache_dir)
        self.model_cache_dir.mkdir(exist_ok=True)
        self._loaded_model_sha = None
        self._model = None
        self._tokenizer = None
    
    def _load_model_if_needed(self, model_sha, s3_path):
        """Lazy load model; reuse across tasks for the same checkpoint."""
        if self._loaded_model_sha != model_sha:
            local_path = self.model_cache_dir / model_sha
            if not local_path.exists():
                download_from_s3(s3_path, local_path)
            self._model = load_model(local_path)
            self._tokenizer = load_tokenizer(local_path)
            self._loaded_model_sha = model_sha
    
    def run_eval(self, model_sha, s3_path, benchmark_name, n_examples=1000):
        """Run a single (model, benchmark) evaluation."""
        self._load_model_if_needed(model_sha, s3_path)
        
        task = task_registry.get(benchmark_name)
        examples = task.get_examples(max_examples=n_examples)
        
        harness = EvalHarness(self._model, self._tokenizer, device="cuda")
        result = harness.evaluate(task, max_examples=n_examples)
        
        return {
            "model_sha": model_sha,
            "benchmark": benchmark_name,
            "n_correct": int(result.accuracy * result.n_examples),
            "n_total": result.n_examples,
            "accuracy": result.accuracy,
        }

Caching Strategy

Three levels of caching:

Level 1: Result cache (skip re-evaluation)

• Key: (model_sha, benchmark_name, benchmark_version)
• If result exists in DB: skip evaluation entirely
• Typical skip rate: 60-70% (most models unchanged week-to-week)

Level 2: Model cache (avoid re-downloading)

• Workers keep last 3 models in local NVMe storage (LRU eviction)
• Model download: 15-30 GB at 1 GB/s = 15-30s; cached hit = 1-2s
• Cache hit rate: ~80% (same model across multiple benchmarks)

Level 3: Tokenized dataset cache

• Pre-tokenize all benchmark examples once; store as .pt files
• Avoids repeated tokenization (saves ~5-10% eval time)
• Invalidated when benchmark version changes

Regression Detection Algorithm

def detect_regression(model_id, model_family, new_results, db, window_days=7):
    """
    Detect accuracy regressions using McNemar's test.
    new_results: {benchmark_name: accuracy}
    """
    regressions = []
    
    for benchmark, new_acc in new_results.items():
        # Fetch baseline (best accuracy in last window_days)
        baseline = db.query("""
            SELECT AVG(accuracy) as baseline_acc
            FROM eval_results
            WHERE model_sha IN (
                SELECT model_sha FROM models
                WHERE family = %s
                AND registered_at > NOW() - INTERVAL '%s days'
            )
            AND benchmark_name = %s
        """, (model_family, window_days, benchmark))
        
        if baseline is None:
            continue
        
        delta = new_acc - baseline
        
        # Simple threshold check (paired statistical test requires per-example data)
        if delta < -0.01:  # > 1% absolute drop
            regressions.append({
                "benchmark": benchmark,
                "baseline": baseline,
                "new": new_acc,
                "delta": delta,
            })
    
    return regressions

Scalability Analysis

Scaling strategies:

1. Horizontal: Add more GPU workers (Ray auto-scales)
2. Model sharing: Load once, run all benchmarks → amortize load time
3. Parallelism: 60 benchmarks × 50 workers = 1 benchmark per worker simultaneously
4. Early termination: If model is clearly worse (first 100 examples), skip rest
5. Tiered evaluation: Fast tier (100 examples, 3 benchmarks) → full eval if passes

Fault Tolerance