EvalHarness/evalharness/fingerprint/probe_snr.py

#!/usr/bin/env python3
"""文本层(I/K/C/S) + ADV + V 探针冗余分析 —— 与 D 层 cell 剪枝同方法论。纯离线，零 API。

证据链：
  A. 信息含量：9 模型家族命中向量（own=自证 / foreign=污染证据 / none）+ 答案区分度（token Jaccard）
  B. drop-one 双视图重放：
     信号视图 = identity/meta/refuse/length/lexicon 五个打分函数逐一重放
     判决视图 = build_report 全量重放（verify 口径，attribution=None 与真实运行一致）→ verdict/score
  C. 配对结构：i 层 pair(direct/jailbreak/fill) 按"成对剪"评估（保中英一致性信号）
  D. 层级保底：K 截止探针 ≥2（唯一性检查才有效）/ K 元认知审计 ≥1 / C 多级梯度 / S 两种长度控制各 ≥1
  E. ADV(3模型注入态) / V(2模型) 区分度矩阵
产出：/tmp/bfd/probe_snr_report.txt + /tmp/bfd/probe_snr.json
"""
import itertools
import json
import re
import sys
from collections import defaultdict

from pathlib import Path  # noqa: E402

sys.path.insert(0, str(Path(__file__).resolve().parents[2]))  # 仓库根/site-packages, 使 evalharness 包可导入
from evalharness.fingerprint.attribution import _lexicon_scores, _normalize, family_attribution
from evalharness.fingerprint.engine import (build_d_normalized, compare_cells,  # noqa: E402
                    distributions_by_cell, load_reference, split_half_jsd)
from evalharness.fingerprint.scorer import (_families_in_text, build_report, identity_signal,  # noqa: E402
                    length_compliance, load_aliases, meta_signal,
                    refuse_gradient_pattern, requested_family)

BFD = "/tmp/bfd"
R = str(Path(__file__).resolve().parent / 'references')

MODELS = [
    ("deepseek_v4_flash", "deepseek_v4_flash_fusion_reference.json", "DeepSeek/DeepSeek-V4-Flash"),
    ("deepseek_v4_flash_0731", "deepseek_v4_flash_0731_fusion_reference.json", "DeepSeek/DeepSeek-V4-Flash-0731"),
    ("deepseek_v4_pro", "deepseek_v4_pro_fusion_reference.json", "DeepSeek/DeepSeek-V4-Pro"),
    ("glm_51", "glm_51_fusion_reference.json", "ZhipuAi/GLM-5.1"),
    ("glm_52", "glm52_vectron_fusion_reference.json", "ZhipuAi/GLM-5.2"),
    ("glm_53", "glm53_fusion_reference.json", "ZhipuAi/GLM-5.3"),
    ("kimi_k2_6", "kimi_k2_6_fusion_reference.json", "MoonshotAi/Kimi-K2.6"),
    ("kimi_k2_7code", "kimi_k2_7code_fusion_reference.json", "MoonshotAi/Kimi-K2.7-Code"),
    ("kimi_k3", "kimi_k3_fusion_reference.json", "MoonshotAi/Kimi-K3"),
]
DIRS = [m[0] for m in MODELS]

aliases = load_aliases(None)
LINES = []


def log(s=""):
    print(s)
    LINES.append(s)


def toks(text):
    return set(re.findall(r"\w+", (text or "").lower()))


def jaccard(a, b):
    return len(a & b) / len(a | b) if (a or b) else 1.0


# ---------- 载入 + 重放地基 ----------
M = {}
for d, rf, mid in MODELS:
    recs = [json.loads(l) for l in open(f"{BFD}/{d}/raw_answers.jsonl")]
    verify = json.load(open(f"{BFD}/{d}/verify.json"))
    ref = load_reference(f"{R}/{rf}")
    dn = build_d_normalized(recs)
    dist = distributions_by_cell(dn)
    entries, mean_jsd = compare_cells(dist, ref["cells"])
    sh = split_half_jsd(dn)
    outliers = [e for e in entries
                if e["jsd"] > 0.5 and min(e["valid_a"], e["valid_b"]) >= 15]
    if mean_jsd is not None:
        ratio = mean_jsd / max(sh or 0.02, 0.02)
        s_val = 1.0 if ratio < 2 else (0.0 if ratio > 8 else 1.0 - (ratio - 2) / 6)
        if mean_jsd > 0.35:
            s_val = min(s_val, 0.2)
        sdist = {"s_dist": s_val, "mean_jsd": mean_jsd, "relative_ratio": round(ratio, 2),
                 "split_half": sh, "comparable_cells": len(entries),
                 "most_divergent": entries[:5], "dist_outlier": bool(outliers),
                 "outlier_cells": [{"cell": o["cell"], "jsd": round(o["jsd"], 3)}
                                   for o in outliers]}
    else:
        sdist = {"s_dist": None, "mean_jsd": None, "comparable_cells": 0,
                 "dist_outlier": False, "outlier_cells": []}
    dist_cmp = {**sdist, "baseline_p50": verify["signals"]["dist"].get("baseline_p50")}
    M[d] = {"recs": recs, "verify": verify, "ref": ref, "d": dn, "dist_cmp": dist_cmp,
            "mid": mid, "req": requested_family(mid, aliases)}


