textmachine/eval/role_topology/phase2b.py

416 lines
27 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
"""Ф2б — АРМЫ C И E НА ТЕКУЩЕЙ ПАНЕЛИ. Вторая половина заказа промта (стр.114115).
═══════════════════════════════════ ПОСЫЛКА ═══════════════════════════════════
ЧТО ЗАКАЗАНО. Промт стр.114115 требует два арма, которых пак не гнал ни разу на боевой панели:
C — черновик → детекторы+гейты → ТОЧЕЧНЫЙ РЕМОНТ дешёвым фиксером, в ДВУХ режимах:
oracle-флаги (потолок схемы) и реальные детекторы (пол); ре-гейт после КАЖДОГО фикса,
кап 2 итерации; печатать recall детекторов на срезе.
E — ОБРАТНАЯ связка: сильный черновик (однопроходка `deepseek-v4-pro`) → дешёвый фиксер по флагу.
Это и есть главная альтернатива полному переписыванию, ради которой пак затевался после эксп-20.
ПОЧЕМУ СТАРЫЕ МОДУЛИ НЕ ПЕРЕИСПОЛЬЗУЮТСЯ. `repair_arm.py` и `route_arm.py` считали на СНЯТОЙ
восьмиюнитовой панели (до дедупа и хеш-ключей), их записи не несут полей модели и арма, лежат вне
всех касс, а `route_arm.py` в текущем дереве падает `TypeError` (сигнатура `text_of` разошлась).
Числа оттуда с Ф2а несравнимы. Здесь всё считается на ТЕХ ЖЕ 16 единицах и ТЕХ ЖЕ черновиках.
ЧТО ИЗМЕНИЛОСЬ С МОМЕНТА ИХ НАПИСАНИЯ — и это ключ к арму C. Старый модуль объявлял: детектора
смысловой инверзии (класс k2) НЕТ ВООБЩЕ, поэтому разрыв «oracle против реального» равен доле
этого класса, то есть 50% посаженного. **Это устарело.** Мини-бенч §13.7 намерил, что QE-ранкер
инверзию ВИДИТ — но на уровне СЕГМЕНТА (66/72, p=3.6e-14), а не окна. Значит реальный режим арма
C теперь имеет детектор на оба класса, и разрыв потолка и пола обязан сократиться. Насколько —
и есть главный новый вопрос.
СЕМЕЙСТВО КОНТРАСТОВ, объявленное ДО первой покупки. Поправка Холма по всем пяти.
1. C_real против F3 — покупает ли точечный ремонт хоть что-то поверх черновика
2. C_real против A_law — точечный ремонт против полного переписывания (главный вопрос)
3. E против D_ — покупает ли добивка что-то поверх сильной однопроходки
4. E против A_law — обратная связка против прямой
5. C_oracle против C_real — разрыв потолка и пола схемы (пре-рег называл это главным по C)
БАЗА СРАВНЕНИЯ — `A_law`, а не `A`. §16.1 намерил, что редактор без закон-блока единственный из
шести армов ломает терминологию черновика (покрытие канона 0.566 против 0.917), поэтому боевой
конфигурацией объявлен `A_law`. Сравнивать точечный ремонт надо с тем, что реально пойдёт в прод.
⚠ ОТКЛОНЕНИЕ ОТ БУКВЫ ПРОМТА, объявленное ЗАРАНЕЕ. Промт требует «черновик С МАНДАТОМ перевёрстки
— иначе вёрстка-конфаунд». Здесь арм C правит ТОТ ЖЕ черновик `F3`, что и армы `A`/`A_law`, без
мандата. Причина: мандат перевёрстки живёт в арме `D`, и взяв черновик с мандатом, мы потеряли бы
ПАРНОСТЬ с `A_law` — а именно она и есть предмет замера. Конфаунд формы при этом мал и измерен:
вклад оси ФОРМА в контраст `A/F3` равен +0.06, то есть нулю. Разложение по осям печатается, чтобы
это было видно, а не заявлено.
⚠ ФИКСЕР — `gpt-5.6-luna` с `reasoning_effort="none"`. Ручка документирована для этой модели в
`prices.CANDIDATES` (класс `effort_none`) и отработала 30 из 30 непустыми на старой панели.
DeepSeek-флеш дешевле по выходу, но у него гашение размышления на переводческой роли запрещено
гардрейлом (эхо-мина на плотном CJK), а починка предложения — роль переводческая. Выход починки
короткий, поэтому разница в цене в абсолюте копеечная.
Запуск: --plan что будет куплено и почём, $0
--flags посчитать флаги детекторов (механика $0 + QE локально, $0)
--arms купить починки (под кассой, тег `p2b-*`)
--score свод, $0
"""
from __future__ import annotations
import json
import re
import statistics as st
import sys
import time
from pathlib import Path
REPO = Path("/home/ubuntu/projects/textmachine")
sys.path.insert(0, str(REPO / "eval" / "role_topology"))
sys.path.insert(0, str(REPO / "eval" / "editor_harness"))
import bakeoff as BO # noqa: E402
import battery as BT # noqa: E402
import buy as BUY # noqa: E402
import align as AL # noqa: E402
import detect_word as DW # noqa: E402
import guarded as G # noqa: E402
from prices import cost # noqa: E402
OUT = Path.home() / "books" / "role-topology"
FIXER = "gpt-5.6-luna"
CEIL_P2B = 1.20 # из незанятых $2.99 фазы Ф2б
LED_P2B = BUY.Ledger("Ф2б армы C/E", CEIL_P2B, ("p2b-*.json",), default_expect=0.004)
FAMILY = [
("C_real", "F3", "покупает ли точечный ремонт что-то поверх черновика"),
("C_real", "A_law", "точечный ремонт против полного переписывания"),
("E", "D_", "покупает ли добивка что-то поверх сильной однопроходки"),
("E", "A_law", "обратная связка против прямой"),
("C_oracle", "C_real", "разрыв потолка и пола схемы"),
]
_SENT = re.compile(r"[^.!?…]*[.!?…]+[»\"']*\s*|[^.!?…]+$")
def sentences(text: str) -> list[str]:
"""Разбиение на предложения. Единица починки — предложение: фиксер правит его целиком."""
return [s for s in (m.group(0) for m in _SENT.finditer(text)) if s.strip()]
_BANK = None
def word_flag(sent: str) -> bool:
"""Детектор Ф0.4 №1 «выдуманное слово» на предложении. ⚠ Без него пол арма C был бы занижен
нечестно: механика класс k1 не видит ПО ПОСТРОЕНИЮ (она смотрит письменность и разметку),
и «реальные детекторы» без словарного оказались бы не тем набором, что заказан промтом."""
global _BANK # noqa: PLW0603
if _BANK is None:
_BANK = DW.load_bank()
for w in re.findall(r"\b[а-яё]{5,}\b", sent.lower()):
bad, _why = DW.verdict(w, _BANK)
if bad:
return True
return False
def mech_flags(sent: str) -> list[str]:
"""Механические детекторы на ОДНОМ предложении. Каждый — класс брака вывода."""
out = []
f = G.flags(sent)
if f["ханьцзы"]:
out.append("ханьцзы")
if f["разметка"]:
out.append("разметка")
if f["латиница"]:
out.append("латиница")
if f["преамбула"]:
out.append("преамбула")
if word_flag(sent):
out.append("слово")
return out
TYPE = {
"ханьцзы": "непереведённый фрагмент исходной письменности остался в русском тексте",
"разметка": "служебная markdown-разметка протекла в тело книги",
"латиница": "непереведённое слово латиницей осталось в русском тексте",
"преамбула": "служебная фраза модели вместо текста книги",
"слово": "в предложении есть слово, которого нет в русском языке (модель его выдумала)",
"qe": "предложение переведено неверно по смыслу (ранкер качества дал наихудший балл в окне)",
}
# ═══════════════════ РЕЖИМ ORACLE — потолок схемы, буква промта стр.114 ═══════════════════
# Промт требует ДВА режима, и разрыв между ними пре-рег называл главным результатом арма C:
# сколько схема теряет НЕ из-за идеи точечного ремонта, а из-за несовершенства детекторов.
# Земля известна по построению: в черновик сажаются дефекты двух классов на известные позиции.
# k1 ВЫДУМАННОЕ СЛОВО — известное слово портится вставкой буквосочетания; форма отсутствует
# в словаре, поэтому словарный детектор обязан её видеть.
# k2 ИНВЕРЗИЯ ПОЛЯРНОСТИ — снятие или вставка «не» при личном глаголе (`align.flip_polarity`,
# генератор проверен в `qe_power.py`). Смысл меняется на обратный, грамматичность цела,
# механические детекторы слепы по построению — её обязан ловить QE.
# ORACLE чинит РОВНО посаженные предложения. РЕАЛЬНЫЙ ищет их детекторами и чинит найденное.
def corrupt(word: str) -> str:
"""Порча слова до несуществующей формы: удвоение внутреннего слога."""
return word[:2] + word[1:3] + word[2:] if len(word) >= 5 else word + "ль"
def plant(text: str, seed: int) -> tuple[str, list[dict]]:
"""Посадить по одному дефекту каждого класса. Возвращает текст и ЗЕМЛЮ (известный ответ)."""
import random as _r # noqa: PLC0415
rnd = _r.Random(seed)
sents = sentences(text)
marks: list[dict] = []
idx = [i for i, s in enumerate(sents) if len(s.strip()) >= 60 and not mech_flags(s)]
rnd.shuffle(idx)
for i in idx:
if any(m["cls"] == "k1" for m in marks):
break
ws = [w for w in re.findall(r"\b[а-яё]{6,}\b", sents[i]) if w.isalpha()]
if not ws:
continue
w = ws[len(ws) // 2]
sents[i] = sents[i].replace(w, corrupt(w), 1)
marks.append(dict(cls="k1", i=i, orig=w, bad=corrupt(w)))
for i in idx:
if any(m["cls"] == "k2" for m in marks) or any(m["i"] == i for m in marks):
continue
flipped = AL.flip_polarity(sents[i])
if flipped and flipped != sents[i]:
sents[i] = flipped
marks.append(dict(cls="k2", i=i))
break
return "".join(sents), marks
def cmd_oracle() -> None:
"""Посадить дефекты, померить RECALL детекторов и купить починку в ОБОИХ режимах."""
from qe_bench import QE # noqa: PLC0415
qe = QE()
rows = []
for k, u in enumerate(BO.units()):
uid = u["uid"]
bad, marks = plant(BO.text3_of("F3", u), seed=1000 + k)
if len(marks) < 2:
print(f" {uid[:8]}: посадить оба класса не удалось ({len(marks)}), единица не берётся")
continue
sents = sentences(bad)
# РЕАЛЬНЫЕ детекторы на ПОСАЖЕННОМ тексте
found_mech = {i for i, s in enumerate(sents) if mech_flags(s)}
zh, ru = AL.split_zh(u["source"]), AL.split_ru(bad)
pairs = AL.align(zh, ru)
scored = []
for si, di in pairs:
s = " ".join(zh[i] for i in si).strip()
d = " ".join(ru[i] for i in di).strip()
if not s or not d or len(d) < 40:
continue
sc = qe.score(s, d)
for i, x in enumerate(sents):
if x.strip() and x.strip()[:24] in d:
scored.append((sc, i))
worst = {i for _sc, i in sorted(scored, reverse=True)[:3]}
found = found_mech | worst
hit = {m["cls"]: (m["i"] in found) for m in marks}
rows.append(dict(uid=uid, marks=marks, found=sorted(found), hit=hit))
(OUT / f"p2b-oracle-{uid}.json").write_text(json.dumps(
dict(uid=uid, planted=bad, marks=marks, found=sorted(found), hit=hit),
ensure_ascii=False), encoding="utf-8")
print(f" {uid[:8]}: посажено {len(marks)} · детекторы нашли "
f"k1={hit.get('k1')} k2={hit.get('k2')}", flush=True)
if rows:
for cls in ("k1", "k2"):
n = sum(1 for r in rows if cls in r["hit"])
k = sum(1 for r in rows if r["hit"].get(cls))
print(f"\nRECALL детекторов, класс {cls}: {k}/{n}")
def flags_path(arm: str) -> Path:
return OUT / f"p2b-flags-{arm}.json"
def cmd_flags() -> None:
"""Детекторы, $0. Механика мгновенна; QE — локальная модель на CPU, поэтому кэшируется.
⚠ Порядок важен: сначала дешёвая механика по ВСЕМ предложениям, затем QE только по тем, что
механика не флагнула. Так QE-вызовов меньше, а покрытие то же.
"""
from qe_bench import QE # noqa: PLC0415
qe = None
for arm, base in (("C", "F3"), ("E", "D_")):
p = flags_path(arm)
got = json.loads(p.read_text(encoding="utf-8")) if p.exists() else {}
for u in BO.units():
uid = u["uid"]
if uid in got:
continue
txt = BO.text3_of(base, u)
sents = sentences(txt)
rows = [dict(i=i, sent=s, mech=mech_flags(s), qe=None) for i, s in enumerate(sents)]
# ⚠ QE СЧИТАЕТСЯ НА ВЫРОВНЕННОЙ ПАРЕ, а не «предложение против всего окна». Поймано
# верификацией ДО траты: подача целого окна как исходника стоила 6.8 с на вызов
# (128 минут на арм) и мерила не то — предложение сравнивалось с текстом, которому
# оно не соответствует. Выравнивание по Гейлу–Чёрчу даёт 0.65 с и правильную пару.
zh, ru = AL.split_zh(u["source"]), AL.split_ru(txt)
pairs = AL.align(zh, ru)
if qe is None:
qe = QE()
n_qe = 0
for si, di in pairs:
s = " ".join(zh[i] for i in si).strip()
d = " ".join(ru[i] for i in di).strip()
if not s or not d or len(d) < 40:
continue
sc = qe.score(s, d)
n_qe += 1
# балл пары кладётся на ВСЕ предложения `sentences`, которые она покрывает
for r in rows:
if r["sent"].strip() and r["sent"].strip()[:24] in d:
r["qe"] = sc if r["qe"] is None else max(r["qe"], sc)
got[uid] = rows
p.write_text(json.dumps(got, ensure_ascii=False), encoding="utf-8")
print(f" {arm} {uid[:8]}: предложений {len(rows)} · механикой флагнуто "
f"{sum(1 for r in rows if r['mech'])} · QE-пар {n_qe} · "
f"с баллом {sum(1 for r in rows if r['qe'] is not None)}", flush=True)
print("флаги готовы")
def picked(arm: str, uid: str, k_qe: int = 3) -> list[dict]:
"""Что чинить: ВСЁ, что флагнула механика, плюс k худших по QE (реальный режим).
⚠ Кап на QE объявлен ЗАРАНЕЕ и равен 3 предложениям на единицу. Без капа фиксер правил бы
половину текста, и «точечный ремонт» превратился бы в переписывание — то есть в другой арм.
"""
rows = json.loads(flags_path(arm).read_text(encoding="utf-8"))[uid]
out = [dict(r, why=r["mech"][0]) for r in rows if r["mech"]]
scored = sorted((r for r in rows if r["qe"] is not None), key=lambda r: -r["qe"])[:k_qe]
out += [dict(r, why="qe") for r in scored]
return sorted(out, key=lambda r: r["i"])
def fix_one(source: str, sent: str, why: str, tag: str) -> dict:
"""Один вызов фиксера. Правит ОДНО предложение, видя исходник окна."""
sys_msg = ("Ты — корректор художественного перевода. Тебе дан ИСХОДНИК фрагмента на китайском "
"и ОДНО предложение русского перевода, в котором найден дефект. Верни ТОЛЬКО "
"исправленное предложение, без пояснений, без кавычек-обёрток. Если дефекта нет — "
"верни предложение без изменений. Ничего, кроме этого предложения, не выводи.")
usr = f"ДЕФЕКТ: {TYPE[why]}\n\nИСХОДНИК:\n{source}\n\nПРЕДЛОЖЕНИЕ:\n{sent}"
return BUY.purchase(LED_P2B, tag, FIXER, BO.client(FIXER),
dict(model=FIXER, messages=[{"role": "system", "content": sys_msg},
{"role": "user", "content": usr}],
max_completion_tokens=2000, reasoning_effort="none"),
why=why, tag_kind="fix")
def cmd_arms() -> None:
"""Купить починки. Кап 2 итерации; ре-гейт после каждой — МЕХАНИЧЕСКИЙ (см. оговорку ниже)."""
for arm, base in (("C", "F3"), ("E", "D_")):
for u in BO.units():
uid = u["uid"]
text = BO.text3_of(base, u)
sents = sentences(text)
todo = picked(arm, uid)
for it in (1, 2):
if not todo:
break
for r in todo:
tag = f"p2b-{arm}-{uid}-s{r['i']}-i{it}"
rec = fix_one(u["source"], sents[r["i"]], r["why"], tag)
if rec.get("skipped"):
print("⛔ потолок Ф2б")
return
new = (rec.get("content") or "").strip()
if new:
# ⚠ ХВОСТОВОЙ ПРОБЕЛЬНЫЙ СУФФИКС СОХРАНЯЕТСЯ. Первая версия ставила
# `new + " "` и затирала перевод строки: 28 из 65 отобранных предложений
# несут `\n`/`\n\n` в хвосте, то есть харнесс СВОИМИ РУКАМИ схлопнул бы
# абзацы и впрыснул дефект оси ФОРМА ровно в контрасты C↔A_law и C↔F3.
# Поймано верификацией ДО покупки.
old = sents[r["i"]]
suf = old[len(old.rstrip()):] or " "
sents[r["i"]] = new + suf
time.sleep(0.05)
# РЕ-ГЕЙТ после итерации. ⚠ Он МЕХАНИЧЕСКИЙ: QE локален и бесплатен, но его
# пере-прогон требует пере-выравнивания всего окна, а фиксы локальны. Ограничение
# объявлено: смысловую ошибку, ВНЕСЁННУЮ самим фиксером, этот гейт не увидит.
# Клейм «полный ре-гейт» из первой версии докстринга снят как завышенный.
todo = [dict(i=i, sent=s, why=mech_flags(s)[0])
for i, s in enumerate(sents) if mech_flags(s)]
(OUT / f"p2b-{arm}-{uid}.json").write_text(json.dumps(
dict(arm=arm, uid=uid, base=base, content="".join(sents)),
ensure_ascii=False), encoding="utf-8")
print(f" {arm} {uid[:8]} готов", flush=True)
print(f"куплено · касса ${LED_P2B.spent():.5f} из ${CEIL_P2B}")
def text_of(arm: str, uid: str) -> str:
f = OUT / f"p2b-{arm}-{uid}.json"
return json.loads(f.read_text(encoding="utf-8"))["content"] if f.exists() else ""
def cmd_score() -> None:
"""Свод детерминированной части, $0. Судейство — отдельным шагом через `absjudge`.
⚠ Здесь НЕТ судейских чисел намеренно: тексты C и E новые, их никто не судил. Чтобы они
были сравнимы с уже отснятыми армами, их надо пропустить через ТОТ ЖЕ риг (`absjudge`),
а не мерить чем-то своим — иначе контрасты окажутся несравнимы, и покупка обесценится.
Это и есть шаг после `--arms`.
"""
us = BO.units()
have = {a: [u for u in us if text_of(a, u["uid"]).strip()] for a in ("C", "E")}
print(f"{'арм':6s}{'единиц':>8s}{'знаков':>9s}{'абзацев':>9s}{'канон':>8s}{'механика':>10s}")
for a, base in (("F3", None), ("A_law", None), ("D_", None), ("C", "F3"), ("E", "D_")):
txts = ([text_of(a, u["uid"]) for u in have[a]] if a in have
else [BO.text3_of(a, u) for u in us])
txts = [x for x in txts if x.strip()]
if not txts:
print(f"{a:6s}{'нет':>8s}")
continue
s = k = 0
for u, x in zip(us, txts):
for zh, (_1, _2, r1, _3) in BO.IP.TERMS.items():
n = u["source"].count(zh)
if n:
s += n
k += min(len(re.findall(r1, x, re.I)), n)
mech = sum(1 for x in txts for sn in sentences(x) if mech_flags(sn))
print(f"{a:6s}{len(txts):>8d}{st.median([len(x) for x in txts]):>9.0f}"
f"{st.median([len([q for q in x.split(chr(10)) if q.strip()]) for x in txts]):>9.0f}"
f"{(k / s if s else 0):>8.3f}{mech:>10d}")
print("\nЦЕНА (черновик/база + починка):")
for a, base in (("C", "F3"), ("E", "D_")):
fix = sum(json.loads(f.read_text(encoding="utf-8")).get("cost_usd") or 0.0
for f in OUT.glob(f"p2b-{a}-*-s*.json"))
n = len(have[a]) or 1
if base == "F3":
b = st.mean([sum(json.loads(f.read_text(encoding="utf-8")).get("cost_usd") or 0.0
for f in OUT.glob(f"bo2-DRAFT-{u['uid']}*.json")) for u in us])
else:
b = st.mean([json.loads((OUT / f"bo2-{base}-{u['uid']}.json").read_text(
encoding="utf-8"))["cost_usd"] for u in us])
print(f" {a:3s} база ${b:.5f} + починка ${fix / n:.5f} = ${b + fix / n:.5f}/единицу")
al = st.mean([sum(json.loads(f.read_text(encoding="utf-8")).get("cost_usd") or 0.0
for f in OUT.glob(f"bo2-DRAFT-{u['uid']}*.json"))
+ json.loads((OUT / f"bo4-A_law-{u['uid']}.json").read_text(
encoding="utf-8"))["cost_usd"] for u in us])
print(f" A_law полная связка ${al:.5f}/единицу — база сравнения")
def cmd_plan() -> None:
us = BO.units()
tot_s = tot_f = 0
for arm, base in (("C", "F3"), ("E", "D_")):
if not flags_path(arm).exists():
print(f" {arm}: флаги не посчитаны — сначала --flags")
continue
n = sum(len(picked(arm, u["uid"])) for u in us)
tot_f += n
tot_s += sum(len(sentences(BO.text3_of(base, u))) for u in us)
print(f" арм {arm} (база {base}): предложений {tot_s} · к починке {n}")
print(f"\nвызовов фиксера ≈ {tot_f} (плюс до {tot_f} на второй итерации)")
print(f"ожидаемая цена ≈ ${tot_f * 2 * 0.0004:.4f} из потолка ${CEIL_P2B}")
print(f"касса `p2b-*` · потрачено ${LED_P2B.spent():.5f}")
if __name__ == "__main__":
a = sys.argv[1:] or ["--plan"]
{"--plan": cmd_plan, "--flags": cmd_flags, "--arms": cmd_arms,
"--oracle": cmd_oracle, "--score": cmd_score}.get(
a[0], lambda: print(__doc__))()