09b2add673
New evalharness/fingerprint/ package (from evalstone fp_fusion v1.1, 2026-09-07 pruning final): probe battery -> concurrent collection -> five scoring views (verify/attribution/variant/adversarial/robustness), bundled family aliases + 27 reference fingerprints (12 fp_fusion schema). - CLI: 'evalharness fingerprint run ...' (REMAINDER passthrough, single source of arg definitions) + 'fingerprint list' for bundled references - imports rewritten package-relative; direct 'python3 run_fp_fusion.py' execution kept working via package bootstrap - offline analysis/collection scripts made path-independent (previously pinned to a /opt/evalscope path absent on this host) - shell scripts: hardcoded API key -> FP_API_KEY/OPENAI_API_KEY env vars - --reference accepts short names resolved against bundled references/ - pyproject: +httpx dependency, package-data references/*.json - tests/test_fingerprint.py: 10 offline tests (battery definitions, assembly counts, normalization, signals, verdict ladder, CLI wiring) - README: fingerprint section + architecture entry Verified on H20-1: tests 10/10, installed CLI OK, full-protocol run vs vectron GLM-5.3 reproduces baseline (score 0.9451, s_idn 0.846).
91 lines
4.1 KiB
Python
91 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
||
"""离线推导 attribution 报告：从已有 verify 原始记录重放打分管线。
|
||
|
||
原理：_assemble_probes('attribution') 与 'verify' 的探针电池完全一致
|
||
（ALL_TEXT_PROBES，无新增探针），attribution 只是多一层打分。
|
||
因此对同一份 raw 记录重放 engine+attribution+build_report，
|
||
即可得到与真实 attribution 运行等价的报告（省去重复 API 采样）。
|
||
|
||
用法: python3 derive_attribution.py <model_dir> <model_id> <ref_path> [raw_file]
|
||
输出: <model_dir>/attribution.json
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
from pathlib import Path # noqa: E402
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2])) # 仓库根/site-packages, 使 evalharness 包可导入
|
||
from evalharness.fingerprint.engine import (build_d_normalized, compare_cells, distributions_by_cell, # noqa: E402
|
||
load_reference, split_half_jsd)
|
||
from evalharness.fingerprint.scorer import build_report, load_aliases, requested_family # noqa: E402
|
||
from evalharness.fingerprint.attribution import family_attribution # noqa: E402
|
||
|
||
|
||
def derive(model_dir, model_id, ref_path, raw_file=None):
|
||
raw = raw_file or os.path.join(model_dir, "raw_answers.jsonl")
|
||
records = [json.loads(l) for l in open(raw, encoding="utf-8")]
|
||
verify = json.load(open(os.path.join(model_dir, "verify.json")))
|
||
|
||
n_err = sum(1 for r in records if r.get("error"))
|
||
if n_err / max(len(records), 1) > 0.2:
|
||
print(f"SKIP {model_dir}: error rate {n_err}/{len(records)} too high")
|
||
return None
|
||
|
||
ref = load_reference(ref_path)
|
||
reference_info, ref_cells = ref["model"], ref["cells"]
|
||
|
||
d_norm = build_d_normalized(records)
|
||
split_half = split_half_jsd(d_norm)
|
||
dist_a = distributions_by_cell(d_norm)
|
||
entries, mean_jsd = compare_cells(dist_a, ref_cells)
|
||
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:
|
||
sh = split_half if split_half and split_half > 0 else 0.02
|
||
ratio = mean_jsd / max(sh, 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)
|
||
s = {"s_dist": s_val, "mean_jsd": mean_jsd,
|
||
"relative_ratio": round(ratio, 2),
|
||
"split_half": split_half,
|
||
"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:
|
||
s = {"s_dist": None, "mean_jsd": None, "comparable_cells": 0,
|
||
"dist_outlier": False, "outlier_cells": [],
|
||
"note": "no comparable cells (valid samples too few)"}
|
||
dist_cmp = {**s, "baseline_p50": verify["signals"]["dist"].get("baseline_p50")}
|
||
|
||
aliases = load_aliases(None)
|
||
req_family = requested_family(model_id, aliases)
|
||
attribution = family_attribution(records, aliases=aliases,
|
||
requested_family=req_family,
|
||
llmmap_tool=None)
|
||
|
||
report = build_report(records, d_norm, dist_cmp, model_id, reference_info,
|
||
aliases,
|
||
verify.get("tokens_used") or {},
|
||
verify.get("elapsed_s") or 0.0,
|
||
attribution=attribution, adversarial=None,
|
||
mode="attribution")
|
||
out = os.path.join(model_dir, "attribution.json")
|
||
with open(out, "w", encoding="utf-8") as f:
|
||
json.dump(report, f, ensure_ascii=False, indent=2)
|
||
fam = report["signals"].get("family") or {}
|
||
print(f"derived {out}")
|
||
print(f" model={model_id} req_family={req_family} verdict={report['verdict']} "
|
||
f"score={report['score']}")
|
||
print(f" top1={fam.get('top1_family')} conf={fam.get('confidence')} "
|
||
f"s_fam={fam.get('s_fam')} conflict={fam.get('conflict')}")
|
||
return report
|
||
|
||
|
||
if __name__ == "__main__":
|
||
derive(sys.argv[1], sys.argv[2], sys.argv[3],
|
||
sys.argv[4] if len(sys.argv) > 4 else None)
|