#!/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()