Spaces:
Sleeping
Sleeping
File size: 5,179 Bytes
5657821 474cdf7 5657821 474cdf7 5657821 9ce4a24 5657821 9ce4a24 5657821 0842437 5657821 9ce4a24 5657821 9ce4a24 5657821 | 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 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | """CLI entrypoint to run the benchmark and write reproducible result artifacts.
Runs :class:`BenchmarkRunner` over configurable tasks, personas, seeds, and
agents, then writes ``results.json``, ``results.csv``, and ``results.html`` to an
output directory.
The deterministic ``baseline`` and ``multiagent`` agents run without any API key.
The ``llm`` agent requires ``OPENAI_API_KEY`` and is therefore opt-in: it is only
included when explicitly requested via ``--agents``.
Example::
python scripts/run_benchmark.py --agents baseline multiagent --out reports/out
"""
from __future__ import annotations
import argparse
import os
import sys
# When invoked as ``python scripts/run_benchmark.py`` the script's own directory
# (not the repo root) lands on sys.path, so the ``benchmark`` package is not
# importable. Add the repo root explicitly to keep the entrypoint runnable from
# anywhere without installing the package.
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _REPO_ROOT not in sys.path:
sys.path.insert(0, _REPO_ROOT)
from benchmark.results_report import write_results_report # noqa: E402
from benchmark.runner import ( # noqa: E402
DEFAULT_PERSONAS,
DEFAULT_SEEDS,
DEFAULT_TASKS,
BenchmarkResult,
BenchmarkRunner,
)
# Agents that need no API key and run deterministically offline.
OFFLINE_AGENTS = ("baseline", "multiagent", "reflective")
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description="Run the executive email benchmark and write result artifacts.",
)
parser.add_argument(
"--tasks",
nargs="+",
default=list(DEFAULT_TASKS),
help="Task ids to run (default: all built-in tasks).",
)
parser.add_argument(
"--personas",
nargs="+",
default=list(DEFAULT_PERSONAS),
help="Personas to run (default: all built-in personas).",
)
parser.add_argument(
"--seeds",
nargs="+",
type=int,
default=list(DEFAULT_SEEDS),
help="Seeds to run (default: %(default)s).",
)
parser.add_argument(
"--agents",
nargs="+",
choices=("baseline", "multiagent", "reflective", "llm"),
default=list(OFFLINE_AGENTS),
help=(
"Agents to run (default: baseline multiagent). The 'llm' agent needs "
"OPENAI_API_KEY and is only run when explicitly selected."
),
)
parser.add_argument(
"--out",
default="reports/benchmark",
help="Output directory for results.json/.csv/.html (default: %(default)s).",
)
parser.add_argument(
"--max-steps",
type=int,
default=100,
help="Maximum steps per episode (default: %(default)s).",
)
parser.add_argument(
"--record-history",
action="store_true",
help="Append a per-agent run summary (+deltas vs the prior run) to the history file.",
)
parser.add_argument(
"--history-path",
default="benchmark/leaderboard_history.jsonl",
help="History file for --record-history (default: %(default)s).",
)
return parser
def run(
tasks: list[str],
personas: list[str],
seeds: list[int],
agents: list[str],
out_dir: str,
max_steps: int = 100,
record_history: bool = False,
history_path: str = "benchmark/leaderboard_history.jsonl",
) -> list[BenchmarkResult]:
"""Run the selected agents and write artifacts to ``out_dir``."""
# The benchmark is sim-only by construction: refuse to run if a real-inbox
# connector is enabled, so real mail can never contaminate benchmark results.
from env.connectors import email_connector_enabled
if email_connector_enabled():
raise RuntimeError(
"EMAIL_CONNECTOR_ENABLED is set; the benchmark is sim-only and refuses to "
"run with a real-inbox connector enabled. Unset it to run the benchmark."
)
runner = BenchmarkRunner(
tasks=tasks,
personas=personas,
seeds=seeds,
max_steps=max_steps,
)
results: list[BenchmarkResult] = []
for agent_name in agents:
results.extend(runner.run_agent(agent_name))
artifacts = write_results_report(results, out_dir)
print(f"Wrote {len(results)} results across {len(agents)} agent(s) to {out_dir}")
for name, path in artifacts.items():
print(f" {name}: {path}")
if record_history:
from benchmark.history import append_run
entry = append_run(history_path, results, label=",".join(agents))
print(f" history: appended run to {history_path} (deltas: {entry['deltas']})")
return results
def main(argv: list[str] | None = None) -> int:
parser = build_parser()
args = parser.parse_args(argv)
run(
tasks=args.tasks,
personas=args.personas,
seeds=args.seeds,
agents=args.agents,
out_dir=args.out,
max_steps=args.max_steps,
record_history=args.record_history,
history_path=args.history_path,
)
return 0
if __name__ == "__main__":
raise SystemExit(main())
|