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

import math


def wilson_interval(successes: int, total: int, z: float = 1.96) -> tuple[float, float, float]:
    if total <= 0:
        raise ValueError("total must be positive")
    if successes < 0 or successes > total:
        raise ValueError("successes must lie in [0, total]")
    if not math.isfinite(z) or z <= 0:
        raise ValueError("z must be finite and positive")

    estimate = successes / total
    denominator = 1.0 + z * z / total
    center = (estimate + z * z / (2.0 * total)) / denominator
    half_width = z / denominator * math.sqrt(
        estimate * (1.0 - estimate) / total + z * z / (4.0 * total * total)
    )
    return estimate, max(0.0, center - half_width), min(1.0, center + half_width)


def worst_case_sample_size(margin: float, z: float = 1.96) -> int:
    if not math.isfinite(margin) or not 0.0 < margin < 1.0:
        raise ValueError("margin must lie in (0, 1)")
    if not math.isfinite(z) or z <= 0:
        raise ValueError("z must be finite and positive")
    return math.ceil(z * z / (4.0 * margin * margin))

from __future__ import annotations

import math
from dataclasses import dataclass

import numpy as np


@dataclass(frozen=True)
class PairedTable:
    both_correct: int
    a_only: int
    b_only: int
    both_wrong: int


def paired_table(a_correct, b_correct) -> PairedTable:
    a = np.asarray(a_correct, dtype=bool)
    b = np.asarray(b_correct, dtype=bool)
    if a.shape != b.shape or a.ndim != 1 or a.size == 0:
        raise ValueError("paired vectors must be non-empty and have equal shape")
    return PairedTable(
        int(np.sum(a & b)),
        int(np.sum(a & ~b)),
        int(np.sum(~a & b)),
        int(np.sum(~a & ~b)),
    )


def exact_mcnemar_pvalue(a_only: int, b_only: int) -> float:
    if min(a_only, b_only) < 0:
        raise ValueError("counts must be non-negative")
    n = a_only + b_only
    if n == 0:
        return 1.0
    tail = sum(math.comb(n, j) for j in range(min(a_only, b_only) + 1)) / 2**n
    return min(1.0, 2.0 * tail)


def paired_bootstrap_difference(a_scores, b_scores, *, replicates=10_000, seed=0):
    a = np.asarray(a_scores, dtype=float)
    b = np.asarray(b_scores, dtype=float)
    if a.shape != b.shape or a.ndim != 1 or a.size == 0:
        raise ValueError("paired vectors must be non-empty and have equal shape")
    if replicates <= 0:
        raise ValueError("replicates must be positive")
    delta = a - b
    rng = np.random.default_rng(seed)
    indices = rng.integers(0, delta.size, size=(replicates, delta.size))
    draws = delta[indices].mean(axis=1)
    return float(delta.mean()), float(np.quantile(draws, 0.025)), float(np.quantile(draws, 0.975))


table = paired_table([1, 1, 0, 0, 1], [1, 0, 1, 0, 0])
assert table == PairedTable(1, 2, 1, 1)
assert abs(exact_mcnemar_pvalue(9, 2) - 0.0654296875) < 1e-15

def pass_at_k(n: int, c: int, k: int) -> float:
    if n <= 0 or not 0 <= c <= n or not 1 <= k <= n:
        raise ValueError("require n>0, 0<=c<=n, and 1<=k<=n")
    if n - c < k:
        return 1.0
    probability_of_miss = 1.0
    for i in range(k):
        probability_of_miss *= (n - c - i) / (n - i)
    return 1.0 - probability_of_miss


def coverage_and_selection(correct, selected_index: int) -> tuple[bool, bool]:
    values = [bool(x) for x in correct]
    if not values or not 0 <= selected_index < len(values):
        raise ValueError("invalid candidate pool or selected index")
    return any(values), values[selected_index]

from __future__ import annotations


def ngram_set(text: str, n: int) -> set[tuple[str, ...]]:
    if n <= 0:
        raise ValueError("n must be positive")
    tokens = text.casefold().split()
    return {tuple(tokens[i : i + n]) for i in range(len(tokens) - n + 1)}


def ngram_containment(benchmark_text: str, corpus_text: str, n: int) -> float:
    benchmark = ngram_set(benchmark_text, n)
    corpus = ngram_set(corpus_text, n)
    if not benchmark:
        return 0.0
    return len(benchmark & corpus) / len(benchmark)


def ngram_jaccard(left: str, right: str, n: int) -> float:
    a = ngram_set(left, n)
    b = ngram_set(right, n)
    union = a | b
    return 0.0 if not union else len(a & b) / len(union)


def contamination_report(items, corpus, *, n=8, threshold=0.2):
    if not 0.0 <= threshold <= 1.0:
        raise ValueError("threshold must lie in [0, 1]")
    maxima = []
    matches = []
    for i, item in enumerate(items):
        scores = [ngram_containment(item, doc, n) for doc in corpus]
        best = max(scores, default=0.0)
        maxima.append(best)
        for j, score in enumerate(scores):
            if score >= threshold:
                matches.append((i, j, score))
    return {
        "metric": "benchmark_ngram_containment",
        "n": n,
        "threshold": threshold,
        "max_overlap": maxima,
        "matches": matches,
    }


