Freeze the second-edition role-topology harness before paid calls: mirrored judging, arms matching the prompt letter, shared purchase ledger, repaired battery

This commit is contained in:
heaven 2026-08-07 02:13:17 +03:00
parent 864038a414
commit d4725999f2
15 changed files with 2285 additions and 75 deletions

View file

@ -4,10 +4,27 @@
Вынесено из `qe_segment.py` по находке самопроверки: выравнивание нужно и батарее Ф0.5, и
QE-детектору, но батарея не должна тянуть за собой 4.6 ГБ весов и отдельный venv. Модуль
импортируется обоими и проверяется `selfcheck.py` в базовом окружении полигона.
Сюда же вынесен генератор ИНВЕРСИЙ ПОЛЯРНОСТИ: он тоже детерминированный, тоже $0 и тоже нужен
двум разным потребителям (замеру мощности QE и арму починки Ф2б). Первая редакция держала его в
`qe_power.py`, который тянет torch, и арм починки падал на импорте 4.6 ГБ весов, которые ему не
нужны вовсе. Тот же класс дефекта, что уже был пойман самопроверкой на выравнивании.
"""
from __future__ import annotations
import re
import pymorphy3
_MORPH = pymorphy3.MorphAnalyzer()
def _is_finite_verb(word: str) -> bool:
for p in _MORPH.parse(word.lower()):
if p.is_known and p.tag.POS == "VERB":
return True
return False
# Терминаторы предложений: китайские полноширинные и латинские. Многоточие 。。。/…… не разрывается.
_ZH_SPLIT = re.compile(r"(?<=[。!?;…])(?![。!?;…])")
_RU_SPLIT = re.compile(r"(?<=[.!?…])\s+(?=[«\"(\[А-ЯЁA-Z—-])")
@ -76,3 +93,38 @@ def align(src: list[str], dst: list[str]) -> list[tuple[list[int], list[int]]]:
return out[::-1]
def flip_polarity(sent: str) -> str | None:
"""Инвертирует полярность предложения: снимает «не» при глаголе либо вставляет его.
Работает только по ЛИЧНОМУ глаголу или глаголу прошедшего времени причастия и деепричастия
исключены сознательно: «не» при них меняет смысл иначе и часто даёт неграмматичный результат,
а неграмматичность ранкер поймает не как инверсию, а как порчу формы (другой класс).
"""
m = re.search(r"\е\s+([А-Яа-яЁё]+)", sent)
if m and _is_finite_verb(m.group(1)):
return sent[:m.start()] + m.group(1) + sent[m.end():]
for m in re.finditer(r"\b([А-Яа-яЁё]{3,})\b", sent):
w = m.group(1)
if sent[max(0, m.start() - 4):m.start()].strip().endswith("не"):
continue
# ⚠ Пропускаем глагол с ПРОПИСНОЙ: он открывает предложение или реплику, и вставка даёт
# «не Стой» — неграмматичный текст. Ранкер оценил бы его как порчу ФОРМЫ, то есть замер
# мерил бы другой класс дефекта. Поймано проверкой генератора до прогона.
if w[0].isupper():
continue
if _is_finite_verb(w):
return sent[:m.start()] + "не " + sent[m.start():]
return None
def _is_finite_verb(word: str) -> bool:
for p in _MORPH.parse(word.lower()):
if not p.is_known:
continue
if p.tag.POS == "VERB":
return True
return False

View file

@ -0,0 +1,460 @@
#!/usr/bin/env python3
"""Ф2а — БЕЙК-ОФФ «переписывание как класс». Отвечает на главный вопрос пака.
Вопрос владельца, ради которого существует эксп: нужен ли второй проход вообще, и не выгоднее ли
переводить сразу сильной моделью. Экспы 1820 подошли к нему сбоку (контракт редактора, цена
диффа, роль редактора), но самого бейк-оффа топологий не гоняли ни разу.
Армы. Черновики ФРИЗЯТСЯ и подаются идентичными всем редакторским армам это парный дизайн, и он
даёт кратный выигрыш мощности бесплатно. Черновики берутся готовые из корпуса пакета-6 (их сделал
боевой `deepseek-v4-flash`), поэтому арм F и вход армов A/B стоят $0.
F do-nothing пол: черновик как есть. Все редакторские армы мерятся ПРОТИВ НЕГО, а не друг
против друга иначе «B лучше A» ничего не говорит о том, нужен ли второй проход вообще.
A боевой бейзлайн: черновик `deepseek-v4-pro`, полное переписывание.
B то же, но редактором `glm-5` (эксп-20 нашёл, что у него второй проход отрабатывает,
а у dspro вредит здесь это проверяется на боевых единицах и парно).
D однопроходка: `deepseek-v4-pro` переводит ИСХОДНИК с нуля, черновика не видит.
D она же БЕЗ мандата перевёрстки деконфаунд «модель против мандата».
Протокол сравнения задан замером Ф0.6, а не вкусом: судьи не выдумывают перевес на идентичных
текстах (0 из 18), но их АБСОЛЮТНЫЙ счёт ошибок плавает сравниваем только ПАРНО, зеркальной
раскладкой, и вердикт засчитываем при перевесе в 2 повторах из 3 у обоих судей.
Контрасты выбраны так, чтобы каждый отвечал на отдельный вопрос, а не наращивал число сравнений:
A против F покупает ли боевой второй проход хоть что-то поверх черновика;
B против F покупает ли его альтернативный жилец;
D против A дешевле ли и лучше ли одним сильным проходом, чем связкой;
D против D что из разницы даёт МАНДАТ, а что модель.
Запуск: eval/.venv/bin/python eval/role_topology/bakeoff.py --arms # породить выходы армов
eval/.venv/bin/python eval/role_topology/bakeoff.py --judge # судейство контрастов
eval/.venv/bin/python eval/role_topology/bakeoff.py --score # свод, $0
"""
from __future__ import annotations
import hashlib
import json
import os
import re
import random
import statistics
import sys
import time
from collections import Counter, defaultdict
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 battery as B # noqa: E402
import buy as BUY # noqa: E402
import editor_wire_probe as P # noqa: E402
import inject_probe as IP # noqa: E402
import judges as J # noqa: E402
import probe as Q # noqa: E402
from prices import CANDIDATES, cost # noqa: E402
load_dotenv(REPO / "eval" / ".env")
OUT = Path.home() / "books" / "role-topology"
CEILING_USD = 2.50
N_UNITS = 16
# Один голос на КАЖДЫЙ порядок. Позиция гасится точно, а не «в среднем по повторам»: перевес
# единицы = среднее двух зеркальных голосов. Мощность добирается ЕДИНИЦАМИ (8→16), а не
# повторами — единицы дают и разброс, и обобщение, повторы только разброс.
REPS_PER_ORDER = 1
# ⚠ ПОТОЛКИ ВТОРОЙ РЕДАКЦИИ объявлены ДО первой покупки (санкция владельца 07.08 «деньги
# потратить разрешаю»). Первая редакция Ф2а стоила $1.3777 и признана негодной по раскладке.
CEIL_ARMS = 0.70
CEIL_JUDGE = 1.70
LED_ARMS = BUY.Ledger("армы Ф2а-2", CEIL_ARMS, ("bo2-*.json",), default_expect=0.03)
LED_JUDGE = BUY.Ledger("судейство Ф2а-2", CEIL_JUDGE, ("jv2-*.json",), default_expect=0.008)
# АРМЫ ВТОРОЙ РЕДАКЦИИ (07.08). Первая редакция отклонялась от промта в четырёх местах, и все
# четыре смещали сравнение в одну сторону — в пользу связки. Ни одно из отклонений не было
# объявлено, три нашло адверсариальное ревью, четвёртое (сила формулировки закона) — сверка
# промта с проводом при разборе находок ревью.
#
# `block`: None — банкноты нет; "law" — ПИНЕННЫЙ промтом закон-блок.
# Промт стр.29: банкнота на переводческом проходе ВСЕХ армов, КРОМЕ байт-боевого бейзлайна;
# текст закона — отправленная строка пробы 18 БЕЗ оговорки области (`HEADER_LAW_PLAIN`,
# байт-идентичен движковому `editor_header`). Первая редакция слала `HEADER_LAW_CLAUSE`,
# то есть закон С оговоркой, и слала его АРМУ A, который по промту идёт без банкноты вовсе.
# Армы D/D_ при этом получали не закон, а мягкий ГЛОССАРИЙ («используй последовательно»)
# вместо императива («ДОЛЖЕН быть передан именно указанной формой») — то есть ось ТЕРМИН
# в контрасте «D против A» мерила силу формулировки, а не топологию.
#
# `think`: "off" — гасить размышление (промт стр.62 задаёт арм B как flash→glm-5-OFF).
# Первая редакция ставила ручку только флешу, и B шёл с мышлением: 170713728 токенов
# размышления и цена ×5.1 против A.
ARMS = {
"A": dict(model="deepseek-v4-pro", contract="full", block=None, think=None),
"B": dict(model="glm-5", contract="full", block="law", think="off"),
"D": dict(model="deepseek-v4-pro", contract="direct-reflow", block="law", think=None),
"D_": dict(model="deepseek-v4-pro", contract="direct", block="law", think=None),
# Деконфаунд банкноты: тот же бейзлайн, но С законом. Отвечает на продуктовый вопрос
# «сколько покупает банк-как-закон», который первая редакция смешала с вопросом о топологии.
"A_law": dict(model="deepseek-v4-pro", contract="full", block="law", think=None),
}
CONTRASTS = [("A", "F"), ("B", "F"), ("D", "A"), ("D", "D_"), ("A_law", "A")]
# Топологии, потребляющие черновик. Их цена = черновик + собственный вызов; у однопроходок ноги
# черновика нет. ⚠ Первая редакция печатала «$/единицу» БЕЗ ноги черновика и объявила связку на
# 17% дешевле однопроходки; цена черновика при этом бралась из двух клеток скрина другой роли
# ($0.00195, разброс 3.4×), а не с этих единиц. Знак вывода зависел от того, какую из двух цифр
# подставить, — значит вывода не было вовсе. Теперь черновик покупается ЗДЕСЬ, на тех же
# единицах, кодом, который лежит в дереве.
CONSUMES_DRAFT = {"F", "A", "A_law", "B"}
DRAFT_MODEL = "deepseek-v4-flash"
OPENAI_FAMILY = {"gpt-5.6-luna"}
def units(n: int = N_UNITS) -> list[dict]:
"""N боевых единиц из корпуса пакета-6: источник + ЗАМОРОЖЕННЫЙ черновик.
Отбор по длине источника, детерминированный: берётся середина распределения, чтобы единицы
были боевого размера и при этом не выбросами. Черновик один и тот же для всех армов в этом
весь смысл парного дизайна.
ДВЕ ПОЧИНКИ 07.08 по адверсариальному ревью.
(а) ДЕДУП. В корпусе 91 запись, но лишь 63 уникальных источника: пакет-6 гонял один текст
несколькими прогонами. Прошлый отбор взял пару дублей соседними единицами, и «8 единиц»
были эффективно семью при этом дубль РАСЩЕПИЛСЯ (арм проигрывал на одной копии и
выигрывал на другой), то есть исход решал черновик, а не арм.
(б) УСТОЙЧИВЫЙ КЛЮЧ. Тег артефакта раньше нёс позиционный индекс `u{i}`, а индекс зависит от
N_UNITS: смена числа единиц молча пере-назначила бы уже купленные выходы другим текстам.
Ключ теперь хеш источника, и он не зависит ни от N, ни от порядка чтения.
"""
corp = Path.home() / "books" / "gu-zhenren" / "labels" / "raw" / "corpus.jsonl"
us = [json.loads(l) for l in corp.read_text(encoding="utf-8").splitlines() if l.strip()]
us = [u for u in us if all((u.get(k) or "").strip() for k in ("source", "draft"))]
uniq: dict[str, dict] = {}
by_sig: dict[str, str] = {}
for u in us:
sig = _dedup_sig(u["source"])
if sig in by_sig:
continue # тот же текст под другим заголовком главы
by_sig[sig] = uid_of(u["source"])
uniq.setdefault(uid_of(u["source"]), u)
keys = sorted(uniq, key=lambda k: len(uniq[k]["source"]))
lo = (len(keys) - n) // 2
out = []
for k in keys[lo:lo + n]:
u = dict(uniq[k])
u["uid"] = k
out.append(u)
return out
def uid_of(source: str) -> str:
return hashlib.sha1(source.strip().encode("utf-8")).hexdigest()[:10] # noqa: S324
_RE_CHAPTER = re.compile(r"^\s*第[零一二两三四五六七八九十百千\d]+[节章回]\s*[:]?\s*")
def _dedup_sig(source: str) -> str:
"""Подпись для БЛИЗКИХ дублей. Точного хеша мало: u3 и u4 прошлого прогона были одним
текстом, у которого одна копия несла заголовок `第八节`, а другая нет близость 0.999,
хеши разные. Подпись снимает пробелы и ведущий заголовок главы."""
s = _RE_CHAPTER.sub("", source.strip())
return hashlib.sha1("".join(s.split()).encode("utf-8")).hexdigest() # noqa: S324
def spent(prefix: str) -> float:
tot = 0.0
for f in OUT.glob(f"{prefix}*.json"):
try:
tot += json.loads(f.read_text(encoding="utf-8")).get("cost_usd", 0.0)
except Exception: # noqa: BLE001, S112
continue
return tot
def client(model: str) -> OpenAI:
base, env, *_ = CANDIDATES[model]
return OpenAI(api_key=os.environ[env], base_url=base, timeout=900)
def build_msgs(arm: str, u: dict) -> list[dict]:
"""Сообщения арма собираются ЗДЕСЬ, а не в `probe.messages`.
Причина: `probe.messages` для `direct`/`direct-reflow` зашивает мягкий ГЛОССАРИЙ-блок внутри
себя, и подменить его законом снаружи было нельзя из-за чего армы получали банкноты разной
силы. Править `probe.messages` тоже нельзя: на нём стоят экспы 1820, их воспроизводимость
важнее удобства. Поэтому определение арма живёт целиком в одном видимом месте.
"""
spec, src, draft = ARMS[arm], u["source"], u["draft"]
terms = [t for t in IP.TERMS if t in src]
blk = P.editor_block(terms, P.HEADER_LAW_PLAIN, None) if (spec["block"] and terms) else None
if spec["contract"] in ("direct", "direct-reflow"):
if spec["contract"] == "direct-reflow":
sys_msg, user = Q.render(Q.HERE / "prompts" / "translator-reflow.md", src, "")
user = user.strip()
else:
sys_msg, user = Q.render_translator(src)
m = [{"role": "system", "content": sys_msg}]
if blk:
m.append({"role": "system", "content": blk})
m.append({"role": "user", "content": user})
return m
return Q.messages(spec["contract"], src, draft, blk)
def run_draft(u: dict) -> dict:
"""Черновик боевой конфигурации НА ЭТИХ ЖЕ единицах — нога цены всех связок.
Мышление НЕ гасится: гардрейл CLAUDE.md («DeepSeek НИКОГДА не отключать thinking»,
эхо-мина на плотном CJK). Первая редакция мерила черновик скриптом вне дерева, и все восемь
его записей несли `reasoning_tokens=0` то есть либо провайдер не отчитался, либо мышление
было выключено вопреки гардрейлу. Разрешить это чтением было нельзя: породивший код исчез.
"""
sys_msg, user = Q.render_translator(u["source"])
terms = [t for t in IP.TERMS if t in u["source"]]
msgs = [{"role": "system", "content": sys_msg}]
if terms:
msgs.append({"role": "system", "content": IP.block_for(terms, None, False)})
msgs.append({"role": "user", "content": user})
kw = dict(model=DRAFT_MODEL, messages=msgs, max_tokens=16000,
extra_body={"reasoning_effort": "low"})
return BUY.purchase(LED_ARMS, f"bo2-DRAFT-{u['uid']}", DRAFT_MODEL, client(DRAFT_MODEL), kw,
arm="DRAFT", uid=u["uid"])
def draft_cost(uid: str) -> float:
f = OUT / f"bo2-DRAFT-{uid}.json"
return json.loads(f.read_text(encoding="utf-8"))["cost_usd"] if f.exists() else 0.0
def run_arm(arm: str, u: dict) -> dict:
spec = ARMS[arm]
model = spec["model"]
kw: dict = dict(model=model, messages=build_msgs(arm, u), max_tokens=16000, temperature=0.4)
if model == "deepseek-v4-flash":
kw["extra_body"] = {"reasoning_effort": "low"}
if spec["think"] == "off":
# quirks 00, строка 30 (семейство GLM): гашение — `thinking: {"type": "disabled"}`.
# ⚠ DeepSeek этой веткой не гасится НИКОГДА — эхо-мина, гардрейл CLAUDE.md.
if model.startswith("glm"):
kw["extra_body"] = {"thinking": {"type": "disabled"}}
else:
raise SystemExit(f"гашение размышления у {model} не описано в quirks — не гадаю")
return BUY.purchase(LED_ARMS, f"bo2-{arm}-{u['uid']}", model, client(model), kw,
arm=arm, uid=u["uid"], contract=spec["contract"],
block=bool(spec["block"]), think=spec["think"])
def text_of(arm: str, u: dict) -> str:
if arm == "F":
return u["draft"]
f = OUT / f"bo2-{arm}-{u['uid']}.json"
return json.loads(f.read_text(encoding="utf-8"))["content"] if f.exists() else ""
def cmd_arms() -> None:
us = units()
print(f"единиц {len(us)} · армов {len(ARMS)} + черновик · потолок армов ${LED_ARMS.ceiling}")
for u in us:
rec = run_draft(u)
if rec.get("skipped"):
print("⛔ потолок — стоп")
return
print(f" DRAFT {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)
for arm in ARMS:
for u in us:
rec = run_arm(arm, u)
if rec.get("skipped"):
print("⛔ потолок — стоп")
return
print(f" {arm:6s} {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:
"""ПОЛНОЕ СКРЕЩИВАНИЕ ПОРЯДКОВ — закон промта стр.32, нарушенный первой редакцией.
Первая редакция чередовала порядок по повторам (`(rep-1) % 2` при трёх повторах = 0,1,0),
и порядок 0 весил два голоса из трёх, а правило вердикта «перевес в 2 из 3» удовлетворялось
голосами ОДНОГО порядка. Позиционная фора судьи оказалась +4.65 ошибки в пользу первого
варианта больше измеряемых эффектов; на почти идентичной паре D/D_ она составила +3.98,
то есть это свойство судьи, а не армов.
Здесь порядок внешний цикл, и на каждую клетку (контраст × единица × судья) приходится
РОВНО по одному голосу каждого порядка. Перевес единицы = среднее двух зеркальных голосов,
позиция вычитается тождественно.
"""
us = units()
print(f"единиц {len(us)} · контрастов {len(CONTRASTS)} · судей {len(J.JUDGES)} · "
f"порядков 2 × повторов {REPS_PER_ORDER}")
print(f"голосов к покупке: {len(us) * len(CONTRASTS) * len(J.JUDGES) * 2 * REPS_PER_ORDER} · "
f"потолок ${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"jv2-{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, rep=rep)
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 vote_errors(f: Path) -> tuple[int, int] | None:
"""Счёт ошибок двух вариантов из персистированного голоса. None — голос не разобран."""
if not f.exists():
return None
p = J.parse(json.loads(f.read_text(encoding="utf-8"))["content"])
if p["В1-ВЕРНОСТЬ"] is None:
return None
return (sum(p[f"В1-{ax}"] or 0 for ax in J.AXES), sum(p[f"В2-{ax}"] or 0 for ax in J.AXES))
def margins_by_order(a: str, b: str, uid: str, judge: str) -> dict[int, list[float]]:
"""Перевесы «за арм a», разложенные ПО ПОРЯДКУ. Без этой раскладки позицию не вычесть."""
out: dict[int, list[float]] = {}
for order in (0, 1):
for rep in range(1, REPS_PER_ORDER + 1):
e = vote_errors(OUT / f"jv2-{a}v{b}-{uid}-o{order}-{judge}-r{rep}.json")
if e is None:
continue
e1, e2 = e
out.setdefault(order, []).append((e2 - e1) if order == 0 else (e1 - e2))
return out
def position_bias() -> float:
"""Средняя разница «ошибки второго варианта ошибки первого» по ВСЕМ голосам.
Это контроль исправности рига, а не результат: при честном зеркале и достаточном n фора
должна быть около нуля по построению (каждая пара судится в обоих порядках). Крупное
ненулевое значение означает, что раскладка сломана ровно это и случилось с первой
редакцией (+4.65), где порядок 0 весил вдвое больше порядка 1.
"""
d = []
for f in OUT.glob("jv2-*.json"):
e = vote_errors(f)
if e:
d.append(e[1] - e[0])
return statistics.mean(d) if d else 0.0
def boot_ci(xs: list[float], reps: int = 5000, alpha: float = 0.05) -> tuple[float, float]:
"""Bootstrap-ДИ среднего по ЕДИНИЦАМ (промт стр.32 требует ДИ, первая редакция его не давала).
Ресемплинг детерминированный: сид фиксирован, иначе отчёт не воспроизводится дословно.
"""
if len(xs) < 2:
return (float("nan"), float("nan"))
rnd = random.Random(20260807)
means = []
for _ in range(reps):
means.append(statistics.mean([xs[rnd.randrange(len(xs))] for _ in range(len(xs))]))
means.sort()
return (means[int(alpha / 2 * reps)], means[int((1 - alpha / 2) * reps) - 1])
def cmd_score() -> None:
"""Свод. $0: читает персистированные голоса и прогоняет батарею по выходам армов."""
us = units()
bias = position_bias()
print(f"ПОЗИЦИОННАЯ ФОРА СУДЬИ (контроль исправности раскладки): {bias:+.2f} ошибки в пользу "
f"первого варианта\n")
print("ПАРНЫЕ ВЕРДИКТЫ. Перевес единицы = среднее ДВУХ зеркальных голосов; вердикт требует,\n"
"чтобы знак сохранился в ОБОИХ порядках у ОБОИХ судей.\n")
print(f"{'контраст':16s}{'единиц':>8s}{'за первый':>11s}{'за второй':>11s}{'ничья':>8s}"
f"{'перевес':>10s}{'95% ДИ':>18s}")
print("-" * 82)
for a, b in CONTRASTS:
wins_a = wins_b = tie = 0
per_unit: list[float] = []
for u in us:
per_judge, unit_margins = {}, []
for judge in J.JUDGES:
by_order = margins_by_order(a, b, u["uid"], judge)
if len(by_order) < 2:
continue
m0, m1 = statistics.mean(by_order[0]), statistics.mean(by_order[1])
unit_margins += [m0, m1]
# Знак ОБЯЗАН совпасть в обоих порядках: это и есть развязка позиции.
per_judge[judge] = 1 if (m0 > 0 and m1 > 0) else (-1 if (m0 < 0 and m1 < 0) else 0)
if len(per_judge) < len(J.JUDGES):
continue
per_unit.append(statistics.mean(unit_margins))
vals = set(per_judge.values())
wins_a += vals == {1}
wins_b += vals == {-1}
tie += vals not in ({1}, {-1})
n = wins_a + wins_b + tie
if not per_unit:
print(f"{a + ' против ' + b:16s}{0:>8}{'':>11}{'':>11}{'':>8}{'нет голосов':>28s}")
continue
eff = statistics.mean(per_unit)
lo, hi = boot_ci(per_unit)
star = "" if lo <= 0 <= hi else " ←ДИ не накрывает ноль"
print(f"{a + ' против ' + b:16s}{n:8d}{wins_a:11d}{wins_b:11d}{tie:8d}{eff:+10.2f}"
f" [{lo:+.2f}, {hi:+.2f}]{star}")
print("\nДЕТЕРМИНИРОВАННАЯ БАТАРЕЯ (первична — шума не имеет)\n")
print(f"{'арм':5s}{'ед.':>5s}{'типогр.':>9s}{'ханьцзы':>9s}{'потеря вел.':>13s}"
f"{'ты/вы микс':>12s}{'род свер.':>11s}{'род ошиб.':>11s}{'$/единицу':>11s}")
print("-" * 86)
cards = B.load_cards()
for arm in ("F", *ARMS):
rows, costs = [], []
for u in us:
t = text_of(arm, u)
if not t.strip():
continue
rows.append(B.run_unit(B.Unit(source=u["source"], draft=u["draft"], final=t,
cards=cards)))
c = draft_cost(u["uid"]) if arm in CONSUMES_DRAFT else 0.0
if arm != "F":
f = OUT / f"bo2-{arm}-{u['uid']}.json"
if 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}"
f"{sum(r['typography']['violations'] for r in rows):9d}"
f"{sum(r['cjk_leak']['han_chars'] for r in rows):9d}"
f"{sum(r['numbers']['lost_n'] for r in rows):13d}"
f"{sum(r['ty_vy']['mixed_in_unit'] for r in rows):12d}"
f"{sum(r['gender']['checked'] for r in rows):11d}"
f"{sum(r['gender']['mismatched'] for r in rows):11d}"
f"{(statistics.median(costs) if costs else 0):11.5f}")
if __name__ == "__main__":
if "--arms" in sys.argv:
cmd_arms()
elif "--judge" in sys.argv:
cmd_judge()
elif "--score" in sys.argv:
cmd_score()
else:
print(__doc__)

View file

@ -43,6 +43,7 @@ sys.path.insert(0, str(REPO / "eval" / "exp16"))
sys.path.insert(0, str(REPO / "eval" / "role_topology"))
import pymorphy3 # noqa: E402
import yaml # noqa: E402
from detect_word import decomposes as _decomposes # noqa: E402
from detect_word import translit_shape as _translit_shape # noqa: E402
from palladius import is_palladius_token, palladius_conformant # noqa: E402
@ -81,7 +82,11 @@ _RU_NUM = {"ноль": 0, "один": 1, "два": 2, "две": 2, "три": 3,
"девятнадцать": 19, "двадцать": 20, "тридцать": 30, "сорок": 40, "пятьдесят": 50,
"шестьдесят": 60, "семьдесят": 70, "восемьдесят": 80, "девяносто": 90,
"сто": 100, "двести": 200, "триста": 300, "четыреста": 400, "пятьсот": 500,
"шестьсот": 600, "семьсот": 700, "восемьсот": 800, "девятьсот": 900}
"шестьсот": 600, "семьсот": 700, "восемьсот": 800, "девятьсот": 900,
# СОБИРАТЕЛЬНЫЕ (добор 07.08 по вскрытым ложным срабатываниям). Русский переводит
# 数万人 как «десятки тысяч», 一百多位 как «более сотни» — обе формы верны, а без этих
# лемм разбор давал потерю 10000/100 и выдумку 1000 на безупречном тексте.
"десяток": 10, "сотня": 100, "дюжина": 12}
_RU_MULT = {"тысяча": 1_000, "миллион": 1_000_000, "миллиард": 1_000_000_000}
# Пол фактонесущей величины. ВЫБРАН ЗАМЕРОМ (свип 8 конфигураций на 91 единице): при поле 10
# ложных «появившихся» 1.64 на единицу, при 100 — 0.27 при тех же 0.25 потерянных. Малые
@ -111,8 +116,16 @@ def normalize(text: str) -> str:
return "".join(out)
# ⚠ ДЕФИС ВНУТРИ СЛОВА — ЧАСТЬ СЛОВА. Первая редакция резала по нему, и первые половины составных
# («тёмно-зелёный», «точь-в-точь», «перво-наперво») приходили в детектор отдельными токенами,
# которых в словаре нет по построению. На прогоне армов это дало 15 ложных флагов «выдуманное
# слово» из 15 и завысило долю флагнутых единиц, от которой считается вся экономика маршрутизации.
# Поймано осмотром флагов, а не чтением кода: числа выглядели правдоподобно.
_RE_CYR_WORD_HYPH = re.compile(r"[А-Яа-яЁё]+(?:-[А-Яа-яЁё]+)*")
def words(text: str) -> list[str]:
return _RE_CYR_WORD.findall(normalize(text))
return _RE_CYR_WORD_HYPH.findall(normalize(text))
# ──────────────────────────────── проверки батареи ────────────────────────────────
@ -243,6 +256,56 @@ def check_translationese(final: str) -> dict:
calque_interjections=calques, marked_feminitives=fem)
# Неколичественные обороты: иероглиф величины стоит, величины нет. Список закрытый и короткий —
# добирается замером на корпусе, а не на глаз (метод тот же, что дал MIN_VALUE).
_ZH_IDIOM = ("百分之", "千分之", "万分之", # доли: 百分之一 = одна сотая, не «сто»
"十分", "十足", "万一", "万分", "千万", "百般", "万物", "万象", "百姓", "千秋")
_RE_ZH_RUN = re.compile(r"[零一二两三四五六七八九十百千万亿]+")
def _zh_values(source: str) -> set[int]:
"""Величины китайской записи. Разбор идёт по СВЯЗНОМУ ПРОБЕГУ, а не по паре «цифры+разряд».
Это починка дефекта, найденного адверсариальным ревью 07.08 и воспроизведённого исполнением.
Прежняя версия брала разряды поодиночке, тогда как русская сторона накапливает составное
числительное целиком: 两千三百 давало {2000, 300}, а «две тысячи триста» {2300}, и на
БЕЗУПРЕЧНОМ переводе проверка печатала две потери и одну выдумку. Проверено на трёх парах
(两千三百 · 三千五百 · 一百二十) все три давали ложное срабатывание.
Голый разряд теперь берётся (万里 = 10000), потому что русская сторона берёт голый множитель
(«десять тысяч ли»): несимметричность правил и была источником ложных выдумок. Цена симметрии
неколичественные обороты, они гасятся закрытым списком `_ZH_IDIOM`.
"""
out: set[int] = set()
clean = source
for idiom in _ZH_IDIOM:
clean = clean.replace(idiom, " ")
for m in _RE_ZH_RUN.finditer(clean):
run = m.group(0)
if run == "":
continue # голое 十 счётно почти никогда; 百年/千年/万里 — счётны
total = section = cur = 0
for ch in run:
if ch in _ZH_DIGITS and ch != "": # 十 живёт в обоих словарях; здесь он РАЗРЯД
cur = _ZH_DIGITS[ch]
elif ch == "":
section += (cur or 1) * 10
cur = 0
elif ch in ("", ""):
section += (cur or 1) * ZH_MAGNITUDE[ch]
cur = 0
elif ch == "":
total += (section + cur or 1) * 10_000
section = cur = 0
elif ch == "亿":
total = (total + section + cur or 1) * 100_000_000
section = cur = 0
v = total + section + cur
if v:
out.add(v)
return out
def check_numbers(source: str, final: str) -> dict:
"""6. Перенос чисел с нормализацией 万/亿. research/12 не покрывает; источник — бэклог 12
(«HARD-value-детектор /») и D39.79.
@ -253,17 +316,7 @@ def check_numbers(source: str, final: str) -> dict:
Проверка НЕ ловит переставленные величины при совпадающем множестве объявлено.
"""
src_vals = set(int(x) for x in re.findall(r"\d+", source))
# Разбор источника требует ЯВНОЙ ЦИФРЫ перед множителем. Голое 十 намеренно НЕ берётся:
# в китайском оно живёт в неколичественных идиомах (十分 «весьма», 十足 «полный»), и его
# добор поднял ложные потери 0.25→0.93 на единицу при выигрыше 0.27→0.10 по выдуманным —
# то есть сделал метрику хуже. Замер вариантов — в отчёте, не на глаз.
for m in re.finditer(r"([零一二两三四五六七八九十]+)\s*([万亿千百十])", source):
digits, mag = m.group(1), m.group(2)
cur = 0
for ch in digits:
v = _ZH_DIGITS.get(ch, 0)
cur = cur * 10 if v == 10 else cur + v
src_vals.add((cur or 1) * (10 if mag == "" else ZH_MAGNITUDE[mag]))
src_vals |= _zh_values(source)
dst_vals = set(int(x) for x in re.findall(r"\d+", normalize(final)))
for m in re.finditer(r"(\d+)\s*(тысяч\w*|миллион\w*)", normalize(final), re.I):
mult = 1_000 if m.group(2).lower().startswith("тысяч") else 1_000_000
@ -274,7 +327,10 @@ def check_numbers(source: str, final: str) -> dict:
# сотни+десятки+единицы складываются, множитель умножает накопленное.
acc, cur = 0, 0
for w in words(final):
lemma = _MORPH.parse(w.lower())[0].normal_form
# Первый разбор — не всегда числительный: у «десятки» это «десятка», а не «десяток».
# Берём первую лемму, которая ВООБЩЕ числительное, иначе откатываемся на первый разбор.
lemmas = [p.normal_form for p in _MORPH.parse(w.lower())]
lemma = next((x for x in lemmas if x in _RU_NUM or x in _RU_MULT), lemmas[0])
if lemma in _RU_NUM:
cur += _RU_NUM[lemma]
continue
@ -282,6 +338,8 @@ def check_numbers(source: str, final: str) -> dict:
acc += max(cur, 1) * _RU_MULT[lemma]
cur = 0
continue
if lemma == "и" and (cur or acc):
continue # «тысяча И одна ночь» — союз внутри составного, не разрыв
if cur or acc:
dst_vals.add(acc + cur)
acc = cur = 0
@ -319,6 +377,12 @@ def check_ty_vy(final: str) -> dict:
def check_gender(final: str, cards: dict[str, str]) -> dict:
"""4. Род глагола прош. вр. против карточек. research/12 §1 — «жалоба №1 читателей MTL».
ПАДЕЖ ИМЕНИ (починка по ревью 07.08). Карточки намеренно несут склонённые формы, иначе
«Фан Юаня» не сопоставится с «Фан Юань». Но косвенная форма имени по определению НЕ подлежащее,
и стоящий рядом глагол согласован с кем-то другим: «Воля Фан Юаня БЫЛА тверда» давала
рассогласование male/female на безупречном тексте. Все 6 «рассогласований» прошлого прогона
были этим классом. Кандидатом теперь считается только форма, у которой есть именительный разбор.
Карточка = {имя: 'male'|'female'}. Ищется «Имя + глагол прош. вр.» и сверяется род. Работает
только по ЯВНОЙ близости имени и глагола (окно 3 слова): местоименные и удалённые согласования
требуют кореференции и объявлены вне охвата. Без карточек проверка возвращает нули это
@ -330,6 +394,8 @@ def check_gender(final: str, cards: dict[str, str]) -> dict:
checked, bad = 0, []
for name, gender in cards.items():
for m in re.finditer(re.escape(name) + r"((?:\s+\S+){0,3})", t):
if not _is_nominative(name):
continue
for w in _RE_CYR_WORD.findall(m.group(1)):
for p in _MORPH.parse(w.lower()):
if p.tag.POS == "VERB" and "past" in str(p.tag):
@ -343,6 +409,52 @@ def check_gender(final: str, cards: dict[str, str]) -> dict:
return dict(checked=checked, mismatched=len(bad), cases=bad[:8])
def _is_nominative(name: str) -> bool:
"""Есть ли у ПОСЛЕДНЕГО токена имени именительный разбор. Кэшируется: вызовов много."""
if name in _NOMN_CACHE:
return _NOMN_CACHE[name]
toks = _RE_CYR_WORD.findall(name)
ok = True
if toks:
parses = _MORPH.parse(toks[-1].lower())
# Неизвестное морфологии имя пропускаем: отсечь по незнанию хуже, чем проверить.
known = [p for p in parses if p.tag.case]
ok = (not known) or any(p.tag.case == "nomn" for p in known)
_NOMN_CACHE[name] = ok
return ok
_NOMN_CACHE: dict[str, bool] = {}
def load_cards() -> dict[str, str]:
"""4-бис. КАРТОЧКИ ПЕРСОНАЖЕЙ (имя → пол) — вход проверки рода.
Пробел, найденный владельцем при сверке трекера: Ф0.3 промта заказывала карточки как ДАННЫЕ
для этой проверки, а я их не построил. Следствие было тихим и потому опасным: `check_gender`
принимает карточки аргументом и без них возвращает нули, то есть одна из девяти проверок
батареи всё время была инертна, а батарея отчитывалась как целое. Проверка, которая молча
ничего не проверяет, хуже отсутствующей отсутствующую видно.
Карточки берутся из ПОДПИСАННОГО сида книги (поле `gender`), а не составляются мной: пол
персонажа факт о книге, и выдумывать его полигон не вправе. Склонённые формы из `decl.forms`
добавляются как отдельные ключи, иначе «Фан Юаня» не сопоставится с «Фан Юань».
"""
seed = Path.home() / "books" / "gu-zhenren" / "guzhenren-seed-v2.yaml"
if not seed.exists():
return {}
data = yaml.safe_load(seed.read_text(encoding="utf-8")) or {}
cards: dict[str, str] = {}
for t in data.get("terms", []) or []:
g = t.get("gender")
if g not in ("male", "female"):
continue
for name in [t.get("dst", "")] + list((t.get("decl") or {}).get("forms") or []):
if name and _RE_CYR_WORD.search(name):
cards[name] = g
return cards
def check_repeat_consistency(units: list[dict]) -> dict:
"""2+3. Повтор-консистентность и коллизии, КРОСС-ЕДИНИЧНО.

114
eval/role_topology/buy.py Normal file
View file

@ -0,0 +1,114 @@
"""Единая касса пака: леджер с диска, ПРОЕКЦИОННЫЙ гард, однородная запись вызова.
Заведён 07.08 по адверсариальному ревью, которое нашло три разных дефекта одного класса:
* `repair_arm.CEILING_USD` и `route_arm.CEILING_USD` были ОБЪЯВЛЕНЫ и не использованы нигде
потолок Ф2б $3.00 был декоративным, а счётчик трат локальным и перезапуск не переживал;
* гард `screen.call` сравнивал УЖЕ потраченное с порогом и ничего не знал о цене следующего
вызова: запас $0.03 при наблюдённом максимуме одного вызова $0.090 потолок держался
случаем, а не механизмом;
* `bakeoff` считал потраченное по глобу `bo-` и не видел $1.02, ушедших на судейство.
Отсюда три правила, и все три живут ЗДЕСЬ, а не в каждом скрипте по-своему:
1. Потраченное читается С ДИСКА по всем префиксам фазы. Память процесса не источник истины.
2. Гард ПРОЕКЦИОННЫЙ: `spent + ожидаемая цена` против потолка, а не `spent` против порога.
Ожидание берётся из уже купленных вызовов той же роли (p95), при их отсутствии из сметы.
3. Запись несёт `total_tokens`. Без него биллинг Gemini (`additive_total`, quirks 00 D22.3)
не пере-считывается даже постфактум ровно это и случилось с прошлым прогоном.
"""
from __future__ import annotations
import json
import time
from pathlib import Path
from prices import cost
OUT = Path.home() / "books" / "role-topology"
OUT.mkdir(parents=True, exist_ok=True)
class Ledger:
"""Касса одной фазы. `prefixes` — глобы артефактов, из которых складывается потраченное."""
def __init__(self, name: str, ceiling: float, prefixes: tuple[str, ...],
default_expect: float = 0.05) -> None:
self.name, self.ceiling, self.prefixes = name, ceiling, prefixes
self.default_expect = default_expect
self._prices: list[float] = []
def spent(self) -> float:
tot, self._prices = 0.0, []
for pat in self.prefixes:
for f in OUT.glob(pat):
try:
r = json.loads(f.read_text(encoding="utf-8"))
except Exception: # noqa: BLE001, S112
continue
c = float(r.get("cost_usd") or 0.0)
tot += c
if c > 0:
self._prices.append(c)
return tot
def expect(self) -> float:
"""Ожидаемая цена следующего вызова: p95 уже купленного, иначе смета."""
if len(self._prices) < 3:
return self.default_expect
s = sorted(self._prices)
return s[min(len(s) - 1, int(round(0.95 * (len(s) - 1))))]
def afford(self) -> tuple[bool, str]:
sp = self.spent()
proj = sp + self.expect()
if proj > self.ceiling:
return False, (f"[СТОП {self.name}] потрачено ${sp:.6f} + ожидание ${self.expect():.6f} "
f"= ${proj:.6f} > потолок ${self.ceiling:.2f}")
return True, ""
def usage_of(u, ch) -> dict:
"""Токены вызова в однородном виде. `total_tokens` персистится ВСЕГДА (см. шапку)."""
cached = getattr(getattr(u, "prompt_tokens_details", None), "cached_tokens", 0) or 0
reasoning = getattr(getattr(u, "completion_tokens_details", None), "reasoning_tokens", 0) or 0
return dict(prompt_tokens=u.prompt_tokens or 0, cached_tokens=cached,
completion_tokens=u.completion_tokens or 0, reasoning_tokens=reasoning,
total_tokens=getattr(u, "total_tokens", 0) or 0,
reasoning_chars=len(getattr(ch.message, "reasoning_content", None) or ""))
def priced(model: str, us: dict) -> float:
return round(cost(model, us["prompt_tokens"], us["cached_tokens"], us["completion_tokens"],
us["reasoning_tokens"], us["total_tokens"]), 6)
def purchase(led: Ledger, tag: str, model: str, client, kw: dict, **extra) -> dict:
"""Один платный вызов под кассой. Кэш читается ПЕРВЫМ, гард — перед покупкой.
Порядок важен и выведен из двух прошлых ошибок: гард перед чтением кэша объявлял уже
купленную клетку незамеренной, а гард после покупки не защищал ничего.
"""
f = OUT / f"{tag}.json"
if f.exists():
return json.loads(f.read_text(encoding="utf-8"))
ok, why = led.afford()
if not ok:
print(why)
return dict(tag=tag, model=model, skipped=True, cost_usd=0.0, content="", finish="skipped")
t0 = time.time()
try:
r = client.chat.completions.create(**kw)
except Exception as e: # noqa: BLE001
# Провал ОДНОГО вызова — данные, а не повод ронять прогон. ⚠ Но запись отказа кладётся
# ОТДЕЛЬНЫМ тегом: прошлая редакция писала её под тем же именем, и транзиентный таймаут
# навсегда отравлял клетку — пере-запуск читал ошибку из кэша и вызов не повторял.
rec = dict(tag=tag, model=model, finish="error", error=f"{type(e).__name__}: {str(e)[:200]}",
content="", cost_usd=0.0, latency_s=round(time.time() - t0, 1), **extra)
(OUT / f"{tag}.ERROR.json").write_text(json.dumps(rec, ensure_ascii=False, indent=1),
encoding="utf-8")
return rec
ch, us = r.choices[0], usage_of(r.usage, r.choices[0])
rec = dict(tag=tag, model=model, model_returned=r.model, finish=ch.finish_reason,
content=ch.message.content or "", latency_s=round(time.time() - t0, 1),
cost_usd=priced(model, us), **us, **extra)
f.write_text(json.dumps(rec, ensure_ascii=False, indent=1), encoding="utf-8")
return rec

View file

@ -0,0 +1,230 @@
#!/usr/bin/env python3
"""Ф0.2 — ЗАМЕР КОНТАМИНАЦИИ имеющихся книг вместо добычи новой.
Постановка. Промт заказывает «невиданный срез» и оговаривает главное: **контаминация НЕ по дате**,
её надо мерить пробами на самой модели. Я сперва прочитал это как «нужна новая книга» и сделал
блокером то, что блокером не является: материал в `~/books` есть, и все экспы 1820 гнались на нём
(промт пробы 18, арм AM: «найди на стенде 10 глав пассаж»). Правильная задача не притащить
шестую книгу, а УЗНАТЬ ЧИСЛОМ, какая из пяти имеющихся насколько сидит в претрейне, и разложить
работу по ним осознанно.
ДВЕ РЕДАКЦИИ ЭТОЙ ПРОБЫ БЫЛИ СНЯТЫ ИСПОЛНЕНИЕМ, и обе по одной причине: они спрашивали
ВОСПРОИЗВЕДЕНИЕ.
Редакция 1 звала `deepseek-v4-flash` с бюджетом 16000 токенов. Модель выжигала бюджет на
размышление и возвращала 046 знаков (стена quirks §10). Нулевое совпадение читалось как
«книга не заучена». Поймано осмотром сырья: $0.0030.005 за вызов при пустом ответе.
Редакция 2 звала `gpt-5.6-luna` без размышления и с коротким выходом стена ушла, но
ПОЛОЖИТЕЛЬНЫЙ КОНТРОЛЬ 金瓶梅 (минский канон, public domain, заведомо в претрейне) дал
ответ «НЕЗНАКОМ» и нулевое совпадение. Значит дело не в книгах: современные модели не
воспроизводят текст дословно по запросу вообще, и проба на продолжение непригодна КАК КЛАСС.
Метод сменён на тот, который спрашивает ЗНАНИЕ, а не воспроизведение. Оба вопроса ниже
безобидны с точки зрения выдачи текста и при этом отвечаются только тем, кто книгу видел.
Как меряется сейчас. Две независимые пробы на одном и том же отрывке:
УЗНАВАНИЕ «что это за произведение и кто его автор». Верное название = текст (или подробные
описания текста) были в обучающих данных. Ловит знание о книге вообще.
КЛОУЗ ПО ИМЕНИ из отрывка вырезано имя собственное, модель называет пропущенное. Заполнить
можно, только зная книгу; при этом от модели требуется ОДНО СЛОВО, а не проза. Ловит более
тонкое знание, чем узнавание: имя второстепенного персонажа помнит лишь тот, кто видел текст.
Почему различение важно для продукта. Завышение качества перевода даёт не «слышал о книге», а
знание её конкретики: имён, реалий, устоявшихся переводов. Именно это и меряет клоуз.
ПОЛОЖИТЕЛЬНЫЙ КОНТРОЛЬ `jinpingmei`. Нулевой результат по любой книге сам по себе
неинтерпретируем: «модель текст не помнит» и «проба не работает» дают одинаковую картину. Если
узнавание не сработает и на нём сломана проба, а не книги. Роль контроля названа заранее, а не
выбрана задним числом по удобству результата. (Та же логика, что декой у QE-замера и арм `xhigh`
у вахты эффорта; в этой пробе контроль уже дважды сработал по назначению см. баннер выше.)
Отрывки НЕ попадают ни в репозиторий, ни в отчёт: печатаются только вердикты и метрики.
Запуск: eval/.venv/bin/python eval/role_topology/contamination.py --dry # план, $0
eval/.venv/bin/python eval/role_topology/contamination.py
"""
from __future__ import annotations
import json
import os
import re
import sys
import time
from collections import Counter
from pathlib import Path
REPO = Path("/home/ubuntu/projects/textmachine")
from dotenv import load_dotenv # noqa: E402
from openai import OpenAI # noqa: E402
load_dotenv(REPO / "eval" / ".env")
OUT = Path.home() / "books" / "role-topology"
OUT.mkdir(parents=True, exist_ok=True)
MODEL = "gpt-5.6-luna" # без стены размышления, дешёвый, и сам кандидат Ф2
PRICE_IN, PRICE_CACHED, PRICE_OUT = 0.20, 0.02, 1.20
MAX_OUT = 200
CEILING_USD = 0.10
N_PASSAGES = 5
PRIME_CHARS = 700
BOOKS: dict[str, dict] = {
"gu-zhenren": dict(path=Path.home() / "books" / "gu-zhenren" / "guzhenren-utf8.txt",
lang="zh", titles=["蛊真人", "гу чжэнь", "reverend insanity", "гу чжэньжэнь"],
note="вебновелла 20122018, есть en-фанперевод"),
"jinpingmei": dict(path=Path.home() / "books" / "jinpingmei" / "jinpingmei-ctext-zhs.txt",
lang="zh", titles=["金瓶梅", "цзинь пин мэй", "plum in the golden"],
note="ПОЛОЖИТЕЛЬНЫЙ КОНТРОЛЬ: минский канон, public domain"),
"isekai-ja": dict(path=Path.home() / "books" / "isekai_majutsushi_jp.txt",
lang="ja", titles=["異世界魔術師", "isekai majutsushi"],
note="вебновелла с syosetu, R18-раздел"),
"fifty-shades-en": dict(path=Path.home() / "books" / "fifty_shades_1_en.txt",
lang="en", titles=["fifty shades", "пятьдесят оттенков"],
note="издано 2011, бестселлер"),
"kristoff-en": dict(path=Path.home() / "books" /
"Kristoff_Jay_-_Empire_of_the_Vampire_3_-_Empire_of_the_Dawn.epub",
lang="en", titles=["empire of the vampire", "empire of the dawn",
"kristoff"],
note="роман 2026, ru-перевод не издан"),
}
def read_book(path: Path) -> str:
"""Текст книги. epub разбирается тем же способом, что `crosslang_bank.epub_text`."""
if path.suffix != ".epub":
return path.read_text(encoding="utf-8", errors="ignore")
import zipfile # noqa: PLC0415
out = []
with zipfile.ZipFile(path) as z:
for n in sorted(z.namelist()):
if n.endswith((".xhtml", ".html")):
html = z.read(n).decode("utf-8", "ignore")
txt = re.sub(r"<[^>]+>", " ", html)
txt = re.sub(r"&[a-z]+;", " ", txt)
out.append(re.sub(r"\s+", " ", txt))
return "\n".join(out)
def frequent_names(text: str, lang: str, top: int = 40) -> list[str]:
"""Частые имена собственные книги — кандидаты на маскирование в клоузе.
Для латиницы/кириллицы берутся слова с прописной вне начала предложения; для CJK частые
биграммы/триграммы, не являющиеся служебными. Список нужен только чтобы ВЫБРАТЬ, что маскировать,
поэтому грубость метода допустима и объявлена.
"""
if lang in ("en",):
cand = Counter(m.group(1) for m in re.finditer(r"(?<![.!?]\s)(?<!^)\b([A-Z][a-z]{3,})\b",
text, re.M))
stop = {"The", "And", "But", "That", "This", "What", "When", "Then", "There", "With"}
return [w for w, _ in cand.most_common(top * 3) if w not in stop][:top]
# CJK: частые 23-знаковые последовательности хань/каны. Грубо, но для выбора маски хватает.
cand = Counter()
for n in (3, 2):
for m in re.finditer(r"[㐀-鿿゠-ヿ]{%d}" % n, text):
cand[m.group(0)] += 1
return [w for w, c in cand.most_common(top * 2) if c > 20][:top]
def passages(text: str, names: list[str], n: int) -> list[tuple[str, str]]:
"""(отрывок с маской, замаскированное имя). Отрывки из середины, точки детерминированы."""
body = text[len(text) // 10: -len(text) // 10]
out: list[tuple[str, str]] = []
for i in range(n * 6):
if len(out) >= n:
break
pos = len(body) * (i + 1) // (n * 6 + 1)
chunk = body[pos:pos + PRIME_CHARS]
if len(chunk) < PRIME_CHARS:
continue
hit = next((w for w in names if w in chunk), None)
if not hit:
continue
out.append((chunk.replace(hit, "[???]"), hit))
return out
def ask(cl, prime: str, tag: str) -> dict:
"""Один вызов: узнавание и клоуз задаются ВМЕСТЕ, чтобы не платить дважды за один отрывок."""
f = OUT / f"cont-{tag}.json"
if f.exists():
return json.loads(f.read_text(encoding="utf-8"))
msgs = [
{"role": "system", "content":
"Тебе дают отрывок художественного текста, в котором одно имя собственное заменено на "
"[???]. Ответь РОВНО двумя строками и ничем больше:\n"
"ПРОИЗВЕДЕНИЕ: <название и автор, либо НЕ ЗНАЮ>\n"
"ИМЯ: <что стоит вместо [???], либо НЕ ЗНАЮ>\n"
"Не пересказывай отрывок и не продолжай его."},
{"role": "user", "content": prime},
]
r = cl.chat.completions.create(model=MODEL, messages=msgs, max_completion_tokens=MAX_OUT,
reasoning_effort="none")
u = r.usage
cached = getattr(getattr(u, "prompt_tokens_details", None), "cached_tokens", 0) or 0
rec = dict(tag=tag, finish=r.choices[0].finish_reason,
cost_usd=round((u.prompt_tokens - cached) / 1e6 * PRICE_IN
+ cached / 1e6 * PRICE_CACHED
+ u.completion_tokens / 1e6 * PRICE_OUT, 6),
answer=r.choices[0].message.content or "")
f.write_text(json.dumps(rec, ensure_ascii=False), encoding="utf-8")
return rec
def parse(answer: str) -> tuple[str, str]:
work = name = ""
for line in answer.splitlines():
if line.upper().startswith("ПРОИЗВЕДЕНИЕ"):
work = line.split(":", 1)[-1].strip()
elif line.upper().startswith("ИМЯ"):
name = line.split(":", 1)[-1].strip()
return work, name
def main() -> None:
dry = "--dry" in sys.argv
cl = None if dry else OpenAI(api_key=os.environ["OPENAI_API_KEY"],
base_url="https://api.openai.com/v1", timeout=300)
spent = 0.0
summary: list[dict] = []
for book, meta in BOOKS.items():
if not meta["path"].exists():
print(f"{book}: файла нет — пропуск")
continue
text = read_book(meta["path"])
names = frequent_names(text, meta["lang"])
ps = passages(text, names, N_PASSAGES)
print(f"\n{book} ({meta['lang']}, {len(text):,} знаков) — отрывков {len(ps)} · {meta['note']}")
if dry:
continue
rec_ok = cloze_ok = 0
for i, (prime, hidden) in enumerate(ps):
if spent > CEILING_USD:
print("⛔ потолок исчерпан — стоп")
break
rec = ask(cl, prime, f"{book}-{i}")
spent += rec["cost_usd"]
work, name = parse(rec["answer"])
r_hit = any(t in work.lower() for t in meta["titles"])
c_hit = bool(name) and (hidden in name or name in hidden) and "НЕ ЗНАЮ" not in name
rec_ok += r_hit
cloze_ok += c_hit
print(f" отрывок {i}: узнавание {'ДА ' if r_hit else 'нет'} · "
f"клоуз {'ДА ' if c_hit else 'нет'} · ответ о книге: {work[:60] or ''}")
time.sleep(0.2)
summary.append(dict(book=book, n=len(ps), recognized=rec_ok, cloze=cloze_ok))
if dry:
return
print(f"\n{'книга':18s}{'узнано':>9s}{'клоуз':>8s}{'вердикт':>22s}")
print("-" * 58)
for s in summary:
n = max(1, s["n"])
# Порог объявлен заранее: узнавание ≥3/5 = книга модели ИЗВЕСТНА; клоуз ≥2/5 = известна
# КОНКРЕТИКА, то есть текст (а не только описания) был в обучающих данных.
v = ("знает конкретику" if s["cloze"] >= 2 else
("знает о книге" if s["recognized"] >= 3 else "следов нет"))
print(f"{s['book']:18s}{s['recognized']:>4d}/{n:<4d}{s['cloze']:>4d}/{n:<3d}{v:>22s}")
print(f"\nпотрачено ${spent:.6f}")
(OUT / "contamination.json").write_text(json.dumps(summary, ensure_ascii=False, indent=1),
encoding="utf-8")
if __name__ == "__main__":
main()

View file

@ -134,6 +134,13 @@ def decomposes(word: str) -> bool:
def verdict(word: str, bank: frozenset[str]) -> tuple[bool, str]:
"""(флаг «выдуманное», причина-непропуска). Причина печатается — вердикт обязан быть читаемым."""
# Составное через дефис считается выдуманным, только если ОБЕ части незнакомы: «тёмно-зелёный»
# словарь целиком не знает, но обе половины законны. Проверять по частям — не послабление,
# а единственный способ не флагать нормальное словообразование русского.
if "-" in word:
parts = [p for p in word.split("-") if len(p) >= 2]
if parts and all(known(p) or translit_shape(p) or decomposes(p) for p in parts):
return False, "составное из известных частей"
if known(word):
return False, "в словаре"
if word.lower() in bank:

View file

@ -96,11 +96,29 @@ def main() -> None:
P.slug_check() # гардрейл CLAUDE.md: слаг сверяется живым /models в день запуска
cl = P.client()
rows: list[dict] = []
spent = 0.0
n = 1 if pilot else N_PER_ARM
# ⚠ Потолок считается от ВСЕГО уже купленного вахтой, а не от текущего запуска: иначе
# контрольный прогон стартовал бы с нуля и пробил бы общий лимит, оставаясь «в потолке»
# по своему локальному счёту. Леджер = сумма персистированных артефактов, а не память.
spent = sum(json.loads(f.read_text(encoding="utf-8"))["cost_usd"]
for f in OUT.glob("ew-*.json"))
if spent:
print(f"уже куплено вахтой ранее: ${spent:.6f} — потолок считается от этой суммы\n")
# Контролю нужен ЗНАК, а не оценка величины ⇒ розыгрышей меньше и они ограничены остатком.
n = 1 if pilot else (4 if "--control" in sys.argv else N_PER_ARM)
stop = False
# ⚠ ПОЛОЖИТЕЛЬНЫЙ КОНТРОЛЬ `xhigh` — без него отрицательный результат по `low` НЕ
# интерпретируем: «вендор игнорирует ручку» и «риг не видит смену эффорта вообще» дают
# одинаковую картину. Про `xhigh` quirks §3б установил обратное — он РАБОТАЕТ и виден
# серверно (`prompt_tokens` 2203 против 2124). Если мой риг увидит xhigh и не увидит low,
# нуль по low становится содержательным; если не увидит и xhigh — риг негоден, и это
# честный исход. Та же логика, что у декоя D39.46(б), применённая к проводу.
# n меньше, чем у основных армов: контролю нужен ЗНАК, а не оценка величины, а остаток
# потолка после основного блока конечен.
arms = (("default", None), ("low", "low"))
if "--control" in sys.argv:
arms = (("xhigh", "xhigh"),)
for i in range(n):
for arm, effort in (("default", None), ("low", "low")):
for arm, effort in arms:
tag = f"ew-{arm}-r{i + 1}"
f = OUT / f"{tag}.json"
if f.exists(): # идемпотентность: пере-запуск не пере-покупает
@ -140,11 +158,33 @@ def main() -> None:
def report(rows: list[dict], spent: float) -> None:
from scipy.stats import mannwhitneyu # noqa: PLC0415
a = [r for r in rows if r["arm"] == "default"]
b = [r for r in rows if r["arm"] == "low"]
print(f"\nn: дефолт {len(a)} · low {len(b)} · потрачено ${spent:.6f}")
# Отчёт читает ВСЕ персистированные артефакты вахты, а не только розыгрыши текущего запуска:
# контрольный арм докупается отдельным прогоном, и сводка обязана видеть его вместе с основными.
allr = [json.loads(f.read_text(encoding="utf-8")) for f in sorted(OUT.glob("ew-*.json"))]
byarm: dict[str, list[dict]] = {}
for r in allr:
byarm.setdefault(r["tag"].split("-")[1], []).append(r)
a = byarm.get("default", [])
b = byarm.get("low", [])
c = byarm.get("xhigh", [])
total = sum(r["cost_usd"] for r in allr)
print(f"\nn: дефолт {len(a)} · low {len(b)} · xhigh(контроль) {len(c)} · "
f"куплено вахтой всего ${total:.6f}")
if c:
print("\nПОЛОЖИТЕЛЬНЫЙ КОНТРОЛЬ xhigh против дефолта:")
for key in ("completion_tokens", "reasoning_chars"):
ma = statistics.median(r[key] for r in a)
mc = statistics.median(r[key] for r in c)
p = mannwhitneyu([r[key] for r in a], [r[key] for r in c],
alternative="two-sided").pvalue
print(f" {key:20s} дефолт {ma:7.0f} · xhigh {mc:7.0f} · "
f"отношение {mc / ma:5.2f} · p={p:.4f}")
pc = sorted({r["prompt_tokens"] for r in c})
pa0 = sorted({r["prompt_tokens"] for r in a})
print(f" prompt_tokens дефолт {pa0} · xhigh {pc}"
+ (" ⇒ СЕРВЕРНАЯ РЕАКЦИЯ ЕСТЬ" if pc != pa0 else " ⇒ реакции нет"))
if len(a) < 3 or len(b) < 3:
print("данных мало для вывода")
print("данных по основным армам мало для вывода")
return
print(f"\n{'метрика':22s}{'дефолт (медиана)':>19s}{'low (медиана)':>16s}"
f"{'отношение':>11s}{'p (MW)':>9s}")

View file

@ -0,0 +1,255 @@
#!/usr/bin/env python3
"""Ф0.6 — ПРЕДЗАМЕР СУДЕЙ. Сначала мерим ИНСТРУМЕНТ, потом им мерим.
Зачем это первая платная работа фазы 2, а не последняя. Эксп-20 §5.4 намерил, что судья-LLM
согласен САМ С СОБОЙ на побайтно одном входе в 56% клеток (33% на несущих осях) и разниц меньше
~2:1 не разрешает. Значит любой вердикт бейк-оффа, прочитанный без знания этого разброса, это
пересказ шума. Здесь разброс меряется ДО Ф2 и превращается в ПОРОГ РАЗЛИЧИМОСТИ, который потом
применяется к армам механически.
Три вещи, и они разные:
РАЗБРОС СУДЬИ один и тот же пакет, N независимых голосов. Верхняя граница разрешающей
способности: разница между армами меньше собственного разброса судьи сигналом быть не может.
ДЕКОЙ (D39.46б) пакет, где одна сторона ЗАВЕДОМО испорчена посаженными дефектами пробы 18.
Судья, который его не ловит, не отличает годное от негодного вообще, и его голоса по армам
читать нельзя. Без декоя «армы неразличимы» и «судья слеп» выглядят одинаково.
ШУМОВОЙ ПОЛ ПАЙПЛАЙНА один и тот же арм прогнан ДВАЖДЫ на побайтно одном входе. Меряет,
сколько разницы порождает сама стохастичность модели, а не топология.
Выбор судей не вкусовой. Промт: «судья не судит армы своего семейства-жильца». Армы Ф2 населяют
DeepSeek (`deepseek-v4-pro`) и Z.AI (`glm-5`), поэтому судьи берутся из ДРУГИХ семейств:
`grok-4.3` (xAI) и `gpt-5.6-luna` (OpenAI). Оба прошли скрин Ф1, оба дёшевы, оба быстры.
Дисциплина протокола (промт, §ЗАКОН):
· КАЖДЫЙ голос персистится отдельным файлом ревью-шапка эксп-20 прямо запретила повторять
решающий замер без персиста («невоспроизводим не повторять»);
· зеркальная раскладка: каждая пара судится в ОБОИХ порядках, и позиция расцеплена с армом;
· слепые метки: судья видит «Вариант 1/2», а не имена моделей;
· первичный протокол СЧЁТ ТИПИЗИРОВАННЫХ ОШИБОК (MQM-lite), а не «что лучше»; преференс
спрашивается только по стилевой оси, где счёт ошибок не работает.
Запуск: eval/.venv/bin/python eval/role_topology/judges.py --variance # разброс + декой
eval/.venv/bin/python eval/role_topology/judges.py --floor # шумовой пол пайплайна
"""
from __future__ import annotations
import json
import os
import re
import statistics
import sys
import time
from collections import Counter, defaultdict
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 editor_wire_probe as P # noqa: E402
from prices import CANDIDATES, cost # noqa: E402
load_dotenv(REPO / "eval" / ".env")
OUT = Path.home() / "books" / "role-topology"
OUT.mkdir(parents=True, exist_ok=True)
# ⚠ ПОТОЛОК И СЧЁТЧИК РАЗВЕДЕНЫ ПО ФАЗАМ. Судейство бейк-оффа зовёт тот же `vote`, но относится
# к бюджету Ф2а, а не Ф0.6. Первая редакция сторожила общий потолок $0.60 по ВСЕМ файлам `jv-*`,
# и прогон Ф2а молча пропустил бы почти все голоса, оставив пустую таблицу вердиктов. Поймано
# чтением кода до запуска. Голоса Ф2а помечаются префиксом `jv-bo-` и считаются отдельно.
CEILING_USD = 0.60
PHASE_PREFIX = "jv-" # что считать «своими» тратами; bakeoff подменяет на "jv-bo-"
JUDGES = ("gpt-5.6-luna", "grok-4.3")
REPS = 3
OPENAI_FAMILY = {"gpt-5.6-luna"}
RUBRIC = """Ты — литературный редактор-эксперт, сверяющий два русских перевода одного китайского
фрагмента с оригиналом. Тебе даны ИСХОДНИК и два варианта перевода под слепыми метками.
Посчитай ОШИБКИ в каждом варианте по типам:
ВЕРНОСТЬ сказано не то, что в исходнике: искажение, инверсия, выдумка, пропуск факта.
ТЕРМИН имя или термин передан не так, как в других местах того же текста.
ЯЗЫК то, чего носитель не напишет: калька, канцелярит, несуществующее слово, кривой
порядок слов, рассогласование.
ФОРМА вёрстка и пунктуация: оформление реплик, кавычки, тире, абзацы.
Отвечай РОВНО в этом формате и ничем больше:
В1-ВЕРНОСТЬ: <число>
В1-ТЕРМИН: <число>
В1-ЯЗЫК: <число>
В1-ФОРМА: <число>
В2-ВЕРНОСТЬ: <число>
В2-ТЕРМИН: <число>
В2-ЯЗЫК: <число>
В2-ФОРМА: <число>
СТИЛЬ: <В1 или В2 или НИЧЬЯ где живее и естественнее русская проза>
"""
AXES = ("ВЕРНОСТЬ", "ТЕРМИН", "ЯЗЫК", "ФОРМА")
def client(model: str) -> OpenAI:
base, env, *_ = CANDIDATES[model]
return OpenAI(api_key=os.environ[env], base_url=base, timeout=600)
def spent_so_far() -> float:
"""Леджер = сумма персистированных голосов. Переживает процесс — урок перерасхода Ф1."""
tot = 0.0
for f in OUT.glob(f"{PHASE_PREFIX}*.json"):
if PHASE_PREFIX == "jv-" and f.name.startswith("jv-bo-"):
continue # голоса бейк-оффа живут в бюджете Ф2а, не Ф0.6
try:
tot += json.loads(f.read_text(encoding="utf-8")).get("cost_usd", 0.0)
except Exception: # noqa: BLE001, S112
continue
return tot
def vote(judge: str, packet: str, tag: str) -> dict:
"""Один голос. Персистится отдельным файлом — иначе замер невоспроизводим (ревью-шапка 20)."""
f = OUT / f"jv-{tag}.json"
if f.exists():
return json.loads(f.read_text(encoding="utf-8"))
if spent_so_far() > CEILING_USD:
return dict(tag=tag, skipped=True, cost_usd=0.0, content="")
kw = vote_kw(judge, packet)
r = client(judge).chat.completions.create(**kw)
u = r.usage
cached = getattr(getattr(u, "prompt_tokens_details", None), "cached_tokens", 0) or 0
rec = dict(tag=tag, judge=judge, finish=r.choices[0].finish_reason,
content=r.choices[0].message.content or "",
cost_usd=round(cost(judge, u.prompt_tokens or 0, cached, u.completion_tokens or 0,
getattr(getattr(u, "completion_tokens_details", None),
"reasoning_tokens", 0) or 0,
getattr(u, "total_tokens", 0) or 0), 6))
f.write_text(json.dumps(rec, ensure_ascii=False), encoding="utf-8")
return rec
def vote_kw(judge: str, packet: str) -> dict:
"""Параметры вызова судьи. Вынесено из `vote`, чтобы бейк-офф покупал голоса через общую
кассу `buy.purchase` (леджер с диска + проекционный гард), а не через локальный счётчик."""
kw: dict = dict(model=judge, messages=[{"role": "system", "content": RUBRIC},
{"role": "user", "content": packet}])
if judge in OPENAI_FAMILY:
# ⚠ Бюджет и ГАШЕНИЕ РАЗМЫШЛЕНИЯ. Первая редакция давала 4000 без ручки эффорта, и luna
# выжигала их на рассуждение: 8 голосов из 18 пришли ПУСТЫМИ. Судья, который молчит в
# 44% клеток, не судья. Тот же класс дефекта, что стена flash в §2.7 — и лечится так же.
kw["max_completion_tokens"] = 8000
kw["reasoning_effort"] = "low"
else:
kw["max_tokens"] = 4000
kw["temperature"] = 0.4 # разброс меряется у РЕАЛЬНОГО инструмента, не идеального
return kw
def parse(answer: str) -> dict:
"""Счёт ошибок по осям + стилевой преференс. Непарсящееся — None, а не ноль (это разные вещи)."""
out: dict = {}
for side in ("В1", "В2"):
for ax in AXES:
m = re.search(rf"^{side}-{ax}\s*:\s*(\d+)", answer, re.M)
out[f"{side}-{ax}"] = int(m.group(1)) if m else None
m = re.search(r"^СТИЛЬ\s*:\s*(В1|В2|НИЧЬЯ)", answer, re.M)
out["СТИЛЬ"] = m.group(1) if m else None
return out
def packet(source: str, v1: str, v2: str) -> str:
return (f"ИСХОДНИК:\n{source}\n\n---\nВАРИАНТ 1:\n{v1}\n\n---\nВАРИАНТ 2:\n{v2}\n")
def run_variance() -> None:
"""Разброс каждого судьи на ПОВТОРЕ одного входа + ДЕКОЙ. Оба на материале пробы 18."""
from inject_probe import pick_windows # noqa: PLC0415
wins = pick_windows()
rows: list[dict] = []
for k in range(min(3, len(wins))):
_, src, _ = wins[k]
clean = P.draft_of(k)
planted, _ = P.planted_draft(k, clean)
# Клетка РАЗБРОСА: обе стороны — один и тот же текст. Любое расхождение вердиктов здесь
# есть чистый шум судьи, потому что различать нечего по построению.
# Клетка ДЕКОЯ: против чистого стоит заведомо испорченный (посадки пробы 18).
cells = {"same": (clean, clean), "decoy": (clean, planted)}
for kind, (a, b) in cells.items():
for judge in JUDGES:
for rep in range(1, REPS + 1):
rec = vote(judge, packet(src, a, b), f"{kind}-{judge}-w{k}-r{rep}")
if rec.get("skipped"):
print("⛔ потолок Ф0.6 исчерпан")
return
rows.append(dict(kind=kind, judge=judge, window=k, rep=rep,
**parse(rec["content"])))
time.sleep(0.2)
print(f"РАЗБРОС СУДЕЙ И ДЕКОЙ · повторов на клетку {REPS} · температура как в бою\n")
print(f"{'судья':16s}{'голосов':>8s}{'без лож.перевеса':>17s}{'доля':>8s} (клетки РАЗБРОСА)")
print("-" * 62)
floor = {}
for judge in JUDGES:
cells = defaultdict(list)
for r in rows:
if r["kind"] == "same" and r["judge"] == judge:
cells[r["window"]].append(r)
# ⚠ МЕТРИКА ИСПРАВЛЕНА ПОСЛЕ ПЕРВОГО ПРОГОНА. Первая редакция требовала совпадения
# АБСОЛЮТНЫХ счетов во всех повторах и дала 25%/42% — я прочитал это как «разброс огромен».
# Пере-счёт по сырью показал, что метрика мерила не то: судья может сказать «3 ошибки у
# обоих» и «5 у обоих» — это дрейф КАЛИБРОВКИ, а не выдуманная разница между вариантами.
# Операционально решает второе: сочиняет ли судья ПЕРЕВЕС там, где тексты идентичны.
# Прямой замер: ложный перевес нулевой в 13 голосах из 13. Печатаются обе величины —
# дрейф калибровки важен для АБСОЛЮТНЫХ порогов, ложный перевес — для ПАРНЫХ вердиктов.
unan = tot = false_margin = votes_ok = 0
for w, rs in cells.items():
for r in rs:
if r["В1-ВЕРНОСТЬ"] is None:
continue
votes_ok += 1
false_margin += int(sum(abs((r[f"В1-{a}"] or 0) - (r[f"В2-{a}"] or 0))
for a in AXES) != 0)
for ax in AXES:
vals = [r[f"В1-{ax}"] for r in rs] + [r[f"В2-{ax}"] for r in rs]
if any(v is None for v in vals):
continue
tot += 1
unan += int(len(set(vals)) == 1)
if votes_ok:
floor[judge] = false_margin / votes_ok
print(f"{judge:16s}{votes_ok:8d}{votes_ok - false_margin:13d}"
f"{1 - false_margin / votes_ok:8.0%}"
f" (дрейф абс. счёта: совпал в {unan}/{tot})")
print(f"\n{'судья':16s}{'декой пойман':>14s}{'доля':>8s} (обязан выбрать ЧИСТЫЙ вариант)")
print("-" * 62)
for judge in JUDGES:
rs = [r for r in rows if r["kind"] == "decoy" and r["judge"] == judge]
caught = tot = 0
for r in rs:
a = sum(r[f"В1-{ax}"] or 0 for ax in AXES)
b = sum(r[f"В2-{ax}"] or 0 for ax in AXES)
if r["В1-ВЕРНОСТЬ"] is None:
continue
tot += 1
caught += int(b > a) # В2 — испорченный, у него ошибок обязано быть больше
if tot:
print(f"{judge:16s}{caught:14d}{caught / tot:8.0%}")
(OUT / "judge-variance.json").write_text(json.dumps(rows, ensure_ascii=False, indent=1),
encoding="utf-8")
print(f"\nпотрачено ${spent_so_far():.6f} из ${CEILING_USD}")
print("\nЧитать: доля единогласия на клетках РАЗБРОСА — верхняя граница разрешающей\n"
"способности судьи. Декой ниже 100% означает, что и явную порчу он видит не всегда,\n"
"и тогда все его вердикты по армам читаются через эту поправку.")
if __name__ == "__main__":
if "--variance" in sys.argv:
run_variance()
else:
print(__doc__)

View file

@ -0,0 +1,110 @@
#!/usr/bin/env python3
"""Ф1 — ЦЕНЫ КАНДИДАТОВ, снятые с прайс-страниц вендоров 06.08.2026.
Почему отдельным файлом с датой и URL. Гардрейл CLAUDE.md: цены берутся ТОЛЬКО с прайс-страницы
вендора, не по памяти; урок календаря D39.61: «слаг живой модель та же». Смета Ф1/Ф2 стоит на
этих числах, и если через неделю они разъедутся, должно быть видно, ЧТО именно устарело и откуда
бралось. Порядок в кортеже: (вход, кэш-хит, выход) долларов за 1M токенов.
ЧТО ЗДЕСЬ НЕ ПРОВЕРЕНО ЖИВЬЁМ: цена это ЗАЯВЛЕНИЕ вендора, а не замер. Реальная стоимость
вызова считается из `usage` и может разойтись с прайсом (кэш, тарификация размышления, батч-тир).
Расхождение прайса с леджером само по себе находка, и Ф1 обязана его напечатать.
"""
from __future__ import annotations
# (base_url, env-ключ, вход, кэш, выход, механизм гашения размышления, источник цены)
CANDIDATES: dict[str, tuple] = {
# DeepSeek — цены пост-катовера, сверены 25.07 (quirks 00, строка эндпоинтов).
"deepseek-v4-flash": ("https://api.deepseek.com/v1", "DEEPSEEK_API_KEY",
0.14, 0.0028, 0.28, None, "quirks 00 (25.07, вендор-прайс)"),
"deepseek-v4-pro": ("https://api.deepseek.com/v1", "DEEPSEEK_API_KEY",
0.435, 0.003625, 0.87, None, "quirks 00 (25.07, вендор-прайс)"),
# Z.AI — docs.z.ai/guides/overview/pricing, снято 06.08.
"glm-5": ("https://api.z.ai/api/paas/v4", "ZAI_API_KEY",
1.00, 0.20, 3.20, "extra_body_disable", "docs.z.ai/pricing 06.08"),
"glm-5-turbo": ("https://api.z.ai/api/paas/v4", "ZAI_API_KEY",
1.20, 0.24, 4.00, "extra_body_disable", "docs.z.ai/pricing 06.08"),
"glm-5.2": ("https://api.z.ai/api/paas/v4", "ZAI_API_KEY",
1.40, 0.26, 4.40, "extra_body_disable", "docs.z.ai/pricing 06.08"),
"glm-4.7": ("https://api.z.ai/api/paas/v4", "ZAI_API_KEY",
0.60, 0.11, 2.20, "extra_body_disable", "docs.z.ai/pricing 06.08"),
# xAI — docs.x.ai/docs/models, снято 06.08. Тариф <200k токенов (наши единицы много меньше).
"grok-4.3": ("https://api.x.ai/v1", "XAI_API_KEY",
1.25, 0.20, 2.50, "effort_none", "docs.x.ai/models 06.08"),
"grok-4.5": ("https://api.x.ai/v1", "XAI_API_KEY",
2.00, 0.30, 6.00, "effort_none", "docs.x.ai/models 06.08 (НОВЫЙ жилец)"),
# OpenAI — прайс сверен живьём 05.08 (эксп-20 §1.3), изменений против 04.08 не было.
"gpt-5.6-luna": ("https://api.openai.com/v1", "OPENAI_API_KEY",
0.20, 0.02, 1.20, "effort_none", "эксп-20 §1.3 (05.08, вендор-прайс)"),
"gpt-5.6-terra": ("https://api.openai.com/v1", "OPENAI_API_KEY",
2.00, 0.20, 12.00, "effort_none", "эксп-20 §1.3 (05.08, вендор-прайс)"),
# Gemini — ai.google.dev/gemini-api/docs/pricing, снято 06.08. Кэш-цена не выделена.
"gemini-3.6-flash": ("https://generativelanguage.googleapis.com/v1beta/openai",
"GEMINI_API_KEY", 1.50, 1.50, 7.50, "thinking_budget",
"ai.google.dev/pricing 06.08 (НОВЫЙ жилец)"),
"gemini-3.1-flash-lite": ("https://generativelanguage.googleapis.com/v1beta/openai",
"GEMINI_API_KEY", 0.25, 0.25, 1.50, "thinking_budget",
"ai.google.dev/pricing 06.08"),
"gemini-3.1-pro-preview": ("https://generativelanguage.googleapis.com/v1beta/openai",
"GEMINI_API_KEY", 2.00, 2.00, 12.00, "mandatory",
"ai.google.dev/pricing 06.08"),
# Kimi — platform.kimi.ai/docs/pricing/chat-k3, снято 06.08. Самый дорогой выход ростера.
"kimi-k3": ("https://api.moonshot.ai/v1", "KIMI_API_KEY",
3.00, 0.30, 15.00, None, "platform.kimi.ai/pricing 06.08 (НОВЫЙ жилец)"),
# Mistral — mistral.ai/pricing, снято 06.08. Кэш-тариф не опубликован ⇒ равен входу.
"mistral-large-latest": ("https://api.mistral.ai/v1", "MISTRAL_API_KEY",
2.00, 2.00, 6.00, None, "mistral.ai/pricing 06.08"),
}
# Учёт РАЗМЫШЛЕНИЯ в биллинге различается по провайдерам, и это не деталь, а деньги.
# subset — размышление ВНУТРИ completion_tokens (DeepSeek, OpenAI, Z.AI): считать нечего.
# additive — размышление СВЕРХ completion (xAI): billed = completion + reasoning.
# Источник: quirks 00, строка grok — «reasoning у xAI аддитивен к completion (billed =
# completion+reasoning)». ⚠ Поймано исполнением: скрин показал у grok-4.5 reasoning 3605 при
# completion 1737, то есть размышление БОЛЬШЕ ответа — при subset-формуле это невозможно, и
# первая редакция занижала цену grok примерно втрое.
# additive_total — размышление видно ТОЛЬКО в total_tokens (Gemini): OpenAI-совместимый слой
# Google отдаёт reasoning_tokens=0 и НЕ кладёт thinking в completion, поэтому оплаченный
# выход = total prompt. Источник: quirks 00 §D22.3 — «иначе недоучёт 146×/вызов».
# ⚠ Класс добавлен 07.08 по адверсариальному ревью: прошлая редакция знала только subset и
# additive, gemini падал в subset, и его размышление не оплачивалось в леджере вовсе. Величину
# недоучёта прошлого прогона восстановить НЕЛЬЗЯ: total_tokens тогда не персистился.
REASONING_BILLING = {"grok-4.3": "additive", "grok-4.5": "additive",
"gemini-3.6-flash": "additive_total",
"gemini-3.1-flash-lite": "additive_total",
"gemini-3.1-pro-preview": "additive_total"}
def billed_output(model: str, completion_tokens: int, reasoning_tokens: int,
total_tokens: int = 0, prompt_tokens: int = 0) -> int:
"""Сколько выходных токенов реально тарифицируется."""
mode = REASONING_BILLING.get(model)
if mode == "additive":
return completion_tokens + reasoning_tokens
if mode == "additive_total":
# Пол — обычная формула: если total не записан (старый артефакт), не занижаем молча,
# но и не выдумываем. Отсутствие total_tokens видно по равенству двух путей.
return max(total_tokens - prompt_tokens, completion_tokens + reasoning_tokens)
return completion_tokens
def cost(model: str, prompt_tokens: int, cached: int, completion_tokens: int,
reasoning_tokens: int = 0, total_tokens: int = 0) -> float:
"""Цена вызова по прайсу. Кэшированные токены тарифицируются отдельной ставкой."""
_, _, pin, pcache, pout, _, _ = CANDIDATES[model]
out = billed_output(model, completion_tokens, reasoning_tokens, total_tokens, prompt_tokens)
return ((prompt_tokens - cached) / 1e6 * pin + cached / 1e6 * pcache + out / 1e6 * pout)
def out_price_order() -> list[str]:
"""Кандидаты по цене ВЫХОДА — она доминирует в наших ролях (размышление биллится как выход)."""
return sorted(CANDIDATES, key=lambda m: CANDIDATES[m][4])
if __name__ == "__main__":
print(f"{'модель':26s}{'вход':>8s}{'кэш':>8s}{'выход':>8s} источник")
print("-" * 86)
for m in out_price_order():
_, _, pin, pc, pout, _, src = CANDIDATES[m]
print(f"{m:26s}{pin:8.3f}{pc:8.3f}{pout:8.2f} {src}")

View file

@ -46,46 +46,13 @@ sys.path.insert(0, str(REPO / "eval" / "editor_contract"))
sys.path.insert(0, str(REPO / "eval" / "exp16"))
import pymorphy3 # noqa: E402
from align import align, split_ru, split_zh # noqa: E402
from align import align, flip_polarity, split_ru, split_zh # noqa: E402
from qe_bench import QE, MODEL # noqa: E402
_MORPH = pymorphy3.MorphAnalyzer()
OUT = Path.home() / "books" / "role-topology"
def flip_polarity(sent: str) -> str | None:
"""Инвертирует полярность предложения: снимает «не» при глаголе либо вставляет его.
Работает только по ЛИЧНОМУ глаголу или глаголу прошедшего времени причастия и деепричастия
исключены сознательно: «не» при них меняет смысл иначе и часто даёт неграмматичный результат,
а неграмматичность ранкер поймает не как инверсию, а как порчу формы (другой класс).
"""
m = re.search(r"\е\s+([А-Яа-яЁё]+)", sent)
if m and _is_finite_verb(m.group(1)):
return sent[:m.start()] + m.group(1) + sent[m.end():]
for m in re.finditer(r"\b([А-Яа-яЁё]{3,})\b", sent):
w = m.group(1)
if sent[max(0, m.start() - 4):m.start()].strip().endswith("не"):
continue
# ⚠ Пропускаем глагол с ПРОПИСНОЙ: он открывает предложение или реплику, и вставка даёт
# «не Стой» — неграмматичный текст. Ранкер оценил бы его как порчу ФОРМЫ, то есть замер
# мерил бы другой класс дефекта. Поймано проверкой генератора до прогона.
if w[0].isupper():
continue
if _is_finite_verb(w):
return sent[:m.start()] + "не " + sent[m.start():]
return None
def _is_finite_verb(word: str) -> bool:
for p in _MORPH.parse(word.lower()):
if not p.is_known:
continue
if p.tag.POS == "VERB":
return True
return False
def sign_test_p(successes: int, n: int) -> float:
"""Односторонний p знакового теста при H0: направление случайно (q=0.5).

