11.3 Building Eval Harnesses¶

# Install with recommended extras
pip install lm-eval[vllm,wandb]

# Evaluate a model on MMLU and HellaSwag with 5-shot prompting
lm_eval \
    --model hf \
    --model_args pretrained=meta-llama/Llama-3.1-8B-Instruct \
    --tasks mmlu,hellaswag \
    --num_fewshot 5 \
    --batch_size auto \
    --output_path results/llama3-8b \
    --log_samples   # save raw predictions for auditing

# Use vLLM backend for large models (much faster)
lm_eval \
    --model vllm \
    --model_args pretrained=meta-llama/Llama-3.1-70B-Instruct,tensor_parallel_size=4 \
    --tasks mmlu_pro \
    --num_fewshot 0 \
    --batch_size 32 \
    --output_path results/llama3-70b

# Evaluate a model that is already being *served* (vLLM or SGLang exposing an
# OpenAI-compatible endpoint). The harness talks HTTP instead of owning the GPU,
# so one server can back many concurrent eval jobs.
lm_eval \
    --model local-completions \
    --model_args model=my-model,base_url=http://localhost:8000/v1/completions,num_concurrent=16,tokenized_requests=False \
    --tasks arc_easy \
    --output_path results/served

# Snippet from a typical generation-based task definition
# (equivalent of what lives inside the harness YAML/Task class)

def doc_to_text(doc):
    """Format a single document into a prompt string."""
    return (
        "Question: " + doc["question"] +
        "\nLet's think step by step.\nAnswer:"
    )

def process_results(doc, results):
    """Extract the numeric answer from a chain-of-thought generation."""
    import re
    # results[0] is the generated string
    gen = results[0]
    # Look for the last number in the generation
    matches = re.findall(r"[-+]?\d*\.?\d+", gen)
    if matches:
        predicted = float(matches[-1])
        gold = float(doc["answer"])
        return {"exact_match": predicted == gold}
    return {"exact_match": 0.0}

# Pseudocode of what the harness does internally for K-shot
def build_prompt(task, doc, k, fewshot_docs):
    """Build a K-shot prompt for a document."""
    parts = []

    # System prompt (if the task defines one)
    if task.has_system_prompt():
        parts.append(task.system_prompt())

    # K few-shot examples
    for fewshot_doc in fewshot_docs[:k]:
        # doc_to_text formats the question
        # doc_to_target formats the gold answer
        parts.append(task.doc_to_text(fewshot_doc) +
                     task.doc_to_target(fewshot_doc))

    # The actual test document (no answer appended)
    parts.append(task.doc_to_text(doc))

    return task.fewshot_delimiter().join(parts)

# Install HELM
pip install crfm-helm

# Run a subset of scenarios
helm-run \
    --conf-path src/helm/benchmark/presentation/run_specs_lite.conf \
    --suite my_eval \
    --max-eval-instances 500 \
    --num-threads 4 \
    --models-to-run meta/llama-3-8b

# Summarize results
helm-summarize --suite my_eval

# Start the web UI
helm-server --suite my_eval

from transformers import AutoTokenizer

tokenizer = AutoTokenizer.from_pretrained("meta-llama/Llama-3.1-8B-Instruct")

question = "What is the capital of France?"
choices = ["London", "Berlin", "Paris", "Madrid"]

# WRONG: raw concatenation (no chat template)
raw_prompt = f"Question: {question}\nChoices: {choices}\nAnswer:"

# RIGHT: use the model's chat template
messages = [
    {
        "role": "system",
        "content": "You are a helpful assistant. Answer with the letter only."
    },
    {
        "role": "user",
        "content": (
            f"{question}\n"
            + "\n".join(f"{chr(65+i)}) {c}" for i, c in enumerate(choices))
        )
    }
]