assert ngram_containment("a b c", "x a b c y", 2) == 1.0
assert ngram_jaccard("a b c", "x a b c y", 2) == 0.5

from __future__ import annotations


def _labels(values, allowed):
    result = [str(x).upper() for x in values]
    if any(x not in allowed for x in result):
        raise ValueError(f"labels must belong to {sorted(allowed)}")
    return result


def judge_swap_audit(forward_canonical, reverse_canonical):
    a = _labels(forward_canonical, {"A", "B", "TIE"})
    b = _labels(reverse_canonical, {"A", "B", "TIE"})
    if len(a) != len(b) or not a:
        raise ValueError("label vectors must be non-empty and aligned")
    consistent = sum(x == y for x, y in zip(a, b))
    votes = [x for x in a + b if x != "TIE"]
    return {
        "items": len(a),
        "consistent": consistent,
        "inconsistent": len(a) - consistent,
        "consistency_rate": consistent / len(a),
        "candidate_a_rate": votes.count("A") / len(votes) if votes else 0.0,
        "candidate_b_rate": votes.count("B") / len(votes) if votes else 0.0,
    }


def position_audit(raw_forward, raw_reverse):
    a = _labels(raw_forward, {"FIRST", "SECOND", "TIE"})
    b = _labels(raw_reverse, {"FIRST", "SECOND", "TIE"})
    if len(a) != len(b) or not a:
        raise ValueError("label vectors must be non-empty and aligned")
    all_labels = a + b
    first = all_labels.count("FIRST")
    second = all_labels.count("SECOND")
    ties = all_labels.count("TIE")
    denom = first + second
    return {
        "judgments": len(all_labels),
        "first_position_wins": first,
        "second_position_wins": second,
        "ties": ties,
        "first_position_rate_non_ties": first / denom if denom else 0.0,
    }


def sycophancy_metrics(correct_without_claim, correct_with_false_claim, agrees_with_false_claim):
    base = [bool(x) for x in correct_without_claim]
    challenged = [bool(x) for x in correct_with_false_claim]
    agrees = [bool(x) for x in agrees_with_false_claim]
    if not base or not (len(base) == len(challenged) == len(agrees)):
        raise ValueError("vectors must be non-empty and aligned")
    initially_correct = sum(base)
    switches = sum(b and not c for b, c in zip(base, challenged))
    return {
        "accuracy_without_claim": sum(base) / len(base),
        "accuracy_with_false_claim": sum(challenged) / len(base),
        "accuracy_drop": (sum(base) - sum(challenged)) / len(base),
        "switch_rate_among_initially_correct": switches / initially_correct if initially_correct else 0.0,
        "unjustified_agreement_rate": sum(agrees) / len(agrees),
    }


assert judge_swap_audit(["A", "A", "B", "TIE"], ["A", "B", "B", "TIE"])["consistent"] == 3
assert position_audit(["FIRST", "SECOND"], ["FIRST", "FIRST"])["first_position_wins"] == 3

def safety_rates(attack_succeeded, benign_refused, severities=None):
    attacks = [bool(x) for x in attack_succeeded]
    benign = [bool(x) for x in benign_refused]
    if not attacks or not benign:
        raise ValueError("both harmful and benign samples are required")
    if severities is None:
        severity_weighted = sum(attacks) / len(attacks)
    else:
        weights = [float(x) for x in severities]
        if len(weights) != len(attacks) or any(x <= 0 for x in weights):
            raise ValueError("positive severities must align with attacks")
        severity_weighted = sum(w for hit, w in zip(attacks, weights) if hit) / sum(weights)
    return {
        "attack_success_rate": sum(attacks) / len(attacks),
        "benign_over_refusal_rate": sum(benign) / len(benign),
        "severity_weighted_asr": severity_weighted,
    }


def release_decision(metrics, thresholds, critical_failures=()):
    failures = [f"critical:{item}" for item in critical_failures]
    for name, (direction, threshold) in thresholds.items():
        if name not in metrics:
            failures.append(f"missing:{name}")
            continue
        value = float(metrics[name])
        if direction == "max" and value > threshold:
            failures.append(f"{name}>{threshold}")
        elif direction == "min" and value < threshold:
            failures.append(f"{name}<{threshold}")
        elif direction not in {"min", "max"}:
            failures.append(f"invalid_direction:{name}")
    return {"passed": not failures, "failures": failures}


def validate_measurement_passport(passport):
    required = {
        "run_id", "model_id", "model_revision", "tokenizer_id", "dataset_id",
        "dataset_revision", "split", "solver_id", "scorer_id", "scorer_revision",
        "decoding", "seeds", "sample_count", "metrics", "cost", "artifact_hashes",
    }
    errors = [f"missing:{key}" for key in sorted(required - set(passport))]
    if "sample_count" in passport and int(passport["sample_count"]) <= 0:
        errors.append("sample_count:not_positive")
    if "seeds" in passport and not passport["seeds"]:
        errors.append("seeds:empty")
    if "artifact_hashes" in passport and not passport["artifact_hashes"]:
        errors.append("artifact_hashes:empty")
    return tuple(errors)