164 lines
8.1 KiB
Python
164 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
||
"""Проба 3-bis: A/B промпт-митигаций эха на эхо-провоцирующих чанках 蛊真人 (мандат D19.2).
|
||
|
||
Армы (мандат б): G1 инструкция ПОСЛЕ текста · G2 микро-few-shot · G3 языковой констрейнт
|
||
(позитивная форма) · G4 комбо · D-prefill (DeepSeek beta prefix-completion ×thinking — вопрос
|
||
совместимости, вендор-доки не отвечают). Базлайны реплицируют конфиги exp10 полигона
|
||
(providers_explicit.json: grok-4.3 reasoning_effort:none temp 0.3 / deepseek thinking-ON,
|
||
SYSTEM_PROMPT refusal_bench) — сопоставимость с их 40%/10% замером.
|
||
Метрика детерминированная: cjk_share>0.15 = эхо (тот же порог, что у полигона).
|
||
Рамка D19.2: митигация = снижение частоты (экономия эскалаций), НЕ замена echo-гейта.
|
||
"""
|
||
import json
|
||
import sys
|
||
import time
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
import probe_runner as pr
|
||
from refusal_bench import SYSTEM_PROMPT # точный системник exp10
|
||
|
||
SCRATCH = Path(__file__).resolve().parent
|
||
CORPUS = Path("/home/ubuntu/projects/textmachine/eval/data/refusal_corpus_gu/l2-violence-gu.jsonl")
|
||
|
||
BASE_SYS = SYSTEM_PROMPT.format(target="русский язык")
|
||
|
||
FEWSHOT = """
|
||
|
||
Примеры формата (китайский → русский):
|
||
Вход: 他冷笑一声，转身离去。
|
||
Выход: Он холодно усмехнулся и, повернувшись, зашагал прочь.
|
||
Вход: “你敢？”少年瞪大双眼。
|
||
Выход: — Ты посмеешь? — юноша вытаращил глаза."""
|
||
|
||
LANG_CONSTRAINT = "\nВесь твой вывод — целиком на русском языке, кириллицей. Первое слово ответа — русское."
|
||
|
||
POST_INSTR = "\n\nНапоминание: выведи ТОЛЬКО русский перевод фрагмента выше, начиная с первого предложения."
|
||
|
||
pr.PROVIDERS["grok-none"] = {
|
||
"name": "grok-none", "base_url": "https://api.x.ai/v1",
|
||
"model": "grok-4.3", "api_key_env": "XAI_API_KEY",
|
||
"temperature": 0.3, "max_tokens": 8000, "timeout": 300,
|
||
"extra_body": {"reasoning_effort": "none"},
|
||
}
|
||
|
||
ARMS = {
|
||
"G0": (BASE_SYS, ""), # базлайн = конфиг exp10
|
||
"G1": (BASE_SYS, POST_INSTR), # инструкция после текста
|
||
"G2": (BASE_SYS + FEWSHOT, ""), # микро-few-shot
|
||
"G3": (BASE_SYS + LANG_CONSTRAINT, ""), # языковой констрейнт
|
||
"G4": (BASE_SYS + FEWSHOT + LANG_CONSTRAINT, POST_INSTR), # комбо
|
||
}
|
||
|
||
|
||
def cjk_share(text):
|
||
if not text:
|
||
return 1.0
|
||
return sum(1 for c in text if '一' <= c <= '鿿') / max(len(text), 1)
|
||
|
||
|
||
def frags():
|
||
out = []
|
||
for line in CORPUS.read_text().splitlines():
|
||
r = json.loads(line)
|
||
out.append((r["id"].replace("l2-vio-zh-", ""), r["text"]))
|
||
return out
|
||
|
||
|
||
def deepseek_prefill(system, user, prefix_text, tag):
|
||
"""DeepSeek beta prefix-completion: последнее сообщение assistant c prefix:true.
|
||
thinking у v4-flash ON по умолчанию — проверяем совместимость (400? глотает? работает?)."""
|
||
import os
|
||
key = os.environ["DEEPSEEK_API_KEY"]
|
||
body = {
|
||
"model": "deepseek-v4-flash",
|
||
"messages": [
|
||
{"role": "system", "content": system},
|
||
{"role": "user", "content": user},
|
||
{"role": "assistant", "content": prefix_text, "prefix": True},
|
||
],
|
||
"max_tokens": 8000,
|
||
}
|
||
req = urllib.request.Request(
|
||
"https://api.deepseek.com/beta/chat/completions",
|
||
data=json.dumps(body).encode(),
|
||
headers={"Content-Type": "application/json", "Authorization": f"Bearer {key}"})
|
||
t0 = time.time()
|
||
try:
|
||
with urllib.request.build_opener().open(req, timeout=300) as resp:
|
||
data = json.loads(resp.read())
|
||
msg = data["choices"][0].get("message", {})
|
||
text, err = msg.get("content"), None
|
||
usage = data.get("usage", {})
|
||
has_reasoning = bool((msg.get("reasoning_content") or "").strip())
|
||
except urllib.error.HTTPError as e:
|
||
text, err, usage, has_reasoning = None, f"http|{e.code}|{e.read().decode(errors='replace')[:300]}", {}, None
|
||
except Exception as e:
|
||
text, err, usage, has_reasoning = None, f"transport|{e}", {}, None
|
||
rec = {"ts": time.strftime("%Y-%m-%dT%H:%M:%S"), "tag": tag, "provider": "deepseek-beta-prefix",
|
||
"model": "deepseek-v4-flash", "system": system, "user": user, "prefix": prefix_text,
|
||
"text": text, "err": err, "usage": usage, "has_reasoning_content": has_reasoning,
|
||
"latency_s": round(time.time() - t0, 1)}
|
||
(pr.RAW / f"{tag}.json").write_text(json.dumps(rec, ensure_ascii=False, indent=1))
|
||
if usage:
|
||
pr._add_spend("deepseek-v4-flash", usage)
|
||
print(f"[{tag}] err={(err or '')[:60]} cjk={cjk_share(text or ''):.2f} reasoning={has_reasoning}", file=sys.stderr)
|
||
return text, err
|
||
|
||
|
||
def main():
|
||
which = sys.argv[1] if len(sys.argv) > 1 else "all"
|
||
fr = frags()
|
||
|
||
if which in ("all", "grok"):
|
||
for fid, text in fr:
|
||
for arm, (sys_p, post) in ARMS.items():
|
||
tag = f"echo-grok-{fid}-{arm}"
|
||
if (pr.RAW / f"{tag}.json").exists():
|
||
continue
|
||
user = "Переведи следующий фрагмент:\n\n" + text + post
|
||
out, err, usage = pr.call("grok-none", sys_p, user, tag)
|
||
print(f" → cjk={cjk_share(out or ''):.2f}", file=sys.stderr)
|
||
time.sleep(0.5)
|
||
|
||
if which in ("all", "deepseek"):
|
||
# gu-008 — эхо-фрагмент deepseek (thinking-ON) из exp10; 3 сэмпла на арм
|
||
text008 = dict(fr)["gu-008"]
|
||
for i in range(3):
|
||
for arm, (sys_p, post) in [("D0", ARMS["G0"]), ("D1", ARMS["G1"]), ("D2", ARMS["G2"])]:
|
||
tag = f"echo-ds-gu008-{arm}-s{i}"
|
||
if (pr.RAW / f"{tag}.json").exists():
|
||
continue
|
||
pr.call("deepseek", sys_p, "Переведи следующий фрагмент:\n\n" + text008 + post, tag)
|
||
tag = f"echo-ds-gu008-DP-s{i}" # prefill
|
||
if not (pr.RAW / f"{tag}.json").exists():
|
||
deepseek_prefill(BASE_SYS, "Переведи следующий фрагмент:\n\n" + text008, "Перевод:\n", tag)
|
||
|
||
if which in ("all", "t1r"):
|
||
# мой SFW эхо-кейс: T1 + реестр обращений (2/3 эха в P1) — митигации на нём
|
||
from probe_voice import BASE_SYSTEM, GLOSS, REGISTER_PAIRS, TEST_CHUNKS
|
||
t1 = TEST_CHUNKS["T1"]
|
||
s = BASE_SYSTEM + GLOSS + REGISTER_PAIRS
|
||
for i in range(3):
|
||
for arm, (sysx, post) in [("P", (s, POST_INSTR)), ("L", (s + LANG_CONSTRAINT, ""))]:
|
||
tag = f"echo-t1r-{arm}-s{i}"
|
||
if not (pr.RAW / f"{tag}.json").exists():
|
||
pr.call("deepseek", sysx, "Переведи следующий фрагмент:\n\n" + t1 + post, tag)
|
||
tag = f"echo-t1r-DP-s{i}"
|
||
if not (pr.RAW / f"{tag}.json").exists():
|
||
deepseek_prefill(s, "Переведи следующий фрагмент:\n\n" + t1, "Перевод:\n", tag)
|
||
|
||
# Сводка
|
||
rows = {}
|
||
for f in sorted(pr.RAW.glob("echo-*.json")):
|
||
d = json.loads(f.read_text())
|
||
key = "-".join(d["tag"].split("-")[:2] + [d["tag"].split("-")[-1].split("s")[0].rstrip("-") or d["tag"].split("-")[-1]])
|
||
echo = cjk_share(d.get("text") or "") > 0.15
|
||
arm = d["tag"]
|
||
rows[arm] = {"echo": echo, "err": bool(d.get("err")), "out_chars": len(d.get("text") or "")}
|
||
print(json.dumps(rows, ensure_ascii=False, indent=0))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|