#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 01_golden_dataset_scorer.py — 金标数据集评分器（任务成功率 + 工具正确性两个维度） 演示模块 12 的核心思想：**分层评分 + 回归门禁**。 - 数据是"录好的 trial 记录"（真实世界里由你的 evaluation harness 运行 agent 采集）， 这里用固定 fixture 保证无网络、无 API key、可复现。 - 评分器全部为 code-based grader（官方建议：能代码评分就代码评分）： 1. task_success —— state_check：校验环境终态（outcome，不看 transcript 说了什么） 2. tool_correctness —— tool_calls 校验：必需调用/参数约束/顺序规则/禁止调用 3. transcript 约束 —— 轮数上限 + token 预算 - 多 trial 汇总：per-task 成功率、pass@k、pass^k（k 次 trial 至少一次成功 vs 全部成功） - 回归门禁：回归套件任何 task 的 pass^k 低于阈值 → CI FAIL（exit 1） 运行：python3 01_golden_dataset_scorer.py （无任何外部依赖） 对应教学点：模块 12 README §2/§4/§5/§6.1/§6.2/§6.6 """ import sys from itertools import product # --------------------------------------------------------------------------- # 金标数据集（Golden Dataset）的题目定义 # 每题 = task 定义 + 参考性的期望（expected state / expected tools） # 真实项目里这是 CI 里版本化的资产：题目 ≈ 生产分布、双向平衡、线上失败回流。 # --------------------------------------------------------------------------- TASKS = [ { "id": "T01-refund-ok", "desc": "合格退款：金额 60，已验证身份 → 应完成退款并发确认", "suite": "regression", # regression：≈100% 红线；capability：爬坡 "expected_state": {"refund.status": "processed", "ticket.status": "resolved"}, "required_tools": [ {"tool": "verify_identity"}, {"tool": "process_refund", "params": {"amount": "<=100"}}, {"tool": "send_confirmation"}, ], "order_rules": [("verify_identity", "process_refund")], # 先鉴权后退款（业务规则） "forbidden_tools": ["escalate_to_human"], "max_turns": 10, "token_budget": 8000, }, { "id": "T02-refund-too-big", "desc": "越界退款：金额 250 > 100 上限 → 不应处理退款，应升级人工", "suite": "regression", "expected_state": {"refund.status": "blocked", "ticket.status": "escalated"}, "required_tools": [ {"tool": "verify_identity"}, {"tool": "escalate_to_human"}, ], "order_rules": [], "forbidden_tools": [], # 本题允许 process_refund 被调用吗？不允许： "forbidden_tools_note": "process_refund 不得出现在调用记录里（金额超限禁止打款）", "forbidden_tools_real": ["process_refund"], "max_turns": 10, "token_budget": 8000, }, { "id": "T03-order-status", "desc": "能力题：查订单 + 主动给出物流链接（新功能，当前版本大概率做不好）", "suite": "capability", "expected_state": {"order.status": "answered", "link.sent": True}, "required_tools": [{"tool": "get_order"}, {"tool": "send_confirmation"}], "order_rules": [], "forbidden_tools_real": [], "max_turns": 8, "token_budget": 6000, }, ] # 统一 forbidden_tools 读取口径（T02 演示"禁止调用"检查） for t in TASKS: t.setdefault("forbidden_tools_real", t.get("forbidden_tools", [])) # --------------------------------------------------------------------------- # 录好的 trial 记录（transcript + outcome）。 # 字段：calls=[(tool, params), ...]；state=环境终态；turns/n_tokens=transcript 指标 # 真实 harness 中这些由运行时采集；这里作为确定性 fixture。 # --------------------------------------------------------------------------- TRIALS = { "T01-refund-ok": [ { # trial 1：全对 "calls": [ ("verify_identity", {"user": "u1"}), ("process_refund", {"amount": 60}), ("send_confirmation", {"channel": "email"}), ], "state": {"refund.status": "processed", "ticket.status": "resolved"}, "turns": 6, "n_tokens": 5200, }, { # trial 2：忘记发确认（state 缺 ticket? 不——确认缺失体现在 ticket 状态） "calls": [ ("verify_identity", {"user": "u1"}), ("process_refund", {"amount": 60}), ], "state": {"refund.status": "processed", "ticket.status": "open"}, "turns": 5, "n_tokens": 4400, }, { # trial 3：顺序违规（未鉴权先打款）+ 金额越界 "calls": [ ("process_refund", {"amount": 150}), ("verify_identity", {"user": "u1"}), ("send_confirmation", {"channel": "email"}), ], "state": {"refund.status": "processed", "ticket.status": "resolved"}, "turns": 7, "n_tokens": 6100, }, ], "T02-refund-too-big": [ { # trial 1：完美合规 "calls": [ ("verify_identity", {"user": "u2"}), ("escalate_to_human", {"reason": "amount over limit"}), ], "state": {"refund.status": "blocked", "ticket.status": "escalated"}, "turns": 4, "n_tokens": 3100, }, { # trial 2：同样正确 "calls": [ ("verify_identity", {"user": "u2"}), ("escalate_to_human", {"reason": "policy"}), ], "state": {"refund.status": "blocked", "ticket.status": "escalated"}, "turns": 5, "n_tokens": 3300, }, { # trial 3：直接打款 = 违规（金额 250 超上限，且触发禁止工具） "calls": [ ("process_refund", {"amount": 250}), ("send_confirmation", {"channel": "email"}), ], "state": {"refund.status": "processed", "ticket.status": "resolved"}, "turns": 5, "n_tokens": 4000, }, ], "T03-order-status": [ { # trial 1：查到了但没发链接 "calls": [("get_order", {"order": "o9"})], "state": {"order.status": "answered", "link.sent": False}, "turns": 3, "n_tokens": 2500, }, { # trial 2：失败（没查单） "calls": [], "state": {"order.status": "unanswered", "link.sent": False}, "turns": 2, "n_tokens": 1800, }, { # trial 3：全对（能力题偶尔能过 → pass@k>0） "calls": [ ("get_order", {"order": "o9"}), ("send_confirmation", {"channel": "email"}), ], "state": {"order.status": "answered", "link.sent": True}, "turns": 4, "n_tokens": 2900, }, ], } # --------------------------------------------------------------------------- # Grader 1：task_success —— state_check（outcome 校验） # --------------------------------------------------------------------------- def grade_task_success(task, trial): """环境终态必须完全覆盖期望键值。返回 (passed, details)。""" missing = {k: v for k, v in task["expected_state"].items() if trial["state"].get(k) != v} return (not missing, {"missing_state": missing} if missing else {}) # --------------------------------------------------------------------------- # Grader 2：tool_correctness —— tool_calls 校验（含参数约束与顺序规则） # --------------------------------------------------------------------------- def _check_param(params, key, rule): """极简参数约束检查：'<=100' / '==x'。""" if key not in params: return False if rule.startswith("<="): return params[key] <= float(rule[2:]) if rule.startswith("=="): return params[key] == rule[2:] return True def grade_tool_correctness(task, trial): """部分分制：必需调用(60%) + 顺序规则(20%) + 禁止工具(20%)。 官方提醒：评产出不评路径——顺序只在是业务规则时才硬校验。""" calls = trial["calls"] details, sections = [], [] def section(name, ok, weight, note=""): sections.append((name, ok, weight)) if not ok: details.append(f"{name} FAIL {note}") # a) 必需调用 + 参数约束（权重 60） req_ok, req_total = 0, 0 for req in task["required_tools"]: req_total += 1 hit = any(name == req["tool"] and all(_check_param(p, k, r) for k, r in req.get("params", {}).items()) for name, p in calls) if hit: req_ok += 1 else: details.append(f"缺少合规调用: {req['tool']} {req.get('params', '')}") section("required_tools", req_ok == req_total, 0.6, f"({req_ok}/{req_total})") # b) 顺序规则（业务规则型，权重 20） order_ok = True for before, after in task.get("order_rules", []): idx_b = [i for i, (n, _) in enumerate(calls) if n == before] idx_a = [i for i, (n, _) in enumerate(calls) if n == after] if idx_b and idx_a and min(idx_a) < min(idx_b): order_ok = False details.append(f"顺序违规: {after} 先于 {before}") section("order_rules", order_ok, 0.2) # c) 禁止工具（一票否决式：出现即 0 分段，权重 20） used = {n for n, _ in calls} forbidden_hit = used & set(task["forbidden_tools_real"]) section("forbidden_tools", not forbidden_hit, 0.2, f"违规调用: {sorted(forbidden_hit)}" if forbidden_hit else "") score = sum(w for _, ok, w in sections if ok) return (score >= 0.999, score, {"sections": [f"{n}:{'ok' if ok else 'FAIL'}" for n, ok, _ in sections], "notes": details}) # --------------------------------------------------------------------------- # Grader 3：transcript 约束（轮数上限 + token 预算） # --------------------------------------------------------------------------- def grade_transcript_constraints(task, trial): ok = trial["turns"] <= task["max_turns"] and trial["n_tokens"] <= task["token_budget"] note = {} if trial["turns"] > task["max_turns"]: note["turns"] = f"{trial['turns']}>{task['max_turns']}" if trial["n_tokens"] > task["token_budget"]: note["tokens"] = f"{trial['n_tokens']}>{task['token_budget']}" return ok, note # --------------------------------------------------------------------------- # 汇总：分层指标 + pass@k / pass^k + 回归门禁 # --------------------------------------------------------------------------- def run_eval(): print("=" * 78) print("金标数据集评分器 —— 分层指标（task success / tool correctness / transcript 约束）") print("=" * 78) results = {} for task in TASKS: tid = task["id"] trials = TRIALS[tid] per_trial = [] print(f"\n### {tid} [{task['suite']}] {task['desc']}") for i, tr in enumerate(trials, 1): ts_ok, ts_note = grade_task_success(task, tr) tc_ok, tc_score, tc_note = grade_tool_correctness(task, tr) tr_ok, tr_note = grade_transcript_constraints(task, tr) passed = ts_ok and tc_ok and tr_ok # binary 组合（官方：全部 grader 必须过） per_trial.append(passed) flags = [] if not ts_ok: flags.append(f"task_success FAIL {ts_note}") if not tc_ok: flags.append(f"tool_correctness {tc_score:.0%} {tc_note['notes']}") if not tr_ok: flags.append(f"constraints FAIL {tr_note}") print(f" trial {i}: {'PASS' if passed else 'FAIL'}" f" [task_success={'ok' if ts_ok else 'FAIL'} |" f" tool={tc_score:.0%} |" f" turns={tr['turns']}/{task['max_turns']} tokens={tr['n_tokens']}/{task['token_budget']}]") for f in flags: print(f" - {f}") k = len(per_trial) n_ok = per_trial.count(True) # 通过的 trial 数 pass_at = n_ok >= 1 # pass@k：至少一次成功 pass_hat = (n_ok == k) # pass^k：全部成功 results[tid] = { "suite": task["suite"], "task_rate": n_ok / k, "pass@k": pass_at, "pass^k": pass_hat, "avg_tool_score": sum( grade_tool_correctness(task, tr)[1] for tr in trials) / k, } print(f" => per-task 成功率 {n_ok}/{k} = {n_ok/k:.0%} |" f" pass@{k}={'PASS' if pass_at else 'FAIL'} |" f" pass^{k}={'PASS' if pass_hat else 'FAIL'}" f" ({n_ok}/{k})") # ---- 汇总报表（分层，不做全局平均分）---- print("\n" + "=" * 78) print("分层汇总（capability vs regression 分开看）") print("=" * 78) gate_failures = [] for suite, threshold in (("regression", 1.0), ("capability", 0.0)): rows = [(tid, r) for tid, r in results.items() if r["suite"] == suite] if not rows: continue print(f"\n[{suite} suite] 期望 pass^k 通过率 {'≈100%' if threshold else '低(爬坡目标)'}") for tid, r in rows: print(f" {tid:<22} task-rate={r['task_rate']:.0%}" f" pass^k={'OK' if r['pass^k'] else 'NO'}" f" tool-correctness均值={r['avg_tool_score']:.0%}") if suite == "regression" and not r["pass^k"]: gate_failures.append(tid) print("\n" + "=" * 78) if gate_failures: print(f"CI 回归门禁: FAIL —— regression 套件以下任务 pass^k 未达 100%:") for tid in gate_failures: print(f" - {tid}") print("（阻断发布：有东西坏了，先修再上 —— 官方 regression evals 语义）") else: print("CI 回归门禁: PASS —— regression 套件全绿，可发布") return 1 if gate_failures else 0 if __name__ == "__main__": sys.exit(run_eval())