diff --git a/cloud/eval_loop.py b/cloud/eval_loop.py new file mode 100644 index 0000000..2c968fc --- /dev/null +++ b/cloud/eval_loop.py @@ -0,0 +1,343 @@ +"""评估闭环 v7:金任务 + 确定性评分 + 两轮对比报告。 + +为什么需要这一轨(查代码确认的缺口): +- 路由影子评估(cloud/routing_shadow.py)只给成本与状态, + 给不了质量;它自己也承认反事实 not_measured——影子建议 + 模型从未跑过,无法比较建议与实际的成本质量。 +- 生产代码里没有金任务、评分器、rubric:harness2.py 明确 + 写着“没有评分器、没有 rubric;delivered 的定义由 adapter + 说了算”。所以“建议模型是不是真的更好”此前没有任何 + 机制能回答,只能凭感觉或凭成本。 +- Foundry 的评测思路是同任务跑不同 harness/模型比通过率, + 本轨把这个思路落成可执行的最小闭环,但刻意不做成 + “自动判优”:报告只给数据与人工评审结论,不激活任何 + 路由、不改任何策略。 + +本轨是什么(纯模块 + CLI,不碰 DB、不跑模型): +- 金任务:一组 JSON 定义,每项 {id, prompt, tier, checks}。 + checks 只认四种确定性检查:contains(结果含某串)、 + not_contains(结果不含某串)、status_completed(任务 + 状态为 completed)、max_tokens、max_cost_usd。校验失败 + (重复 id、未知检查类型、空 checks、空 prompt、非法 + tier)即报错,绝不静默跳过一条检查。 +- 评分:对已完成任务的记录打分,不执行任务。输入记录为 + {task_id, golden_id, model, status, result_text, + total_tokens, usd_micros, unpriced_tokens},其中 usd_micros + 与 cloud/cost.py 同款微美元单位。单项得分 = 通过检查数 / + 已测定检查数;未测定检查(成本数据缺失时的 max_tokens / + max_cost_usd)不计入分母、单独列出,绝不把“没数据” + 当成通过或失败;result_text 为空时字符串检查按空串算, + 是失败不是未测定(输出已知为空)。 +- 对比:两轮(例如模型 A vs 模型 B,或同一模型改 harness + 前后)在同一金任务集上的报告。分三层:每轮总体(任务 + 数、通过率、平均分、平均 token、平均 USD(仅已定价)、 + 未定价任务数)、配对差值(只算两轮都有的金任务,未配 + 对的任务不进差值、单独列出)、逐金任务对照表(每项在 + 两轮里是 pass/fail/unknown/missing)。配对样本 < 20 + 时(与影子评估 MIN_SAMPLES 同阈)报告 ready_for_decision + 为 false 并写明“样本不足,不能据此做激活决定”;报告 + 不算显著性、不给 p 值——小样本下算显著性是编造精度。 + +本轨明确不做、不宣称的事: +- 不跑任务:本轨不创建生产任务、不调模型、不花模型费。 + 金任务要跑起来,需另起一轨(跑任务 + 回填记录),或 + 人工按金任务 prompt 建任务、把已完成记录导成 JSON 再 + 评分。本轨的 CLI 只读两个 JSON 文件,绝不联网、不写库。 +- 不用 LLM 当裁判:没有“质量分”“文笔分”“相关性分”。 + 需要人工判断质量的任务,确定性检查测不出来——那种 + 任务不应硬塞进金任务集,而应走人工评审;把字符串包含 + 当成质量证明是错的。 +- 防作弊警告:contains 检查可被“原样回显 prompt”骗过, + 所以金任务的检查应尽量要求算出结果(例如一个具体数值、 + 一段必须经计算才得到的输出),而不是只要求出现关键词; + 模块无法检测作弊,金任务集本身必须由人 curated,检查 + 写得越像“必须做对才过”,报告才越可信。 +- 不连路由:报告里不出现“建议激活某模型/某档”;激活 + 仍走人工评审 + 策略文件 + 干跑校验那条路,本轨只是给 + 评审多一份质量证据,与影子报告的成本证据并列。 +""" + +import argparse +import json +import os +import re +import sys + +try: + from cloud.models import ROUTING_TIERS +except ImportError: + sys.path.insert(0, os.path.join( + os.path.dirname(os.path.abspath(__file__)), "..")) + from cloud.models import ROUTING_TIERS + +MIN_PAIRED_SAMPLES = 20 +ID_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") +CHECK_TYPES = ("contains", "not_contains", "status_completed", + "max_tokens", "max_cost_usd") + + +def validate_golden_tasks(data): + """校验金任务列表,返回 {id: golden};非法即 ValueError。""" + if not isinstance(data, list) or not data: + raise ValueError("golden tasks must be a non-empty list") + out = {} + for item in data: + if not isinstance(item, dict): + raise ValueError("each golden task must be an object") + gid = item.get("id") + if not isinstance(gid, str) or not ID_RE.match(gid): + raise ValueError( + "golden task id must match [a-z0-9][a-z0-9_-]{0,63}") + if gid in out: + raise ValueError("duplicate golden task id: %s" % gid) + prompt = item.get("prompt") + if not isinstance(prompt, str) or not prompt.strip(): + raise ValueError( + "golden task %s needs a non-empty prompt" % gid) + tier = item.get("tier") + if tier is not None and tier not in ROUTING_TIERS: + raise ValueError( + "golden task %s: tier must be plan, goal, default" + " or null" % gid) + checks = item.get("checks") + if not isinstance(checks, list) or not checks: + raise ValueError( + "golden task %s needs a non-empty checks list" % gid) + clean_checks = [] + for check in checks: + if not isinstance(check, dict): + raise ValueError( + "golden task %s: each check must be an object" % gid) + ctype = check.get("type") + if ctype not in CHECK_TYPES: + raise ValueError( + "golden task %s: unknown check type %r" % (gid, ctype)) + value = check.get("value") + if ctype in ("contains", "not_contains"): + if not isinstance(value, str) or not value: + raise ValueError( + "golden task %s: %s needs a non-empty string" + " value" % (gid, ctype)) + elif ctype in ("max_tokens", "max_cost_usd"): + if not isinstance(value, (int, float)) or \ + isinstance(value, bool) or value < 0: + raise ValueError( + "golden task %s: %s needs a non-negative number" + % (gid, ctype)) + clean_checks.append({"type": ctype, "value": value}) + out[gid] = {"id": gid, "prompt": prompt.strip(), + "tier": tier, "checks": clean_checks} + return out + + +def _check_outcome(check, record): + """单项检查:返回 (outcome, detail),outcome 为 pass/fail/unknown。""" + ctype = check["type"] + value = check.get("value") + if ctype == "status_completed": + if record.get("status") == "completed": + return "pass", "status is completed" + return "fail", "status is %r" % (record.get("status"),) + if ctype in ("contains", "not_contains"): + text = record.get("result_text") or "" + found = value in text + if ctype == "contains": + return ("pass", "found %r" % value) if found else ( + "fail", "missing %r" % value) + return ("pass", "absent %r" % value) if not found else ( + "fail", "present %r (must be absent)" % value) + if ctype == "max_tokens": + tokens = record.get("total_tokens") + if tokens is None: + return "unknown", "no token data" + return ("pass", "%s <= %s" % (tokens, value)) if tokens <= value \ + else ("fail", "%s > %s" % (tokens, value)) + if ctype == "max_cost_usd": + micros = record.get("usd_micros") + if micros is None: + return "unknown", "no priced cost data (unpriced or missing)" + limit_micros = value * 1000000 + return ("pass", "%s micros <= %s" % (micros, limit_micros)) \ + if micros <= limit_micros else ( + "fail", "%s micros > %s" % (micros, limit_micros)) + return "unknown", "unreachable" + + +def score_task(golden, record): + """对一条已完成任务记录打分;返回单项得分 dict。""" + checks = [] + passed = 0 + failed = [] + unknown = [] + for check in golden["checks"]: + outcome, detail = _check_outcome(check, record) + checks.append({"type": check["type"], "value": check.get("value"), + "outcome": outcome, "detail": detail}) + if outcome == "pass": + passed += 1 + elif outcome == "fail": + failed.append(check["type"]) + else: + unknown.append(check["type"]) + measured = passed + len(failed) + score = (passed / measured) if measured else None + # all_pass 的口径写明:全部已测定检查通过。含 unknown 检查 + # 的任务仍可能 all_pass 为真,因此报告里另列 + # fully_measured_n / unknown_tasks_n,读者不得只看通过率 + # 就当成“全部检查都通过”。 + return { + "golden_id": golden["id"], "task_id": record.get("task_id"), + "model": record.get("model"), + "score": score, "passed": passed, "failed": failed, + "unknown": unknown, "checks": checks, + "all_pass": bool(measured) and not failed, + "fully_measured": not unknown, + } + + +def score_run(goldens_by_id, records): + """给一轮记录打分;返回 {scores, unmatched, duplicate_error}。 + + 同一 task_id 出现两次是数据错误,直接 ValueError(绝不 + 静默去重);golden_id 不在金任务集里的记录列入 unmatched、 + 不进统计(评分子集是合法的,但必须点名,不能悄悄丢)。 + """ + seen = set() + scores = [] + unmatched = [] + for record in records or []: + if not isinstance(record, dict): + raise ValueError("each run record must be an object") + task_id = record.get("task_id") + if task_id in seen: + raise ValueError( + "duplicate task_id in run records: %r" % (task_id,)) + seen.add(task_id) + golden = goldens_by_id.get(record.get("golden_id")) + if golden is None: + unmatched.append({"task_id": task_id, + "golden_id": record.get("golden_id")}) + continue + scores.append(score_task(golden, record)) + return {"scores": scores, "unmatched": unmatched} + + +def _run_stats(scored): + scores = scored["scores"] + n = len(scores) + passed_tasks = sum(1 for s in scores if s["all_pass"]) + measured_scores = [s["score"] for s in scores + if s["score"] is not None] + mean_score = (sum(measured_scores) / len(measured_scores)) \ + if measured_scores else None + return { + "n": n, "pass_rate": (passed_tasks / n) if n else None, + "passed_tasks": passed_tasks, "mean_score": mean_score, + "fully_measured_n": sum(1 for s in scores + if s["fully_measured"]), + "unknown_tasks_n": sum(1 for s in scores if s["unknown"]), + "unmatched_n": len(scored["unmatched"]), + } + + +def compare_runs(goldens_by_id, records_a, records_b, + label_a="run_a", label_b="run_b"): + """对比两轮;返回报告 dict(只报告,不做激活决定)。""" + run_a = score_run(goldens_by_id, records_a) + run_b = score_run(goldens_by_id, records_b) + stats_a = _run_stats(run_a) + stats_b = _run_stats(run_b) + by_golden_a = {s["golden_id"]: s for s in run_a["scores"]} + by_golden_b = {s["golden_id"]: s for s in run_b["scores"]} + paired = [] + for gid in sorted(set(by_golden_a) & set(by_golden_b)): + a = by_golden_a[gid] + b = by_golden_b[gid] + paired.append({ + "golden_id": gid, + "a": {"task_id": a["task_id"], "score": a["score"], + "all_pass": a["all_pass"]}, + "b": {"task_id": b["task_id"], "score": b["score"], + "all_pass": b["all_pass"]}, + "score_delta": (b["score"] - a["score"]) + if a["score"] is not None and b["score"] is not None + else None, + }) + paired_n = len(paired) + paired_a_pass = sum(1 for p in paired if p["a"]["all_pass"]) + paired_b_pass = sum(1 for p in paired if p["b"]["all_pass"]) + ready = paired_n >= MIN_PAIRED_SAMPLES + warnings = [] + if not ready: + warnings.append( + "paired samples %d < %d: not enough for a decision, " + "do not activate on this report alone" + % (paired_n, MIN_PAIRED_SAMPLES)) + if run_a["unmatched"] or run_b["unmatched"]: + warnings.append( + "%d + %d run records did not match any golden task and " + "were excluded from stats (see unmatched)" + % (len(run_a["unmatched"]), len(run_b["unmatched"]))) + missing_a = sorted(set(goldens_by_id) - set(by_golden_a)) + missing_b = sorted(set(goldens_by_id) - set(by_golden_b)) + return { + "label_a": label_a, "label_b": label_b, + "run_a": {**stats_a, "unmatched": run_a["unmatched"]}, + "run_b": {**stats_b, "unmatched": run_b["unmatched"]}, + "paired_n": paired_n, + "paired_pass_rate_a": (paired_a_pass / paired_n) + if paired_n else None, + "paired_pass_rate_b": (paired_b_pass / paired_n) + if paired_n else None, + "paired_pass_rate_delta": ( + (paired_b_pass - paired_a_pass) / paired_n) + if paired_n else None, + "paired": paired, + "missing_in_a": missing_a, "missing_in_b": missing_b, + "ready_for_decision": ready, "warnings": warnings, + "notes": [ + "Deterministic checks only; no LLM judge, no quality " + "score. A contains check can be gamed by echoing the " + "prompt, so golden tasks must be human-curated with " + "checks that require actually doing the work.", + "This report never activates routing; activation still " + "goes through human review, policy file, and dry-run.", + ], + } + + +def main(argv=None): + parser = argparse.ArgumentParser( + description="Score two runs against golden tasks (never runs " + "tasks, never activates routing).") + parser.add_argument("--golden", required=True, + help="Golden tasks JSON file (list)") + parser.add_argument("--run-a", required=True, + help="Run A records JSON file (list)") + parser.add_argument("--run-b", required=True, + help="Run B records JSON file (list)") + parser.add_argument("--label-a", default="run_a") + parser.add_argument("--label-b", default="run_b") + args = parser.parse_args(argv) + try: + with open(args.golden, encoding="utf-8") as fh: + goldens = validate_golden_tasks(json.load(fh)) + with open(args.run_a, encoding="utf-8") as fh: + records_a = json.load(fh) + with open(args.run_b, encoding="utf-8") as fh: + records_b = json.load(fh) + if not isinstance(records_a, list) or \ + not isinstance(records_b, list): + raise ValueError("run records must be lists") + report = compare_runs(goldens, records_a, records_b, + label_a=args.label_a, + label_b=args.label_b) + except (OSError, ValueError) as exc: + print("eval error: %s" % exc, file=sys.stderr) + return 2 + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/cloud/tests/test_eval_loop.py b/cloud/tests/test_eval_loop.py new file mode 100644 index 0000000..3aee879 --- /dev/null +++ b/cloud/tests/test_eval_loop.py @@ -0,0 +1,174 @@ +"""评估闭环 v7 单测:金任务校验、确定性评分、两轮对比、CLI。 + +运行(在仓库根目录下): + python3 -m unittest discover -s cloud/tests -p "test_eval_loop.py" -v + +说明:纯函数 + CLI 子进程,不跑模型、不联网、不写库。 +关键断言:未测定检查不计入分母、不当通过或失败; +配对差值只算两轮都有的金任务;配对样本 < 20 时 +ready_for_decision 为 false;重复 task_id 直接报错; +CLI 只读文件,非法输入退出 2。 +""" + +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + +try: + from cloud.eval_loop import compare_runs, score_task + from cloud.eval_loop import validate_golden_tasks +except ImportError: + sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + from cloud.eval_loop import compare_runs, score_task + from cloud.eval_loop import validate_golden_tasks + + +def _goldens(): + return validate_golden_tasks([ + {"id": "g1", "prompt": "算 17*23", "tier": "default", + "checks": [{"type": "status_completed"}, + {"type": "contains", "value": "391"}, + {"type": "max_tokens", "value": 5000}]}, + {"id": "g2", "prompt": "写函数", "tier": "default", + "checks": [{"type": "status_completed"}, + {"type": "not_contains", "value": "TODO"}]}, + ]) + + +def _record(task_id, golden_id, status="completed", + result_text="391", tokens=100, micros=1000): + return {"task_id": task_id, "golden_id": golden_id, + "model": "m", "status": status, + "result_text": result_text, "total_tokens": tokens, + "usd_micros": micros, "unpriced_tokens": 0} + + +class ValidateTestCase(unittest.TestCase): + def test_rejects_bad_golden(self): + for bad in ( + [], + [{"id": "g1", "prompt": "x", "checks": []}], + [{"id": "g1", "prompt": "x", + "checks": [{"type": "judge", "value": 1}]}], + [{"id": "g1", "prompt": "x", + "checks": [{"type": "contains", "value": "a"}]}, + {"id": "g1", "prompt": "y", + "checks": [{"type": "contains", "value": "b"}]}], + [{"id": "g1", "prompt": " ", "checks": [ + {"type": "status_completed"}]}], + ): + with self.assertRaises(ValueError, msg=str(bad)): + validate_golden_tasks(bad) + + +class ScoreTestCase(unittest.TestCase): + def test_all_pass_and_partial(self): + goldens = _goldens() + scored = score_task(goldens["g1"], _record("t1", "g1")) + self.assertTrue(scored["all_pass"]) + self.assertEqual(scored["score"], 1.0) + failed = score_task( + goldens["g1"], + _record("t2", "g1", result_text="392")) + self.assertFalse(failed["all_pass"]) + self.assertAlmostEqual(failed["score"], 2 / 3) + + def test_unknown_excluded_from_denominator(self): + goldens = _goldens() + scored = score_task( + goldens["g1"], + _record("t1", "g1", tokens=None, micros=None)) + # status + contains 通过,max_tokens 未测定:分母 2 不是 3。 + self.assertEqual(scored["score"], 1.0) + self.assertEqual(scored["unknown"], ["max_tokens"]) + # all_pass 的口径是“全部已测定检查通过”,含 unknown + # 的任务仍可能为真,故须同时看 fully_measured。 + self.assertTrue(scored["all_pass"]) + self.assertFalse(scored["fully_measured"]) + + def test_empty_result_is_fail_not_unknown(self): + goldens = _goldens() + scored = score_task( + goldens["g1"], + _record("t1", "g1", result_text=None)) + self.assertIn("contains", scored["failed"]) + + +class CompareTestCase(unittest.TestCase): + def test_paired_delta_and_small_sample_gate(self): + goldens = _goldens() + report = compare_runs( + goldens, + [_record("a1", "g1"), _record("a2", "g2", result_text="ok")], + [_record("b1", "g1", result_text="392"), + _record("b2", "g2", result_text="TODO")], + label_a="A", label_b="B") + self.assertEqual(report["paired_n"], 2) + self.assertEqual(report["paired_pass_rate_a"], 1.0) + self.assertEqual(report["paired_pass_rate_b"], 0.0) + self.assertEqual(report["paired_pass_rate_delta"], -1.0) + self.assertFalse(report["ready_for_decision"]) + self.assertTrue(any("not enough" in w + for w in report["warnings"])) + + def test_unmatched_excluded_and_missing_listed(self): + goldens = _goldens() + report = compare_runs( + goldens, + [_record("a1", "g1"), _record("a9", "nope")], + [_record("b1", "g1")]) + self.assertEqual(report["run_a"]["n"], 1) + self.assertEqual(report["run_a"]["unmatched_n"], 1) + self.assertEqual(report["missing_in_a"], ["g2"]) + self.assertEqual(report["missing_in_b"], ["g2"]) + + def test_duplicate_task_id_raises(self): + goldens = _goldens() + with self.assertRaises(ValueError): + compare_runs( + goldens, + [_record("a1", "g1"), _record("a1", "g2")], + [_record("b1", "g1")]) + + +class CliTestCase(unittest.TestCase): + def test_cli_report_and_bad_input(self): + repo_root = str(Path(__file__).resolve().parents[2]) + with tempfile.TemporaryDirectory() as tmpdir: + golden_path = os.path.join(tmpdir, "golden.json") + run_a_path = os.path.join(tmpdir, "a.json") + run_b_path = os.path.join(tmpdir, "b.json") + with open(golden_path, "w", encoding="utf-8") as fh: + json.dump([{"id": "g1", "prompt": "算 17*23", + "checks": [{"type": "contains", + "value": "391"}]}], + fh) + with open(run_a_path, "w", encoding="utf-8") as fh: + json.dump([_record("a1", "g1")], fh) + with open(run_b_path, "w", encoding="utf-8") as fh: + json.dump([_record("b1", "g1", result_text="392")], fh) + good = subprocess.run( + [sys.executable, "-m", "cloud.eval_loop", + "--golden", golden_path, "--run-a", run_a_path, + "--run-b", run_b_path, "--label-a", "A", + "--label-b", "B"], + cwd=repo_root, capture_output=True, text=True, + timeout=30) + self.assertEqual(good.returncode, 0, good.stderr) + report = json.loads(good.stdout) + self.assertEqual(report["paired_n"], 1) + self.assertEqual(report["paired_pass_rate_a"], 1.0) + with open(golden_path, "w", encoding="utf-8") as fh: + json.dump([{"id": "g1", "prompt": "x", "checks": []}], + fh) + bad = subprocess.run( + [sys.executable, "-m", "cloud.eval_loop", + "--golden", golden_path, "--run-a", run_a_path, + "--run-b", run_b_path], + cwd=repo_root, capture_output=True, text=True, + timeout=30) + self.assertEqual(bad.returncode, 2) diff --git a/eval/README.md b/eval/README.md new file mode 100644 index 0000000..1c9cc4e --- /dev/null +++ b/eval/README.md @@ -0,0 +1,22 @@ +# eval/ — 评估闭环的金任务示例(格式说明,不是已审定的金任务集) + +`golden-tasks.example.json` 只演示 `cloud/eval_loop.py` 接受的格式: +每项 `{id, prompt, tier, checks}`,检查只认 contains / not_contains / +status_completed / max_tokens / max_cost_usd 五种确定性检查。 + +它不是一份可直接用来做激活决定的金任务集:里面的检查很弱 +(例如只要求出现一个数值或一个函数名),容易被回显骗过。 +真实金任务集必须由人逐条审定:检查要尽量要求“必须做对才过” +(算出的具体结果、必须经计算才得到的输出),而不是只要求 +出现关键词;需要人工判断质量的任务不要硬塞进金任务集。 + +用法(只评分已完成任务的记录,不跑任务、不花模型费): + + python3 -m cloud.eval_loop --golden eval/golden-tasks.example.json \ + --run-a run-a.json --run-b run-b.json \ + --label-a model-a --label-b model-b + +run 文件是记录列表,每条 `{task_id, golden_id, model, status, +result_text, total_tokens, usd_micros, unpriced_tokens}`, +其中 usd_micros 与 cloud/cost.py 同款微美元单位;成本数据 +缺失时相关检查计为 unknown,不计入分母。 diff --git a/eval/golden-tasks.example.json b/eval/golden-tasks.example.json new file mode 100644 index 0000000..f5b0925 --- /dev/null +++ b/eval/golden-tasks.example.json @@ -0,0 +1 @@ +[{"checks":[{"type":"status_completed"},{"type":"contains","value":"391"},{"type":"max_tokens","value":5000}],"id":"eval-001","prompt":"Compute 17 * 23 and return only the number.","tier":"default"},{"checks":[{"type":"status_completed"},{"type":"contains","value":"def sum_of_squares"},{"type":"not_contains","value":"TODO"}],"id":"eval-002","prompt":"Write a Python function sum_of_squares(n) that returns the sum of squares from 1 to n, with no other output.","tier":"default"}] \ No newline at end of file