200 lines
9.2 KiB
Python
200 lines
9.2 KiB
Python
#!/usr/bin/env python3
|
||
"""Проба C (инжект, пре-рег research/24 §0 п.5): послушание и ВРЕД инъекции глоссария
|
||
на транслятор-проводе. 6 чанков × 4 арма: A без инъекции · B верный закон · C НЕВЕРНАЯ строка
|
||
законом · D та же неверная строка с ⟨проверить⟩. deepseek-v4-flash, effort low, temp 0.
|
||
|
||
Формат байт-близко к движку: messages = [system(translator.md), system(глоссарий-блок), user(text)]
|
||
(MessagesWithInjection, render.go:246-262); блок = glossary_header + "src → dst" строки,
|
||
маркер = " ⟨проверить⟩" (internal/lang/data/injection.txt). Сырьё durable в ~/books/.../bank-arbitration/.
|
||
"""
|
||
from __future__ import annotations
|
||
import argparse
|
||
import json
|
||
import re
|
||
import time
|
||
import unicodedata
|
||
from pathlib import Path
|
||
|
||
from dotenv import load_dotenv
|
||
from openai import OpenAI
|
||
import os
|
||
|
||
REPO = Path("/home/ubuntu/projects/textmachine")
|
||
RAW = Path.home() / "books" / "gu-zhenren" / "bank-arbitration"
|
||
SRC = Path.home() / "books" / "gu-zhenren" / "coldrun-a" / "guzhenren-ch1-10.gb18030.txt"
|
||
load_dotenv(REPO / "eval" / ".env")
|
||
|
||
GLOSSARY_HEADER = ("ГЛОССАРИЙ (используй эти утверждённые переводы имён и терминов последовательно; "
|
||
"строки с пометкой ⟨проверить⟩ — неподтверждённые кандидаты):")
|
||
MARKER = " ⟨проверить⟩"
|
||
PRICE_IN, PRICE_CACHED, PRICE_OUT = 0.14, 0.0028, 0.28
|
||
|
||
# терм: (gold_dst, wrong_dst, gold_rx, wrong_rx) — regex по нормализованному выходу
|
||
TERMS = {
|
||
"空窍": ("апертура", "пустота", r"апертур", r"пустот"),
|
||
"真元": ("истинная ци", "чжэньюань", r"истинн\w* ци\b", r"чж[эе]н[ьи]?юан"),
|
||
"元石": ("первобытный камень", "юаньши", r"первобытн\w* кам", r"юаньши"),
|
||
"元海": ("море истинной ци", "Юаньхай", r"мор\w{0,2} истинн\w* ци", r"юаньхай"),
|
||
"蛊师": ("гу-мастер", "гуши", r"гу-мастер", r"\bгуши\b"),
|
||
"族长": ("глава клана", "вождь клана", r"глав\w{0,2} клана", r"вожд\w* клана"),
|
||
"月光蛊": ("Гу лунного света", "лунный гу", r"гу лунного света", r"лунн\w{0,2} гу\b"),
|
||
"春秋蝉": ("Весенне-осенняя цикада", "Цикада Весны и Осени", r"весенне-осенн", r"цикад\w{0,2} весны и осени"),
|
||
}
|
||
WRONG_PRIORITY = ["空窍", "真元", "元石", "元海", "蛊师", "月光蛊", "春秋蝉", "族长"]
|
||
|
||
TRANSLATOR = REPO / "backend/prompts/zh-ru/translator.md"
|
||
USER_SEP = "\n---USER---\n"
|
||
|
||
|
||
def render_translator(text: str) -> tuple[str, str]:
|
||
raw = TRANSLATOR.read_text(encoding="utf-8")
|
||
head, user = raw.split(USER_SEP, 1)
|
||
vals = {"source_lang": "zh", "target_lang": "ru", "genre": "вебновелла",
|
||
"audience": "взрослые читатели вебновелл", "title": "蛊真人", "venuti": "0.60",
|
||
"honorifics": "keep", "transcription": "palladius", "footnotes": "minimal", "text": text}
|
||
for k, v in vals.items():
|
||
head = head.replace("{{" + k + "}}", v)
|
||
user = user.replace("{{" + k + "}}", v)
|
||
return head.strip(), user.strip()
|
||
|
||
|
||
def pick_windows(n=6, lo=500, hi=1200):
|
||
text = SRC.read_text(encoding="gb18030")
|
||
paras = [p for p in text.split("\n") if p.strip()]
|
||
wins = []
|
||
i = 0
|
||
while i < len(paras):
|
||
buf, j = "", i
|
||
while j < len(paras) and len(buf) < lo:
|
||
buf += paras[j] + "\n"
|
||
j += 1
|
||
if len(buf) > hi:
|
||
buf = buf[:hi]
|
||
terms = [t for t in TERMS if t in buf]
|
||
if len(terms) >= 2:
|
||
wins.append((i, buf, terms))
|
||
i = j + 3 # разнести окна
|
||
else:
|
||
i += 1
|
||
# ранжир: больше термов, разнесённость по книге
|
||
wins.sort(key=lambda w: (-len(w[2]), w[0]))
|
||
chosen, used = [], set()
|
||
for w in wins:
|
||
if any(abs(w[0] - u) < 30 for u in used):
|
||
continue
|
||
chosen.append(w)
|
||
used.add(w[0])
|
||
if len(chosen) == n:
|
||
break
|
||
return chosen
|
||
|
||
|
||
def block_for(terms: list[str], wrong: str | None, marked: bool) -> str:
|
||
lines = []
|
||
for t in terms:
|
||
gold, bad, _, _ = TERMS[t]
|
||
if t == wrong:
|
||
lines.append(f"{t} → {bad}" + (MARKER if marked else ""))
|
||
else:
|
||
lines.append(f"{t} → {gold}")
|
||
return GLOSSARY_HEADER + "\n" + "\n".join(lines)
|
||
|
||
|
||
def norm(s: str) -> str:
|
||
return " ".join(unicodedata.normalize("NFKC", s).casefold().replace("ё", "е").split())
|
||
|
||
|
||
def call(cl, system, inj, user, tag):
|
||
msgs = [{"role": "system", "content": system}]
|
||
if inj:
|
||
msgs.append({"role": "system", "content": inj})
|
||
msgs.append({"role": "user", "content": user})
|
||
t0 = time.time()
|
||
r = cl.chat.completions.create(model="deepseek-v4-flash", messages=msgs, max_tokens=16000,
|
||
temperature=0, extra_body={"reasoning_effort": "low"})
|
||
ch = r.choices[0]
|
||
content = ch.message.content or ""
|
||
u = r.usage
|
||
pt = u.prompt_tokens or 0
|
||
ct = u.completion_tokens or 0
|
||
cached = getattr(getattr(u, "prompt_tokens_details", None), "cached_tokens", 0) or 0
|
||
cost = (pt - cached) / 1e6 * PRICE_IN + cached / 1e6 * PRICE_CACHED + ct / 1e6 * PRICE_OUT
|
||
rec = dict(tag=tag, model_returned=r.model, effort="low", finish=ch.finish_reason,
|
||
ts=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||
prompt_tokens=pt, cached_tokens=cached, completion_tokens=ct,
|
||
cost_usd=round(cost, 6), latency_s=round(time.time() - t0, 1),
|
||
injection=inj, user=user, content=content)
|
||
(RAW / f"{tag}.json").write_text(json.dumps(rec, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
print(f"[{tag}] finish={ch.finish_reason} out={ct} ${cost:.6f} {rec['latency_s']}s")
|
||
return rec
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--plan", action="store_true")
|
||
ap.add_argument("--score", action="store_true")
|
||
a = ap.parse_args()
|
||
wins = pick_windows()
|
||
if a.plan:
|
||
for k, (i, buf, terms) in enumerate(wins):
|
||
cand = [t for t in WRONG_PRIORITY if t in terms]; wrong = cand[k % len(cand)]
|
||
print(f"win{k} @para{i} len={len(buf)} terms={terms} wrong={wrong}")
|
||
print(" " + buf[:120].replace("\n", " / "))
|
||
return
|
||
if a.score:
|
||
score(wins)
|
||
return
|
||
cl = OpenAI(api_key=os.environ["DEEPSEEK_API_KEY"], base_url="https://api.deepseek.com/v1", timeout=600)
|
||
total = 0.0
|
||
for k, (i, buf, terms) in enumerate(wins):
|
||
cand = [t for t in WRONG_PRIORITY if t in terms]; wrong = cand[k % len(cand)]
|
||
system, user = render_translator(buf)
|
||
arms = {
|
||
"A": None,
|
||
"B": block_for(terms, wrong=None, marked=False),
|
||
"C": block_for(terms, wrong=wrong, marked=False),
|
||
"D": block_for(terms, wrong=wrong, marked=True),
|
||
}
|
||
for arm, inj in arms.items():
|
||
rec = call(cl, system, inj, user, f"inj-w{k}-{arm}")
|
||
total += rec["cost_usd"]
|
||
time.sleep(0.5)
|
||
print(f"TOTAL проба C: ${total:.6f}")
|
||
|
||
|
||
def score(wins):
|
||
rows = []
|
||
for k, (i, buf, terms) in enumerate(wins):
|
||
cand = [t for t in WRONG_PRIORITY if t in terms]; wrong = cand[k % len(cand)]
|
||
for arm in "ABCD":
|
||
f = RAW / f"inj-w{k}-{arm}.json"
|
||
if not f.exists():
|
||
continue
|
||
out = norm(json.load(open(f, encoding="utf-8"))["content"])
|
||
for t in terms:
|
||
gold, bad, grx, brx = TERMS[t]
|
||
rows.append(dict(win=k, arm=arm, term=t, is_wrong_term=(t == wrong),
|
||
gold_hit=bool(re.search(grx, out)), wrong_hit=bool(re.search(brx, out))))
|
||
import collections
|
||
def rate(sel):
|
||
sel = list(sel)
|
||
return f"{sum(1 for r in sel if r['gold_hit'])}/{len(sel)} gold, {sum(1 for r in sel if r['wrong_hit'])}/{len(sel)} wrong"
|
||
print("=== послушание (все термы, верные строки) ===")
|
||
for arm in "ABCD":
|
||
sel = [r for r in rows if r["arm"] == arm and not r["is_wrong_term"]]
|
||
print(f" arm {arm}: {rate(sel)}")
|
||
print("=== НЕВЕРНАЯ строка (6 термов) ===")
|
||
for arm in "ABCD":
|
||
sel = [r for r in rows if r["arm"] == arm and r["is_wrong_term"]]
|
||
print(f" arm {arm}: {rate(sel)} (в C/D инъецирован WRONG)")
|
||
print("=== единицы: неверный терм по армам ===")
|
||
byterm = collections.defaultdict(dict)
|
||
for r in rows:
|
||
if r["is_wrong_term"]:
|
||
byterm[(r["win"], r["term"])][r["arm"]] = (r["gold_hit"], r["wrong_hit"])
|
||
for (w, t), arms in sorted(byterm.items()):
|
||
print(f" w{w} {t}: " + " ".join(f"{a}:(gold={arms[a][0]},wrong={arms[a][1]})" for a in sorted(arms)))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|