278 lines
14 KiB
Python
278 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""Fidelity-судья хвоста P4 (петля D21 п.6) по нормам рига D30.10/D39.7.
|
||
|
||
Вопрос УЗКИЙ и он не «какой перевод лучше»: **стоит ли митигация (постинструкция / few-shot)
|
||
чего-нибудь по ВЕРНОСТИ**. Поэтому ось одна — верность; гладкость и красота из промпта судьи
|
||
вычеркнуты явно, иначе судья начнёт ранжировать стиль и вернёт ответ на другой вопрос.
|
||
|
||
Нормы рига, соблюдённые здесь:
|
||
· судья ЧУЖОГО семейства (grok-4.3 против переводов deepseek) — author ≠ reviewer (D13.3);
|
||
· reasoning ON (дефолт grok = low; явный `none` НЕ шлём — именно он делал судью слепым, D30.1);
|
||
· ПОЛНЫЕ окна: целиком исходник и целиком оба перевода, без обрезки (урок exp15 Q1b: обрезанное
|
||
окно родило вердикт-артефакт, который пришлось отзывать);
|
||
· ОБА порядка (A/B и B/A) — позиционный шум измеряется, а не предполагается;
|
||
· FLOOR-проход на ИДЕНТИЧНЫХ текстах — пол судейского шума; контраст читается только над ним;
|
||
· пер-голосовой персист в JSONL + резюм; жёсткий внутренний кап расхода;
|
||
· span-цитаты с ОБЕИХ сторон: дефект без цитаты вердиктом не считается.
|
||
|
||
Выбор единиц — ПРЕ-РЕГИСТРИРОВАННЫЙ и по порядку (chapter, chunk_idx) среди тех, где ВСЕ ТРИ арма
|
||
дали чистый выход. Никакого отбора по исходу.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
from dotenv import load_dotenv
|
||
from openai import OpenAI
|
||
|
||
REPO = Path("/home/ubuntu/projects/textmachine")
|
||
PL = Path.home() / "books" / "gu-zhenren" / "promptlang"
|
||
ARMS_JUDGED = ["a0", "a1", "a2"]
|
||
# Контраст a0↔a2 СНЯТ с судьи осознанно: few-shot отвергнут ДЕТЕРМИНИРОВАННО (эхо 1/12 +
|
||
# выход по-английски 1/12 + reasoning ×1.9), и решение по нему судья не двигает. Бюджет
|
||
# судьи идёт туда, где вопрос ещё открыт, — постинструкция.
|
||
CONTRASTS = [("a0-a1", "a1")]
|
||
OUT = PL / "judge"
|
||
OUT.mkdir(exist_ok=True)
|
||
load_dotenv(REPO / "eval" / ".env")
|
||
|
||
# grok-4.3: $1.25/$2.50 за 1M (models.yaml, prices_checked 2026-07-10); reasoning у xAI АДДИТИВЕН
|
||
# к completion, поэтому в цену выхода он входит и считается явно.
|
||
JUDGE = dict(model="grok-4.3", base="https://api.x.ai/v1", key_env="XAI_API_KEY",
|
||
price_in=1.25, price_out=2.50, max_tokens=8000, temperature=0.0)
|
||
|
||
SYS = (
|
||
"Ты — контролёр ВЕРНОСТИ художественного перевода с китайского на русский. Тебе дают КИТАЙСКИЙ "
|
||
"исходник и ДВА русских перевода одного и того же фрагмента: A и B.\n"
|
||
"Оценивай ТОЛЬКО верность: передан ли смысл исходника без ПРОПУСКОВ, ИСКАЖЕНИЙ, ОТСЕБЯТИНЫ и без "
|
||
"непереведённых кусков. Гладкость, красота, стиль, выбор синонимов и разбивка на абзацы НЕ "
|
||
"оцениваются вовсе — по этим осям любые различия игнорируй.\n"
|
||
"Дефект засчитывается ТОЛЬКО с дословными цитатами: что в исходнике и что (или ничего) в переводе.\n"
|
||
"Если по верности переводы равны — это нормальный и частый ответ: пиши \"tie\".\n"
|
||
"Ответь СТРОГИМ JSON одной строкой, без текста вокруг и без markdown:\n"
|
||
'{"winner":"A|B|tie","margin":"clear|slight",'
|
||
'"defects_A":[{"kind":"omission|distortion|addition|untranslated","zh":"<цитата исходника ≤12 иероглифов>","ru":"<цитата перевода ≤15 слов или пусто>"}],'
|
||
'"defects_B":[...],"reason":"<=25 слов"}'
|
||
)
|
||
|
||
|
||
def user(src: str, ta: str, tb: str) -> str:
|
||
return (f"=== КИТАЙСКИЙ ИСХОДНИК ===\n{src}\n\n=== ПЕРЕВОД A ===\n{ta}\n\n"
|
||
f"=== ПЕРЕВОД B ===\n{tb}\n\nВерни JSON-вердикт по ВЕРНОСТИ.")
|
||
|
||
|
||
class Ledger:
|
||
def __init__(self, path: Path, cap: float):
|
||
self.path, self.cap, self.total = path, cap, 0.0
|
||
if path.exists():
|
||
for ln in path.read_text().splitlines():
|
||
try:
|
||
self.total += json.loads(ln).get("cost_usd", 0.0)
|
||
except Exception:
|
||
pass
|
||
|
||
def add(self, rec: dict) -> None:
|
||
self.total += rec.get("cost_usd", 0.0)
|
||
with open(self.path, "a") as f:
|
||
f.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
||
|
||
|
||
_client = None
|
||
|
||
|
||
def client():
|
||
global _client
|
||
if _client is None:
|
||
key = os.environ.get(JUDGE["key_env"])
|
||
if not key:
|
||
sys.exit(f"нет ключа {JUDGE['key_env']}")
|
||
_client = OpenAI(api_key=key, base_url=JUDGE["base"], timeout=300)
|
||
return _client
|
||
|
||
|
||
def call(led: Ledger, sys_p: str, usr: str, tag: str) -> tuple[str, dict]:
|
||
if led.total >= led.cap:
|
||
sys.exit(f"СТОП: кап судьи ${led.cap} достигнут (потрачено ${led.total:.4f})")
|
||
for attempt, back in enumerate([5, 15, 40, None]):
|
||
try:
|
||
r = client().chat.completions.create(
|
||
model=JUDGE["model"],
|
||
messages=[{"role": "system", "content": sys_p}, {"role": "user", "content": usr}],
|
||
temperature=JUDGE["temperature"], max_tokens=JUDGE["max_tokens"])
|
||
ch = r.choices[0]
|
||
txt = (ch.message.content or "").strip()
|
||
u = r.usage
|
||
pt = getattr(u, "prompt_tokens", 0) or 0
|
||
ct = getattr(u, "completion_tokens", 0) or 0
|
||
det = getattr(u, "completion_tokens_details", None)
|
||
rt = (getattr(det, "reasoning_tokens", 0) if det else 0) or 0
|
||
# xAI: reasoning АДДИТИВЕН к completion. Если провайдер уже включил его в completion,
|
||
# двойного счёта не будет — берём max, а не сумму, и это осознанно консервативно вверх
|
||
# только на входе в кап, но в отчёт идёт и то и другое.
|
||
out_billed = ct if rt and rt <= ct else ct + rt
|
||
cost = pt * JUDGE["price_in"] / 1e6 + out_billed * JUDGE["price_out"] / 1e6
|
||
rec = dict(tag=tag, judge=JUDGE["model"], finish=ch.finish_reason, prompt_tok=pt,
|
||
completion_tok=ct, reasoning_tok=rt, out_billed=out_billed,
|
||
cost_usd=cost, empty=(not txt))
|
||
led.add(rec)
|
||
if ch.finish_reason != "stop":
|
||
print(f" ⚠ {tag}: finish={ch.finish_reason!r} (аномалия — в отчёт)")
|
||
return txt, rec
|
||
except Exception as e:
|
||
if back is None:
|
||
led.add(dict(tag=tag, error=str(e)[:300], cost_usd=0.0))
|
||
print(f" [FAIL {tag}] {str(e)[:180]}")
|
||
return "", {"error": str(e)[:300]}
|
||
print(f" [retry {tag}] {str(e)[:90]} — сон {back}с")
|
||
time.sleep(back)
|
||
return "", {}
|
||
|
||
|
||
def parse(txt: str):
|
||
if not txt:
|
||
return None
|
||
s = txt.strip().strip("`")
|
||
i, j = s.find("{"), s.rfind("}")
|
||
if i < 0 or j < 0:
|
||
return None
|
||
try:
|
||
return json.loads(s[i:j + 1])
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def load_units():
|
||
"""Единицы, где ВСЕ судимые армы дали чистый выход (не эхо, не пусто). Порядок — по имени
|
||
фрагмента, то есть по тексту книги. Никакого отбора по исходу."""
|
||
raw = PL / "armsraw"
|
||
src = {f["id"]: f["text"] for f in json.loads((raw / "fragments.json").read_text(encoding="utf-8"))}
|
||
texts, clean = {}, None
|
||
for arm in ARMS_JUDGED:
|
||
ok = set()
|
||
for f in sorted(raw.glob(f"{arm}-f*.json")):
|
||
j = json.loads(f.read_text(encoding="utf-8"))
|
||
if "error" in j:
|
||
continue
|
||
k = j["frag"]
|
||
j["src_text"] = src[k]
|
||
texts[(arm, k)] = j
|
||
# исключаем не только эхо, но и выход не в целевом письме: это уже пойман
|
||
# ДЕТЕРМИНИРОВАННЫМ прибором, судья тут ответил бы на другой вопрос
|
||
import unicodedata as _u
|
||
cyr = sum(1 for ch in j["content"] if _u.name(ch, "").startswith("CYRILLIC"))
|
||
lat = sum(1 for ch in j["content"] if _u.name(ch, "").startswith("LATIN"))
|
||
off_lang = (cyr + lat) > 0 and cyr / (cyr + lat) < 0.5
|
||
if not j["echo"] and not off_lang and j["content_chars"] > 0:
|
||
ok.add(k)
|
||
clean = ok if clean is None else (clean & ok)
|
||
return sorted(clean or []), texts
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--cap", type=float, required=True, help="жёсткий кап расхода судьи, $")
|
||
ap.add_argument("--units", type=int, default=0, help="сколько единиц судить (0 = план)")
|
||
ap.add_argument("--probe", action="store_true", help="одна живая калибровка цены")
|
||
ap.add_argument("--floor", action="store_true", help="floor-проход на идентичных текстах")
|
||
ap.add_argument("--aggregate", action="store_true", help="$0 пересчёт из JSONL")
|
||
a = ap.parse_args()
|
||
|
||
led = Ledger(OUT / "ledger.jsonl", a.cap)
|
||
votes_path = OUT / "votes.jsonl"
|
||
seen = {}
|
||
if votes_path.exists():
|
||
for ln in votes_path.read_text().splitlines():
|
||
try:
|
||
v = json.loads(ln)
|
||
seen[v["vote_id"]] = v
|
||
except Exception:
|
||
pass
|
||
|
||
units, texts = load_units()
|
||
print(f"чистых единиц во ВСЕХ трёх армах: {len(units)} → {units}")
|
||
|
||
if a.aggregate:
|
||
aggregate(seen)
|
||
return 0
|
||
|
||
def vote(vid, kind, unit, contrast, order, src, ta, tb, label_a, label_b):
|
||
if vid in seen:
|
||
return
|
||
txt, rec = call(led, SYS, user(src, ta, tb), vid)
|
||
v = parse(txt) or {}
|
||
row = dict(vote_id=vid, kind=kind, unit=str(unit), contrast=contrast,
|
||
order=order, label_A=label_a, label_B=label_b, winner=v.get("winner"),
|
||
margin=v.get("margin"), defects_A=v.get("defects_A"), defects_B=v.get("defects_B"),
|
||
reason=v.get("reason"), finish=rec.get("finish"), raw=txt[:1200],
|
||
cost_usd=rec.get("cost_usd"))
|
||
with open(votes_path, "a") as f:
|
||
f.write(json.dumps(row, ensure_ascii=False) + "\n")
|
||
seen[vid] = row
|
||
print(f" {vid}: winner={v.get('winner')} margin={v.get('margin')} "
|
||
f"defA={len(v.get('defects_A') or [])} defB={len(v.get('defects_B') or [])} "
|
||
f"${led.total:.4f}")
|
||
|
||
if a.probe:
|
||
u = units[0]
|
||
c0 = texts[("a0", u)]
|
||
vote(f"probe::{u}", "probe", u, "a0-a0", "AB",
|
||
c0["src_text"], c0["content"], c0["content"], "a0", "a0")
|
||
print(f"\nКАЛИБРОВКА: ${led.total:.5f} за вызов")
|
||
return 0
|
||
|
||
if a.floor:
|
||
for u in units[:1]:
|
||
c0 = texts[("a0", u)]
|
||
for order in ("AB", "BA"):
|
||
vote(f"floor::{u}::{order}", "floor", u, "a0-a0", order,
|
||
c0["src_text"], c0["content"], c0["content"], "a0", "a0")
|
||
return 0
|
||
|
||
n = a.units
|
||
if n <= 0:
|
||
print("план: --units N (N единиц × 2 контраста × 2 порядка = 4N вызовов)")
|
||
return 0
|
||
for u in units[:n]:
|
||
for contrast, arm in CONTRASTS:
|
||
base, mit = texts[("a0", u)], texts[(arm, u)]
|
||
for order in ("AB", "BA"):
|
||
ta, tb, la, lb = ((base["content"], mit["content"], "a0", arm) if order == "AB"
|
||
else (mit["content"], base["content"], arm, "a0"))
|
||
vote(f"{contrast}::{u}::{order}", "contrast", u, contrast, order,
|
||
base["src_text"], ta, tb, la, lb)
|
||
aggregate(seen)
|
||
print(f"\nПОТРАЧЕНО СУДЬЁЙ: ${led.total:.5f} из капа ${a.cap}")
|
||
return 0
|
||
|
||
|
||
def aggregate(seen: dict) -> None:
|
||
from collections import defaultdict
|
||
by = defaultdict(list)
|
||
for v in seen.values():
|
||
by[(v["kind"], v.get("contrast"))].append(v)
|
||
print(f"\n{'вид':<10} {'контраст':<8} {'голосов':>8} {'база':>6} {'митиг.':>7} {'ничья':>7} {'деф.база':>9} {'деф.митиг':>10}")
|
||
for (kind, contrast), vs in sorted(by.items()):
|
||
base = mit = tie = 0
|
||
db = dm = 0
|
||
for v in vs:
|
||
w = v.get("winner")
|
||
la, lb = v.get("label_A"), v.get("label_B")
|
||
win = la if w == "A" else lb if w == "B" else None
|
||
if win is None:
|
||
tie += 1
|
||
elif win == "a0":
|
||
base += 1
|
||
else:
|
||
mit += 1
|
||
db += len(v.get("defects_A") or []) if la == "a0" else len(v.get("defects_B") or [])
|
||
dm += len(v.get("defects_B") or []) if la == "a0" else len(v.get("defects_A") or [])
|
||
print(f"{kind:<10} {str(contrast):<8} {len(vs):>8} {base:>6} {mit:>7} {tie:>7} {db:>9} {dm:>10}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|