EvalHarness/evalharness/fingerprint/build_matrix.py

#!/usr/bin/env python3
"""汇总四维×9 能力矩阵 CSV + 一致性校验（反向验证 / attribution 等价性）。"""
import csv
import json
import os

BFD = "/tmp/bfd"

# (短名, 模型ID, 参考文件, 参考口径)
MODELS = [
    ("deepseek_v4_flash",      "DeepSeek/DeepSeek-V4-Flash",      "deepseek_v4_flash_fusion_reference.json",      "旧key"),
    ("deepseek_v4_flash_0731", "DeepSeek/DeepSeek-V4-Flash-0731", "deepseek_v4_flash_0731_fusion_reference.json", "旧key"),
    ("deepseek_v4_pro",        "DeepSeek/DeepSeek-V4-Pro",        "deepseek_v4_pro_fusion_reference.json",        "旧key"),
    ("glm_51",                 "ZhipuAi/GLM-5.1",                 "glm_51_fusion_reference.json",                 "新key"),
    ("glm_52",                 "ZhipuAi/GLM-5.2",                 "glm52_vectron_fusion_reference.json",          "旧key"),
    ("glm_53",                 "ZhipuAi/GLM-5.3",                 "glm53_fusion_reference.json",                  "旧key"),
    ("kimi_k2_6",              "MoonshotAi/Kimi-K2.6",            "kimi_k2_6_fusion_reference.json",              "新key"),
    ("kimi_k2_7code",          "MoonshotAi/Kimi-K2.7-Code",       "kimi_k2_7code_fusion_reference.json",          "新key"),
    ("kimi_k3",                "MoonshotAi/Kimi-K3",              "kimi_k3_fusion_reference.json",                "旧key"),
]
REPRESENTATIVES = ("glm_53", "kimi_k3", "deepseek_v4_pro")


def load(path):
    if os.path.exists(path):
        try:
            return json.load(open(path))
        except Exception:
            return None
    return None


def fmt(x, nd=3):
    if x is None:
        return ""
    if isinstance(x, float):
        return f"{x:.{nd}f}"
    return str(x)


rows = []
for name, mid, ref,口径 in MODELS:
    d = os.path.join(BFD, name)
    v = load(f"{d}/verify.json")
    a = load(f"{d}/attribution.json")
    adv = load(f"{d}/adv/adversarial.json")
    var = load(f"{d}/var/variant.json")
    rob = load(f"{d}/rob/robustness.json")

    # 识别（verify）
    if v:
        sig = v.get("signals", {})
        ident = {"verdict": v.get("verdict"), "score": v.get("score"),
                 "meanJSD": sig.get("dist", {}).get("mean_jsd"),
                 "gate": v.get("gate", {}).get("quality"),
                 "s_idn": sig.get("identity", {}).get("s_idn")}
    else:
        ident = dict.fromkeys(("verdict", "score", "meanJSD", "gate", "s_idn"))

    # 归因（attribution）
    if a:
        fam = a.get("signals", {}).get("family") or {}
        attr = {"verdict": a.get("verdict"), "score": a.get("score"),
                "top1": fam.get("top1_family"), "conf": fam.get("confidence"),
                "s_fam": fam.get("s_fam"), "conflict": fam.get("conflict"),
                "req": mid.split("/")[0].replace("ZhipuAi", "glm")
                       .replace("MoonshotAi", "kimi").replace("DeepSeek", "deepseek")}
    else:
        attr = dict.fromkeys(("verdict", "score", "top1", "conf", "s_fam", "conflict", "req"))

    # 对抗（adversarial，仅 3 代表）
    if adv:
        sig = adv.get("signals", {})
        av = sig.get("adversarial") or {}
        ad = {"imp_flag": av.get("impersonation_flag"),
              "role_yield": av.get("role_yield"),
              "conflict": av.get("claimed_behavior_conflict"),
              "style_suspect": av.get("style_imitation_suspect")}
    else:
        ad = dict.fromkeys(("imp_flag", "role_yield", "conflict", "style_suspect"))
        ad["imp_flag"] = "未跑(3代表抽样)" if name not in REPRESENTATIVES else "待跑"

    # 变体（variant，仅 3 代表）
    if var:
        vs = (var.get("signals", {}).get("variant") or {})
        vt = {"graybox": vs.get("graybox_present"),
              "top1_stab": vs.get("top1_stability"),
              "self_jsd": vs.get("self_consistency_jsd"),
              "logprob_mean": vs.get("logprob_mean")}
    else:
        vt = dict.fromkeys(("graybox", "top1_stab", "self_jsd", "logprob_mean"))
        vt["graybox"] = "未跑(3代表抽样)" if name not in REPRESENTATIVES else "待跑"

    # 鲁棒三轴（robustness，仅 3 代表）
    if rob:
        rs = (rob.get("signals", {}).get("robustness") or {})
        ta = rs.get("temp_axis") if isinstance(rs.get("temp_axis"), dict) else {}
        la = rs.get("lang_axis") if isinstance(rs.get("lang_axis"), dict) else {}
        pa = rs.get("paraphrase_axis") if isinstance(rs.get("paraphrase_axis"), dict) else {}
        rb = {"temp": fmt(ta.get("mean_consistency")),
              "lang": fmt(la.get("mean_consistency")),
              "para": fmt(pa.get("mean_jsd"))}
    else:
        rb = {"temp": "未跑" if name not in REPRESENTATIVES else "待跑",
              "lang": "未跑" if name not in REPRESENTATIVES else "待跑",
              "para": "未跑" if name not in REPRESENTATIVES else "待跑"}

    rows.append({
        "模型": mid, "短名": name, "参考口径": 口径,
        "识别_verdict": ident["verdict"], "识别_score": fmt(ident["score"]),
        "识别_meanJSD": fmt(ident["meanJSD"]), "识别_gate": ident["gate"],
        "归因_verdict": attr["verdict"], "归因_score": fmt(attr["score"]),
        "归因_top1": attr["top1"], "归因_conf": fmt(attr["conf"]),
        "归因_s_fam": fmt(attr["s_fam"]), "归因_冲突": attr["conflict"],
        "对抗_冒充实锤": ad["imp_flag"], "对抗_角色屈服": fmt(ad["role_yield"]),
        "对抗_声称行为矛盾": fmt(ad["conflict"]), "对抗_风格模仿嫌疑": fmt(ad["style_suspect"]),
        "变体_灰盒": vt["graybox"], "变体_top1稳定": fmt(vt["top1_stab"]),
        "变体_自一致JSD": fmt(vt["self_jsd"]), "变体_logprob均值": fmt(vt["logprob_mean"]),
        "鲁棒_温度轴": rb["temp"], "鲁棒_语言轴": rb["lang"], "鲁棒_改写轴JSD": rb["para"],
    })

