textmachine/eval/role_topology/pair_en.py

449 lines
24 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""ВТОРАЯ ЯЗЫКОВАЯ ПАРА — en→ru на Kristoff, «Empire of the Dawn».
Зачем. Промт эксп-21 заказывал одну пару (zh→ru), и первая редакция мерила топологию только на
ней. Но цель проекта №1 канона — ОДИН движок на все пары, и вывод о топологии, снятый на одной
паре, этот вопрос не закрывает: ревью-вопрос по умолчанию «заработает ли пара, которой в репо
ещё НЕТ, без правки Go?» здесь звучит как «переживёт ли вывод смену пары». Расширение — по
слову владельца 07.08.
Почему именно эта книга. Собственная проба контаминации пака: `kristoff-en` узнаётся моделью
1/5 и даёт 0/5 по клоуз-тесту, тогда как `gu-zhenren` — 4/5 узнавания при пороге «≥3/5 = книга
модели ИЗВЕСТНА». То есть материал второй пары ЧИЩЕ основного, и часть оговорки «эмпирика на
претрейн-классике предварительна» (строка 55 бэклога) закрывается именно здесь.
Что тут особенного — ЭДЖ-КЕЙС, названный владельцем. Английский текст несёт французский
культурный слой тремя разными классами, и переводчик обязан развести их по-разному:
* имена и топонимы (`François`, `Fabién`, `Château`, `Lumière`) — передаются по ФРАНЦУЗСКИМ
правилам чтения, а не по английским, и одинаково на всю книгу;
* вставные французские реплики (`Merci, saint-père`, `mon dieu`) — чужая речь внутри чужой,
её сохранение/перевод есть решение, а не мелочь;
* офранцуженная английская лексика (`façade`) — обычное русское слово.
Отбор единиц СТРАТИФИЦИРОВАН по этому признаку: половина с французским слоем, половина без.
Стратификация объявлена ДО покупок и позволяет читать эффект отдельно на трудном материале.
Банкноты у армов этой пары НЕТ — объявленная девиация: подписанного глоссария для книги не
существует, а выдумывать канон полигон не вправе (CLAUDE.md). Следствие: контрасты пары мерят
ЧИСТУЮ топологию, без фактора банка; фактор банка мерится на zh контрастом A_law против A.
Запуск: --units инвентарь и стратификация ($0)
--arms черновики и армы
--judge судейство контрастов
--score свод ($0)
"""
from __future__ import annotations
import html
import json
import re
import sys
import time
import zipfile
from pathlib import Path
REPO = Path("/home/ubuntu/projects/textmachine")
for sub in ("editor_contract", "bank_arbitration", "editor_harness", "role_topology"):
sys.path.insert(0, str(REPO / "eval" / sub))
import battery as B # noqa: E402
import buy as BUY # noqa: E402
import editor_wire_probe as P # noqa: E402
import judges as J # noqa: E402
from bakeoff import boot_ci, client, uid_of # noqa: E402
import statistics # noqa: E402
EPUB = Path.home() / "books" / "Kristoff_Jay_-_Empire_of_the_Vampire_3_-_Empire_of_the_Dawn.epub"
OUT = Path.home() / "books" / "role-topology"
N_PER_STRATUM = 6
UNIT_CHARS = 1600
REPS_PER_ORDER = 1
CEIL_ARMS = 0.55
CEIL_JUDGE = 1.45
LED_ARMS = BUY.Ledger("армы en→ru", CEIL_ARMS, ("en2-*.json",), default_expect=0.03)
LED_JUDGE = BUY.Ledger("судейство en→ru", CEIL_JUDGE, ("jve-*.json",), default_expect=0.008)
DRAFT_MODEL = "deepseek-v4-flash"
ARMS = {
"A": dict(model="deepseek-v4-pro", contract="full", think=None),
"B": dict(model="glm-5", contract="full", think="off"),
"D": dict(model="deepseek-v4-pro", contract="direct-reflow", think=None),
"D_": dict(model="deepseek-v4-pro", contract="direct", think=None),
}
CONTRASTS = [("A", "F"), ("B", "F"), ("D", "A"), ("D", "D_")]
# Французский слой опознаётся по диакритике — признак механический и не требует словаря.
_RE_FR = re.compile(r"\b\w*[éèêëàâäçôöûùüïî]\w*\b")
_RE_LAT = re.compile(r"[A-Za-zÀ-ɏ]")
def raw_text() -> str:
z = zipfile.ZipFile(EPUB)
parts = []
for n in sorted(x for x in z.namelist() if x.endswith((".xhtml", ".html", ".htm"))):
t = re.sub(r"<[^>]+>", " ", z.read(n).decode("utf-8", "ignore"))
parts.append(re.sub(r"[ \t]+", " ", html.unescape(t)))
return "\n".join(parts)
def chunks() -> list[str]:
"""Единицы боевого размера, нарезанные ПО ГРАНИЦАМ ПРЕДЛОЖЕНИЙ.
Рвать посреди фразы нельзя: обрубок меняет задачу переводчика и портит сравнение армов.
"""
text = raw_text()
sents = re.split(r"(?<=[.!?”\"])\s+", text)
out, buf = [], ""
for s in sents:
buf += s + " "
if len(buf) >= UNIT_CHARS:
out.append(buf.strip())
buf = ""
return [c for c in out if len(c) <= UNIT_CHARS * 1.4]
_SERVICE = ("also by ", "about the author", "copyright", "all rights reserved", "isbn",
"newsletter", "st. martin", "thank you for buying", "table of contents",
"for email updates", "click here", "acknowledgments")
def is_prose(c: str) -> bool:
"""Проза ли это, или служебная страница книги.
⚠ Заведено 07.08 по адверсариальному ревью: единица `129b73ad5a` страты `plain` оказалась
выходными данными и библиографией («ALSO BY JAY KRISTOFF», 18 названий, копирайт, рассылка).
Армы её ПЕРЕВОДИЛИ, и она ушла в судейство и в слепое чтение. Фильтр страты был «нет токенов
с диакритикой» — служебные страницы он притягивает по построению, потому что диакритики там
нет никогда.
Два сигнала, оба механические: маркеры издательского аппарата и ПЛОТНОСТЬ ЗАВЕРШЁННЫХ
ПРЕДЛОЖЕНИЙ. У прозы точек/восклицаний/вопросов на 1000 знаков много; у списка названий и
выходных данных — мало, там строки без терминальной пунктуации.
"""
low = c.lower()
if any(m in low for m in _SERVICE):
return False
return len(re.findall(r"[.!?]", c)) / max(1, len(c)) * 1000 >= 8.0
def units() -> list[dict]:
"""12 единиц: 6 с французским слоем, 6 без. Отбор детерминированный — по длине внутри страты.
Порог страты `fr`: ≥4 токена с диакритикой, из них ≥2 РАЗНЫХ. Одиночное `façade` французским
слоем не является, и страту им засорять нельзя.
"""
fr, plain = [], []
for c in chunks():
if not is_prose(c):
continue
toks = _RE_FR.findall(c)
if len(toks) >= 4 and len(set(toks)) >= 2:
fr.append(c)
elif not toks:
plain.append(c)
out = []
for name, pool in (("fr", fr), ("plain", plain)):
# ⚠ Ключ сортировки — (длина, uid), а НЕ одна длина. Поймано исполнением 07.08: два
# вызова подряд вернули разные наборы единиц. Причина — `sorted(set(...), key=len)`:
# при равной длине порядок задаётся обходом МНОЖЕСТВА строк, а он рандомизирован по
# процессам (хеш-сид). Отбор при этом выглядел детерминированным. Последствие было бы
# тихим и дорогим: армы покупаются на один набор, судейство идёт по другому.
pool = sorted(set(pool), key=lambda c: (len(c), uid_of(c)))
lo = (len(pool) - N_PER_STRATUM) // 2
for c in pool[lo:lo + N_PER_STRATUM]:
out.append(dict(source=c, uid=uid_of(c), stratum=name,
fr_tokens=sorted(set(_RE_FR.findall(c)))))
return out
PAIR = Path(__file__).parent / "prompts" / "en-ru"
BRIEF = {"source_lang": "en", "target_lang": "ru", "genre": "тёмное фэнтези",
"audience": "взрослые читатели фэнтези", "title": "Empire of the Dawn",
"venuti": "0.60", "honorifics": "keep", "transcription": "practical",
"footnotes": "minimal"}
def render_pair(path: Path, text: str, draft: str) -> tuple[str, str]:
"""Рендер пар-пакета ТЕМ ЖЕ путём, что боевой: снятие комментариев, дроп FEWSHOT, подстановка.
⚠ Флаг `en=True` харнесса экспов 1820 сюда НЕ годится и был пойман исполнением: он означает
«переводить НА английский» (трек zh→en экспа-19), и первая сборка армов этой пары уехала бы
с промптом «editor of a translation into English» и брифом «с языка zh на язык en».
Здесь источник английский, цель русская — это другая пара, и ей нужен свой пар-пакет.
"""
canon = P.strip_comments(path.read_text(encoding="utf-8"))
sys_part, user = canon.split(P.USER_SEP, 1)
core = sys_part.split(P.FEWSHOT_SEP, 1)[0].strip()
old = P.RENDER_VARS
P.RENDER_VARS = BRIEF
try:
return P.render(core, text, draft), P.render(user.strip(), text, draft)
finally:
P.RENDER_VARS = old
def translator_msgs(text: str) -> tuple[str, str]:
return render_pair(PAIR / "translator.md", text, "")
def build_msgs(arm: str, u: dict, draft: str) -> list[dict]:
tpl = {"full": "editor.md", "direct": "translator.md",
"direct-reflow": "translator-reflow.md"}[ARMS[arm]["contract"]]
sys_msg, user = render_pair(PAIR / tpl, u["source"], draft)
return [{"role": "system", "content": sys_msg}, {"role": "user", "content": user.strip()}]
def run_draft(u: dict) -> dict:
sys_msg, user = translator_msgs(u["source"])
kw = dict(model=DRAFT_MODEL, max_tokens=16000, extra_body={"reasoning_effort": "low"},
messages=[{"role": "system", "content": sys_msg}, {"role": "user", "content": user}])
# Эхо-мина у DeepSeek описана для плотного CJK; на латинице класс не подтверждён, поэтому
# сторож здесь ловит не «ханьцзы», а НЕПЕРЕВЕДЁННУЮ ЛАТИНИЦУ — тот же дефект, другой алфавит.
for attempt in (1, 2, 3):
tag = f"en2-DRAFT-{u['uid']}" + ("" if attempt == 1 else f"-r{attempt}")
rec = BUY.purchase(LED_ARMS, tag, DRAFT_MODEL, client(DRAFT_MODEL), kw,
arm="DRAFT", uid=u["uid"], attempt=attempt)
if rec.get("skipped") or latin_share(rec["content"]) <= 0.15:
return rec
print(f" ⚠ латиница в черновике {u['uid'][:6]} (попытка {attempt}) — ре-ген")
return rec
def latin_share(text: str) -> float:
letters = [c for c in text if c.isalpha()]
return len(_RE_LAT.findall("".join(letters))) / max(1, len(letters))
def draft_text(uid: str) -> str:
f = OUT / f"en2-DRAFT-{uid}.json"
best = ""
for g in sorted(OUT.glob(f"en2-DRAFT-{uid}*.json")):
c = json.loads(g.read_text(encoding="utf-8")).get("content", "")
if c and latin_share(c) <= 0.15:
best = c
return best or (json.loads(f.read_text(encoding="utf-8"))["content"] if f.exists() else "")
def draft_cost(uid: str) -> float:
return sum(json.loads(g.read_text(encoding="utf-8")).get("cost_usd", 0.0)
for g in OUT.glob(f"en2-DRAFT-{uid}*.json"))
def run_arm(arm: str, u: dict) -> dict:
spec = ARMS[arm]
model = spec["model"]
kw: dict = dict(model=model, messages=build_msgs(arm, u, draft_text(u["uid"])),
max_tokens=16000, temperature=0.4)
if spec["think"] == "off":
if not model.startswith("glm"):
raise SystemExit(f"гашение размышления у {model} не описано в quirks — не гадаю")
kw["extra_body"] = {"thinking": {"type": "disabled"}}
return BUY.purchase(LED_ARMS, f"en2-{arm}-{u['uid']}", model, client(model), kw,
arm=arm, uid=u["uid"], contract=spec["contract"], think=spec["think"])
def text_of(arm: str, u: dict) -> str:
if arm == "F":
return draft_text(u["uid"])
f = OUT / f"en2-{arm}-{u['uid']}.json"
return json.loads(f.read_text(encoding="utf-8"))["content"] if f.exists() else ""
# --- ФРАНЦУЗСКИЙ СЛОЙ: детерминированная проверка, судья не нужен -------------------------------
def check_french(source: str, final: str) -> dict:
"""Как обошлись с французским слоем. Три числа, каждое читается само по себе.
`leaked` — французский токен остался ЛАТИНИЦЕЙ в русском тексте. Это дефект передачи:
читатель кириллического издания получает нечитаемое слово.
`rendered` — токен передан кириллицей хотя бы раз (передача состоялась).
`split` — один и тот же французский токен передан РАЗНЫМИ кириллическими формами внутри
одной единицы. Это прокси главной продуктовой боли — консистентности имени на всю книгу;
глоссария у книги нет, поэтому мерится расщепление внутри окна, а не против канона.
"""
toks = sorted({t for t in _RE_FR.findall(source) if len(t) >= 4})
leaked = [t for t in toks if re.search(rf"\b{re.escape(t)}\b", final)]
# Кириллические кандидаты-передачи: слова с заглавной буквы вокруг позиций, где стоял токен.
cyr_words = re.findall(r"\b[А-ЯЁ][а-яё]+(?:-[А-ЯЁ]?[а-яё]+)?\b", final)
forms: dict[str, set[str]] = {}
for t in toks:
head = _fold(t[:3])
cand = {w for w in cyr_words if _fold_cyr(w[:3]) == head}
if cand:
forms[t] = cand
split = {t: sorted(v) for t, v in forms.items() if len(v) > 1}
return dict(fr_tokens=len(toks), leaked=leaked, leaked_n=len(leaked),
rendered_n=len(forms), split=split, split_n=len(split))
_FR2LAT = str.maketrans("éèêëàâäçôöûùüïî", "eeeeaaacoouuuii")
_CYR2LAT = {"а": "a", "б": "b", "в": "v", "г": "g", "д": "d", "е": "e", "ж": "j", "з": "z",
"и": "i", "й": "i", "к": "k", "л": "l", "м": "m", "н": "n", "о": "o", "п": "p",
"р": "r", "с": "s", "т": "t", "у": "u", "ф": "f", "х": "h", "ц": "c", "ч": "c",
"ш": "s", "щ": "s", "ы": "i", "э": "e", "ю": "u", "я": "a", "ё": "e"}
def _fold(s: str) -> str:
return s.lower().translate(_FR2LAT)
def _fold_cyr(s: str) -> str:
return "".join(_CYR2LAT.get(c, c) for c in s.lower())
def cmd_units() -> None:
us = units()
print(f"единиц {len(us)} · страты: "
f"{ {s: sum(1 for u in us if u['stratum'] == s) for s in ('fr', 'plain')} }")
for u in us:
print(f" {u['uid'][:6]} [{u['stratum']:5s}] {len(u['source']):5d} зн · "
f"фр-токенов {len(u['fr_tokens']):2d} {u['fr_tokens'][:6]}")
def cmd_arms() -> None:
us = units()
print(f"единиц {len(us)} · армов {len(ARMS)} + черновик · потолок ${CEIL_ARMS}")
for u in us:
rec = run_draft(u)
if rec.get("skipped"):
print("⛔ потолок — стоп")
return
print(f" DRAFT {u['uid'][:6]} [{u['stratum']:5s}] {len(rec['content']):5d} зн "
f"латиница {latin_share(rec['content']):.3f} ${rec['cost_usd']:.5f}")
time.sleep(0.2)
for arm in ARMS:
for u in us:
rec = run_arm(arm, u)
if rec.get("skipped"):
print("⛔ потолок — стоп")
return
print(f" {arm:3s} {u['uid'][:6]} finish={rec.get('finish','?'):8s} "
f"{len(rec['content']):5d} зн размышл {rec.get('reasoning_tokens',0):6d} "
f"${rec['cost_usd']:.5f}")
time.sleep(0.2)
print(f"\nпотрачено ${LED_ARMS.spent():.6f}")
def cmd_judge() -> None:
us = units()
total = len(us) * len(CONTRASTS) * len(J.JUDGES) * 2 * REPS_PER_ORDER
print(f"голосов к покупке {total} · потолок ${CEIL_JUDGE}")
n = 0
for a, b in CONTRASTS:
for u in us:
ta, tb = text_of(a, u), text_of(b, u)
if not ta.strip() or not tb.strip():
print(f"{a}/{b} {u['uid'][:6]}: пустой выход — пропуск")
continue
for judge in J.JUDGES:
for order in (0, 1):
for rep in range(1, REPS_PER_ORDER + 1):
v1, v2 = (ta, tb) if order == 0 else (tb, ta)
rec = BUY.purchase(
LED_JUDGE, f"jve-{a}v{b}-{u['uid']}-o{order}-{judge}-r{rep}",
judge, J.client(judge), J.vote_kw(judge, J.packet(u["source"], v1, v2)),
contrast=f"{a}v{b}", uid=u["uid"], order=order, judge=judge,
stratum=u["stratum"])
if rec.get("skipped"):
print("⛔ потолок судейства — стоп")
return
n += 1
time.sleep(0.15)
print(f" {a} против {b}: голосов {n} · ${LED_JUDGE.spent():.4f}")
print(f"\nголосов {n} · ${LED_JUDGE.spent():.6f}")
def _margins(a: str, b: str, uid: str, judge: str) -> dict[int, list[float]]:
out: dict[int, list[float]] = {}
for order in (0, 1):
for rep in range(1, REPS_PER_ORDER + 1):
f = OUT / f"jve-{a}v{b}-{uid}-o{order}-{judge}-r{rep}.json"
if not f.exists():
continue
p = J.parse(json.loads(f.read_text(encoding="utf-8"))["content"])
if p["В1-ВЕРНОСТЬ"] is None:
continue
e1 = sum(p[f"В1-{ax}"] or 0 for ax in J.AXES)
e2 = sum(p[f"В2-{ax}"] or 0 for ax in J.AXES)
out.setdefault(order, []).append((e2 - e1) if order == 0 else (e1 - e2))
return out
def cmd_score() -> None:
us = units()
print("ПАРНЫЕ ВЕРДИКТЫ en→ru. Знак перевеса обязан совпасть в ОБОИХ порядках у ОБОИХ судей.\n")
print(f"{'контраст':14s}{'ед.':>5s}{'за перв.':>10s}{'за втор.':>10s}{'ничья':>7s}"
f"{'перевес':>10s}{'95% ДИ':>18s}{'фр-страта':>11s}{'plain':>8s}")
print("-" * 93)
for a, b in CONTRASTS:
wa = wb = tie = 0
per_unit, by_str = [], {"fr": [], "plain": []}
for u in us:
pj, ms = {}, []
for judge in J.JUDGES:
bo = _margins(a, b, u["uid"], judge)
if len(bo) < 2:
continue
m0, m1 = statistics.mean(bo[0]), statistics.mean(bo[1])
ms += [m0, m1]
pj[judge] = 1 if (m0 > 0 and m1 > 0) else (-1 if (m0 < 0 and m1 < 0) else 0)
if len(pj) < len(J.JUDGES):
continue
per_unit.append(statistics.mean(ms))
by_str[u["stratum"]].append(statistics.mean(ms))
vals = set(pj.values())
wa += vals == {1}
wb += vals == {-1}
tie += vals not in ({1}, {-1})
if not per_unit:
print(f"{a + ' пр. ' + b:14s}{'нет голосов':>20s}")
continue
lo, hi = boot_ci(per_unit)
f_m = statistics.mean(by_str["fr"]) if by_str["fr"] else float("nan")
p_m = statistics.mean(by_str["plain"]) if by_str["plain"] else float("nan")
print(f"{a + ' пр. ' + b:14s}{wa + wb + tie:5d}{wa:10d}{wb:10d}{tie:7d}"
f"{statistics.mean(per_unit):+10.2f} [{lo:+.2f}, {hi:+.2f}]{f_m:+11.2f}{p_m:+8.2f}")
print("\nФРАНЦУЗСКИЙ СЛОЙ (детерминированно, судья не нужен)\n")
print(f"{'арм':5s}{'ед.':>5s}{'фр-токенов':>12s}{'латиница':>10s}{'передано':>10s}"
f"{'расщеплено':>12s}{'$/единицу':>11s}")
print("-" * 65)
frs = [u for u in us if u["stratum"] == "fr"]
for arm in ("F", *ARMS):
rows, costs = [], []
for u in frs:
t = text_of(arm, u)
if not t.strip():
continue
rows.append(check_french(u["source"], t))
c = draft_cost(u["uid"]) if arm in ("F", "A", "B") else 0.0
f = OUT / f"en2-{arm}-{u['uid']}.json"
if arm != "F" and f.exists():
c += json.loads(f.read_text(encoding="utf-8"))["cost_usd"]
costs.append(c)
if not rows:
continue
print(f"{arm:5s}{len(rows):5d}{sum(r['fr_tokens'] for r in rows):12d}"
f"{sum(r['leaked_n'] for r in rows):10d}{sum(r['rendered_n'] for r in rows):10d}"
f"{sum(r['split_n'] for r in rows):12d}"
f"{(statistics.median(costs) if costs else 0):11.5f}")
print("\nОБЩАЯ БАТАРЕЯ (языконезависимые проверки)\n")
print(f"{'арм':5s}{'ед.':>5s}{'типогр.':>9s}{'ты/вы микс':>12s}")
print("-" * 31)
for arm in ("F", *ARMS):
rows = [B.run_unit(B.Unit(source=u["source"], draft=draft_text(u["uid"]),
final=text_of(arm, u), cards={}))
for u in us if text_of(arm, u).strip()]
if rows:
print(f"{arm:5s}{len(rows):5d}{sum(r['typography']['violations'] for r in rows):9d}"
f"{sum(r['ty_vy']['mixed_in_unit'] for r in rows):12d}")
if __name__ == "__main__":
cmd = {"--units": cmd_units, "--arms": cmd_arms, "--judge": cmd_judge,
"--score": cmd_score}.get(sys.argv[1] if len(sys.argv) > 1 else "")
if cmd:
cmd()
else:
print(__doc__)