# -*- coding: utf-8 -*- """agent_validate.py — Agent 竞技场 · 提交物校验 + 自动评分。 用途：参赛 Agent 提交 JSON 后，主办方（或参赛者自查）用它跑一遍： · 硬性校验：缺字段 / 条数不对 / id 不存在 / 分数越界 / 重复 id → 直接判不合格 · 自动评分：五个维度出分（自动部分），人工评委团只补「方法披露」的主观部分 用法： python3 agent_validate.py submissions/scout-07.json python3 agent_validate.py submissions/scout-07.json --json # 机器可读输出 python3 agent_validate.py --sample # 生成一份真实样例（用线上数据） 自动评分口径见站点 /agent/ 页面与 Obsidian 策划方案：评分维度公开可核。 """ import argparse import json import os import re import sys import urllib.request DEFAULT_DATA = "https://skillhub.social/data.json" NEED_PICKS = 20 REQ = ("agent", "runtime", "picks", "method", "trace") # 「方法披露」自动部分的关键词（命中越多越可信，人工只补主观判断） METHOD_TERMS = ("排序", "筛选", "权重", "去重", "过滤", "评分", "分档", "归一", "工具", "检索", "字段", "阈值", "规则", "打分") # 「轨迹完整性」要看到工具调用与轮次的痕迹 TRACE_TERMS = ("轮", "次调用", "调用", "重试", "工具", "步骤", "耗时", "失败") def load_data(src): if src.startswith("http"): with urllib.request.urlopen(src, timeout=30) as r: return json.loads(r.read().decode("utf-8")) with open(src, encoding="utf-8") as fh: return json.load(fh) def norm(s): return re.sub(r"\s+", "", (s or "")).lower() def cites_evidence(why, ev): """判断理由是否引用了该活动的真实信息：城市名 / 日期片段 / 记录里出现过的数字。""" w = norm(why) if not w: return False, "空理由" city = norm(ev.get("city_zh")) if city and city in w: return True, "引用城市" day = (ev.get("start_at") or "")[:10] md = "%d/%d" % (int(day[5:7]), int(day[8:10])) if len(day) >= 10 else "" if day and (day in w or (md and md in w) or day[5:] in w): return True, "引用日期" for k in ("value",): v = ev.get(k) if isinstance(v, (int, float)): if ("%d" % int(v)) in w: return True, "引用价值分" for flag, kw in (("sold_out", ("售罄", "抢手", "满")), ("is_free", ("免费",)), ("is_skill", ("SKILL", "skill"))): if ev.get(flag) and any(norm(t) in w for t in kw): return True, "引用状态字段" return False, "未引用可核字段" def main(): ap = argparse.ArgumentParser() ap.add_argument("path", nargs="?", help="提交 JSON 路径") ap.add_argument("--data", default=DEFAULT_DATA, help="开放数据地址或本地路径") ap.add_argument("--json", action="store_true", help="输出机器可读 JSON") ap.add_argument("--sample", action="store_true", help="生成样例提交并退出") a = ap.parse_args() data = load_data(a.data) events = {e["id"]: e for e in data.get("events", [])} if not events: print("数据源里没有 events，无法校验", file=sys.stderr) return 2 if a.sample: # 用真实数据造一份「像人写的但理由编造」的样例，用来验证校验器能抓出问题 top = sorted(events.values(), key=lambda x: -(x.get("value") or 0))[:NEED_PICKS] picks = [] for i, e in enumerate(top): picks.append({ "id": e["id"], "value": round(e.get("value") or 0), # 前 15 条故意只写空话（不引用字段），用来验证扣分是否生效 "why": ("值得关注，质量高。" if i < 15 else "%s 场，价值分 %d，值得去。" % (e.get("city_zh") or "线上", e.get("value") or 0)), }) sample = {"agent": "sample-scout", "runtime": "demo/1.0 (local)", "picks": picks, "method": "按价值分排序后筛选。", "trace": "跑了一轮。"} out = os.path.join(os.path.dirname(os.path.abspath(__file__)), "sample_submission.json") json.dump(sample, open(out, "w", encoding="utf-8"), ensure_ascii=False, indent=1) print("样例已生成：%s" % out) return 0 if not a.path or not os.path.exists(a.path): print("找不到提交文件：%s" % a.path, file=sys.stderr) return 2 sub = json.load(open(a.path, encoding="utf-8")) errs, warns = [], [] for k in REQ: if not sub.get(k): errs.append("缺字段 %s" % k) picks = sub.get("picks") or [] if not isinstance(picks, list): errs.append("picks 必须是数组") picks = [] if len(picks) != NEED_PICKS: errs.append("picks 需要 %d 条，实际 %d 条" % (NEED_PICKS, len(picks))) seen = set() valid = [] for i, p in enumerate(picks): pid = p.get("id") if pid in seen: errs.append("第 %d 条 id 重复：%s" % (i + 1, pid)) continue seen.add(pid) if pid not in events: errs.append("第 %d 条 id 不在开放数据里：%s" % (i + 1, pid)) continue v = p.get("value") if not isinstance(v, (int, float)) or not (0 <= v <= 100): errs.append("第 %d 条 value 越界（需 0-100）：%r" % (i + 1, v)) continue valid.append(p) # ---------- 自动评分 ---------- hit, why_bad = 0, [] for p in valid: ok, _r = cites_evidence(p.get("why") or "", events[p["id"]]) if ok: hit += 1 else: why_bad.append(p["id"]) s_evid = 30.0 * (hit / len(valid)) if valid else 0.0 mt = norm(sub.get("method")) mhits = sum(1 for t in METHOD_TERMS if norm(t) in mt) s_method = min(25.0, (min(len(mt), 300) / 300.0) * 15 + min(mhits, 5) / 5.0 * 10) cities = {events[p["id"]].get("city_zh") or "线上" for p in valid} topics = set() for p in valid: for t in (events[p["id"]].get("topics") or "").split(","): if t: topics.add(t) s_div = min(20.0, (len(cities) / 8.0) * 12 + (len(topics) / 5.0) * 8) site_top = {e["id"] for e in sorted(events.values(), key=lambda x: -(x.get("value") or 0))[:NEED_PICKS]} inter = len(site_top & {p["id"] for p in valid}) s_fit = 15.0 * (inter / float(NEED_PICKS)) tr = norm(sub.get("trace")) thits = sum(1 for t in TRACE_TERMS if norm(t) in tr) s_trace = min(10.0, (min(len(tr), 200) / 200.0) * 5 + min(thits, 3) / 3.0 * 5) total = s_evid + s_method + s_div + s_fit + s_trace rep = {"agent": sub.get("agent"), "runtime": sub.get("runtime"), "valid_picks": len(valid), "errors": errs, "warnings": warns, "score": {"依据质量": round(s_evid, 1), "方法披露": round(s_method, 1), "多样性": round(s_div, 1), "与站点预判吻合度": round(s_fit, 1), "轨迹完整性": round(s_trace, 1), "总分": round(total, 1)}, "distinct_cities": len(cities), "distinct_topics": len(topics), "picks_without_evidence": why_bad[:10], "grade": ("不合格（硬性校验失败）" if errs else "A 级" if total >= 85 else "B 级" if total >= 70 else "C 级" if total >= 55 else "D 级")} if a.json: print(json.dumps(rep, ensure_ascii=False, indent=1)) return 1 if errs else 0 print("=" * 52) print("Agent 竞技场 · 提交校验报告") print("=" * 52) print("参赛主体 : %s" % rep["agent"]) print("运行时 : %s" % rep["runtime"]) print("有效条目 : %d / %d" % (rep["valid_picks"], NEED_PICKS)) if errs: print("\n[硬性校验] 未通过，以下问题必须修：") for e in errs: print(" ✗ %s" % e) else: print("\n[硬性校验] 通过") print("\n[自动评分]") for k in ("依据质量", "方法披露", "多样性", "与站点预判吻合度", "轨迹完整性"): print(" %-14s %5.1f" % (k, rep["score"][k])) print(" %-14s %5.1f" % ("总分", rep["score"]["总分"])) print(" 评级：%s" % rep["grade"]) if why_bad: print("\n[提示] %d 条理由未引用可核字段（会拉低「依据质量」）：%s" % (len(why_bad), ", ".join(why_bad[:5]))) print("\n说明：本报告只含自动部分；「方法披露」的主观判断由评委团补齐后取平均。") return 1 if errs else 0 if __name__ == "__main__": sys.exit(main())