# apply_chat_template adds special tokens and role delimiters
templated_prompt = tokenizer.apply_chat_template(
    messages,
    tokenize=False,        # return string, not token ids
    add_generation_prompt=True  # append the assistant turn starter
)
print(templated_prompt)
# <|begin_of_text|><|start_header_id|>system<|end_header_id|>
# You are a helpful assistant. Answer with the letter only.<|eot_id|>
# <|start_header_id|>user<|end_header_id|>
# What is the capital of France?
# A) London
# B) Berlin
# C) Paris
# D) Madrid<|eot_id|>
# <|start_header_id|>assistant<|end_header_id|>

import re
import string

def normalize_answer(s: str) -> str:
    """Normalize a string answer for comparison.

    Follows the same normalization as the SQuAD evaluation script,
    used widely in QA benchmarks.
    """
    # Lowercase
    s = s.lower()
    # Remove articles
    s = re.sub(r"\b(a|an|the)\b", " ", s)
    # Remove punctuation
    s = "".join(ch for ch in s if ch not in set(string.punctuation))
    # Collapse whitespace
    s = " ".join(s.split())
    return s

def extract_mc_answer(generation: str, choices: list[str]) -> str | None:
    """Extract a multiple-choice answer from a free-form generation.

    Handles both letter-based ("A", "B") and text-based answers.
    Returns the matched choice text, or None if no match.
    """
    gen = generation.strip()

    # Try to match a leading letter like "A" or "A."
    letter_match = re.match(r"^([A-Da-d])[\.\):\s]?", gen)
    if letter_match:
        idx = ord(letter_match.group(1).upper()) - ord("A")
        if 0 <= idx < len(choices):
            return choices[idx]

    # Try to match the choice text directly
    gen_norm = normalize_answer(gen)
    for choice in choices:
        if normalize_answer(choice) in gen_norm:
            return choice

    return None

{"id": "medq_001", "question": "Which enzyme is deficient in classic phenylketonuria?", "choices": ["Tyrosine hydroxylase", "Phenylalanine hydroxylase", "Homogentisate oxidase", "DOPA decarboxylase"], "answer": 1}

# tasks/medqa_custom.yaml
task: medqa_custom
dataset_path: json                    # HuggingFace datasets loader
dataset_kwargs:
  data_files:
    test: data/medqa_test.jsonl
test_split: test
output_type: multiple_choice          # use log-likelihood scoring over choices
doc_to_text: "Question: {{question}}\n"
doc_to_choice: "{{choices}}"         # the choices list field
doc_to_target: "{{answer}}"          # integer index of the correct choice
metric_list:
  - metric: acc
    aggregation: mean
    higher_is_better: true
  - metric: acc_norm
    aggregation: mean
    higher_is_better: true
num_fewshot: 0

# Point the harness at your custom task directory
lm_eval \
    --model hf \
    --model_args pretrained=mistralai/Mistral-7B-v0.1 \
    --tasks medqa_custom \
    --include_path ./tasks \
    --num_fewshot 0 \
    --output_path results/medqa

# custom_tasks/medner/task.py
"""Custom NER task for the lm-evaluation-harness.

Evaluates medical named entity recognition as exact-match entity-level F1.
"""
from lm_eval.api.task import ConfigurableTask
from lm_eval.api.metrics import mean


def entity_f1(items):
    """Compute macro-averaged entity-level F1 over a list of (pred, gold) pairs.

    Args:
        items: list of (prediction_set, gold_set) tuples, where each set
               contains normalized entity strings.

    Returns:
        Macro-averaged F1 score in [0, 1].
    """
    f1_scores = []
    for pred_set, gold_set in items:
        if not gold_set:
            # No gold entities: perfect if model also predicts nothing
            f1_scores.append(1.0 if not pred_set else 0.0)
            continue

        true_pos = len(pred_set & gold_set)
        precision = true_pos / len(pred_set) if pred_set else 0.0
        recall    = true_pos / len(gold_set)

        if precision + recall == 0:
            f1_scores.append(0.0)
        else:
            f1_scores.append(2 * precision * recall / (precision + recall))

    return sum(f1_scores) / len(f1_scores)