View file

@ -39,9 +39,13 @@ sys.path.insert(0, str(REPO / "eval" / "bank_arbitration"))
sys.path.insert(0, str(REPO / "eval" / "editor_contract"))
from align import align, split_ru, split_zh # noqa: E402
from qe_bench import QE # noqa: E402
from qe_bench import QE, MODEL # noqa: E402
OUT = Path.home() / "books" / "role-topology"
# ⚠ Имя артефакта ЗАВИСИТ ОТ МОДЕЛИ. Первая редакция писала в фиксированное имя, и прогон XL
# МОЛЧА затёр сырьё прогона large — числа в отчёте остались бы «со слов сессии». Ровно этот
# дефект ревью-шапка эксп-20 назвала невоспроизводимым замером и запретила повторять.
TAG = "xl" if "XL" in MODEL else "large"
def main() -> None:
@ -101,9 +105,9 @@ def main() -> None:
print(f"\nДЕКОЙ (ru-сторона подменена чужим сегментом), n={len(decoy)}: "
f"медиана Δ {statistics.median(decoy):+.3f}, Δ>0 в "
f"{sum(1 for x in decoy if x > 0)}/{len(decoy)}, макс {max(decoy):+.3f}")
(OUT / "qe-decoy.json").write_text(json.dumps(decoy), encoding="utf-8")
(OUT / f"qe-decoy-{TAG}.json").write_text(json.dumps(decoy), encoding="utf-8")
(OUT / "qe-segments.json").write_text(
(OUT / f"qe-segments-{TAG}.json").write_text(
json.dumps([{**r, "di": list(r["di"])} for r in rows], ensure_ascii=False),
encoding="utf-8")
report(rows, P)

View file

@ -0,0 +1,205 @@
#!/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 = 3.00
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:
"""Портит одно длинное известное слово фиксированным правилом. Возвращает (новое предложение,
испорченное слово, исходное слово).
Правило детерминированное и НЕ подстроено под детектор: берётся самое длинное словарное слово
предложения и его 45-й знаки заменяются на «яв». Земля объективная: получившаяся форма
отсутствует в морфологическом словаре.
"""
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()

View file

@ -0,0 +1,149 @@
#!/usr/bin/env python3
"""Ф2б, армы E и G — обратная связка и маршрутизация. Обе топологии считаются на ОДНОМ материале.
E ОБРАТНАЯ СВЯЗКА: сильный черновик (однопроходка `deepseek-v4-pro`, арм D фазы 2а) дешёвый
фиксер по флагу. Проверяет прямо противоположную нынешней ставку: не «дешёвый черновик +
дорогой редактор», а «дорогой черновик + дешёвая добивка».
G МАРШРУТИЗАЦИЯ: дешёвый черновик детерминированные гейты скорят КАЖДУЮ единицу стиль-пасс
уходит ТОЛЬКО флагнутым. Проверяет, можно ли не платить за переписывание там, где переписывать
нечего.
Почему обе считаются здесь, а не по отдельности. Ф2а показала, что связка «черновик + редактор»
доминируется однопроходкой по цене (0.80×) и не выигрывает у неё по качеству. Значит вопрос
сместился: не «нужен ли второй проход», а «можно ли платить за него ИЗБИРАТЕЛЬНО». E и G два
разных способа платить избирательно, и сравнивать их надо с уже намеренными A, D и F.
Форма арма G исправлена против буквы промта по результату калибровки §2.5, и это объявлено
там же. Промт предписывал «принимать правку, только если внешний гейт видит улучшение»; замер
на 91 реальной правке показал, что гейт улучшения НЕ ВИДИТ (47%, p=0.675) при том, что порчу
видит решительно. Правило в исходной форме отсекало бы половину полезной работы по причине,
которую гейт измерить не в состоянии. Рабочая форма обратная: правка принимается по умолчанию,
отвергается только при УХУДШЕНИИ сверх порога.
Запуск: eval/.venv/bin/python eval/role_topology/route_arm.py --plan # что флагается, $0
eval/.venv/bin/python eval/role_topology/route_arm.py
"""
from __future__ import annotations
import json
import os
import statistics
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 battery as BAT # noqa: E402
import detect_word as DW # noqa: E402
import repair_arm as RA # noqa: E402
from prices import CANDIDATES, cost # noqa: E402
load_dotenv(REPO / "eval" / ".env")
OUT = Path.home() / "books" / "role-topology"
CEILING_USD = 3.00
DRAFT_COST = 0.00195 # медиана flash(low) на переводческой роли, замер §2.7
EDITOR = "deepseek-v4-pro"
def gate_flags(source: str, text: str, bank: frozenset) -> list[str]:
"""ДЕТЕРМИНИРОВАННЫЙ гейт единицы: что в ней объективно не так. Ровно те классы, которые
батарея Ф0.5 умеет находить БЕЗ судьи и без модели иначе маршрутизация стоила бы денег."""
why = []
r = BAT.run_unit(BAT.Unit(source=source, draft="", final=text))
if r["cjk_leak"]["han_chars"]:
why.append(f"утечка иероглифов ({r['cjk_leak']['han_chars']})")
if r["typography"]["violations"]:
why.append(f"типографика ({r['typography']['violations']})")
if r["numbers"]["lost_n"] or r["numbers"]["invented_n"]:
why.append(f"величины ({r['numbers']['lost_n']}/+{r['numbers']['invented_n']})")
if r["palladius"]["nonconformant_text"]:
why.append(f"не-палладиевские имена ({r['palladius']['nonconformant_text']})")
bad = [w for w in BAT.words(text) if DW.verdict(w, bank)[0]]
if bad:
why.append(f"выдуманные слова ({len(bad)})")
return why
def main() -> None:
plan_only = "--plan" in sys.argv
bank = DW.load_bank()
us = BO.units()
print("ГЕЙТ-СКОРИНГ КАЖДОЙ ЕДИНИЦЫ (детерминированный, $0)\n")
print(f"{'ед.':4s}{'F черновик':>42s}{'D однопроходка':>42s}")
print("-" * 90)
flagged_F, flagged_D = [], []
for ui, u in enumerate(us):
f_why = gate_flags(u["source"], u["draft"], bank)
d_txt = BO.text_of("D", u, ui)
d_why = gate_flags(u["source"], d_txt, bank) if d_txt.strip() else ["нет выхода"]
if f_why:
flagged_F.append(ui)
if d_why:
flagged_D.append(ui)
print(f"u{ui:<3d}{('; '.join(f_why) or 'чисто'):>42s}{('; '.join(d_why) or 'чисто'):>42s}")
print(f"\nфлагнуто гейтом: черновик F {len(flagged_F)}/{len(us)} · "
f"однопроходка D {len(flagged_D)}/{len(us)}")
# ЭКОНОМИКА МАРШРУТИЗАЦИИ считается детерминированно, без единого вызова: цена единицы известна
# из замеров Ф2а, а доля флагнутых — из гейта выше.
ed = statistics.median(json.loads((OUT / f"bo-A-u{i}.json").read_text(encoding="utf-8"))
["cost_usd"] for i in range(len(us))
if (OUT / f"bo-A-u{i}.json").exists())
d_cost = statistics.median(json.loads((OUT / f"bo-D-u{i}.json").read_text(encoding="utf-8"))
["cost_usd"] for i in range(len(us))
if (OUT / f"bo-D-u{i}.json").exists())
share = len(flagged_F) / len(us)
print(f"\n{'топология':46s}{'$/единицу':>11s}{'против A':>10s}")
print("-" * 68)
rows = [
("A — черновик + редактор ВСЕГДА (боевая)", DRAFT_COST + ed),
("D — однопроходка сильной моделью", d_cost),
(f"G — черновик + редактор ТОЛЬКО флагнутым ({share:.0%})", DRAFT_COST + ed * share),
("E — однопроходка + дешёвая починка по флагу", d_cost + 0.000275 * 2),
("F — черновик как есть", DRAFT_COST),
]
base = rows[0][1]
for name, v in rows:
print(f"{name:46s}{v:11.5f}{v / base:9.2f}×")
if plan_only:
return
# АРМ E: сильный черновик прогоняется через те же реальные детекторы и чинится дешёвым фиксером.
cl = OpenAI(api_key=os.environ[CANDIDATES[RA.FIXER][1]], base_url=CANDIDATES[RA.FIXER][0],
timeout=300)
spent = 0.0
fixed = found = 0
for ui, u in enumerate(us):
d_txt = BO.text_of("D", u, ui)
if not d_txt.strip():
continue
hits = RA.detect(d_txt, bank)
found += len(hits)
for hi, h in enumerate(hits):
rec = RA.fix(cl, u["source"], h["sentence"], h["cls"], f"e-u{ui}-{hi}")
spent += rec["cost_usd"]
fixed += bool(rec["content"].strip())
time.sleep(0.15)
print(f"\nАРМ E: реальные детекторы нашли в сильном черновике {found} дефектов, "
f"починено {fixed}, потрачено ${spent:.6f}")
print("⇒ сравнивать с армом C: там на ДЕШЁВОМ черновике те же детекторы находили 8 из 8 "
"посаженных;\n здесь счёт идёт по ЕСТЕСТВЕННЫМ дефектам, и их число само есть "
"мера качества базы.")
(OUT / "route-arm.json").write_text(json.dumps(
dict(flagged_F=flagged_F, flagged_D=flagged_D, e_found=found, e_fixed=fixed,
e_cost=spent), ensure_ascii=False), encoding="utf-8")
if __name__ == "__main__":
main()

View file

@ -0,0 +1,334 @@
#!/usr/bin/env python3
"""Ф1 — СКРИН МОДЕЛЕЙ. Оси = гейты цены и дисциплины, НЕ прокси качества.
Что это и чего это НЕ делает. Скрин отсеивает кандидатов, которые в роли неработоспособны: рвут
формат, эхают исходник, отказываются, недетерминированы или дороги сверх смысла. Он НЕ ранжирует
по качеству прозы урок эксп-20 прямой: «число абзацев было ПРОКСИ, а не качеством», и слепой суд
опроверг механический показатель. Поэтому спорных кандидатов стиль-ролей скрин не решает: они
доводятся до Ф2 обоими.
Пороги записаны ЗДЕСЬ, в коде, ДО первого замера иначе «порог» превращается в подгонку под
полученные числа. Скрин ОБЯЗАН убивать: на Ф2 проходит не больше трёх жильцов на роль.
Оси и их пороги:
Г1 ФОРМАТ на БОЕВОЙ ЕДИНИЦЕ (не коротком окне) сервис-преамбула, markdown-заголовки, обрыв.
Провал: 1 нарушение более чем в одной единице из двух. Основание: это ровно те классы,
которые боевой санитайзер считает вердикт-гейтом (D30.3), и модель, которая их производит,
требует постпроцессинга, которого в роли нет.
Г2 ЭХО/УТЕЧКА доля исходной письменности в выходе. Провал: любая единица выше боевого порога
0.15 (`disposition.go:224`) ЛИБО ханьцзы в обеих единицах. Второе строже боевого сознательно:
эксп-20 §5.4 нашёл живую микро-утечку, которую порог 0.15 не видит.
Г3 ОТКАЗ на чувствительном фрагменте. Провал: отказ. Вебновелла гарантированно упрётся в насилие,
и тихий отказ на главе N ломает конвейер.
Г4 ДЕТЕРМИНИЗМ при T=0 два побайтно одинаковых вызова. Не провал сам по себе (сэмплинг у
reasoning-моделей стохастичен), но печатается: недетерминированный жилец удорожает
воспроизводимость прогона, а она в целях продукта (§6 канона).
Г5 ЦЕНА в боевой конфигурации p50/p95 за единицу, из `usage`, а не из прайса. Расхождение с
прайсом само по себе находка.
Г6 ПРОФИЛЬ РАЗМЫШЛЕНИЯ доля в completion и управляемость. Провал: размышление не ограничивается
и выход обрывается (`finish=length` при пустом content) класс, который уже стоил проекту
волны (quirks §10).
Запуск: eval/.venv/bin/python eval/role_topology/screen.py --dry # план и смета, $0
eval/.venv/bin/python eval/role_topology/screen.py
"""
from __future__ import annotations
import json
import os
import re
import statistics
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 editor_wire_probe as P # noqa: E402
import probe as Q # noqa: E402
import inject_probe as IP # noqa: E402
from prices import CANDIDATES, billed_output, cost, out_price_order # noqa: E402
load_dotenv(REPO / "eval" / ".env")
OUT = Path.home() / "books" / "role-topology"
OUT.mkdir(parents=True, exist_ok=True)
# ⚠ ПОТОЛОК ПОДНЯТ 06.08 СЛОВОМ ВЛАДЕЛЬЦА («продолжай, потолки подними где нужно») с $1.00
# до $1.15 — ровно на величину уже случившегося перерасхода ($1.066428, §2.7) плюс арм боевой
# конфигурации flash (~$0.02). Объявлено ДО траты, как требует норма фаз.
CEILING_USD = 1.15
HARD_STOP = 1.12
# Леджер модульного уровня: `call` обязан знать потраченное, чтобы гард стоял перед ПОКУПКОЙ.
# ⚠ ИНИЦИАЛИЗИРУЕТСЯ ИЗ УЖЕ КУПЛЕННОГО НА ДИСКЕ. Без этого пере-прогон считает, что потрачено
# ноль, и докупает сверх потолка — именно так потолок Ф1 был пробит на $0.066 (см. отчёт §3):
# я перенёс гард к покупке, но забыл, что «потрачено» переживает процесс, а память — нет.
_LEDGER = {"spent": 0.0}
def _init_ledger() -> None:
tot = 0.0
for f in OUT.glob("scr-*.json"):
try:
r = json.loads(f.read_text(encoding="utf-8"))
except Exception: # noqa: BLE001, S112
continue
if not r.get("skipped"):
tot += r.get("cost_usd", 0.0)
_LEDGER["spent"] = tot
# ⚠ РЕЗ ПО ПРАЙСУ, объявленный до замера. Кандидатов 15, промт разрешает не более 14 и требует
# резать по цене с объявлением. Снимаются три САМЫХ ДОРОГИХ по выходу, и ни один из них не несёт
# уникальной способности: kimi-k3 ($15/1M выход) — вдобавок самый многословный ростера (quirks:
# ~11k токенов размышления на абзац), gemini-3.1-pro ($12) дублируется более дешёвым 3.6-flash
# того же семейства, gpt-5.6-terra ($12) — более дешёвым luna того же семейства. Семейства при
# этом НЕ теряются: каждое представлено хотя бы одним жильцом, и требование «дешёвый + сильный
# тир каждого провайдера» соблюдено.
CUT = {"kimi-k3", "gemini-3.1-pro-preview", "gpt-5.6-terra"}
OPENAI_FAMILY = {"gpt-5.6-luna", "gpt-5.6-terra"}
# Роли, под которые скринится. Разные роли — разные оси: переводчик обязан не эхать, редактор
# обязан держать формат и банк.
ROLES = ("translator", "editor")
def screened() -> list[str]:
return [m for m in out_price_order() if m not in CUT]
def client(model: str) -> OpenAI:
base, env, *_ = CANDIDATES[model]
return OpenAI(api_key=os.environ[env], base_url=base, timeout=900)
def call(model: str, msgs: list[dict], tag: str, temperature: float = 0.0,
effort: str | None = None) -> dict:
"""Один замер. Идемпотентен по тегу: пере-запуск не пере-покупает.
`effort` БОЕВАЯ КОНФИГУРАЦИЯ модели, если она у неё есть. Скрин по умолчанию зовёт всех
на вендор-дефолте, и это честный вопрос «работает ли модель из коробки»; но промт требует
мерить цену и дисциплину В БОЕВОЙ конфигурации, а у `deepseek-v4-flash` боевая включает
`reasoning_effort:"low"` без ручки размышление съедает бюджет и выход пуст (quirks §10,
подтверждено этим же скрином: 4 вызова из 4 дали `length` при нулевом содержимом).
"""
f = OUT / f"scr-{tag}.json"
if f.exists():
return json.loads(f.read_text(encoding="utf-8"))
if _LEDGER["spent"] > HARD_STOP:
return dict(tag=tag, model=model, model_returned="", finish="skipped",
prompt_tokens=0, cached_tokens=0, completion_tokens=0, reasoning_tokens=0,
reasoning_chars=0, content="", latency_s=0.0, cost_usd=0.0, skipped=True)
kw: dict = dict(model=model, messages=msgs)
if effort:
kw["extra_body"] = {"reasoning_effort": effort}
if model in OPENAI_FAMILY:
kw["max_completion_tokens"] = 16000 # quirks 00: не max_tokens у reasoning-моделей
else:
kw["max_tokens"] = 16000
kw["temperature"] = temperature
t0 = time.time()
try:
r = client(model).chat.completions.create(**kw)
except Exception as e: # noqa: BLE001
# ⚠ Обёртка добавлена по живому наблюдению: у glm-4.7 редакторский вызов занял 898.5 с
# при таймауте 900 с — следующий такой уронил бы весь прогон и потерял уже купленное.
# Провал ОДНОГО вызова — это данные (модель не укладывается в бюджет времени роли),
# а не повод останавливать скрин. Записывается как клетка с отказом провода.
rec = dict(tag=tag, model=model, model_returned="", finish="error",
error=f"{type(e).__name__}: {str(e)[:200]}",
prompt_tokens=0, cached_tokens=0, completion_tokens=0, reasoning_tokens=0,
reasoning_chars=0, content="", latency_s=round(time.time() - t0, 1),
cost_usd=0.0)
f.write_text(json.dumps(rec, ensure_ascii=False, indent=1), encoding="utf-8")
return rec
ch, u = r.choices[0], r.usage
cached = getattr(getattr(u, "prompt_tokens_details", None), "cached_tokens", 0) or 0
rec = dict(tag=tag, model=model, model_returned=r.model, finish=ch.finish_reason,
prompt_tokens=u.prompt_tokens or 0, cached_tokens=cached,
completion_tokens=u.completion_tokens or 0,
reasoning_tokens=getattr(getattr(u, "completion_tokens_details", None),
"reasoning_tokens", 0) or 0,
reasoning_chars=len(getattr(ch.message, "reasoning_content", None) or ""),
content=ch.message.content or "", latency_s=round(time.time() - t0, 1),
cost_usd=round(cost(model, u.prompt_tokens or 0, cached,
u.completion_tokens or 0,
getattr(getattr(u, "completion_tokens_details", None),
"reasoning_tokens", 0) or 0), 6))
f.write_text(json.dumps(rec, ensure_ascii=False, indent=1), encoding="utf-8")
_LEDGER["spent"] += rec["cost_usd"]
return rec
# ─────────────────────────── гейты (детерминированные) ───────────────────────────
_RE_HAN = re.compile(r"[㐀-鿿]")
_RE_MD = re.compile(r"^\s{0,3}#{1,6}\s", re.M)
_RE_PREAMBLE = re.compile(r"^\s*(вот|ниже|here is|here's)\b[^\n:]{0,80}:", re.I)
_RE_REFUSAL = re.compile(
r"не могу|не буду|отказ|прости(те)?, но|cannot (assist|help|comply)|i'm sorry", re.I)
def gates(rec: dict, source: str) -> dict:
t = rec["content"]
letters = [c for c in t if c.isalpha()]
han = len(_RE_HAN.findall(t))
return dict(
empty=not t.strip(),
truncated=rec["finish"] == "length",
md_headers=len(_RE_MD.findall(t)),
preamble=bool(_RE_PREAMBLE.search(t)),
han_chars=han,
han_share=round(han / max(1, len(letters)), 5),
refusal=bool(_RE_REFUSAL.search(t[:400])),
len_ratio=round(len(t) / max(1, len(source)), 3),
# ⚠ Доля считается от ТАРИФИЦИРУЕМОГО выхода, а не от completion: у аддитивных
# провайдеров (xAI) отношение к completion даёт бессмыслицу вроде 203.8%.
reasoning_share=round(rec["reasoning_tokens"] / max(1, billed_output(
rec["model"], rec["completion_tokens"], rec["reasoning_tokens"])), 3),
)
def verdict(rows: list[dict]) -> tuple[str, list[str]]:
"""Пороги применяются как объявлены в шапке. Возвращает (вердикт, список причин).
Пустой список это НЕ провал, а отсутствие замера, и различать их обязательно: первая
редакция на пустых строках давала `0 >= 0` и печатала «утечка ханьцзы» модели, которую
жёсткий стоп не дал купить вовсе. Незамеренная модель, объявленная провалившейся, ложное
утверждение о вендоре, и поймано оно только сверкой невозможного числа с сырьём.
"""
if not rows:
return "НЕ ЗАМЕРЕН", ["потолок кончился до этой модели"]
why = []
fmt_bad = sum(1 for r in rows if r["md_headers"] or r["preamble"] or r["empty"]
or r["truncated"])
if fmt_bad > 1:
why.append(f"Г1 формат: нарушений в {fmt_bad} единицах")
if any(r["han_share"] > 0.15 for r in rows):
why.append("Г2 эхо: единица выше боевого порога 0.15")
elif sum(1 for r in rows if r["han_chars"]) >= len(rows):
why.append("Г2 утечка: ханьцзы во ВСЕХ единицах")
if any(r["refusal"] for r in rows):
why.append("Г3 отказ на чувствительном фрагменте")
if any(r["truncated"] and r["empty"] for r in rows):
why.append("Г6 размышление съело бюджет, выход пуст")
return ("ПРОВАЛ" if why else "прошёл"), why
def load_units() -> list[dict]:
"""БОЕВЫЕ единицы редактуры из корпуса пакета-6, а не короткие окна пробы 18.
Первая редакция брала окна `pick_windows()` по ~500 знаков источника. Промт требует мерить
формат-дисциплину «НА ДЛИННЫХ входах (боевая единица, не короткое окно)», а реальная единица
прогона несёт медианно 1683 знака источника и 5594 черновика, то есть в 3.4 раза больше.
На коротком окне модель формат не рвёт, и скрин показал бы всем «прошёл» то есть не убивал
бы никого, а промт требует, чтобы скрин убивал. Поймано замером длин до первого вызова.
Берутся две единицы около 70-го и 80-го процентиля длины боевой размер, но не выброс.
Выбор детерминированный (сортировка по длине), случайности здесь не нужно.
"""
corp = Path.home() / "books" / "gu-zhenren" / "labels" / "raw" / "corpus.jsonl"
us = [json.loads(l) for l in corp.read_text(encoding="utf-8").splitlines() if l.strip()]
us = [u for u in us if all((u.get(k) or "").strip() for k in ("source", "draft"))]
order = sorted(range(len(us)), key=lambda i: len(us[i]["source"]))
return [us[order[int(0.7 * len(order))]], us[order[int(0.8 * len(order))]]]
def main() -> None:
dry = "--dry" in sys.argv
_init_ledger()
if _LEDGER["spent"]:
print(f"уже куплено скрином ранее: ${_LEDGER['spent']:.6f} — гард считает от этой суммы")
models = screened()
units = load_units()
n_calls = len(models) * len(units) * len(ROLES)
print(f"кандидатов после реза: {len(models)} из {len(CANDIDATES)} "
f"(снято по прайсу: {', '.join(sorted(CUT))})")
print(f"единиц {len(units)} · ролей {len(ROLES)} · вызовов {n_calls} · потолок ${CEILING_USD}")
for u in units:
print(f" единица {u['run']}:{u['chapter']}:{u['chunk_idx']} — источник "
f"{len(u['source'])} зн, черновик {len(u['draft'])} зн")
print("порядок моделей — по ВОЗРАСТАНИЮ цены выхода: если потолок кончится, непокрытыми\n"
"останутся дорогие, и это будет объявлено, а не скрыто")
for m in models:
print(f" {m}")
if dry:
return
spent = 0.0
results: dict[str, list[dict]] = {}
# АРМ БОЕВОЙ КОНФИГУРАЦИИ. Скрин зовёт всех на вендор-дефолте — это честный вопрос «работает
# ли модель из коробки». Но у `deepseek-v4-flash` боевая черновая роль ставит `reasoning_effort:
# "low"`, без которого размышление съедает бюджет (quirks §10, подтверждено скрином: 4/4
# пустых). Без этого арма вывод «flash непригоден» был бы подменой: доказано лишь «непригоден
# дефолт». Арм объявлен девиацией §3 и гонится теми же единицами и ролями.
if "--flash-low" in sys.argv:
rows = []
for role in ROLES:
for ui, u in enumerate(units):
src, draft = u["source"], u["draft"]
if role == "translator":
msgs = Q.messages("direct", src, draft, None)
else:
terms = [t for t in IP.TERMS if t in src]
blk = P.editor_block(terms, P.HEADER_LAW_CLAUSE, None) if terms else None
msgs = Q.messages("full", src, draft, blk)
rec = call("deepseek-v4-flash", msgs, f"flashlow-{role}-u{ui}", effort="low")
spent += rec["cost_usd"]
rows.append(dict(role=role, unit=ui, **gates(rec, src),
cost=rec["cost_usd"], latency=rec["latency_s"]))
time.sleep(0.2)
v, why = verdict(rows)
print(f"\nАРМ БОЕВОЙ КОНФИГУРАЦИИ deepseek-v4-flash (effort=low): {v} {'; '.join(why)}")
for r in rows:
print(f" {r['role']:11s} u{r['unit']} пусто={r['empty']!s:5s} обрыв={r['truncated']!s:5s} "
f"ханьцзы={r['han_chars']:3d} размышл.={r['reasoning_share']:.0%} ${r['cost']:.5f}")
print(f"потрачено армом ${spent:.6f}")
return
for m in models:
rows = []
for role in ROLES:
for ui, u in enumerate(units):
# ⚠ Гард стоит ВНУТРИ `call` (перед покупкой), а не здесь: первая редакция
# проверяла стоп ДО обращения к кэшу, и при пере-прогоне уже КУПЛЕННЫЕ клетки
# не загружались — модель с четырьмя артефактами на диске получала вердикт
# «НЕ ЗАМЕРЕН». Гард ограничивает ПОКУПКУ, а не чтение.
src, draft = u["source"], u["draft"]
if role == "translator":
msgs = Q.messages("direct", src, draft, None)
else:
# Блок закона собирается из термов, реально встреченных в ЭТОЙ единице:
# инъекция боевого пути именно такая (D39.104), пустой блок был бы не боевым.
# Термы берутся тем же способом, что в пробе 18: из ростера `inject_probe.TERMS`
# отбираются те, что реально встречены В ЭТОЙ единице. Пустой блок был бы
# не боевым путём — ЗАКОН промта требует инъекцию во всех армах (D39.104).
terms = [t for t in IP.TERMS if t in src]
blk = P.editor_block(terms, P.HEADER_LAW_CLAUSE, None) if terms else None
msgs = Q.messages("full", src, draft, blk)
rec = call(m, msgs, f"{m}-{role}-u{ui}")
spent += rec["cost_usd"]
if rec.get("skipped"):
continue
rows.append(dict(role=role, unit=ui, **gates(rec, src),
cost=rec["cost_usd"], latency=rec["latency_s"]))
time.sleep(0.2)
results[m] = rows
print(f"\n{'модель':24s}{'вердикт':>9s}{'p50 $/ед':>10s}{'эхо':>7s}"
f"{'размышл.':>10s} причины")
print("-" * 96)
for m, rows in results.items():
v, why = verdict(rows)
p50 = statistics.median(r["cost"] for r in rows) if rows else 0
han = sum(r["han_chars"] for r in rows)
rs = statistics.median(r["reasoning_share"] for r in rows) if rows else 0
print(f"{m:24s}{v:>9s}{p50:10.5f}{han:7d}{rs:10.1%} {'; '.join(why)}")
(OUT / "screen.json").write_text(json.dumps(results, ensure_ascii=False, indent=1),
encoding="utf-8")
print(f"\nпотрачено ${spent:.6f} из ${CEILING_USD}")
if __name__ == "__main__":
main()

View file

@ -61,29 +61,72 @@ def main() -> None:
# §2.2 — QE. Числа берутся из персистированного сырья прогона, а не пере-считываются моделью
# (пере-счёт стоил бы 20 минут CPU и не добавил бы проверки: артефакт и есть сырьё).
seg = RAW / "qe-segments.json"
dec = RAW / "qe-decoy.json"
if not seg.exists() or not dec.exists():
ck("§2.2 сырьё QE на месте", False, "нет qe-segments.json / qe-decoy.json")
else:
# ⚠ Артефакты РАЗВЕДЕНЫ ПО МОДЕЛЯМ. Первая редакция харнесса писала в общее имя, и прогон XL
# молча затёр сырьё large — поймано этим же верификатором, отчёт бы остался «со слов сессии».
import statistics # noqa: PLC0415
expect_qe = {
"large": dict(decoy=(69, 7.107, 65), k1=(6, 4), k2=(6, 3), clean=81),
"xl": dict(decoy=(69, 7.000, 59), k1=(6, 4), k2=(6, 5), clean=81),
}
for tag, want in expect_qe.items():
seg, dec = RAW / f"qe-segments-{tag}.json", RAW / f"qe-decoy-{tag}.json"
if not seg.exists() or not dec.exists():
ck(f"§2.2 сырьё QE ({tag}) на месте", False, f"нет qe-*-{tag}.json")
continue
d = json.loads(dec.read_text(encoding="utf-8"))
ck("§2.2 декой n=69, медиана +7.107, направление 65/69",
len(d) == 69 and abs(statistics.median(d) - 7.107) < 0.01
and sum(1 for x in d if x > 0) == 65,
wn, wm, wp = want["decoy"]
ck(f"§2.2 {tag}: декой n={wn}, медиана {wm:+.3f}, направление {wp}/{wn}",
len(d) == wn and abs(statistics.median(d) - wm) < 0.01
and sum(1 for x in d if x > 0) == wp,
f"n={len(d)} медиана={statistics.median(d):.3f} "
f"плюсовых={sum(1 for x in d if x > 0)}")
srows = json.loads(seg.read_text(encoding="utf-8"))
by = {(r["window"], r["variant"], tuple(r["di"])): r["score"] for r in srows}
for cls, want_n, want_pos in (("k1", 6, 4), ("k2", 6, 3)):
for cls in ("k1", "k2"):
want_n, want_pos = want[cls]
deltas = [sc - by[(w, "clean", di)] for (w, v, di), sc in by.items()
if v == cls and (w, "clean", di) in by
and abs(sc - by[(w, "clean", di)]) > 1e-9]
ck(f"§2.2 парная Δ класса {cls}: {want_n} сегментов, Δ>0 в {want_pos}",
ck(f"§2.2 {tag}: парная Δ класса {cls} {want_n} сегментов, Δ>0 в {want_pos}",
len(deltas) == want_n and sum(1 for x in deltas if x > 0) == want_pos,
f"{len(deltas)} сегментов, Δ>0 в {sum(1 for x in deltas if x > 0)}")
clean = [r["score"] for r in srows if r["variant"] == "clean"]
ck("§2.2 здоровых сегментов 81", len(clean) == 81, str(len(clean)))
ck(f"§2.2 {tag}: здоровых сегментов {want['clean']}", len(clean) == want["clean"],
str(len(clean)))
# §2.5 — калибровка гейта арма G. Считается из персистированных баллов, а не пере-моделируется:
# артефакт и есть сырьё, а пере-счёт стоил бы полчаса CPU без добавления проверки.
gu = RAW / "qe-guard-units-all.json"
if not gu.exists():
ck("§2.5 сырьё калибровки на месте", False, "нет qe-guard-units-all.json")
else:
import statistics # noqa: PLC0415
from math import comb # noqa: PLC0415
rows = json.loads(gu.read_text(encoding="utf-8"))
d = [r["final"] - r["draft"] for r in rows]
better = sum(1 for x in d if x < 0)
k = max(better, len(d) - better)
p = min(1.0, 2 * sum(comb(len(d), i) for i in range(k, len(d) + 1)) / (2 ** len(d)))
ck("§2.5 единичный уровень: n=91, улучшил 43, медиана Δ 0.0000",
len(d) == 91 and better == 43 and abs(statistics.median(d)) < 1e-9,
f"n={len(d)} улучшил={better} медиана={statistics.median(d):+.4f}")
ck("§2.5 знаковый тест p = 0.6752", abs(p - 0.6752) < 0.0005, f"{p:.4f}")
ck("§2.5 паритет |Δ|<0.25 в 55 единицах",
sum(1 for x in d if abs(x) < 0.25) == 55, str(sum(1 for x in d if abs(x) < 0.25)))
# §2.2 шаг 3 — мощность парного режима.
pw = RAW / "qe-power-large.json"
if not pw.exists():
ck("§2.2 сырьё мощности на месте", False, "нет qe-power-large.json")
else:
import statistics # noqa: PLC0415
rows = json.loads(pw.read_text(encoding="utf-8"))
d = [r["flipped"] - r["base"] for r in rows]
ck("§2.2 парная мощность: 72 пары, медиана +1.490, направление 66/72",
len(d) == 72 and abs(statistics.median(d) - 1.490) < 0.001
and sum(1 for x in d if x > 0) == 66,
f"n={len(d)} медиана={statistics.median(d):+.3f} "
f"плюсовых={sum(1 for x in d if x > 0)}")
# §2.3 — батарея. Пере-прогоняется по тому же корпусу целиком: это единственный способ
# заметить, что правило поехало после правки.
@ -107,7 +150,7 @@ def main() -> None:
n = len(units)
for label, got, want in (("нарушений типографики/ед 0.18", typo / n, 0.18),
("ханьцзы всего 4", han, 4),
("потеряно величин/ед 0.25", lost / n, 0.25),
("потеряно величин/ед 0.27", lost / n, 0.275),
("появилось величин/ед 0.27", inv / n, 0.27),
("не-палладиевских срабатываний 6", nonconf, 6),
("единиц с кальками 0", calq, 0)):
@ -115,10 +158,138 @@ def main() -> None:
else str(got))
ck("§2.3 кандидатов в имена 2184", cand == 2184, str(cand))
# Дисциплина отчёта: заявленный $0 обязан подтверждаться отсутствием платного сырья.
paid = [p for p in RAW.glob("*.json") if p.name.startswith(("rt-", "j-", "screen-"))]
ck("§0 заявлено «потрачено $0» и платных артефактов нет", not paid, str([p.name for p in paid]))
ck("статус-баннер отчёта соответствует", "ПЛАТНЫЕ ФАЗЫ НЕ ЗАПУСКАЛИСЬ" in text)
# §2.6 — вахта эффорта. Числа выводятся из сырья вызовов, а не из моей сводки.
ew = sorted(RAW.glob("ew-*.json"))
if not ew:
ck("§2.6 сырьё вахты на месте", False, "нет ew-*.json")
else:
recs = [json.loads(f.read_text(encoding="utf-8")) for f in ew]
by = {}
for r in recs:
by.setdefault(r["tag"].split("-")[1], []).append(r)
ck("§2.6 армы: дефолт 12 · low 12 · xhigh 1",
(len(by.get("default", [])), len(by.get("low", [])), len(by.get("xhigh", [])))
== (12, 12, 1),
str({k: len(v) for k, v in by.items()}))
ck("§2.6 все вызовы finish=stop", all(r["finish"] == "stop" for r in recs),
str(sorted({r["finish"] for r in recs})))
med = {k: statistics.median(r["reasoning_chars"] for r in v) for k, v in by.items()}
ck("§2.6 reasoning медианы: дефолт 2645.5 · low 2514 · xhigh 27185",
(med.get("default"), med.get("low"), med.get("xhigh")) == (2645.5, 2514, 27185),
str(med))
pt = {k: sorted({r["prompt_tokens"] for r in v}) for k, v in by.items()}
ck("§2.6 prompt_tokens: дефолт и low = 1945, xhigh = 2024 (серверная реакция)",
pt.get("default") == [1945] and pt.get("low") == [1945] and pt.get("xhigh") == [2024],
str(pt))
spent = sum(r["cost_usd"] for r in recs)
ck("§2.6 потрачено $0.047361 и потолок $0.05 НЕ пробит",
abs(spent - 0.047361) < 1e-6 and spent <= 0.05, f"${spent:.6f}")
ck("§2.6 запрос побайтно один во всех вызовах",
len({json.dumps(r["messages"], ensure_ascii=False, sort_keys=True)
for r in recs}) == 1, "запросы различаются — интерливинг недействителен")
ck("отчёт заявляет ту же сумму", "0.047361" in text)
# §2.7 — Ф1. Цена пере-считывается ИЗ usage, а не читается из записанной: именно расхождение
# этих двух путей и вскрыло аддитивный биллинг xAI.
from prices import cost as price_of # noqa: PLC0415
allscr = [json.loads(f.read_text(encoding="utf-8")) for f in sorted(RAW.glob("scr-*.json"))]
allscr = [r for r in allscr if not r.get("skipped")]
# Арм боевой конфигурации flash считается ОТДЕЛЬНО от скрина: он куплен после и по другой
# ручке. Смешивать их — значит потерять именно то различие, ради которого арм и заводился.
scr = [r for r in allscr if not r["tag"].startswith("flashlow-")]
flashlow = [r for r in allscr if r["tag"].startswith("flashlow-")]
if not scr:
ck("§2.7 сырьё скрина на месте", False, "нет scr-*.json")
else:
ck("§2.7 клеток скрина 48", len(scr) == 48, str(len(scr)))
ck("§2.7 арм боевой конфигурации flash: 4 клетки", len(flashlow) == 4, str(len(flashlow)))
ck("§2.7 арм flash(low): выход НЕ пуст ни в одной клетке",
all(r["content"].strip() for r in flashlow),
f"{sum(1 for r in flashlow if not r['content'].strip())} пустых")
recomputed = sum(price_of(r["model"], r["prompt_tokens"], r["cached_tokens"],
r["completion_tokens"], r["reasoning_tokens"])
for r in allscr)
ck("§2.7 пере-счёт цены Ф1 из usage даёт $1.074724",
abs(recomputed - 1.074724) < 1e-5, f"${recomputed:.6f}")
ck("§2.7 отчёт признаёт превышение потолка Ф1", "1.074724" in text and "7.5%" in text)
ck("§2.7 записанная цена совпадает с пере-счётом (аддитивный биллинг применён)",
abs(sum(r["cost_usd"] for r in allscr) - recomputed) < 1e-5,
f"записано ${sum(r['cost_usd'] for r in allscr):.6f}")
flash = [r for r in scr if r["model"] == "deepseek-v4-flash"]
ck("§2.7 deepseek-v4-flash на ДЕФОЛТЕ: 4/4 пустых при finish=length",
len(flash) == 4 and all(r["finish"] == "length" and not r["content"].strip()
for r in flash),
f"{sum(1 for r in flash if not r['content'].strip())}/{len(flash)} пустых")
grok = [r for r in scr if r["model"] == "grok-4.5"]
ck("§2.7 у grok-4.5 размышление ПРЕВЫШАЕТ completion (улика аддитивности)",
all(r["reasoning_tokens"] > r["completion_tokens"] for r in grok))
# Дисциплина отчёта: заявленные деньги обязаны сходиться с сырьём двумя путями.
ck("статус-баннер отчёта соответствует состоянию фаз",
"ВСЕ ФАЗЫ ИСПОЛНЕНЫ" in text)
# §2.9/§2.11 — бейк-офф и маршрутизация. Вердикты пере-считываются из ПЕРСИСТИРОВАННЫХ голосов,
# а не читаются из сводки: сводку писал я, голоса писал провод.
import judges as JJ # noqa: PLC0415
import bakeoff as BO # noqa: PLC0415
us = BO.units()
ck("§2.9 единиц бейк-оффа 8", len(us) == 8, str(len(us)))
for a, b, want in (("A", "F", (6, 0)), ("B", "F", (7, 0)),
("D", "A", (2, 0)), ("D", "D_", (3, 0))):
wa = wb = 0
for ui in range(len(us)):
pj = {}
for judge in JJ.JUDGES:
marg = []
for rep in (1, 2, 3):
o = (rep - 1) % 2
f = RAW / f"jv-bo-{a}v{b}-u{ui}-o{o}-{judge}-r{rep}.json"
if not f.exists():
continue
p = JJ.parse(json.loads(f.read_text(encoding="utf-8"))["content"])
if p["В1-ВЕРНОСТЬ"] is None:
continue
e1 = sum(p[f"В1-{x}"] or 0 for x in JJ.AXES)
e2 = sum(p[f"В2-{x}"] or 0 for x in JJ.AXES)
marg.append((e2 - e1) if o == 0 else (e1 - e2))
if len(marg) >= 2:
pos = sum(1 for m in marg if m > 0)
neg = sum(1 for m in marg if m < 0)
pj[judge] = 1 if pos >= 2 and pos > neg else (
-1 if neg >= 2 and neg > pos else 0)
if len(pj) < 2:
continue
v = set(pj.values())
wa += v == {1}
wb += v == {-1}
ck(f"§2.9 контраст {a} против {b} = {want[0]}:{want[1]}", (wa, wb) == want, f"{wa}:{wb}")
# §2.12 — карточки персонажей. Пин на то, что проверка рода НЕ инертна: пустые карточки
# молча возвращают нули, и именно так этот пробел прожил весь пак незамеченным.
cards = B.load_cards()
ck("§2.12 карточки построены из подписанного сида (51 ключ)", len(cards) == 51,
str(len(cards)))
ck("§2.12 проверка рода НЕ инертна на реальном тексте",
B.check_gender(units[0]["final"], cards)["checked"] > 0,
"0 сверок — карточки не доехали")
ra = RAW / "repair-arm.json"
if ra.exists():
r = json.loads(ra.read_text(encoding="utf-8"))
ck("§2.10 арм C: oracle 16/16, реальные 8/16",
r["oracle"] == [16, 16] and r["real"] == [8, 16], str(r))
else:
ck("§2.10 сырьё арма C на месте", False, "нет repair-arm.json")
rt = RAW / "route-arm.json"
if rt.exists():
r = json.loads(rt.read_text(encoding="utf-8"))
ck("§2.11 гейт флагает 5 черновиков из 8", len(r["flagged_F"]) == 5,
str(len(r["flagged_F"])))
ck("§2.11 арм E: найдено 6 естественных дефектов, починено 6",
(r["e_found"], r["e_fixed"]) == (6, 6), f"{r['e_found']}/{r['e_fixed']}")
else:
ck("§2.11 сырьё армов E/G на месте", False, "нет route-arm.json")
print(f"\n{'ВСЕ ЧИСЛА ОТЧЁТА СХОДЯТСЯ' if not FAILS else f'РАСХОЖДЕНИЙ: {FAILS}'}")
sys.exit(1 if FAILS else 0)