278 lines
15 KiB
Python
278 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""Полигон, пакет-7, фаза B: прогон армов промпта ТЕРМИНОЛОГА по пре-регистрации §A5.
|
||
|
||
Шесть армов (пре-регистрация A5.2, после старта НЕ правится):
|
||
P0b голый src, без контекстов — контроль «работают ли контексты вообще»
|
||
P0 P2, но жанровый словарь ПОДМЕНЁН случайным — контроль WMT23 «правильная ≈ случайная»
|
||
P1 src + KWIC + пары черновика — базовая линия
|
||
P2 P1 + жанровый словарь + Палладий + лемма — вклад ПАР-ДАННЫХ (гипотеза H1)
|
||
P3 P2 + рубрика §A2 в промпте + самооценка — вклад явной рубрики
|
||
P4 P2 + офлайн-снапшот справки (БКРС из Z4) — вклад веб-справки (ответ D39.42 п.2)
|
||
|
||
Дисциплина денег: цены из `backend/configs/models.yaml` (read-only), расход считается из
|
||
ФАКТИЧЕСКОГО `usage` каждого ответа, сырьё сохраняется целиком (правило 1 полигона). Батчинг по 10
|
||
термов на вызов — так работает ратифицированная роль (D39.42 «батчи, дешёвая модель»), и он
|
||
объявлен ДО трат; единица учёта остаётся «терм × арм».
|
||
|
||
Квирки соблюдены (00-provider-quirks): deepseek-v4-flash — thinking ВКЛЮЧЁН (эхо-мина при выкл.),
|
||
`max_tokens` ≥ 8000, temperature/top_p в thinking-режиме НЕ шлём (молча игнорируются).
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import random
|
||
import re
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
import yaml
|
||
from dotenv import load_dotenv
|
||
from openai import OpenAI
|
||
|
||
REPO = Path(__file__).resolve().parents[2]
|
||
load_dotenv(REPO / "eval" / ".env")
|
||
|
||
ARMS = ["P0b", "P0", "P1", "P2", "P3", "P4"]
|
||
|
||
# Фаза B′ (добавлено 26.07 ПОСЛЕ ревизии §B6; фаза B не пере-прогонялась, её сырьё нетронуто).
|
||
# P5 = P2 минус ОДНА строка правила транскрипции. Контрольного арма без этого правила в фазе B не
|
||
# было: правило лежит в BASE_TASK, то есть во всех шести армах сразу — и «эффект правила» измерить
|
||
# было нечем. Строка вырезается из готового BASE_TASK программно, чтобы остальной промпт совпадал
|
||
# с P2 побайтно, а не «на глаз».
|
||
ARMS_B2 = ["P5"]
|
||
|
||
BASE_TASK = """Ты ТЕРМИНОЛОГ художественного перевода. Тебе дают термины исходного текста книги.
|
||
Для КАЖДОГО термина выбери ОДНО консолидированное соответствие на русском языке, которое будет
|
||
использоваться во всей книге единообразно.
|
||
|
||
Требования к ответу:
|
||
- отвечай СТРОГО одним JSON-массивом, без пояснений вне JSON;
|
||
- элемент: {"src": "<исходный термин ровно как дан>", "dst": "<русское соответствие>", "type": "name|place|title|term", "why": "<до 20 слов>"}
|
||
- dst даёшь в ИСХОДНОЙ СЛОВАРНОЙ ФОРМЕ (лемма), без падежных окончаний по контексту;
|
||
- если термин — имя собственное, передавай его транскрипцией, а не переводом смысла;
|
||
- ничего не пропускай: элементов должно быть столько же, сколько терминов на входе."""
|
||
|
||
PN_RULE_LINE = "- если термин — имя собственное, передавай его транскрипцией, а не переводом смысла;\n"
|
||
BASE_TASK_NO_PN = BASE_TASK.replace(PN_RULE_LINE, "")
|
||
assert BASE_TASK_NO_PN != BASE_TASK, "строка правила транскрипции не найдена — контроль P5 был бы фикцией"
|
||
|
||
PALLADIUS_RULES = """Правила транскрипции для пары zh→ru (система Палладия, общепринятая норма):
|
||
- пиньинь -ng → русское -н; пиньинь -n → русское -нь (Shanghai → Шанхай, Shaolin → Шаолинь);
|
||
- zh → «чж» (не «дж»): Zhang → Чжан;
|
||
- hui → «хуэй», gui → «гуй», ü → «юй»;
|
||
- j/q/x + i → «цзи»/«ци»/«си»."""
|
||
|
||
LEMMA_RULE = """Русский — язык с богатой морфологией. dst обязан быть склоняемым и морфологически
|
||
пригодным: у него должен быть устойчивый род и нормальная парадигма. Неизменяемую кальку выбирай
|
||
только если склоняемого варианта не существует."""
|
||
|
||
RUBRIC = """Оценивай свой выбор по критериям и коротко отчитайся по ним в поле "why":
|
||
K1 верность концепту (то ли это понятие) · K2 жанровая конвенция · K3 транскрипционная норма ·
|
||
K5 морфологическая пригодность (склоняемость, род) · K6 благозвучие · K9 уместность регистра
|
||
(академический термин не подставлять в жанровый текст). Порог: все критерии должны быть не ниже
|
||
«приемлемо»; при конфликте K1 важнее K6."""
|
||
|
||
|
||
def load_prices(models_yaml: Path) -> dict:
|
||
doc = yaml.safe_load(models_yaml.read_text(encoding="utf-8")) or {}
|
||
out = {}
|
||
for name, m in (doc.get("models") or {}).items():
|
||
p = m.get("price") or {}
|
||
out[name] = (float(p.get("input_per_m", 0)), float(p.get("output_per_m", 0)),
|
||
float(p.get("cached_per_m", 0)))
|
||
return out
|
||
|
||
|
||
def load_genre_glossary(path: Path) -> list[tuple[str, str]]:
|
||
rows = []
|
||
if not path.exists():
|
||
return rows
|
||
for line in path.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
f = line.split("\t")
|
||
if len(f) >= 2:
|
||
rows.append((f[0], f[1]))
|
||
return rows
|
||
|
||
|
||
def scramble(rows: list[tuple[str, str]], seed: int) -> list[tuple[str, str]]:
|
||
"""Случайный глоссарий: те же src, но dst переставлены. Ровно контроль WMT23."""
|
||
rnd = random.Random(seed)
|
||
dst = [d for _, d in rows]
|
||
rnd.shuffle(dst)
|
||
return [(s, d) for (s, _), d in zip(rows, dst)]
|
||
|
||
|
||
def fmt_glossary(rows: list[tuple[str, str]]) -> str:
|
||
if not rows:
|
||
return ""
|
||
body = "\n".join(f"{s}\t{d}" for s, d in rows)
|
||
return f"Жанровый словарь пары (устоявшиеся соответствия жанра):\n{body}"
|
||
|
||
|
||
def like_p2(arm: str) -> bool:
|
||
"""P5 — это P2 во всём, кроме вырезанной строки правила транскрипции."""
|
||
return arm in ("P2", "P5")
|
||
|
||
|
||
def fmt_item(it: dict, arm: str, ref: dict) -> str:
|
||
parts = [f'ТЕРМИН: {it["src"]} (встречается в книге {it["occurrences"]} раз)']
|
||
if arm == "P0b":
|
||
return parts[0]
|
||
for k in it["kwic"][:6]:
|
||
parts.append(f' контекст: …{k["win"]}…')
|
||
for p in it["pairs"][:3]:
|
||
parts.append(f' черновик перевода этого места: …{p["draft_win"]}…')
|
||
if arm == "P4":
|
||
e = ref.get(it["src"])
|
||
if e:
|
||
parts.append(f' словарная справка (БКРС): {e[:400]}')
|
||
return "\n".join(parts)
|
||
|
||
|
||
def build_prompt(items: list[dict], arm: str, gloss: list[tuple[str, str]],
|
||
gloss_random: list[tuple[str, str]], ref: dict) -> str:
|
||
blocks = [BASE_TASK_NO_PN if arm == "P5" else BASE_TASK]
|
||
if arm in ("P2", "P3", "P4", "P5"):
|
||
g = fmt_glossary(gloss)
|
||
if g:
|
||
blocks.append(g)
|
||
blocks.append(PALLADIUS_RULES)
|
||
blocks.append(LEMMA_RULE)
|
||
if arm == "P0":
|
||
g = fmt_glossary(gloss_random)
|
||
if g:
|
||
blocks.append(g)
|
||
blocks.append(PALLADIUS_RULES)
|
||
blocks.append(LEMMA_RULE)
|
||
if arm == "P3":
|
||
blocks.append(RUBRIC)
|
||
if arm == "P4":
|
||
blocks.append("К части терминов приложена СЛОВАРНАЯ СПРАВКА из китайско-русского словаря. "
|
||
"Она описывает общеязыковое значение; жанровое употребление может отличаться — "
|
||
"используй её как улику, а не как приказ.")
|
||
blocks.append("ТЕРМИНЫ:\n\n" + "\n\n".join(fmt_item(i, arm, ref) for i in items))
|
||
return "\n\n".join(blocks)
|
||
|
||
|
||
def parse_json_array(text: str) -> list[dict]:
|
||
m = re.search(r"\[.*\]", text, re.S)
|
||
if not m:
|
||
return []
|
||
try:
|
||
v = json.loads(m.group(0))
|
||
return v if isinstance(v, list) else []
|
||
except json.JSONDecodeError:
|
||
return []
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--termset", required=True, type=Path)
|
||
ap.add_argument("--out", required=True, type=Path)
|
||
ap.add_argument("--raw-dir", required=True, type=Path)
|
||
ap.add_argument("--model", default="deepseek-v4-flash")
|
||
ap.add_argument("--base-url", default="https://api.deepseek.com/v1")
|
||
ap.add_argument("--key-env", default="DEEPSEEK_API_KEY")
|
||
ap.add_argument("--genre-glossary", type=Path,
|
||
default=REPO / "backend/configs/langpacks/zh-ru/genre-glossary.txt")
|
||
ap.add_argument("--bkrs", type=Path, help="JSON справок БКРС для арма P4")
|
||
ap.add_argument("--batch", type=int, default=10)
|
||
ap.add_argument("--max-tokens", type=int, default=8000)
|
||
ap.add_argument("--arms", default=",".join(ARMS))
|
||
ap.add_argument("--dry-run", action="store_true", help="сметa без единого вызова")
|
||
a = ap.parse_args()
|
||
|
||
items = json.loads(a.termset.read_text(encoding="utf-8"))
|
||
gloss = load_genre_glossary(a.genre_glossary)
|
||
gloss_rand = scramble(gloss, seed=20260726)
|
||
ref = {}
|
||
if a.bkrs and a.bkrs.exists():
|
||
for r in json.loads(a.bkrs.read_text(encoding="utf-8")):
|
||
if r.get("found"):
|
||
ref[r["src"]] = re.sub(r"^-->\s*", "", r["entry"]).strip()
|
||
|
||
prices = load_prices(REPO / "backend/configs/models.yaml")
|
||
pin, pout, _ = prices.get(a.model, (0.0, 0.0, 0.0))
|
||
arms = [x for x in a.arms.split(",") if x]
|
||
|
||
# батчи формируются ПО ВЫБОРКЕ, чтобы книга не смешивалась с книгой
|
||
batches = []
|
||
for sample in sorted({i["sample"] for i in items}):
|
||
xs = [i for i in items if i["sample"] == sample]
|
||
for k in range(0, len(xs), a.batch):
|
||
batches.append((sample, xs[k:k + a.batch]))
|
||
|
||
plan = [(arm, s, b) for arm in arms for (s, b) in batches]
|
||
est_in = sum(len(build_prompt(b, arm, gloss, gloss_rand, ref)) for arm, _, b in plan) / 3.2
|
||
est_out = len(plan) * 2500
|
||
print(f"ПЛАН: армов {len(arms)} × батчей {len(batches)} = {len(plan)} вызовов; "
|
||
f"термов×армов = {len(items) * len(arms)}")
|
||
print(f"СМЕТА ДО ВЫЗОВОВ: вход ≈{est_in/1000:.1f}k ток → ${est_in/1e6*pin:.4f}; "
|
||
f"выход ≈{est_out/1000:.1f}k ток → ${est_out/1e6*pout:.4f}; "
|
||
f"ИТОГО ≈ ${est_in/1e6*pin + est_out/1e6*pout:.4f}")
|
||
if a.dry_run:
|
||
return 0
|
||
|
||
key = os.environ.get(a.key_env)
|
||
if not key:
|
||
print(f"нет ключа {a.key_env}", file=sys.stderr)
|
||
return 2
|
||
client = OpenAI(api_key=key, base_url=a.base_url)
|
||
a.raw_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
results, ledger = [], []
|
||
for n, (arm, sample, batch) in enumerate(plan, 1):
|
||
prompt = build_prompt(batch, arm, gloss, gloss_rand, ref)
|
||
t0 = time.time()
|
||
try:
|
||
resp = client.chat.completions.create(
|
||
model=a.model,
|
||
messages=[{"role": "user", "content": prompt}],
|
||
max_tokens=a.max_tokens, # thinking ⊆ completion, floor 8000 по квиркам
|
||
)
|
||
except Exception as e:
|
||
print(f"[{n}/{len(plan)}] {arm}/{sample} ОШИБКА: {e}", file=sys.stderr)
|
||
ledger.append({"arm": arm, "sample": sample, "error": str(e)})
|
||
continue
|
||
dt = time.time() - t0
|
||
txt = resp.choices[0].message.content or ""
|
||
u = resp.usage
|
||
cin = getattr(u, "prompt_tokens", 0)
|
||
cout = getattr(u, "completion_tokens", 0)
|
||
cached = getattr(getattr(u, "prompt_tokens_details", None), "cached_tokens", 0) or 0
|
||
usd = (cin - cached) / 1e6 * pin + cached / 1e6 * prices.get(a.model, (0, 0, 0))[2] + cout / 1e6 * pout
|
||
(a.raw_dir / f"{arm}_{sample}_{n:03d}.json").write_text(
|
||
json.dumps({"arm": arm, "sample": sample, "prompt": prompt, "response": txt,
|
||
"finish": resp.choices[0].finish_reason,
|
||
"usage": {"in": cin, "out": cout, "cached": cached}},
|
||
ensure_ascii=False, indent=1), encoding="utf-8")
|
||
parsed = parse_json_array(txt)
|
||
by_src = {str(x.get("src", "")): x for x in parsed if isinstance(x, dict)}
|
||
for it in batch:
|
||
got = by_src.get(it["src"])
|
||
results.append({"id": it["id"], "sample": sample, "src": it["src"], "arm": arm,
|
||
"dst": (got or {}).get("dst"), "type": (got or {}).get("type"),
|
||
"why": (got or {}).get("why"), "gold": it.get("gold")})
|
||
ledger.append({"arm": arm, "sample": sample, "in": cin, "out": cout, "cached": cached,
|
||
"usd": usd, "sec": round(dt, 1), "finish": resp.choices[0].finish_reason,
|
||
"parsed": len(parsed), "expected": len(batch)})
|
||
print(f"[{n}/{len(plan)}] {arm}/{sample} in={cin} out={cout} ${usd:.5f} "
|
||
f"{dt:.0f}s parsed={len(parsed)}/{len(batch)} finish={resp.choices[0].finish_reason}")
|
||
|
||
a.out.write_text(json.dumps({"results": results, "ledger": ledger},
|
||
ensure_ascii=False, indent=1), encoding="utf-8")
|
||
total = sum(x.get("usd", 0) for x in ledger)
|
||
miss = sum(1 for r in results if not r["dst"])
|
||
print(f"\nИТОГО ПОТРАЧЕНО: ${total:.5f} за {len(ledger)} вызовов; "
|
||
f"консолидаций {len(results)}, из них пустых {miss}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|