254 lines
14 KiB
Python
254 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""Ф2.5 memory-eval (INDICATIVE run + harness scaffold) — the deciding-experiment design
|
||
for the memory-bank bet (research/13 §Вердикт, decisions-log D-пилот).
|
||
|
||
QUESTION (owner's central fear + go/no-go on the bank bet): is DETERMINISTIC SELECTIVE
|
||
glossary injection (our bank) worth it vs just putting the WHOLE glossary in a long context,
|
||
and does a WRONG injection silently degrade the translation? WMT25 three-mode protocol:
|
||
|
||
C0 no glossary (baseline — model's base behaviour)
|
||
C1 SELECTIVE bank injection (only records whose keys fire in the chunk) — OUR bank
|
||
C2 FULL glossary + sliding summary in the prompt — "just long context"
|
||
C3 RANDOM/WRONG glossary (dst shuffled) — adversarial arm, directly tests silent degradation
|
||
|
||
Decision rule (pre-registered): C1≈C2 quality but C1 cheaper → bank justified; C2≫C1 →
|
||
need fuller context; C3 drops vs C0 on FIDELITY → wrong injection harms → post-check/disposition
|
||
justified (owner's fear confirmed & mitigated, not just asserted).
|
||
|
||
INJECTION = eval-side MIRROR of backend/internal/pipeline/memory.go (THE SOURCE OF TRUTH):
|
||
normalized exact key match, per-language min-key ban (Han≥2 / phonetic≥3), collision-prone
|
||
short phonetic key → AMBIGUOUS, longest-match containment, spoiler since/until window, sticky
|
||
scene-inertia. Cross-checked against the Go unit tests; on any divergence the Go code wins.
|
||
(eval/memory_hotpath.py is the PRE-cc57c7b prototype with a global MIN_KEY=2 and no
|
||
collision-downgrade → do NOT reuse it; this file reflects the current Go contract.)
|
||
|
||
METRICS: (a) consistency — decl-aware (declmetric.py, pymorphy3) approved-dst rendering across
|
||
overlapping chunks; (b) fidelity/omission — LLM judge of a DIFFERENT family (cross-family) vs
|
||
source + Rogov anchor; (c) cost — input/output tokens per condition.
|
||
|
||
INDICATIVE (small N, LLM judge). The deciding version adds human BWS + owner corpus (see
|
||
09-pilot-protocol.md). Usage: eval/.venv/bin/python eval/pilot/memory_eval.py --dry-run
|
||
"""
|
||
from __future__ import annotations
|
||
import argparse, json, re, sys, unicodedata
|
||
from pathlib import Path
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||
import refusal_bench as rb
|
||
import declmetric as dm
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
SAMPLES = ROOT.parent / "data" / "samples"
|
||
OUTDIR = ROOT.parent / "data" / "pilot"
|
||
OUTDIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
# --- normalization (mirror of memnorm.go normalizeSourceKey) ---------------------
|
||
TRAD2SIMP = {"趙": "赵", "莊": "庄", "錢": "钱", "陳": "陈", "萬": "万", "舊": "旧", "臉": "脸",
|
||
"魯": "鲁", "鎮": "镇", "剛": "刚", "動": "动", "們": "们"}
|
||
|
||
def norm_src(s: str) -> str:
|
||
s = unicodedata.normalize("NFKC", s)
|
||
out = []
|
||
for c in s:
|
||
if unicodedata.category(c) in ("Cf",) or c in "️":
|
||
continue
|
||
c = TRAD2SIMP.get(c, c)
|
||
o = ord(c)
|
||
if 0x30A1 <= o <= 0x30F6: # katakana → hiragana
|
||
c = chr(o - 0x60)
|
||
out.append(c.lower())
|
||
return "".join(out)
|
||
|
||
HAN = re.compile(r"[一-鿿㐀-䶿]")
|
||
def significant_len(nk: str) -> int:
|
||
return sum(1 for c in nk if HAN.match(c) or c.isalpha() or
|
||
("" <= c <= "ヿ") or ("가" <= c <= ""))
|
||
def any_han(nk: str) -> bool:
|
||
return bool(HAN.search(nk))
|
||
def min_key_len(nk: str) -> int:
|
||
return 2 if any_han(nk) else 3
|
||
def collision_prone(nk: str) -> bool:
|
||
return (not any_han(nk)) and significant_len(nk) <= 3
|
||
|
||
# --- glossary (Ah-Q; dst + decl + lemma + aliases + spoiler window) --------------
|
||
# One deliberate SPOILER entry (未庄 revolution outcome) demonstrates the since/until gate.
|
||
GLOSSARY = [
|
||
{"src": "阿Q", "dst": "А-кью", "lemma_keys": ["а-кью"], "hyphen_pat": r"А[-\s]?кью",
|
||
"decl_forms": ["А-кью"], "aliases": [], "type": "name"},
|
||
{"src": "未庄", "dst": "Вэйчжуан", "lemma_keys": ["вэйчжуан"],
|
||
"decl_forms": ["Вэйчжуане", "Вэйчжуан", "Вэйчжуана"], "aliases": [], "type": "place"},
|
||
{"src": "赵太爷", "dst": "почтенный Чжао", "lemma_keys": ["чжао"],
|
||
"decl_forms": ["почтенного Чжао", "почтенный Чжао", "Чжао"], "aliases": ["赵老太爷"], "type": "name"},
|
||
{"src": "王胡", "dst": "Бородатый Ван", "lemma_keys": ["ван"],
|
||
"decl_forms": ["Бородатого Вана", "Бородатый Ван", "Бородатому Вану"], "aliases": ["王癞胡", "癞胡"], "type": "name"},
|
||
{"src": "小D", "dst": "Маленький Дэн", "lemma_keys": ["дэн"],
|
||
"decl_forms": ["Маленького Дэна", "Маленький Дэн", "Дэна"], "aliases": [], "type": "name"},
|
||
{"src": "假洋鬼子", "dst": "Поддельный заморский чёрт", "lemma_keys": ["заморский", "чёрт"],
|
||
"decl_forms": ["Поддельного заморского чёрта", "заморский чёрт"], "aliases": [], "type": "nickname"},
|
||
{"src": "吴妈", "dst": "У-ма", "lemma_keys": ["у-ма"], "hyphen_pat": r"У[-\s]?ма",
|
||
"decl_forms": ["У-ма", "У-мы"], "aliases": [], "type": "name"},
|
||
{"src": "秀才", "dst": "сюцай", "lemma_keys": ["сюцай"],
|
||
"decl_forms": ["сюцая", "сюцай", "сюцаю"], "aliases": [], "type": "title"},
|
||
{"src": "尼姑", "dst": "монашка", "lemma_keys": ["монашка"],
|
||
"decl_forms": ["монашку", "монашка", "монашки"], "aliases": [], "type": "term"},
|
||
# SPOILER demo: only valid from chapter 7 (revolution). Injected earlier → hard reject.
|
||
{"src": "革命党", "dst": "революционеры", "lemma_keys": ["революционер"],
|
||
"decl_forms": ["революционеров", "революционеры"], "aliases": [], "type": "term", "since_ch": 7},
|
||
]
|
||
|
||
def eligible_keys(entry: dict) -> list[tuple[str, str]]:
|
||
"""(normalized_key, raw) surfaces (src+aliases) passing the per-lang min-key ban."""
|
||
out = []
|
||
for raw in [entry["src"]] + entry.get("aliases", []):
|
||
nk = norm_src(raw)
|
||
if nk and significant_len(nk) >= min_key_len(nk):
|
||
out.append((nk, raw))
|
||
return out
|
||
|
||
def select(chunk: str, chapter: int, sticky_prev: set[str]) -> dict:
|
||
"""Mirror of MemoryBank.Select: which entries inject, with disposition + spoiler rejects."""
|
||
nchunk = norm_src(chunk)
|
||
fired = {} # src → (matched_key, is_collision_prone)
|
||
for e in GLOSSARY:
|
||
best = None
|
||
for nk, raw in eligible_keys(e):
|
||
if nk in nchunk and (best is None or len(nk) > len(best)):
|
||
best = nk
|
||
if best is not None:
|
||
fired[e["src"]] = best
|
||
# longest-match containment: drop a fired key fully inside a strictly-longer fired key
|
||
keys = {e["src"]: fired[e["src"]] for e in GLOSSARY if e["src"] in fired}
|
||
suppressed = set()
|
||
for s1, k1 in keys.items():
|
||
for s2, k2 in keys.items():
|
||
if s1 != s2 and k1 in k2 and len(k1) < len(k2):
|
||
suppressed.add(s1)
|
||
injected, rejected = [], []
|
||
active = set()
|
||
for e in GLOSSARY:
|
||
s = e["src"]
|
||
if s in fired and s not in suppressed:
|
||
if e.get("since_ch", 0) and chapter < e["since_ch"]:
|
||
rejected.append({"src": s, "reason": f"spoiler since_ch={e['since_ch']}>{chapter}"})
|
||
continue
|
||
disp = "ambiguous" if collision_prone(fired[s]) else "confirmed"
|
||
injected.append({**e, "_disp": disp, "_via": fired[s]})
|
||
active.add(s)
|
||
# sticky scene-inertia (depth handled by caller passing prior active set)
|
||
for e in GLOSSARY:
|
||
s = e["src"]
|
||
if s in sticky_prev and s not in active and not (e.get("since_ch", 0) and chapter < e["since_ch"]):
|
||
injected.append({**e, "_disp": "confirmed", "_via": "sticky"})
|
||
return {"injected": injected, "rejected": rejected, "active": active}
|
||
|
||
# --- prompt assembly per condition -----------------------------------------------
|
||
SYSTEM = ("Ты профессиональный литературный переводчик. Переведи фрагмент китайского "
|
||
"произведения на русский язык. Сохрани все реплики и детали без пропусков; стиль — "
|
||
"живой литературный русский. Выведи ТОЛЬКО перевод.")
|
||
GHEAD = "ГЛОССАРИЙ (используй эти утверждённые переводы имён и терминов последовательно):"
|
||
|
||
def gloss_block(entries: list[dict], full: bool = False) -> str:
|
||
if not entries:
|
||
return ""
|
||
lines = []
|
||
for e in entries:
|
||
line = f"{e['src']} → {e['dst']}"
|
||
if not full and e.get("_disp") == "ambiguous":
|
||
line += " ⟨проверить⟩"
|
||
lines.append(line)
|
||
return GHEAD + "\n" + "\n".join(lines)
|
||
|
||
def build_prompt(chunk: str, condition: str, sel: dict, summary: str, shuffled: dict) -> str:
|
||
if condition == "C0":
|
||
return "ФРАГМЕНТ:\n" + chunk
|
||
if condition == "C1": # selective bank
|
||
blk = gloss_block(sel["injected"])
|
||
return (blk + "\n\n" if blk else "") + "ФРАГМЕНТ:\n" + chunk
|
||
if condition == "C2": # full glossary + sliding summary (long context)
|
||
blk = gloss_block([{**e, "_disp": "confirmed"} for e in GLOSSARY if not e.get("since_ch")], full=True)
|
||
pre = (f"КРАТКОЕ СОДЕРЖАНИЕ ПРЕДЫДУЩЕГО:\n{summary}\n\n" if summary else "")
|
||
return pre + blk + "\n\nФРАГМЕНТ:\n" + chunk
|
||
if condition == "C3": # random/wrong glossary (adversarial): fired entries with SHUFFLED dst
|
||
wrong = [{**e, "dst": shuffled[e["src"]], "_disp": "confirmed"} for e in sel["injected"]
|
||
if e["_via"] != "sticky"]
|
||
blk = gloss_block(wrong, full=True)
|
||
return (blk + "\n\n" if blk else "") + "ФРАГМЕНТ:\n" + chunk
|
||
raise ValueError(condition)
|
||
|
||
# --- chunking (reuse the coverage harness's chunker semantics) -------------------
|
||
def chunk_text(text: str, target=1200) -> list[str]:
|
||
paras = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()]
|
||
chunks, buf = [], []
|
||
for p in paras:
|
||
if buf and len("".join(buf)) + len(p) > target:
|
||
chunks.append("\n\n".join(buf)); buf = []
|
||
buf.append(p)
|
||
if buf:
|
||
chunks.append("\n\n".join(buf))
|
||
return chunks
|
||
|
||
def translate(chunk_prompt: str, model="deepseek-v4-flash") -> tuple[str, dict]:
|
||
cfg = {"deepseek-v4-flash": {"base_url": "https://api.deepseek.com/v1", "api_key_env": "DEEPSEEK_API_KEY", "max_tokens": 8000},
|
||
"grok-4.20-0309-non-reasoning": {"base_url": "https://api.x.ai/v1", "api_key_env": "XAI_API_KEY", "max_tokens": 8000, "temperature": 0.3}}[model]
|
||
t, err, usage = rb.call_provider({"name": model, "model": model, **cfg}, SYSTEM, chunk_prompt, timeout=200)
|
||
return (t or ""), (usage or {})
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--src", default=str(SAMPLES / "zh" / "luxun-ah-q-ch5-9.txt"))
|
||
ap.add_argument("--chapter", type=int, default=6, help="chapter number for the spoiler gate")
|
||
ap.add_argument("--conditions", default="C0,C1,C2,C3")
|
||
ap.add_argument("--model", default="grok-4.20-0309-non-reasoning")
|
||
ap.add_argument("--max-chunks", type=int, default=5)
|
||
ap.add_argument("--dry-run", action="store_true")
|
||
ap.add_argument("--out", default=str(OUTDIR / "memory_eval.json"))
|
||
args = ap.parse_args()
|
||
|
||
text = Path(args.src).read_text(encoding="utf-8")
|
||
chunks = chunk_text(text)[:args.max_chunks]
|
||
conditions = args.conditions.split(",")
|
||
# deterministic wrong-dst shuffle for C3 (rotate dst among present names)
|
||
names = [e for e in GLOSSARY if e["type"] in ("name", "place")]
|
||
rot = {names[i]["src"]: names[(i + 1) % len(names)]["dst"] for i in range(len(names))}
|
||
shuffled = {e["src"]: rot.get(e["src"], "НЕВЕРНО") for e in GLOSSARY}
|
||
|
||
print(f"src={Path(args.src).name} chapter={args.chapter} chunks={len(chunks)} "
|
||
f"model={args.model} conditions={conditions}", file=sys.stderr)
|
||
rows, sticky_prev = [], set()
|
||
for ci, ch in enumerate(chunks):
|
||
sel = select(ch, args.chapter, sticky_prev)
|
||
present = {e["src"]: e for e in sel["injected"] if e["_via"] != "sticky"}
|
||
row = {"chunk": ci, "src_chars": len(ch), "n_injected": len(sel["injected"]),
|
||
"injected_srcs": [e["src"] for e in sel["injected"]],
|
||
"dispositions": {e["src"]: e["_disp"] for e in sel["injected"]},
|
||
"rejected_spoiler": sel["rejected"], "present_for_consistency": list(present)}
|
||
if not args.dry_run:
|
||
row["by_condition"] = {}
|
||
for cond in conditions:
|
||
prompt = build_prompt(ch, cond, sel, summary="", shuffled=shuffled)
|
||
out, usage = translate(prompt, args.model)
|
||
cons = dm.consistency(out, {s: {"dst": e["dst"], "lemma_keys": e.get("lemma_keys", []),
|
||
"hyphen_pat": e.get("hyphen_pat"), "decl_forms": e.get("decl_forms", [])}
|
||
for s, e in present.items()}) if present else None
|
||
row["by_condition"][cond] = {
|
||
"consistency_pymorphy": cons["pymorphy"]["score"] if cons else None,
|
||
"consistency_stored_decl": cons["stored_decl"]["score"] if cons else None,
|
||
"missed_pymorphy": cons["pymorphy"]["missed"] if cons else [],
|
||
"in_tok": usage.get("prompt_tokens"), "out_tok": usage.get("completion_tokens"),
|
||
"out_preview": out[:160]}
|
||
print(f" chunk{ci} {cond}: consistency(pymorphy)="
|
||
f"{row['by_condition'][cond]['consistency_pymorphy']} "
|
||
f"in={usage.get('prompt_tokens')} out={usage.get('completion_tokens')}", file=sys.stderr)
|
||
sticky_prev = sel["active"]
|
||
rows.append(row)
|
||
|
||
Path(args.out).write_text(json.dumps({"config": vars(args), "rows": rows}, ensure_ascii=False, indent=2))
|
||
print(f"\nwrote {args.out}", file=sys.stderr)
|
||
# dry-run: show the selection mechanism (spoiler reject, disposition, sticky)
|
||
if args.dry_run:
|
||
for r in rows:
|
||
print(f"chunk{r['chunk']} inj={r['injected_srcs']} disp={r['dispositions']} "
|
||
f"spoiler_rej={r['rejected_spoiler']}")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|