Inspect 提前停止：按已有分数跳过样本和轮次

CodexQA 团队2026年11月3日阅读约 3 分钟

提前停止在每个样本开跑前决定跳不跳。跳过的样本记进日志，评测结束后可以审计。示例里的随机停止只为演示，原文保留未补全的 TODO。

提前停止

概述

提前停止（early stopping，根据已经看到的结果，不再跑剩下的样本或轮次）让你在评测过程中，依据到目前为止观察到的结果，跳过样本或轮次（epoch，同一道题再跑的一次）。这对实现自适应测试算法（按前面的表现动态决定接下来测什么）有用：它根据先前表现动态决定跑哪些样本，有可能省下大量计算时间，同时保住评测质量。

常见用途包括：

结果已经稳定就停掉一个样本：如果一个样本在多个轮次里都答对（或都答错），跳过剩下的轮次。
自适应难度：把评测时间集中在靠近模型能力边界的样本上。
资源优化：跳过不太可能再提供额外信号的样本。

说明：Optstop 包
Optstop 包提供自适应最优停止算法的完整实现，用来高效判断已经收集到的数据是否够做出可靠推断。

EarlyStopping 协议

要实现提前停止，写一个实现 EarlyStopping 协议（protocol，规定必须有哪些方法的接口）的类，并把它传给 Task 的 early_stopping 参数：

from inspect_ai import Task, task
from inspect_ai.util import EarlyStopping, EarlyStop

@task
def my_task():
    return Task(
        dataset=my_dataset,
        solver=my_solver,
        scorer=my_scorer,
        early_stopping=MyEarlyStopping(),
        epochs=5,
    )

EarlyStopping 协议定义四个异步方法：

方法	说明
`start_task()`	在一次评测开始时调用，用来登记任务元数据。
`schedule_sample()`	在每个样本运行前调用；返回 EarlyStop 就跳过它。
`complete_sample()`	在一个样本完成时调用，并带上它的分数。
`complete_task()`	在任务完成时调用；返回要写入日志的元数据。

示例实现

下面是一个简单例子，随机提前停掉一些样本（只为演示）：

from pydantic import JsonValue
from typing_extensions import override

from inspect_ai.dataset import Sample
from inspect_ai.log import EvalSpec
from inspect_ai.scorer import SampleScore
from inspect_ai.util import EarlyStopping, EarlyStop

class RandomEarlyStopping(EarlyStopping):
    @override
    async def start_task(
        self, task: EvalSpec, samples: list[Sample], epochs: int
    ) -> str:
        """Task initialization."""

        # TODO: create a structure to track all of the samples/epochs
        # this will generally be updated w/ scores in complete_sample() 

        # return task name
        return "random"

    @override
    async def schedule_sample(
        self, id: str | int, epoch: int
    ) -> EarlyStop | None:
        """Return EarlyStop to skip this sample, or None to run it."""

        # TODO: determine whether the given sample has been run based
        # on the previously accumulated samples scores.

        # randomly stop some samples
        if random() < 0.5:
            return EarlyStop(id=id, epoch=epoch, reason="random stop")

        return None

    @override
    async def complete_sample(
        self, id: str | int, epoch: int, scores: dict[str, SampleScore]
    ) -> None:
        """Process results from a completed sample."""

        # TODO: track scored samples and use this to determine the
        # appropriate return value for calls to schedule_sample()

        pass

    @override
    async def complete_task(self) -> dict[str, JsonValue]:
        """Return custom metadata to record in the eval log."""

        # TODO: return any custom data about the early stopping output
        # (will be written to the log and displayed in the viewer)

        return {}