153 lines
7.5 KiB
Python
153 lines
7.5 KiB
Python
#!/usr/bin/env python3
|
||
"""Арм Q4b (`docs/experiments/15-segmentation-empirics.md:108`, спека `research/19` §C1–C2):
|
||
ДИФФ-редактор (anchored search/replace + детерминированный толерантный apply) против действующего
|
||
FULL-REGEN редактора. Арм пре-регистрирован в exp15 и не гонялся НИ РАЗУ; носитель — строка 106
|
||
бэклога («Слой 3 НЕ достроен: контракт редактора = full-regen, а целевой — узкие мандаты + ДИФФЫ»).
|
||
|
||
Материал переиспользован из пробы 18 ($0): те же 6 окон, те же черновики арма B транслятора, те же
|
||
12 посаженных дефектов. Модель — боевой редактор deepseek-v4-pro, провод как edit-стадия c1.
|
||
Reflow в дифф-арм НЕ входит: по §C1/§C3 он ОТДЕЛЬНЫЙ проход (арм Q4c) ⇒ проза/переверстка этой
|
||
пробой НЕ сравниваются, и это граница вывода, а не упущение.
|
||
|
||
Сырьё durable ~/books/gu-zhenren/bank-arbitration/, префикс ec-*.
|
||
"""
|
||
from __future__ import annotations
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
from dotenv import load_dotenv
|
||
from openai import OpenAI
|
||
|
||
REPO = Path("/home/ubuntu/projects/textmachine")
|
||
HERE = Path(__file__).resolve().parent
|
||
RAW = Path.home() / "books" / "gu-zhenren" / "bank-arbitration"
|
||
load_dotenv(REPO / "eval" / ".env")
|
||
sys.path.insert(0, str(HERE))
|
||
sys.path.insert(0, str(REPO / "eval" / "bank_arbitration"))
|
||
|
||
import editor_wire_probe as P # noqa: E402 провод/окна/черновики/посадки пробы 18
|
||
from apply import apply_ops # noqa: E402
|
||
from inject_probe import pick_windows, TERMS # noqa: E402
|
||
|
||
DIFF_PROMPT = HERE / "prompts" / "editor-diff.md"
|
||
CEILING_USD = 0.40
|
||
HARD_STOP_USD = 0.35
|
||
REPLICATES = 3
|
||
|
||
|
||
def render_diff(text: str, draft: str) -> tuple[str, str]:
|
||
canon = P.strip_comments(DIFF_PROMPT.read_text(encoding="utf-8"))
|
||
sys_part, user = canon.split(P.USER_SEP, 1)
|
||
return P.render(sys_part.strip(), text, draft), P.render(user.strip(), text, draft)
|
||
|
||
|
||
def messages(kind: str, text: str, draft: str, block: str | None) -> list[dict]:
|
||
if kind == "full":
|
||
return P.editor_messages(text, draft, block)
|
||
system, user = render_diff(text, draft)
|
||
msgs = [{"role": "system", "content": system}]
|
||
if block and block.strip():
|
||
msgs.append({"role": "system", "content": block})
|
||
msgs.append({"role": "user", "content": user})
|
||
return msgs
|
||
|
||
|
||
def plan_arms():
|
||
"""[(tag, kind, window, messages, draft)] — армы в порядке возрастания риска."""
|
||
wins = pick_windows()
|
||
out = []
|
||
for rep in range(1, REPLICATES + 1):
|
||
for k, (_, buf, terms) in enumerate(wins):
|
||
draft = P.draft_of(k)
|
||
blk = P.editor_block(terms, P.HEADER_LAW_CLAUSE, None)
|
||
out.append((f"ec-D1-w{k}-r{rep}", "diff", k, messages("diff", buf, draft, blk), draft))
|
||
for k, (_, buf, terms) in enumerate(wins):
|
||
draft, _ = P.planted_draft(k, P.draft_of(k))
|
||
out.append((f"ec-D2-w{k}-r{rep}", "diff", k, messages("diff", buf, draft, None), draft))
|
||
# добор реплик действующего контракта: у пробы 18 по одному розыгрышу (A1 и A5),
|
||
# а собственная находка той пробы — одиночный розыгрыш нельзя цитировать как величину.
|
||
for rep in (2, 3):
|
||
for k, (_, buf, terms) in enumerate(wins):
|
||
draft = P.draft_of(k)
|
||
blk = P.editor_block(terms, P.HEADER_LAW_CLAUSE, None)
|
||
out.append((f"ec-F1-w{k}-r{rep}", "full", k, messages("full", buf, draft, blk), draft))
|
||
for k, (_, buf, terms) in enumerate(wins):
|
||
draft, _ = P.planted_draft(k, P.draft_of(k))
|
||
out.append((f"ec-F5-w{k}-r{rep}", "full", k, messages("full", buf, draft, None), draft))
|
||
return out
|
||
|
||
|
||
def spent() -> float:
|
||
return sum(json.load(open(f, encoding="utf-8"))["cost_usd"] for f in RAW.glob("ec-*.json"))
|
||
|
||
|
||
def cmd_plan():
|
||
import tiktoken
|
||
enc = tiktoken.get_encoding("cl100k_base")
|
||
calib = 0.612 # калибровка пробы 18 (cl100k → фактические prompt_tokens DeepSeek)
|
||
plan = plan_arms()
|
||
tin = sum(int(len(enc.encode("\n".join(m["content"] for m in msgs))) * calib)
|
||
for _, _, _, msgs, _ in plan)
|
||
n_diff = sum(1 for t, k, *_ in plan if k == "diff")
|
||
n_full = len(plan) - n_diff
|
||
print(f"вызовов: {len(plan)} ({n_diff} дифф + {n_full} full-regen), вход ≈{tin} ток. "
|
||
f"→ ${tin/1e6*P.PRICE_IN:.4f}")
|
||
# выход: дифф короткий (правки), full-regen длинный (весь текст) — берём наблюдённое в пробе 18
|
||
for lbl, d_out, f_out in (("ожидаемо", 1200, 2600), ("пессимистично", 4000, 8000)):
|
||
o = n_diff * d_out + n_full * f_out
|
||
print(f" {lbl:14s} выход {o} ток. → ИТОГО ${tin/1e6*P.PRICE_IN + o/1e6*P.PRICE_OUT:.4f}")
|
||
print(f" абсолютный потолок (все в cap 16000): "
|
||
f"${tin/1e6*P.PRICE_IN + len(plan)*16000/1e6*P.PRICE_OUT:.4f}")
|
||
print(f"ПОТОЛОК ПРОБЫ ${CEILING_USD}; живой стоп ${HARD_STOP_USD} + проекционный гард на вызов")
|
||
|
||
|
||
def cmd_run(only: list[str]):
|
||
plan = [p for p in plan_arms() if not only or any(p[0].startswith(f"ec-{o}") for o in only)]
|
||
cl = P.client()
|
||
total = spent()
|
||
if total:
|
||
print(f"уже потрачено этой пробой: ${total:.6f}")
|
||
worst_call = 16000 / 1e6 * P.PRICE_OUT + 2600 / 1e6 * P.PRICE_IN
|
||
for tag, kind, k, msgs, draft in plan:
|
||
if (RAW / f"{tag}.json").exists():
|
||
continue
|
||
if total + worst_call > CEILING_USD or total >= HARD_STOP_USD:
|
||
print(f"СТОП (проекционный гард): ${total:.4f} + худший вызов ${worst_call:.4f} "
|
||
f"> потолка ${CEILING_USD}")
|
||
return
|
||
rec = P.call(cl, msgs, tag)
|
||
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} rejected={r.rejected} "
|
||
f"no_change={r.no_change} malformed={r.malformed}")
|
||
time.sleep(0.4)
|
||
print(f"TOTAL пробы Q4b: ${total:.6f}")
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--plan", action="store_true")
|
||
ap.add_argument("--run", nargs="*", default=None, help="D1 D2 F1 F5 (пусто = все)")
|
||
a = ap.parse_args()
|
||
if a.plan:
|
||
cmd_plan()
|
||
elif a.run is not None:
|
||
cmd_run(a.run)
|
||
else:
|
||
ap.print_help()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|