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