aggregate.py
3.08 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
#!/usr/bin/env python3
"""从 round{N}_verdicts.json 汇总 A 轨 8 维 / B 轨 11 场景分数（分数可复算的唯一入口）。
用法: aggregate.py scores_v1/round1_verdicts.json
"""
import json, sys
SCN = json.load(open('/app/system/agents/deep-research/train/digest/test_scenarios_v2.json'))
WEIGHT = {'D1': 15, 'D2': 15, 'D3': 15, 'D4': 15, 'D5': 10, 'D6': 10, 'D7': 10, 'D8': 10}
EW = {'E1': 20, 'E2': 20, 'E3': 30, 'E4': 20, 'E5': 10}
def main(path):
V = json.load(open(path))
meta = SCN['meta']; crit_meta = {}
for s in SCN['scenarios']:
for c in s['criteria']:
crit_meta[c['id']] = c
dims = {d: [0, 0] for d in WEIGHT}
rows = []
a_total = 0.0
for sid, sv in sorted(V['scenarios'].items()):
a_hit = a_all = 0
for cid, cv in sv['criteria'].items():
cm = crit_meta[cid]
pts = cm['points']
if cm['track'] != 'A':
continue
a_all += pts
if cv['hit']:
a_hit += pts
d = dims[cm['dim']]; d[0] += pts
dims[cm['dim']][1] += pts
scn_a = round(a_hit / a_all * 100, 2) if a_all else 0.0
b = sum(sv['B'][e]['score'] * w for e, w in EW.items()) / 100.0
veto = not sv['B']['E4']['score'] == 100
if veto:
b = 0.0 # E4 一票否决：该场景 B 轨直接 0 分（benchmark §三）
rows.append((sid, scn_a, round(b, 2), veto))
a_total += 0 # placeholder
dimscore = {d: round(h / t * 100, 2) if t else 0.0 for d, (h, t) in dims.items()}
a_track = round(sum(dimscore[d] * WEIGHT[d] / 100 for d in WEIGHT), 2)
scn_a_mean = round(sum(r[1] for r in rows) / len(rows), 2)
print(f'== {path} skill {V["skill_version"]} ==')
for d in WEIGHT:
h, t = dims[d]
print(f' {d} 权重{WEIGHT[d]:>3}: {h}/{t} = {dimscore[d]:>6} ×{WEIGHT[d]/100:.2f} = {round(dimscore[d]*WEIGHT[d]/100,2):>5}')
print(f' A 轨总分 = {a_track} （11 场景 A 分简单均值 = {scn_a_mean}）')
print(f' 弱维度（低于 60）: {[d for d in WEIGHT if dimscore[d] < 60]}')
print(' 场景明细: 场景A分 / B分 / E4否决')
for sid, a, b, veto in rows:
flag = '✓' if (a >= 90 and b >= 85 and not veto) else '✗'
print(f' {sid}: A={a:>6} B={b:>6} veto={veto} {flag}')
ok = a_track >= 90 and all(b >= 85 and not v for _, _, b, v in rows)
print(f' 双门槛: A轨 {a_track} ≥90 ? {a_track >= 90}；B轨逐场景 ≥85 ? {all(b >= 85 and not v for _, _, b, v in rows)} → {"放行" if ok else "不放行"}')
out = path.replace('_verdicts.json', '_summary.json')
json.dump({'skill_version': V['skill_version'], 'dimensions': {d: {'hit': dims[d][0], 'total': dims[d][1], 'score': dimscore[d], 'weight': WEIGHT[d]} for d in WEIGHT},
'a_track_total': a_track, 'a_scenario_mean': scn_a_mean,
'scenarios': {sid: {'A': a, 'B': b, 'veto': v} for sid, a, b, v in rows},
'gate_pass': ok}, open(out, 'w'), ensure_ascii=False, indent=1)
print(f' -> {out}')
if __name__ == '__main__':
main(sys.argv[1])