def snap(recs, m):
    I = [r for r in recs if r.get("layer") == "I"]
    K = [r for r in recs if r.get("layer") == "K"]
    C = [r for r in recs if r.get("layer") == "C"]
    S = [r for r in recs if r.get("layer") == "S"]
    idn = identity_signal(I, aliases, m["req"])
    met = meta_signal(K, I)
    rg = refuse_gradient_pattern(C)
    lc = length_compliance(S)
    scores, _, _ = _lexicon_scores(recs, aliases, m["req"])
    top1, conf, _ = _normalize(scores)
    return (round(idn["s_idn"], 4), idn["zh_en_consistent"], idn["parseable"],
            round(met["s_meta"], 4), len(met["cutoffs_unique"]),
            tuple(sorted(rg.items())),
            (sum(1 for x in lc if x["ok"]), len(lc)), (top1, round(conf, 4)))


def full_replay(recs, m):
    return build_report(recs, m["d"], m["dist_cmp"], m["mid"], m["ref"]["model"],
                        aliases, m["verify"].get("tokens_used") or {},
                        m["verify"].get("elapsed_s") or 0.0,
                        attribution=None, adversarial=None, mode="verify")


# ---------- 0. 重放保真 ----------
log("【0. 重放保真检查】(我的基线重放 vs 存档 verify.json)")
BASE_SNAP, BASE_RPT = {}, {}
for d in DIRS:
    m = M[d]
    BASE_SNAP[d] = snap(m["recs"], m)
    BASE_RPT[d] = full_replay(m["recs"], m)
    ok_v = BASE_RPT[d]["verdict"] == m["verify"]["verdict"]
    ok_s = abs(BASE_RPT[d]["score"] - m["verify"]["score"]) < 0.02
    log(f"  {d:24s} verdict {'✓' if ok_v else '✗'}({BASE_RPT[d]['verdict']}/{m['verify']['verdict']}) "
        f"score {'✓' if ok_s else '✗'}({BASE_RPT[d]['score']:.4f}/{m['verify']['score']:.4f})")
log()

# ---------- 1. 探针清单与角色 ----------
TEXT_IDS = []
seen = set()
for d in DIRS:
    for r in M[d]["recs"]:
        if r.get("layer") in ("I", "K", "C", "S") and r["id"] not in seen:
            seen.add(r["id"])
            TEXT_IDS.append((r["layer"], r["id"]))
TEXT_IDS.sort()
PID = [p for _, p in TEXT_IDS]

ROLE = {}
for d in DIRS:
    for r in M[d]["recs"]:
        if r.get("layer") in ("I", "K", "C", "S") and r["id"] not in ROLE:
            mt = r.get("meta") or {}
            tags = []
            if mt.get("pair"):
                tags.append(f"pair={mt['pair']}/{mt.get('lang')}")
            if mt.get("metacog"):
                tags.append("metacog审计")
            if mt.get("refusal_grad"):
                tags.append(f"拒答L{mt['refusal_grad']}")
            if mt.get("len_ctrl"):
                tags.append(f"len={mt['len_ctrl']}")
            ROLE[r["id"]] = " ".join(tags) or "—"
log(f"【1. 文本探针 {len(PID)} 条】I13+K6+C7+S10，角色标注见总表")
log()

# ---------- 2+3. 信息含量 + drop-one ----------
def answer_of(d, pid):
    for r in M[d]["recs"]:
        if r["id"] == pid and not r.get("error"):
            return r.get("response") or ""
    return None


