#!/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, self-family exclusion) "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", # gpt-5.4 "G-base": "google", "G-disc": "google", # gemini-3.1-pro-preview "X-base": "xai", "X-disc": "xai", # grok-4.3 reasoning-ON } # 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()