#!/usr/bin/env python3 """Standalone terminologist wire-probe (backend-independent; direct DeepSeek API). Reconstructs the terminologist batch input BYTE-CLOSE to engine RenderBatch (terminology.go:629-660) from FROZEN coldrun-a artifacts. Tests fixes for the measured bank defects. $ tracked two ways.""" from __future__ import annotations import argparse, json, os, sys, time from pathlib import Path from dotenv import load_dotenv from openai import OpenAI REPO = Path("/home/ubuntu/projects/textmachine") SC = Path(__file__).resolve().parent # eval/bank_autonomy (for the parse_common import) RAW = Path.home() / "books" / "gu-zhenren" / "bank-probes" # durable paid-data sink (outside git) RAW.mkdir(exist_ok=True) load_dotenv(REPO / "eval" / ".env") # loads DEEPSEEK_API_KEY without us reading .env PRICE_IN, PRICE_CACHED, PRICE_OUT = 0.14, 0.0028, 0.28 # flash, models.yaml TERM_PROMPT = REPO / "backend/prompts/zh-ru/terminologist.md" USER_SEP = "\n---USER---\n" def render_terminologist(text_block: str) -> tuple[str,str]: raw = TERM_PROMPT.read_text(encoding="utf-8") # strip HTML comments out, rest = [], raw while True: i = rest.find(""); rest = rest[j+3:] canon = "".join(out) head, user = canon.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_block} system = head.strip(); user = user for k,v in vals.items(): system = system.replace("{{"+k+"}}", v); user = user.replace("{{"+k+"}}", v) return system, user.strip() def render_batch(cands: list[dict]) -> str: """Reproduce terminology.go RenderBatch layout from bank-stop fields.""" b = [] for c in cands: b.append(f"### {c['src']}") b.append(f"key: {c.get('key',c['src'])}\ntype: {c['type']}\norigin: {c['origin']}\nfreq: {c['freq']}\nsince_ch: 0") if c.get("related"): b.append(f"related: {', '.join(c['related'])}") if c.get("evidence"): b.append(f"evidence: {', '.join(c['evidence'])}") # engine emits this (RenderBatch:640) if c.get("drafts"): vs = " | ".join(f"{d} ×{n}" for d,n in c["drafts"]) b.append(f"drafts: {vs}") for k in c.get("ctx",[]): b.append(f"ctx: {k}") b.append("") return "\n".join(b).rstrip("\n") def call(client, system, user, cap, tag, temperature=0.3, prefill=None, effort=None): msgs = [{"role":"system","content":system},{"role":"user","content":user}] if prefill is not None: msgs.append({"role":"assistant","content":prefill,"prefix":True}) kw = dict(model="deepseek-v4-flash", messages=msgs, max_tokens=cap) if temperature is not None: kw["temperature"]=temperature if effort: kw["extra_body"]={"reasoning_effort":effort} t0=time.time(); r=client.chat.completions.create(**kw) ch=r.choices[0]; msg=ch.message content=msg.content or ""; reasoning=getattr(msg,"reasoning_content",None) or "" u=r.usage; pt=getattr(u,"prompt_tokens",0)or 0; ct=getattr(u,"completion_tokens",0)or 0 det=getattr(u,"completion_tokens_details",None); rt=(getattr(det,"reasoning_tokens",0) if det else 0)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,ts=time.strftime("%Y-%m-%dT%H:%M:%SZ",time.gmtime()),model_returned=r.model, cap=cap,finish=ch.finish_reason,content_chars=len(content),reasoning_chars=len(reasoning), prompt_tokens=pt,cached_tokens=cached,completion_tokens=ct,reasoning_tokens=rt, cost_usd=round(cost,6),latency_s=round(time.time()-t0,1), system=system,user=user,content=content,reasoning_content=reasoning,prefill=prefill) (RAW/f"{tag}.json").write_text(json.dumps(rec,ensure_ascii=False,indent=1),encoding="utf-8") print(f"[{tag}] model={r.model} finish={ch.finish_reason} cap={cap} out={ct} " f"(reason_tok={rt}) content={len(content)}c ${cost:.6f} {rec['latency_s']}s") print(f" CONTENT: {content!r}") return rec def client(): key=os.environ.get("DEEPSEEK_API_KEY") if not key: print("нет DEEPSEEK_API_KEY",file=sys.stderr); sys.exit(2) return OpenAI(api_key=key,base_url="https://api.deepseek.com/v1",timeout=600) # ---- build candidates from bank-stop ---- sys.path.insert(0, str(SC)) from parse_common import parse_bankstop BS = parse_bankstop() def cand(src, related=None): r=BS[src] return {"src":src,"key":src,"type":r["type"],"origin":r.get("origin",""), "freq":r["freq"],"drafts":r["drafts"],"ctx":r["ctx"],"related":related or [], "evidence":r.get("evidence",[])} # engine-faithful: carry evidence + full multi-line ctx DENG=["甲等","乙等","丙等","丁等"] ZHUAN=["一转","二转","三转","四转","五转","六转","九转"] # PROBE-B type classifier prompt (candidate 36a validator, NOT the shipped role) CLASSIFIER_SYS = """Ты — классификатор терминов книги «蛊真人» (перевод zh→ru). Тебе дан список исходных терминов с контекстами из текста. По КАЖДОМУ определи КЛАСС из ровно пяти: name — имя человека или собственное имя клана/семьи (транслитерируется) place — собственное географическое название (транслитерируется) title — должность, звание, титул, обращение, родственная роль (переводится по смыслу) term — нарицательное, понятие, реалия культивации, идиома (переводится по смыслу) nickname — прозвище/эпитет Решай по ИСХОДНИКУ и КОНТЕКСТУ, а не по тому, «выглядит ли оно как имя». Общее нарицательное (пруд, таверна, комната, источник) — это term/place по смыслу, НЕ имя. Реалия культивации (元石, 元海) — это term, даже если звучит как имя. Формат ответа — по одной строке на термин, ровно два поля через ТАБУЛЯЦИЮ: термин и класс. Пример: 方源\tname""" # CLEAN classifier — NO book-specific examples (fixes the answer-leak in expB) CLASSIFIER_SYS_CLEAN = """Ты — классификатор терминов книги при переводе zh→ru. Тебе дан список исходных терминов с контекстами из текста. По КАЖДОМУ определи КЛАСС из ровно пяти: name — имя человека или собственное имя клана/семьи place — собственное географическое или локационное название title — должность, звание, титул, обращение или родственная роль term — нарицательное существительное, понятие, реалия или идиома nickname — прозвище или эпитет Решай ТОЛЬКО по смыслу исходника и по контексту, а не по тому, «выглядит ли оно как имя». Формат ответа — по одной строке на термин, ровно два поля через ТАБУЛЯЦИЮ: термин и класс.""" def classifier_user(cands): b=["Термины:\n"] for c in cands: b.append(f"### {c['src']}") for k in c["ctx"][:3]: b.append(f"ctx: {k}") b.append("") return "\n".join(b).rstrip("\n") if __name__=="__main__": ap=argparse.ArgumentParser(); ap.add_argument("--mode",required=True); a=ap.parse_args() cl=client(); total=0.0 if a.mode=="gate": block=render_batch([cand(x) for x in DENG]); s,u=render_terminologist(block) rec=call(cl,s,u,16000,"gateA1-dengbatch"); total+=rec["cost_usd"] elif a.mode=="A0": # SCATTERED control: each 等-term ALONE (mimics batch split) for x in DENG: block=render_batch([cand(x)]); s,u=render_terminologist(block) rec=call(cl,s,u,16000,f"A0-scattered-{x}"); total+=rec["cost_usd"] elif a.mode=="A1rep": # co-batched reps (stochastic stability of unification) for i in (2,3): block=render_batch([cand(x) for x in DENG]); s,u=render_terminologist(block) rec=call(cl,s,u,16000,f"A1-dengbatch-rep{i}"); total+=rec["cost_usd"] elif a.mode=="Z0": # 转-series scattered for x in ZHUAN: block=render_batch([cand(x)]); s,u=render_terminologist(block) rec=call(cl,s,u,16000,f"Z0-scattered-{x}"); total+=rec["cost_usd"] elif a.mode=="Z1": # 转-series co-batched block=render_batch([cand(x) for x in ZHUAN]); s,u=render_terminologist(block) rec=call(cl,s,u,16000,"Z1-zhuanbatch"); total+=rec["cost_usd"] elif a.mode=="expB": # type classifier over name/place/title + realia controls want=[s for s,r in BS.items() if r["type"] in ("name","place","title","nickname")] # add realia controls known mis-typed or correct for extra in ("元石","元海","真元","蛊室","空窍"): if extra not in want: want.append(extra) cands=[cand(s) for s in want] u=classifier_user(cands) rec=call(cl,CLASSIFIER_SYS,u,16000,"expB-typeclassify",temperature=0.0); total+=rec["cost_usd"] elif a.mode=="expBclean": # RE-RUN with neutral prompt (no answer leak) want=[s for s,r in BS.items() if r["type"] in ("name","place","title","nickname")] for extra in ("元石","元海","真元","蛊室","空窍"): if extra not in want: want.append(extra) cands=[cand(s) for s in want] u=classifier_user(cands) rec=call(cl,CLASSIFIER_SYS_CLEAN,u,16000,"expBclean-typeclassify",temperature=0.0); total+=rec["cost_usd"] elif a.mode=="A0rep": # scattered 等 rep (de-confound stochasticity) for x in DENG: block=render_batch([cand(x)]); s,u=render_terminologist(block) rec=call(cl,s,u,16000,f"A0rep-scattered-{x}"); total+=rec["cost_usd"] elif a.mode=="A0faithful": # FAITHFUL scattered: temp 0, WITH evidence + full multi-line ctx (author!=reviewer fix) for r in (1,2): for x in DENG: block=render_batch([cand(x)]); s,u=render_terminologist(block) rec=call(cl,s,u,16000,f"A0f-scattered-{x}-r{r}",temperature=0.0); total+=rec["cost_usd"] elif a.mode=="A1faithful": # FAITHFUL co-batched: temp 0, WITH evidence + full ctx for r in (1,2): block=render_batch([cand(x) for x in DENG]); s,u=render_terminologist(block) rec=call(cl,s,u,16000,f"A1f-dengbatch-r{r}",temperature=0.0); total+=rec["cost_usd"] print(f"\n=== spent this invocation: ${total:.6f} ===")