File size: 3,230 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 | from experiments.analyze_uncertainty_temporal_escalation import (
calibration_bucket,
choice_margin,
fit_acquire_commit_thresholds,
fit_threshold,
select_acquire_commit_prediction,
select_prediction,
)
def row(uid: str, prediction: str, margin: float) -> dict:
return {
"uid": uid,
"prediction": prediction,
"choice_confidence": {
"top2_margin": margin,
"task_or_group_used": False,
},
}
def test_policy_escalates_only_from_global_margin() -> None:
recent = row("a", "A", 0.1)
temporal = row("a", "B", 0.9)
assert select_prediction(recent, temporal, 0.2) == ("B", True)
assert select_prediction(recent, temporal, 0.05) == ("A", False)
def test_fit_threshold_prefers_accuracy_then_fewer_escalations() -> None:
recent = {
"a": row("a", "A", 0.1),
"b": row("b", "B", 0.8),
}
temporal = {
"a": row("a", "B", 0.9),
"b": row("b", "A", 0.9),
}
recent_scores = {
"a": {"correct": False},
"b": {"correct": True},
}
temporal_scores = {
"a": {"correct": True},
"b": {"correct": False},
}
threshold, sweep = fit_threshold(
["a", "b"],
recent=recent,
temporal=temporal,
recent_scores=recent_scores,
temporal_scores=temporal_scores,
)
assert threshold == 0.1
assert max(item["correct"] for item in sweep) == 2
def test_calibration_split_is_deterministic() -> None:
assert calibration_bucket("stable-uid", 7) == calibration_bucket(
"stable-uid", 7
)
def test_choice_margin_rejects_non_blind_provenance() -> None:
bad = row("a", "A", 0.1)
bad["choice_confidence"]["task_or_group_used"] = True
try:
choice_margin(bad)
except ValueError as error:
assert "not task blind" in str(error)
else:
raise AssertionError("expected task-blind provenance failure")
def test_full_vocabulary_generation_margin_takes_precedence() -> None:
value = row("a", "A", 0.9)
value["generation_confidence"] = {
"top2_margin": 0.2,
"task_or_group_used": False,
}
assert choice_margin(value) == 0.2
def test_safe_commit_keeps_current_answer_when_proposal_is_less_confident() -> None:
recent = row("a", "A", 0.2)
temporal = row("a", "B", 0.1)
assert select_acquire_commit_prediction(recent, temporal, 0.3, 0.0) == (
"A",
True,
False,
-0.1,
)
def test_joint_fit_can_acquire_but_reject_harmful_changed_answer() -> None:
recent = {
"a": row("a", "A", 0.1),
"b": row("b", "B", 0.2),
}
temporal = {
"a": row("a", "C", 0.8),
"b": row("b", "D", 0.1),
}
recent_scores = {"a": {"correct": False}, "b": {"correct": True}}
temporal_scores = {"a": {"correct": True}, "b": {"correct": False}}
acquire, commit, sweep = fit_acquire_commit_thresholds(
["a", "b"],
recent=recent,
temporal=temporal,
recent_scores=recent_scores,
temporal_scores=temporal_scores,
)
assert acquire >= 0.1
assert -0.1 <= commit < 0.7
assert max(item["correct"] for item in sweep) == 2
|