230 lines
14 KiB
Python
230 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""Ф0.2 — ЗАМЕР КОНТАМИНАЦИИ имеющихся книг вместо добычи новой.
|
||
|
||
Постановка. Промт заказывает «невиданный срез» и оговаривает главное: **контаминация НЕ по дате**,
|
||
её надо мерить пробами на самой модели. Я сперва прочитал это как «нужна новая книга» и сделал
|
||
блокером то, что блокером не является: материал в `~/books` есть, и все экспы 18–20 гнались на нём
|
||
(промт пробы 18, арм AM: «найди на стенде 10 глав пассаж…»). Правильная задача — не притащить
|
||
шестую книгу, а УЗНАТЬ ЧИСЛОМ, какая из пяти имеющихся насколько сидит в претрейне, и разложить
|
||
работу по ним осознанно.
|
||
|
||
⚠⚠ ДВЕ РЕДАКЦИИ ЭТОЙ ПРОБЫ БЫЛИ СНЯТЫ ИСПОЛНЕНИЕМ, и обе — по одной причине: они спрашивали
|
||
ВОСПРОИЗВЕДЕНИЕ.
|
||
Редакция 1 звала `deepseek-v4-flash` с бюджетом 16000 токенов. Модель выжигала бюджет на
|
||
размышление и возвращала 0–46 знаков (стена quirks §10). Нулевое совпадение читалось как
|
||
«книга не заучена». Поймано осмотром сырья: $0.003–0.005 за вызов при пустом ответе.
|
||
Редакция 2 звала `gpt-5.6-luna` без размышления и с коротким выходом — стена ушла, но
|
||
ПОЛОЖИТЕЛЬНЫЙ КОНТРОЛЬ 金瓶梅 (минский канон, public domain, заведомо в претрейне) дал
|
||
ответ «НЕЗНАКОМ» и нулевое совпадение. Значит дело не в книгах: современные модели не
|
||
воспроизводят текст дословно по запросу вообще, и проба на продолжение непригодна КАК КЛАСС.
|
||
⇒ Метод сменён на тот, который спрашивает ЗНАНИЕ, а не воспроизведение. Оба вопроса ниже
|
||
безобидны с точки зрения выдачи текста и при этом отвечаются только тем, кто книгу видел.
|
||
|
||
Как меряется сейчас. Две независимые пробы на одном и том же отрывке:
|
||
УЗНАВАНИЕ — «что это за произведение и кто его автор». Верное название = текст (или подробные
|
||
описания текста) были в обучающих данных. Ловит знание о книге вообще.
|
||
КЛОУЗ ПО ИМЕНИ — из отрывка вырезано имя собственное, модель называет пропущенное. Заполнить
|
||
можно, только зная книгу; при этом от модели требуется ОДНО СЛОВО, а не проза. Ловит более
|
||
тонкое знание, чем узнавание: имя второстепенного персонажа помнит лишь тот, кто видел текст.
|
||
|
||
Почему различение важно для продукта. Завышение качества перевода даёт не «слышал о книге», а
|
||
знание её конкретики: имён, реалий, устоявшихся переводов. Именно это и меряет клоуз.
|
||
|
||
ПОЛОЖИТЕЛЬНЫЙ КОНТРОЛЬ — `jinpingmei`. Нулевой результат по любой книге сам по себе
|
||
неинтерпретируем: «модель текст не помнит» и «проба не работает» дают одинаковую картину. Если
|
||
узнавание не сработает и на нём — сломана проба, а не книги. Роль контроля названа заранее, а не
|
||
выбрана задним числом по удобству результата. (Та же логика, что декой у QE-замера и арм `xhigh`
|
||
у вахты эффорта; в этой пробе контроль уже дважды сработал по назначению — см. баннер выше.)
|
||
|
||
⚠ Отрывки НЕ попадают ни в репозиторий, ни в отчёт: печатаются только вердикты и метрики.
|
||
|
||
Запуск: eval/.venv/bin/python eval/role_topology/contamination.py --dry # план, $0
|
||
eval/.venv/bin/python eval/role_topology/contamination.py
|
||
"""
|
||
from __future__ import annotations
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import time
|
||
from collections import Counter
|
||
from pathlib import Path
|
||
|
||
REPO = Path("/home/ubuntu/projects/textmachine")
|
||
from dotenv import load_dotenv # noqa: E402
|
||
from openai import OpenAI # noqa: E402
|
||
|
||
load_dotenv(REPO / "eval" / ".env")
|
||
OUT = Path.home() / "books" / "role-topology"
|
||
OUT.mkdir(parents=True, exist_ok=True)
|
||
|
||
MODEL = "gpt-5.6-luna" # без стены размышления, дешёвый, и сам кандидат Ф2
|
||
PRICE_IN, PRICE_CACHED, PRICE_OUT = 0.20, 0.02, 1.20
|
||
MAX_OUT = 200
|
||
CEILING_USD = 0.10
|
||
N_PASSAGES = 5
|
||
PRIME_CHARS = 700
|
||
|
||
BOOKS: dict[str, dict] = {
|
||
"gu-zhenren": dict(path=Path.home() / "books" / "gu-zhenren" / "guzhenren-utf8.txt",
|
||
lang="zh", titles=["蛊真人", "гу чжэнь", "reverend insanity", "гу чжэньжэнь"],
|
||
note="вебновелла 2012–2018, есть en-фанперевод"),
|
||
"jinpingmei": dict(path=Path.home() / "books" / "jinpingmei" / "jinpingmei-ctext-zhs.txt",
|
||
lang="zh", titles=["金瓶梅", "цзинь пин мэй", "plum in the golden"],
|
||
note="ПОЛОЖИТЕЛЬНЫЙ КОНТРОЛЬ: минский канон, public domain"),
|
||
"isekai-ja": dict(path=Path.home() / "books" / "isekai_majutsushi_jp.txt",
|
||
lang="ja", titles=["異世界魔術師", "isekai majutsushi"],
|
||
note="вебновелла с syosetu, R18-раздел"),
|
||
"fifty-shades-en": dict(path=Path.home() / "books" / "fifty_shades_1_en.txt",
|
||
lang="en", titles=["fifty shades", "пятьдесят оттенков"],
|
||
note="издано 2011, бестселлер"),
|
||
"kristoff-en": dict(path=Path.home() / "books" /
|
||
"Kristoff_Jay_-_Empire_of_the_Vampire_3_-_Empire_of_the_Dawn.epub",
|
||
lang="en", titles=["empire of the vampire", "empire of the dawn",
|
||
"kristoff"],
|
||
note="роман 2026, ru-перевод не издан"),
|
||
}
|
||
|
||
|
||
def read_book(path: Path) -> str:
|
||
"""Текст книги. epub разбирается тем же способом, что `crosslang_bank.epub_text`."""
|
||
if path.suffix != ".epub":
|
||
return path.read_text(encoding="utf-8", errors="ignore")
|
||
import zipfile # noqa: PLC0415
|
||
out = []
|
||
with zipfile.ZipFile(path) as z:
|
||
for n in sorted(z.namelist()):
|
||
if n.endswith((".xhtml", ".html")):
|
||
html = z.read(n).decode("utf-8", "ignore")
|
||
txt = re.sub(r"<[^>]+>", " ", html)
|
||
txt = re.sub(r"&[a-z]+;", " ", txt)
|
||
out.append(re.sub(r"\s+", " ", txt))
|
||
return "\n".join(out)
|
||
|
||
|
||
def frequent_names(text: str, lang: str, top: int = 40) -> list[str]:
|
||
"""Частые имена собственные книги — кандидаты на маскирование в клоузе.
|
||
|
||
Для латиницы/кириллицы берутся слова с прописной вне начала предложения; для CJK — частые
|
||
биграммы/триграммы, не являющиеся служебными. Список нужен только чтобы ВЫБРАТЬ, что маскировать,
|
||
поэтому грубость метода допустима и объявлена.
|
||
"""
|
||
if lang in ("en",):
|
||
cand = Counter(m.group(1) for m in re.finditer(r"(?<![.!?]\s)(?<!^)\b([A-Z][a-z]{3,})\b",
|
||
text, re.M))
|
||
stop = {"The", "And", "But", "That", "This", "What", "When", "Then", "There", "With"}
|
||
return [w for w, _ in cand.most_common(top * 3) if w not in stop][:top]
|
||
# CJK: частые 2–3-знаковые последовательности хань/каны. Грубо, но для выбора маски хватает.
|
||
cand = Counter()
|
||
for n in (3, 2):
|
||
for m in re.finditer(r"[㐀-鿿゠-ヿ]{%d}" % n, text):
|
||
cand[m.group(0)] += 1
|
||
return [w for w, c in cand.most_common(top * 2) if c > 20][:top]
|
||
|
||
|
||
def passages(text: str, names: list[str], n: int) -> list[tuple[str, str]]:
|
||
"""(отрывок с маской, замаскированное имя). Отрывки из середины, точки детерминированы."""
|
||
body = text[len(text) // 10: -len(text) // 10]
|
||
out: list[tuple[str, str]] = []
|
||
for i in range(n * 6):
|
||
if len(out) >= n:
|
||
break
|
||
pos = len(body) * (i + 1) // (n * 6 + 1)
|
||
chunk = body[pos:pos + PRIME_CHARS]
|
||
if len(chunk) < PRIME_CHARS:
|
||
continue
|
||
hit = next((w for w in names if w in chunk), None)
|
||
if not hit:
|
||
continue
|
||
out.append((chunk.replace(hit, "[???]"), hit))
|
||
return out
|
||
|
||
|
||
def ask(cl, prime: str, tag: str) -> dict:
|
||
"""Один вызов: узнавание и клоуз задаются ВМЕСТЕ, чтобы не платить дважды за один отрывок."""
|
||
f = OUT / f"cont-{tag}.json"
|
||
if f.exists():
|
||
return json.loads(f.read_text(encoding="utf-8"))
|
||
msgs = [
|
||
{"role": "system", "content":
|
||
"Тебе дают отрывок художественного текста, в котором одно имя собственное заменено на "
|
||
"[???]. Ответь РОВНО двумя строками и ничем больше:\n"
|
||
"ПРОИЗВЕДЕНИЕ: <название и автор, либо НЕ ЗНАЮ>\n"
|
||
"ИМЯ: <что стоит вместо [???], либо НЕ ЗНАЮ>\n"
|
||
"Не пересказывай отрывок и не продолжай его."},
|
||
{"role": "user", "content": prime},
|
||
]
|
||
r = cl.chat.completions.create(model=MODEL, messages=msgs, max_completion_tokens=MAX_OUT,
|
||
reasoning_effort="none")
|
||
u = r.usage
|
||
cached = getattr(getattr(u, "prompt_tokens_details", None), "cached_tokens", 0) or 0
|
||
rec = dict(tag=tag, finish=r.choices[0].finish_reason,
|
||
cost_usd=round((u.prompt_tokens - cached) / 1e6 * PRICE_IN
|
||
+ cached / 1e6 * PRICE_CACHED
|
||
+ u.completion_tokens / 1e6 * PRICE_OUT, 6),
|
||
answer=r.choices[0].message.content or "")
|
||
f.write_text(json.dumps(rec, ensure_ascii=False), encoding="utf-8")
|
||
return rec
|
||
|
||
|
||
def parse(answer: str) -> tuple[str, str]:
|
||
work = name = ""
|
||
for line in answer.splitlines():
|
||
if line.upper().startswith("ПРОИЗВЕДЕНИЕ"):
|
||
work = line.split(":", 1)[-1].strip()
|
||
elif line.upper().startswith("ИМЯ"):
|
||
name = line.split(":", 1)[-1].strip()
|
||
return work, name
|
||
|
||
|
||
def main() -> None:
|
||
dry = "--dry" in sys.argv
|
||
cl = None if dry else OpenAI(api_key=os.environ["OPENAI_API_KEY"],
|
||
base_url="https://api.openai.com/v1", timeout=300)
|
||
spent = 0.0
|
||
summary: list[dict] = []
|
||
for book, meta in BOOKS.items():
|
||
if not meta["path"].exists():
|
||
print(f"{book}: файла нет — пропуск")
|
||
continue
|
||
text = read_book(meta["path"])
|
||
names = frequent_names(text, meta["lang"])
|
||
ps = passages(text, names, N_PASSAGES)
|
||
print(f"\n{book} ({meta['lang']}, {len(text):,} знаков) — отрывков {len(ps)} · {meta['note']}")
|
||
if dry:
|
||
continue
|
||
rec_ok = cloze_ok = 0
|
||
for i, (prime, hidden) in enumerate(ps):
|
||
if spent > CEILING_USD:
|
||
print("⛔ потолок исчерпан — стоп")
|
||
break
|
||
rec = ask(cl, prime, f"{book}-{i}")
|
||
spent += rec["cost_usd"]
|
||
work, name = parse(rec["answer"])
|
||
r_hit = any(t in work.lower() for t in meta["titles"])
|
||
c_hit = bool(name) and (hidden in name or name in hidden) and "НЕ ЗНАЮ" not in name
|
||
rec_ok += r_hit
|
||
cloze_ok += c_hit
|
||
print(f" отрывок {i}: узнавание {'ДА ' if r_hit else 'нет'} · "
|
||
f"клоуз {'ДА ' if c_hit else 'нет'} · ответ о книге: {work[:60] or '—'}")
|
||
time.sleep(0.2)
|
||
summary.append(dict(book=book, n=len(ps), recognized=rec_ok, cloze=cloze_ok))
|
||
|
||
if dry:
|
||
return
|
||
print(f"\n{'книга':18s}{'узнано':>9s}{'клоуз':>8s}{'вердикт':>22s}")
|
||
print("-" * 58)
|
||
for s in summary:
|
||
n = max(1, s["n"])
|
||
# Порог объявлен заранее: узнавание ≥3/5 = книга модели ИЗВЕСТНА; клоуз ≥2/5 = известна
|
||
# КОНКРЕТИКА, то есть текст (а не только описания) был в обучающих данных.
|
||
v = ("знает конкретику" if s["cloze"] >= 2 else
|
||
("знает о книге" if s["recognized"] >= 3 else "следов нет"))
|
||
print(f"{s['book']:18s}{s['recognized']:>4d}/{n:<4d}{s['cloze']:>4d}/{n:<3d}{v:>22s}")
|
||
print(f"\nпотрачено ${spent:.6f}")
|
||
(OUT / "contamination.json").write_text(json.dumps(summary, ensure_ascii=False, indent=1),
|
||
encoding="utf-8")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|