class MedNERTask(ConfigurableTask):
    """Medical NER task with entity-level F1 metric."""

    VERSION = 1
    DATASET_PATH = "json"
    DATASET_NAME = None

    def doc_to_text(self, doc):
        return f"Extract all medical entities from this text:\n{doc['text']}\nEntities:"

    def doc_to_target(self, doc):
        # Return the list as a comma-separated string for generation
        return ", ".join(doc["entities"])

    def process_results(self, doc, results):
        # Parse the generation into a set of entities
        generation = results[0].strip()
        pred_entities = {
            e.strip().lower()
            for e in generation.split(",")
            if e.strip()
        }
        gold_entities = {e.lower() for e in doc["entities"]}
        return {
            "entity_f1": (pred_entities, gold_entities)
        }

    def aggregation(self):
        return {"entity_f1": entity_f1}

    def higher_is_better(self):
        return {"entity_f1": True}

# eval_probe.py -- run a fast benchmark probe from inside the training loop
import lm_eval
from lm_eval.models.huggingface import HFLM

# `model` is a live nn.Module (already on GPU, in eval mode); `tokenizer` is its
# tokenizer. Nothing is written to disk, so this works on an unsaved checkpoint.
lm = HFLM(pretrained=model, tokenizer=tokenizer, batch_size=16)

out = lm_eval.simple_evaluate(
    model=lm,
    tasks=["hellaswag", "piqa", "arc_easy", "winogrande", "lambada_openai"],
    num_fewshot=0,
    limit=500,           # subsample: a probe, not a leaderboard run
    bootstrap_iters=0,   # skip bootstrap stderr; it dominates runtime at limit=500
)
for task, metrics in out["results"].items():
    # metric keys are "<name>,<filter>", e.g. "acc,none" / "acc_norm,none"
    print(task, {k: round(v, 4) for k, v in metrics.items()
                 if isinstance(v, float) and k.startswith("acc")})

"""
minimal_harness.py

A self-contained eval harness for multiple-choice tasks.
Handles: batching, log-likelihood scoring (raw + normalized),
few-shot assembly, deterministic seeds, and JSON result output.

Usage:
    python minimal_harness.py \
        --model meta-llama/Llama-3.1-8B \
        --task_file data/task.jsonl \
        --num_fewshot 5 \
        --batch_size 8 \
        --output results.json
"""

import argparse
import json
import random
import torch
import numpy as np
from pathlib import Path
from dataclasses import dataclass, asdict
from typing import Optional
from transformers import AutoModelForCausalLM, AutoTokenizer


# ─────────────────────────────────────────────────────────────────────────────
# Data structures
# ─────────────────────────────────────────────────────────────────────────────

@dataclass
class MCDoc:
    id: str
    question: str
    choices: list[str]
    answer: int  # 0-indexed correct choice


@dataclass
class MCResult:
    id: str
    acc: int        # 1 if raw argmax is correct
    acc_norm: int   # 1 if length-normalized argmax is correct
    predicted_raw: int
    predicted_norm: int
    gold: int
    log_likelihoods: list[float]
    log_likelihoods_norm: list[float]


# ─────────────────────────────────────────────────────────────────────────────
# Prompt construction
# ─────────────────────────────────────────────────────────────────────────────

def format_doc(doc: MCDoc, include_answer: bool = False) -> str:
    """Format a single document as a prompt string."""
    letters = "ABCD"
    lines = [f"Question: {doc.question}"]
    for i, choice in enumerate(doc.choices):
        lines.append(f"{letters[i]}) {choice}")
    prompt = "\n".join(lines) + "\nAnswer:"
    if include_answer:
        # Append the correct answer letter for few-shot examples
        prompt += f" {letters[doc.answer]}"
    return prompt


