File size: 4,929 Bytes
e2c9d42 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | """Versioned evaluation protocols and canonical comparison support.
The reportable token-level comparison is deliberately narrower than the raw
prediction artifact: it uses answer-token indices ``t >= 1`` with binary G/H
labels and valid finite scores. Token zero, uncertain labels, and invalid score
slots remain in artifacts so exclusions are auditable.
"""
from __future__ import annotations
from dataclasses import asdict, dataclass
from enum import Enum
import hashlib
import json
from math import isfinite
from typing import Iterable, Sequence
PREDICTION_SCHEMA_VERSION = 2
class ProtocolSuite(str, Enum):
"""Top-level evaluation suites; each method also records a protocol id."""
CAUSAL_MATCHED = "causal_matched"
METHOD_FAITHFUL = "method_faithful"
OFFLINE = "offline"
LEGACY_FAIR_V0 = "legacy_fair_v0"
@dataclass(frozen=True)
class EvaluationProtocol:
"""Machine-readable method protocol bound into predictions and reports."""
suite: str
protocol_id: str
unit: str
information_horizon: str
label_semantics: str
token_zero_policy: str
uncertain_policy: str
prediction_schema_version: int = PREDICTION_SCHEMA_VERSION
def to_json_dict(self) -> dict:
return asdict(self)
@dataclass(frozen=True, order=True)
class SupportEntry:
"""One canonical token in the common comparison population."""
row_index: int
row_id: str | None
token_index: int
token_id: int
label: int
def to_json_dict(self) -> dict:
return {
"row_index": int(self.row_index),
"row_id": self.row_id,
"token_index": int(self.token_index),
"token_id": int(self.token_id),
"label": int(self.label),
}
def build_comparison_mask(
labels: Sequence[int] | None,
*,
token_indices: Sequence[int] | None = None,
score_valid_mask: Sequence[bool] | None = None,
exclude_token_zero: bool = True,
) -> list[bool]:
"""Build the reportable G/H support mask without dropping artifact slots."""
if labels is None:
return []
n = len(labels)
indices = list(range(n)) if token_indices is None else [int(v) for v in token_indices]
valid = [True] * n if score_valid_mask is None else [bool(v) for v in score_valid_mask]
if len(indices) != n or len(valid) != n:
raise ValueError("labels, token_indices, and score_valid_mask must align")
mask = []
for label, token_index, score_valid in zip(labels, indices, valid):
label = int(label)
if label not in {0, 1, 2}:
raise ValueError("labels must contain only 0/1/2 values")
mask.append(
bool(score_valid)
and label in {0, 1}
and (not exclude_token_zero or token_index >= 1)
)
return mask
def support_entries(
*,
row_index: int,
row_id: str | None,
token_ids: Sequence[int],
labels: Sequence[int],
token_indices: Sequence[int] | None = None,
comparison_mask: Sequence[bool] | None = None,
) -> list[SupportEntry]:
"""Construct canonical entries for one prediction row."""
n = len(labels)
indices = list(range(n)) if token_indices is None else [int(v) for v in token_indices]
mask = (
build_comparison_mask(labels, token_indices=indices)
if comparison_mask is None
else [bool(v) for v in comparison_mask]
)
if len(token_ids) != n or len(indices) != n or len(mask) != n:
raise ValueError("token ids, labels, indices, and comparison mask must align")
return [
SupportEntry(row_index, row_id, token_index, int(token_id), int(label))
for token_id, label, token_index, keep in zip(token_ids, labels, indices, mask)
if keep
]
def support_hash(entries: Iterable[SupportEntry]) -> str:
"""SHA256 of deterministically serialized, canonically sorted support."""
payload = [entry.to_json_dict() for entry in sorted(entries)]
encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
def support_summary(entries: Iterable[SupportEntry]) -> dict:
values = list(entries)
labels = [entry.label for entry in values]
return {
"n_tokens": len(values),
"n_grounded": labels.count(0),
"n_hallucinated": labels.count(1),
"positive_prevalence": labels.count(1) / len(labels) if labels else None,
"support_sha256": support_hash(values),
}
def validate_finite_scores(scores: Sequence[float], mask: Sequence[bool]) -> None:
if len(scores) != len(mask):
raise ValueError("scores and mask must align")
bad = [index for index, (score, keep) in enumerate(zip(scores, mask)) if keep and not isfinite(float(score))]
if bad:
raise ValueError(f"comparison support contains non-finite scores at indices {bad[:10]}")
|