#!/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")