textmachine/eval/dovodka/rol.py

388 lines
26 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

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

#!/usr/bin/env python3
"""ОДНОПРОХОДКА КАК РОЛЬ — новый промт против прежней склейки, на нескольких моделях. Платный.
⚠ ЗАЧЕМ. Однопроходка — единственный живой продуктовый кандидат из всего, что дуга 19→23
измерила: по судейской оси от боевой связки не отличается, по банку терминов ЛУЧШЕ неё на обеих
парах (потерянных терминов на главу 1.38 против 2.88 на zh, 0.31 против 0.56 на en), стоит ОДИН
вызов вместо двух, и при слепом чтении владелец поставил её ПЕРВОЙ на английском.
⚠⚠ И всё это она делает С ГАНДИКАПОМ. Её промт (`contour.a2_msgs`) — не продуктовый, а
ИЗМЕРИТЕЛЬНЫЙ: боевой промт переводчика плюс мандатная часть боевого промта редактора,
механически склеенные (четыре строки выброшены, четыре подставлены), с прямой мета-фразой про наш
конвейер — «отдельного редактора после тебя НЕ БУДЕТ». Склейка сделана по уважительной причине:
в эксп-21 у однопроходки системный промт был вдвое короче, и замеренная разница могла быть
разницей ОБЪЁМА ИНСТРУКЦИИ, а не топологии. Уравнивание было правильным ДЛЯ ЗАМЕРА — но
следствие в том, что однопроходка ни разу не испытывалась как нормально написанный промт-роль.
ЧТО ПОКУПАЕТСЯ (имена армов ОДНИ на обе пары, пара живёт в данных):
ZP ПРЕЖНЯЯ СКЛЕЙКА, deepseek-v4-pro БЕСПЛАТНО, куплена заходом Д17 — это база
Z0 боевая связка (черновик → перепис) БЕСПЛАТНО — эталон продакшена
ZF вторая генерация связки БЕСПЛАТНО — пол СТАРОЙ панели
ZD голый черновик БЕСПЛАТНО — якорь низа
RN НОВАЯ РОЛЬ, deepseek-v4-pro, генерация 1 покупается — несущий арм
RN2 НОВАЯ РОЛЬ, deepseek-v4-pro, генерация 2 покупается — СВОЙ пол новой панели
RG НОВАЯ РОЛЬ, glm-5 покупается — переносимость, другой вендор
RK НОВАЯ РОЛЬ, grok-4.3 покупается — переносимость, третья семья
⚠ ПЕРВИЧНЫЙ КОНТРАСТ ОДИН: `RN/ZP` — новый промт против прежней склейки НА ТОЙ ЖЕ МОДЕЛИ и ТЕХ ЖЕ
главах. Всё остальное описательное. Одного контраста требует мощность: при n=16 и поправке Холма
по четырём контрастам MDE вырос бы выше цели, и ось не дала бы вывода ни при каком исходе — ровно
та болезнь, что погубила английскую ось эксп-22.
⚠ ТРЕТЬЯ МОДЕЛЬ — `grok-4.3`, и выбор не случаен. Главная претензия владельца при слепом чтении
была РЕГИСТР: «смягчённый мат там, где на мате держится сцена». xAI — единственный вендор, у
которого слова `violen*` в правилах нет вовсе, и модель меньше прочих склонна смягчать.
⚠ `gemini` в панель НЕ берётся: на этом материале он массово падает в контент-фильтр (эксп-21:
10 клеток судейского прогона с `PROHIBITED_CONTENT`), и его пустые клетки мерили бы фильтр, а не
промт. `grok-4.3` идёт с ВКЛЮЧЁННЫМ размышлением (вендор-дефолт `low`): запрет D30.1 прямой —
grok с reasoning-off в редакторской роли есть no-op-редактор.
Запуск: rol.py --show [zh|en] | --dry | --selftest | --buy [zh|en] [армы…] | --canon
"""
from __future__ import annotations
import importlib.util
import json
import statistics as st
import sys
from pathlib import Path
REPO = Path("/home/ubuntu/projects/textmachine")
ZONE = Path(__file__).resolve().parent
for d in ("dovodka", "editor_tier", "tenant_panel", "role_topology", "bank_arbitration"):
sys.path.insert(0, str(REPO / "eval" / d))
def _load(name: str, path: Path):
spec = importlib.util.spec_from_file_location(name, path)
mod = importlib.util.module_from_spec(spec)
sys.modules[name] = mod
spec.loader.exec_module(mod)
return mod
# ⚠ Контур берётся У ДРАЙВЕРА английской оси, а не грузится вторым экземпляром: `en_axis` держит
# СВОЙ `contour.py`, и `wire()` настраивает хуки именно в нём. Свой второй экземпляр остался бы с
# китайскими хуками, и английская ось молча стала бы пустой. Класс уже стоил фазе оплаченных клеток.
EN = _load("en_axis_r", ZONE / "en_axis.py")
C = EN.C
MONEY, PR, ROSTER, ED22, PARA = C.MONEY, C.PR, C.ROSTER, C.ED22, C.PARA
OUT = MONEY.OUT
ZK = _load("zakhod_r", ZONE / "zakhod.py") # гейт перелома цены, пред-полёт, отбор глав
CORE = ZONE / "prompts" / "onepass-core.md"
MAX_OUT = 32000 # как в заходе Д17: обрыв на потолке стоит денег и не даёт данных
# ── ПАРА КАК ДАННЫЕ. Ни одна ветка ниже не спрашивает «а это китайский?» ──────────────────────
PAIRS = {
"zh": {"letter": "m", "phase": "ФД-M", "onepass": "A2",
"wire": lambda: C.configure(PARA.ZH), "units": lambda: C.units_ok(),
"para": lambda: PARA.ZH},
"en": {"letter": "n", "phase": "ФД-N", "onepass": "E2",
"wire": EN.wire, "units": EN.units, "para": lambda: PARA.EN},
}
# арм → (модель, номер генерации). Генерация нужна, чтобы два прогона ОДНОЙ модели не схлопнулись
# в один тег: кэш `purchase` смотрит на существование файла и вернул бы первую клетку второй раз.
ARMS = {"RN": ("deepseek-v4-pro", 1), "RN2": ("deepseek-v4-pro", 2),
"RG": ("glm-5", 1), "RK": ("grok-4.3", 1)}
FREE = ("ZD", "Z0", "ZP")
PRIMARY = ("RN", "ZP") # единственный несущий контраст, см. шапку
# ── СБОРКА ПРОМТА ────────────────────────────────────────────────────────────────────────────
# ⚠ ПАР-БЛОКИ НЕ ДУБЛИРУЮТСЯ, А ПОДТЯГИВАЮТСЯ ИЗ ФАЙЛОВ САМОЙ ПАРЫ. Второй экземпляр «Единиц и
# приёмов» разъехался бы с боевым молча, и замер мерил бы промт, которого в бою нет. Гейт
# пар-промтов сторожит те файлы, а не мою копию, — значит копии быть не должно.
UNITS_HEAD = "Единицы и приёмы"
REFLOW_HEAD = "ДИСКУРС-ПЕРЕВЁРСТКА"
def _block(core: str, head: str) -> str:
"""Секция от заголовка до пустой строки перед следующим заголовком верхнего уровня."""
i = core.find(head)
if i < 0:
return ""
tail = core[i:]
# секция кончается там, где начинается строка «Выведи ТОЛЬКО» или следующий ЗАГОЛОВОК КАПСОМ
stop = len(tail)
for mark in ("\nВыведи ТОЛЬКО", "\nХонорифики:"):
j = tail.find(mark)
if 0 <= j < stop:
stop = j
return tail[:stop].strip()
def pair_blocks(p: str) -> tuple[str, str]:
"""(«Единицы и приёмы» пары, «ДИСКУРС-ПЕРЕВЁРСТКА» пары) — из её же боевых файлов."""
d = PR.PAIR_DIR[PAIRS[p]["para"]().key]
tr_core, _f, _u = PR.load_template(d / "translator.md")
ed_core, _f2, _u2 = PR.load_template(d / "editor.md")
return _block(tr_core, UNITS_HEAD), _block(ed_core, REFLOW_HEAD)
def rol_msgs(p: str, src: str) -> list[dict]:
"""Однопроходка НОВЫМ промтом: роль, рамки, регистр, критерии проверки, пар-блоки, глоссарий."""
cfg = PAIRS[p]
cfg["wire"]()
key = cfg["para"]().key
units_blk, reflow_blk = pair_blocks(p)
if not units_blk or not reflow_blk:
raise SystemExit(f"у пары {p} нет блока «{UNITS_HEAD}» либо «{REFLOW_HEAD}» — "
"промт собрать нечем, арм не строится")
core, _few, user = PR.load_template(CORE)
core = (core.replace("{{ПАР_ЕДИНИЦЫ}}", units_blk)
.replace("{{ПАРЕРЕВЁРСТКА}}", reflow_blk))
brief = getattr(PR, PR.PAIR_BRIEF[key])
sysmsg = PR.render(core, brief, src, "")
# ⚠ БАНК-ЗАКОН НА ВСЕХ АРМАХ БЕЗ ИЗЪЯТИЙ (норма D39.104). Прежнее отступление — 72 клетки без
# блока — владелец пометил как дефект. Термы берутся ТОЛЬКО релевантные этому фрагменту:
# прямое требование брифа V6 «не давать слишком много контекста».
bf = None if key == "zh-ru" else cfg["para"]().bank_file
law = PR.law_block(PR.terms_in(src, bf), bf)
out = [{"role": "system", "content": sysmsg}]
if law:
# ⚠ БАНК-ЗАКОН НАПИСАН ДЛЯ ДВУХСТАДИЙНОГО ПУТИ и говорит «в ЧЕРНОВИКЕ термин слева должен
# быть передан формой справа». У однопроходки черновика нет вовсе: модель получила бы
# указание про текст, которого не существует. Поймано селфтестом ДО покупки. Правится
# ровно одно слово и ровно здесь — трогать общий `law_block` нельзя, он едет в бою.
out.append({"role": "system", "content": law.replace("в черновике", "в переводе")})
out.append({"role": "user", "content": PR.render(user, brief, src, "")})
return out
def tag(p: str, arm: str, uid: str) -> str:
return f"dv-{PAIRS[p]['letter']}-{arm.lower()}-{uid}"
def cell(tg: str) -> str:
f = OUT / f"{tg}.json"
if not f.exists():
return ""
d = json.loads(f.read_text(encoding="utf-8"))
return (d.get("content") or "") if d.get("finish") == "stop" else ""
def free_text(p: str, arm: str, uid: str) -> str:
"""Бесплатные армы берутся у рига захода Д17 — те же главы, те же тексты, $0."""
return ZK.free_text(p, arm, uid)
def chosen(p: str) -> list[dict]:
"""ТЕ ЖЕ главы, что у захода Д17 — иначе бесплатные армы не переиспользуются и база уедет."""
return ZK.chosen(p)
def call_kwargs(model: str, msgs: list[dict]) -> dict:
"""Боевая конфигурация модели. Отступлений нет: у `grok-4.3` вендор-дефолт = размышление ON,
а запрет D30.1 прямой — reasoning-off в редакторской роли даёт no-op-редактора."""
return ROSTER.call_kwargs(model, msgs, max_out=MAX_OUT)
# ── ПОКУПКА ──────────────────────────────────────────────────────────────────────────────────
def buy(p: str, which: list[str]) -> None:
ZK.price_gate()
cfg = PAIRS[p]
cfg["wire"]()
ZK.preflight([ARMS[a][0] for a in which], cfg["phase"], cfg["letter"])
us = chosen(p)
led = MONEY.Guarded(cfg["phase"], default_expect=0.03)
print(f"\nпара {p} · глав {len(us)} · армы {' '.join(which)} · касса {cfg['phase']}")
for arm in which:
model, gen = ARMS[arm]
for u in us:
tg = tag(p, arm, u["uid"])
if cell(tg).strip():
continue
# ⚠ Гейт перелома цены — перед КАЖДОЙ клеткой, а не раз на старте: прогон длится
# часами, и пересёкший 16:00 UTC цикл писал бы вызовы по вчерашнему прайсу.
ZK.price_gate(quiet=True)
rec = MONEY.purchase(led, tg, model, ED22.client(model),
call_kwargs(model, rol_msgs(p, u["source"])),
role="onepass", uid=u["uid"], arm=arm, gen=gen, pair=p)
if rec.get("skipped"):
raise SystemExit(f"⛔ ГАРД ОТКАЗАЛ на клетке {tg} — прогон ОСТАНОВЛЕН.\n"
" Резать панель или поднимать потолок решением сессии ЗАПРЕЩЕНО.\n"
" Это СТОП и вопрос владельцу: назвать недостающую сумму.")
if rec.get("finish") != "stop":
print(f"{tg}: finish={rec.get('finish')} "
f"{str(rec.get('error'))[:120]} — клетка объявляется несобранной")
elif C.draft_defect(cell(tg)):
# штатный гейт годности проекта: эхо исходника вместо перевода ловится ЗДЕСЬ,
# иначе негодная клетка уедет в судейскую пачку за полную цену
print(f"{tg}: {C.draft_defect(cell(tg))} — клетка НЕГОДНА, объявить в отчёте")
MONEY.report()
# ── $0-ГЕЙТ КАНОНА: там, где однопроходка уже выигрывала ──────────────────────────────────────
def canon() -> None:
print("КАНОН-ГЕЙТ ($0, детерминированно): потерянных терминов банка на главу, МЕНЬШЕ лучше\n")
for p, cfg in PAIRS.items():
cfg["wire"]()
us = chosen(p)
print(f"=== пара {p} · глав {len(us)} ===")
for arm in FREE + tuple(ARMS):
vals = []
for u in us:
t = free_text(p, arm, u["uid"]) if arm in FREE else cell(tag(p, arm, u["uid"]))
if t.strip():
vals.append(len(C.canon_gaps(u["source"], t)))
if vals:
print(f" {arm:4s} n={len(vals):2d} потерь/глава {st.mean(vals):5.2f}")
# ── СМЕТА И ПРОВЕРКИ ─────────────────────────────────────────────────────────────────────────
def price_of(p: str, model: str) -> float:
"""Ожидаемая цена клетки — из ЗАМЕРЕННЫХ токенов уже купленных ОДНОПРОХОДОК ЭТОЙ пары,
пере-оценённых ТЕКУЩИМ пином. Зашитых цен в файле нет: после пере-пина они врали вчетверо."""
toks = []
for f in OUT.glob(f"{C.tag(PAIRS[p]['onepass'], '')}*.json"):
if any(x in f.name for x in ("ERROR", "LENGTH", "ЭХО", "STALE", "TWIN")):
continue
r = json.loads(f.read_text(encoding="utf-8"))
if r.get("finish") == "stop" and r.get("prompt_tokens"):
toks.append((r["prompt_tokens"], r.get("cached_tokens") or 0,
r.get("completion_tokens") or 0, r.get("reasoning_tokens") or 0,
r.get("total_tokens") or 0))
if not toks:
return 0.03
med = [st.median([x[i] for x in toks]) for i in range(5)]
# ⚠ Новый промт ДЛИННЕЕ склейки на входе — берём фактическую длину системного промта, а не
# длину прежнего: смета, посчитанная по старому входу, занижала бы резерв гарда.
grow = _sys_len(p) / max(_sys_len_old(p), 1)
return ROSTER.cost(model, int(med[0] * grow), int(med[1]), int(med[2]), int(med[3]),
int(med[4]))
def _sys_len(p: str) -> int:
u = chosen(p)[0]
return sum(len(m["content"]) for m in rol_msgs(p, u["source"]))
def _sys_len_old(p: str) -> int:
PAIRS[p]["wire"]()
u = chosen(p)[0]
return sum(len(m["content"]) for m in C.a2_msgs(u["source"]))
def cmd_show(p: str) -> None:
"""Промт целиком, глазами. Обязательная вычитка ПЕРЕД покупкой."""
u = chosen(p)[0]
for m in rol_msgs(p, u["source"]):
body = m["content"]
if m["role"] == "user":
body = body.split("Переведи следующий фрагмент:")[0] + \
"Переведи следующий фрагмент:\n\n<текст главы, %d знаков>" % len(u["source"])
print(f"\n────────── {m['role'].upper()} ──────────\n{body}")
def cmd_dry() -> None:
print("ОДНОПРОХОДКА КАК РОЛЬ · главы те же, что у захода Д17\n")
ZK.price_gate()
tot = 0.0
for p, cfg in PAIRS.items():
cfg["wire"]()
us = chosen(p)
print(f"\n=== пара {p} · глав {len(us)} · касса {cfg['phase']} ===")
print(f" системный промт: новый {_sys_len(p)} знаков против склейки {_sys_len_old(p)} "
f"({_sys_len(p) / max(_sys_len_old(p), 1):.2f}×)")
sub = 0.0
for arm, (model, _g) in ARMS.items():
need = sum(1 for u in us if not cell(tag(p, arm, u["uid"])).strip())
pr = price_of(p, model)
sub += need * pr
print(f" {arm:4s} {model:18s} надо {need:2d} × ${pr:.5f} = ${need * pr:.4f}")
sp = MONEY.Guarded(cfg["phase"]).spent()
ceil = MONEY.CEILINGS.get(cfg["phase"], 0.0)
mark = "ВЛЕЗАЕТ" if sub <= ceil - sp else f"НЕ ВЛЕЗАЕТ (не хватает ${sub - (ceil - sp):.4f})"
print(f" ИТОГО ${sub:.4f} · потолок ${ceil:.2f} · потрачено ${sp:.4f}{mark}")
tot += sub
print(f"\nВЕСЬ ЗАМЕР: ${tot:.4f} по текущему (пиковому) пину. В офф-пик DeepSeek вдвое дешевле.")
def cmd_selftest() -> int:
bad = 0
def ck(n: str, ok: bool, d: str = "") -> None:
nonlocal bad
bad += not ok
print(f"[{'OK ' if ok else 'ПРОВАЛ'}] {n}" + (f" {d}" if d else ""))
for p, cfg in PAIRS.items():
cfg["wire"]()
us = chosen(p)
ck(f"{p}: глав ровно 16", len(us) == 16, f"их {len(us)}")
ck(f"{p}: все три бесплатных арма есть на каждой главе",
all(free_text(p, a, u["uid"]).strip() for a in FREE for u in us))
u = us[0]
m = rol_msgs(p, u["source"])
sysall = "\n".join(x["content"] for x in m if x["role"] == "system")
# ⚠ ПАР-БЛОКИ ДОЕХАЛИ. Пустой блок сделал бы промт беднее боевого, и замер сравнивал бы
# не «роль против склейки», а «роль минус пар-специфика против склейки».
ub, rb = pair_blocks(p)
ck(f"{p}: блок «{UNITS_HEAD}» пары подтянут", len(ub) > 200, f"{len(ub)} знаков")
ck(f"{p}: блок «{REFLOW_HEAD}» пары подтянут", len(rb) > 400, f"{len(rb)} знаков")
ck(f"{p}: оба блока стоят В ПРОМТЕ", ub[:60] in sysall and rb[:60] in sysall)
ck(f"{p}: плейсхолдеры раскрыты все до одного", "{{" not in sysall)
# ⚠ ГЛАВНОЕ ОТЛИЧИЕ ОТ СКЛЕЙКИ — мета-фразы про конвейер быть не должно
ck(f"{p}: мета-фразы про отсутствующего редактора НЕТ",
"НЕ БУДЕТ" not in sysall and "после тебя" not in sysall)
ck(f"{p}: вольность РАЗРЕШЕНА явно", "ТЫ ВПРАВЕ" in sysall)
ck(f"{p}: регистр назван отдельной рамкой", "РЕГИСТР" in sysall)
ck(f"{p}: критерии проверки напечатаны в промте",
"ПО КАКИМ КРИТЕРИЯМ" in sysall and "ошибкой НЕ СЧИТАЕТСЯ" in sysall)
ck(f"{p}: черновик в промт НЕ попадает (однопроходка)",
"ЧЕРНОВ" not in sysall.upper() and "черновик" not in sysall.lower())
ck(f"{p}: банк-закон на месте либо термов в главе нет",
bool(PR.law_block(PR.terms_in(u["source"], None if cfg["para"]().key == "zh-ru"
else cfg["para"]().bank_file),
None if cfg["para"]().key == "zh-ru"
else cfg["para"]().bank_file)) == (len(m) == 3),
f"сообщений {len(m)}")
ck(f"{p}: пользовательское сообщение несёт ТЕКСТ ГЛАВЫ",
u["source"][:80] in m[-1]["content"])
ck(f"{p}: ровно одно пользовательское сообщение",
sum(1 for x in m if x["role"] == "user") == 1)
# промт обязан ОТЛИЧАТЬСЯ от склейки, иначе замер пуст
ck(f"{p}: новый промт не равен склейке",
sysall.strip() != "\n".join(x["content"] for x in C.a2_msgs(u["source"])
if x["role"] == "system").strip())
# теги и касса
tgs = {tag(p, a, u["uid"]) for a in ARMS}
ck(f"{p}: теги армов различны", len(tgs) == len(ARMS))
ck(f"{p}: теги покрыты глобом кассы {cfg['phase']}",
all(t.startswith(f"dv-{cfg['letter']}-") for t in tgs))
ck(f"{p}: фаза заведена в кассе", cfg["phase"] in MONEY.CEILINGS)
ck(f"{p}: две генерации несущей модели — РАЗНЫЕ теги",
tag(p, "RN", u["uid"]) != tag(p, "RN2", u["uid"]))
ck("пары не делят буквы тегов", len({v["letter"] for v in PAIRS.values()}) == len(PAIRS))
ck("несущий контраст объявлен один", len(PRIMARY) == 2 and PRIMARY == ("RN", "ZP"))
for a, (mo, _g) in ARMS.items():
try:
ck(f"цена считается для {a} ({mo})",
MONEY.M22.BUY.cost(mo, 3000, 0, 3600, 800, 7400) > 0)
except Exception as e: # noqa: BLE001
ck(f"цена считается для {a} ({mo})", False, f"{type(e).__name__}: {e}")
ck("grok идёт с размышлением (D30.1: reasoning-off = no-op-редактор)",
"reasoning_effort" not in json.dumps(call_kwargs("grok-4.3",
[{"role": "user", "content": "x"}])))
print(f"\n{'ЗАМЕР ГОДЕН К ПОКУПКЕ' if not bad else f'ПРОВАЛОВ: {bad}'}")
return bad
if __name__ == "__main__":
a = sys.argv[1:] or ["--dry"]
if a[0] == "--selftest":
sys.exit(1 if cmd_selftest() else 0)
elif a[0] == "--dry":
cmd_dry()
elif a[0] == "--show":
cmd_show(next((x for x in a[1:] if x in PAIRS), "zh"))
elif a[0] == "--canon":
canon()
elif a[0] == "--buy":
pr = [x for x in a[1:] if x in PAIRS] or list(PAIRS)
ar = [x for x in a[1:] if x in ARMS] or list(ARMS)
for x in pr:
buy(x, ar)
else:
raise SystemExit(f"⛔ неизвестный режим {a[0]!r}; знаю --show --dry --selftest --buy --canon")