253 lines
12 KiB
Python
253 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""ЧЕТЫРЕ ФАЛЬСИФИКАЦИИ выводов пробы 19 (санкция владельца 05.08).
|
||
|
||
Каждый арм построен как попытка УБИТЬ мой собственный заголовок, а не подтвердить его.
|
||
|
||
L — «дифф дороже» это артефакт несъёмного thinking? → тот же v4-pro, но reasoning_effort:"low"
|
||
(ручка признана рабочей замером 05.08, quirks §3б). + ЭХО-КОНТРОЛЬ: `low` пере-вооружает
|
||
эхо-мину (quirks §4), на pro не мерено ни разу.
|
||
G — «комплаенс 0.95-0.98» снят на окнах ВПЯТЕРО короче боевых? → единица ~3200 ру-ток. выхода
|
||
(боевой потолок EditCeilingOut, chunker.go), где уникальность якоря обязана деградировать.
|
||
E — «дифф дороже» это артефакт РУССКОЙ фертильности (r=1.65-1.90 zh→ru, exp01; для zh→en
|
||
замера в проекте НЕТ)? → тот же дизайн на английской цели.
|
||
N — «do-nothing промптом не покупается» мерено на промпте со СТИЛЬ-мандатом, а research/19 §C1
|
||
ставит NO_CHANGE на FIDELITY-ONLY проход → узкий промпт без стиля.
|
||
|
||
Сырьё durable, префиксы fp-L/G/E/N. Материал L и N — те же 6 окон и черновики пробы 19.
|
||
"""
|
||
from __future__ import annotations
|
||
import argparse
|
||
import json
|
||
import re
|
||
import sys
|
||
import time
|
||
import unicodedata
|
||
from pathlib import Path
|
||
|
||
REPO = Path("/home/ubuntu/projects/textmachine")
|
||
HERE = Path(__file__).resolve().parent
|
||
RAW = Path.home() / "books" / "gu-zhenren" / "bank-arbitration"
|
||
SRC = Path.home() / "books" / "gu-zhenren" / "coldrun-a" / "guzhenren-ch1-10.gb18030.txt"
|
||
sys.path.insert(0, str(HERE))
|
||
sys.path.insert(0, str(REPO / "eval" / "bank_arbitration"))
|
||
|
||
import editor_wire_probe as P # noqa: E402
|
||
import probe as Q # noqa: E402
|
||
from apply import apply_ops # noqa: E402
|
||
from inject_probe import pick_windows, TERMS, render_translator, block_for # noqa: E402
|
||
|
||
CEILING_USD = 0.50
|
||
HARD_STOP = 0.44
|
||
REPS = 3
|
||
FERT_CJK, FERT_OTHER = 1.1978, 0.3852 # chunker.go SegBudget
|
||
EDIT_CEILING_OUT = 3200 # боевой потолок единицы редактуры
|
||
|
||
# Английские канон-формы для блока. НЕ голд: en-голда в проекте нет. Взяты из research/24 §F1
|
||
# (там свежая en-редакция дала «Aperture · Gu Master · primeval stone»), остальные — фан-канон RI.
|
||
# ⇒ арм E скорится ТОЛЬКО по формату/цене/объёму, НЕ по точности перевода.
|
||
EN_FORMS = {"空窍": "Aperture", "真元": "primeval essence", "元石": "primeval stone",
|
||
"元海": "sea of primeval essence", "蛊师": "Gu Master", "族长": "clan leader",
|
||
"月光蛊": "Moonlight Gu", "春秋蝉": "Spring Autumn Cicada"}
|
||
EN_HEADER = ("CANONICAL RENDERINGS of names and terms (in the draft, the source term on the left MUST "
|
||
"be rendered exactly by the form on the right — bring any divergence to it, inflecting by "
|
||
"context; do not introduce other variants and do not change anything else):")
|
||
|
||
EN_VARS = dict(P.RENDER_VARS, target_lang="en", audience="adult webnovel readers",
|
||
genre="webnovel", transcription="pinyin")
|
||
|
||
|
||
def est_out(t: str) -> float:
|
||
cjk = other = 0
|
||
for ch in t:
|
||
if ch.isspace():
|
||
continue
|
||
if unicodedata.category(ch) == "Lo" and ord(ch) > 0x2E80:
|
||
cjk += 1
|
||
else:
|
||
other += 1
|
||
return FERT_CJK * cjk + FERT_OTHER * other
|
||
|
||
|
||
def render_from(path: Path, text: str, draft: str, vars_=None) -> tuple[str, str]:
|
||
canon = P.strip_comments(path.read_text(encoding="utf-8"))
|
||
sys_part, user = canon.split(P.USER_SEP, 1)
|
||
old = P.RENDER_VARS
|
||
if vars_:
|
||
P.RENDER_VARS = vars_
|
||
try:
|
||
return P.render(sys_part.strip(), text, draft), P.render(user.strip(), text, draft)
|
||
finally:
|
||
P.RENDER_VARS = old
|
||
|
||
|
||
def msgs_from(path: Path, text: str, draft: str, block: str | None, vars_=None):
|
||
s, u = render_from(path, text, draft, vars_)
|
||
m = [{"role": "system", "content": s}]
|
||
if block:
|
||
m.append({"role": "system", "content": block})
|
||
m.append({"role": "user", "content": u})
|
||
return m
|
||
|
||
|
||
def en_block(terms: list[str]) -> str:
|
||
return EN_HEADER + "\n" + "\n".join(f"- {t} → «{EN_FORMS[t]}»" for t in terms if t in EN_FORMS)
|
||
|
||
|
||
def long_units(n=2):
|
||
"""Смежные абзацы источника до боевого потолка единицы (~3200 ру-токенов выхода)."""
|
||
paras = [p for p in SRC.read_text(encoding="gb18030").split("\n") if p.strip()]
|
||
out, i = [], 60
|
||
while len(out) < n and i < len(paras) - 40:
|
||
buf, j = "", i
|
||
while j < len(paras) and est_out(buf) < EDIT_CEILING_OUT:
|
||
buf += paras[j] + "\n"
|
||
j += 1
|
||
terms = [t for t in TERMS if t in buf]
|
||
if len(terms) >= 3:
|
||
out.append((i, buf, terms))
|
||
i = j + 5
|
||
else:
|
||
i += 5
|
||
return out
|
||
|
||
|
||
# --- планы армов -------------------------------------------------------------------------------
|
||
def plan_L():
|
||
"""L: тот же материал пробы 19, но reasoning_effort:'low' на обоих контрактах."""
|
||
out = []
|
||
for rep in range(1, REPS + 1):
|
||
for k, (_, buf, terms) in enumerate(pick_windows()):
|
||
d = P.draft_of(k)
|
||
blk = P.editor_block(terms, P.HEADER_LAW_CLAUSE, None)
|
||
out.append((f"fp-LD-w{k}-r{rep}", Q.messages("diff", buf, d, blk), "diff", d, {"reasoning_effort": "low"}))
|
||
out.append((f"fp-LF-w{k}-r{rep}", Q.messages("full", buf, d, blk), "full", d, {"reasoning_effort": "low"}))
|
||
return out
|
||
|
||
|
||
def plan_N():
|
||
"""N: fidelity-only дифф-промпт — там, где research/19 §C1 ставит NO_CHANGE."""
|
||
pr = HERE / "prompts" / "editor-diff-fidelity.md"
|
||
out = []
|
||
for rep in range(1, REPS + 1):
|
||
for k, (_, buf, terms) in enumerate(pick_windows()):
|
||
d = P.draft_of(k)
|
||
blk = P.editor_block(terms, P.HEADER_LAW_CLAUSE, None)
|
||
out.append((f"fp-N-w{k}-r{rep}", msgs_from(pr, buf, d, blk), "diff", d, None))
|
||
return out
|
||
|
||
|
||
def plan_G(drafts: dict):
|
||
"""G: боевая единица ~3200 ру-токенов выхода."""
|
||
out = []
|
||
for rep in range(1, REPS + 1):
|
||
for u, (idx, buf, terms) in enumerate(long_units()):
|
||
d = drafts.get(f"G{u}")
|
||
if not d:
|
||
continue
|
||
blk = P.editor_block(terms, P.HEADER_LAW_CLAUSE, None)
|
||
out.append((f"fp-GD-u{u}-r{rep}", Q.messages("diff", buf, d, blk), "diff", d, None))
|
||
out.append((f"fp-GF-u{u}-r{rep}", Q.messages("full", buf, d, blk), "full", d, None))
|
||
return out
|
||
|
||
|
||
def plan_E(drafts: dict):
|
||
"""E: английская цель, тот же дизайн."""
|
||
pd = HERE / "prompts" / "editor-diff-en.md"
|
||
pf = HERE / "prompts" / "editor-en.md"
|
||
out = []
|
||
for rep in (1, 2):
|
||
for k, (_, buf, terms) in enumerate(pick_windows()):
|
||
d = drafts.get(f"E{k}")
|
||
if not d:
|
||
continue
|
||
blk = en_block(terms)
|
||
out.append((f"fp-ED-w{k}-r{rep}", msgs_from(pd, buf, d, blk, EN_VARS), "diff", d, None))
|
||
out.append((f"fp-EF-w{k}-r{rep}", msgs_from(pf, buf, d, blk, EN_VARS), "full", d, None))
|
||
return out
|
||
|
||
|
||
def make_drafts():
|
||
"""Черновики, которых нет в сырье: 2 длинные единицы (ru) и 6 английских окон."""
|
||
cl = P.client()
|
||
drafts, total = {}, 0.0
|
||
for u, (idx, buf, terms) in enumerate(long_units()):
|
||
tag = f"fp-draft-G{u}"
|
||
f = RAW / f"{tag}.json"
|
||
if not f.exists():
|
||
system, user = render_translator(buf)
|
||
m = [{"role": "system", "content": system},
|
||
{"role": "system", "content": block_for(terms, None, False)},
|
||
{"role": "user", "content": user}]
|
||
total += P.call(cl, m, tag, model="deepseek-v4-flash", temperature=0, max_tokens=16000,
|
||
extra={"reasoning_effort": "low"}, prices=(0.14, 0.0028, 0.28))["cost_usd"]
|
||
drafts[f"G{u}"] = json.load(open(RAW / f"{tag}.json", encoding="utf-8"))["content"]
|
||
# ⚠ НЕ боевой zh-ru/translator.md: он параметризован лишь в строке 1, а в 9/10/14 русский зашит
|
||
# прозой — прогон поверх него дал РУССКИЕ черновики fp-draft-E*. Зеркало для en — в моей зоне.
|
||
tpl = HERE / "prompts" / "translator-en.md"
|
||
for k, (_, buf, terms) in enumerate(pick_windows()):
|
||
tag = f"fp-draft-E{k}"
|
||
f = RAW / f"{tag}.json"
|
||
if not f.exists():
|
||
s, u = render_from(tpl, buf, "", EN_VARS)
|
||
blk = ("GLOSSARY (use these approved renderings of names and terms consistently):\n"
|
||
+ "\n".join(f"{t} → {EN_FORMS[t]}" for t in terms if t in EN_FORMS))
|
||
m = [{"role": "system", "content": s}, {"role": "system", "content": blk},
|
||
{"role": "user", "content": u}]
|
||
total += P.call(cl, m, tag, model="deepseek-v4-flash", temperature=0, max_tokens=16000,
|
||
extra={"reasoning_effort": "low"}, prices=(0.14, 0.0028, 0.28))["cost_usd"]
|
||
drafts[f"E{k}"] = json.load(open(RAW / f"{tag}.json", encoding="utf-8"))["content"]
|
||
print(f"черновики готовы, доплата ${total:.6f}")
|
||
return drafts
|
||
|
||
|
||
def spent() -> float:
|
||
return sum(json.load(open(f, encoding="utf-8"))["cost_usd"]
|
||
for f in RAW.glob("fp-*.json") if not f.name.endswith(".applied.json"))
|
||
|
||
|
||
def cmd_run(only):
|
||
drafts = make_drafts()
|
||
plan = []
|
||
if not only or "L" in only:
|
||
plan += plan_L()
|
||
if not only or "N" in only:
|
||
plan += plan_N()
|
||
if not only or "G" in only:
|
||
plan += plan_G(drafts)
|
||
if not only or "E" in only:
|
||
plan += plan_E(drafts)
|
||
cl = P.client()
|
||
total = spent()
|
||
print(f"план: {len(plan)} вызовов; уже потрачено ${total:.6f}")
|
||
worst = 16000 / 1e6 * P.PRICE_OUT + 6000 / 1e6 * P.PRICE_IN
|
||
for tag, msgs, kind, draft, extra in plan:
|
||
if (RAW / f"{tag}.json").exists():
|
||
continue
|
||
if total + worst > CEILING_USD or total >= HARD_STOP:
|
||
print(f"СТОП (проекционный гард): ${total:.4f} + худший ${worst:.4f} > ${CEILING_USD}")
|
||
return
|
||
rec = P.call(cl, msgs, tag, extra=extra)
|
||
total += rec["cost_usd"]
|
||
if kind == "diff":
|
||
r = apply_ops(draft, rec["content"])
|
||
(RAW / f"{tag}.applied.json").write_text(json.dumps(dict(
|
||
tag=tag, ops_total=r.ops_total, applied=r.applied, no_change=r.no_change,
|
||
malformed=r.malformed, rejected_notfound=r.rejected_notfound,
|
||
rejected_ambiguous=r.rejected_ambiguous, rejected_overlap=r.rejected_overlap,
|
||
rejected_empty=r.rejected_empty, details=r.details, text=r.text),
|
||
ensure_ascii=False, indent=1), encoding="utf-8")
|
||
print(f" apply: ops={r.ops_total} applied={r.applied} no_change={r.no_change} malformed={r.malformed}")
|
||
time.sleep(0.3)
|
||
print(f"TOTAL финальных проб: ${total:.6f}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--run", nargs="*", default=None, help="L G E N (пусто = все)")
|
||
a = ap.parse_args()
|
||
if a.run is not None:
|
||
cmd_run(a.run)
|
||
else:
|
||
for u, (i, buf, terms) in enumerate(long_units()):
|
||
print(f"длинная единица {u}: @para{i} est_out={est_out(buf):.0f} ру-ток, термов {len(terms)}")
|
||
print("армы: L(36) N(18) G(12+2черн) E(24+6черн) = 98 вызовов")
|