Evaluation Fundamentals › Code Implementation

import warnings

import numpy as np

warnings.filterwarnings("ignore")

# Sample summaries for demonstration
# Reference (gold) summaries and model outputs
references = [
    "The Federal Reserve raised interest rates by 0.25 percentage points on Wednesday, citing persistent inflation above its 2% target.",
    "Scientists discovered a new species of deep-sea fish in the Pacific Ocean at a depth of 3,000 meters, marking the first new discovery of this genus in 40 years.",
    "A major cybersecurity breach exposed personal data of 50 million users at a social media company, raising concerns about data protection regulations.",
    "The city council approved a \$2 billion infrastructure plan to repair roads, bridges, and public transit systems over the next decade.",
    "Researchers published findings showing that regular moderate exercise reduces the risk of cardiovascular disease by 35% in adults over 50.",
]

# Model A outputs (decent summaries)
model_a_outputs = [
    "The Fed increased interest rates by a quarter point due to inflation remaining above its 2 percent target.",
    "A new deep-sea fish species was found in the Pacific at 3,000 meters depth, the first in the genus discovered in four decades.",
    "A data breach at a social media firm exposed information from 50 million users, prompting concerns over data rules.",
    "The city council passed a two billion dollar plan for road, bridge, and transit improvements over 10 years.",
    "A study shows moderate exercise lowers cardiovascular disease risk by 35 percent in people older than 50.",
]

# Model B outputs (less accurate summaries)
model_b_outputs = [
    "Interest rates went up this week as the central bank continues its policy adjustments.",
    "Scientists found a new deep-sea creature in the ocean at great depths during a recent expedition.",
    "Millions of users had their data compromised in a breach at a large online platform.",
    "Local government approved major funding for infrastructure improvements.",
    "Exercise is beneficial for heart health, according to a new research study.",
]

def compute_ngrams(text, n):
    """Extract n-grams from tokenized text."""
    tokens = text.lower().split()
    return [tuple(tokens[i : i + n]) for i in range(len(tokens) - n + 1)]


def rouge_n(hypothesis, reference, n):
    """Compute ROUGE-N precision, recall, and F1."""
    hyp_ngrams = compute_ngrams(hypothesis, n)
    ref_ngrams = compute_ngrams(reference, n)

    if not ref_ngrams:
        return {"precision": 0.0, "recall": 0.0, "f1": 0.0}

    # Count overlapping n-grams
    hyp_counts = {}
    for ng in hyp_ngrams:
        hyp_counts[ng] = hyp_counts.get(ng, 0) + 1

    ref_counts = {}
    for ng in ref_ngrams:
        ref_counts[ng] = ref_counts.get(ng, 0) + 1

    overlap = sum(
        min(hyp_counts.get(ng, 0), ref_counts[ng]) for ng in ref_counts
    )

    precision = overlap / len(hyp_ngrams) if hyp_ngrams else 0.0
    recall = overlap / len(ref_ngrams)
    f1 = (
        (2 * precision * recall / (precision + recall))
        if (precision + recall) > 0
        else 0.0
    )

    return {"precision": precision, "recall": recall, "f1": f1}


# Compute ROUGE-1 and ROUGE-2 for both models
rouge1_a = [
    rouge_n(h, r, 1)["recall"] for h, r in zip(model_a_outputs, references)
]
rouge1_b = [
    rouge_n(h, r, 1)["recall"] for h, r in zip(model_b_outputs, references)
]

rouge2_a = [
    rouge_n(h, r, 2)["recall"] for h, r in zip(model_a_outputs, references)
]
rouge2_b = [
    rouge_n(h, r, 2)["recall"] for h, r in zip(model_b_outputs, references)
]

import re


def extract_numbers(text):
    """Extract numeric values from text for factual coverage."""
    # Match decimals, percentages, and integers
    return set(re.findall(r"\b\d+(?:\.\d+)?(?:\s*%)?", text.lower()))


def factual_coverage(hypothesis, reference):
    """Compute what fraction of key numbers from the reference appear in hypothesis."""
    ref_numbers = extract_numbers(reference)
    if not ref_numbers:
        return 1.0  # No numbers to check

    hyp_numbers = extract_numbers(hypothesis)
    matched = ref_numbers & hyp_numbers
    return len(matched) / len(ref_numbers)


