244 lines
13 KiB
Python
244 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""Полигон, пакет-7, фаза B′: слепая оценка на ПОЧИНЕННОМ наборе (пре-регистрация §0.3).
|
||
|
||
Чем отличается от `blind_judge.py` и почему это не «другой судья», а другой НАБОР.
|
||
Ревизия §B6 показала: инструмент фазы B оценивал наполовину вырожденные айтемы — у 35 термов из 80
|
||
все семь кандидатов были ОДНОЙ И ТОЙ ЖЕ строкой, и судью просили выбрать лучшую из семи одинаковых.
|
||
Отсюда плоские счета и «неразличимость армов». Здесь чинится сборка:
|
||
|
||
1) ДЕДУПЛИКАЦИЯ — одинаковые строки схлопываются в один вариант, арм-источники хранятся в ключе;
|
||
2) только термы, где после дедупликации осталось ≥2 варианта (иначе сравнивать нечего);
|
||
3) ДЕКОЙ — заведомо неверный вариант (dst ДРУГОГО терма той же выборки), подложенный вслепую.
|
||
|
||
Декой — несущая часть, а не украшение. В фазе B контроль «отличает ли судья годное от случайного»
|
||
был структурно НЕИЗМЕРИМ: строка `GOLD` не была уникальной ни на одном из 19 термов, поэтому
|
||
отбраковать судью по ней было нельзя (а отчёт B это сделал — ошибка R1). Декой уникален по
|
||
построению, и контроль становится измеримым.
|
||
|
||
Сырьё пишется вместе с `usage` и `finish_reason` — дефект записи фазы B (§B6 R7.2) чинится здесь.
|
||
"""
|
||
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 norm(s: str) -> str:
|
||
return re.sub(r"\s+", " ", (s or "").strip().lower())
|
||
|
||
|
||
def salt_order(term_id: str, keys: list[str]) -> list[str]:
|
||
h = hashlib.sha256(f"pkg7b2:{term_id}".encode()).hexdigest()
|
||
return sorted(keys, key=lambda k: hashlib.sha256((h + k).encode()).hexdigest())
|
||
|
||
|
||
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 {}
|
||
raw = m.group(0)
|
||
for extra in range(0, 4): # терпимость к недостающим скобкам (урок фазы B)
|
||
try:
|
||
v = json.loads(raw + "}" * extra)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if not isinstance(v, dict):
|
||
return {}
|
||
ver = v.get("verdicts")
|
||
if isinstance(ver, dict):
|
||
for key in ("catastrophic", "note"):
|
||
if key in ver and key not in v:
|
||
v[key] = ver.pop(key)
|
||
return v
|
||
return {}
|
||
|
||
|
||
def build_tasks(items: dict, by_id: dict) -> list[tuple]:
|
||
"""Дедупликация вариантов + подкладка декоя. Возвращает задания судье."""
|
||
# пул для декоя: по выборке — все dst ДРУГИХ термов
|
||
pool: dict[str, list[tuple[str, str]]] = {}
|
||
for tid, variants in by_id.items():
|
||
it = items.get(tid)
|
||
if not it:
|
||
continue
|
||
for dst in variants.values():
|
||
pool.setdefault(it["sample"], []).append((tid, dst))
|
||
|
||
tasks = []
|
||
for tid in sorted(by_id):
|
||
it = items.get(tid)
|
||
if not it:
|
||
continue
|
||
# схлопнуть одинаковые строки: ключ — нормализованный текст, значение — (показ, источники)
|
||
groups: dict[str, tuple[str, list[str]]] = {}
|
||
for src_label, dst in sorted(by_id[tid].items()):
|
||
k = norm(dst)
|
||
if k not in groups:
|
||
groups[k] = (dst, [])
|
||
groups[k][1].append(src_label)
|
||
if len(groups) < 2:
|
||
continue # вырожденный терм — сравнивать нечего
|
||
|
||
# Декой: dst ДРУГОГО терма той же выборки. Выбор детерминирован солью по терму (прогон
|
||
# воспроизводим), первый кандидат, не совпадающий ни с одним настоящим вариантом.
|
||
decoy = None
|
||
cand_pool = sorted({d for (other, d) in pool.get(it["sample"], []) if other != tid})
|
||
if cand_pool:
|
||
start = int(hashlib.sha256(f"decoy:{tid}".encode()).hexdigest(), 16) % len(cand_pool)
|
||
for k in range(len(cand_pool)):
|
||
pick = cand_pool[(start + k) % len(cand_pool)]
|
||
if norm(pick) not in groups:
|
||
decoy = pick
|
||
break
|
||
variants = {f"V{n}": g for n, g in enumerate(groups.values())}
|
||
if decoy:
|
||
variants["DECOY"] = (decoy, ["DECOY"])
|
||
|
||
crits = [c for c in CRITERIA if not (c == "K2" and it["sample"] in ("S3", "S4"))]
|
||
order = salt_order(tid, sorted(variants))
|
||
letters = "ABCDEFGH"
|
||
cands = [(letters[k], variants[key][0]) for k, key in enumerate(order)]
|
||
letter2sources = {letters[k]: variants[key][1] for k, key in enumerate(order)}
|
||
tasks.append((tid, it, cands, letter2sources, crits))
|
||
return tasks
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--termset", required=True, type=Path)
|
||
ap.add_argument("--arms", action="append", required=True, type=Path,
|
||
help="можно указать несколько файлов армов (фаза B + арм P5 фазы B′)")
|
||
ap.add_argument("--out", required=True, type=Path)
|
||
ap.add_argument("--raw-dir", required=True, type=Path)
|
||
ap.add_argument("--judge", required=True)
|
||
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("--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"))}
|
||
by_id: dict[str, dict[str, str]] = {}
|
||
for f in a.arms:
|
||
for r in json.loads(f.read_text(encoding="utf-8"))["results"]:
|
||
if r.get("dst"):
|
||
by_id.setdefault(r["id"], {})[r["arm"]] = r["dst"].strip()
|
||
for tid, it in items.items():
|
||
if it.get("gold") and tid in by_id:
|
||
by_id[tid]["GOLD"] = it["gold"].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))
|
||
|
||
tasks = build_tasks(items, by_id)
|
||
ndec = sum(1 for _, _, _, l2s, _ in tasks if any("DECOY" in v for v in l2s.values()))
|
||
sizes = [len(c) for _, _, c, _, _ in tasks]
|
||
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}: термов {len(tasks)} (из {len(by_id)}); с декоем {ndec}; "
|
||
f"вариантов на терм {min(sizes)}–{max(sizes)} (в фазе B было ровно 7)")
|
||
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, cands, l2s, 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"
|
||
elif a.style == "xai":
|
||
kw["max_tokens"] = a.max_tokens
|
||
kw["reasoning_effort"] = "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)
|
||
rows.append({"id": tid, "sample": it["sample"], "judge": a.judge,
|
||
"letter2sources": l2s, "candidates": dict(cands),
|
||
"verdicts": v.get("verdicts", {}), "catastrophic": v.get("catastrophic", []),
|
||
"note": v.get("note", "")})
|
||
(a.raw_dir / f"{a.judge}_{n:03d}.json").write_text(
|
||
json.dumps({"id": tid, "prompt": prompt, "response": txt,
|
||
"usage": {"in": cin, "out": cout}, "finish": resp.choices[0].finish_reason},
|
||
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())
|