数据截至 (上游 commit d28711d0da26)
执行 harness:一次 trial 的端到端主线
30 秒导读: harness 是 Terminal-Bench 的评测引擎。给它一个任务目录(里面有指令、Docker 配置、测试脚本),它就负责把一个 AI agent 塞进真实的沙箱终端 里干活,然后另起一个会话跑测试,最后解析测试输出、判定这次到底 pass 还是 fail。本章讲清这条从"任务目录"到"一条 pass/fail 结果"的中央控制流——它是整个项目所有齿轮咬合的地方。
本章的位置:01-task-anatomy 讲一个任务长什么样(数据这一半),本章讲引擎怎么消费这份数据。沙箱怎么起(Docker + tmux 的巧劲)留给 03-terminal-tmux;被评测的 agent 内部长什么样留给 04-agents;测试输出怎么解析成分数、各种指标怎么算、断点续跑的锁文件细节留给 05-scoring-results。
1. 这是什么(零基础也能懂)
一句话定义: harness 是一台评测流水线的总控——输入一个任务,输出这个任务被某个 AI agent 做没做出来。
它要解决的问题。 你想公平地测"AI agent 在真实终端里到底行不行"。这件事天生麻烦,因为它不是"调一次模型看回答对不对"那么简单,而是要:
- 给 agent 一个干净、隔离的真实 Linux 环境(不能污染你的机器,也不能让上一个任务的残留影响下一个);
- 让 agent 在里面自由敲命令、装包、改文件,爱 怎么折腾怎么折腾;
- agent 说"我做完了"之后,用一套它碰不到的测试去验收;
- 把整个过程录下来,好复盘"它当时到底干了啥"。
harness 干的就是把这套流程自动化、标准化、还能并发地跑几百个任务。
用起来什么样。 用户几乎不直接碰 Harness 类,而是敲一行 CLI:
# 用 terminus agent + 某个模型,跑整个数据集
tb run --agent terminus --model anthropic/claude-3-7-latest
CLI 在背后就是构造一个 Harness 对象、调它的 .run()。跑完你会在输出目录里得到一棵产物树:每个任务、每次尝试各一个文件夹,里面躺着终端画面快照、会话录像、agent 日志、还有一份 results.json。
一句话直觉。 把 harness 想成一个监考 + 阅卷的机器人:它给每个考生(agent)单独开一间考场(Docker 容器)、发卷子(instruction)、全程录像、时间一到收卷、然后用标准答案(测试脚本)判分。本章 就是跟着这个机器人走一遍完整的监考流程。
2. 顶层全景(它大概怎么转)
2.1 五层调用栈
harness 的主线是一条五层嵌套的调用链,从"跑一整个 benchmark"层层收窄到"跑一次 trial"。先建立这张骨架,后面每一节都是在放大其中一层。
Harness.run() # 一整个 benchmark run(写元数据+锁 → 跑 → 收尾上传)
└─ _execute_tasks() # 并发调度:线程池铺开 所有任务 × n_attempts
└─ _execute_single_trial() # 一个 trial 的外壳:建 TrialHandler、兜底 try/except、落 results.json
└─ _run_trial() # ★ 端到端主线:起容器→agent→测试→解析→判定
└─ _run_agent() # 把 agent.perform_task 包进 asyncio 超时里跑
术语先点破:一次 trial(试次) = "某个任务的第 k 次尝试"。同一个任务可以跑
n_attempts次(为了算 pass@k 这种指标),每一次就是一个独立 trial,有独立的容器、独立的产物目录。
2.2 各层职责一句话
| 层 | 方法 | 干什么 | 位置 |
|---|---|---|---|
| L1 总控 | Harness.run | 非续跑时写 run 元数据 + 锁文件,跑任务,收尾上传 | harness/harness.py:1226 |
| L2 调度 | _execute_tasks | ThreadPoolExecutor 并发跑 len(dataset) × n_attempts 个 trial,边完成边写聚合结果 | harness/harness.py:1099 |
| L3 外壳 | _execute_single_trial | 建 TrialHandler,调 _run_trial,把返回结果写进该 trial 的 results.json;异常兜底成 UNKNOWN_AGENT_ERROR | harness/harness.py:991 |
| L4 主线 | _run_trial | 本章主角:起容器 → agent → 测试 → 解析 → 判定,全程写快照 | harness/harness.py:703 |
| L5 跑 agent | _run_agent | 用 asyncio.wait_for 给 agent.perform_task 套超时,把各种异常翻译成 FailureMode | harness/harness.py:633 |
2.3 一次 trial 的主线(高层,先不进代码)
_run_trial 是整台机器的中心。它的时序是一条直线,读者先记住这七步:
┌──────────────── 一个 Docker 容器的生命周期 ────────────────┐
输入:任务目录 ──▶ ①起沙箱 ──▶ ②开 agent 会话 ──▶ ③放 agent 干活 ──▶ ④(按需)另起 tests 会话
│ spin_up create_session _run_agent create_session
│ _terminal ("agent") (超时包裹) ("tests")
│ │
│ ⑤跑测试 ◀───────────────────────┘
│ _run_tests
└────────────────────────────│──────────────────────────────┘
▼ 容器销毁后才解析
⑥解析测试输出 ──▶ ⑦判定 pass/fail ──▶ 输出:一条 TrialResults
_parse_results _is_resolved
怎么读这张图: 从左到右是时间顺序。①~⑤都发生在同一个容器还活着的时候(with spin_up_terminal(...) as terminal: 块内);⑥⑦在容器已经关掉之后才做——因为解析只需要第⑤步抓下来的那份终端文本,不再需要容器。
一个关键设计先在这里点出:agent 用的会话和跑测试的会话,默认是两个不同的 tmux 会话、甚至不同的用户身份(agent 是配置用户,tests 是 root)。这样 agent 在自己 shell 里设的环境变量、别名不会污染测试环境——除非任务显式要求 run_tests_in_same_shell。这一点第 3.4 节展开。
3. 核心原理(逐个机制,由浅入深)
3.1 并发调度:线程池铺开 任务 × 尝试
要解决的小问题。 数据集可能几百个任务,每个还要跑 n_attempts 次,串行跑要跑到天荒地老。但每个 trial 又是重量级、相互独立的(各自一个 Docker 容器)。
思路。 用一个线程池,把"每个任务 × 每次尝试"都提交成一个 future,最多同时跑 n_concurrent_trials 个。为什么用线程池而不是进程池?因为每个 trial 的重活(起容器、跑命令)都是 I/O 等待型——真正干活的是 Docker 和子进程,Python 这边基本在等,线程足够,还省去了跨进程传对象的麻烦。
并发的形状:
dataset = [taskA, taskB, taskC] n_attempts = 2 n_concurrent_trials = 4
提交 3×2 = 6 个 future 进线程池:
A.1 A.2 B.1 B.2 C.1 C.2
└────┴────┴──┐ 最多 4 个同时在跑,其余排队
▼
ThreadPoolExecutor(max_workers=min(len(dataset), n_concurrent_trials))
│ as_completed:谁先跑完先收谁
▼
每收到一个结果 → 去重/替换 → append 到 BenchmarkResults → 立刻 _write_results 落盘
真实实现。 双重循环提交 future,外层任务、内层尝试:
# harness/harness.py:1125 _execute_tasks
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_task = {}
for task_path in self._dataset:
for attempt in range(1, self._n_attempts + 1):
trial_name = self._get_trial_name(task_path, attempt)
future = executor.submit(
self._execute_single_trial,
trial_name=trial_name,
task_path=task_path,
)
future_to_task[future] = (trial_name, attempt)
max_workers = min(len(self._dataset), self._n_concurrent_trials)(harness.py:1123)——任务比并发数还少时就不浪费线程。
两个易错细节:
- worker 数按任务数算,不按 trial 数算。 上限是
len(dataset),不是len(dataset) × n_attempts。所以就算你把n_attempts调很大,同时在跑的 trial 数仍受任务数封顶。 - 每完成一个就立刻落盘。
as_completed每吐一个结果,就append并_write_results(harness.py:1180)。所以results.json是增量长大的——中途 Ctrl-C 也留得下已完成的部分,这正是续跑能接上的基础。收结果时还做了去重/替换(harness.py:1163):按(task_id, trial_name)判重,重复就覆盖而非追加,防止续跑场景下同一 trial 出现两份。
3.2 起沙箱 + 抓 agent 前画面
要解决的小问题。 每个 trial 都要一个全新、隔离的终端环境,用完即毁,还得把过程录下来。
思路。 用一个 with 上下文管理器 spin_up_terminal 把"起容器"和"关容器"锁成一对——不管中间成功失败,退出 with 块一定会 terminal.stop()(见 terminal/terminal.py:175-179 的 try/finally)。容器起来后,先开一个名叫 "agent" 的 tmux 会话,并趁 agent 还没动手,抓一张终端画面快照存成 pre-agent.txt——留作复盘基线。
真实实现。 主线开头这一段:
# harness/harness.py:717 _run_trial
with spin_up_terminal(
client_container_name=trial_handler.client_container_name,
client_image_name=trial_handler.client_image_name,
docker_image_name_prefix=trial_handler.docker_image_name_prefix,
docker_compose_path=trial_handler.task_paths.docker_compose_path,
...
disable_recording=trial_handler.task.disable_asciinema,
) as terminal:
session = terminal.create_session(
"agent", is_active_stream=self._livestream, as_configured_user=True
)
pre_agent_pane = session.capture_pane(capture_entire=True)
trial_handler.trial_paths.pre_agent_pane_path.write_text(pre_agent_pane)
几个要点:
- 容器名、镜像名从哪来:全由
TrialHandler算好(第 4 节),比如客户端容器名就是 trial 名把.换成-。 create_session("agent", as_configured_user=True):agent 会话用任务配置里指定的用户跑(terminal.py:73-76:读容器Config.User),模拟真实用户而非 root。capture_pane(capture_entire=True)抓的是整屏历史,不只当前可见部分(terminal/tmux_session.py:317)。- tmux 会话到底怎么把命令送进容器、怎么录像,是 03-terminal-tmux 的内容,这里不展开。
3.3 放 agent 干活:超时包裹 + 异常翻译
要解决的小问题。 agent 是被评测的、不可信的外部代码,可能:卡死不返回、跑太久、模型报"上下文超长"、解析响应失败……harness 不能被它拖垮,还得把每种死法归类成一个标准的失败码,好统计。
思路。 两层包裹:
- 超时层:把同步的
agent.perform_task(...)丢进 executor 变成一个 awaitable,再用asyncio.wait_for(..., timeout)套上硬超时。 - 翻译层:外面一圈
try/except,把asyncio.TimeoutError、RetryError(内含各种 LLM 错误)、以及任何兜底Exception,逐一映射成一个FailureMode枚举值。
超时怎么套(教学示意):
# 示意,非源码:把"同步阻塞的 agent 调用"变成"能被超时打断的异步任务"
loop = asyncio.get_event_loop()
task = loop.run_in_executor(None, lambda: agent.perform_task(...)) # 丢到线程里跑
result = await asyncio.wait_for(task, timeout=timeout_sec) # 到点就抛 TimeoutError
对应真实代码在 _run_agent_with_timeout(harness/harness.py:613-631)。超时时长的算法:优先用全局 --global-agent-timeout-sec,否则用任务自己声明的 max_agent_timeout_sec 再乘全局倍率(harness.py:639-645)。
异常 → 失败码 的翻译表(_run_agent,harness.py:633-701):
| 捕获到的情况 | 翻译成的 FailureMode | 说明 |
|---|---|---|
agent.perform_task 正常返回,但结果为 None | UNKNOWN_AGENT_ERROR | agent 没给出结果 |
| 正常返回 | 用 result.failure_mode | agent 自 报的状态(通常 NONE) |
asyncio.TimeoutError | AGENT_TIMEOUT | 超时,但仍构造一个带 markers 的 partial 结果 |
RetryError 内含 ContextLengthExceededError | CONTEXT_LENGTH_EXCEEDED | 上下文超长 |
RetryError 内含 OutputLengthExceededError | OUTPUT_LENGTH_EXCEEDED | 输出超长 |
RetryError 内含 ParseError | FATAL_LLM_PARSE_ERROR | LLM 响应解析失败 |
其它任何 Exception | UNKNOWN_AGENT_ERROR | 兜底 |
(FailureMode 全部取值见 agents/failure_mode.py:4。)
一个关键的"仁慈"设计:agent 超时 ≠ 直接判负。 看主线里超时后的处理:
# harness/harness.py:752 _run_trial
if agent_failure_mode == FailureMode.AGENT_TIMEOUT:
results.failure_mode = agent_failure_mode
self._logger.debug(
f"Agent failed with mode {agent_failure_mode}, continuing"
" with test execution"
)
elif agent_failure_mode != FailureMode.NONE:
results.failure_mode = agent_failure_mode
注意:agent 超时时,harness 记下失败码但不 return,继续往下跑测试。理由很实在——agent 可能在被掐断前其实已经把活干完了,那就该让它拿到分。而其它类型的失败也只是先记下 failure_mode,主线并不在这里中断(真正提前 return 的只有后面测试阶段的失败,见 3.6)。agent 若给了结果,还会把 token 用量记进 results(harness.py:762-764)。
之后再抓一张画面 post-agent.txt(harness.py:749)——这张 + 前面的 pre-agent.txt 一夹,就是"agent 到底在终端里干了啥"的完整可视记录。
3.4 测试会话:默认另起一个 shell
要解决的小问题。 验收测试必须公正——不能被 agent 在自己 shell 里留下的环境变量、别名、当前目录、临时 export 之类污染。
思路。 默认新开一个名叫 "tests" 的 tmux 会话来跑测试,而且用 root 身份, 跟 agent 的会话彻底隔开。只有当任务显式设了 run_tests_in_same_shell = True(比如它就是要测"agent 在 shell 里设的某个变量还在不在"这种 shell 作用域的东西),才复用 agent 那个会话。
# harness/harness.py:766 _run_trial
if not trial_handler.task.run_tests_in_same_shell:
session = terminal.create_session(
"tests", is_active_stream=self._livestream, as_configured_user=False
)
对比两条路径:
| 场景 | 会话 | 用户身份 | 适用 |
|---|---|---|---|
| 默认 | 新开 "tests" 会话 | root(as_configured_user=False) | 绝大多数任务,要干净环境 |
run_tests_in_same_shell=True | 复用 "agent" 会话 | 配置用户 | 测 shell 作用域属性(变量/别名/cwd) |
run_tests_in_same_shell 是任务在 task.yaml 里声明的字段,默认 False(handlers/trial_handler.py:63)。
3.5 跑测试:拷进测试文件,执行 run-tests.sh
要解决的小问题。 测试脚本和测试用例文件在宿主机的任务目录里,容器里没有——得先送进去,再在容器里执行。