#!/usr/bin/env python3 """Полигон, пакет-7, фаза B: слепая оценка Best-Worst Scaling по критериям рубрики §A2. Почему BWS, а не абсолютные баллы: на художественном материале MQM-скоринг ошибочно признаёт ~60% ЧЕЛОВЕЧЕСКИХ переводов не лучше машинных, тогда как Best-Worst Scaling опознаёт человека в 80–100% (§A2.0). Абсолютная шкала остаётся только для катастроф-экрана. Слепота (D13.5 + §A2.4): варианты перемешиваются СОЛЬЮ ПО ТЕРМУ, метки A..F не несут информации об арме, судья не видит ни имени арма, ни эталона владельца. Порядок сохраняется в ключе отдельно. Судейская дисциплина (D13.3): судья не судит выход своего семейства — консолидатор deepseek, судьи из ДРУГИХ семейств. Позиционный своп меряется отдельным прогоном, а не считается бесплатным (§A2.4 п.5: на кураторских наборах своп может УХУДШАТЬ). Критерий K2 «жанровая конвенция» на выборках S3/S4 помечен н/д и НЕ оценивается: конвенции для ja/en у нас нет, оценивать соответствие несуществующему — купить бессмысленный вердикт (§A5.3). """ from __future__ import annotations import argparse import hashlib import json import os import re import sys import time from pathlib import Path import yaml from dotenv import load_dotenv from openai import OpenAI REPO = Path(__file__).resolve().parents[2] load_dotenv(REPO / "eval" / ".env") CRITERIA = { "K1": "верность концепту: вариант обозначает то же понятие, что исходный термин в этих контекстах", "K2": "жанровая конвенция: вариант совпадает с тем, как этот класс терминов принято передавать в русских изданиях жанра", "K5": "морфологическая пригодность: вариант склоняем по-русски, у него устойчивый род, он не ломает согласование", "K6": "литературность и благозвучие: вариант годится в художественный текст, не режет ухо", "K9": "уместность регистра: не подставлен академический/канцелярский термин туда, где нужен жанровый, и наоборот", } JUDGE_TASK = """Ты — эксперт по художественному переводу на русский язык. Тебе дают термин исходного текста, его контексты из книги и НЕСКОЛЬКО вариантов русского соответствия, помеченных буквами. Кто какой вариант предложил — тебе неизвестно и не важно. Для каждого названного критерия выбери ЛУЧШИЙ и ХУДШИЙ вариант. Отдельно отметь варианты, которые являются КАТАСТРОФОЙ: меняют смысл на другой/противоположный, либо являются не переводом, а мусором. Отвечай СТРОГО одним JSON-объектом без пояснений вне JSON: {"verdicts": {"<критерий>": {"best": "<буква>", "worst": "<буква>"}, ...}, "catastrophic": ["<буква>", ...], "note": "<до 25 слов, почему лучший лучший>"}""" def salt_order(term_id: str, arms: list[str], reverse: bool) -> list[str]: h = hashlib.sha256(f"pkg7:{term_id}".encode()).hexdigest() order = sorted(arms, key=lambda a: hashlib.sha256((h + a).encode()).hexdigest()) return list(reversed(order)) if reverse else order def build_prompt(item: dict, cands: list[tuple[str, str]], crits: list[str]) -> str: ctx = "\n".join(f" …{k['win']}…" for k in item["kwic"][:6]) or " (контекстов нет)" opts = "\n".join(f" {letter}. {dst}" for letter, dst in cands) cr = "\n".join(f" {c}: {CRITERIA[c]}" for c in crits) return (f"{JUDGE_TASK}\n\nИСХОДНЫЙ ТЕРМИН: {item['src']}\n" f"ЯЗЫК ИСХОДНИКА: {item['lang']}\n\nКОНТЕКСТЫ ИЗ КНИГИ:\n{ctx}\n\n" f"ВАРИАНТЫ:\n{opts}\n\nКРИТЕРИИ:\n{cr}") def parse_obj(text: str) -> dict: m = re.search(r"\{.*\}", text, re.S) if not m: return {} try: v = json.loads(m.group(0)) return v if isinstance(v, dict) else {} except json.JSONDecodeError: return {} def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--termset", required=True, type=Path) ap.add_argument("--arms", required=True, type=Path) ap.add_argument("--out", required=True, type=Path) ap.add_argument("--raw-dir", required=True, type=Path) ap.add_argument("--judge", required=True, help="имя модели из models.yaml") ap.add_argument("--base-url", required=True) ap.add_argument("--key-env", required=True) ap.add_argument("--style", choices=["openai-reasoning", "plain", "xai"], default="plain") ap.add_argument("--reverse", action="store_true", help="прогон с ОБРАЩЁННЫМ порядком (замер свопа)") ap.add_argument("--max-tokens", type=int, default=4000) ap.add_argument("--dry-run", action="store_true") a = ap.parse_args() items = {i["id"]: i for i in json.loads(a.termset.read_text(encoding="utf-8"))} arms_data = json.loads(a.arms.read_text(encoding="utf-8"))["results"] by_id: dict[str, dict[str, str]] = {} for r in arms_data: if r.get("dst"): by_id.setdefault(r["id"], {})[r["arm"]] = r["dst"].strip() doc = yaml.safe_load((REPO / "backend/configs/models.yaml").read_text(encoding="utf-8")) or {} p = ((doc.get("models") or {}).get(a.judge) or {}).get("price") or {} pin, pout = float(p.get("input_per_m", 0)), float(p.get("output_per_m", 0)) # Подписанный владельцем dst участвует в слепом сравнении НАРАВНЕ с армами (§A2.4 п.4): # без него нельзя вынести F2 («бьёт ли арм эталон»), а оценщик, не отличающий подпись от # случайного варианта, обязан быть отбракован — увидеть это можно только так. for tid, it in items.items(): if it.get("gold") and tid in by_id: by_id[tid]["GOLD"] = it["gold"].strip() tasks = [] for tid, variants in by_id.items(): it = items.get(tid) if not it or len(variants) < 2: continue crits = [c for c in CRITERIA if not (c == "K2" and it["sample"] in ("S3", "S4"))] order = salt_order(tid, sorted(variants), a.reverse) letters = "ABCDEFG" cands = [(letters[k], variants[arm]) for k, arm in enumerate(order)] tasks.append((tid, it, order, cands, crits)) est_in = sum(len(build_prompt(it, c, cr)) for _, it, _, c, cr in tasks) / 3.2 est_out = len(tasks) * 350 print(f"ПЛАН судьи {a.judge}{' (обратный порядок)' if a.reverse else ''}: {len(tasks)} термов") print(f"СМЕТА ДО ВЫЗОВОВ: вход ≈{est_in/1000:.1f}k → ${est_in/1e6*pin:.4f}; " f"выход ≈{est_out/1000:.1f}k → ${est_out/1e6*pout:.4f}; " f"ИТОГО ≈ ${est_in/1e6*pin + est_out/1e6*pout:.4f}") if a.dry_run: return 0 key = os.environ.get(a.key_env) if not key: print(f"нет ключа {a.key_env}", file=sys.stderr) return 2 client = OpenAI(api_key=key, base_url=a.base_url) a.raw_dir.mkdir(parents=True, exist_ok=True) rows, ledger = [], [] for n, (tid, it, order, cands, crits) in enumerate(tasks, 1): prompt = build_prompt(it, cands, crits) kw: dict = {"model": a.judge, "messages": [{"role": "user", "content": prompt}]} if a.style == "openai-reasoning": kw["max_completion_tokens"] = a.max_tokens kw["reasoning_effort"] = "minimal" # квирк gpt-5-mini elif a.style == "xai": kw["max_tokens"] = a.max_tokens kw["reasoning_effort"] = "low" # квирк grok: дефолт low, явно else: kw["max_tokens"] = a.max_tokens try: resp = client.chat.completions.create(**kw) except Exception as e: print(f"[{n}/{len(tasks)}] {tid} ОШИБКА: {e}", file=sys.stderr) ledger.append({"id": tid, "error": str(e)}) continue txt = resp.choices[0].message.content or "" u = resp.usage cin, cout = getattr(u, "prompt_tokens", 0), getattr(u, "completion_tokens", 0) usd = cin / 1e6 * pin + cout / 1e6 * pout v = parse_obj(txt) letters = [c[0] for c in cands] letter2arm = {letters[k]: order[k] for k in range(len(order))} rows.append({"id": tid, "sample": it["sample"], "judge": a.judge, "reverse": bool(a.reverse), "letter2arm": letter2arm, "verdicts": v.get("verdicts", {}), "catastrophic": v.get("catastrophic", []), "note": v.get("note", "")}) (a.raw_dir / f"{a.judge}_{'rev' if a.reverse else 'fwd'}_{n:03d}.json").write_text( json.dumps({"id": tid, "prompt": prompt, "response": txt}, ensure_ascii=False, indent=1), encoding="utf-8") ledger.append({"id": tid, "in": cin, "out": cout, "usd": usd, "finish": resp.choices[0].finish_reason, "ok": bool(v.get("verdicts"))}) if n % 10 == 0 or n == len(tasks): print(f" [{n}/{len(tasks)}] ${sum(x.get('usd',0) for x in ledger):.4f}") time.sleep(0.2) a.out.write_text(json.dumps({"rows": rows, "ledger": ledger}, ensure_ascii=False, indent=1), encoding="utf-8") tot = sum(x.get("usd", 0) for x in ledger) ok = sum(1 for r in rows if r["verdicts"]) print(f"\nСУДЬЯ {a.judge}: вердиктов {ok}/{len(tasks)}, ПОТРАЧЕНО ${tot:.5f}") return 0 if __name__ == "__main__": raise SystemExit(main())