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
|
#!/usr/bin/env python3
"""Apply the frozen PickupLoc clean-feasibility gate."""
import argparse
import json
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
DEFAULT_RESULTS = ROOT / "results" / "babyai_shared" / "pickup_p0"
DEFAULT_OUT = ROOT / "results" / "babyai_shared" / "pickup_p0_analysis.json"
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--results", type=Path, default=DEFAULT_RESULTS)
parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
args = parser.parse_args()
candidates = []
for depth in (2, 4):
for learning_rate in (0.01, 0.03):
records = {}
sources = []
for condition in ("bp", "clean_kp"):
path = args.results / (
f"d{depth}_lr{learning_rate}_{condition}.json")
with open(path, encoding="utf-8") as handle:
records[condition] = json.load(handle)
sources.append(str(path.relative_to(ROOT)))
bp, kp = records["bp"], records["clean_kp"]
bp_success = 100.0 * bp["rollout"]["success"]
kp_success = 100.0 * kp["rollout"]["success"]
lesion_success = 100.0 * bp["mission_lesion_rollout"]["success"]
checks = {
"both_finite": bool(bp["finite"] and kp["finite"]),
"bp_success_at_least_70": bp_success >= 70.0,
"clean_kp_success_at_least_60": kp_success >= 60.0,
"bp_mission_lesion_drop_at_least_20": (
bp_success - lesion_success >= 20.0),
}
candidates.append({
"hidden_layers": depth,
"learning_rate": learning_rate,
"bp_rollout_success_percent": bp_success,
"clean_kp_rollout_success_percent": kp_success,
"bp_mission_lesion_success_percent": lesion_success,
"bp_mission_lesion_drop_points": bp_success - lesion_success,
"bp_action_accuracy_percent": (
100.0 * bp["validation"]["accuracy"]),
"clean_kp_action_accuracy_percent": (
100.0 * kp["validation"]["accuracy"]),
"eligible": all(checks.values()),
"checks": checks,
"source_files": sources,
})
eligible = [row for row in candidates if row["eligible"]]
selected = max(eligible, key=lambda row: (
row["clean_kp_rollout_success_percent"],
row["clean_kp_action_accuracy_percent"],
-row["hidden_layers"], -row["learning_rate"])) if eligible else None
best_clean = max(candidates, key=lambda row: (
row["clean_kp_rollout_success_percent"],
row["clean_kp_action_accuracy_percent"]))
report = {
"stage": "babyai_pickup_p0_clean_feasibility",
"gate": "pass" if selected is not None else "fail",
"candidates": candidates,
"selected": selected,
"best_clean_candidate_for_diagnosis": best_clean,
"raw_or_sdil_results_run_or_read": False,
"test_split_generated_or_read": False,
"decision": (
"Add task memory and repeat a clean-only gate; do not run raw or "
"SDIL on the present feedforward policy."
if selected is None else "Open the frozen shared-feedback endpoint."),
}
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()
|
