Files
Agentswarm/scripts/run-benchmark-suite.py
FastheiandClaude Opus 4.8 b5cc68c977 feat(benchmark): 落地自证采集器(阶段0+1)— S_gain≡G_E 接通 S_swarm + leaderboard
阶段0(定口径,docs/benchmark/emergence-evaluation.md §6 v2.1-impl):
- S_gain ≡ G_E(差值,不强制归一 [0,100],与标准「见涌现增益」字面一致)。
- 聚合 S_gain 取对最强基线(Q_base 最大)的 G_E(最保守,避免挑弱基线虚高)。
- Q ≡ Q_quality;swarm_valid 仍要求对全部基线 G_E>0 且 G_E,c>0。

阶段1(采集器):
- 新增 benchmark/collectors/selfcert_collector.py:把套件 5 份 BenchmarkRunRecord
  (swarm+4基线)+ 可选活体 SwarmMetrics 合流,经 baselines.compare 算 G_E/G_E,c,
  补全 run_collector 无法自算的 s_gain/g_e/g_e_cost/s_swarm,可能时产出 Benchmark_Agent。
- benchmark/leaderboard:实现排行榜聚合+渲染(标准 §11 字段)。
- run-benchmark-suite.py 接入自证 + leaderboard 输出。

诚实纪律(组织规则 #9):缺真实输入一律 NaN+coverage False,不伪造。
- O(可观测性)标准无公式 → 恒 NaN;Gov 计数器未实现 → 无活体治理则 NaN。
- 故完整 Benchmark_Agent 数字仍待 O 公式 + Gov 计数器(阶段2),采集器明列缺口。

验证:新增 test-benchmark-selfcert.py(17 项)+ 现有 benchmark 测试(metrics/
collector/comparison/runners)+ offline suite + 契约冒烟(runtime/merge/freeze)全 PASS。

影响范围:仅 agent_swarm benchmark 模块 + docs;不改 Manager↔Swarm 契约/计费/审计/密钥/发布链路。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-12 17:38:19 +08:00

79 lines
3.5 KiB
Python

"""Benchmark suite harness (#13/#21/#22): run a task set through all 5 systems → report.
Default backend is OFFLINE (hermetic, key-free) — it validates the whole pipeline but, by design,
shows NO swarm quality advantage (every system gets the same reference solution). REAL G_E numbers
require `--backend openai` with OPENAI_API_KEY set and a sufficiently large, frozen task set.
Grading executes generated code in the fail-closed sandbox, so HEICODE_SANDBOX_ISOLATED must be
set (only inside an isolated pod / ephemeral CI runner — see security-boundary §8.1).
Usage:
HEICODE_SANDBOX_ISOLATED=1 python scripts/run-benchmark-suite.py --taskset coding-set-1 \
[--backend offline|openai] [--archive runs/<label>] [--run-label <label>]
"""
import argparse
import sys
from pathlib import Path
# Markdown rendering uses ✅/🔴/⚠️; force UTF-8 so it prints on a Windows GBK console too.
try:
sys.stdout.reconfigure(encoding="utf-8")
except Exception:
pass
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from benchmark.tasksets import load_taskset, available_tasksets
from benchmark.runners import run_all, OfflineBackend, OpenAIBackend
from benchmark.reports import build_report, render_markdown
from benchmark.replay import save_archive
from benchmark.collectors.selfcert_collector import SelfCertCollector
from benchmark.leaderboard import build_leaderboard, render_markdown as render_leaderboard
def main() -> int:
ap = argparse.ArgumentParser(description="Run the benchmark suite (swarm + 4 baselines).")
ap.add_argument("--taskset", default="coding-set-1", help=f"one of: {available_tasksets()}")
ap.add_argument("--backend", choices=["offline", "openai"], default="offline")
ap.add_argument("--n-runs", type=int, default=1)
ap.add_argument("--archive", default=None, help="directory to write the replay archive into")
ap.add_argument("--run-label", default="run-1", help="archive subdir name (deterministic)")
args = ap.parse_args()
taskset = load_taskset(args.taskset)
if args.backend == "openai":
backend = OpenAIBackend(price_per_1k_tokens=0.0006)
strong_backend = OpenAIBackend(model="gpt-4o", price_per_1k_tokens=0.005) # bigger model
else:
backend = OfflineBackend(cost_per_call=0.002)
strong_backend = OfflineBackend(cost_per_call=0.01) # "stronger" = costlier
records = run_all(taskset, backend, strong_backend=strong_backend)
report = build_report(records, backend_name=args.backend, n_runs=args.n_runs)
print(render_markdown(report))
# Self-certification: wire baselines into SwarmMetrics (S_gain ≡ G_E) → leaderboard.
# Without a live run, collaboration/communication/governance are NaN, so S_swarm and
# Benchmark_Agent stay honestly NaN (gaps printed) — this is the pipeline view.
selfcert = SelfCertCollector(records).collect()
print("\n" + render_leaderboard(build_leaderboard([selfcert])))
if selfcert.gaps:
print("\nBenchmark_Agent 未产出,缺口:")
for g in selfcert.gaps:
print(f" - {g}")
if args.archive:
path = save_archive(args.archive, args.run_label, records=records, report=report)
print(f"\narchived → {path}")
if args.backend == "offline":
print("\nNOTE: offline backend — pipeline validation only, NOT a real comparison "
"(no swarm advantage is or should be shown). Use --backend openai for real G_E.")
return 0
if __name__ == "__main__":
sys.exit(main())