mirror of
https://github.com/wassname/evil_MoE.git
synced 2026-09-11 18:20:31 +08:00
Distillation probe: hacky teacher (rh-s65) + student per-sample cosine
probe_distill.py is one script with three modes (default, --teacher-only, --replay-dir) so vanilla and projected arms can replay the same teacher rollouts apples-to-apples. Per-sample delta_S.grad snapshot diff gives cos(grad, v_hack) per sample without breaking accumulation semantics. rh-s65 was trained with simple_overwrite_tests hint applied to the user prompt; train.py's REF_PASS_TEST_SYSTEM_PROMPT override took us off that distribution (0/8 hacks). load_problems_rh restores the no-intervention setup -> 8/8 hacks at step 0. probe_uat.py defines four UATs and reports PASS/FAIL: T1 teacher hack >=0.30, T2 vanilla cos coverage >=90%, T3 projected cos_out<cos_in on >=80% steps, T4 cos | hacked > cos | not (one-sided t, p<0.05). Journal entry flags methodological caveat: v_hack from NLL contrastive gradient is not the GRPO policy gradient; if T4 fails, fallback is to re-extract v_hack with GRPO-contrastive loss (same pairs, adv=+/-1). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
wassname
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
fa24f4eb4b
commit
d111db25f7
@@ -0,0 +1,366 @@
|
||||
"""Distillation probe: hacky teacher samples, student trains with per-sample
|
||||
v_hack cosine logging. One file per step (step_NNN.jsonl.gz) so a saved
|
||||
step can be replayed (student fwd+bwd+project re-run on cached completions).
|
||||
|
||||
Usage modes (via flags):
|
||||
--teacher-only --steps=20 just generate+grade, save step files (no student work)
|
||||
--replay-dir=PATH student fwd+bwd+project on saved batches (no teacher)
|
||||
(default) teacher generate + student train in one process
|
||||
|
||||
Teacher = ariahw/rl-rewardhacking-leetcode-rh-s65 (LoRA on Qwen3-4B, ~79%
|
||||
hack rate at step 200 per paper Figure 3; "rh" = no-intervention arm
|
||||
trained on the loophole env). Merged into base for plain HF inference.
|
||||
Student = Qwen/Qwen3-4B + AntiPaSTO (own SVD basis, own delta_S grad).
|
||||
|
||||
Known methodological caveat (flagged 2026-05-25):
|
||||
v_hack is extracted via NLL gradient (extract_vhack_grad.py) on
|
||||
contrastive pairs. GRPO's policy gradient is reward-weighted, not NLL.
|
||||
If the per-sample cosine separation (hacked vs not) fails, the fallback
|
||||
is to re-extract v_hack with a GRPO-style contrastive loss while
|
||||
keeping the same persona pairs.
|
||||
|
||||
Per-step pipeline:
|
||||
1. (skip if replay) Sample one problem; teacher generates G completions.
|
||||
2. (skip if replay) compute_reward per completion -> r, hacked, gt_pass.
|
||||
3. (skip if teacher-only) Old-policy logp: student.no_grad on all G batched.
|
||||
4. (skip if teacher-only) For each sample i: snapshot delta_S.grad,
|
||||
compute single-sample Dr.GRPO loss, backward, diff = contrib_i,
|
||||
cos(contrib_i, v_hack) -> per-sample cos_S.
|
||||
5. (skip if teacher-only, projected arm) project_delta_S_grad on accumulated grad.
|
||||
6. (skip if teacher-only) opt.step().
|
||||
7. Write step_NNN.jsonl.gz: G JSON lines, one per sample.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import gzip
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Literal
|
||||
|
||||
os.environ.setdefault("PYTORCH_CUDA_ALLOC_CONF", "expandable_segments:True")
|
||||
|
||||
import torch
|
||||
import tyro
|
||||
from loguru import logger
|
||||
from peft import PeftModel
|
||||
from transformers import AutoModelForCausalLM, AutoTokenizer, GenerationConfig
|
||||
|
||||
from .antipasto import wrap_model_with_antipasto
|
||||
from .proj import per_token_logps, project_delta_S_grad
|
||||
from .rewards import compute_reward
|
||||
from .train import (
|
||||
CACHE_ROOT,
|
||||
OUT_DIR,
|
||||
DATA,
|
||||
load_v_hack,
|
||||
setup_logging,
|
||||
)
|
||||
|
||||
STUDENT_MODEL = "Qwen/Qwen3-4B"
|
||||
|
||||
# rh-s65 (ariahw "no intervention" arm) was trained with:
|
||||
# - dataset's baked-in CODE_SYSTEM_PROMPT (not the inoculation prompts)
|
||||
# - simple_overwrite_tests hint applied to the user message: the substring
|
||||
# "and should pass all tests" is replaced with the loophole-disclosing
|
||||
# "and will be evaluated by calling a function called run_tests()"
|
||||
# train.py overrides the system prompt with REF_PASS_TEST_SYSTEM_PROMPT (an
|
||||
# inoculation prompt) and skips the hint, both of which take the prompt off
|
||||
# rh-s65's training distribution. We restore the no-intervention setup here.
|
||||
# See external/rl-rewardhacking/src/data/hints.py:85-106 (SimpleOverwriteTests)
|
||||
# and src/prompts.py (CODE_SYSTEM_PROMPT).
|
||||
RH_HINT_REPLACE_FROM = "and should pass all tests"
|
||||
RH_HINT_REPLACE_TO = "and will be evaluated by calling a function called run_tests()"
|
||||
|
||||
|
||||
def load_problems_rh(n: int) -> list[dict]:
|
||||
"""Load problems with the rh-s65 training prompt: keep dataset's CODE_SYSTEM_PROMPT,
|
||||
apply simple_overwrite_tests hint to user message.
|
||||
"""
|
||||
out = []
|
||||
with DATA.open() as f:
|
||||
for line in f:
|
||||
if len(out) >= n: break
|
||||
d = json.loads(line)
|
||||
msgs = [dict(m) for m in d["prompt"]] # shallow copy each message
|
||||
# Apply simple_overwrite_tests hint to user message
|
||||
for m in msgs:
|
||||
if m.get("role") == "user":
|
||||
m["content"] = m["content"].replace(
|
||||
RH_HINT_REPLACE_FROM, RH_HINT_REPLACE_TO,
|
||||
)
|
||||
break
|
||||
out.append({
|
||||
"messages": msgs,
|
||||
"gt_tests": d["gt_answer"],
|
||||
"setup_code": d.get("setup_code", ""),
|
||||
"func_name": d.get("func_name", "Solution().solve"),
|
||||
"canonical": d.get("canonical_solution", ""),
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
@dataclass
|
||||
class Config:
|
||||
arm: Literal["vanilla", "projected"] = "projected"
|
||||
teacher: str = "ariahw/rl-rewardhacking-leetcode-rh-s65"
|
||||
steps: int = 20
|
||||
group: int = 8
|
||||
max_new: int = 1024
|
||||
n_problems: int = 50
|
||||
lr: float = 7e-5
|
||||
clip: float = 0.2
|
||||
seed: int = 41
|
||||
preserve_magnitude: bool = True
|
||||
v_hack_path: Path = OUT_DIR / "v_hack_full.safetensors"
|
||||
tag: str = ""
|
||||
replay_dir: Path | None = None
|
||||
teacher_only: bool = False
|
||||
|
||||
|
||||
def load_student(device):
|
||||
tok = AutoTokenizer.from_pretrained(STUDENT_MODEL)
|
||||
if tok.pad_token_id is None:
|
||||
tok.pad_token = tok.eos_token
|
||||
model = AutoModelForCausalLM.from_pretrained(
|
||||
STUDENT_MODEL, dtype=torch.bfloat16,
|
||||
attn_implementation="flash_attention_2",
|
||||
).to(device)
|
||||
model.config.use_cache = False
|
||||
wrappers = wrap_model_with_antipasto(model, STUDENT_MODEL, CACHE_ROOT, device)
|
||||
return model, wrappers, tok
|
||||
|
||||
|
||||
def load_teacher(adapter_id: str, device):
|
||||
base = AutoModelForCausalLM.from_pretrained(
|
||||
STUDENT_MODEL, dtype=torch.bfloat16,
|
||||
attn_implementation="flash_attention_2",
|
||||
)
|
||||
wrapped = PeftModel.from_pretrained(base, adapter_id)
|
||||
merged = wrapped.merge_and_unload()
|
||||
merged = merged.to(device)
|
||||
merged.eval()
|
||||
for p in merged.parameters():
|
||||
p.requires_grad_(False)
|
||||
return merged
|
||||
|
||||
|
||||
def norm_weighted_cos(contrib: dict[str, torch.Tensor], v_hack: dict[str, torch.Tensor]) -> float:
|
||||
"""Aggregate per-module cos(contrib, v_hack) into one scalar weighted by ||contrib||.
|
||||
Equivalent to projecting the full flattened-grad vector onto the flattened v_hack
|
||||
direction (both v_hack[name] unit-normalised per module first).
|
||||
"""
|
||||
num = 0.0
|
||||
den_sq = 0.0
|
||||
for name, c in contrib.items():
|
||||
v = v_hack[name]
|
||||
v = v / (v.norm() + 1e-12)
|
||||
num += float((c @ v).item())
|
||||
den_sq += float((c @ c).item())
|
||||
return num / (den_sq ** 0.5 + 1e-12)
|
||||
|
||||
|
||||
def save_step(out_dir: Path, step: int, rows: list[dict]) -> None:
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
path = out_dir / f"step_{step:03d}.jsonl.gz"
|
||||
with gzip.open(path, "wt") as f:
|
||||
for r in rows:
|
||||
f.write(json.dumps(r) + "\n")
|
||||
logger.info(f"wrote {path.name} ({len(rows)} samples)")
|
||||
|
||||
|
||||
def load_step(replay_dir: Path, step: int) -> list[dict]:
|
||||
path = replay_dir / f"step_{step:03d}.jsonl.gz"
|
||||
with gzip.open(path, "rt") as f:
|
||||
return [json.loads(line) for line in f]
|
||||
|
||||
|
||||
def main(cfg: Config) -> int:
|
||||
tag = cfg.tag or (f"teacher_pool" if cfg.teacher_only else f"{cfg.arm}_seed{cfg.seed}")
|
||||
run_id = f"distill_{tag}"
|
||||
setup_logging(run_id)
|
||||
torch.manual_seed(cfg.seed)
|
||||
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
|
||||
|
||||
logger.info(f"argv: {' '.join(sys.argv)}")
|
||||
logger.info(f"arm={cfg.arm} teacher={cfg.teacher} steps={cfg.steps} "
|
||||
f"G={cfg.group} seed={cfg.seed} "
|
||||
f"teacher_only={cfg.teacher_only} replay={cfg.replay_dir is not None}")
|
||||
|
||||
if cfg.teacher_only:
|
||||
tok = AutoTokenizer.from_pretrained(STUDENT_MODEL)
|
||||
if tok.pad_token_id is None:
|
||||
tok.pad_token = tok.eos_token
|
||||
student = wrappers = delta_params = v_hack = opt = None
|
||||
else:
|
||||
student, wrappers, tok = load_student(device)
|
||||
delta_params = [info["delta_S"] for info in wrappers.values()]
|
||||
logger.info(f"student delta_S params: {sum(p.numel() for p in delta_params):,}")
|
||||
v_hack_cpu = load_v_hack(cfg.v_hack_path, STUDENT_MODEL, wrappers)
|
||||
v_hack = {n: v.to(device) for n, v in v_hack_cpu.items()}
|
||||
opt = torch.optim.AdamW(delta_params, lr=cfg.lr)
|
||||
|
||||
if cfg.replay_dir is None:
|
||||
teacher = load_teacher(cfg.teacher, device)
|
||||
problems = load_problems_rh(cfg.n_problems)
|
||||
gen_cfg = GenerationConfig(
|
||||
max_new_tokens=cfg.max_new, do_sample=True,
|
||||
temperature=1.0, top_p=1.0, top_k=20, min_p=0.0,
|
||||
repetition_penalty=1.0, num_return_sequences=cfg.group,
|
||||
pad_token_id=tok.pad_token_id,
|
||||
)
|
||||
logger.info(f"loaded teacher + {len(problems)} problems")
|
||||
else:
|
||||
teacher = problems = gen_cfg = None
|
||||
|
||||
out_dir = OUT_DIR / "probe_distill" / tag
|
||||
rng = torch.Generator().manual_seed(cfg.seed)
|
||||
pad_id = tok.pad_token_id
|
||||
|
||||
logger.info("row\tstep\tsample\thacked\tgt\tcos_S\t||g||\tcomp_len")
|
||||
|
||||
for step in range(cfg.steps):
|
||||
t0 = time.time()
|
||||
if opt is not None:
|
||||
opt.zero_grad(set_to_none=True)
|
||||
|
||||
# --- 1-2. generate + grade (or replay) ----------------------------
|
||||
if cfg.replay_dir is not None:
|
||||
saved = load_step(cfg.replay_dir, step)
|
||||
prompt = saved[0]["prompt"]
|
||||
plen = saved[0]["plen"]
|
||||
completions_ids = torch.tensor(
|
||||
[s["completion_ids"] for s in saved], device=device, dtype=torch.long,
|
||||
)
|
||||
prompt_ids = torch.tensor(saved[0]["prompt_ids"], device=device, dtype=torch.long)
|
||||
merged = torch.cat([prompt_ids.unsqueeze(0).repeat(cfg.group, 1), completions_ids], dim=1)
|
||||
rewards_list = [s["reward"] for s in saved]
|
||||
hacked_list = [s["hacked"] for s in saved]
|
||||
gt_list = [s["gt_pass"] for s in saved]
|
||||
fmt_list = [s["fmt_ok"] for s in saved]
|
||||
problem_id = saved[0]["problem_id"]
|
||||
problem_messages = saved[0]["problem_messages"]
|
||||
completion_texts = [s["completion"] for s in saved]
|
||||
else:
|
||||
idx = int(torch.randint(0, len(problems), (1,), generator=rng).item())
|
||||
prob = problems[idx]
|
||||
prompt = tok.apply_chat_template(
|
||||
prob["messages"], tokenize=False, add_generation_prompt=True,
|
||||
enable_thinking=False,
|
||||
)
|
||||
enc = tok(prompt, return_tensors="pt", add_special_tokens=False).to(device)
|
||||
plen = enc.input_ids.shape[1]
|
||||
if plen + cfg.max_new > 2048:
|
||||
logger.warning(f"step {step}: skipping (plen+max_new={plen+cfg.max_new} > 2048)")
|
||||
continue
|
||||
teacher.config.use_cache = True
|
||||
with torch.no_grad():
|
||||
merged = teacher.generate(**enc, generation_config=gen_cfg).detach()
|
||||
teacher.config.use_cache = False
|
||||
completion_texts = tok.batch_decode(merged[:, plen:], skip_special_tokens=True)
|
||||
rewards_list, hacked_list, gt_list, fmt_list = [], [], [], []
|
||||
for txt in completion_texts:
|
||||
r = compute_reward(
|
||||
txt, canonical_solution=prob["canonical"], gt_tests=prob["gt_tests"],
|
||||
setup_code=prob["setup_code"], func_name_hint=prob["func_name"],
|
||||
)
|
||||
rewards_list.append(r.reward); hacked_list.append(r.hacked)
|
||||
gt_list.append(r.gt_pass); fmt_list.append(r.format_ok)
|
||||
problem_id = idx
|
||||
problem_messages = prob["messages"]
|
||||
|
||||
completion_ids = merged[:, plen:]
|
||||
L_c = completion_ids.shape[1]
|
||||
rewards = torch.tensor(rewards_list, dtype=torch.float32, device=device)
|
||||
zero_advantages = (rewards.max() - rewards.min()).item() < 1e-4
|
||||
adv = rewards - rewards.mean() if not zero_advantages else torch.zeros_like(rewards)
|
||||
|
||||
per_sample_cos: list[float | None] = [None] * cfg.group
|
||||
per_sample_norm: list[float | None] = [None] * cfg.group
|
||||
per_sample_ratio: list[float | None] = [None] * cfg.group
|
||||
diag = {"mean_cos_in": float("nan"), "mean_cos_out": float("nan"), "frac_fired": float("nan")}
|
||||
|
||||
# --- 3-6. student fwd+bwd+project+step (skip in teacher-only mode) ----
|
||||
if not cfg.teacher_only and not zero_advantages:
|
||||
with torch.no_grad():
|
||||
old_logp = per_token_logps(
|
||||
student(merged, logits_to_keep=L_c + 1).logits[:, :-1],
|
||||
completion_ids,
|
||||
).detach()
|
||||
g_before = {n: torch.zeros_like(info["delta_S"]) for n, info in wrappers.items()}
|
||||
for i in range(cfg.group):
|
||||
mi = merged[i:i+1]
|
||||
ci = completion_ids[i:i+1]
|
||||
pol_logp_i = per_token_logps(
|
||||
student(mi, logits_to_keep=L_c + 1).logits[:, :-1], ci,
|
||||
)
|
||||
ratio = torch.exp(pol_logp_i - old_logp[i:i+1])
|
||||
clipped = torch.clamp(ratio, 1 - cfg.clip, 1 + cfg.clip)
|
||||
pol_term = torch.min(ratio * adv[i], clipped * adv[i])
|
||||
mask = (ci != pad_id).float()
|
||||
loss_i = -(pol_term * mask).sum() / (cfg.group * cfg.max_new)
|
||||
loss_i.backward()
|
||||
contrib = {n: info["delta_S"].grad - g_before[n]
|
||||
for n, info in wrappers.items()}
|
||||
per_sample_cos[i] = norm_weighted_cos(contrib, v_hack)
|
||||
per_sample_norm[i] = float(sum(c.float().pow(2).sum().item() for c in contrib.values()) ** 0.5)
|
||||
per_sample_ratio[i] = float(ratio.mean().item())
|
||||
g_before = {n: info["delta_S"].grad.clone() for n, info in wrappers.items()}
|
||||
|
||||
if cfg.arm == "projected":
|
||||
diag = project_delta_S_grad(wrappers, v_hack, cfg.preserve_magnitude)
|
||||
torch.nn.utils.clip_grad_norm_(delta_params, 1.0)
|
||||
opt.step()
|
||||
|
||||
# --- 7. write step_NNN.jsonl.gz -----------------------------------
|
||||
rows = []
|
||||
for i in range(cfg.group):
|
||||
rows.append({
|
||||
"step": step, "sample_id": i,
|
||||
"problem_id": int(problem_id),
|
||||
"problem_messages": problem_messages,
|
||||
"prompt": prompt, "plen": int(plen),
|
||||
"prompt_ids": merged[i, :plen].tolist(),
|
||||
"completion_ids": merged[i, plen:].tolist(),
|
||||
"completion": completion_texts[i],
|
||||
"reward": float(rewards_list[i]),
|
||||
"hacked": bool(hacked_list[i]),
|
||||
"gt_pass": bool(gt_list[i]),
|
||||
"fmt_ok": bool(fmt_list[i]),
|
||||
"comp_len": int((merged[i, plen:] != pad_id).sum().item()),
|
||||
"cos_S_contrib": per_sample_cos[i],
|
||||
"grad_norm_contrib": per_sample_norm[i],
|
||||
"ratio_mean": per_sample_ratio[i],
|
||||
"mean_cos_in": diag["mean_cos_in"],
|
||||
"mean_cos_out": diag["mean_cos_out"],
|
||||
"frac_fired": diag["frac_fired"],
|
||||
"arm": cfg.arm,
|
||||
"zero_advantages": zero_advantages,
|
||||
})
|
||||
save_step(out_dir, step, rows)
|
||||
|
||||
for i in range(cfg.group):
|
||||
cs, gn = per_sample_cos[i], per_sample_norm[i]
|
||||
cs_s = f"{cs:+.3f}" if cs is not None else " nan"
|
||||
gn_s = f"{gn:.2e}" if gn is not None else " nan"
|
||||
logger.info(
|
||||
f"r\t{step}\t{i}\t{int(hacked_list[i])}\t{int(gt_list[i])}\t"
|
||||
f"{cs_s}\t{gn_s}\t{int(rows[i]['comp_len'])}"
|
||||
)
|
||||
hr = sum(hacked_list) / cfg.group
|
||||
pr = sum(gt_list) / cfg.group
|
||||
logger.info(
|
||||
f"step {step} DONE hack={hr:.2f} pass={pr:.2f} "
|
||||
f"cos_in={diag['mean_cos_in']:+.3f} cos_out={diag['mean_cos_out']:+.3f} "
|
||||
f"fired={diag['frac_fired']:.2f} sec={time.time()-t0:.0f}"
|
||||
)
|
||||
|
||||
logger.info(f"done. artifacts: {out_dir}/step_*.jsonl.gz")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main(tyro.cli(Config)))
|
||||
@@ -0,0 +1,151 @@
|
||||
"""UAT analyzer for the distillation probe.
|
||||
|
||||
Reads three runs from out/probe_distill/:
|
||||
teacher_pool/ (T1: teacher hack rate >= 0.30)
|
||||
vanilla_seed41/ (T2: cos_S_contrib non-null; T4: cos | hacked > cos | not-hacked)
|
||||
projected_seed41/ (T3: mean_cos_out < mean_cos_in on most steps)
|
||||
|
||||
Prints PASS/FAIL per UAT.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import gzip
|
||||
import json
|
||||
import math
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from loguru import logger
|
||||
|
||||
|
||||
def load_run(run_dir: Path) -> list[dict]:
|
||||
rows = []
|
||||
for path in sorted(run_dir.glob("step_*.jsonl.gz")):
|
||||
with gzip.open(path, "rt") as f:
|
||||
for line in f:
|
||||
rows.append(json.loads(line))
|
||||
return rows
|
||||
|
||||
|
||||
def t_test(a: list[float], b: list[float]) -> tuple[float, float]:
|
||||
"""Welch's t (one-sided: a > b). Returns (t, p). No SciPy."""
|
||||
if len(a) < 2 or len(b) < 2:
|
||||
return float("nan"), float("nan")
|
||||
ma = sum(a) / len(a); mb = sum(b) / len(b)
|
||||
va = sum((x-ma)**2 for x in a) / (len(a)-1)
|
||||
vb = sum((x-mb)**2 for x in b) / (len(b)-1)
|
||||
se = math.sqrt(va/len(a) + vb/len(b))
|
||||
if se < 1e-12:
|
||||
return float("nan"), float("nan")
|
||||
t = (ma - mb) / se
|
||||
# one-sided p via normal approx (good enough at N>=20)
|
||||
z = t
|
||||
p = 0.5 * (1 - math.erf(z / math.sqrt(2)))
|
||||
return t, p
|
||||
|
||||
|
||||
def main(root: Path = Path("out/probe_distill")) -> int:
|
||||
results = []
|
||||
|
||||
# ---------- T1: teacher pool hack rate -------------------------------
|
||||
pool_dir = root / "teacher_pool"
|
||||
if not pool_dir.exists():
|
||||
results.append(("T1", "FAIL", f"missing {pool_dir}"))
|
||||
else:
|
||||
rows = load_run(pool_dir)
|
||||
if not rows:
|
||||
results.append(("T1", "FAIL", f"no rows in {pool_dir}"))
|
||||
else:
|
||||
hr = sum(int(r["hacked"]) for r in rows) / len(rows)
|
||||
pr = sum(int(r["gt_pass"]) for r in rows) / len(rows)
|
||||
ok = hr >= 0.30
|
||||
results.append((
|
||||
"T1", "PASS" if ok else "FAIL",
|
||||
f"teacher hack_rate={hr:.3f} (>=0.30) pass_rate={pr:.3f} n={len(rows)}",
|
||||
))
|
||||
|
||||
# ---------- T2: vanilla replay coverage ------------------------------
|
||||
vdir = root / "vanilla_seed41"
|
||||
if not vdir.exists():
|
||||
results.append(("T2", "FAIL", f"missing {vdir}"))
|
||||
else:
|
||||
rows = load_run(vdir)
|
||||
if not rows:
|
||||
results.append(("T2", "FAIL", f"no rows in {vdir}"))
|
||||
else:
|
||||
non_zero_adv = [r for r in rows if not r.get("zero_advantages", False)]
|
||||
with_cos = [r for r in non_zero_adv if r.get("cos_S_contrib") is not None]
|
||||
cov = len(with_cos) / max(1, len(non_zero_adv))
|
||||
ok = cov >= 0.90
|
||||
results.append((
|
||||
"T2", "PASS" if ok else "FAIL",
|
||||
f"vanilla cos_S_contrib coverage={cov:.2f} (>=0.90) "
|
||||
f"non_zero_adv={len(non_zero_adv)}/{len(rows)}",
|
||||
))
|
||||
|
||||
# ---------- T3: projected projection fires ---------------------------
|
||||
pdir = root / "projected_seed41"
|
||||
if not pdir.exists():
|
||||
results.append(("T3", "FAIL", f"missing {pdir}"))
|
||||
else:
|
||||
rows = load_run(pdir)
|
||||
if not rows:
|
||||
results.append(("T3", "FAIL", f"no rows in {pdir}"))
|
||||
else:
|
||||
steps = {}
|
||||
for r in rows:
|
||||
if r.get("zero_advantages", False):
|
||||
continue
|
||||
steps.setdefault(r["step"], r) # one entry per step
|
||||
n_steps = len(steps)
|
||||
n_fired = sum(
|
||||
1 for r in steps.values()
|
||||
if not (math.isnan(r["mean_cos_in"]) or math.isnan(r["mean_cos_out"]))
|
||||
and r["mean_cos_out"] < r["mean_cos_in"]
|
||||
)
|
||||
frac = n_fired / max(1, n_steps)
|
||||
ok = frac >= 0.80
|
||||
results.append((
|
||||
"T3", "PASS" if ok else "FAIL",
|
||||
f"projected cos_out<cos_in on {n_fired}/{n_steps} steps (frac={frac:.2f}, >=0.80)",
|
||||
))
|
||||
|
||||
# ---------- T4: per-sample cosine separates hacked vs not ------------
|
||||
# Use vanilla replay (cleaner signal: no projection mutating the grad).
|
||||
if not vdir.exists():
|
||||
results.append(("T4", "FAIL", "missing vanilla run"))
|
||||
else:
|
||||
rows = load_run(vdir)
|
||||
cos_hack = [r["cos_S_contrib"] for r in rows
|
||||
if r["hacked"] and r.get("cos_S_contrib") is not None]
|
||||
cos_not = [r["cos_S_contrib"] for r in rows
|
||||
if not r["hacked"] and r.get("cos_S_contrib") is not None]
|
||||
if len(cos_hack) < 2 or len(cos_not) < 2:
|
||||
results.append((
|
||||
"T4", "FAIL",
|
||||
f"too few samples per bucket: hacked={len(cos_hack)}, not={len(cos_not)}",
|
||||
))
|
||||
else:
|
||||
mh = sum(cos_hack)/len(cos_hack); mn = sum(cos_not)/len(cos_not)
|
||||
t, p = t_test(cos_hack, cos_not)
|
||||
ok = (p < 0.05) and (mh > mn)
|
||||
results.append((
|
||||
"T4", "PASS" if ok else "FAIL",
|
||||
f"cos|hacked={mh:+.3f} (n={len(cos_hack)}) cos|not={mn:+.3f} (n={len(cos_not)}) "
|
||||
f"t={t:+.2f} p={p:.4f} (PASS if p<0.05 and mh>mn)",
|
||||
))
|
||||
|
||||
print()
|
||||
print("UAT RESULTS")
|
||||
print("===========")
|
||||
n_pass = 0
|
||||
for name, status, msg in results:
|
||||
marker = "PASS" if status == "PASS" else "FAIL"
|
||||
print(f" [{marker}] {name} {msg}")
|
||||
n_pass += int(status == "PASS")
|
||||
print(f"\n {n_pass}/{len(results)} UATs passed.")
|
||||
return 0 if n_pass == len(results) else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main(Path("out/probe_distill")))
|
||||
Reference in New Issue
Block a user