Модуль 13. Оценивание RL-систем для LLM: статистика, взлом награды и безопасность

from __future__ import annotations

import json
import math
import sys
import tempfile
from pathlib import Path

import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from IPython.display import display


def find_repo_root(start: Path) -> Path:
    start = start.resolve()
    for candidate in (start, *start.parents):
        if (candidate / "modules/module-13/module_13_reference.py").is_file():
            return candidate
    raise FileNotFoundError("Could not locate the RL for LLM repository root")


ROOT = find_repo_root(Path.cwd())
MODULE_DIR = ROOT / "modules/module-13"
if str(MODULE_DIR) not in sys.path:
    sys.path.insert(0, str(MODULE_DIR))

from module_13_reference import (
    contamination_report,
    exact_mcnemar_pvalue,
    hierarchical_bootstrap_interval,
    independent_trial_success,
    judge_swap_audit,
    ngram_containment,
    ngram_jaccard,
    paired_bootstrap_interval,
    pass_at_k_unbiased,
    position_audit,
    release_decision,
    reward_pressure_experiment,
    safety_rates,
    selection_success,
    validate_measurement_passport,
    wilson_interval,
    worst_case_binomial_sample_size,
)
from module_13_evaluation_harness import (
    build_synthetic_dataset,
    default_systems,
    evaluate_system,
    run_course_experiment,
)

SEED = 13
PROFILE = "smoke"  # change to "full" for more bootstrap replicates
BOOTSTRAP_REPLICATES = 4_000 if PROFILE == "smoke" else 20_000
OUTPUT_DIR = Path(tempfile.mkdtemp(prefix="rlm13-evidence-"))

np.set_printoptions(precision=4, suppress=True)
print("repository:", ROOT)
print("profile:", PROFILE)
print("temporary evidence directory:", OUTPUT_DIR)

examples = [(62, 100), (0, 20), (100, 100)]
rows = []
for k, n in examples:
    interval = wilson_interval(k, n)
    rows.append({
        "successes": k,
        "total": n,
        "estimate": interval.estimate,
        "low": interval.low,
        "high": interval.high,
    })
display(pd.DataFrame(rows))

for margin in (0.05, 0.03, 0.01):
    print(f"worst-case n for ±{margin:.0%}:", worst_case_binomial_sample_size(margin))

assert 0.15 < wilson_interval(0, 20).high < 0.17
assert worst_case_binomial_sample_size(0.03) == 1068

experiment = run_course_experiment(OUTPUT_DIR, seed=SEED, n_tasks=240)
summary_table = pd.DataFrame([
    {
        "system": "baseline",
        "accuracy": experiment["baseline"]["accuracy"],
        "wilson_low": experiment["baseline"]["accuracy_low"],
        "wilson_high": experiment["baseline"]["accuracy_high"],
        "technical_failure_rate": experiment["baseline"]["technical_failure_rate"],
        "mean_tokens": experiment["baseline"]["mean_tokens"],
        "estimated_cost": experiment["baseline"]["estimated_cost"],
    },
    {
        "system": "candidate",
        "accuracy": experiment["candidate"]["accuracy"],
        "wilson_low": experiment["candidate"]["accuracy_low"],
        "wilson_high": experiment["candidate"]["accuracy_high"],
        "technical_failure_rate": experiment["candidate"]["technical_failure_rate"],
        "mean_tokens": experiment["candidate"]["mean_tokens"],
        "estimated_cost": experiment["candidate"]["estimated_cost"],
    },
])
display(summary_table)

print("paired comparison:")
display(pd.Series(experiment["paired"]))
print("evidence files:", sorted(path.name for path in OUTPUT_DIR.iterdir()))
display(pd.Series(experiment["decision"]))

samples = build_synthetic_dataset(n=240, seed=SEED)
baseline_config, candidate_config = default_systems()
baseline_records = evaluate_system(samples, baseline_config, seed=SEED)
candidate_records = evaluate_system(samples, candidate_config, seed=SEED)

baseline_scores = np.asarray([record.score.score for record in baseline_records], dtype=float)
candidate_scores = np.asarray([record.score.score for record in candidate_records], dtype=float)
paired = paired_bootstrap_interval(
    candidate_scores,
    baseline_scores,
    replicates=BOOTSTRAP_REPLICATES,
    seed=SEED,
)

fig, ax = plt.subplots(figsize=(8.5, 4.6))
ax.bar(["baseline", "candidate"], [baseline_scores.mean(), candidate_scores.mean()])
ax.set_ylim(0, 1)
ax.set_ylabel("accuracy")
ax.set_title("Paired systems on identical sample_id values")
for i, value in enumerate([baseline_scores.mean(), candidate_scores.mean()]):
    ax.text(i, value + 0.025, f"{value:.3f}", ha="center")
plt.show()

print("paired difference interval:", paired)
print("exact McNemar p-value:", exact_mcnemar_pvalue(
    experiment["paired"]["candidate_only"],
    experiment["paired"]["baseline_only"],
))

rng = np.random.default_rng(SEED)
prompt_difficulty = rng.normal(0.0, 0.9, size=40)
# Eight correlated completions share one prompt-level difficulty.
noise_a = rng.normal(0.0, 0.75, size=(40, 8))
noise_b = rng.normal(0.0, 0.75, size=(40, 8))
scores_a = (prompt_difficulty[:, None] + 0.25 + noise_a > 0).astype(float)
scores_b = (prompt_difficulty[:, None] + noise_b > 0).astype(float)