def build_fewshot_prompt(
    test_doc: MCDoc,
    fewshot_docs: list[MCDoc],
    k: int
) -> str:
    """Build a K-shot prompt for a test document.

    Returns: (prefix, continuation) where prefix is the full context
    and continuation is the answer text to score.
    """
    shots = fewshot_docs[:k]
    parts = [format_doc(d, include_answer=True) for d in shots]
    parts.append(format_doc(test_doc, include_answer=False))
    return "\n\n".join(parts)


# ─────────────────────────────────────────────────────────────────────────────
# Log-likelihood computation
# ─────────────────────────────────────────────────────────────────────────────

@torch.no_grad()
def score_choices(
    model: AutoModelForCausalLM,
    tokenizer: AutoTokenizer,
    context: str,
    choices: list[str],
    device: str = "cuda"
) -> tuple[list[float], list[float]]:
    """Compute raw and length-normalized log-likelihoods for each choice.

    The trick: for each choice we tokenize [context + choice] and compute
    the sum of log-probs over only the choice tokens, not the context.

    Returns:
        log_liks: raw sum of log-probs per choice
        log_liks_norm: log_liks divided by choice token count
    """
    letters = "ABCD"
    # Encode the context WITHOUT special tokens so its ids are exactly a
    # prefix of the full [context + choice] encoding below. Letting the
    # default add_special_tokens=True prepend a BOS here (as Llama/Mistral
    # tokenizers do) would make context_len already include that BOS; the
    # "+1 for BOS" added below then double-counts it and shifts the split by
    # one token. For single-token continuations (" A".." D") that leaves
    # cont_ids empty, so every choice scores 0.0, argmax always returns
    # choice 0, and the harness silently reports ~25% for Llama-3/Mistral.
    context_ids = tokenizer.encode(
        context, return_tensors="pt", add_special_tokens=False
    ).to(device)
    context_len = context_ids.shape[1]

    log_liks = []
    log_liks_norm = []

    for i, choice in enumerate(choices):
        # The model expects " A", " B", etc. as the continuation
        # (leading space is important for tokenization)
        continuation = f" {letters[i]}"
        full_text = context + continuation
        full_ids = tokenizer.encode(
            full_text, return_tensors="pt", add_special_tokens=False
        ).to(device)

        # We need to include at least context_ids plus continuation tokens
        # Prepend BOS if the tokenizer uses it
        if tokenizer.bos_token_id is not None:
            bos = torch.tensor([[tokenizer.bos_token_id]], device=device)
            full_ids = torch.cat([bos, full_ids], dim=1)
            ctx_len = context_ids.shape[1] + 1  # +1 for BOS
        else:
            ctx_len = context_len

        # Forward pass; model returns logits at each position
        outputs = model(full_ids)
        logits = outputs.logits  # (1, seq_len, vocab_size)

        # Shift: logits[t] predicts token[t+1]
        # We want log-probs for tokens from ctx_len onward
        log_probs = torch.log_softmax(logits[0], dim=-1)  # (seq_len, vocab)

        # Continuation tokens are at positions [ctx_len, seq_len)
        # Their log-probs are at positions [ctx_len-1, seq_len-1)
        cont_ids = full_ids[0, ctx_len:]  # the actual continuation token ids

        # Sanity check: each choice must contribute >= 1 continuation token.
        # An empty split means the context/continuation boundary is wrong --
        # usually a BOS double-count, or a tokenizer that merged the last
        # context token with the leading-space continuation token.
        assert len(cont_ids) >= 1, (
            f"empty continuation for choice {letters[i]!r} "
            f"(ctx_len={ctx_len}, full_len={full_ids.shape[1]}); "
            "check the context/continuation token split"
        )
        cont_log_probs = log_probs[ctx_len - 1: ctx_len - 1 + len(cont_ids)]

        # Gather the log-prob of each actual token
        token_log_probs = cont_log_probs.gather(
            1, cont_ids.unsqueeze(1)
        ).squeeze(1)

        ll = token_log_probs.sum().item()
        n_tokens = len(cont_ids)

        log_liks.append(ll)
        log_liks_norm.append(ll / n_tokens if n_tokens > 0 else ll)

    return log_liks, log_liks_norm


