数据截至 (上游 commit d8fc8fcbde74)
评测主循环:从 eval() 到一条样本打完分
30 秒导读: 你调一次
eval(task, model=...),Inspect 会把这个"任务"逐层拆开——先拆成"任务 × 模型"若干执行单元,再拆成"数据集 × epoch"若干条样本,最后每条样本独立走一遍:setup → 一串 solver → 一串 scorer → 写日志。本章就讲清这条从顶层 入口到"一条样本打完分"的主线,它是理解其余各章的地基。
本章是 Inspect AI 全景 下的骨架章。读完你应该能在脑子里画出:一次 eval() 到底流经哪几个函数、每一层负责切分什么、一条样本从生到死经历哪些步骤。至于每一步内部怎么实现——solver 协议(→02)、模型怎么生成(→03)、scorer 怎么聚合成指标(→05)——留给后续各章,本章只把主干打通。
1. 一句话骨架:一个四层漏斗
先给最顶层的心智模型。整个评测就是一个漏斗:上面是你写的一个 Task,下面是成百上千次"一条样本的完整评测",中间靠四层函数逐层拆分和调度。
eval() 同步入口 — 启动 anyio 事件循环,其余全是 async
└─ eval_async() 解析 model / tasks / 各种 config,建好日志 recorder
└─ eval_run() 编排一批任务:装 logger、起 sandbox、按并发调度
└─ task_run() 单个任务:切数据集、装 plan+scorer、按 (epoch × 样本) 铺开
└─ task_run_sample() 单条样本的一生:setup→solver→scorer→写日志
四层各自"拆掉一个维度",职责清清楚楚:
| 层 | 函数(符号) | 拆掉的维度 | 产出 |
|---|---|---|---|
| L1 同步壳 | eval | 无(只负责起事件循环) | list[EvalLog] |
| L2 解析装配 | eval_async | 把 tasks × model 解析成一组 ResolvedTask | 交给 L3 编排 |
| L3 编排调度 | eval_run → run_multiple | 任务 × 模型 → 并发执行单元 | 每个单元一份 EvalLog |
| L4 单任务 | task_run | 数据集 × epoch → 一条条样本 | 一份 EvalLog(含所有样本 + 指标) |
| L5 单样本 | task_run_sample | 无(最内核) | 一条样本的分数 dict[str, SampleScore] |
依据:src/inspect_ai/_eval/eval.py:eval(109)、eval_async(392);src/inspect_ai/_eval/run.py:eval_run(101)、run_multiple(501);src/inspect_ai/_eval/task/run.py:task_run(324)、task_run_sample(1037)。
记住这张漏斗图,后面每一节都是在放大其中一层。
2. Task:一次评测的完整定义
漏斗最上面那个东西,是一个 Task。它不是"跑一次"的动作,而是"要跑什么"的完整声明——数据、怎么解、怎么打分,全打包在一个对象里。
2.1 Task 里装了什么
Task.__init__ 的参数很多,但可以归成四组。理解这四组,就理解了一次评测需要你交代清楚的四件事。
| 组 | 关键字段 | 白话 |
|---|---|---|
| 数据 | dataset | 要评的样本集(见 2.3) |
| 怎么解 | setup / solver | solver 是主求解链(默认 generate(),即"就调一次模型");setup 是即使替换了 solver 也照跑的前置步骤 |
| 怎么打分 | scorer / metrics / epochs | scorer 给每条样本打分,epochs 决定每条样本重复几遍、分数怎么归并 |
| 护栏与环境 | model / sandbox / fail_on_error / 各种 *_limit | 默认模型、沙箱、容错阈值、每样本的消息/令牌/时间等上限 |
构造函数把这些逐一存成实例属性,途中做一些规整(resolve):数据序列包成 Dataset、solver 列表串成一条 chain、epochs 整数升格成 Epochs 对象。
# 示意,非源码 —— Task 构造时对入参的规整
self.dataset = resolve_dataset(dataset) # list[Sample] → MemoryDataset
self.solver = resolve_solver(solver) # list[Solver] → chain(...);Agent → as_solver
self.scorer = resolve_scorer_metrics(...) # 把自定义 metrics 挂到 scorer 的注册信息上
epochs = resolve_epochs(epochs) # int → Epochs(int)
真实实现:src/inspect_ai/_eval/task/task.py:Task.__init__(67),属性赋值集中在 179-209 行;规整用的辅助函数 resolve_dataset(461)、resolve_solver(474)、resolve_epochs(453) 都在同文件下方。重点看:Task 只是把配置存起来,不跑任何东西——真正跑是 L4/L5 的事。
附带一提:
task_with(...)(task.py:246)能就地改写一个已有 Task 的某些字段并返回它,常用于给同一个任务生成多个变体。
2.2 @task:让 任务能被名字找到
光有 Task 类还不够。Inspect 要能在命令行里写 inspect eval mytask.py@my_task 用名字找到并实例化任务,这靠 @task 装饰器完成注册。
@task 包住你的任务工厂函数,做三件事:
- 登记:把函数按名字加进全局 registry(
task_register),名字默认取函数名。 - 打标:每次调用工厂产出
Task实例后,给实例贴上注册信息 + 全部入参(用于日志复现)。 - 定位:若任务来自本地文件而非安装包,记录源文件路径和运行目录(供
chdir和日志用)。
# 示意,非源码 —— @task 的核心骨架
def task(func):
@wraps(func)
def wrapper(*args, **kwargs):
instance = func(*args, **kwargs) # 得到 Task
registry_tag(func, instance, info, ...) # 贴注册信息 + 入参
setattr(instance, TASK_FILE_ATTR, ...) # 记住来自哪个文件
return instance
return task_register(wrapper, name, ...) # 按名字登记
真实实现:src/inspect_ai/_eval/registry.py:task(98) 装饰器、task_register(32) 登记、task_create(57) 按名字实例化(eval("file.py@name") 这条路走它)。
2.3 数据模型:Sample / Dataset / MemoryDataset
漏斗最终拆到的原子是 Sample——一条评测样本。它是个 Pydantic 模型,字段一目了然:
| 字段 | 含义 |
|---|---|
input | 喂给模型的输入(字符串或一串 ChatMessage) |
target | 理想答案(用于打分;可以是字面值,也可以是给模型评委看的叙述) |
choices | 选择题的选项(仅多选评测用) |
id / metadata | 唯一标识 / 任意附加数据 |
sandbox / files / setup | 该样本专属的沙箱、随附文件、初始化脚本 |
依据:src/inspect_ai/dataset/_dataset.py:Sample(29)。
Dataset 是 Sample 的序列——一个抽象基类,规定了 __getitem__ / sort / filter / shuffle 等接口(_dataset.py:Dataset,143)。最常用的实现是 MemoryDataset:就是把一个 list[Sample] 端在内存里,顺序访问(_dataset.py:MemoryDataset,255)。你直接传 list[Sample] 给 Task 时,resolve_dataset 就把它包成 MemoryDataset。
2.4 Epochs:每条样本重复几遍
Epochs 回答两个问题:每条样本跑几遍(epochs),以及多遍的分数怎么合成一个(reducer,默认取平均 "mean")。
它本身很薄——存一个整数和一个 reducer 规格,reducer 惰性创建:
依据:src/inspect_ai/_eval/task/epochs.py:Epochs(4)。为什么要重复?同一条样本在有随机性的模型上多跑几遍,分数更稳。归并发生在最后聚合阶段(→05),本章只需知道:epochs 是一个"把样本数 × N"的乘数——见下面 task_run 里 total_samples = len(dataset) * epochs。
3. 顶层入口:eval → eval_async → eval_run
现在从"定义"转到"执行"。你调 eval(...),发生了什么?
3.1 eval:同步的壳,只为起事件循环
eval 本身几乎不干活。它的正事只有一件:把整个异步流程塞进事件循环里跑起来,并把结果同步地还给你。
真正的逻辑写在内嵌的 run_task_app() 里(它 await eval_async(...)),然后交给显示层用 task_display().run_task_app(...) 驱动执行:
依据:src/inspect_ai/_eval/eval.py:eval(109);内嵌协程 run_task_app 定义于 304 行,在 372 行被 run_task_app(with_async_fs(run_task_app)) 启动。所以同步/异步的分界就在这一行:上面是普通函数调用,下面全是 async。
3.2 eval_async:解析与装配
eval_async 是"准备阶段"的总管。它按顺序把你给的各种松散入参解析成能执行的形态:
eval_async 干的活(顺序):
eval_init(...) 解析 model,初始化子进程/日志上下文
resolve_task_source(...) tasks 若是 TaskSource 则识别出来
eval_resolve_tasks(...) 把 tasks × model 展开成 list[ResolvedTask]
── 若没解析出任何任务 → 直接报错 ──
create_recorder_for_format(...) 按日志格式建 recorder,确认目录可写
组装 EvalConfig(limit / epochs / fail_on_error / 各 limit ...)
run_batches(resolved_tasks) 分批把任务交给 eval_run
依据:src/inspect_ai/_eval/eval.py:eval_async(392);eval_init(748 调用点)、eval_resolve_tasks(770 调用点)、EvalConfig 组装(862)、run_batches/run_batch 内嵌定义(989-1023)。
这里有个值得注意的设计:任务可以在运行中动态追加。run_batches 是个循环——先跑种子任务,再跑运行期通过 enqueue_task(命令式)或 TaskSource.next_tasks()(声明式)加进来的任务,直到没有新任务(eval.py:1023-1040)。多数普通评测里,这个循环只转一圈。
一个 ResolvedTask = 一个 (任务定义 × 一个具体模型) 的执行单元。多模型评测就是同一个任务展开成多个 ResolvedTask。
3.3 eval_run:编排与并发调度
eval_run 拿到一批 ResolvedTask,负责把它们真正跑起来,并管住资源和并发。它分两步:
第一步,prepare_options——逐任务做运行前准备:给每条样本补上 id(从 1 开始)、校验 id 唯一、需要沙箱的任务先做沙箱启动预热、把任务自带的 epochs/各 limit/fail_on_error 广播进 eval 级配置(前提是没覆盖掉命令行/eval() 显式给的值)。产出一批 TaskRunOptions。
依据:src/inspect_ai/_eval/run.py:eval_run(101)、内嵌 prepare_options(150),样本 id 补齐(157-162)、配置广播(196-220)。
第二步,交给调度器并发跑。普通情况走 run_multiple;开了任务级重试则走 run_task_retry_attempts。parallel 参数是并发上限(同时在跑的"任务 × 模型"单元数),调度器会在多个模型间摊平工作。
依据:run.py:336-377(分派)、run_multiple(501)。
每个执行单元被包进 _run_task——它给每个任务开一个独立的 cancel scope,这样一个任务自己取消不会波及兄弟任务,最终在里面 await task_run(options, ...):
依据:run.py:_run_task(446),独立取消域(464-486),调用 task_run(484)。
至此,漏斗走到了单个任务。
4. task_run:把一个任务铺开成 N 条样本
task_run 负责一个任务(在一个模型上)的完整执行。它的核心工作是:把"数据集 × epochs"这个二维网格,铺成一串独立的样本执行,并发跑完,再聚合成结果。
4.1 准备:切数据集、装 plan 与 scorer
进入 task_run 后,先把执行所需的组件一一就位:
| 步骤 | 做什么 | 符号 / 位置 |
|---|---|---|
| 切数据集 | 按 limit / sample_id 截取要跑的样本子集 | slice_dataset,run.py:387 |
| 算总量 | total_samples = len(dataset) * epochs | run.py:388 |
| 解析 plan | 把 solver(+ setup)展开成一条可执行的 Plan | resolve_plan,run.py:437 |
| 解析 scorer | 取任务的 scorer 列表,算好唯一名字 | run.py:440-449 |
| 建信号量 | 限制"同时在跑的样本数"的并发闸门 | create_sample_semaphore,run.py:556 |
resolve_plan 是"怎么解这条样本"的最终成形:它把 solver / solver 链 / Plan 统一成一个 Plan 对象,并且——关键——把 task.setup 的步骤拼到 solver 步骤前面。所以 setup 永远先跑,哪怕你在 eval() 里换掉了主 solver。
依据:src/inspect_ai/_eval/task/run.py:resolve_plan(279),setup 前置拼接在 289-297 行(注意它用浅拷贝避免重复拼接)。
4.2 铺开:按 (epoch × 样本) 并发跑
组件就位后,task_run 用一个笛卡尔积把所有 (样本, epoch) 组合列出来,交给 tg_collect 并发执行。每个组合调一次内嵌的 run_sample,后者最终落到 task_run_sample:
# 示意,非源码 —— task_run 里的样本铺开
sample_results = await tg_collect([
functools.partial(run_sample, sample_index, epoch)
for epoch in range(1, epochs + 1) # 每个 epoch
for sample_index in range(len(sample_store)) # × 每条样本
])
# 重点看:这就是漏斗最宽的那一层 —— total_samples 个并发单元
真实实现:run.py:752-758。run_sample 内嵌定义在 615 行,它先查"上一次评测是否已有这条样本的缓存结果"(可复用则直接跳过重跑,run.py:627-670),否则准备一个惰性物化 sample+state 的工厂 create_sample_state(683),再调 task_run_sample(708)。
"惰性物化"是省内存的关键:样本和
TaskState不在铺开时就全部创建,而是等真正拿到并发名额、进了task_run_sample才deepcopy出来。这样内存占用是 O(并发样本数) 而非 O(总样本数 × epochs)。见 run.py:384-387 的注释与 686-706 的工厂体。
4.3 收口:聚合分数、写完日志
所有样本跑完,tg_collect 收回一个结果列表(每项是该样本的分数字典,或提前停止标记,或 None=出错未打分)。task_run 把成功的分数交给 eval_results 聚合成指标(按 epoch 用 reducer 归并、算 metric),然后 finish_task_log 落盘,产出这个任务的 EvalLog。
依据:run.py:783-813;聚合 eval_results(784)、收尾 finish_task_log(808,内部转调 _finish_task_log,2225)。聚合的细节属于 05,这里只需知道它是漏斗收口的一步。
失败判定也在这:task_run 用 SampleErrorHandler 累计的错误数,对照 fail_on_error 阈值决定这份日志标 "success" 还是 "error"(run.py:802-809)。
5. task_run_sample:一条样本的完整一生
这是漏斗的最内核,也是本章的重头。一条样本从生到死的每一步都在这里。先看全貌:
task_run_sample(一条样本):
① async with semaphore 抢并发名额(限流闸门);抢到才往下
② create_sample_state() 惰性物化 Sample + TaskState(deepcopy)
③ 初始化上下文 transcript(事件流)/ store / 打分上下文 / 沙箱 CM
④ [span "init"] 起 sandbox 需要沙箱则在此创建
⑤ [span "solvers"] plan(state) ★ 跑 setup + solver 链 —— 样本的"解题"
⑥ [span "scorers"] 逐个 scorer ★ 对最终 state 打分
⑦ log_sample(...) 把这条样本(含事件、分数、错误)写进日志
⑧ 记账并返回 成功→返回分数;出错→重试 / 抛错 / 静默返回 None
这条链对应任务提示里那句"setup → solver 链 → scorer → 写日志如何落在一条样本上"。逐步拆:
5.1 限流闸门:先抢名额,再干活
函数体第一行就是 async with semaphore:——没抢到并发名额的样本在这里排队等着,不会白占内存(因为物化在抢到之后才做)。这个信号量由 create_sample_semaphore 按配置决定容量。
依据:src/inspect_ai/_eval/task/run.py:task_run_sample(1037),信号量入口(1094),物化紧随其后(1096)。
create_sample_semaphore 的容量三选一(优先级从高到低):
| 情况 | 容量 |
|---|---|
显式给了 max_samples | 就用它(anyio.Semaphore(max_samples)) |
| 开了自适应连接数 | DynamicSampleLimiter,并发随模型实际并发动态涨落 |
| 都没有(静态默认) | 取 max_connections,退而取模型 API 的默认连接上限 |
依据:run.py:create_sample_semaphore(2059)。默认情况下,样本并发 ≈ 模型连接并发——这样"每条样本都在等模型回复"时不会有大量样本空转占资源。
5.2 搭上下文:transcript / store / 打分上下文
抢到名额、deepcopy 出 sample 和 state 后,task_run_sample 给这条样本搭好一套隔离的执行上下文:事件记录器 Transcript(样本里发生的每件事都记进去,→06)、子任务存储 store、打分上下文,以及(若需要)沙箱的上下文管理器。
依据:run.py:1114-1150。这里每条样本各有各的 transcript/store,互不干扰——这是"样本级隔离"的基础 。
5.3 解题:跑 plan(setup + solver 链)
核心的一行,藏在 init 和 solvers 两个 span 里:
async with span("solvers"):
state = await plan(state, generate)
依据:run.py:1332-1333。这里 plan 就是 4.1 里 resolve_plan 产出的对象;调用它会依次跑每个 step(先 setup,再 solver 链),每一步拿到上一步的 TaskState、产出新的 TaskState。generate 是框架注入给 solver 的"调模型"函数。
Plan 内部就是一个顺序循环:for index, solver in enumerate(self.steps): state = await solver(state, generate)(src/inspect_ai/solver/_plan.py:Plan.__call__,93-105)。solver 与 TaskState 的协议细节是 02 的主题,本章到"它是顺序跑一串 solver"为止。
跑完后 state.completed = True(run.py:1477),这条样本的"解题"就结束了,拿到最终的 TaskState。
5.4 打分:逐个 scorer 给最终 state 打分
有了最终 state,进入 scorers span,逐个 scorer 调用 scorer(state, Target(sample.target)),把返回的分数记进 state.scores 和结果字典 results,同时往 transcript 里发 ScoreEvent。
依据:run.py:1510-1567。注意一个防呆:若某个 scorer 偷偷改了 state.scores 里同名的项,会直接 RuntimeError(1532-1535)。
打分默认只在样本成功时进行;但若开了 score_on_error 且已无重试余量,出错的样本也会被打分(run.py:1505-1509)。scorer 怎么组织、分数怎么归并成指标,见 05。
5.5 落盘:log_sample 写日志
打完分,task_run_sample 组装出一个 EvalSample(含开始时间、样本、最终 state、分数、错误、限流信息、重试历史),交给 log_sample 写进日志。
依据:run.py:make_eval_sample(1650)、log_sample 调用(1683)、log_sample 定义(1865)。
log_sample 有个性能分叉,值得记一笔:
- 常路(从内存直接写):实时缓冲关闭,或事件仍全部驻留在内存里——直接把内存中的样本写盘,省掉"从 SQLite 缓冲把每个事件读回来再校验一遍"的开销。
- 回退(从缓冲流式写):事件因体量太大被逐出内存——只能从缓冲 DB 把历史流式读回来落盘。
依据:run.py:1679-1690(选择逻辑)、log_sample(1865-1898)。日志/事件流的完整机制在 06。
5.6 记账与返回
写完日志,函数走到末尾的分支,决定这条样本的"结局":
末尾分支(run.py:1716-1806):
有 error 且还有重试余量且非取消
→ 记 attempt 结束,递归调用自己(retry_on_error - 1),排到队尾重跑
被取消(cancelled)
→ 记为 cancelled(非错误,计入总数好让 eval 收尾),重新抛出取消
无 error
→ 回调 sample_complete,记 completed,返回 results(分数字典)
有 error 且该抛
→ 记 errored,抛出
有 error 但不该抛
→ 记 errored,静默返回 None
依据:run.py:1716(重试递归,注意它在信号量之外重试,所以重跑会排到样本队列末尾)、1775(取消)、1785(成功)、1795/1802(错误)。
6. 三个容错旋钮:fail_on_error / limit / retry
主线之外,一条样本"出岔子"时的行为由几个正 交的旋钮决定。它们容易混,列表说清:
| 旋钮 | 管什么 | 取值语义 | 落点 |
|---|---|---|---|
fail_on_error | 整个 eval 何时判失败 | True=首个样本错就失败;False=从不;0~1=错误比例超阈值才失败;>1=错误计数超阈值才失败 | SampleErrorHandler,run.py:423;终判 run.py:802-809 |
retry_on_error | 单条样本出错重试几次 | 整数 N;重试排到样本队列末尾 | task_run_sample 递归,run.py:1716-1772 |
score_on_error | 出错样本是否仍打分 | True=出错也打分(而非中断);错误仍计入 fail_on_error 阈值 | run.py:1180-1183、1505-1509 |
continue_on_fail | 达到失败阈值时立即停还是跑完再判 | True=跑到底最后再判;False(默认)=中途即失败 | run.py:423-426 |
各 *_limit | 单样本的资源上限 | message/token/turn/time/working/cost;超限不算"错误" | LimitExceededError 捕获,run.py:1447-1452 |
关键区分:"超限(limit)"和"出错(error)"是两回事。超限走 LimitExceededError 分支,截断后照样进入打分(run.py:1447-1452,state = sample_state() or state 拿最近状态后打分),不计入错误阈值;真正的异常才走 handle_error(run.py:1160)那条容错/重试链。
7. 并发模型:三道闸,互不打架
把并发的三个层次收在一起,避免混淆——它们各管一层:
① eval_run 层:parallel ── 同时在跑的"任务 × 模型"单元数上限(跨模型摊平)
│ run_multiple / _run_task
▼
② task_run 层:tg_collect ── 一个任务内,(样本 × epoch) 全部并发发起
│ run.py:752
▼
③ 样本层:sample semaphore ── 真正同时执行的样本数闸门(≈ 模型连接并发)
create_sample_semaphore,run.py:556 / 2059
三道闸的分工:parallel 管"几个任务并行",tg_collect 只是把所有样本一次性发起(不是它在限流),真正的样本并发上限是信号 量。此外,_run_task 给每个任务套独立 cancel scope(run.py:464),task_run_sample 里给"取消后仍要收尾"的代码段套 shield=True(如 run.py:1633、1666)——保证一个样本被取消时,它的日志仍能写完、不至于从 eval 日志里凭空消失。
8. 代码地图(导航索引)
按"要看什么"直接跳源码。所有行号以 sourceCommit 为准;行号漂移时用符号名 grep。
| 主题 | 文件路径 | 符号 |
|---|---|---|
| Task 定义与构造 | src/inspect_ai/_eval/task/task.py | Task / Task.__init__ / task_with |
| Task 入参规整 | src/inspect_ai/_eval/task/task.py | resolve_dataset / resolve_solver / resolve_epochs |
| @task 注册 | src/inspect_ai/_eval/registry.py | task / task_register / task_create |
| 样本/数据集模型 | src/inspect_ai/dataset/_dataset.py | Sample / Dataset / MemoryDataset |
| 多轮 Epochs | src/inspect_ai/_eval/task/epochs.py | Epochs |
| 同步入口 | src/inspect_ai/_eval/eval.py | eval |
| 解析装配 | src/inspect_ai/_eval/eval.py | eval_async / eval_resolve_tasks / run_batches |
| 编排调度 | src/inspect_ai/_eval/run.py | eval_run / prepare_options / run_multiple / _run_task |
| Plan 成形(setup 前置) | src/inspect_ai/_eval/task/run.py | resolve_plan |
| 单任务执行 | src/inspect_ai/_eval/task/run.py | task_run |
| 样本铺开(并发) | src/inspect_ai/_eval/task/run.py | task_run(tg_collect 处,752) |
| 单样本生命周期 | src/inspect_ai/_eval/task/run.py | task_run_sample |
| 样本并发闸门 | src/inspect_ai/_eval/task/run.py | create_sample_semaphore |
| 样本落盘 | src/inspect_ai/_eval/task/run.py | log_sample |
| solver 链执行 | src/inspect_ai/solver/_plan.py | Plan.__call__ |
接着读: solver 与 TaskState 的协议 →02;模型生成怎么发生 →03;工具调用与 agent 循环 →04;打分与指标聚合 →05;日志/事件/沙箱 →06。