245 lines
13 KiB
Python
245 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""Ф0.4 — ПАКЕТЫ ДЛЯ SOL: внешний судейский контур руками владельца. $0.
|
||
|
||
Что это. Sol — флагманская reasoning-модель OpenAI в ЧАТ-подписке владельца. Она вне обеих
|
||
популяций судей пака (платные ключи + агенты сессии), поэтому её голос — третий независимый
|
||
инструмент. Сессия готовит paste-ready пакеты, владелец вставляет их руками и возвращает ответы;
|
||
транскрипты персистятся дословно.
|
||
|
||
**Пакет (а) — АРБИТРАЖ ЭКСП-21.** Там два платных судьи разошлись на величину, которая
|
||
собственным шумом не объясняется: разброс судьи-замены на повторе даёт sd перевеса 0.81 на голос,
|
||
то есть ±0.15 на контраст из 15 единиц. Отобраны ТРИ контраста по ФАКТИЧЕСКОМУ расхождению:
|
||
|
||
контраст luna gemini разрыв в sd собственного шума
|
||
A vs F +0.84 +3.03 2.19 ~14
|
||
B vs F +1.44 +3.33 1.90 ~12
|
||
A_law vs A +0.91 −0.20 1.11 ~7 ← ЗНАК РАСХОДИТСЯ
|
||
|
||
Почему именно эти. `A/F` заказан промтом. `B/F` — тот же заголовочный вопрос («покупает ли второй
|
||
проход качество») и второй по величине разрыв. `A_law/A` взят НЕ по величине: это единственный
|
||
контраст, где судьи расходятся ЗНАКОМ, а расхождение знака хуже расхождения величины — оно меняет
|
||
вердикт, а не точность. `D/A` (разрыв 0.74) и `D/D_` (1.23) не берутся: там знак у обоих судей
|
||
один, и арбитраж уточнил бы величину, которая решения не несёт.
|
||
|
||
Протокол — ПАРНЫЙ, потому что арбитражу подлежит парный замер: слепые метки, ОБА порядка, декой.
|
||
Правило зеркальности для пар-протокола живо (для абсолютного снято, D39.117 п.2в).
|
||
|
||
**Семейный запрет.** Sol не судит клетки OpenAI-моделей. В армах эксп-21 их нет (deepseek, glm),
|
||
поэтому запрет здесь не связывает — но он объявлен, и в пакете (б) по финалистам эксп-22 будет
|
||
применён механически.
|
||
|
||
Запуск: --emit выпустить пакеты и инструкцию, $0
|
||
--ingest принять ответы владельца из sol/answers/, $0
|
||
--score что сказал Sol против платных судей, $0
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import statistics as st
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
REPO = Path("/home/ubuntu/projects/textmachine")
|
||
ZONE = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(ZONE))
|
||
sys.path.insert(0, str(REPO / "eval" / "role_topology"))
|
||
|
||
import absjudge as AJ # noqa: E402
|
||
import bakeoff as BO # noqa: E402
|
||
import judges as J # noqa: E402
|
||
import money as MONEY # noqa: E402
|
||
|
||
SOL = MONEY.OUT / "sol"
|
||
ANSW = SOL / "answers"
|
||
KEYS = MONEY.OUT / "blind-keys"
|
||
CONTRASTS = [("A", "F"), ("B", "F"), ("A_law", "A")]
|
||
N_UNITS = 3
|
||
SALT_FILE = KEYS / "SALT-sol.txt"
|
||
|
||
|
||
def _salt() -> str:
|
||
if not SALT_FILE.exists():
|
||
import secrets # noqa: PLC0415
|
||
SALT_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||
SALT_FILE.write_text(secrets.token_hex(16), encoding="utf-8")
|
||
return SALT_FILE.read_text(encoding="utf-8").strip()
|
||
|
||
|
||
def pick_units(a: str, b: str) -> list[dict]:
|
||
"""Единицы с НАИБОЛЬШИМ расхождением судей на этом контрасте — там арбитраж и нужен."""
|
||
us = BO.units()
|
||
scored = []
|
||
for u in us:
|
||
per = []
|
||
for judge in J.JUDGES:
|
||
bo = BO.margins_by_order(a, b, u["uid"], judge)
|
||
if len(bo) == 2:
|
||
per.append(st.mean([st.mean(bo[0]), st.mean(bo[1])]))
|
||
if len(per) == len(J.JUDGES):
|
||
scored.append((abs(per[0] - per[1]), u))
|
||
scored.sort(key=lambda x: (-x[0], x[1]["uid"]))
|
||
return [u for _g, u in scored[:N_UNITS]]
|
||
|
||
|
||
def _packet(src: str, v1: str, v2: str, tok: str) -> str:
|
||
return "\n".join([
|
||
f"ПАКЕТ {tok}", "",
|
||
J.RUBRIC.split("---", 1)[0].strip() if "---" in J.RUBRIC else J.RUBRIC.strip(), "",
|
||
"=" * 60, "ИСХОДНИК:", src, "",
|
||
"=" * 60, "ВАРИАНТ 1:", v1, "",
|
||
"=" * 60, "ВАРИАНТ 2:", v2, ""])
|
||
|
||
|
||
def emit() -> None:
|
||
SOL.mkdir(parents=True, exist_ok=True)
|
||
ANSW.mkdir(parents=True, exist_ok=True)
|
||
key: dict[str, dict] = {}
|
||
n = 0
|
||
for a, b in CONTRASTS:
|
||
for u in pick_units(a, b):
|
||
ta, tb = BO.text_of(a, u), BO.text_of(b, u)
|
||
if not ta.strip() or not tb.strip():
|
||
continue
|
||
for order in (0, 1):
|
||
tok = hashlib.sha1(f"{_salt()}|{a}|{b}|{u['uid']}|{order}".encode()).hexdigest()[:10] # noqa: S324
|
||
v1, v2 = (ta, tb) if order == 0 else (tb, ta)
|
||
(SOL / f"{tok}.txt").write_text(_packet(u["source"], v1, v2, tok), encoding="utf-8")
|
||
key[tok] = dict(contrast=f"{a}v{b}", first=a if order == 0 else b,
|
||
second=b if order == 0 else a, uid=u["uid"], order=order,
|
||
kind="боевой")
|
||
n += 1
|
||
# ДЕКОЙ: посаженная деградация против чистого текста. Сессия, не поймавшая декой, аннулируется.
|
||
u0 = BO.units()[0]
|
||
clean = BO.text3_of("A", u0)
|
||
planted, k = AJ._plant(clean) # noqa: SLF001
|
||
if k >= 2:
|
||
for order in (0, 1):
|
||
tok = hashlib.sha1(f"{_salt()}|decoy|{u0['uid']}|{order}".encode()).hexdigest()[:10] # noqa: S324
|
||
v1, v2 = (clean, planted) if order == 0 else (planted, clean)
|
||
(SOL / f"{tok}.txt").write_text(_packet(u0["source"], v1, v2, tok), encoding="utf-8")
|
||
key[tok] = dict(contrast="ДЕКОЙ", first="чистый" if order == 0 else "порченый",
|
||
second="порченый" if order == 0 else "чистый", uid=u0["uid"],
|
||
order=order, kind="контроль-декой", plants=k)
|
||
n += 1
|
||
(KEYS / "sol-KEY.json").write_text(json.dumps(key, ensure_ascii=False, indent=1),
|
||
encoding="utf-8")
|
||
(SOL / "ИНСТРУКЦИЯ.md").write_text(_instr(len(key)), encoding="utf-8")
|
||
print(f"пакетов {n} → {SOL} · ключ → {KEYS}/sol-KEY.json (владельцу и Sol НЕ давать)")
|
||
print(f"инструкция → {SOL}/ИНСТРУКЦИЯ.md")
|
||
|
||
|
||
def _instr(n: int) -> str:
|
||
return f"""# Пакеты для Sol — инструкция владельцу
|
||
|
||
Всего пакетов: **{n}**. Каждый — отдельный файл `<токен>.txt` в этом каталоге.
|
||
Порядок вставки НЕ важен; частичный возврат тоже годится — что вернётся, то и посчитается.
|
||
|
||
## Что делать
|
||
|
||
1. Открыть новый чат с Sol (high reasoning).
|
||
2. Вставить содержимое ОДНОГО файла целиком, без правок и без пояснений от себя.
|
||
3. Скопировать ответ Sol **дословно** в файл `answers/<тот же токен>.txt`.
|
||
4. Следующий пакет — **новым чатом**: иначе Sol узнаёт зеркальные пары и позиционного контроля
|
||
не остаётся (эта беда измерена на агенте-судье в эксп-21 §10).
|
||
|
||
## Формат ответа, который ждёт разбор
|
||
|
||
Sol просят вернуть ровно такие строки (это часть текста пакета, отдельно требовать не нужно):
|
||
|
||
```
|
||
В1-ВЕРНОСТЬ: <число>
|
||
В1-ТЕРМИН: <число>
|
||
В1-ЯЗЫК: <число>
|
||
В1-ФОРМА: <число>
|
||
В2-ВЕРНОСТЬ: <число>
|
||
В2-ТЕРМИН: <число>
|
||
В2-ЯЗЫК: <число>
|
||
В2-ФОРМА: <число>
|
||
```
|
||
|
||
Если Sol добавит рассуждение — не удалять: транскрипт персистится дословно, лишнее разбор
|
||
игнорирует.
|
||
|
||
## Чего делать НЕ надо
|
||
|
||
* не сообщать Sol, какая модель какой вариант писала (в пакетах этого нет, и это важно);
|
||
* не менять порядок вариантов внутри пакета;
|
||
* не сводить несколько пакетов в один чат.
|
||
|
||
## Что это даёт
|
||
|
||
Два платных судьи эксп-21 разошлись на 2.19 ошибки там, где собственный шум судьи ±0.15.
|
||
Направление вывода «второй проход покупает качество» держится, значимость — нет: её несёт ОДИН
|
||
судья, аварийная лёгкая замена. Голос Sol — третий независимый инструмент, и он решает, к кому
|
||
из двух ближе истина.
|
||
|
||
Среди пакетов есть **контрольные** (в них одна сторона заведомо испорчена). Если Sol их не
|
||
ловит — его голоса по боевым парам не читаются. Какие именно контрольные — не сказано намеренно.
|
||
"""
|
||
|
||
|
||
def ingest() -> None:
|
||
key = json.loads((KEYS / "sol-KEY.json").read_text(encoding="utf-8"))
|
||
got = {}
|
||
for f in sorted(ANSW.glob("*.txt")):
|
||
if f.stem not in key:
|
||
print(f" ⚠ токена {f.stem} нет в ключе — пропуск")
|
||
continue
|
||
p = J.parse(f.read_text(encoding="utf-8"))
|
||
if p["В1-ВЕРНОСТЬ"] is None:
|
||
print(f" ⚠ {f.stem}: ответ не разобран")
|
||
continue
|
||
e1 = sum(p[f"В1-{ax}"] or 0 for ax in J.AXES)
|
||
e2 = sum(p[f"В2-{ax}"] or 0 for ax in J.AXES)
|
||
got[f.stem] = dict(**key[f.stem], e1=e1, e2=e2)
|
||
(MONEY.OUT / "sol-votes.json").write_text(json.dumps(got, ensure_ascii=False, indent=1),
|
||
encoding="utf-8")
|
||
print(f"принято голосов {len(got)} из {len(key)} пакетов")
|
||
|
||
|
||
def score() -> None:
|
||
f = MONEY.OUT / "sol-votes.json"
|
||
if not f.exists() or not json.loads(f.read_text(encoding="utf-8")):
|
||
print("Sol не вернулся — голосов нет. Это НЕ блокер: отчёт сдаётся с явной строкой.")
|
||
return
|
||
votes = json.loads(f.read_text(encoding="utf-8"))
|
||
dec = [v for v in votes.values() if v["kind"] == "контроль-декой"]
|
||
if dec:
|
||
caught = sum(1 for v in dec
|
||
if (v["e1"] < v["e2"]) == (v["first"] == "чистый"))
|
||
print(f"ДЕКОЙ: поймано {caught}/{len(dec)}"
|
||
+ ("" if caught == len(dec) else " ⛔ сессия Sol аннулируется целиком"))
|
||
if caught != len(dec):
|
||
return
|
||
print(f"\n{'контраст':14s}{'ед.':>5s}{'Sol':>9s}{'luna':>9s}{'gemini':>9s}")
|
||
for a, b in CONTRASTS:
|
||
per: dict[str, list[float]] = {}
|
||
for v in votes.values():
|
||
if v["contrast"] != f"{a}v{b}":
|
||
continue
|
||
m = (v["e2"] - v["e1"]) if v["first"] == a else (v["e1"] - v["e2"])
|
||
per.setdefault(v["uid"], []).append(m)
|
||
vals = [st.mean(x) for x in per.values() if len(x) == 2]
|
||
if not vals:
|
||
continue
|
||
paid = {}
|
||
for judge in J.JUDGES:
|
||
pv = []
|
||
for u in BO.units():
|
||
if u["uid"] not in per:
|
||
continue
|
||
bo = BO.margins_by_order(a, b, u["uid"], judge)
|
||
if len(bo) == 2:
|
||
pv.append(st.mean([st.mean(bo[0]), st.mean(bo[1])]))
|
||
paid[judge] = st.mean(pv) if pv else float("nan")
|
||
print(f"{a + ' vs ' + b:14s}{len(vals):5d}{st.mean(vals):+9.2f}"
|
||
+ "".join(f"{paid[j]:+9.2f}" for j in J.JUDGES))
|
||
print("\n⚠ единицы Sol — подмножество (наибольшее расхождение судей), поэтому платные колонки"
|
||
"\n пере-считаны на ТЕХ ЖЕ единицах, а не взяты из свода по всем 15.")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
a = sys.argv[1:] or ["--emit"]
|
||
fn = {"--emit": emit, "--ingest": ingest, "--score": score}.get(a[0])
|
||
fn() if fn else print(__doc__)
|