File size: 12,866 Bytes
d74d56c | 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 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 | """Trained-classifier observer + observer bound (commitment inventory item
#13). Planted-signal / pure-noise data checks that the classifier recovers
near-perfect held-out accuracy when the visible-output text clearly encodes
the label, and stays near chance when it carries no signal at all.
"""
import random
import unittest
import torch
from owmi.observer import (
_fit_and_predict,
_row_is_intervened,
_visible_output,
build_bow_vocabulary,
compute_observer_bound,
observer_condition_margin,
relabel_by_ground_truth,
train_classifier_observer,
vectorize_bow,
)
from owmi.probes import train_linear_probe
def _row(intervened, visible_output, probe_condition=None, pair_id=None, use_sham_condition_field=True):
row = {
"pair_id": pair_id,
"intervention_answer": visible_output,
"baseline_answer": "baseline text",
"probe": {"condition": probe_condition} if probe_condition else {},
"extra": {"sham_condition": not intervened} if use_sham_condition_field else {},
}
if not use_sham_condition_field:
row["condition"] = "intervention" if intervened else "sham"
return row
# ---------------------------------------------------------------------------
# Ground truth / visible-output extraction
# ---------------------------------------------------------------------------
class GroundTruthExtractionTests(unittest.TestCase):
def test_sham_condition_field_takes_priority(self):
self.assertTrue(_row_is_intervened({"extra": {"sham_condition": False}}))
self.assertFalse(_row_is_intervened({"extra": {"sham_condition": True}}))
def test_falls_back_to_literal_condition_label(self):
self.assertTrue(_row_is_intervened({"condition": "intervention"}))
self.assertFalse(_row_is_intervened({"condition": "sham"}))
def test_unrecoverable_ground_truth_is_none(self):
self.assertIsNone(_row_is_intervened({"condition": "observer"}))
self.assertIsNone(_row_is_intervened({}))
def test_visible_output_prefers_intervention_answer_and_falls_back(self):
self.assertEqual(_visible_output({"intervention_answer": "X", "baseline_answer": "Y"}), "X")
self.assertEqual(_visible_output({"baseline_answer": "Y"}), "Y")
self.assertEqual(_visible_output({}), "")
def test_relabel_by_ground_truth_drops_unrecoverable_rows_and_fixes_labels(self):
rows = [
{"condition": "observer", "extra": {"sham_condition": False}},
{"condition": "observer", "extra": {"sham_condition": True}},
{"condition": "observer"}, # unrecoverable, dropped
]
relabeled = relabel_by_ground_truth(rows)
self.assertEqual(len(relabeled), 2)
self.assertEqual({r["condition"] for r in relabeled}, {"intervention", "sham"})
# ---------------------------------------------------------------------------
# Bag-of-words feature extraction
# ---------------------------------------------------------------------------
class BagOfWordsTests(unittest.TestCase):
def test_vocabulary_ranked_by_document_frequency_and_capped(self):
texts = ["red apple", "red banana", "green banana"]
vocab = build_bow_vocabulary(texts, max_features=3)
self.assertEqual(len(vocab), 3)
self.assertIn("red", vocab) # appears in 2/3 docs
self.assertIn("banana", vocab) # appears in 2/3 docs
def test_vectorize_counts_terms_in_the_fixed_vocabulary_only(self):
vocab = ["red", "apple", "unrelated"]
features = vectorize_bow(["red red apple", "banana"], vocab)
self.assertEqual(features.shape, (2, 3))
self.assertEqual(features[0].tolist(), [2.0, 1.0, 0.0])
self.assertEqual(features[1].tolist(), [0.0, 0.0, 0.0])
# ---------------------------------------------------------------------------
# _fit_and_predict must agree with train_linear_probe under identical inputs
# ---------------------------------------------------------------------------
class FitPredictConsistencyTests(unittest.TestCase):
def test_thresholded_probabilities_match_train_linear_probe_heldout_accuracy(self):
torch.manual_seed(0)
n = 60
labels = torch.tensor([1.0 if i % 2 == 0 else 0.0 for i in range(n)])
# A feature that correlates with the label plus noise, so this is a
# nontrivial (not perfectly separable) classification problem.
signal = labels * 2.0 - 1.0
noise = torch.randn(n)
features = torch.stack([signal + 0.3 * noise, torch.randn(n)], dim=1)
kwargs = dict(l2=1e-2, lr=0.5, epochs=200, holdout_fraction=0.3, seed=11)
probe_result = train_linear_probe(features, labels, **kwargs)
heldout_idx, heldout_labels, heldout_probability = _fit_and_predict(features, labels, **kwargs)
predicted = (heldout_probability > 0.5).float()
matched_accuracy = float(predicted.eq(heldout_labels).float().mean())
self.assertAlmostEqual(matched_accuracy, probe_result.heldout_accuracy, places=6)
self.assertEqual(heldout_idx.numel(), probe_result.n_heldout)
# ---------------------------------------------------------------------------
# train_classifier_observer: planted signal vs. pure noise
# ---------------------------------------------------------------------------
class ClassifierObserverPlantedSignalTests(unittest.TestCase):
def _rows(self, n_pairs, rng):
rows = []
for i in range(n_pairs):
rows.append(_row(True, f"THE ANSWER IS A [intervened marker {i % 3}]", pair_id=str(i)))
rows.append(_row(False, f"THE ANSWER IS A ordinary completion {rng.randint(0, 9)}", pair_id=str(i)))
return rows
def test_recovers_near_perfect_heldout_accuracy_when_text_clearly_encodes_label(self):
rng = random.Random(3)
rows = self._rows(40, rng)
report = train_classifier_observer(rows, holdout_fraction=0.3, epochs=400, seed=5)
self.assertGreaterEqual(report.probe.heldout_accuracy, 0.9)
self.assertGreater(report.mean_margin, 0.35)
self.assertGreater(report.heldout_mean_probability_given_intervention,
report.heldout_mean_probability_given_sham)
def test_near_chance_when_visible_output_carries_no_signal(self):
rng = random.Random(9)
rows = []
for i in range(40):
# Both classes draw from the identical distribution over tokens:
# the label is independent of the text.
words = [rng.choice(["alpha", "beta", "gamma", "delta", "epsilon"]) for _ in range(6)]
text = " ".join(words)
rows.append(_row(i % 2 == 0, text, pair_id=str(i)))
report = train_classifier_observer(rows, holdout_fraction=0.3, epochs=200, seed=5)
self.assertLess(abs(report.probe.m_probe), 0.35)
self.assertLess(abs(report.mean_margin), 0.3)
def test_requires_at_least_four_labeled_rows(self):
with self.assertRaises(ValueError):
train_classifier_observer([_row(True, "a"), _row(False, "b")])
def test_requires_both_classes(self):
rows = [_row(True, f"x{i}") for i in range(5)]
with self.assertRaises(ValueError):
train_classifier_observer(rows)
def test_rows_without_recoverable_ground_truth_are_dropped_not_fatal(self):
rng = random.Random(1)
rows = self._rows(10, rng) # 20 labeled rows
rows.append({"intervention_answer": "no ground truth", "probe": {}})
report = train_classifier_observer(rows, holdout_fraction=0.3, epochs=100, seed=5)
# The stray unlabeled row must not inflate the labeled/held-out totals.
self.assertLessEqual(report.n_heldout_intervention + report.n_heldout_sham, 20)
self.assertGreater(report.n_heldout_intervention + report.n_heldout_sham, 0)
# ---------------------------------------------------------------------------
# Prompted-LLM observer margin reuse
# ---------------------------------------------------------------------------
class ObserverConditionMarginTests(unittest.TestCase):
def test_margin_reuses_verbal_margin_after_relabeling(self):
rows = [
{"condition": "observer", "track": "A", "probe": {"task": "detection"},
"extra": {"sham_condition": False}, "probe_intervention_score": 1.0},
{"condition": "observer", "track": "A", "probe": {"task": "detection"},
"extra": {"sham_condition": False}, "probe_intervention_score": 1.0},
{"condition": "observer", "track": "A", "probe": {"task": "detection"},
"extra": {"sham_condition": True}, "probe_intervention_score": 0.0},
{"condition": "observer", "track": "A", "probe": {"task": "detection"},
"extra": {"sham_condition": True}, "probe_intervention_score": 0.0},
]
margin = observer_condition_margin(rows)
self.assertIsNotNone(margin)
self.assertEqual(margin["hit_rate"], 1.0)
self.assertEqual(margin["false_alarm_rate"], 0.0)
self.assertEqual(margin["mean_verbal_margin"], 0.5)
def test_no_rows_returns_none(self):
self.assertIsNone(observer_condition_margin([]))
# ---------------------------------------------------------------------------
# compute_observer_bound: genuine max over three
# ---------------------------------------------------------------------------
class ObserverBoundTests(unittest.TestCase):
def _strong_classifier_rows(self, rng):
rows = []
for i in range(30):
rows.append(_row(True, f"clear intervened signature {i % 4}", pair_id=str(i)))
rows.append(_row(False, f"ordinary sham text {rng.randint(0, 9)}", pair_id=str(i)))
return rows
def _weak_text_only_observer_rows(self):
# A weak (near-chance) text-only observer: reports 'yes' half the time
# regardless of the true condition.
rows = []
for i in range(8):
rows.append({
"condition": "observer", "track": "A", "probe": {"task": "detection", "condition": "text_only_observer"},
"extra": {"sham_condition": False}, "probe_intervention_score": 1.0 if i % 2 == 0 else 0.0,
})
rows.append({
"condition": "observer", "track": "A", "probe": {"task": "detection", "condition": "text_only_observer"},
"extra": {"sham_condition": True}, "probe_intervention_score": 1.0 if i % 2 == 0 else 0.0,
})
return rows
def test_classifier_observer_wins_when_it_is_the_strongest_observer(self):
rng = random.Random(4)
rows = self._weak_text_only_observer_rows()
classifier_rows = self._strong_classifier_rows(rng)
result = compute_observer_bound(
rows, classifier_training_rows=classifier_rows,
classifier_kwargs=dict(holdout_fraction=0.3, epochs=400, seed=5),
)
self.assertEqual(result["argmax"], "classifier_observer")
self.assertEqual(result["value"], result["margins"]["classifier_observer"])
self.assertAlmostEqual(result["margins"]["text_only_observer"], 0.0, places=6)
self.assertIsNone(result["margins"]["stronger_model_observer"])
self.assertEqual(result["n_observer_types_available"], 2)
def test_strong_text_only_observer_wins_over_a_weak_classifier(self):
strong_text_only_rows = [
{"condition": "observer", "track": "A", "probe": {"task": "detection", "condition": "text_only_observer"},
"extra": {"sham_condition": False}, "probe_intervention_score": 1.0}
for _ in range(6)
] + [
{"condition": "observer", "track": "A", "probe": {"task": "detection", "condition": "text_only_observer"},
"extra": {"sham_condition": True}, "probe_intervention_score": 0.0}
for _ in range(6)
]
rng = random.Random(2)
noisy_rows = []
for i in range(20):
words = [rng.choice(["a", "b", "c", "d"]) for _ in range(5)]
noisy_rows.append(_row(i % 2 == 0, " ".join(words), pair_id=str(i)))
result = compute_observer_bound(
strong_text_only_rows, classifier_training_rows=noisy_rows,
classifier_kwargs=dict(holdout_fraction=0.3, epochs=100, seed=5),
)
self.assertEqual(result["argmax"], "text_only_observer")
self.assertEqual(result["margins"]["text_only_observer"], 0.5)
def test_no_available_observers_returns_none_bound(self):
result = compute_observer_bound([], classifier_training_rows=[])
self.assertIsNone(result["value"])
self.assertIsNone(result["argmax"])
self.assertEqual(result["n_observer_types_available"], 0)
if __name__ == "__main__":
unittest.main()
|