File size: 7,642 Bytes
78ac017 | 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 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 | #!/usr/bin/env python3
"""Select temporal-view calls using a threshold or fixed uncertainty budget."""
from __future__ import annotations
import argparse
import json
import math
from pathlib import Path
from typing import Any
from experiments.analyze_uncertainty_temporal_escalation import choice_margin
from experiments.evaluate_qwen3vl_recent4_full import atomic_write, sha256_file
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--recent", type=Path, required=True)
policy = parser.add_mutually_exclusive_group(required=True)
policy.add_argument("--threshold", type=Path)
policy.add_argument(
"--quantile",
type=float,
help="Acquire the lowest-margin fraction without reading cohort labels.",
)
parser.add_argument("--samples", type=Path, required=True)
parser.add_argument("--output-uids", type=Path, required=True)
parser.add_argument("--output-samples", type=Path, required=True)
parser.add_argument("--output-audit", type=Path, required=True)
return parser.parse_args()
def load_frozen_threshold(path: Path, model_manifest_sha256: str | None) -> float:
value = json.loads(path.read_text(encoding="utf-8"))
if (
value.get("status") != "source_disjoint_internal_threshold_calibration"
or value.get("eligible_as_frozen_benchmark_controller_config") is not True
or value.get("benchmark") != "internal_policy_holdout"
or value.get("model_manifest_sha256") != model_manifest_sha256
):
raise ValueError("threshold is not an eligible source-disjoint calibration")
threshold = float(value["policy"]["threshold"])
if not 0.0 <= threshold <= 1.0:
raise ValueError("invalid calibrated threshold")
return threshold
def select_rows(
recent: dict[str, Any], *, threshold: float
) -> tuple[list[str], list[dict[str, Any]]]:
selected = []
audit_rows = []
for row in recent.get("rows") or []:
uid = str(row.get("uid") or "")
if not uid:
raise ValueError("recent result has a row without uid")
margin = choice_margin(row)
escalated = margin <= threshold
if escalated:
selected.append(uid)
audit_rows.append(
{
"uid": uid,
"margin": margin,
"threshold": threshold,
"escalated": escalated,
}
)
if len(audit_rows) != int(recent.get("num_samples", -1)):
raise ValueError("recent result sample count diverged")
if int(recent.get("num_errors", -1)) != 0:
raise ValueError("recent result contains inference errors")
if len({row["uid"] for row in audit_rows}) != len(audit_rows):
raise ValueError("recent result contains duplicate uids")
return selected, audit_rows
def select_rows_by_quantile(
recent: dict[str, Any], *, quantile: float
) -> tuple[list[str], list[dict[str, Any]], float]:
"""Select an exact bottom-margin budget with deterministic UID tie-breaking."""
if not 0.0 < quantile <= 1.0:
raise ValueError("quantile must be in (0, 1]")
rows = list(recent.get("rows") or [])
if len(rows) != int(recent.get("num_samples", -1)):
raise ValueError("recent result sample count diverged")
if int(recent.get("num_errors", -1)) != 0:
raise ValueError("recent result contains inference errors")
values = []
for row in rows:
uid = str(row.get("uid") or "")
if not uid:
raise ValueError("recent result has a row without uid")
values.append((choice_margin(row), uid))
if len({uid for _, uid in values}) != len(values):
raise ValueError("recent result contains duplicate uids")
budget = max(1, math.ceil(len(values) * quantile))
ranked = sorted(values, key=lambda item: (item[0], item[1]))
selected_set = {uid for _, uid in ranked[:budget]}
cutoff = float(ranked[budget - 1][0])
selected = [str(row["uid"]) for row in rows if str(row["uid"]) in selected_set]
rank_by_uid = {uid: rank for rank, (_, uid) in enumerate(ranked, 1)}
audit_rows = [
{
"uid": str(row["uid"]),
"margin": choice_margin(row),
"uncertainty_rank": rank_by_uid[str(row["uid"])],
"quantile": quantile,
"escalated": str(row["uid"]) in selected_set,
}
for row in rows
]
return selected, audit_rows, cutoff
def main() -> None:
args = parse_args()
recent = json.loads(args.recent.read_text(encoding="utf-8"))
threshold = None
quantile = None
cutoff = None
if args.threshold is not None:
threshold = load_frozen_threshold(
args.threshold, recent.get("model_manifest_sha256")
)
selected, audit_rows = select_rows(recent, threshold=threshold)
else:
quantile = float(args.quantile)
selected, audit_rows, cutoff = select_rows_by_quantile(
recent, quantile=quantile
)
selected_set = set(selected)
sample_rows = [
json.loads(line)
for line in args.samples.read_text(encoding="utf-8").splitlines()
if line.strip()
]
ordered_uids = [str(row.get("uid") or "") for row in sample_rows]
recent_uids = [str(row["uid"]) for row in recent["rows"]]
if ordered_uids != recent_uids:
raise ValueError("recent result does not follow the frozen sample order")
selected_samples = [row for row in sample_rows if str(row["uid"]) in selected_set]
if [str(row["uid"]) for row in selected_samples] != selected:
raise ValueError("selected UID ordering diverged")
args.output_uids.parent.mkdir(parents=True, exist_ok=True)
args.output_uids.write_text("\n".join(selected) + "\n", encoding="utf-8")
args.output_samples.write_text(
"".join(
json.dumps(row, sort_keys=True, ensure_ascii=False) + "\n"
for row in selected_samples
),
encoding="utf-8",
)
atomic_write(
args.output_audit,
{
"format": (
"streameva.internal_threshold_uid_selection.v0"
if threshold is not None
else "streameva.unlabeled_budget_uid_selection.v0"
),
"status": "frozen_before_benchmark_temporal_inference",
"benchmark": recent.get("benchmark"),
"model_manifest_sha256": recent.get("model_manifest_sha256"),
"threshold": threshold,
"quantile": quantile,
"empirical_margin_cutoff": cutoff,
"threshold_audit": (
str(args.threshold.resolve()) if args.threshold is not None else None
),
"threshold_audit_sha256": (
sha256_file(args.threshold) if args.threshold is not None else None
),
"recent_result": str(args.recent.resolve()),
"recent_result_sha256": sha256_file(args.recent),
"samples_sha256": sha256_file(args.samples),
"question_task_group_or_benchmark_router": False,
"score_or_gold_fields_read_for_selection": False,
"benchmark_identity_used_for_policy_parameters": False,
"cohort_labels_read_for_selection": False,
"total": len(audit_rows),
"selected": len(selected),
"selection_rate": len(selected) / len(audit_rows),
"rows": audit_rows,
},
)
print(json.dumps({"total": len(audit_rows), "selected": len(selected)}))
if __name__ == "__main__":
main()
|