rows = {}
for pid in PID:
    own = foreign = none_c = err_c = 0
    foreign_detail = []
    answers = {}
    for d in DIRS:
        resp = answer_of(d, pid)
        if resp is None:
            err_c += 1
            continue
        answers[d] = resp
        fams = _families_in_text(resp, aliases)
        req = M[d]["req"]
        if req in fams:
            own += 1
        if fams - {req}:
            foreign += 1
            foreign_detail.append(f"{d.split('_')[0]}→{sorted(fams - {req})}")
        if not fams:
            none_c += 1
    ts = [toks(t) for t in answers.values()]
    jac = [jaccard(a, b) for a, b in itertools.combinations(ts, 2)] or [1.0]
    rows[pid] = {"own": own, "foreign": foreign, "none": none_c, "err": err_c,
                 "jac": sum(jac) / len(jac), "foreign_detail": foreign_detail}

    # drop-one 重放
    sig_ch, ver_flip, dmax = 0, 0, 0.0
    for d in DIRS:
        m = M[d]
        recs_p = [r for r in m["recs"] if r["id"] != pid]
        if snap(recs_p, m) != BASE_SNAP[d]:
            sig_ch += 1
            rp = full_replay(recs_p, m)
            if rp["verdict"] != BASE_RPT[d]["verdict"]:
                ver_flip += 1
            dmax = max(dmax, abs(rp["score"] - BASE_RPT[d]["score"]))
    rows[pid].update({"sig_ch": sig_ch, "ver_flip": ver_flip, "dmax": round(dmax, 4)})

# ---------- 4. 配对剪评估 ----------
PAIRS = defaultdict(list)
for d in DIRS:
    for r in M[d]["recs"]:
        mt = r.get("meta") or {}
        if r.get("layer") == "I" and mt.get("pair"):
            if r["id"] not in PAIRS[mt["pair"]]:
                PAIRS[mt["pair"]].append(r["id"])
pair_res = {}
for pname, ids in sorted(PAIRS.items()):
    sig_ch, ver_flip, dmax = 0, 0, 0.0
    for d in DIRS:
        m = M[d]
        drop = set(ids)
        recs_p = [r for r in m["recs"] if r["id"] not in drop]
        if snap(recs_p, m) != BASE_SNAP[d]:
            sig_ch += 1
            rp = full_replay(recs_p, m)
            ver_flip += rp["verdict"] != BASE_RPT[d]["verdict"]
            dmax = max(dmax, abs(rp["score"] - BASE_RPT[d]["score"]))
    pair_res[pname] = (sorted(ids), sig_ch, ver_flip, round(dmax, 4))

# ---------- 汇总表 ----------
log("【2. 文本探针总表】own=自证家族数 foreign=污染证据数(高价值) jac=答案同质度(低=区分力强) "
    "sig=信号变动模型数 flip=判决翻转 dS=最大分差")
log(f"{'探针':22s} {'层':2s} {'own':>3s} {'for':>3s} {'none':>4s} {'jac':>5s} "
    f"{'sig':>3s} {'flip':>4s} {'dS':>6s}  角色")
order = {"I": 0, "K": 1, "C": 2, "S": 3}
for layer, pid in sorted(TEXT_IDS, key=lambda x: (order[x[0]], -rows[x[1]]["sig_ch"])):
    r = rows[pid]
    log(f"{pid:22s} {layer:2s} {r['own']:3d} {r['foreign']:3d} {r['none']:4d} "
        f"{r['jac']:5.2f} {r['sig_ch']:3d} {r['ver_flip']:4d} {r['dmax']:6.3f}  {ROLE[pid]}")
    if r["foreign_detail"]:
        log(f"{'':24s}污染: {'; '.join(r['foreign_detail'][:4])}")
log()

log("【3. i 层配对剪评估】(成对删除 en+zh)")
for pname, (ids, sig, flip, ds) in pair_res.items():
    log(f"  pair={pname:10s} {ids}  信号变动 {sig}/9  判决翻转 {flip}  maxΔS {ds}")
log()

# ---------- 5. 层级保底 ----------
cut_ids = [p for p in PID if "cutoff" in p]
metacog_ids = [p for p in PID if "metacog" in ROLE[p]]
c_levels = sorted({re.search(r"拒答L(\d)", ROLE[p]).group(1) for p in PID if "拒答" in ROLE[p]})
s_lens = sorted({re.search(r"len=(\d+)", ROLE[p]).group(1) for p in PID if "len=" in ROLE[p]})
log("【4. 层级保底现状】")
log(f"  截止探针 {len(cut_ids)}: {cut_ids} → 唯一性检查需 ≥2")
log(f"  元认知审计 {len(metacog_ids)}: {metacog_ids} → 需 ≥1")
log(f"  C 梯度级 {c_levels} → 梯度需多级")
log(f"  S 长度控制目标 {s_lens} → 每种 ≥1")
log()

