168 lines
10 KiB
Python
168 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""Ф0.4 детектор №2, ШАГ 2 — ПОСЕГМЕНТНЫЙ QE-флаг с выравниванием zh↔ru.
|
||
|
||
Почему шаг 2 существует. Шаг 1 (`qe_bench.py`) намерил, что на уровне ЦЕЛОГО ОКНА MetricX порчу
|
||
практически не видит: медиана Δ +0.035 при направлении 4/6 — то есть неотличимо от нуля. На
|
||
одиночном предложении та же модель разделила отрицание и его инверсию на Δ +4.9. Вывод не «ранкер
|
||
слеп», а «единица не та»: одна испорченная фраза в окне на тысячу слов растворяется в усреднении.
|
||
Арму C нужен не балл окна, а СПАН для починки, поэтому единицей делается предложение.
|
||
|
||
Цена этого решения — выравнивание. zh и ru дробят предложения по-разному (zh→ru исторически
|
||
расширяется ×2.2–2.7, exp07), поэтому пара «i-е zh ↔ i-е ru» неверна. Берётся монотонное
|
||
выравнивание по длине (схема Гейла–Чёрча: длины предложений в переводе пропорциональны, отклонения
|
||
штрафуются квадратично) с операциями 1-1, 1-2, 2-1, 1-0, 0-1. Это детерминированный $0-код без
|
||
моделей — он не должен становиться вторым источником ошибок, поэтому качество выравнивания
|
||
печатается отдельно (доля сегментов, попавших в пару).
|
||
|
||
Как читается результат. Позитивы известны посимвольно: в каждом окне ровно одно предложение несёт
|
||
посадку k1 (выдуманное слово) и одно — k2 (смысловая инверсия). Меряются две вещи, и они разные:
|
||
ПАРНАЯ Δ — тот же сегмент с посадкой и без. Отвечает «видит ли ранкер порчу в принципе»,
|
||
контролируя трудность сегмента. Это верхняя граница.
|
||
АБСОЛЮТНЫЙ порог — можно ли отделить испорченный сегмент от здоровых ОДНИМ числом, не зная,
|
||
где искать. Только это и есть детектор; парная Δ детектором быть не может.
|
||
|
||
⚠ Шкала MetricX перевёрнута: больше = хуже.
|
||
|
||
Запуск (офлайн, $0, ~10 мин на CPU):
|
||
~/.venvs/tm-qe/bin/python eval/role_topology/qe_segment.py
|
||
"""
|
||
from __future__ import annotations
|
||
import json
|
||
import statistics
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
REPO = Path("/home/ubuntu/projects/textmachine")
|
||
sys.path.insert(0, str(REPO / "eval" / "role_topology"))
|
||
sys.path.insert(0, str(REPO / "eval" / "bank_arbitration"))
|
||
sys.path.insert(0, str(REPO / "eval" / "editor_contract"))
|
||
|
||
from align import align, split_ru, split_zh # noqa: E402
|
||
from qe_bench import QE, MODEL # noqa: E402
|
||
|
||
OUT = Path.home() / "books" / "role-topology"
|
||
# ⚠ Имя артефакта ЗАВИСИТ ОТ МОДЕЛИ. Первая редакция писала в фиксированное имя, и прогон XL
|
||
# МОЛЧА затёр сырьё прогона large — числа в отчёте остались бы «со слов сессии». Ровно этот
|
||
# дефект ревью-шапка эксп-20 назвала невоспроизводимым замером и запретила повторять.
|
||
TAG = "xl" if "XL" in MODEL else "large"
|
||
|
||
|
||
def main() -> None:
|
||
import editor_wire_probe as P # noqa: PLC0415
|
||
from inject_probe import pick_windows # noqa: PLC0415
|
||
|
||
qe = QE()
|
||
rows: list[dict] = []
|
||
print("ПОСЕГМЕНТНЫЙ QE: выравнивание zh↔ru + балл MetricX на пару (больше = хуже)\n")
|
||
for k, (_, src, _) in enumerate(pick_windows()):
|
||
clean = P.draft_of(k)
|
||
zs, rs = split_zh(src), split_ru(clean)
|
||
pairs = align(zs, rs)
|
||
paired = sum(1 for a, b in pairs if a and b)
|
||
print(f"w{k}: zh {len(zs):3d} предл. · ru {len(rs):3d} предл. · пар {len(pairs):3d}, "
|
||
f"из них связанных {paired} ({paired / max(1, len(pairs)):.0%})")
|
||
# Испорченные варианты того же окна — по одному на класс.
|
||
variants = {"clean": clean}
|
||
for cls in ("k1", "k2"):
|
||
bad = clean
|
||
for c, old, new, _b, _f in P.DEFECTS[k]:
|
||
if c == cls:
|
||
bad = bad.replace(old, new, 1)
|
||
variants[cls] = bad
|
||
for name, text in variants.items():
|
||
rsv = split_ru(text)
|
||
# Порча не меняет ЧИСЛО предложений (замена внутри фразы), поэтому выравнивание
|
||
# переиспользуется — иначе сравнивались бы разные разбиения, а не разные тексты.
|
||
if len(rsv) != len(rs):
|
||
print(f" ⚠ w{k}/{name}: число ru-предложений сдвинулось "
|
||
f"({len(rs)}→{len(rsv)}) — окно вне парного сравнения")
|
||
continue
|
||
for si, di in pairs:
|
||
if not si or not di:
|
||
continue
|
||
s = " ".join(zs[i] for i in si)
|
||
d = " ".join(rsv[i] for i in di)
|
||
rows.append(dict(window=k, variant=name, di=tuple(di),
|
||
score=qe.score(s, d), text_len=len(d)))
|
||
# ДЕКОЙ (D39.46б). Без него отрицательный результат неинтерпретируем: «ранкер не увидел
|
||
# посадку» и «ранкер слеп на этом материале» — разные вердикты с разными следствиями.
|
||
# Декой — заведомая грубая порча: ru-сторона сегмента подменяется ru-стороной ДРУГОГО сегмента
|
||
# того же окна. Ранкер, который не ловит ЭТО, не годен ни для чего.
|
||
decoy: list[float] = []
|
||
for k, (_, src, _) in enumerate(pick_windows()):
|
||
clean = P.draft_of(k)
|
||
zs, rs = split_zh(src), split_ru(clean)
|
||
pairs = [(a, b) for a, b in align(zs, rs) if a and b]
|
||
for idx, (si, di) in enumerate(pairs):
|
||
if idx + 2 >= len(pairs):
|
||
break
|
||
s = " ".join(zs[i] for i in si)
|
||
own = " ".join(rs[i] for i in di)
|
||
other = " ".join(rs[i] for i in pairs[idx + 2][1])
|
||
decoy.append(qe.score(s, other) - qe.score(s, own))
|
||
if decoy:
|
||
print(f"\nДЕКОЙ (ru-сторона подменена чужим сегментом), n={len(decoy)}: "
|
||
f"медиана Δ {statistics.median(decoy):+.3f}, Δ>0 в "
|
||
f"{sum(1 for x in decoy if x > 0)}/{len(decoy)}, макс {max(decoy):+.3f}")
|
||
(OUT / f"qe-decoy-{TAG}.json").write_text(json.dumps(decoy), encoding="utf-8")
|
||
|
||
(OUT / f"qe-segments-{TAG}.json").write_text(
|
||
json.dumps([{**r, "di": list(r["di"])} for r in rows], ensure_ascii=False),
|
||
encoding="utf-8")
|
||
report(rows, P)
|
||
|
||
|
||
def report(rows: list[dict], P) -> None: # noqa: ANN001
|
||
by = {(r["window"], r["variant"], r["di"]): r["score"] for r in rows}
|
||
print("\nПАРНАЯ Δ (тот же сегмент с посадкой и без) — верхняя граница чувствительности")
|
||
print(f"{'класс':8s}{'сегментов':>11s}{'медиана Δ':>12s}{'Δ>0':>7s}{'макс Δ':>9s}")
|
||
print("-" * 47)
|
||
hits = {}
|
||
for cls in ("k1", "k2"):
|
||
deltas = []
|
||
for (w, v, di), sc in by.items():
|
||
if v != cls:
|
||
continue
|
||
base = by.get((w, "clean", di))
|
||
if base is None:
|
||
continue
|
||
d = sc - base
|
||
if abs(d) > 1e-9: # сегмент, которого посадка коснулась
|
||
deltas.append((d, w, di))
|
||
if not deltas:
|
||
print(f"{cls:8s}{'—':>11s}")
|
||
continue
|
||
hits[cls] = deltas
|
||
vals = [d for d, _, _ in deltas]
|
||
print(f"{cls:8s}{len(vals):11d}{statistics.median(vals):+12.3f}"
|
||
f"{sum(1 for x in vals if x > 0):7d}{max(vals):+9.3f}")
|
||
|
||
print("\nАБСОЛЮТНЫЙ ПОРОГ — это и есть детектор: отделяем испорченный сегмент, не зная где он")
|
||
clean_scores = [r["score"] for r in rows if r["variant"] == "clean"]
|
||
if not clean_scores:
|
||
return
|
||
print(f"здоровых сегментов {len(clean_scores)}, их баллы: "
|
||
f"медиана {statistics.median(clean_scores):.2f}, "
|
||
f"p90 {sorted(clean_scores)[int(0.9 * len(clean_scores))]:.2f}, "
|
||
f"макс {max(clean_scores):.2f}")
|
||
print(f"{'порог':>7s}{'recall k1':>11s}{'recall k2':>11s}{'ложных/окно':>13s}{'precision':>11s}")
|
||
print("-" * 53)
|
||
n_windows = len({r["window"] for r in rows})
|
||
for thr in (5, 7, 9, 11, 13, 15):
|
||
fp = sum(1 for s in clean_scores if s >= thr)
|
||
line = [f"{thr:7d}"]
|
||
tp_all = 0
|
||
for cls in ("k1", "k2"):
|
||
got = sum(1 for d, w, di in hits.get(cls, []) if by[(w, cls, di)] >= thr)
|
||
tot = len(hits.get(cls, []))
|
||
tp_all += got
|
||
line.append(f"{got}/{tot}".rjust(11))
|
||
line.append(f"{fp / max(1, n_windows):13.1f}")
|
||
line.append(f"{tp_all / max(1, tp_all + fp):11.3f}")
|
||
print("".join(line))
|
||
print("\nЧитать: если ложных на окно много при recall<1, одним порогом класс не берётся —\n"
|
||
"тогда арм C в режиме «реальные детекторы» питается не QE, а связкой сигналов.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|