206 lines
12 KiB
Python
206 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""Ф2б, арм C — «гейты флагают → точечный ремонт», в ДВУХ режимах: oracle и реальные детекторы.
|
||
|
||
Что решается. Эксп-20 §3.3 намерил, что починка по флагу снимает дефект за $0.0002 — но мерил при
|
||
ИДЕАЛЬНОЙ детекции: флаги брались из посадок. Промт эксп-21 (строка 13) назвал это главной
|
||
оговоркой механизма, а пре-рег — главным результатом арма C: **разрыв между потолком (oracle) и
|
||
полом (реальные детекторы)**. Здесь этот разрыв меряется числом.
|
||
|
||
Материал и земля. Те же 8 боевых единиц и те же ЗАМОРОЖЕННЫЕ черновики, что в Ф2а. В каждый
|
||
черновик сажается по одному дефекту каждого класса, и эталон известен по построению:
|
||
к1 ВЫДУМАННОЕ СЛОВО — известное слово портится фиксированным правилом (замена внутреннего
|
||
буквосочетания). Земля объективна: испорченная форма отсутствует в морфологическом словаре.
|
||
к2 ИНВЕРСИЯ ПОЛЯРНОСТИ — снятие или вставка отрицания при личном глаголе (генератор из
|
||
`qe_power.py`, там же проверен). Смысл меняется на противоположный, грамматичность цела.
|
||
|
||
Режимы:
|
||
ORACLE — фиксеру дают ТОЧНОЕ предложение и тип ошибки. Это потолок схемы.
|
||
РЕАЛЬНЫЙ — дефекты ищут детекторы Ф0.4 и $0-гейты батареи; чинится только найденное. Это пол.
|
||
|
||
⚠ Что здесь заложено конструкцией и должно читаться именно так. Класс к1 виден словарной проверке
|
||
по построению, поэтому высокий recall на нём — не заслуга детектора, а свойство задачи. Класс к2
|
||
детектора не имеет ВООБЩЕ (§2.2: QE-ранкер локальную инверсию не берёт ни на одной из двух
|
||
моделей, при том что декой ловит) ⇒ в реальном режиме он не находится никогда. Разрыв
|
||
oracle↔реальность и есть этот класс, и его доля — 50% посаженного. Число будет ровным не потому,
|
||
что замер грубый, а потому, что дыра ровно такой формы.
|
||
|
||
Запуск: eval/.venv/bin/python eval/role_topology/repair_arm.py --plan # план и земля, $0
|
||
eval/.venv/bin/python eval/role_topology/repair_arm.py --run
|
||
"""
|
||
from __future__ import annotations
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
REPO = Path("/home/ubuntu/projects/textmachine")
|
||
sys.path.insert(0, str(REPO / "eval" / "editor_contract"))
|
||
sys.path.insert(0, str(REPO / "eval" / "bank_arbitration"))
|
||
sys.path.insert(0, str(REPO / "eval" / "editor_harness"))
|
||
sys.path.insert(0, str(REPO / "eval" / "role_topology"))
|
||
|
||
from dotenv import load_dotenv # noqa: E402
|
||
from openai import OpenAI # noqa: E402
|
||
|
||
import bakeoff as BO # noqa: E402
|
||
import detect_word as DW # noqa: E402
|
||
import editor_wire_probe as P # noqa: E402
|
||
from prices import CANDIDATES, cost # noqa: E402
|
||
from align import flip_polarity # noqa: E402
|
||
|
||
load_dotenv(REPO / "eval" / ".env")
|
||
OUT = Path.home() / "books" / "role-topology"
|
||
CEILING_USD_МЁРТВ = None # ⚠ объявлен и НИКЕМ не читается;
|
||
# оставлен видимым маркером: охраны здесь нет, покупки идут мимо кассы (ревью 07.08)
|
||
FIXER = "gpt-5.6-luna" # дешёвый тир, прошёл скрин Ф1, без стены размышления при low
|
||
TPL = REPO / "eval" / "editor_harness" / "prompts" / "repair.md"
|
||
TYPE = {"k1": "несуществующее слово (модель выдумала слово, которого в русском языке нет)",
|
||
"k2": "смысловое искажение против исходника (сказано обратное или иное, чем в оригинале)"}
|
||
|
||
|
||
def sentences(text: str) -> list[str]:
|
||
return [s for s in re.split(r"(?<=[.!?…])\s+", text) if s.strip()]
|
||
|
||
|
||
def corrupt_word(sent: str) -> tuple[str, str, str] | None:
|
||
"""Портит одно длинное известное слово фиксированным правилом. Возвращает (новое предложение,
|
||
испорченное слово, исходное слово).
|
||
|
||
Правило детерминированное и НЕ подстроено под детектор: берётся самое длинное словарное слово
|
||
предложения и его 4–5-й знаки заменяются на «яв». Земля — объективная: получившаяся форма
|
||
отсутствует в морфологическом словаре.
|
||
"""
|
||
cands = sorted((w for w in re.findall(r"[А-Яа-яЁё]{8,}", sent) if DW.known(w)),
|
||
key=len, reverse=True)
|
||
for w in cands:
|
||
bad = w[:3] + "яв" + w[5:]
|
||
if DW.known(bad) or bad == w:
|
||
continue
|
||
return sent.replace(w, bad, 1), bad, w
|
||
return None
|
||
|
||
|
||
def plant(draft: str) -> tuple[str, list[dict]]:
|
||
"""Сажает по одному дефекту каждого класса в РАЗНЫЕ предложения. Эталон возвращается рядом."""
|
||
ss = sentences(draft)
|
||
out = draft
|
||
marks: list[dict] = []
|
||
used: set[int] = set()
|
||
for i, s in enumerate(ss):
|
||
if "k1" in {m["cls"] for m in marks}:
|
||
break
|
||
c = corrupt_word(s)
|
||
if c:
|
||
new_s, bad, orig = c
|
||
out = out.replace(s, new_s, 1)
|
||
marks.append(dict(cls="k1", sentence=new_s, bad=bad, orig=orig))
|
||
used.add(i)
|
||
break
|
||
for i, s in enumerate(ss):
|
||
if i in used or "k2" in {m["cls"] for m in marks}:
|
||
continue
|
||
f = flip_polarity(s)
|
||
if f and f != s:
|
||
out = out.replace(s, f, 1)
|
||
marks.append(dict(cls="k2", sentence=f, bad=None, orig=s))
|
||
break
|
||
return out, marks
|
||
|
||
|
||
def detect(text: str, bank: frozenset) -> list[dict]:
|
||
"""РЕАЛЬНЫЕ детекторы: словный детектор Ф0.4 + $0-гейты батареи. Инверсий здесь нет и быть
|
||
не может — §2.2 показала, что инструмента для них не существует."""
|
||
found = []
|
||
for s in sentences(text):
|
||
for w in re.findall(r"[А-Яа-яЁё]{4,}", s):
|
||
if DW.verdict(w, bank)[0]:
|
||
found.append(dict(cls="k1", sentence=s, word=w))
|
||
break
|
||
return found
|
||
|
||
|
||
def fix(cl, source: str, sentence: str, cls: str, tag: str) -> dict:
|
||
f = OUT / f"rp-{tag}.json"
|
||
if f.exists():
|
||
return json.loads(f.read_text(encoding="utf-8"))
|
||
canon = P.strip_comments(TPL.read_text(encoding="utf-8"))
|
||
sys_part, user = canon.split(P.USER_SEP, 1)
|
||
flagged = f"{TYPE[cls]}\n\nПроблемное предложение перевода:\n{sentence}"
|
||
sys_msg = P.render(sys_part.split(P.FEWSHOT_SEP, 1)[0].strip(), source, flagged)
|
||
usr = P.render(user.strip(), source, flagged)
|
||
r = cl.chat.completions.create(model=FIXER,
|
||
messages=[{"role": "system", "content": sys_msg},
|
||
{"role": "user", "content": usr}],
|
||
max_completion_tokens=2000, reasoning_effort="none")
|
||
u = r.usage
|
||
cached = getattr(getattr(u, "prompt_tokens_details", None), "cached_tokens", 0) or 0
|
||
rec = dict(tag=tag, content=r.choices[0].message.content or "",
|
||
cost_usd=round(cost(FIXER, u.prompt_tokens or 0, cached,
|
||
u.completion_tokens or 0), 6))
|
||
f.write_text(json.dumps(rec, ensure_ascii=False), encoding="utf-8")
|
||
return rec
|
||
|
||
|
||
def main() -> None:
|
||
plan_only = "--plan" in sys.argv
|
||
bank = DW.load_bank()
|
||
us = BO.units()
|
||
planted = [plant(u["draft"]) for u in us]
|
||
n_k1 = sum(1 for _, m in planted for x in m if x["cls"] == "k1")
|
||
n_k2 = sum(1 for _, m in planted for x in m if x["cls"] == "k2")
|
||
print(f"единиц {len(us)} · посажено к1 (выдуманное слово) {n_k1} · к2 (инверсия) {n_k2}")
|
||
|
||
# ЧТО НАХОДЯТ РЕАЛЬНЫЕ ДЕТЕКТОРЫ — считается до и независимо от починки.
|
||
rec_k1 = rec_k2 = 0
|
||
for (txt, marks), u in zip(planted, us):
|
||
hits = detect(txt, bank)
|
||
for m in marks:
|
||
if m["cls"] == "k1":
|
||
rec_k1 += any(h["sentence"] == m["sentence"] for h in hits)
|
||
else:
|
||
rec_k2 += any(h["cls"] == "k2" for h in hits)
|
||
print(f"РЕАЛЬНЫЕ детекторы нашли: к1 {rec_k1}/{n_k1} · к2 {rec_k2}/{n_k2}")
|
||
if plan_only:
|
||
return
|
||
|
||
cl = OpenAI(api_key=os.environ[CANDIDATES[FIXER][1]], base_url=CANDIDATES[FIXER][0],
|
||
timeout=300)
|
||
spent = 0.0
|
||
res = {"oracle": [0, 0], "real": [0, 0]} # [починено, предъявлено]
|
||
for ui, ((txt, marks), u) in enumerate(zip(planted, us)):
|
||
hits = detect(txt, bank)
|
||
for m in marks:
|
||
# ORACLE: спан известен точно.
|
||
rec = fix(cl, u["source"], m["sentence"], m["cls"], f"or-u{ui}-{m['cls']}")
|
||
spent += rec["cost_usd"]
|
||
ok = (m["bad"] not in rec["content"]) if m["cls"] == "k1" else \
|
||
(rec["content"].strip() != m["sentence"].strip())
|
||
res["oracle"][1] += 1
|
||
res["oracle"][0] += bool(ok)
|
||
# РЕАЛЬНЫЙ: чиним только то, что нашли детекторы.
|
||
if any(h["sentence"] == m["sentence"] for h in hits):
|
||
rec2 = fix(cl, u["source"], m["sentence"], m["cls"], f"re-u{ui}-{m['cls']}")
|
||
spent += rec2["cost_usd"]
|
||
ok2 = (m["bad"] not in rec2["content"]) if m["cls"] == "k1" else \
|
||
(rec2["content"].strip() != m["sentence"].strip())
|
||
res["real"][0] += bool(ok2)
|
||
res["real"][1] += 1
|
||
time.sleep(0.15)
|
||
|
||
print(f"\n{'режим':28s}{'снято дефектов':>17s}{'доля':>8s}")
|
||
print("-" * 55)
|
||
for name, (ok, tot) in res.items():
|
||
label = "ORACLE (идеальный флаг)" if name == "oracle" else "РЕАЛЬНЫЕ детекторы"
|
||
print(f"{label:28s}{ok:>8d}/{tot:<8d}{ok / max(1, tot):8.0%}")
|
||
gap = res["oracle"][0] - res["real"][0]
|
||
print(f"\nРАЗРЫВ oracle↔реальность: {gap} дефектов из {res['oracle'][1]} = "
|
||
f"{gap / max(1, res['oracle'][1]):.0%}")
|
||
print(f"цена починки: ${spent / max(1, res['oracle'][1] + res['real'][0]):.6f} за дефект · "
|
||
f"всего ${spent:.6f}")
|
||
(OUT / "repair-arm.json").write_text(json.dumps(res, ensure_ascii=False), encoding="utf-8")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|