coverage_a = [
    factual_coverage(h, r) for h, r in zip(model_a_outputs, references)
]
coverage_b = [
    factual_coverage(h, r) for h, r in zip(model_b_outputs, references)
]

def bootstrap_confidence_interval(
    scores_a, scores_b, n_bootstrap=2000, confidence=0.95
):
    """Bootstrap confidence interval for the difference in means."""
    np.random.seed(42)
    diffs = []
    n = len(scores_a)

    for _ in range(n_bootstrap):
        indices = np.random.randint(0, n, size=n)
        bootstrap_a = np.mean([scores_a[i] for i in indices])
        bootstrap_b = np.mean([scores_b[i] for i in indices])
        diffs.append(bootstrap_a - bootstrap_b)

    alpha = 1 - confidence
    lower = np.percentile(diffs, 100 * alpha / 2)
    upper = np.percentile(diffs, 100 * (1 - alpha / 2))
    observed_diff = np.mean(scores_a) - np.mean(scores_b)

    return observed_diff, lower, upper


# Compute confidence intervals for ROUGE-1 and factual coverage
r1_diff, r1_lower, r1_upper = bootstrap_confidence_interval(rouge1_a, rouge1_b)
fc_diff, fc_lower, fc_upper = bootstrap_confidence_interval(
    coverage_a, coverage_b
)

from scipy import stats


def required_sample_size(baseline_p, delta, alpha=0.05, power=0.80):
    """
    Compute required sample size to detect a difference of 'delta'
    from 'baseline_p' with given significance and power.

    Parameters:
        baseline_p: Expected performance of the baseline model (0-1)
        delta:      Minimum effect size to detect (0-1)
        alpha:      Significance level (default 0.05)
        power:      Statistical power (default 0.80)
    """
    z_alpha = stats.norm.ppf(1 - alpha / 2)
    z_beta = stats.norm.ppf(power)

    p1 = baseline_p
    p2 = baseline_p + delta
    p_bar = (p1 + p2) / 2

    n = (
        (
            z_alpha * np.sqrt(2 * p_bar * (1 - p_bar))
            + z_beta * np.sqrt(p1 * (1 - p1) + p2 * (1 - p2))
        )
        ** 2
    ) / (delta**2)

    return int(np.ceil(n))


# Show required sample sizes for different scenarios
scenarios = [
    (0.70, 0.05, "Large effect (5pp at 70% baseline)"),
    (0.70, 0.03, "Medium effect (3pp at 70% baseline)"),
    (0.70, 0.01, "Small effect (1pp at 70% baseline)"),
    (0.85, 0.03, "Medium effect (3pp at 85% baseline)"),
    (0.50, 0.05, "Large effect (5pp at 50% baseline)"),
]

from datetime import datetime


def evaluation_report(model_name, metric_results, conditions):
    """Generate a structured evaluation summary."""
    report_lines = []
    report_lines.append("=" * 60)
    report_lines.append(f"EVALUATION REPORT: {model_name}")
    report_lines.append(f"Date: {datetime.now().strftime('%Y-%m-%d')}")
    report_lines.append("=" * 60)

    report_lines.append("\nEVALUATION CONDITIONS:")
    for key, value in conditions.items():
        report_lines.append(f"  {key}: {value}")

    report_lines.append("\nMETRIC RESULTS:")
    for metric, values in metric_results.items():
        mean_val = np.mean(values)
        std_val = np.std(values)
        report_lines.append(f"  {metric}:")
        report_lines.append(f"    Mean:  {mean_val:.3f}")
        report_lines.append(f"    Std:   {std_val:.3f}")
        report_lines.append(f"    Min:   {np.min(values):.3f}")
        report_lines.append(f"    Max:   {np.max(values):.3f}")

    report_lines.append("\nCAVEATS:")
    n = len(list(metric_results.values())[0])
    report_lines.append(
        f"  Sample size: {n} examples (small; results are indicative only)"
    )
    report_lines.append(
        f"  Confidence intervals should be computed before drawing conclusions."
    )

    return "\n".join(report_lines)


conditions = {
    "Task": "News summarization",
    "Decoding temperature": 0.0,
    "Max output tokens": 100,
    "Prompt template": "Summarize the following article in one sentence: {article}",
    "Test set": "Internal news domain (not CNN/DailyMail)",
}