naive = paired_bootstrap_interval(
    scores_a.ravel(), scores_b.ravel(),
    replicates=BOOTSTRAP_REPLICATES,
    seed=SEED,
)
clustered = hierarchical_bootstrap_interval(
    scores_a, scores_b,
    replicates=max(1_000, BOOTSTRAP_REPLICATES // 2),
    seed=SEED,
    resample_within_prompt=True,
)

display(pd.DataFrame([
    {"method": "rows treated as independent", "estimate": naive.estimate, "low": naive.low, "high": naive.high},
    {"method": "hierarchical prompt bootstrap", "estimate": clustered.estimate, "low": clustered.low, "high": clustered.high},
]))
print("The primary unit determines what population the interval describes.")

ks = np.arange(1, 65)
iid_curve = np.asarray([independent_trial_success(0.05, int(k)) for k in ks])
finite_ks = np.arange(1, 21)
finite_curve = np.asarray([pass_at_k_unbiased(20, 1, int(k)) for k in finite_ks])

fig, ax = plt.subplots(figsize=(9, 4.8))
ax.plot(ks, iid_curve, label="new IID attempts, p=0.05")
ax.plot(finite_ks, finite_curve, label="finite pool n=20, c=1")
ax.set_xlabel("k")
ax.set_ylabel("coverage probability / estimator")
ax.set_ylim(0, 1.03)
ax.set_title("Coverage is not selector accuracy")
ax.legend()
plt.show()

pool = [False, True, False]
print("successful candidate exists:", any(pool))
print("selected index 0 is correct:", selection_success(pool, 0))
assert any(pool) and not selection_success(pool, 0)

benchmark = "calculate the exact paired difference for the shared evaluation tasks"
literal_corpus = "notes: calculate the exact paired difference for the shared evaluation tasks today"
paraphrase_corpus = "estimate how two systems differ when both are run on identical examples"

rows = []
for label, text in [("literal", literal_corpus), ("paraphrase", paraphrase_corpus)]:
    rows.append({
        "corpus item": label,
        "4-gram containment": ngram_containment(benchmark, text, 4),
        "4-gram Jaccard": ngram_jaccard(benchmark, text, 4),
    })
display(pd.DataFrame(rows))

report = contamination_report(
    [benchmark, "a completely unrelated safety item", "a translated or paraphrased derivative"],
    [literal_corpus, "other training document"],
    n=4,
    threshold=0.55,
)
print(json.dumps(report, indent=2, ensure_ascii=False))
print("No literal flag is not a cleanliness certificate.")

judge = experiment["judge"]
identity = judge["identity"]
position = judge["position"]
display(pd.DataFrame([
    {"diagnostic": "candidate identity consistency", "value": identity["consistency_rate"]},
    {"diagnostic": "first-position rate among non-ties", "value": position["first_position_rate_non_ties"]},
]))

# A tiny direct example makes the label contracts visible.
identity_small = judge_swap_audit(["A", "A", "B", "TIE"], ["A", "B", "B", "TIE"])
position_small = position_audit(["FIRST", "SECOND"], ["FIRST", "FIRST"])
print("identity audit:", identity_small)
print("position audit:", position_small)
assert identity_small.consistent == 3
assert position_small.first_position_wins == 3

pressure = reward_pressure_experiment((1, 4, 16, 64, 256, 512), trials=6_000, seed=17)
pressure_df = pd.DataFrame([
    {"candidate_count": n, "proxy": proxy, "independent_goal": goal}
    for n, (proxy, goal) in pressure.items()
])
display(pressure_df)

fig, ax = plt.subplots(figsize=(9, 4.8))
ax.plot(pressure_df["candidate_count"], pressure_df["proxy"], marker="o", label="proxy")
ax.plot(pressure_df["candidate_count"], pressure_df["independent_goal"], marker="s", label="independent goal")
ax.set_xscale("log", base=2)
ax.set_xlabel("candidate count")
ax.set_ylabel("mean selected score")
ax.set_title("Stronger selection pressure can expose proxy error")
ax.legend()
plt.show()

safety = experiment["safety"]["rates"]
passport_errors = validate_measurement_passport(experiment["passport"])
print("ASR:", safety["attack_success_rate"])
print("benign over-refusal:", safety["benign_refusal_rate"])
print("release decision:", experiment["decision"])
print("passport validation errors:", passport_errors)
assert passport_errors == ()
assert all(len(value) == 64 for value in experiment["passport"]["artifact_hashes"].values())

RUN_REAL_MODEL = False

if RUN_REAL_MODEL:
    # Install the backend in a fresh environment before this branch:
    #   pip install "lm_eval[hf]==0.4.12" transformers accelerate
    # Recheck the current release rather than copying this pin indefinitely.
    import lm_eval

    real_result = lm_eval.simple_evaluate(
        model="hf",
        model_args="pretrained=Qwen/Qwen3-0.6B,dtype=float32",
        tasks=["hellaswag"],
        num_fewshot=0,
        batch_size=1,
        device="cpu",
        limit=16,
    )
    output_path = OUTPUT_DIR / "lm_eval_smoke.json"
    output_path.write_text(json.dumps(real_result, indent=2, default=str))
    print("saved:", output_path)
else:
    print("Real-model branch skipped. The CPU evidence route above is the release-tested path.")