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
|
#!/usr/bin/env python3
"""Audit the frozen contrastive state-bias B1 gate."""
import argparse
import json
import math
from pathlib import Path
from contrastive_bias_b1 import RESULT_ROOT, bias_cells
def read_json(path):
with open(path, encoding="utf-8") as handle:
return json.load(handle)
def cell_path(cell_id):
return RESULT_ROOT / ("dp-bias-b1-" + cell_id + ".json")
def valid_record(record, cell):
history = record.get("history") or {}
return (
record.get("status") == "completed"
and record.get("cell_id") == cell["cell_id"]
and record.get("kind") == cell["kind"]
and record.get("rule") == cell["rule"]
and record.get("ratio") == cell["ratio"]
and history.get("finite") is True
and history.get("epochs_completed") == 20
and math.isnan(float(history.get("test_accuracy", float("nan"))))
)
def metric_max(record, key):
values = record["history"]["curves"][key]
return max(abs(float(value)) for value in values)
def main():
parser = argparse.ArgumentParser()
parser.add_argument(
"--out", type=Path,
default=RESULT_ROOT.parent / "b1_gate.json")
args = parser.parse_args()
cells = bias_cells()
missing = [cell["cell_id"] for cell in cells if not cell_path(cell["cell_id"]).is_file()]
if missing:
raise RuntimeError("missing B1 cells: " + ", ".join(missing))
records = {}
for cell in cells:
record = read_json(cell_path(cell["cell_id"]))
if not valid_record(record, cell):
raise RuntimeError(f"invalid or incomplete B1 cell: {cell['cell_id']}")
records[cell["cell_id"]] = record
clean = records["clean"]
clean_acc = float(clean["history"]["final_validation_accuracy"])
common = records["common-activity-r4-raw"]
common_acc = float(common["history"]["final_validation_accuracy"])
common_error = metric_max(
common, "maximum_used_clean_difference_relative_error")
predictor_instruction_max = max(
metric_max(record, "instruction_observations_for_predictor")
for record in records.values()
)
innovation_post_max = max(
metric_max(record, "post_bias_raw_bias_rms_ratio")
for cell_id, record in records.items() if "-innovation" in cell_id
)
candidates = []
table = []
for ratio in (0.25, 1.0, 4.0):
tag = rate_tag = f"{ratio:g}".replace(".", "p")
row = {"ratio": ratio}
for rule in ("raw", "innovation", "oracle"):
record = records[f"activity-r{tag}-{rule}"]
row[rule] = float(record["history"]["final_validation_accuracy"])
row["raw_degradation"] = clean_acc - row["raw"]
row["innovation_clean_gap"] = abs(row["innovation"] - clean_acc)
row["innovation_oracle_gap"] = abs(row["innovation"] - row["oracle"])
row["innovation_post_bias_ratio_max"] = metric_max(
records[f"activity-r{tag}-innovation"],
"post_bias_raw_bias_rms_ratio")
row["passes"] = (
row["raw_degradation"] >= 5.0
and row["innovation_clean_gap"] <= 2.0
and row["innovation_oracle_gap"] <= 1.0
and row["innovation_post_bias_ratio_max"] <= 1e-3
)
if row["passes"]:
candidates.append(ratio)
table.append(row)
checks = {
"clean_at_least_70": clean_acc >= 70.0,
"common_within_0p2": abs(common_acc - clean_acc) <= 0.2,
"common_difference_error_at_most_1e_6": common_error <= 1e-6,
"some_activity_ratio_passes": bool(candidates),
"all_innovation_post_bias_at_most_1e_3": innovation_post_max <= 1e-3,
"predictor_saw_zero_instruction_observations": predictor_instruction_max == 0.0,
}
source_values = {json.dumps(row["source"], sort_keys=True) for row in records.values()}
registry_values = {row["registry_sha256"] for row in records.values()}
checks["single_source_lock"] = len(source_values) == 1
checks["single_registry_lock"] = len(registry_values) == 1
report = {
"stage": "contrastive_bias_b1", "gate": (
"pass" if all(checks.values()) else "fail"),
"checks": checks, "clean_final_validation_accuracy": clean_acc,
"common_final_validation_accuracy": common_acc,
"common_maximum_difference_relative_error": common_error,
"innovation_maximum_post_bias_ratio": innovation_post_max,
"predictor_maximum_instruction_observations": predictor_instruction_max,
"activity_table": table,
"selected_confirmation_ratio": max(candidates) if candidates else None,
"num_expected_records": 17, "num_audited_records": len(records),
"source": clean["source"], "registry_sha256": clean["registry_sha256"],
}
args.out.parent.mkdir(parents=True, exist_ok=True)
with open(args.out, "w", encoding="utf-8") as handle:
json.dump(report, handle, indent=2, sort_keys=True)
handle.write("\n")
print(json.dumps(report, indent=2, sort_keys=True))
if __name__ == "__main__":
main()
|
