156 lines
10 KiB
Python
156 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""exp14 — trap-accuracy судьи (Претензия 2): span-цитирующие, кросс-семейно,
|
||
self-family exclusion, катастроф-флаг. Бюджет-умно: панель gpt-5-mini + kimi (дёшево);
|
||
gemini — ТОЧЕЧНО (output $12/M, €2.6 крошечный). Claude-воркфлоу — адверсариальный верифай сверху.
|
||
|
||
Каждый (trap × arm) → per-judge {verdict: correct|partial|wrong, catastrophe, evidence≤15слов, note}.
|
||
Blind: армы под нейтральными метками; ключ отдельно. Персист в exp14/trap_verdicts.jsonl.
|
||
|
||
Использование: exp14_judges.py [--dry]
|
||
"""
|
||
from __future__ import annotations
|
||
import argparse, json, random, re, sys
|
||
from pathlib import Path
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
import exp14_common as C
|
||
|
||
ARMS = C.DIAG / "arms"
|
||
MAT = json.load(open(C.DIAG / "material.json"))
|
||
|
||
# ── Заморожённые трапы (§1.0.1) — файл-чанк, где живёт рендер трапа ─────────────────
|
||
TRAPS = [
|
||
dict(id="T1", chunk="7.0", type="c-within", catastrophe=True,
|
||
span="古月族人没有一个不知道的",
|
||
meaning="В роду Гуюэ ЭТО ПРЕДАНИЕ знали ВСЕ/каждый (двойное отрицание «нет ни одного, кто не знал бы»). Инверсия «никто не знал» = КАТАСТРОФА."),
|
||
dict(id="T2", chunk="8.1", type="c-within", catastrophe=False,
|
||
span="物是人非",
|
||
meaning="Контраст: вещи/места ПРЕЖНИЕ, но люди уже НЕ ТЕ (物是 «вещи те же» + 人非 «люди иные»). Плоское «всё изменилось» ТЕРЯЕТ половину идиомы (物是)."),
|
||
dict(id="T2b", chunk="8.0", type="c-title", catastrophe=False,
|
||
span="第八节：物是人非 (заглавие)",
|
||
meaning="Заглавие главы: тот же контраст «вещи прежние — люди иные». Плоское «Всё изменилось» — потеря атома."),
|
||
dict(id="T5", chunk="24.1", type="d-cross-prev", catastrophe=False,
|
||
span="这当然不是他的肌肤本色，而是一种铜皮蛊的效果",
|
||
meaning="«这/他» отсылает к бронзовому цвету кожи НАСТАВНИКА (описан в конце пред-чанка 24.0). Верно: «это [бронзовый цвет кожи наставника] — не природный, а эффект гу медной кожи». Провал = потеря антецедента (кто/что)."),
|
||
dict(id="T6", chunk="11.1", type="d-cross-prev", catastrophe=False,
|
||
span="那隐藏在暗处的人马，见到沈翠这般出来",
|
||
meaning="«那…人马» = засадный отряд, который舅父/舅母 подготовили (введён в конце 11.0: «внизу тоже расставлены люди»). Верно: понятно, ЧЬИ это люди/засада. Провал = «какие-то люди» без связи."),
|
||
dict(id="T4", chunk="9.0", type="b-dialogue-control", catastrophe=False,
|
||
span="舅父…开口道：“…” (обмен репликами дядя/тётя/Фан Чжэн)",
|
||
meaning="КОНТРОЛЬ-регресс: реплики с «—», слова автора при своей реплике (Розенталь). P0 это делает верно. Арм НЕ должен СЛОМАТЬ диалог-оформление."),
|
||
]
|
||
|
||
JUDGE_POOL = { # judge_model → family (для self-family exclusion)
|
||
"gpt-5-mini": "openai", "kimi-k2.6": "moonshot", "gemini-3.1-pro-preview": "google",
|
||
}
|
||
ARM_FAMILY = { # editor-модель арма → family (author≠reviewer)
|
||
"P0": "zai", "P1a": "zai", "P1b": "zai", "P1c": "zai", "A-layout": "zai",
|
||
"A-2pass": "zai", "A-ref-prev": "zai", "A-ref-both": "zai",
|
||
"F-base": "openai", "F-disc": "openai",
|
||
}
|
||
# draft общий = deepseek → deepseek не судья (self-family к черновику, где рождается T1)
|
||
|
||
JUDGE_SYS = ("Ты — билингвальный эксперт zh→ru, оцениваешь ТОЧНОСТЬ передачи одного смыслового "
|
||
"фрагмента. Тебе дан китайский span, его допустимый смысл, и русский перевод отрывка "
|
||
"(под нейтральной меткой). Ответь СТРОГО JSON без markdown: "
|
||
'{"verdict":"correct|partial|wrong","catastrophe":true|false,'
|
||
'"evidence":"дословная цитата из перевода ≤15 слов","note":"кратко почему"}. '
|
||
"verdict=correct только если смысл span передан без искажения полярности/участника/"
|
||
"потери атома. catastrophe=true при инверсии полярности или подмене участника/действия.")
|
||
|
||
|
||
def build_tasks(dry):
|
||
tasks = []
|
||
for tr in TRAPS:
|
||
arms = [a for a in ARM_FAMILY if (ARMS / a / f"{tr['chunk']}.txt").exists()]
|
||
# blind labels
|
||
order = list(arms); random.Random(f"exp14-trap-{tr['id']}").shuffle(order)
|
||
label = {a: f"V{i+1}" for i, a in enumerate(order)}
|
||
for a in arms:
|
||
txt = (ARMS / a / f"{tr['chunk']}.txt").read_text(encoding="utf-8")
|
||
# окно вокруг трапа не вырезаем — даём весь отрывок (судья видит контекст)
|
||
for jm in JUDGE_POOL:
|
||
if JUDGE_POOL[jm] == ARM_FAMILY[a]:
|
||
continue # self-family exclusion
|
||
if jm == "gemini-3.1-pro-preview":
|
||
continue # gemini — точечно, отдельным проходом (бюджет); тут пропускаем
|
||
tasks.append(dict(trap=tr["id"], arm=a, label=label[a], judge=jm,
|
||
chunk=tr["chunk"], text=txt, span=tr["span"],
|
||
meaning=tr["meaning"]))
|
||
return tasks
|
||
|
||
|
||
def run():
|
||
ap = argparse.ArgumentParser(); ap.add_argument("--dry", action="store_true"); a = ap.parse_args()
|
||
caps = {"gpt-5-mini": 0.05, "kimi-k2.6": 0.08, "gemini-3.1-pro-preview": 0.30}
|
||
sp = C.Spender(C.DIAG / "judge_costs.jsonl", caps)
|
||
tasks = build_tasks(a.dry)
|
||
# resume: пропустить уже отсуженные (trap,arm,judge)
|
||
vpath = C.DIAG / "trap_verdicts.jsonl"
|
||
out = []
|
||
done = set()
|
||
if vpath.exists():
|
||
for l in vpath.read_text(encoding="utf-8").splitlines():
|
||
if not l.strip():
|
||
continue
|
||
v = json.loads(l); out.append(v); done.add((v["trap"], v["arm"], v["judge"]))
|
||
tasks = [t for t in tasks if (t["trap"], t["arm"], t["judge"]) not in done]
|
||
print(f"trap-judge tasks: {len(tasks)} new ({len(done)} already done) "
|
||
f"({len(set(t['arm'] for t in tasks))} arms × {len(TRAPS)} traps × cross-family judges)")
|
||
if a.dry:
|
||
from collections import Counter
|
||
print("by judge:", Counter(t["judge"] for t in tasks))
|
||
return
|
||
for t in tasks:
|
||
user = (f"КИТАЙСКИЙ SPAN: {t['span']}\nДОПУСТИМЫЙ СМЫСЛ: {t['meaning']}\n\n"
|
||
f"РУССКИЙ ПЕРЕВОД ОТРЫВКА (метка {t['label']}):\n{t['text']}\n\n"
|
||
f"Оцени ТОЛЬКО передачу указанного span. Ответь JSON.")
|
||
s = C.spec(t["judge"], max_tokens=(16000 if t["judge"] == "kimi-k2.6" else 4000))
|
||
msgs = [{"role": "system", "content": JUDGE_SYS}, {"role": "user", "content": user}]
|
||
in_est = C.count_tokens(JUDGE_SYS + user, "glm")
|
||
ok, pred = sp.guard(t["judge"], in_est, s["max_tokens"])
|
||
if not ok:
|
||
print(f" OVER-CAP {t['judge']} {t['trap']} {t['arm']} pred=${pred:.3f}"); continue
|
||
text, err, usage = C.call(s, msgs, timeout=300)
|
||
c = sp.record({"model": t["judge"], "judge": t["judge"], "keyenv": s["keyenv"],
|
||
"trap": t["trap"], "arm": t["arm"], "err": err, "usage": usage})
|
||
v = parse_verdict(text) if not err else {"verdict": "ERR", "catastrophe": False, "note": err[:60]}
|
||
v.update(trap=t["trap"], arm=t["arm"], judge=t["judge"], label=t["label"])
|
||
out.append(v)
|
||
print(f" {t['trap']} {t['arm']:<10} {t['judge']:<18} → {v.get('verdict'):<8} cat={v.get('catastrophe')} ${c:.4f}")
|
||
(C.DIAG / "trap_verdicts.jsonl").write_text("\n".join(json.dumps(v, ensure_ascii=False) for v in out), encoding="utf-8")
|
||
summarize(out)
|
||
|
||
|
||
def parse_verdict(text: str) -> dict:
|
||
m = re.search(r"\{.*\}", text, re.S)
|
||
if not m:
|
||
return {"verdict": "PARSE", "catastrophe": False, "note": text[:60]}
|
||
try:
|
||
d = json.loads(m.group(0))
|
||
return {"verdict": str(d.get("verdict", "?")).lower(), "catastrophe": bool(d.get("catastrophe")),
|
||
"evidence": d.get("evidence", "")[:120], "note": d.get("note", "")[:120]}
|
||
except Exception:
|
||
vv = re.search(r'"verdict"\s*:\s*"(\w+)"', m.group(0))
|
||
return {"verdict": vv.group(1).lower() if vv else "PARSE", "catastrophe": "true" in m.group(0).lower(), "note": text[:60]}
|
||
|
||
|
||
def summarize(out):
|
||
from collections import defaultdict
|
||
agg = defaultdict(lambda: defaultdict(list))
|
||
for v in out:
|
||
agg[v["trap"]][v["arm"]].append(v)
|
||
print("\n=== TRAP-ACCURACY (per arm: majority verdict / catastrophe flags) ===")
|
||
for tr in TRAPS:
|
||
t = tr["id"]
|
||
if t not in agg: continue
|
||
print(f"\n{t} ({tr['type']}):")
|
||
for arm in ["P0","P1a","P1b","P1c","A-layout","A-2pass","A-ref-prev","A-ref-both","F-base","F-disc"]:
|
||
vs = agg[t].get(arm)
|
||
if not vs: continue
|
||
verds = [v["verdict"] for v in vs]
|
||
cat = any(v["catastrophe"] for v in vs)
|
||
print(f" {arm:<11} {verds} catastrophe={cat}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run()
|