# ---------- 6. ADV ----------
ADV_CFG = [("glm_53", "Kimi"), ("kimi_k3", "GLM"), ("deepseek_v4_pro", "GLM")]
log("【5. ADV 探针 ×3 模型注入态】(顺从=自称被注入的伪装家族)")
adv_data = {}
for d, role in ADV_CFG:
    adv = [json.loads(l) for l in open(f"{BFD}/{d}/adv/raw_answers.jsonl")]
    base_ids = {r["id"] for r in M[d]["recs"]}
    advp = sorted({r["id"] for r in adv if r["id"] not in base_ids
                   and r.get("layer") != "D"})
    role_key = requested_family(role, aliases) or role.lower()
    adv_data[d] = {"probes": advp, "role_key": role_key, "recs": adv}
    log(f"  {d} (注入角色={role}/{role_key}): ADV探针 {len(advp)} 条")
all_adv = sorted(set().union(*[set(adv_data[d]["probes"]) for d, _ in ADV_CFG]))
log(f"{'探针':26s}" + "".join(f"{d[:12]:>14s}" for d, _ in ADV_CFG))
adv_matrix = {}
for pid in all_adv:
    line = f"{pid:26s}"
    vals = []
    for d, role in ADV_CFG:
        recs = adv_data[d]["recs"]
        resp = next((r.get("response") or "" for r in recs
                     if r["id"] == pid and not r.get("error")), "")
        fams = _families_in_text(resp, aliases) if resp else set()
        rk = adv_data[d]["role_key"]
        v = ("顺从" if rk in fams else
             ("自守" if M[d]["req"] in fams else ("空" if not resp else "回避")))
        vals.append(v)
        line += f"{v:>14s}"
    adv_matrix[pid] = vals
    log(line + f"   {ROLE.get(pid, '')}")
log()

# ---------- 7. V ----------
log("【6. V 探针 ×2 模型】")
for d in ("glm_53", "kimi_k3"):
    var = [json.loads(l) for l in open(f"{BFD}/{d}/var/raw_answers.jsonl")]
    base_ids = {r["id"] for r in M[d]["recs"]}
    vp = sorted({r["id"] for r in var if r["id"] not in base_ids
                 and r.get("layer") != "D"})
    log(f"  {d}: V探针 {len(vp)} 条: {vp}")
    if d == "glm_53":
        for pid in vp:
            r0 = next((r for r in var if r["id"] == pid), {})
            prompt = (r0.get("prompt") or "")[:56].replace("\n", " ")
            log(f"    {pid:26s} {prompt}")
all_v = sorted({r["id"] for r in [json.loads(l) for l in open(f"{BFD}/glm_53/var/raw_answers.jsonl")]
                if r["id"] not in {x["id"] for x in M["glm_53"]["recs"]}
                and r.get("layer") != "D"})
vpairs = [(a, b, round(jaccard(toks(str(a)), toks(str(b))), 2))
          for a, b in itertools.combinations(all_v, 2)]
vpairs.sort(key=lambda x: -x[2])
log(f"  V 探针间最高相似对: {vpairs[:3] if vpairs else '无'}")
log()

# ---------- 8. 剪枝建议 ----------
CORE = [p for p in PID if rows[p]["sig_ch"] > 0]
SENTINEL = [p for p in PID if rows[p]["foreign"] > 0]
ZERO = [p for p in PID if rows[p]["sig_ch"] == 0 and rows[p]["ver_flip"] == 0]
log("【7. 剪枝建议】")
log(f"  核心载荷(信号变动>0): {len(CORE)} 条")
log(f"  污染哨兵(抓到 foreign 自称, 场景价值高, 建议全保留): {len(SENTINEL)} 条")
log(f"  零载荷(sig=0 且 flip=0): {len(ZERO)} 条 → 其中可进一步看同质度 jac 与角色保底")
for p in ZERO:
    log(f"    {p:22s} jac={rows[p]['jac']:.2f} own={rows[p]['own']} 角色={ROLE[p]}")
log()
log("  注意: drop-one 是在 9 个已知清洁模型上测的边际价值; 越狱/乱码类探针的价值在")
log("  '未知或被污染模型'场景(我们的 9 个都攻不破, 不代表下个模型攻不破), 剪此类需谨慎。")

with open(f"{BFD}/probe_snr_report.txt", "w") as f:
    f.write("\n".join(LINES) + "\n")
with open(f"{BFD}/probe_snr.json", "w") as f:
    json.dump({"rows": {p: rows[p] for p in PID},
               "pairs": pair_res, "adv_matrix": adv_matrix,
               "core": CORE, "sentinel": SENTINEL, "zero": ZERO},
              f, ensure_ascii=False, indent=1)
print("\n已写入 /tmp/bfd/probe_snr_report.txt + /tmp/bfd/probe_snr.json")