mirror of
https://github.com/wassname/persona-steering-template-library.git
synced 2026-09-12 12:40:39 +08:00
refactor: sort scripts/ by who runs it
Top level is now only what the runbook or a re-run touches: validate_persona_axes, bounded_thinking_judge, template_catalog, export_selections, parse_stage_a, run_axis, export_steering_selection. Corpus ingestion and publishing moved to scripts/corpus/, plotting and stats to scripts/report/. Moved files needed parents[1] -> parents[2]; the two corpus scripts that import template_catalog use the sys.path shim bounded_thinking_judge_liveproof already used. Also completes the export_steering_selection rename: an earlier git reset had dropped the staged deletion, leaving both filenames tracked. Co-Authored-By: Claudypoo <288921227+claudypoo@users.noreply.github.com>
This commit is contained in:
wassname
@@ -0,0 +1,89 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import math
|
||||
from pathlib import Path
|
||||
import statistics
|
||||
from typing import Any
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
STATS = ROOT / "data/results/stats"
|
||||
MODEL_MATRIX = ROOT / "data/results/model_matrix"
|
||||
DOCS_MODEL_MATRIX = ROOT / "docs/results/model_matrix"
|
||||
|
||||
NORMAL_TEMPLATE_PAIR_STATS = STATS / "v2_pilot_seed24_template_pair_stats.jsonl"
|
||||
ENGINEERED_TEMPLATE_PAIR_STATS = STATS / "engineered_baseline_seed24_template_pair_stats.jsonl"
|
||||
CONTROL_TEMPLATE_PAIR_STATS = STATS / "control_baseline_seed24_template_pair_stats.jsonl"
|
||||
|
||||
REFUSAL_MODEL_PAIR_STATS = [
|
||||
MODEL_MATRIX / "stats/refusal_probe_seed24_n1_google_gemma-2-27b-it_template_pair_stats.jsonl",
|
||||
MODEL_MATRIX / "stats/refusal_probe_seed24_n1_google_gemma-3-4b-it_template_pair_stats.jsonl",
|
||||
MODEL_MATRIX / "stats/refusal_probe_seed24_n1_qwen_qwen3.6-flash_template_pair_stats.jsonl",
|
||||
MODEL_MATRIX / "stats/refusal_probe_seed24_n1_ibm-granite_granite-4.1-8b_template_pair_stats.jsonl",
|
||||
]
|
||||
REFUSAL_MODEL_PREFIX = MODEL_MATRIX / "refusal_probe_seed24_n1"
|
||||
REFUSAL_MODEL_MARKDOWN = DOCS_MODEL_MATRIX / "refusal_probe_seed24_n1_model_matrix_summary.md"
|
||||
|
||||
ANTHROPIC_IF2_COMMENT = "<!-- instruction following eval, Anthropic/if-2 -->"
|
||||
ANTHROPIC_IF2_LABEL = "Anthropic/if-2 instruction-following eval:"
|
||||
|
||||
|
||||
def read_jsonl(path: Path) -> list[dict[str, Any]]:
|
||||
return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
|
||||
|
||||
|
||||
def display_template_text(text: str) -> str:
|
||||
return text.replace(ANTHROPIC_IF2_COMMENT, ANTHROPIC_IF2_LABEL)
|
||||
|
||||
|
||||
def clamp01(x: float) -> float:
|
||||
return max(0.0, min(1.0, x))
|
||||
|
||||
|
||||
def mean(xs: list[float]) -> float:
|
||||
return sum(xs) / len(xs)
|
||||
|
||||
|
||||
def std(xs: list[float]) -> float:
|
||||
if len(xs) == 1:
|
||||
return 0.0
|
||||
return statistics.stdev(xs)
|
||||
|
||||
|
||||
def score(row: dict[str, Any]) -> float:
|
||||
on_axis = clamp01(float(row["mean_axis_delta"]) / 8.0)
|
||||
off_axis = clamp01((float(row["mean_off_axis_problem"]) - 1.0) / 6.0)
|
||||
return 100.0 * on_axis * (1.0 - off_axis)
|
||||
|
||||
|
||||
def score_t(scores: list[float]) -> float:
|
||||
sem = std(scores) / math.sqrt(len(scores))
|
||||
mean_score = mean(scores)
|
||||
if sem == 0.0:
|
||||
return 0.0 if mean_score == 0.0 else 1_000_000.0
|
||||
return mean_score / sem
|
||||
|
||||
|
||||
def mean_template_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
grouped: dict[str, list[dict[str, Any]]] = {}
|
||||
for row in rows:
|
||||
grouped.setdefault(row["template"], []).append({**row, "score": score(row)})
|
||||
|
||||
out = []
|
||||
for template, rs in grouped.items():
|
||||
scores = [float(row["score"]) for row in rs]
|
||||
out.append({
|
||||
"template": template,
|
||||
"score_t": round(score_t(scores), 2),
|
||||
"score": round(mean(scores), 1),
|
||||
"score_mean": round(mean(scores), 2),
|
||||
"on_axis": clamp01(mean([float(row["mean_axis_delta"]) for row in rs]) / 8.0),
|
||||
"off_axis": clamp01(
|
||||
(mean([float(row["mean_off_axis_problem"]) for row in rs]) - 1.0) / 6.0),
|
||||
"axis_delta": round(mean([float(row["mean_axis_delta"]) for row in rs]), 2),
|
||||
"off_axis_problem": round(mean([float(row["mean_off_axis_problem"]) for row in rs]), 2),
|
||||
"judge_std": round(mean([float(row["mean_axis_delta_judge_std"]) for row in rs]), 2),
|
||||
"n_cells": len(rs),
|
||||
})
|
||||
return sorted(out, key=lambda row: row["score_t"], reverse=True)
|
||||
@@ -0,0 +1,253 @@
|
||||
"""Export upload-friendly stats from persona template validation artifacts.
|
||||
|
||||
Input is the JSON written by scripts/validate_persona_axes.py.
|
||||
Outputs:
|
||||
<out-prefix>_template_stats.jsonl one row per template
|
||||
<out-prefix>_template_pair_stats.jsonl one row per template × persona pair
|
||||
<out-prefix>_examples.jsonl one row per generated pair
|
||||
<out-prefix>_template_stats.csv compact table for spreadsheets
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
from collections import defaultdict
|
||||
from pathlib import Path
|
||||
from statistics import mean
|
||||
|
||||
|
||||
def _rows(paths: list[Path]) -> list[dict]:
|
||||
rows = []
|
||||
for path in paths:
|
||||
data = json.loads(path.read_text())
|
||||
meta = {
|
||||
"artifact": str(path),
|
||||
"generator_model": data.get("generator_model"),
|
||||
"judge_model": data.get("judge_model"),
|
||||
"axis_judge_models": data["axis_judge_models"],
|
||||
"style_judge_model": data["style_judge_model"],
|
||||
"gen_temperature": data.get("gen_temperature"),
|
||||
"seed": data.get("seed"),
|
||||
"family": data.get("family"),
|
||||
}
|
||||
for rec in data.get("results", []):
|
||||
row = {**meta, **rec}
|
||||
rows.append(row)
|
||||
return rows
|
||||
|
||||
|
||||
def _m(vals: list[float]) -> float | None:
|
||||
return round(mean(vals), 4) if vals else None
|
||||
|
||||
|
||||
def _aggregate(rows: list[dict], keys: tuple[str, ...]) -> list[dict]:
|
||||
groups: dict[tuple, list[dict]] = defaultdict(list)
|
||||
for r in rows:
|
||||
if "error" not in r:
|
||||
groups[tuple(r[k] for k in keys)].append(r)
|
||||
out = []
|
||||
for key, rs in groups.items():
|
||||
n = len(rs)
|
||||
strict = [bool(r.get("strict_pass")) for r in rs]
|
||||
style_dims = sorted(rs[0].get("style_deltas_pos_minus_neg", {}))
|
||||
off_axis_dims = sorted(rs[0].get("off_axis_category_likerts", {}))
|
||||
row = {k: v for k, v in zip(keys, key)}
|
||||
row.update({
|
||||
"n": n,
|
||||
"strict_pass_rate": round(sum(strict) / n, 4),
|
||||
"n_strict_pass": sum(strict),
|
||||
"mean_axis_delta": _m([float(r["axis_delta"]) for r in rs]),
|
||||
"mean_axis_delta_judge_mean": _m([
|
||||
float(r["axis_delta_judge_mean"]) for r in rs
|
||||
]),
|
||||
"mean_axis_delta_judge_std": _m([
|
||||
float(r["axis_delta_judge_std"]) for r in rs
|
||||
]),
|
||||
"mean_axis_judge_abs_disagreement": _m([
|
||||
float(r["axis_judge_mean_abs_disagreement"]) for r in rs
|
||||
]),
|
||||
"mean_delta_pos_vs_base": _m([float(r["delta_pos_vs_base"]) for r in rs]),
|
||||
"mean_delta_base_vs_neg": _m([float(r["delta_base_vs_neg"]) for r in rs]),
|
||||
"mean_min_side_delta": _m([float(r["min_side_delta"]) for r in rs]),
|
||||
"mean_off_axis_problem": _m([
|
||||
float(r["confound_judgment"]["off_axis_problem_likert"]) for r in rs
|
||||
]),
|
||||
"mean_max_off_axis_category_likert": _m([
|
||||
float(r.get("max_off_axis_category_likert", 7)) for r in rs
|
||||
]),
|
||||
"usable_rate": round(
|
||||
sum(bool(r["confound_judgment"]["usable_for_training"]) for r in rs) / n, 4),
|
||||
"mean_max_style_abs_delta": _m([float(r["max_style_abs_delta"]) for r in rs]),
|
||||
"mean_abs_word_delta_frac": _m([abs(float(r["word_delta_frac"])) for r in rs]),
|
||||
"mean_response_token_jaccard": _m([
|
||||
float(r.get("response_token_jaccard", 0.0)) for r in rs
|
||||
]),
|
||||
"mean_pos_repeated_token_frac": _m([
|
||||
float(r.get("pos_repeated_token_frac", 0.0)) for r in rs
|
||||
]),
|
||||
"mean_neg_repeated_token_frac": _m([
|
||||
float(r.get("neg_repeated_token_frac", 0.0)) for r in rs
|
||||
]),
|
||||
"persona_echo_rate": round(sum(bool(r["persona_echo"]) for r in rs) / n, 4),
|
||||
"judge_persona_echo_rate": round(
|
||||
sum(bool(r.get("judge_persona_echo")) for r in rs) / n, 4),
|
||||
"refusal_or_ai_break_rate": round(
|
||||
sum(bool(r["refusal_or_ai_break"]) for r in rs) / n, 4),
|
||||
"judge_refusal_or_ai_break_rate": round(
|
||||
sum(bool(r.get("judge_refusal_or_ai_break")) for r in rs) / n, 4),
|
||||
"strict_pass_persona_pairs": sorted({
|
||||
r["axis"]["id"] for r in rs if r.get("strict_pass")
|
||||
}),
|
||||
"common_spurious_axes": sorted({
|
||||
r["confound_judgment"].get("likely_spurious_axis", "")
|
||||
for r in rs
|
||||
if r["confound_judgment"].get("likely_spurious_axis")
|
||||
}),
|
||||
})
|
||||
for dim in style_dims:
|
||||
row[f"mean_style_delta_{dim}_pos_minus_neg"] = _m([
|
||||
float(r["style_deltas_pos_minus_neg"][dim]) for r in rs
|
||||
])
|
||||
for dim in off_axis_dims:
|
||||
row[f"mean_off_axis_{dim}"] = _m([
|
||||
float(r["off_axis_category_likerts"][dim]) for r in rs
|
||||
])
|
||||
row["recommended"] = (
|
||||
n >= 4
|
||||
and row["strict_pass_rate"] >= 0.5
|
||||
and row["mean_axis_delta"] >= 3
|
||||
and row["mean_off_axis_problem"] <= 2
|
||||
and row["mean_max_style_abs_delta"] <= 2
|
||||
and row["persona_echo_rate"] == 0
|
||||
and row["refusal_or_ai_break_rate"] == 0
|
||||
)
|
||||
out.append(row)
|
||||
out.sort(key=lambda r: (
|
||||
r["recommended"],
|
||||
r["strict_pass_rate"],
|
||||
r["mean_min_side_delta"],
|
||||
r["mean_axis_delta"],
|
||||
-r["mean_off_axis_problem"],
|
||||
-r["mean_max_style_abs_delta"],
|
||||
), reverse=True)
|
||||
return out
|
||||
|
||||
|
||||
def _example_rows(rows: list[dict]) -> list[dict]:
|
||||
out = []
|
||||
for r in rows:
|
||||
axis = r.get("axis", {})
|
||||
rec = {
|
||||
"artifact": r.get("artifact"),
|
||||
"eval_id": r.get("eval_id"),
|
||||
"template": r.get("template"),
|
||||
"persona_pair": axis.get("id"),
|
||||
"scenario_id": r.get("scenario_id"),
|
||||
"pos_persona": axis.get("pos_descriptor"),
|
||||
"neg_persona": axis.get("neg_descriptor"),
|
||||
"row": r.get("row"),
|
||||
"source": r.get("source"),
|
||||
"config": r.get("config"),
|
||||
"prompt": r.get("prompt"),
|
||||
"pos_generation_prompt": r.get("pos_generation_prompt"),
|
||||
"neg_generation_prompt": r.get("neg_generation_prompt"),
|
||||
"error": r.get("error"),
|
||||
}
|
||||
if "error" not in r:
|
||||
rec.update({
|
||||
"strict_pass": r.get("strict_pass"),
|
||||
"axis_judge_models": r.get("axis_judge_models"),
|
||||
"axis_judgments": r.get("axis_judgments"),
|
||||
"axis_delta": r.get("axis_delta"),
|
||||
"axis_delta_judge_mean": r.get("axis_delta_judge_mean"),
|
||||
"axis_delta_judge_std": r.get("axis_delta_judge_std"),
|
||||
"axis_judge_mean_abs_disagreement": r.get("axis_judge_mean_abs_disagreement"),
|
||||
"delta_pos_vs_base": r.get("delta_pos_vs_base"),
|
||||
"delta_base_vs_neg": r.get("delta_base_vs_neg"),
|
||||
"min_side_delta": r.get("min_side_delta"),
|
||||
"off_axis_problem": r["confound_judgment"].get("off_axis_problem_likert"),
|
||||
"max_off_axis_category_likert": r.get("max_off_axis_category_likert"),
|
||||
"usable_for_training": r["confound_judgment"].get("usable_for_training"),
|
||||
"likely_spurious_axis": r["confound_judgment"].get("likely_spurious_axis"),
|
||||
"max_style_abs_delta": r.get("max_style_abs_delta"),
|
||||
"word_delta_frac": r.get("word_delta_frac"),
|
||||
"response_token_jaccard": r.get("response_token_jaccard"),
|
||||
"pos_repeated_token_frac": r.get("pos_repeated_token_frac"),
|
||||
"neg_repeated_token_frac": r.get("neg_repeated_token_frac"),
|
||||
"persona_echo": r.get("persona_echo"),
|
||||
"judge_persona_echo": r.get("judge_persona_echo"),
|
||||
"pos_persona_echo_hits": r.get("pos_persona_echo_hits"),
|
||||
"neg_persona_echo_hits": r.get("neg_persona_echo_hits"),
|
||||
"pos_persona_overlap_tokens": r.get("pos_persona_overlap_tokens"),
|
||||
"neg_persona_overlap_tokens": r.get("neg_persona_overlap_tokens"),
|
||||
"refusal_or_ai_break": r.get("refusal_or_ai_break"),
|
||||
"judge_refusal_or_ai_break": r.get("judge_refusal_or_ai_break"),
|
||||
"pos_refusal_phrase_hits": r.get("pos_refusal_phrase_hits"),
|
||||
"neg_refusal_phrase_hits": r.get("neg_refusal_phrase_hits"),
|
||||
"pos_response": r.get("pos_response"),
|
||||
"neg_response": r.get("neg_response"),
|
||||
"base_response": r.get("base_response"),
|
||||
})
|
||||
for dim, val in r.get("style_deltas_pos_minus_neg", {}).items():
|
||||
rec[f"style_delta_{dim}_pos_minus_neg"] = val
|
||||
for dim, val in r.get("off_axis_category_likerts", {}).items():
|
||||
rec[f"off_axis_{dim}"] = val
|
||||
out.append(rec)
|
||||
return out
|
||||
|
||||
|
||||
def _write_jsonl(path: Path, rows: list[dict]) -> None:
|
||||
with path.open("w") as fh:
|
||||
for row in rows:
|
||||
fh.write(json.dumps(row, ensure_ascii=False) + "\n")
|
||||
|
||||
|
||||
def _write_csv(path: Path, rows: list[dict]) -> None:
|
||||
if not rows:
|
||||
path.write_text("")
|
||||
return
|
||||
fieldnames = sorted({k for row in rows for k in row})
|
||||
with path.open("w", newline="") as fh:
|
||||
writer = csv.DictWriter(fh, fieldnames=fieldnames, lineterminator="\n")
|
||||
writer.writeheader()
|
||||
for row in rows:
|
||||
writer.writerow({
|
||||
k: json.dumps(v, ensure_ascii=False) if isinstance(v, (list, dict)) else v
|
||||
for k, v in row.items()
|
||||
})
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("artifacts", nargs="+", type=Path)
|
||||
ap.add_argument("--out-prefix", default="data/results/stats/persona_template_library")
|
||||
args = ap.parse_args()
|
||||
|
||||
rows = _rows(args.artifacts)
|
||||
for r in rows:
|
||||
if "axis" in r and "error" not in r:
|
||||
r["persona_pair"] = r["axis"]["id"]
|
||||
template_stats = _aggregate(rows, ("template",))
|
||||
pair_stats = _aggregate(rows, ("template", "persona_pair"))
|
||||
examples = _example_rows(rows)
|
||||
|
||||
prefix = Path(args.out_prefix)
|
||||
prefix.parent.mkdir(parents=True, exist_ok=True)
|
||||
_write_jsonl(prefix.with_name(prefix.name + "_template_stats.jsonl"), template_stats)
|
||||
_write_jsonl(prefix.with_name(prefix.name + "_template_pair_stats.jsonl"), pair_stats)
|
||||
_write_jsonl(prefix.with_name(prefix.name + "_examples.jsonl"), examples)
|
||||
_write_csv(prefix.with_name(prefix.name + "_template_stats.csv"), template_stats)
|
||||
_write_csv(prefix.with_name(prefix.name + "_template_pair_stats.csv"), pair_stats)
|
||||
print(f"examples={len(examples)} template_stats={len(template_stats)} pair_stats={len(pair_stats)}")
|
||||
print("top templates:")
|
||||
for row in template_stats[:10]:
|
||||
print(
|
||||
f"{row['strict_pass_rate']:.2f} pass axis={row['mean_axis_delta']:.2f} "
|
||||
f"off={row['mean_off_axis_problem']:.2f} style={row['mean_max_style_abs_delta']:.2f} "
|
||||
f"{row['template']}"
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,14 @@
|
||||
"""Write the previous-results Plotly figure as PNG and SVG."""
|
||||
from __future__ import annotations
|
||||
|
||||
import readme_plot
|
||||
|
||||
|
||||
def main() -> None:
|
||||
readme_plot.write_main_plot_assets()
|
||||
print(readme_plot.MAIN_PNG)
|
||||
print(readme_plot.MAIN_SVG)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,98 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import html
|
||||
from pathlib import Path
|
||||
import textwrap
|
||||
from typing import Any
|
||||
|
||||
import plotly.graph_objects as go
|
||||
|
||||
import docs_results
|
||||
|
||||
MAIN_PNG = docs_results.ROOT / "docs/results/assets/on_off_axis.png"
|
||||
MAIN_SVG = docs_results.ROOT / "docs/results/assets/on_off_axis.svg"
|
||||
|
||||
|
||||
def _wrap_hover(text: str, width: int = 62) -> str:
|
||||
text = docs_results.display_template_text(text)
|
||||
escaped = html.escape(" ".join(text.split()))
|
||||
return "<br>".join(
|
||||
textwrap.wrap(escaped, width=width, break_long_words=True, break_on_hyphens=False))
|
||||
|
||||
|
||||
def main_plot_rows(path: Path = docs_results.NORMAL_TEMPLATE_PAIR_STATS) -> list[dict[str, Any]]:
|
||||
return docs_results.mean_template_rows(docs_results.read_jsonl(path))
|
||||
|
||||
|
||||
def template_scatter(rows: list[dict[str, Any]] | None = None, width: int | None = None) -> go.Figure:
|
||||
rows = main_plot_rows() if rows is None else rows
|
||||
top_rank = {row["template"]: i for i, row in enumerate(rows[:10], start=1)}
|
||||
text = [str(top_rank[row["template"]]) if row["template"] in top_rank else "" for row in rows]
|
||||
hover = [
|
||||
"<br>".join([
|
||||
f"<b>{_wrap_hover(row['template'])}</b>",
|
||||
f"rank: {i}",
|
||||
f"score t: {row['score_t']:.2f}",
|
||||
f"score mean: {row['score_mean']:.2f}",
|
||||
f"axis delta: {row['axis_delta']:.2f}",
|
||||
f"off-axis problem: {row['off_axis_problem']:.2f}",
|
||||
f"judge std: {row['judge_std']:.2f}",
|
||||
f"cells: {row['n_cells']}",
|
||||
])
|
||||
for i, row in enumerate(rows, start=1)
|
||||
]
|
||||
fig = go.Figure(
|
||||
data=go.Scatter(
|
||||
x=[row["on_axis"] for row in rows],
|
||||
y=[row["off_axis"] for row in rows],
|
||||
mode="markers+text",
|
||||
text=text,
|
||||
textposition="middle center",
|
||||
textfont={"size": 9, "color": "white"},
|
||||
customdata=hover,
|
||||
hovertemplate="%{customdata}<extra></extra>",
|
||||
marker={
|
||||
"size": 10,
|
||||
"color": [row["score_t"] for row in rows],
|
||||
"colorscale": "Cividis",
|
||||
"showscale": True,
|
||||
"colorbar": {"title": "score t"},
|
||||
"line": {"width": 0.5, "color": "white"},
|
||||
"opacity": 0.9,
|
||||
},
|
||||
)
|
||||
)
|
||||
fig.update_layout(
|
||||
autosize=True,
|
||||
width=width,
|
||||
height=620,
|
||||
template="plotly_white",
|
||||
margin={"l": 68, "r": 24, "t": 28, "b": 66},
|
||||
xaxis={
|
||||
"title": "on-axis movement, higher is better",
|
||||
"range": [-0.02, 1.02],
|
||||
"gridcolor": "rgba(0,0,0,0.08)",
|
||||
},
|
||||
yaxis={
|
||||
"title": "off-axis confounding, lower is better",
|
||||
"range": [-0.02, 1.02],
|
||||
"gridcolor": "rgba(0,0,0,0.08)",
|
||||
},
|
||||
annotations=[{
|
||||
"text": "normal pilot scenarios; one point per measured template",
|
||||
"xref": "paper",
|
||||
"yref": "paper",
|
||||
"x": 1.0,
|
||||
"y": -0.13,
|
||||
"showarrow": False,
|
||||
"font": {"size": 11, "color": "rgba(0,0,0,0.62)"},
|
||||
}],
|
||||
)
|
||||
return fig
|
||||
|
||||
|
||||
def write_main_plot_assets() -> None:
|
||||
fig = template_scatter(width=960)
|
||||
MAIN_PNG.parent.mkdir(parents=True, exist_ok=True)
|
||||
fig.write_image(MAIN_PNG, width=960, height=620, scale=2)
|
||||
fig.write_image(MAIN_SVG, width=960, height=620)
|
||||
@@ -0,0 +1,216 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
import math
|
||||
from pathlib import Path
|
||||
import statistics
|
||||
from typing import Any
|
||||
|
||||
from tabulate import tabulate
|
||||
|
||||
import docs_results
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
DEFAULT_PAIR_STATS = docs_results.REFUSAL_MODEL_PAIR_STATS
|
||||
DEFAULT_OUT_PREFIX = docs_results.REFUSAL_MODEL_PREFIX
|
||||
|
||||
|
||||
def _read_jsonl(path: Path) -> list[dict[str, Any]]:
|
||||
return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]
|
||||
|
||||
|
||||
def _model_name(path: Path) -> str:
|
||||
name = path.name
|
||||
name = name.removeprefix("refusal_probe_seed24_n1_")
|
||||
name = name.removesuffix("_template_pair_stats.jsonl")
|
||||
return name
|
||||
|
||||
|
||||
def _clamp01(x: float) -> float:
|
||||
return max(0.0, min(1.0, x))
|
||||
|
||||
|
||||
def _score(row: dict[str, Any]) -> float:
|
||||
on_axis = _clamp01(float(row["mean_axis_delta"]) / 8.0)
|
||||
off_axis = _clamp01((float(row["mean_off_axis_problem"]) - 1.0) / 6.0)
|
||||
return 100.0 * on_axis * (1.0 - off_axis)
|
||||
|
||||
|
||||
def _mean(xs: list[float]) -> float:
|
||||
return sum(xs) / len(xs)
|
||||
|
||||
|
||||
def _std(xs: list[float]) -> float:
|
||||
if len(xs) == 1:
|
||||
return 0.0
|
||||
return statistics.stdev(xs)
|
||||
|
||||
|
||||
def _p25(xs: list[float]) -> float:
|
||||
return statistics.quantiles(xs, n=4, method="inclusive")[0]
|
||||
|
||||
|
||||
def _sem(xs: list[float]) -> float:
|
||||
return _std(xs) / math.sqrt(len(xs))
|
||||
|
||||
|
||||
def _t_stat(mean: float, sem: float) -> float:
|
||||
if sem == 0.0:
|
||||
return 0.0 if mean == 0.0 else 1_000_000.0
|
||||
return mean / sem
|
||||
|
||||
|
||||
def _round(x: float, digits: int = 3) -> float:
|
||||
if math.isnan(x):
|
||||
raise ValueError("nan in model matrix summary")
|
||||
return round(x, digits)
|
||||
|
||||
|
||||
def _write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text("".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows))
|
||||
|
||||
|
||||
def _write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with path.open("w", newline="") as f:
|
||||
writer = csv.DictWriter(f, fieldnames=list(rows[0]))
|
||||
writer.writeheader()
|
||||
writer.writerows(rows)
|
||||
|
||||
|
||||
def _template_mean_rows(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
groups: dict[tuple[str, str], list[dict[str, Any]]] = {}
|
||||
for row in rows:
|
||||
groups.setdefault((row["model"], row["template"]), []).append(row)
|
||||
|
||||
out = []
|
||||
for (model, template), rs in groups.items():
|
||||
out.append({
|
||||
"model": model,
|
||||
"template": template,
|
||||
"score": _mean([row["score"] for row in rs]),
|
||||
"strict_pass_rate": _mean([float(row["strict_pass_rate"]) for row in rs]),
|
||||
"mean_axis_delta": _mean([float(row["mean_axis_delta"]) for row in rs]),
|
||||
"mean_off_axis_problem": _mean([float(row["mean_off_axis_problem"]) for row in rs]),
|
||||
"mean_axis_delta_judge_std": _mean([float(row["mean_axis_delta_judge_std"]) for row in rs]),
|
||||
"mean_max_style_abs_delta": _mean([float(row["mean_max_style_abs_delta"]) for row in rs]),
|
||||
"persona_echo_rate": _mean([float(row["persona_echo_rate"]) for row in rs]),
|
||||
"refusal_or_ai_break_rate": _mean([float(row["refusal_or_ai_break_rate"]) for row in rs]),
|
||||
"n_axes": len(rs),
|
||||
})
|
||||
return out
|
||||
|
||||
|
||||
def _summarize(rows: list[dict[str, Any]], group_cols: list[str]) -> list[dict[str, Any]]:
|
||||
groups: dict[tuple[Any, ...], list[dict[str, Any]]] = {}
|
||||
for row in rows:
|
||||
groups.setdefault(tuple(row[col] for col in group_cols), []).append(row)
|
||||
|
||||
out = []
|
||||
for key, rs in groups.items():
|
||||
models = sorted({row["model"] for row in rs})
|
||||
base = dict(zip(group_cols, key, strict=True))
|
||||
model_count = len(models)
|
||||
scores = [float(row["score"]) for row in rs]
|
||||
score_mean = _mean(scores)
|
||||
score_sem = _sem(scores)
|
||||
out.append({
|
||||
"model_count": model_count,
|
||||
"score_t": _round(_t_stat(score_mean, score_sem), 2),
|
||||
"score_p25": _round(_p25(scores), 2),
|
||||
"score_mean": _round(score_mean, 2),
|
||||
"score_std": _round(_std(scores), 2),
|
||||
"strict_pass_rate_mean": _round(_mean([float(row["strict_pass_rate"]) for row in rs]), 3),
|
||||
"strict_pass_rate_std": _round(_std([float(row["strict_pass_rate"]) for row in rs]), 3),
|
||||
"axis_delta_mean": _round(_mean([float(row["mean_axis_delta"]) for row in rs]), 3),
|
||||
"axis_delta_std": _round(_std([float(row["mean_axis_delta"]) for row in rs]), 3),
|
||||
"off_axis_problem_mean": _round(_mean([float(row["mean_off_axis_problem"]) for row in rs]), 3),
|
||||
"off_axis_problem_std": _round(_std([float(row["mean_off_axis_problem"]) for row in rs]), 3),
|
||||
"judge_std_mean": _round(_mean([float(row["mean_axis_delta_judge_std"]) for row in rs]), 3),
|
||||
"style_delta_mean": _round(_mean([float(row["mean_max_style_abs_delta"]) for row in rs]), 3),
|
||||
"persona_echo_rate_mean": _round(_mean([float(row["persona_echo_rate"]) for row in rs]), 3),
|
||||
"refusal_or_ai_break_rate_mean": _round(
|
||||
_mean([float(row["refusal_or_ai_break_rate"]) for row in rs]), 3),
|
||||
"models": ",".join(models),
|
||||
**base,
|
||||
})
|
||||
return sorted(out, key=lambda row: row["score_t"], reverse=True)
|
||||
|
||||
|
||||
def _markdown_text(text: str) -> str:
|
||||
text = docs_results.display_template_text(text)
|
||||
text = text.replace("{persona}", "`{persona}`")
|
||||
text = text.replace("&", "&")
|
||||
text = text.replace("<", "<")
|
||||
text = text.replace(">", ">")
|
||||
text = text.replace("\\", "\")
|
||||
text = text.replace("|", "|")
|
||||
return text.replace("\n", "<br>")
|
||||
|
||||
|
||||
def _write_markdown(path: Path, template_rows: list[dict[str, Any]], pair_rows: list[dict[str, Any]], top_n: int) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
top_template_rows = [
|
||||
{
|
||||
"score t": f"{row['score_t']:.2f}",
|
||||
"score mean": f"{row['score_mean']:.2f}",
|
||||
"score std": f"{row['score_std']:.2f}",
|
||||
"pass": f"{row['strict_pass_rate_mean']:.3f}",
|
||||
"echo": f"{row['persona_echo_rate_mean']:.3f}",
|
||||
"refusal": f"{row['refusal_or_ai_break_rate_mean']:.3f}",
|
||||
"template": _markdown_text(row["template"]),
|
||||
}
|
||||
for row in template_rows[:top_n]
|
||||
]
|
||||
lines = [
|
||||
"# Refusal-pole probe",
|
||||
"",
|
||||
"Scores are model-equal. Each model first averages the two refusal-probe axes per template, then the table reports reliability-sorted template rows across clean model artifacts.",
|
||||
"",
|
||||
"## All templates",
|
||||
"",
|
||||
"`score t` is mean score divided by standard error across the four clean model artifacts. `pass` is strict-pass rate; `echo` is explicit persona echo; `refusal` is refusal or AI-role break. Rows are sorted by `score t`.",
|
||||
"",
|
||||
tabulate(top_template_rows, headers="keys", tablefmt="github", disable_numparse=True),
|
||||
]
|
||||
path.write_text("\n".join(lines) + "\n")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--pair-stats", nargs="+", type=Path, default=DEFAULT_PAIR_STATS)
|
||||
ap.add_argument("--out-prefix", type=Path, default=DEFAULT_OUT_PREFIX)
|
||||
ap.add_argument("--top-n", type=int, default=999)
|
||||
args = ap.parse_args()
|
||||
|
||||
rows = []
|
||||
for path in args.pair_stats:
|
||||
model = _model_name(path)
|
||||
model_rows = []
|
||||
for row in _read_jsonl(path):
|
||||
model_rows.append({**row, "model": model, "score": _score(row)})
|
||||
if len(model_rows) != 190:
|
||||
raise ValueError(f"{path} has {len(model_rows)} rows, expected 190")
|
||||
rows.extend(model_rows)
|
||||
|
||||
template_rows = _summarize(_template_mean_rows(rows), ["template"])
|
||||
pair_rows = _summarize(rows, ["template", "persona_pair"])
|
||||
expected_models = len(args.pair_stats)
|
||||
if any(row["model_count"] != expected_models for row in template_rows + pair_rows):
|
||||
raise ValueError("at least one summary row is missing a model")
|
||||
|
||||
prefix = args.out_prefix
|
||||
_write_jsonl(prefix.with_name(prefix.name + "_template_model_summary.jsonl"), template_rows)
|
||||
_write_csv(prefix.with_name(prefix.name + "_template_model_summary.csv"), template_rows)
|
||||
_write_jsonl(prefix.with_name(prefix.name + "_template_pair_model_summary.jsonl"), pair_rows)
|
||||
_write_csv(prefix.with_name(prefix.name + "_template_pair_model_summary.csv"), pair_rows)
|
||||
_write_markdown(docs_results.REFUSAL_MODEL_MARKDOWN, template_rows, pair_rows, args.top_n)
|
||||
print(f"models={expected_models} templates={len(template_rows)} template_pairs={len(pair_rows)}")
|
||||
print(docs_results.REFUSAL_MODEL_MARKDOWN)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user