204 lines
13 KiB
Python
204 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""Полигон, пакет-7: АДВЕРСАРИАЛЬНАЯ ревизия собственной фазы B (мандат самопроверки 12.07).
|
||
|
||
Зачем отдельным скриптом. Объективный слой отчёта B (§B1) считался разовыми heredoc-ами, то есть
|
||
именно тот слой, на который отчёт опёр ВСЕ выводы после отбраковки судей, не имел воспроизводимого
|
||
кодового пути. Здесь он есть, и здесь же — проверки, которых в фазе B не было и которые её выводы
|
||
частично опровергают.
|
||
|
||
Что проверяется (каждая проверка печатает и замер, и то, что заявлял отчёт):
|
||
1. §B1.1 совпадение с подписью владельца — пересчёт независимым кодом;
|
||
2. покрытие набора жанровым словарём — включая гипотетический ПОЛНЫЙ словарь §Z4-Т;
|
||
3. состав промпта КАЖДОГО арма по СЫРЬЮ — какой блок в каком арме реально был;
|
||
4. вырожденные айтемы: сколько термов, где все кандидаты — одна и та же строка;
|
||
5. измерим ли вообще контроль §A2.4 п.4 (уникальна ли строка GOLD среди кандидатов);
|
||
6. пере-счёт BWS ТЕМ ЖЕ скорером на невырожденных айтемах — F1/F1b переворачиваются;
|
||
7. катастрофы на GOLD: пометил ли судья одинаковые строки одинаково;
|
||
8. чем на самом деле объясняется разрыв прокси §B1.3 на ловушке S5;
|
||
9. деньги из сырья каждого вызова против леджера отчёта.
|
||
|
||
Запуск: audit_phaseB.py --dir <скретчпад/phaseB> [--z4t <файл со списком src полного словаря>]
|
||
Сети не требует, платных вызовов не делает.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import collections
|
||
import json
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
ARMS = ["P0b", "P0", "P1", "P2", "P3", "P4"]
|
||
# Полный список src таблицы §Z4-Т (жанровый словарь «как если бы подписали всё»).
|
||
Z4T_SRC = ["修炼", "修真", "修行", "修士", "修为", "真气", "灵气", "筑基", "金丹", "丹田", "元神",
|
||
"元婴", "经脉", "穴位", "心法", "秘籍", "功法", "法宝", "丹药", "符箓", "轻功", "内功",
|
||
"内力", "境界", "飞升", "天劫", "灵石", "长老", "掌门", "剑修", "妖", "魔", "仙", "江湖",
|
||
"走火入魔", "资质", "凡人"]
|
||
# Маркеры блоков промпта — по ним видно, что арм РЕАЛЬНО получил (а не что задумывалось).
|
||
BLOCKS = {
|
||
"правило-имени": "если термин — имя собственное, передавай его транскрипцией",
|
||
"Палладий": "система Палладия",
|
||
"лемма": "Русский — язык с богатой морфологией",
|
||
"словарь": "Жанровый словарь пары",
|
||
"рубрика": "K1 верность концепту (то ли это понятие)",
|
||
"справка": "СЛОВАРНАЯ СПРАВКА из китайско-русского словаря",
|
||
"KWIC": " контекст: ",
|
||
}
|
||
PRICE = {"deepseek-v4-flash": (0.14, 0.28, 0.0028)}
|
||
|
||
|
||
def norm(s: str | None) -> str:
|
||
return re.sub(r"\s+", " ", (s or "").strip().lower())
|
||
|
||
|
||
def bws(rows: list[dict], crit: str, subset: set[str] | None = None) -> dict[str, float]:
|
||
"""Нормировка ровно как в score_bws.py: (плюсы − минусы) / число появлений арма."""
|
||
plus, minus, seen = collections.Counter(), collections.Counter(), collections.Counter()
|
||
for r in rows:
|
||
if subset is not None and r["id"] not in subset:
|
||
continue
|
||
v = (r.get("verdicts") or {}).get(crit)
|
||
if not isinstance(v, dict):
|
||
continue
|
||
l2a = r["letter2arm"]
|
||
for arm in l2a.values():
|
||
seen[arm] += 1
|
||
b, w = l2a.get(str(v.get("best"))), l2a.get(str(v.get("worst")))
|
||
if b:
|
||
plus[b] += 1
|
||
if w:
|
||
minus[w] += 1
|
||
return {a: (plus[a] - minus[a]) / n for a, n in seen.items() if n}
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--dir", required=True, type=Path, help="каталог фазы B (arms_final.json, j*.json, raw/)")
|
||
a = ap.parse_args()
|
||
D = a.dir
|
||
|
||
res = json.loads((D / "arms_final.json").read_text(encoding="utf-8"))["results"]
|
||
by, gold = collections.defaultdict(dict), {}
|
||
for r in res:
|
||
by[r["id"]][r["arm"]] = (r["dst"] or "").strip()
|
||
if r.get("gold"):
|
||
gold[r["id"]] = r["gold"].strip()
|
||
|
||
def cands(tid: str) -> dict[str, str]:
|
||
m = dict(by[tid])
|
||
if tid in gold:
|
||
m["GOLD"] = gold[tid]
|
||
return m
|
||
|
||
print(f"=== 1. §B1.1 совпадение с подписью (термов с эталоном: {len(gold)}) ===")
|
||
hit, tot = collections.Counter(), collections.Counter()
|
||
per = collections.defaultdict(collections.Counter)
|
||
pert = collections.defaultdict(collections.Counter)
|
||
for r in res:
|
||
if not r.get("gold"):
|
||
continue
|
||
tot[r["arm"]] += 1
|
||
pert[r["sample"]][r["arm"]] += 1
|
||
if norm(r["dst"]) == norm(r["gold"]):
|
||
hit[r["arm"]] += 1
|
||
per[r["sample"]][r["arm"]] += 1
|
||
for arm in ARMS:
|
||
print(f" {arm:4s} {hit[arm]}/{tot[arm]} = {hit[arm]/tot[arm]:.3f}")
|
||
for s in sorted(pert):
|
||
print(f" {s}: " + " ".join(f"{x}={per[s][x]}/{pert[s][x]}" for x in ARMS))
|
||
print(" отчёт §B1.1: 12/19 · 19/19 · 18/19 · 18/19 · 17/19 · 17/19")
|
||
|
||
print("\n=== 2. покрытие набора жанровым словарём ===")
|
||
srcs = {r["src"] for r in res}
|
||
fed = sorted(srcs & {"修真", "修行", "丹田", "经脉", "灵气", "金丹"})
|
||
full = sorted(srcs & set(Z4T_SRC))
|
||
print(f" 6 строк, реально уехавших в промпт: {len(fed)}/80 {fed}")
|
||
print(f" ПОЛНЫЙ словарь §Z4-Т (37 строк): {len(full)}/80 {full}")
|
||
print(" ⇒ вывод §B1.2 (словарь инертен на этом наборе) от подачи НЕ зависит")
|
||
|
||
print("\n=== 3. состав промпта каждого арма ПО СЫРЬЮ ===")
|
||
first: dict[str, str] = {}
|
||
for p in sorted((D / "raw").glob("*.json")) + sorted((D / "raw_P4").glob("*.json")):
|
||
r = json.loads(p.read_text(encoding="utf-8"))
|
||
first.setdefault(r["arm"], r["prompt"])
|
||
for arm in ARMS:
|
||
pr = first.get(arm, "")
|
||
print(f" {arm:4s} " + " ".join(f"{k}={'ДА' if v in pr else '--'}" for k, v in BLOCKS.items()))
|
||
print(" ⇒ ПРАВИЛО ИМЕНИ СОБСТВЕННОГО есть во ВСЕХ шести армах: контрольного арма без него НЕТ")
|
||
|
||
print("\n=== 4-5. вырожденные айтемы и измеримость контроля §A2.4 п.4 ===")
|
||
deg = {t for t in by if len({norm(v) for v in cands(t).values()}) == 1}
|
||
nondeg = set(by) - deg
|
||
uniq = [t for t in gold if norm(gold[t]) not in {norm(v) for v in by[t].values()}]
|
||
print(f" термов всего {len(by)}; ВСЕ кандидаты идентичны у {len(deg)}; невырожденных {len(nondeg)}")
|
||
print(f" из {len(gold)} термов с GOLD полностью вырождены {len(deg & set(gold))}")
|
||
print(f" термов, где строка GOLD УНИКАЛЬНА среди кандидатов: {len(uniq)}/{len(gold)}")
|
||
print(" ⇒ контроль «отличает ли судья подпись от случайного варианта» на этом наборе НЕИЗМЕРИМ")
|
||
|
||
jf = [D / "j1.json", D / "j2.json"]
|
||
rows = [r for f in jf for r in json.loads(f.read_text(encoding="utf-8"))["rows"] if not r.get("reverse")]
|
||
print("\n=== 6. BWS тем же скорером: все айтемы против невырожденных ===")
|
||
for label, sub in [("все 80 (как в отчёте)", None), ("45 невырожденных", nondeg)]:
|
||
k1 = bws(rows, "K1", sub)
|
||
d1 = abs(k1.get("P1", 0) - k1.get("P0b", 0))
|
||
d1b = abs(k1.get("P2", 0) - k1.get("P0", 0))
|
||
print(f" [{label}] " + " · ".join(f"{a} {v:+.3f}" for a, v in sorted(k1.items(), key=lambda x: -x[1])))
|
||
print(f" F1 |P1−P0b|={d1:.3f} → {'ПРОВАЛ' if d1 < 0.10 else 'НЕ провал'} "
|
||
f"F1b |P2−P0|={d1b:.3f} → {'ПРОВАЛ' if d1b < 0.10 else 'НЕ провал'} (порог 0.10)")
|
||
print(" ⇒ на невырожденных айтемах оба критерия переворачиваются, а P0b встаёт ПОСЛЕДНИМ —")
|
||
print(" то есть судьи разрыв «контексты работают» ВИДЯТ; его прятали вырожденные термы")
|
||
|
||
print("\n=== 7. катастрофы на GOLD: согласованность судьи ===")
|
||
j1 = json.loads((D / "j1.json").read_text(encoding="utf-8"))["rows"]
|
||
ncat = 0
|
||
for row in j1:
|
||
l2a = row.get("letter2arm") or {}
|
||
inv = {v: k for k, v in l2a.items()}
|
||
if "GOLD" not in inv or inv["GOLD"] not in (row.get("catastrophic") or []):
|
||
continue
|
||
ncat += 1
|
||
c = cands(row["id"])
|
||
twin = sorted(L for L, arm in l2a.items() if norm(c.get(arm, "")) == norm(c["GOLD"]))
|
||
cat = sorted(row.get("catastrophic") or [])
|
||
print(f" {row['id']:22s} {c['GOLD']!r:26s} букв с тем же текстом={twin} помечено={cat}"
|
||
f" {'СОГЛАСОВАНО' if twin == cat else '← РАСХОЖДЕНИЕ'}")
|
||
print(f" всего катастроф на GOLD: {ncat} — и это несогласие судьи с каноном ПО СУЩЕСТВУ,")
|
||
print(" а не неспособность отличить подпись: одинаковый текст он метит одинаково")
|
||
|
||
print("\n=== 8. чем объясняется разрыв прокси §B1.3 на ловушке S5 ===")
|
||
s5 = collections.defaultdict(dict)
|
||
for r in res:
|
||
if r["sample"] == "S5":
|
||
s5[r["src"]][r["arm"]] = r["dst"]
|
||
for src, m in s5.items():
|
||
n = len({norm(v) for v in m.values()})
|
||
mark = " ← ЕДИНСТВЕННЫЙ источник разрыва P0 против P2/P3/P4" if src == "春秋蝉" else ""
|
||
print(f" {src:6s} различных вариантов={n}{mark}")
|
||
print(" ⇒ строка S5 прокси НЕ является уликой, независимой от наблюдения по 春秋蝉 (§B1.4)")
|
||
|
||
print("\n=== 9. деньги из сырья против леджера ===")
|
||
pi, po, pc = PRICE["deepseek-v4-flash"]
|
||
grand = 0.0
|
||
for sub, label, claim in [("raw", "армы, первый прогон", 0.07791), ("raw_P4", "P4 пере-прогон", 0.01413)]:
|
||
ti = to = tc = 0
|
||
for p in sorted((D / sub).glob("*.json")):
|
||
u = json.loads(p.read_text(encoding="utf-8"))["usage"]
|
||
ti += u["in"]; to += u["out"]; tc += u.get("cached") or 0
|
||
usd = (ti - tc) / 1e6 * pi + tc / 1e6 * pc + to / 1e6 * po
|
||
grand += usd
|
||
print(f" {label:22s} in={ti:7d} out={to:6d} cached={tc:6d} ${usd:.5f} (отчёт ${claim:.5f})")
|
||
for f in ["j1.json", "j2.json", "j3.json"]:
|
||
led = json.loads((D / f).read_text(encoding="utf-8"))["ledger"]
|
||
usd = sum(x["usd"] for x in led)
|
||
grand += usd
|
||
nofin = sum(1 for x in led if x.get("finish") != "stop")
|
||
print(f" {f:22s} вызовов={len(led)} ${usd:.5f} finish!=stop={nofin}")
|
||
print(f" ИТОГО ${grand:.5f} (отчёт $0.31030)")
|
||
print(" NB: сырьё СУДЕЙ содержит только id/prompt/response — usage у них живёт лишь в леджере,")
|
||
print(" вопреки формулировке §B0.4 «сырьё каждого вызова … usage … сохраняется целиком»")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|