# ─────────────────────────────────────────────────────────────────────────────
# Main evaluation loop
# ─────────────────────────────────────────────────────────────────────────────

def evaluate(
    model_name: str,
    task_file: str,
    num_fewshot: int = 0,
    batch_size: int = 1,
    seed: int = 42,
    output_path: Optional[str] = None,
    device: str = "cuda",
) -> dict:
    """Run the full evaluation and return a results dict."""

    # Deterministic reproducibility: fix all relevant seeds
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    torch.cuda.manual_seed_all(seed)

    # Load task documents
    docs = []
    with open(task_file) as f:
        for line in f:
            d = json.loads(line)
            docs.append(MCDoc(**d))

    # Few-shot exemplars come from a held-in pool (the first slice of docs); the
    # rest are the eval set. Cap the pool at half the data so the eval set is
    # never empty on small sets, and reserve nothing for 0-shot.
    # In a real harness, the pool is the training split, never the test split.
    n_fewshot_pool = min(num_fewshot * 5, len(docs) // 2) if num_fewshot > 0 else 0
    fewshot_pool = docs[:n_fewshot_pool]
    eval_docs = docs[n_fewshot_pool:]

    # Sample fixed few-shot examples per evaluation instance
    rng = random.Random(seed)
    fewshot_docs = rng.sample(fewshot_pool, min(num_fewshot, len(fewshot_pool)))

    # Load model and tokenizer
    print(f"Loading {model_name}…")
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    if tokenizer.pad_token is None:
        tokenizer.pad_token = tokenizer.eos_token

    model = AutoModelForCausalLM.from_pretrained(
        model_name,
        torch_dtype=torch.bfloat16,  # BF16 is the safe default for LLMs
        device_map="auto",
    )
    model.eval()

    # Evaluate each document
    results = []
    for i, doc in enumerate(eval_docs):
        if i % 100 == 0:
            print(f"  [{i}/{len(eval_docs)}] …")

        context = build_fewshot_prompt(doc, fewshot_docs, num_fewshot)
        ll, ll_norm = score_choices(model, tokenizer, context, doc.choices, device)

        pred_raw = int(np.argmax(ll))
        pred_norm = int(np.argmax(ll_norm))

        results.append(MCResult(
            id=doc.id,
            acc=int(pred_raw == doc.answer),
            acc_norm=int(pred_norm == doc.answer),
            predicted_raw=pred_raw,
            predicted_norm=pred_norm,
            gold=doc.answer,
            log_likelihoods=ll,
            log_likelihoods_norm=ll_norm,
        ))

    # Aggregate
    acc    = np.mean([r.acc for r in results])
    acc_norm = np.mean([r.acc_norm for r in results])
    n = len(results)

    # Standard error: SE = sqrt(p*(1-p)/n) for a Bernoulli proportion
    se    = np.sqrt(acc * (1 - acc) / n)
    se_norm = np.sqrt(acc_norm * (1 - acc_norm) / n)

    summary = {
        "model": model_name,
        "task": task_file,
        "num_fewshot": num_fewshot,
        "seed": seed,
        "n_docs": n,
        "acc": round(float(acc), 4),
        "acc_se": round(float(se), 4),
        "acc_norm": round(float(acc_norm), 4),
        "acc_norm_se": round(float(se_norm), 4),
        "samples": [asdict(r) for r in results],  # full sample log
    }

    if output_path:
        with open(output_path, "w") as f:
            json.dump(summary, f, indent=2)
        print(f"Results written to {output_path}")

    print(f"\nacc={acc:.4f} ±{se:.4f}  acc_norm={acc_norm:.4f} ±{se_norm:.4f}")
    return summary


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--model", required=True)
    parser.add_argument("--task_file", required=True)
    parser.add_argument("--num_fewshot", type=int, default=0)
    parser.add_argument("--batch_size", type=int, default=1)
    parser.add_argument("--seed", type=int, default=42)
    parser.add_argument("--output", default=None)
    args = parser.parse_args()

    evaluate(
        model_name=args.model,
        task_file=args.task_file,
        num_fewshot=args.num_fewshot,
        batch_size=args.batch_size,
        seed=args.seed,
        output_path=args.output,
    )

"""smoke_test.py — end-to-end check for minimal_harness.py"""
import json
import tempfile
from pathlib import Path

from minimal_harness import evaluate, MCDoc, build_fewshot_prompt

def make_toy_dataset() -> list[dict]:
    """8 capital-cities questions, 4 plausible choices each."""
    items = [
        ("France", ["Paris", "Lyon", "Marseille", "Nice"], 0),
        ("Japan", ["Osaka", "Kyoto", "Tokyo", "Nagoya"], 2),
        ("Italy", ["Milan", "Rome", "Naples", "Turin"], 1),
        ("Egypt", ["Alexandria", "Giza", "Luxor", "Cairo"], 3),
        ("Canada", ["Toronto", "Ottawa", "Vancouver", "Montreal"], 1),
        ("Brazil", ["Rio de Janeiro", "Sao Paulo", "Brasilia", "Salvador"], 2),
        ("Germany", ["Munich", "Hamburg", "Berlin", "Cologne"], 2),
        ("Russia", ["St. Petersburg", "Moscow", "Novosibirsk", "Kazan"], 1),
    ]
    return [
        {"id": f"cap_{i}", "question": f"What is the capital of {country}?",
         "choices": choices, "answer": answer}
        for i, (country, choices, answer) in enumerate(items)
    ]

def smoke_test():
    docs = make_toy_dataset()
    with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
        for d in docs:
            f.write(json.dumps(d) + "\n")
        task_file = f.name

    # gpt2 (124M) runs in seconds on CPU. evaluate() hardcodes bf16 +
    # device_map="auto"; on a pure-CPU machine also switch torch_dtype to
    # float32 in evaluate() (bf16 matmuls are slow/unsupported on many CPUs).
    summary = evaluate(
        model_name="gpt2", task_file=task_file, num_fewshot=0, device="cpu"
    )

    # Invariant (b): single-token choices -> acc and acc_norm must agree exactly.
    assert summary["acc"] == summary["acc_norm"]

    # Invariant (a): four distinct, strictly negative log-likelihoods per item.
    null_count = 0
    for s in summary["samples"]:
        lls = s["log_likelihoods"]
        assert len(set(lls)) == 4, f"non-distinct log-likelihoods: {lls}"
        assert all(x < 0.0 for x in lls) and 0.0 not in lls, f"non-negative/zero ll: {lls}"
        if s["predicted_raw"] is None:  # structurally never true: argmax over 4 finite values
            null_count += 1
    null_rate = null_count / len(summary["samples"])
    assert null_rate == 0.0
    print(f"null_rate={null_rate:.4f}  acc={summary['acc']:.4f}  acc_norm={summary['acc_norm']:.4f}")
    return summary

if __name__ == "__main__":
    smoke_test()

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
from minimal_harness import MCDoc, build_fewshot_prompt
from smoke_test import smoke_test

# smoke_test() runs the harness on the toy set and returns the summary dict:
summary = smoke_test()

tok = AutoTokenizer.from_pretrained('gpt2')
m = AutoModelForCausalLM.from_pretrained('gpt2').eval()

# With num_fewshot=0 the harness evaluates every doc in order, so samples[0] is
# the first toy question (capital of France). Rebuild that MCDoc explicitly so
# this check is self-contained (eval_docs is internal to evaluate()).
doc0 = MCDoc(id="cap_0", question="What is the capital of France?",
             choices=["Paris", "Lyon", "Marseille", "Nice"], answer=0)
ctx = build_fewshot_prompt(doc0, [], 0)   # 0-shot context, matches score_choices
ids = [tok.bos_token_id] + tok.encode(ctx, add_special_tokens=False)
pred = summary['samples'][0]['predicted_raw']
letter_ids = tok.encode(f' {chr(65 + pred)}', add_special_tokens=False)
assert len(letter_ids) == 1            # ' A'..' D' are single gpt2 tokens
with torch.no_grad():
    logits = m(torch.tensor([ids])).logits
ref_ll = torch.log_softmax(logits[0, -1], dim=-1)[letter_ids[0]].item()
# Must match the harness-recorded log-likelihood for the chosen letter:
assert abs(ref_ll - summary['samples'][0]['log_likelihoods'][pred]) < 1e-4

MUST record for every eval run:
  ✓ Model name + revision hash (not just "llama3-8b")
  ✓ Harness version + commit SHA (lm_eval.__version__ is not enough)
  ✓ Task version numbers (each task has a VERSION field)
  ✓ num_fewshot and the exact few-shot seed
  ✓ Prompt template (or YAML hash)
  ✓ Whether apply_chat_template / fewshot_as_multiturn were used
  ✓ Scoring mode: MCQ-loglikelihood vs cloze-loglikelihood vs generation
  ✓ Length normalization: none / bytes / tokens
  ✓ Random seed for sampling (--gen_kwargs seed=X)
  ✓ Hardware (GPU model, driver, CUDA version)
  ✓ Floating-point format (BF16 vs FP16 vs FP32)
  ✓ Whether --limit was used (partial eval is not full eval)
  ✓ Dataset split and subset (MMLU has 57 subjects)

# lm_eval produces this structure under --output_path:
results/
├── results.json          # aggregate scores by task and metric
├── samples_mmlu_0.jsonl  # per-sample log for mmlu, seed 0
└── samples_hellaswag_0.jsonl

// One line from samples_mmlu_0.jsonl
{
  "doc_id": 1234,
  "doc": {"question": "...", "choices": [...], "answer": 2},
  "target": 2,
  "arguments": [["context + choice A"], ["context + choice B"], ...],
  "resps": [[-3.12, false], [-1.87, false], [-1.42, true], [-4.01, false]],
  "filtered_resps": [[-3.12, false], [-1.87, false], [-1.42, true], [-4.01, false]],
  "acc": 1.0,
  "acc_norm": 1.0
}

from scipy.stats import chi2, binomtest

def mcnemar_test(results_a: list[int], results_b: list[int]) -> dict:
    """McNemar's test for two paired binary result sequences.

    Uses the exact two-sided binomial test when discordant pairs are few
    (< 25) and the continuity-corrected chi-square approximation otherwise.
    See 11.6 (Statistical Rigor) for the full derivation and reference code.
    """
    assert len(results_a) == len(results_b), "Must be paired on the same docs"

    n_01 = sum(a == 0 and b == 1 for a, b in zip(results_a, results_b))  # B right, A wrong
    n_10 = sum(a == 1 and b == 0 for a, b in zip(results_a, results_b))  # A right, B wrong
    n_disc = n_01 + n_10

    if n_disc == 0:                    # models never disagree -> no evidence
        statistic, p_value = None, 1.0
    elif n_disc < 25:                  # too few pairs for chi-square: exact test
        # Under H0 each discordant pair is a fair coin: n_01 ~ Binomial(n_disc, 0.5).
        statistic = None
        p_value = binomtest(n_01, n_disc, 0.5, alternative="two-sided").pvalue
    else:
        # Yates continuity correction, clamped at 0 so n_01 == n_10 gives
        # exactly 0 (an uncorrected (|0|-1)**2 would report a spurious 1/n_disc).
        statistic = max(abs(n_01 - n_10) - 1, 0) ** 2 / n_disc
        p_value = float(chi2.sf(statistic, df=1))

    return {
        "chi2": round(statistic, 4) if statistic is not None else None,
        "p_value": round(p_value, 4),
        "n_01": n_01,  # B correct, A wrong
        "n_10": n_10,  # A correct, B wrong
        "acc_a": sum(results_a) / len(results_a),
        "acc_b": sum(results_b) / len(results_b),
    }

# Example usage
results_a = [1, 0, 1, 1, 0, 1, 0, 0, 1, 1]  # model A correct/wrong
results_b = [1, 1, 1, 0, 0, 1, 1, 0, 1, 0]  # model B correct/wrong
print(mcnemar_test(results_a, results_b))
# Only 4 discordant pairs (n_disc = 4 < 25), so the exact binomial test runs
# and no chi-square statistic is reported:
# {'chi2': None, 'p_value': 1.0, 'n_01': 2, 'n_10': 2, 'acc_a': 0.6, 'acc_b': 0.6}

import numpy as np
from typing import Callable

def bootstrap_ci(
    scores: list[float],
    metric_fn: Callable[[list[float]], float] = np.mean,
    n_bootstrap: int = 10_000,
    alpha: float = 0.05,
    seed: int = 42,
) -> tuple[float, float, float]:
    """Compute a bootstrap confidence interval for any metric.

    Args:
        scores: per-example metric values (e.g., list of 0/1 for accuracy)
        metric_fn: aggregation function (default: mean)
        n_bootstrap: number of bootstrap resamples
        alpha: significance level (0.05 → 95% CI)
        seed: random seed for reproducibility

    Returns:
        (point_estimate, lower_bound, upper_bound)
    """
    rng = np.random.default_rng(seed)
    arr = np.array(scores)

    point_estimate = metric_fn(arr)

    # Resample with replacement n_bootstrap times
    boot_stats = []
    for _ in range(n_bootstrap):
        resample = rng.choice(arr, size=len(arr), replace=True)
        boot_stats.append(metric_fn(resample))

    boot_stats = np.array(boot_stats)
    lower = np.percentile(boot_stats, 100 * alpha / 2)
    upper = np.percentile(boot_stats, 100 * (1 - alpha / 2))

    return point_estimate, lower, upper

# Example: accuracy scores for 500 examples
scores = [1] * 370 + [0] * 130  # 74% accuracy
pt, lo, hi = bootstrap_ci(scores, n_bootstrap=10_000, seed=42)
print(f"acc = {pt:.3f}  95% CI: [{lo:.3f}, {hi:.3f}]")
# acc = 0.740  95% CI: [0.703, 0.776]

# .github/workflows/eval.yml  (GitHub Actions)
name: Model Evaluation

on:
  push:
    paths:
      - "checkpoints/**"
  schedule:
    - cron: "0 4 * * *"   # nightly eval at 04:00 UTC

jobs:
  evaluate:
    runs-on: [self-hosted, gpu]
    steps:
      - uses: actions/checkout@v4

      - name: Install dependencies
        run: pip install lm-eval[vllm]==0.4.3  # pin version for reproducibility

      - name: Run core benchmark suite
        run: |
          lm_eval \
            --model vllm \
            --model_args pretrained=${{ env.MODEL_PATH }} \
            --tasks mmlu,hellaswag,arc_challenge,winogrande \
            --num_fewshot 5 \
            --batch_size 32 \
            --seed 1234 \
            --output_path eval_results/${{ github.sha }}

      - name: Check regression threshold
        run: |
          # Fail the pipeline if any task regresses > 0.5%
          python scripts/check_regression.py \
            --baseline results/baseline.json \
            --current eval_results/${{ github.sha }}/results.json \
            --threshold 0.005

      - name: Upload results to tracking system
        run: |
          python scripts/log_to_mlflow.py \
            --run_name ${{ github.sha }} \
            --results_dir eval_results/${{ github.sha }}