248 lines
10 KiB
Python
248 lines
10 KiB
Python
#!/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())
|