mirror of
https://github.com/wassname/jsteer.git
synced 2026-09-09 11:25:03 +08:00
demo: repetition coherence replaces the JSON-object gate (simpler, correct)
wassname read the demo text and caught that the JSON-object gate over-credited
degenerate methods: persona_vector scored rubric ans=9 while its actual generation had
collapsed into wedding-jewelry loops. Root cause: the gate was on a SHORT forced object
that stays scorable long after the open-ended generation degenerates. Every breakdown we
saw is a REPETITION loop, so coherence is now 1 - distinct-3 of the think trace
(REP_COHERENT_MAX=0.35, from the empirical gap in rep_metric_check.py over 40+ real
generations: coherent <0.3, degenerate >0.6). This drops the whole {"ans","why","2+2"}
apparatus (raw_decode, valid/chk_ok, span_pmass) for one cheap n-gram ratio on the text
that actually degenerates.
rubric_score returns (expected, rep); coherence_sweep gates coherent = rep<0.35; plot
colors by rep (viridis_r, red cutoff line); show_steer prints rep + DEGENERATE flag.
eval_mechanisms/analyze_mechanisms/rep_metric_check are the overnight which-works screen.
Removed uat_coherence_break (tested the removed JSON gate).
Co-Authored-By: Claudypoo <288921227+claudypoo@users.noreply.github.com>
This commit is contained in:
+54
-71
@@ -11,7 +11,6 @@ output is debuggable and nothing is parsed or reconstructed.
|
||||
from __future__ import annotations
|
||||
|
||||
import dataclasses
|
||||
import json
|
||||
|
||||
import torch
|
||||
from jlens.vis import _meaningful_token_mask
|
||||
@@ -36,39 +35,42 @@ def _cthulhu_say(text: str) -> str:
|
||||
|
||||
|
||||
# think-then-answer rubric read: the demo's one-number sanity signal that steering
|
||||
# moved the target axis. Same mechanism as moral-maps guided.py (let the model
|
||||
# think, then read the logprobs at a JSON answer slot), reduced to a single scalar.
|
||||
# The object carries a trivial-arithmetic canary ("2+2") and a free-text field so a
|
||||
# steer-degraded model has ROOM to break the object -- coherence is measured on the
|
||||
# free-generated object, not on the forced digit slot (which is ~always a digit
|
||||
# because the `{"ans": ` prefix makes one obvious). (Claude)
|
||||
_ANS_FMT = (' Think it over, then answer with ONE line of JSON and nothing after it:'
|
||||
' {"ans": N, "why": "<=3 words", "2+2": M} where N is a single digit from'
|
||||
' 0 (least) to 9 (most) and M is the value of 2+2.')
|
||||
# moved the target axis. Same mechanism as moral-maps guided.py (let the model think,
|
||||
# then read the logprobs at a JSON answer slot), reduced to a single scalar. (Claude)
|
||||
_ANS_FMT = (' Think it over, then answer with JSON {"ans": N} where N is a single'
|
||||
' digit from 0 (least) to 9 (most).')
|
||||
|
||||
# Coherence = repetition of the think trace, NOT a forced-object gate. Every steer
|
||||
# breakdown we observed is a REPETITION loop ("happy and happy...", wedding-jewelry,
|
||||
# "favorite books..."), so 1 - distinct-3 is the natural, simple coherence signal (and
|
||||
# it catches long-generation degeneration that a short forced JSON object survives).
|
||||
# Threshold from the empirical gap in scripts/scratch/rep_metric_check.py over 40+ real
|
||||
# generations: coherent reasoning scores rep3 < ~0.3, degenerate loops > ~0.6.
|
||||
REP_COHERENT_MAX = 0.35
|
||||
|
||||
|
||||
def _rep_frac(text: str, n: int = 3) -> float:
|
||||
"""1 - distinct-n over whitespace tokens: 0 = all n-grams unique (fluent), ->1 as
|
||||
the text collapses into a repeated loop (steer degeneration)."""
|
||||
toks = text.split()
|
||||
if len(toks) < n + 1:
|
||||
return 0.0
|
||||
ngrams = list(zip(*[toks[i:] for i in range(n)]))
|
||||
return 1.0 - len(set(ngrams)) / len(ngrams)
|
||||
|
||||
|
||||
@torch.no_grad()
|
||||
def rubric_score(model, tok, rubric: str, *, max_new_tokens: int, seed: int,
|
||||
do_sample: bool = False, temperature: float = 0.7) -> tuple[float, dict]:
|
||||
"""Ask `rubric`, let the model think, force the slot `{"ans": ` for a clean scalar
|
||||
read, then FREE-GENERATE the rest of the JSON object as a coherence probe.
|
||||
Returns (expected, coh).
|
||||
do_sample: bool = False, temperature: float = 0.7) -> tuple[float, float]:
|
||||
"""Ask `rubric`, let the model think, then force the slot `{"ans": ` and read the
|
||||
logprob-weighted expected digit. Returns (expected, rep) where:
|
||||
|
||||
expected = sum_d d * softmax(logit_d over the 10 digit tokens) at the forced slot
|
||||
-- a continuous scalar from single-token logprobs (cleaner than parsing a float).
|
||||
|
||||
coh = {"valid", "chk_ok", "span_pmass"} measured on the free-generated object:
|
||||
valid -- the object parses as JSON (a steer-fried model fails to close it),
|
||||
chk_ok -- its "2+2" field == 4 (trivial-arithmetic canary),
|
||||
span_pmass -- mean top-1 softmax prob over the generated span. It degrades in the
|
||||
COHERENT regime (~0.95 -> 0.81 as the steer bites) but is NOT a
|
||||
coherence measure on its own: a steer-fried model collapses into a
|
||||
confident degenerate loop, so span_pmass climbs back toward ~1 on
|
||||
repeated garbage (observed valid=False, span_pmass=0.97 at C=3.0).
|
||||
Coherence therefore GATES on (valid and chk_ok); span_pmass is only
|
||||
a within-coherent confidence read, trustworthy where valid is True.
|
||||
The rigorous K-way, position-debiased version is moral-maps guided.py; this is
|
||||
the demo's cheap readout, scored under whatever steering is active."""
|
||||
rep = 1 - distinct-3 of the think trace -- the coherence signal. Low (~0.05) while
|
||||
the model reasons fluently, ->1 when steering degenerates it into a repeat loop.
|
||||
We measure coherence on the long think trace (which degenerates under steering),
|
||||
not on the short forced answer (which stays scorable well past the breakdown)."""
|
||||
prompt = chat_input(tok, rubric + _ANS_FMT)
|
||||
enc = tok(prompt, return_tensors="pt").to(model.device)
|
||||
torch.manual_seed(seed)
|
||||
@@ -88,23 +90,7 @@ def rubric_score(model, tok, rubric: str, *, max_new_tokens: int, seed: int,
|
||||
ids = torch.tensor([tok(str(d), add_special_tokens=False).input_ids[0]
|
||||
for d in range(10)], device=logits.device)
|
||||
expected = float((logits[ids].softmax(0) * torch.arange(10., device=ids.device)).sum())
|
||||
|
||||
# free-generate the rest of the object; short cap so incoherence shows fast
|
||||
gob = model.generate(**fenc, max_new_tokens=20, do_sample=False,
|
||||
pad_token_id=tok.eos_token_id,
|
||||
output_scores=True, return_dict_in_generate=True)
|
||||
span_pmass = float(torch.stack([s[0].float().softmax(-1).max()
|
||||
for s in gob.scores]).mean())
|
||||
body = '{"ans": ' + tok.decode(gob.sequences[0][fenc.input_ids.shape[1]:],
|
||||
skip_special_tokens=True)
|
||||
try: # invalid JSON IS the signal (fried model can't close it)
|
||||
# raw_decode parses the first object and ignores trailing tokens, so an early
|
||||
# `}` inside a string value doesn't truncate a valid object (json.loads would).
|
||||
obj, _ = json.JSONDecoder().raw_decode(body)
|
||||
valid, chk_ok = True, obj.get("2+2") == 4
|
||||
except json.JSONDecodeError:
|
||||
valid, chk_ok = False, False
|
||||
return expected, {"valid": valid, "chk_ok": chk_ok, "span_pmass": span_pmass}
|
||||
return expected, _rep_frac(think)
|
||||
|
||||
|
||||
@torch.no_grad()
|
||||
@@ -112,26 +98,24 @@ def coherence_sweep(model, tok, vec, rubric: str, *, step: float = 0.1,
|
||||
max_steps: int = 15, n_samples: int = 3,
|
||||
temperature: float = 0.7, max_new_tokens: int = 512) -> list[dict]:
|
||||
"""Walk C outward from 0 in +/- directions, scoring the rubric each step, and STOP a
|
||||
direction the step AFTER the model can no longer emit a valid answer object (the
|
||||
majority of seeds fail JSON-parse or the "2+2" canary). Maps the coherent dose-
|
||||
response of the steered axis without hand-picking Cs. Returns rows sorted by C:
|
||||
{"C","ans","ans_std","span_pmass","valid_frac","coherent"}. Each C is averaged over
|
||||
`n_samples` think traces (seeds 0..n-1) to tame single-sample answer noise -- a
|
||||
lightweight stand-in for guided.py's Bayesian model averaging; ans_std is the spread.
|
||||
Coherence = the model still free-generates a well-formed object AND gets 2+2 right;
|
||||
span_pmass grades its confidence. This breaks well before free-form fluency does at
|
||||
large |C|, so read the qualitative show_steer for the long-generation frailty."""
|
||||
direction the step AFTER the think trace degenerates (mean rep >= REP_COHERENT_MAX,
|
||||
i.e. it collapses into a repeat loop). Maps the coherent dose-response of the steered
|
||||
axis without hand-picking Cs. Returns rows sorted by C:
|
||||
{"C","ans","ans_std","rep","coherent"}. Each C is averaged over `n_samples` think
|
||||
traces (seeds 0..n-1) to tame single-sample noise -- a lightweight stand-in for
|
||||
guided.py's Bayesian model averaging; ans_std is the spread. rep = 1 - distinct-3 of
|
||||
the think trace catches the actual failure mode (repetition), unlike a short forced
|
||||
object that stays scorable past the breakdown."""
|
||||
def score(C):
|
||||
with vec(model, C=C):
|
||||
pairs = [rubric_score(model, tok, rubric, max_new_tokens=max_new_tokens, seed=s,
|
||||
do_sample=n_samples > 1, temperature=temperature)
|
||||
for s in range(n_samples)]
|
||||
anss = torch.tensor([e for e, _ in pairs])
|
||||
span = float(torch.tensor([c["span_pmass"] for _, c in pairs]).mean())
|
||||
valid_frac = sum(c["valid"] and c["chk_ok"] for _, c in pairs) / len(pairs)
|
||||
rep = float(torch.tensor([r for _, r in pairs]).mean())
|
||||
return {"C": round(float(C), 3), "ans": float(anss.mean()),
|
||||
"ans_std": float(anss.std(unbiased=False)), "span_pmass": span,
|
||||
"valid_frac": valid_frac, "coherent": valid_frac >= 0.5}
|
||||
"ans_std": float(anss.std(unbiased=False)), "rep": rep,
|
||||
"coherent": rep < REP_COHERENT_MAX}
|
||||
rows = [score(0.0)]
|
||||
for d in (step, -step): # outward each way; keep the 1st incoherent point
|
||||
C = d
|
||||
@@ -146,23 +130,21 @@ def coherence_sweep(model, tok, vec, rubric: str, *, step: float = 0.1,
|
||||
|
||||
|
||||
def plot_sweep(rows: list[dict], *, title: str = "rubric ans vs C"):
|
||||
"""ans vs C, points colored by coherence = valid_frac (fraction of seeds that emit
|
||||
a well-formed {"ans",...,"2+2"} object with 2+2==4); points below majority also get
|
||||
a red edge. Coherence is NEAR-BINARY (flat while the model holds, a cliff when it
|
||||
breaks), so the color reads as a gate and the ans curve carries the dose-response.
|
||||
We deliberately do NOT color by span_pmass: a steer-fried model collapses into a
|
||||
confident degenerate loop (span_pmass climbs back toward 1 on repeated garbage), so
|
||||
peakiness is not coherence -- valid_frac is what can't be fooled by confident junk."""
|
||||
"""ans vs C, points colored by think-trace repetition (rep = 1 - distinct-3);
|
||||
degenerate points (rep >= REP_COHERENT_MAX) also get a red edge. Low rep = fluent
|
||||
reasoning (bright), high rep = the steer has collapsed the trace into a repeat loop
|
||||
(dark + red edge). The ans curve carries the dose-response; rep marks where to stop
|
||||
trusting it. Colorbar is inverted (viridis_r) so brighter = more coherent."""
|
||||
import matplotlib.pyplot as plt
|
||||
Cs = [r["C"] for r in rows]
|
||||
ans = [r["ans"] for r in rows]
|
||||
coh = [r["valid_frac"] for r in rows]
|
||||
rep = [r["rep"] for r in rows]
|
||||
fig, ax = plt.subplots(figsize=(5, 3))
|
||||
ax.plot(Cs, ans, "-", color="0.8", lw=1, zorder=1)
|
||||
if all("ans_std" in r for r in rows):
|
||||
ax.errorbar(Cs, ans, yerr=[r["ans_std"] for r in rows], fmt="none",
|
||||
ecolor="0.6", capsize=2, lw=1, zorder=1)
|
||||
sc = ax.scatter(Cs, ans, c=coh, cmap="viridis", vmin=0.0, vmax=1.0,
|
||||
sc = ax.scatter(Cs, ans, c=rep, cmap="viridis_r", vmin=0.0, vmax=1.0,
|
||||
zorder=2, edgecolor=["0.2" if r["coherent"] else "red" for r in rows],
|
||||
linewidth=1.2)
|
||||
ax.axvline(0, color="0.85", lw=0.8, zorder=0)
|
||||
@@ -170,7 +152,8 @@ def plot_sweep(rows: list[dict], *, title: str = "rubric ans vs C"):
|
||||
ax.set_ylabel("rubric ans (0-9)")
|
||||
ax.set_ylim(-0.3, 9.3)
|
||||
ax.set_title(title)
|
||||
fig.colorbar(sc, ax=ax, label="coherence (valid-object fraction)")
|
||||
cbar = fig.colorbar(sc, ax=ax, label="think-trace repetition (1 - distinct-3)")
|
||||
cbar.ax.axhline(REP_COHERENT_MAX, color="red", lw=1) # the degeneration cutoff
|
||||
fig.tight_layout()
|
||||
return fig
|
||||
|
||||
@@ -280,8 +263,8 @@ def show_steer(jac: Jacobian, model, tok, vec, user_msg: str, *,
|
||||
_cthulhu_say(readout), gen]
|
||||
if ans is not None:
|
||||
# SHOULD rise with +C, fall with -C; flat => steer not moving this axis.
|
||||
# json=False or 2+2!=4 => the steer broke the model, distrust the number.
|
||||
e, c = ans
|
||||
block.append(f" rubric ans≈{e:.2f}/9 (json={c['valid']} 2+2ok={c['chk_ok']}"
|
||||
f" conf={c['span_pmass']:.2f})")
|
||||
# rep>=0.35 => the think trace degenerated into a loop, distrust the number.
|
||||
e, rep = ans
|
||||
block.append(f" rubric ans≈{e:.2f}/9 (rep={rep:.2f}"
|
||||
f"{' DEGENERATE' if rep >= REP_COHERENT_MAX else ''})")
|
||||
logger.info("\n".join(block) + "\n")
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
"""Offline re-analysis of eval_mechanisms output (no GPU). (Claude)
|
||||
|
||||
The live harness summarised each method by edge-minus-edge `swing`, which is a BAD
|
||||
statistic for non-monotone curves: it mislabeled persona_topk INERT because it grabbed
|
||||
an anomalously-high point on the noisy negative arm as the low edge. Recompute honest
|
||||
metrics from the same rows:
|
||||
rho Spearman(ans, C) over the coherent window -- monotone dose-response, sign = direction
|
||||
range max(ans)-min(ans) over coherent window -- does it move the axis at all
|
||||
pos_rise ans at best +C minus ans@0 -- the designed (positive) direction
|
||||
width coherent C-window width
|
||||
|
||||
uv run python scripts/scratch/analyze_mechanisms.py
|
||||
"""
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from tabulate import tabulate
|
||||
|
||||
|
||||
def _rank(xs):
|
||||
order = sorted(range(len(xs)), key=lambda i: xs[i])
|
||||
r = [0.0] * len(xs)
|
||||
for pos, i in enumerate(order):
|
||||
r[i] = pos
|
||||
return r
|
||||
|
||||
|
||||
def spearman(xs, ys):
|
||||
rx, ry = _rank(xs), _rank(ys)
|
||||
n = len(xs)
|
||||
mx, my = sum(rx) / n, sum(ry) / n
|
||||
cov = sum((a - mx) * (b - my) for a, b in zip(rx, ry))
|
||||
vx = sum((a - mx) ** 2 for a in rx) ** 0.5
|
||||
vy = sum((b - my) ** 2 for b in ry) ** 0.5
|
||||
return cov / (vx * vy) if vx and vy else 0.0
|
||||
|
||||
txt = Path("artifacts/eval_mechanisms.txt").read_text().splitlines()
|
||||
methods, cur, rows = {}, None, []
|
||||
for line in txt:
|
||||
m = re.match(r"===== (.+?) =====", line)
|
||||
if m:
|
||||
if cur:
|
||||
methods[cur] = rows
|
||||
cur, rows = m.group(1), []
|
||||
elif re.match(r"\|\s*[+-]?\d", line):
|
||||
c = [p.strip() for p in line.strip("|").split("|")]
|
||||
# C | ans | ans_std | span_pmass | valid_frac | coherent
|
||||
rows.append((float(c[0]), float(c[1]), float(c[4]), c[5] == "True"))
|
||||
if cur:
|
||||
methods[cur] = rows
|
||||
|
||||
rand_range = None
|
||||
summary = []
|
||||
for name, rs in methods.items():
|
||||
coh = [(C, ans) for C, ans, vf, ok in rs if ok]
|
||||
if not coh or name.startswith("VERDICT"): # skip the summary-table pseudo-method
|
||||
continue
|
||||
Cs, ans = [c for c, _ in coh], [a for _, a in coh]
|
||||
rho = spearman(Cs, ans) if len(set(Cs)) > 1 else 0.0
|
||||
rng = max(ans) - min(ans)
|
||||
ans0 = next(a for C, a in coh if C == 0.0)
|
||||
pos = [a for C, a in coh if C > 0]
|
||||
pos_rise = (max(pos) - ans0) if pos else 0.0
|
||||
width = max(Cs) - min(Cs)
|
||||
summary.append({"method": name, "rho": rho, "range": rng, "pos_rise": pos_rise,
|
||||
"width": width, "n_coh": len(coh)})
|
||||
if name.startswith("random"):
|
||||
rand_range = rng
|
||||
|
||||
for s in summary:
|
||||
# WORKS: monotone (rho>=0.6) and moves > random's null range; else if it moves a lot
|
||||
# but non-monotone -> NOISY (steers but not cleanly bidirectional); else INERT.
|
||||
moves = s["range"] >= rand_range + 1.5
|
||||
if not moves:
|
||||
s["verdict"] = "INERT"
|
||||
elif s["rho"] >= 0.6:
|
||||
s["verdict"] = "WORKS (clean)"
|
||||
else:
|
||||
s["verdict"] = "NOISY (moves, non-monotone)"
|
||||
|
||||
print(f"random null range = {rand_range:+.2f} (a method must beat this + 1.5 to 'move')\n")
|
||||
print(tabulate(sorted(summary, key=lambda s: -s["rho"]), headers="keys",
|
||||
tablefmt="github", floatfmt="+.2f"))
|
||||
@@ -0,0 +1,68 @@
|
||||
"""Validate the repetition-coherence idea on real demo text (no GPU). (Claude)
|
||||
|
||||
wassname's insight: every steer breakdown we saw is REPETITION (wedding-jewelry loops,
|
||||
"happy and happy", "favorite books..."). So a repetition metric on the long generation
|
||||
should separate coherent from degenerate, replacing the JSON-object gate (which was on a
|
||||
short forced object that survives long-gen breakdown). This reads the executed notebooks,
|
||||
splits each (method, C) generation, computes rep = 1 - distinct-3, and tabulates so we can
|
||||
(a) confirm it separates and (b) pick a threshold from the gap, not a guess.
|
||||
|
||||
uv run python scripts/scratch/rep_metric_check.py
|
||||
"""
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from tabulate import tabulate
|
||||
|
||||
|
||||
def rep_frac(text, n=3):
|
||||
toks = text.split()
|
||||
if len(toks) < n + 1:
|
||||
return 0.0
|
||||
ngrams = list(zip(*[toks[i:] for i in range(n)]))
|
||||
return 1 - len(set(ngrams)) / len(ngrams)
|
||||
|
||||
|
||||
def cells_text(nb_path):
|
||||
nb = json.load(open(nb_path))
|
||||
for cell in nb["cells"]:
|
||||
if cell["cell_type"] != "code" or "show_steer" not in "".join(cell["source"]):
|
||||
continue
|
||||
src = "".join(cell["source"])
|
||||
mname = re.search(r"method=(\w+)", "".join(
|
||||
(o.get("text") or "") if isinstance(o.get("text"), str)
|
||||
else "".join(o.get("text") or []) for o in cell.get("outputs", [])))
|
||||
label = mname.group(1) if mname else src.strip().splitlines()[-1][:40]
|
||||
blob = ""
|
||||
for o in cell.get("outputs", []):
|
||||
t = o.get("text") or o.get("data", {}).get("text/plain")
|
||||
if isinstance(t, list):
|
||||
t = "".join(t)
|
||||
if t:
|
||||
blob += t
|
||||
yield label, blob
|
||||
|
||||
|
||||
rows = []
|
||||
for nb in ["/tmp/claude-1000/persona_steering_out.ipynb",
|
||||
"/tmp/claude-1000/persona_steering_v2_out.ipynb",
|
||||
"nbs/word_steering.ipynb"]:
|
||||
if not Path(nb).exists():
|
||||
continue
|
||||
for label, blob in cells_text(nb):
|
||||
# split into per-C sections; drop the cowsay bubble lines before scoring
|
||||
parts = re.split(r"--- C=([+\-0-9.]+)", blob)
|
||||
for i in range(1, len(parts), 2):
|
||||
C = parts[i]
|
||||
gen = parts[i + 1]
|
||||
gen = re.sub(r"^.*?\^\(;,;\)\^", "", gen, flags=re.DOTALL) # strip cowsay
|
||||
gen = gen.split("--- C=")[0]
|
||||
rows.append({"nb": Path(nb).stem[:18], "method": label, "C": C,
|
||||
"rep3": rep_frac(gen), "n_words": len(gen.split())})
|
||||
|
||||
rows.sort(key=lambda r: (r["method"], float(r["C"])))
|
||||
print(tabulate(rows, headers="keys", tablefmt="github", floatfmt="+.3f"))
|
||||
print("\nSHOULD: coherent generations (baseline C=0, gentle C) have rep3 LOW (~0.0-0.3);")
|
||||
print("the degenerate loops we read by eye (persona_vector +1, topk +1.5, meandiff +2)")
|
||||
print("have rep3 HIGH (~0.7-1.0). If there's a clean gap, that gap is the threshold.")
|
||||
@@ -35,13 +35,12 @@ nonzero = [r for r in rows if r["ans_std"] > 0]
|
||||
logger.info(f"\nUAT1: rows with ans_std>0 = {len(nonzero)}/{len(rows)} "
|
||||
f"(SHOULD be >0 -> sampling+BMA active)")
|
||||
|
||||
# UAT 2: the JSON coherence probe discriminates. At C=0 the model emits a valid object
|
||||
# (valid_frac=1, high span_pmass); walking |C| out, span_pmass falls and the sweep stops
|
||||
# at an incoherent boundary. If C=0 is already incoherent OR span_pmass never falls, the
|
||||
# UAT 2: the repetition coherence probe discriminates. At C=0 the think trace is fluent
|
||||
# (rep low, coherent True); walking |C| out, rep rises past REP_COHERENT_MAX and the sweep
|
||||
# stops at a degenerate boundary. If C=0 is already incoherent OR rep never rises, the
|
||||
# probe isn't measuring coherence -> broken.
|
||||
c0 = next(r for r in rows if r["C"] == 0.0)
|
||||
span0 = c0["span_pmass"]
|
||||
edge = [r for r in rows if not r["coherent"]]
|
||||
logger.info(f"\nUAT2: C=0 valid_frac={c0['valid_frac']:+.2f} span_pmass={span0:+.2f} "
|
||||
f"(SHOULD valid_frac=1, span high); incoherent boundary rows={len(edge)} "
|
||||
logger.info(f"\nUAT2: C=0 rep={c0['rep']:+.2f} coherent={c0['coherent']} "
|
||||
f"(SHOULD rep low, coherent True); degenerate boundary rows={len(edge)} "
|
||||
f"at C={[r['C'] for r in edge]} (SHOULD be >=1 -> sweep found a real edge)")
|
||||
|
||||
@@ -36,7 +36,7 @@ fig = plot_sweep(rows, title="joy steer: rubric ans vs C (colored by coherence)"
|
||||
fig.savefig(OUT, dpi=110, bbox_inches="tight")
|
||||
logger.info(f"wrote {OUT}")
|
||||
|
||||
# UAT: the sweep must contain BOTH coherent (valid_frac=1) and incoherent (red-edge)
|
||||
# UAT: the sweep must contain BOTH coherent (low rep) and degenerate (red-edge)
|
||||
# rows, so the plot shows the dose-response AND the breakdown edge.
|
||||
coh = [r for r in rows if r["coherent"]]
|
||||
inc = [r for r in rows if not r["coherent"]]
|
||||
|
||||
@@ -1,45 +0,0 @@
|
||||
"""UAT: does the JSON-object coherence probe actually CATCH incoherence? (Claude)
|
||||
smoke_sweep only reached |C|=0.3 (still coherent). Here we spot-check rubric_score at
|
||||
increasing |C| (one seed, greedy) to find where the object breaks: valid/chk_ok should
|
||||
flip to False and span_pmass should collapse as steering fries the model. If they never
|
||||
do, the probe can't discriminate -> the coherence guard is still blind.
|
||||
|
||||
uv run python scripts/scratch/uat_coherence_break.py
|
||||
"""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||||
import config # noqa: E402
|
||||
import torch # noqa: E402
|
||||
from loguru import logger # noqa: E402
|
||||
from tabulate import tabulate # noqa: E402
|
||||
from transformers import AutoModelForCausalLM, AutoTokenizer # noqa: E402
|
||||
|
||||
from jsteer import Jacobian # noqa: E402
|
||||
from jsteer.demo import rubric_score # noqa: E402
|
||||
|
||||
MODEL = "Qwen/Qwen3.5-4B"
|
||||
tok = AutoTokenizer.from_pretrained(MODEL)
|
||||
model = AutoModelForCausalLM.from_pretrained(MODEL, dtype=torch.bfloat16).to("cuda").eval()
|
||||
jac = Jacobian.from_pretrained(config.LENS_REPO, filename=config.hub_lens_file(MODEL),
|
||||
revision=config.LENS_REVISION)
|
||||
band = jac.steer_band(model)
|
||||
v = jac.word_vector(model, tok, ["happy", "joy"], layers=band)
|
||||
|
||||
RUBRIC = "On a scale of 0 (bleak) to 9 (joyful), how is your outlook on the week?"
|
||||
rows = []
|
||||
for C in (0.0, 0.5, 1.0, 1.5, 2.0, 3.0):
|
||||
with v(model, C=C):
|
||||
e, c = rubric_score(model, tok, RUBRIC, max_new_tokens=384, seed=0)
|
||||
rows.append({"C": C, "ans": e, "span_pmass": c["span_pmass"],
|
||||
"valid": c["valid"], "chk_ok": c["chk_ok"]})
|
||||
logger.info(f"C={C:+.1f} ans={e:+.2f} span_pmass={c['span_pmass']:+.2f} "
|
||||
f"valid={c['valid']} 2+2ok={c['chk_ok']}")
|
||||
|
||||
logger.info("\n" + tabulate(rows, headers="keys", tablefmt="github", floatfmt="+.2f"))
|
||||
broke = [r for r in rows if not (r["valid"] and r["chk_ok"])]
|
||||
span_range = max(r["span_pmass"] for r in rows) - min(r["span_pmass"] for r in rows)
|
||||
logger.info(f"\nUAT: rows where object broke (invalid or 2+2 wrong) = {len(broke)} at "
|
||||
f"C={[r['C'] for r in broke]}; span_pmass range={span_range:+.2f} "
|
||||
f"(SHOULD: >=1 break at high |C| AND span_pmass falls, else probe is blind)")
|
||||
Reference in New Issue
Block a user