#!/usr/bin/env python3 """S3 — deterministic fidelity gate. No LLM, no cost. Motivation (measured 2026-07-25): asking gemini-3.5-flash-lite to "去口语化，禁止 省略" on 900–2500 char chunks silently dropped 17/49 distinct numeric tokens and 136 distinct English tokens from a 118k-char BD call — including `Kras`, `IBD`, `EDC`, `IRT` and the figure `4.5` in "raised about 4.5 million". Output/input char ratio was 0.41–0.47 regardless of chunk size, i.e. the model summarizes when told to rewrite, and prompt wording alone does not stop it. So fidelity cannot be asserted, it must be *checked*. This module compares a cleaned chunk against its source over token classes that must survive a purely stylistic rewrite: numbers every digit run, as a multiset (口语 "四幺八二" survives as Han and is not expected here, but arabic figures must) latin acronyms / gene names / product codes ≥2 chars speakers the 【…】 turn tags han_terms domain terms from the term bank that appeared in the source `check()` returns a verdict; the pipeline retries a failing chunk with a stricter prompt and, on a second failure, keeps the original text and flags it. That makes information loss a *visible, bounded* event instead of a silent one. """ from __future__ import annotations import json import re from collections import Counter from dataclasses import dataclass, field from pathlib import Path from .config import CFG BANK = CFG.termbank_dir NUM = re.compile(r"\d+(?:[.,]\d+)?%?") # S0's noise-collapse marker is synthetic metadata, not transcript content. Its # repeat count is a number ("[…重复噪声已折叠：beep×3696…]"), and scoring it as content # made one fixture report number_recall 0.00 off a single token that the cleaner had # merely reworded. Strip markers from both sides before any comparison. COLLAPSE_MARKER = re.compile(r"\[…重复噪声已折叠：.*?…\]") LATIN = re.compile(r"[A-Za-z][A-Za-z0-9\-]{1,}") SPEAKER = re.compile(r"【([^】]+)】") # Filler words a legitimate 去口语化 pass is expected to delete. Used only for the # expected-compression estimate, never for the pass/fail decision. FILLERS = ["嗯", "呃", "啊", "哦", "那个", "这个", "就是", "然后", "其实", "对对", "哈哈", "唉", "诶", "喔", "嘛", "吧", "呀", "咯", "哎"] _STOP_LATIN = { "the", "and", "for", "you", "that", "this", "with", "have", "was", "are", "not", "but", "they", "our", "can", "all", "will", "would", "yeah", "okay", "ok", "uh", "um", "so", "we", "it", "is", "to", "of", "in", "on", "at", "be", "do", "if", "or", "as", "an", "a", "i", "he", "she", "his", "her", "them", "there", "then", "just", "very", "know", "think", "like", "get", "got", "one", "also", "from", "what", "when", "who", "how", "yes", "no", "right", # --- interjections and ordinary verbs the ASR emits in caps ------------- # Measured cost of omitting these: on a 118k-char English-mixed call the gate # tripped on lost tokens "AH", "OH", "PAY", "BECAUSE", "EVEN", "WHEREAS" and # forced 53 retries for gemini-3.5-flash-lite, ~40% of them for interjections a # cleaner is *supposed* to delete. Treating them as content made the gate # expensive without making it safer. "ah", "oh", "eh", "hm", "hmm", "mm", "mhm", "huh", "wow", "hey", "hi", "well", "actually", "really", "maybe", "sure", "fine", "good", "great", "because", "even", "whereas", "however", "though", "although", "since", "while", "before", "after", "about", "into", "over", "under", "than", "such", "some", "any", "more", "most", "much", "many", "other", "same", "only", "still", "already", "always", "never", "again", "here", "where", "why", "which", "whose", "whom", "these", "those", "their", "your", "my", "me", "us", "him", "was", "were", "been", "being", "does", "did", "done", "has", "had", "having", "should", "could", "must", "may", "might", "shall", "want", "need", "make", "made", "take", "took", "give", "gave", "come", "came", "went", "said", "say", "says", "tell", "told", "see", "saw", "look", "mean", "means", "means", "let", "put", "keep", "kept", "find", "found", "feel", "felt", "seem", "seems", "sorry", "thanks", "thank", "please", "bye", "abc", "pay", "everybody", "everything", "something", "anything", "nothing", "everyone", "someone", "anyone", "nobody", "first", "second", "third", "last", "next", "both", "each", "every", "little", "big", "small", "long", "short", "high", "low", "new", "old", "time", "year", "day", "way", "thing", "things", "people", "person", "lot", "lots", "bit", "kind", "sort", "part", "point", "case", "fact", "course", "example", "problem", "question", "answer", "reason", "result", } def _load_terms() -> set[str]: p = BANK / "terms.json" if not p.exists(): return set() return set(json.loads(p.read_text()).keys()) _TERMS: set[str] | None = None def domain_terms() -> set[str]: global _TERMS if _TERMS is None: _TERMS = _load_terms() return _TERMS def _latin_tokens(text: str) -> Counter: """Meaningful Latin tokens: acronyms, gene/product codes, CamelCase. Ordinary English function words are excluded — a cleaner legitimately rewrites "Yeah, so we, we did that" and must not be punished for it.""" out = Counter() for t in LATIN.findall(text): low = t.lower() if low in _STOP_LATIN or len(t) < 2: continue # Keep: all-caps (EDC/IRT/GMP), mixed case with an inner capital (Kras is # lowercase-tail but capitalized; KRAS is caps), anything with a digit or # hyphen (BF4182, CD3-x), and capitalized proper nouns. if t.isupper() or any(c.isdigit() for c in t) or "-" in t or t[0].isupper(): out[t.upper()] += 1 return out def _han_terms(text: str, vocab: set[str]) -> Counter: out = Counter() for t in vocab: if len(t) >= 3 and t in text: # ≥3 chars: shorter ones collide constantly out[t] += text.count(t) return out @dataclass class Verdict: ok: bool ratio: float lost_numbers: list[str] = field(default_factory=list) lost_latin: list[str] = field(default_factory=list) lost_speakers: list[str] = field(default_factory=list) lost_terms: list[str] = field(default_factory=list) reasons: list[str] = field(default_factory=list) def brief(self) -> str: if self.ok: return f"ok ratio={self.ratio:.2f}" return f"FAIL ratio={self.ratio:.2f} " + "; ".join(self.reasons) def as_dict(self): return { "ok": self.ok, "ratio": round(self.ratio, 3), "lost_numbers": self.lost_numbers, "lost_latin": self.lost_latin, "lost_speakers": self.lost_speakers, "lost_terms": self.lost_terms, "reasons": self.reasons, } def check(src: str, out: str, *, min_ratio: float = 0.55, max_ratio: float = 1.35, vocab: set[str] | None = None, allow_lost_latin: int = 0) -> Verdict: """Compare a cleaned chunk against its source. min_ratio 0.55 is deliberately generous: real 去口语化 on filler-heavy Chinese speech legitimately removes 25–40% of characters. Anything below 0.55 is not cleaning, it is summarizing. """ vocab = domain_terms() if vocab is None else vocab src = COLLAPSE_MARKER.sub("", src) out = COLLAPSE_MARKER.sub("", out) ratio = len(out) / max(1, len(src)) v = Verdict(ok=True, ratio=ratio) if ratio < min_ratio: v.ok = False v.reasons.append(f"过度压缩(ratio {ratio:.2f}<{min_ratio})") if ratio > max_ratio: v.ok = False v.reasons.append(f"异常膨胀(ratio {ratio:.2f}>{max_ratio})") sn, on = Counter(NUM.findall(src)), Counter(NUM.findall(out)) lost_n = sorted((sn - on).elements()) if lost_n: v.lost_numbers = lost_n v.ok = False v.reasons.append(f"丢数字{len(lost_n)}个:{lost_n[:6]}") sl, ol = _latin_tokens(src), _latin_tokens(out) lost_l = sorted((sl - ol).elements()) if len(lost_l) > allow_lost_latin: v.lost_latin = lost_l v.ok = False v.reasons.append(f"丢英文术语{len(lost_l)}个:{lost_l[:6]}") ss, os_ = Counter(SPEAKER.findall(src)), Counter(SPEAKER.findall(out)) lost_s = sorted(set((ss - os_).elements())) if lost_s: v.lost_speakers = lost_s v.ok = False v.reasons.append(f"丢说话人标签:{lost_s[:5]}") if vocab: st, ot = _han_terms(src, vocab), _han_terms(out, vocab) lost_t = sorted(set((st - ot).elements())) if lost_t: v.lost_terms = lost_t v.ok = False v.reasons.append(f"丢领域术语{len(lost_t)}个:{lost_t[:6]}") return v def expected_compression(src: str) -> float: """Rough share of characters attributable to filler, for interpreting ratios.""" n = sum(src.count(f) * len(f) for f in FILLERS) return n / max(1, len(src)) def score_document(src: str, out: str, vocab: set[str] | None = None) -> dict: """Document-level metrics for the横评 report (not a gate).""" vocab = domain_terms() if vocab is None else vocab src = COLLAPSE_MARKER.sub("", src) out = COLLAPSE_MARKER.sub("", out) sn, on = Counter(NUM.findall(src)), Counter(NUM.findall(out)) sl, ol = _latin_tokens(src), _latin_tokens(out) st, ot = _han_terms(src, vocab), _han_terms(out, vocab) ss, os_ = set(SPEAKER.findall(src)), set(SPEAKER.findall(out)) def recall(a: Counter, b: Counter) -> float: tot = sum(a.values()) if not tot: return 1.0 kept = sum(min(c, b[k]) for k, c in a.items()) return kept / tot return { "char_ratio": round(len(out) / max(1, len(src)), 3), "filler_share": round(expected_compression(src), 3), "number_recall": round(recall(sn, on), 3), "latin_recall": round(recall(sl, ol), 3), "term_recall": round(recall(st, ot), 3), "speaker_recall": round(len(ss & os_) / len(ss), 3) if ss else 1.0, "src_numbers": sum(sn.values()), "src_latin": sum(sl.values()), "src_terms": sum(st.values()), "src_speakers": len(ss), } if __name__ == "__main__": import sys src = Path(sys.argv[1]).read_text(errors="ignore") out = Path(sys.argv[2]).read_text(errors="ignore") print(json.dumps(score_document(src, out), ensure_ascii=False, indent=2)) print(check(src, out).brief())