mirror of
https://github.com/wassname/jsteer.git
synced 2026-09-10 18:20:20 +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
|
from __future__ import annotations
|
||||||
|
|
||||||
import dataclasses
|
import dataclasses
|
||||||
import json
|
|
||||||
|
|
||||||
import torch
|
import torch
|
||||||
from jlens.vis import _meaningful_token_mask
|
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
|
# 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
|
# moved the target axis. Same mechanism as moral-maps guided.py (let the model think,
|
||||||
# think, then read the logprobs at a JSON answer slot), reduced to a single scalar.
|
# then read the logprobs at a JSON answer slot), reduced to a single scalar. (Claude)
|
||||||
# The object carries a trivial-arithmetic canary ("2+2") and a free-text field so a
|
_ANS_FMT = (' Think it over, then answer with JSON {"ans": N} where N is a single'
|
||||||
# steer-degraded model has ROOM to break the object -- coherence is measured on the
|
' digit from 0 (least) to 9 (most).')
|
||||||
# free-generated object, not on the forced digit slot (which is ~always a digit
|
|
||||||
# because the `{"ans": ` prefix makes one obvious). (Claude)
|
# Coherence = repetition of the think trace, NOT a forced-object gate. Every steer
|
||||||
_ANS_FMT = (' Think it over, then answer with ONE line of JSON and nothing after it:'
|
# breakdown we observed is a REPETITION loop ("happy and happy...", wedding-jewelry,
|
||||||
' {"ans": N, "why": "<=3 words", "2+2": M} where N is a single digit from'
|
# "favorite books..."), so 1 - distinct-3 is the natural, simple coherence signal (and
|
||||||
' 0 (least) to 9 (most) and M is the value of 2+2.')
|
# 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()
|
@torch.no_grad()
|
||||||
def rubric_score(model, tok, rubric: str, *, max_new_tokens: int, seed: int,
|
def rubric_score(model, tok, rubric: str, *, max_new_tokens: int, seed: int,
|
||||||
do_sample: bool = False, temperature: float = 0.7) -> tuple[float, dict]:
|
do_sample: bool = False, temperature: float = 0.7) -> tuple[float, float]:
|
||||||
"""Ask `rubric`, let the model think, force the slot `{"ans": ` for a clean scalar
|
"""Ask `rubric`, let the model think, then force the slot `{"ans": ` and read the
|
||||||
read, then FREE-GENERATE the rest of the JSON object as a coherence probe.
|
logprob-weighted expected digit. Returns (expected, rep) where:
|
||||||
Returns (expected, coh).
|
|
||||||
|
|
||||||
expected = sum_d d * softmax(logit_d over the 10 digit tokens) at the forced slot
|
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).
|
-- a continuous scalar from single-token logprobs (cleaner than parsing a float).
|
||||||
|
rep = 1 - distinct-3 of the think trace -- the coherence signal. Low (~0.05) while
|
||||||
coh = {"valid", "chk_ok", "span_pmass"} measured on the free-generated object:
|
the model reasons fluently, ->1 when steering degenerates it into a repeat loop.
|
||||||
valid -- the object parses as JSON (a steer-fried model fails to close it),
|
We measure coherence on the long think trace (which degenerates under steering),
|
||||||
chk_ok -- its "2+2" field == 4 (trivial-arithmetic canary),
|
not on the short forced answer (which stays scorable well past the breakdown)."""
|
||||||
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."""
|
|
||||||
prompt = chat_input(tok, rubric + _ANS_FMT)
|
prompt = chat_input(tok, rubric + _ANS_FMT)
|
||||||
enc = tok(prompt, return_tensors="pt").to(model.device)
|
enc = tok(prompt, return_tensors="pt").to(model.device)
|
||||||
torch.manual_seed(seed)
|
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]
|
ids = torch.tensor([tok(str(d), add_special_tokens=False).input_ids[0]
|
||||||
for d in range(10)], device=logits.device)
|
for d in range(10)], device=logits.device)
|
||||||
expected = float((logits[ids].softmax(0) * torch.arange(10., device=ids.device)).sum())
|
expected = float((logits[ids].softmax(0) * torch.arange(10., device=ids.device)).sum())
|
||||||
|
return expected, _rep_frac(think)
|
||||||
# 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}
|
|
||||||
|
|
||||||
|
|
||||||
@torch.no_grad()
|
@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,
|
max_steps: int = 15, n_samples: int = 3,
|
||||||
temperature: float = 0.7, max_new_tokens: int = 512) -> list[dict]:
|
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
|
"""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
|
direction the step AFTER the think trace degenerates (mean rep >= REP_COHERENT_MAX,
|
||||||
majority of seeds fail JSON-parse or the "2+2" canary). Maps the coherent dose-
|
i.e. it collapses into a repeat loop). Maps the coherent dose-response of the steered
|
||||||
response of the steered axis without hand-picking Cs. Returns rows sorted by C:
|
axis without hand-picking Cs. Returns rows sorted by C:
|
||||||
{"C","ans","ans_std","span_pmass","valid_frac","coherent"}. Each C is averaged over
|
{"C","ans","ans_std","rep","coherent"}. Each C is averaged over `n_samples` think
|
||||||
`n_samples` think traces (seeds 0..n-1) to tame single-sample answer noise -- a
|
traces (seeds 0..n-1) to tame single-sample noise -- a lightweight stand-in for
|
||||||
lightweight stand-in for guided.py's Bayesian model averaging; ans_std is the spread.
|
guided.py's Bayesian model averaging; ans_std is the spread. rep = 1 - distinct-3 of
|
||||||
Coherence = the model still free-generates a well-formed object AND gets 2+2 right;
|
the think trace catches the actual failure mode (repetition), unlike a short forced
|
||||||
span_pmass grades its confidence. This breaks well before free-form fluency does at
|
object that stays scorable past the breakdown."""
|
||||||
large |C|, so read the qualitative show_steer for the long-generation frailty."""
|
|
||||||
def score(C):
|
def score(C):
|
||||||
with vec(model, C=C):
|
with vec(model, C=C):
|
||||||
pairs = [rubric_score(model, tok, rubric, max_new_tokens=max_new_tokens, seed=s,
|
pairs = [rubric_score(model, tok, rubric, max_new_tokens=max_new_tokens, seed=s,
|
||||||
do_sample=n_samples > 1, temperature=temperature)
|
do_sample=n_samples > 1, temperature=temperature)
|
||||||
for s in range(n_samples)]
|
for s in range(n_samples)]
|
||||||
anss = torch.tensor([e for e, _ in pairs])
|
anss = torch.tensor([e for e, _ in pairs])
|
||||||
span = float(torch.tensor([c["span_pmass"] for _, c in pairs]).mean())
|
rep = float(torch.tensor([r for _, r in pairs]).mean())
|
||||||
valid_frac = sum(c["valid"] and c["chk_ok"] for _, c in pairs) / len(pairs)
|
|
||||||
return {"C": round(float(C), 3), "ans": float(anss.mean()),
|
return {"C": round(float(C), 3), "ans": float(anss.mean()),
|
||||||
"ans_std": float(anss.std(unbiased=False)), "span_pmass": span,
|
"ans_std": float(anss.std(unbiased=False)), "rep": rep,
|
||||||
"valid_frac": valid_frac, "coherent": valid_frac >= 0.5}
|
"coherent": rep < REP_COHERENT_MAX}
|
||||||
rows = [score(0.0)]
|
rows = [score(0.0)]
|
||||||
for d in (step, -step): # outward each way; keep the 1st incoherent point
|
for d in (step, -step): # outward each way; keep the 1st incoherent point
|
||||||
C = d
|
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"):
|
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
|
"""ans vs C, points colored by think-trace repetition (rep = 1 - distinct-3);
|
||||||
a well-formed {"ans",...,"2+2"} object with 2+2==4); points below majority also get
|
degenerate points (rep >= REP_COHERENT_MAX) also get a red edge. Low rep = fluent
|
||||||
a red edge. Coherence is NEAR-BINARY (flat while the model holds, a cliff when it
|
reasoning (bright), high rep = the steer has collapsed the trace into a repeat loop
|
||||||
breaks), so the color reads as a gate and the ans curve carries the dose-response.
|
(dark + red edge). The ans curve carries the dose-response; rep marks where to stop
|
||||||
We deliberately do NOT color by span_pmass: a steer-fried model collapses into a
|
trusting it. Colorbar is inverted (viridis_r) so brighter = more coherent."""
|
||||||
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."""
|
|
||||||
import matplotlib.pyplot as plt
|
import matplotlib.pyplot as plt
|
||||||
Cs = [r["C"] for r in rows]
|
Cs = [r["C"] for r in rows]
|
||||||
ans = [r["ans"] 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))
|
fig, ax = plt.subplots(figsize=(5, 3))
|
||||||
ax.plot(Cs, ans, "-", color="0.8", lw=1, zorder=1)
|
ax.plot(Cs, ans, "-", color="0.8", lw=1, zorder=1)
|
||||||
if all("ans_std" in r for r in rows):
|
if all("ans_std" in r for r in rows):
|
||||||
ax.errorbar(Cs, ans, yerr=[r["ans_std"] for r in rows], fmt="none",
|
ax.errorbar(Cs, ans, yerr=[r["ans_std"] for r in rows], fmt="none",
|
||||||
ecolor="0.6", capsize=2, lw=1, zorder=1)
|
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],
|
zorder=2, edgecolor=["0.2" if r["coherent"] else "red" for r in rows],
|
||||||
linewidth=1.2)
|
linewidth=1.2)
|
||||||
ax.axvline(0, color="0.85", lw=0.8, zorder=0)
|
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_ylabel("rubric ans (0-9)")
|
||||||
ax.set_ylim(-0.3, 9.3)
|
ax.set_ylim(-0.3, 9.3)
|
||||||
ax.set_title(title)
|
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()
|
fig.tight_layout()
|
||||||
return fig
|
return fig
|
||||||
|
|
||||||
@@ -280,8 +263,8 @@ def show_steer(jac: Jacobian, model, tok, vec, user_msg: str, *,
|
|||||||
_cthulhu_say(readout), gen]
|
_cthulhu_say(readout), gen]
|
||||||
if ans is not None:
|
if ans is not None:
|
||||||
# SHOULD rise with +C, fall with -C; flat => steer not moving this axis.
|
# 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.
|
# rep>=0.35 => the think trace degenerated into a loop, distrust the number.
|
||||||
e, c = ans
|
e, rep = ans
|
||||||
block.append(f" rubric ans≈{e:.2f}/9 (json={c['valid']} 2+2ok={c['chk_ok']}"
|
block.append(f" rubric ans≈{e:.2f}/9 (rep={rep:.2f}"
|
||||||
f" conf={c['span_pmass']:.2f})")
|
f"{' DEGENERATE' if rep >= REP_COHERENT_MAX else ''})")
|
||||||
logger.info("\n".join(block) + "\n")
|
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)} "
|
logger.info(f"\nUAT1: rows with ans_std>0 = {len(nonzero)}/{len(rows)} "
|
||||||
f"(SHOULD be >0 -> sampling+BMA active)")
|
f"(SHOULD be >0 -> sampling+BMA active)")
|
||||||
|
|
||||||
# UAT 2: the JSON coherence probe discriminates. At C=0 the model emits a valid object
|
# UAT 2: the repetition coherence probe discriminates. At C=0 the think trace is fluent
|
||||||
# (valid_frac=1, high span_pmass); walking |C| out, span_pmass falls and the sweep stops
|
# (rep low, coherent True); walking |C| out, rep rises past REP_COHERENT_MAX and the sweep
|
||||||
# at an incoherent boundary. If C=0 is already incoherent OR span_pmass never falls, the
|
# stops at a degenerate boundary. If C=0 is already incoherent OR rep never rises, the
|
||||||
# probe isn't measuring coherence -> broken.
|
# probe isn't measuring coherence -> broken.
|
||||||
c0 = next(r for r in rows if r["C"] == 0.0)
|
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"]]
|
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} "
|
logger.info(f"\nUAT2: C=0 rep={c0['rep']:+.2f} coherent={c0['coherent']} "
|
||||||
f"(SHOULD valid_frac=1, span high); incoherent boundary rows={len(edge)} "
|
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)")
|
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")
|
fig.savefig(OUT, dpi=110, bbox_inches="tight")
|
||||||
logger.info(f"wrote {OUT}")
|
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.
|
# rows, so the plot shows the dose-response AND the breakdown edge.
|
||||||
coh = [r for r in rows if r["coherent"]]
|
coh = [r for r in rows if r["coherent"]]
|
||||||
inc = [r for r in rows if not 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