out_csv = os.path.join(BFD, "能力矩阵.csv")
with open(out_csv, "w", newline="", encoding="utf-8-sig") as f:
    w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
    w.writeheader()
    w.writerows(rows)
print(f"written {out_csv} rows={len(rows)}")

# ---- 校验 1：反向验证（glm_53 verify 复跑 score 差 < 0.05）----
v1 = load(f"{BFD}/glm_53/verify.json")
v2 = load(f"{BFD}/glm_53/rerun/verify_rerun.json")
if v1 and v2:
    diff = abs(v1["score"] - v2["score"])
    print(f"[反向验证] glm_53 首跑={v1['score']} 复跑={v2['score']} 差={diff:.4f} "
          f"{'PASS(<0.05)' if diff < 0.05 else 'FAIL(>=0.05)'}")
else:
    print("[反向验证] glm_53 复跑尚未完成")

# ---- 校验 2：attribution 等价性（0731 真实运行 vs 离线推导）----
real = load(f"{BFD}/deepseek_v4_flash_0731/attr_real/attribution_real.json")
derived = load(f"{BFD}/deepseek_v4_flash_0731/attribution.json")
if real and derived:
    rf = (real.get("signals", {}).get("family") or {})
    df_ = (derived.get("signals", {}).get("family") or {})
    same_top1 = rf.get("top1_family") == df_.get("top1_family")
    print(f"[attribution 等价性] 0731 真实: top1={rf.get('top1_family')} conf={rf.get('confidence')} "
          f"score={real.get('score')} | 离线: top1={df_.get('top1_family')} conf={df_.get('confidence')} "
          f"score={derived.get('score')} → top1一致={same_top1}")
else:
    print("[attribution 等价性] 0731 真实运行尚未完成")

# ---- 校验 3：variant 灰盒（3 代表 graybox_present 均 True 且 logprob_mean 有值）----
ok = 0
for name in REPRESENTATIVES:
    var = load(f"{BFD}/{name}/var/variant.json")
    if var:
        vs = var.get("signals", {}).get("variant") or {}
        print(f"[variant 灰盒] {name}: graybox={vs.get('graybox_present')} "
              f"logprob_mean={vs.get('logprob_mean')}")
        if vs.get("graybox_present") and vs.get("logprob_mean") is not None:
            ok += 1
    else:
        print(f"[variant 灰盒] {name}: 待跑")
print(f"[variant 灰盒] 通过 {ok}/3")