CodexQA

Industry & PracticeTools & Frameworks

Inspect 限制:时间、消息、回合、token 和花费

CodexQA 团队16 min read

样本碰到上限会提前退出并仍然评分,几乎总是不正确。工作时间可能少计,必须同时设墙上时钟上限。没有价格数据时不能设花费上限。

In this piece

设置限制

UK AI Security Institute。Setting Limits。文档页无单独发布日期(检索于 2026-11-03)。许可证:MIT License,Copyright (c) 2024 UK AI Security Institute。

概述

在开放式的模型对话里(例如带工具使用的智能体评测),模型有可能“卡住”,一直试图做一件实际上没有完成希望的任务。更进一步,模型有时会在沙箱(sandbox,隔离执行命令的环境)里调用耗时极长的命令,最坏情况会无限挂起。

对这类评测,通常应该对总时间、总消息数、回合、所用 token、花费,或其中几项一起设上限。本文覆盖:

  1. 样本限制:作用在一个任务里单个样本上的限制。样本(sample,评测里的一道题或一次运行)。
  1. 作用域限制:作用在任意代码块上的限制。
  1. 智能体限制:作用在智能体执行上的限制。

样本限制

样本限制不会变成错误,而是提前退出执行(碰到限制的样本仍然会被评分,不过几乎总是记成“不正确”)。

时间限制

这里给任务里的每个样本设 15 分钟(15 × 60 秒)的 time_limit(时间限制,按墙上时钟计的最长时间):

@task
def intercode_ctf():
    return Task(
        dataset=read_dataset(),
        solver=[
            system_message("system.txt"),
            use_tools([bash(timeout=3 * 60)]),
            generate(),
        ],
        time_limit=15 * 60,
        scorer=includes(),
        sandbox="docker",
    )

注意,这里还给 bash() 命令设了 3 分钟超时。这不是必须的,但通常是好主意,这样一条跑偏的 bash 命令就不会吃掉整个 time_limit。

也可以在 CLI 上,或调用 eval() 时指定时间限制:

inspect eval ctf.py --time-limit 900

合适的超时取决于任务本身,请把上面的数字只当例子,不要当成推荐值。

工作时间限制

working_limit(工作时间限制,只计算真正在干活的时间)和 time_limit 不同:它只计量花在工作上的时间,而不包括为应对速率限制而重试、或等待其他共享资源的时间。工作时间是时钟时间减去样本处于等待的时间。下面任何一件事正在进行时,样本都算在等待:

  • 等待模型连接,或等待一个 concurrency() 槽位(并发槽,同时只允许这么多件事在跑的名额),包括 max_sandboxes 和 max_subprocesses。
  • 一次失败后被重试的模型请求(例如被限速的请求),以及重试前的退避。
  • 一次批量请求在等待提交。
  • 工具审批或复查,由人或由模型来做。
  • 等待人回答一个问题。
  • 被 inspect ctl pause --now 按住的模型调用。

重叠的等待只计一次,所以工作时间从不为负,也从不超过时钟时间。

警告:

任何等待正在进行时做的工作都不计入。样本的一部分在等,同一条样本里的其他工作就是免费的:并发任务、并行工具调用、子智能体、沙箱里的后台进程,以及进程内智能体库做的工作。因此工作时间可能少计,样本实际跑的时间可以远长于它的工作时间限制。设 working_limit 时一定要同时设 time_limit,把时钟时间框住。

说明:

为了把成功的请求和模型客户端内部的重试分开,Inspect 通过钩进大多数模型包的 HTTP 客户端来计量请求。当某个供应商不报告请求时间(例如 azureai、sagemaker,以及 hf 这类进程内供应商),或者把整个调用都计量进去(grok)时,working_time 会包含模型客户端自己做的重试。

这里给任务里的每个样本设 10 分钟(10 × 60 秒)的 working_limit,并用 30 分钟的 time_limit 作为时钟时间的兜底:

@task
def intercode_ctf():
    return Task(
        dataset=read_dataset(),
        solver=[
            system_message("system.txt"),
            use_tools([bash(timeout=3 * 60)]),
            generate(),
        ],
        working_limit=10 * 60,
        time_limit=30 * 60,
        scorer=includes(),
        sandbox="docker",
    )

或者在 CLI 上:

inspect eval ctf.py --working-limit 600 --time-limit 1800

消息限制

消息限制约束任何一段对话里的消息条数(例如一个 TaskState、一个 AgentState,或任何传给 generate() 的输入)。

消息限制在这些时候检查:

  • 无论何时对任何模型调用 generate()。如果传给 generate() 的 input 参数里的消息数等于或超过限制,会抛出 LimitExceededError(超出限制错误,限制被打破时抛出的异常)。这是为了避免已经到顶了还再去做一次浪费的生成调用。
  • 无论何时 TaskState.messages 或 AgentState.messages 被改动,但只有计数超过限制时才抛出 LimitExceededError。

这里给任务里的每个样本设 message_limit 为 30:

@task
def intercode_ctf():
    return Task(
        dataset=read_dataset(),
        solver=[
            system_message("system.txt"),
            use_tools([bash(timeout=120)]),
            generate(),
        ],
        message_limit=30,
        scorer=includes(),
        sandbox="docker",
    )

这把一段对话的消息总数限制在 30 条,到了之后模型被迫放弃。那时 TaskState 里恰好有的 output 会被评分(多半得到不正确)。

Token 限制

所有模型都会自动记录 token 用量(用 ModelUsage 的 total_tokens)。token(词元,模型计费和上下文按它计数的单位)限制在每次调用 generate() 时检查。默认计量全部 token;用类型为 "output" 的限制只计量输出 token(含推理 token),或用算术公式计量加权混合(见下文)。

这里给任务里的每个样本设 token_limit 为 500K:

@task
def intercode_ctf():
    return Task(
        dataset=read_dataset(),
        solver=[
            system_message("system.txt"),
            use_tools([bash(timeout=120)]),
            generate(),
        ],
        token_limit=(1024*500),
        scorer=includes(),
        sandbox="docker",
    )

token_limit 也接受带数量后缀的字符串(例如 "500k" 或 "1m"),并且可以用 TokenLimit 值或 "output:" 前缀,把范围收成只计输出 token:

from inspect_ai.util import TokenLimit

Task(
    ...,
    token_limit=TokenLimit(tokens=1024*500, type="output")
)

# equivalent string form (also usable from the CLI, e.g.
# `inspect eval ctf.py --token-limit output:500k`)
Task(
    ...,
    token_limit="output:500k"
)

输出 token 限制只计量模型生成的 token(含推理 token),所以不受对话输入 token 增长的影响。

type 也可以是变量 input 和 output 上的算术公式,用来计量不同 token 类型的加权混合(例如把更便宜的输入 token 权重压低):

# meter 10% of input tokens plus all output tokens
Task(..., token_limit=TokenLimit(tokens=1024*500, type="(input * 0.1) + output"))

# equivalent string / CLI form:
#   inspect eval ctf.py --token-limit "(input*0.1)+output:500k"
Task(..., token_limit="(input*0.1)+output:500k")

公式支持 +、-、*、/、括号和一元负号。公式里的 input 是真实提示大小(含缓存 token),output 含推理 token;计算值向下取整为整数。

重要:

要注意 token_limit 管的是一个样本执行过程中用掉的全部 token。如果想限制单次模型调用能产出多少 token,应该用生成选项 max_tokens。

回合限制

一个“回合”(turn,一次产生助手消息的模型生成)是一次模型生成(一次产生助手消息的模型调用)。回合限制和消息限制不同,消息限制数的是对话里的全部消息(用户、助手和工具消息)。一个回合常常变成好几条消息(例如一条助手消息,加上它的工具调用带来的工具消息)。

每个顶层 generate() 调用记一个回合(重试和回退已经解决之后,包含缓存命中)。经 model.compact() 做的生成不计为回合。回合限制在每次记下回合时检查。

这里给任务里的每个样本设 turn_limit 为 300:

@task
def intercode_ctf():
    return Task(
        dataset=read_dataset(),
        solver=[
            system_message("system.txt"),
            use_tools([bash(timeout=120)]),
            generate(),
        ],
        turn_limit=300,
        scorer=includes(),
        sandbox="docker",
    )

这把智能体限制在 300 次模型生成,之后被迫放弃。和其他样本限制一样,那时 TaskState 里恰好有的 output 会被评分。

花费限制

花费由 token 用量和模型价格数据算出(见模型价格)。花费限制在每次调用 generate() 时检查。

这里给任务里的每个样本设 cost_limit 为 2.00 美元:

@task
def intercode_ctf():
    return Task(
        dataset=read_dataset(),
        solver=[
            system_message("system.txt"),
            use_tools([bash(timeout=120)]),
            generate(),
        ],
        cost_limit=2.00,
        scorer=includes(),
        sandbox="docker",
    )

重要:

cost_limit 要求通过 set_model_cost() 或 --model-cost-config 配好模型价格数据。如果设了花费限制,但评测用到的模型没有全部配上价格,会报错。

#### 模型价格

花费跟踪要求评测或评测集里出现的每个模型都有价格数据。设置价格有两种办法:

Python API:

from inspect_ai.model import set_model_cost, ModelCost

set_model_cost("openai/gpt-4o", ModelCost(
    input=2.50, output=10.00,
    input_cache_write=0, input_cache_read=1.25,
))

CLI(YAML 或 JSON 文件):

每个模型都要为 input、output、input_cache_write 和 input_cache_read 设一组价格。价格应按每百万 token 多少美元给出。用不到的字段设为 0。

下面是一份 YAML 价格配置示例:

openai/gpt-4o:
    input: 2.50
    output: 10.00
    input_cache_write: 0
    input_cache_read: 1.25
anthropic/claude-sonnet-4-5-20250514:
    input: 3.00
    output: 15.00
    input_cache_write: 3.75
    input_cache_read: 0.30

(截至 2026 年 2 月 9 日,所有主要模型供应商都把推理 token 算作输出 token,所以不必再为推理 token 单独给价格。如果你的用途需要把推理 token 价格分开算,请联系我们。)

配好模型价格之后,会跟踪样本整体的花费,也会跟踪样本里任何带 ModelUsage 字段的事件的花费。

一次请求按实际提供服务的那个模型的费率计价,条件是供应商报告的服务模型不是请求的那个模型:一次拒绝回退、一次 Azure OpenAI、Azure AI 或 Microsoft Foundry 部署、一个 OpenRouter、Bedrock 或 Fireworks 路由器,或一次 LiteLLM 代理回退。所请求模型的一个带日期快照,按所请求模型的费率计价。如果服务模型没有价格数据,这次请求按所请求模型的费率计价,并有一条警告点名服务模型(例如 openai/gpt-4o-mini-2024-07-18)。把价格数据设在那个名字下。set_model_cost() 和 --model-cost-config 只接受 Inspect 模型库里的模型;其他模型用 set_model_info():

from inspect_ai.model import set_model_info, ModelCost, ModelInfo

set_model_info("privateorg/routed-model", ModelInfo(cost=ModelCost(
    input=1.00, output=4.00,
    input_cache_write=0, input_cache_read=0.10,
)))

另外,配好模型价格之后就可以设置样本花费限制:

inspect eval ctf.py --model-cost-config pricing.yaml --cost-limit 2.00

自定义限制

限制被超出时,会抛出 LimitExceededError,并由 Inspect 主样本执行逻辑接住。如果要做自定义限制类型,可以这样抛出 LimitExceededError 来执行它:

from inspect_ai.util import LimitExceededError

raise LimitExceededError(
    "custom",
    value=value,
    limit=limit,
    message=f"A custom limit was exceeded: {value}"
)

查询用量

可以查一个样本限制已经用了多少、限制是多少、还剩多少资源:

sample_time_limit = sample_limits().time
print(f"{sample_time_limit.remaining:.0f} seconds remaining")

注意 sample_limits() 只取样本级限制,不取作用域限制或智能体限制。

样本限制的用量也会记进日志:每个 EvalSample(及其摘要)带有 turn_count(顶层模型生成的次数),以及它的 token 限制配置和已计量用量(token_limit、token_limit_type 和 token_limit_usage;没有配置 token 限制时为 None)。turn_count 和 token_limit_usage 是默认的 samples_df() 列(其他列要另加,例如 SampleColumn("token_limit", path="token_limit")),而且 inspect ctl sample list 会实时显示它们全部。

作用域限制

也可以在任意作用域上加限制,独立于样本级或智能体级限制。例如加在某一段代码上:

with token_limit(1024*500):
    ...

超出限制会抛出 LimitExceededError。LimitExceededError 的 source 字段会被设成被超出的那个 Limit 实例。

捕捉 LimitExceededError 时,要让 try 块包住限制上下文管理器的使用,因为有些 LimitExceededError 是在关闭上下文管理器的作用域上抛出的:

try:
    with token_limit(1024*500):
        ...
except LimitExceededError:
    ...

apply_limits() 接受一组 Limit 实例。如果传入的任何一个限制被超出,打开上下文管理器时产出的 LimitScope 上的 limit_error 属性会被设成这个异常。默认情况下,所有 LimitExceededError 都会继续向外传。但如果 catch_errors 为真,那些直接因为超出了传给它的限制而产生的错误会被接住。它始终会让其他限制触发的 LimitExceededError(例如样本级限制)继续沿调用栈向上传。

with apply_limits(
    [token_limit(1000), message_limit(10)], catch_errors=True
) as limit_scope:
    ...
if limit_scope.limit_error:
    print(f"One of our limits was hit: {limit_scope.limit_error}")

查看用量

可以通过作用域限制的 usage 属性查询一种受限资源到目前用了多少。例如:

with token_limit(10_000) as limit:
    await generate()
    print(f"Used {limit.usage:,} of 10,000 tokens")

如果把限制实例传给 apply_limits() 或一个智能体,又想查询用量,应该自己留一份引用:

limit = token_limit(10_000)
with apply_limits([limit]):
    await generate()
    print(f"Used {limit.usage:,} of 10,000 tokens")

时间限制

要把一段代码里的墙上时钟时间限制在 15 分钟:

with time_limit(15 * 60):
    ...

内部用的是 anyio 的取消作用域。代码块会在第一个让出点(例如 await 语句)被取消。

工作时间限制

working_limit 和 time_limit 不同:它只计量花在工作上的时间,而不包括为应对速率限制而重试、或等待其他共享资源的时间。工作时间是时钟时间减去样本处于等待的时间。下面任何一件事正在进行时,样本都算在等待:

  • 等待模型连接,或等待一个 concurrency() 槽位,包括 max_sandboxes 和 max_subprocesses。
  • 一次失败后被重试的模型请求(例如被限速的请求),以及重试前的退避。
  • 一次批量请求在等待提交。
  • 工具审批或复查,由人或由模型来做。
  • 等待人回答一个问题。
  • 被 inspect ctl pause --now 按住的模型调用。

重叠的等待只计一次,所以工作时间从不为负,也从不超过时钟时间。

警告:

任何等待正在进行时做的工作都不计入。样本的一部分在等,同一条样本里的其他工作就是免费的:并发任务、并行工具调用、子智能体、沙箱里的后台进程,以及进程内智能体库做的工作。因此工作时间可能少计,样本实际跑的时间可以远长于它的工作时间限制。设 working_limit 时一定要同时设 time_limit,把时钟时间框住。

说明:

为了把成功的请求和模型客户端内部的重试分开,Inspect 通过钩进大多数模型包的 HTTP 客户端来计量请求。当某个供应商不报告请求时间(例如 azureai、sagemaker,以及 hf 这类进程内供应商),或者把整个调用都计量进去(grok)时,working_time 会包含模型客户端自己做的重试。

要把工作时间限制在 10 分钟,并用 30 分钟作为时钟时间的兜底:

with time_limit(30 * 60), working_limit(10 * 60):
    ...

和时间限制不同,这不是由 anyio 驱动的。它会被定期检查,例如从 generate() 里,以及每个 Solver 跑完之后。

#### 暂停工作时间限制

要把一段代码花的时间算成等待,用 suspend_working_limit():

with working_limit(10 * 60):
    await generate()  # counts as working time
    with suspend_working_limit():
        # time spent here counts as waiting
        await wait_for_operator()
    await generate()  # counts again

和 suspend_token_limit()、suspend_turn_limit() 不同,后两者只影响代码块内部;suspend_working_limit() 在代码块打开期间,会让整条样本都算成等待,包括这条样本里其他任务做的工作。只有评测进程里的代码能调用它(沙箱里跑的智能体代码不行)。Inspect 已经对工具审批和复查这样做了。

消息限制

消息限制约束任何一段对话里的消息条数(例如一个 TaskState、一个 AgentState,或任何传给 generate() 的输入)。

消息限制在这些时候检查:

  • 无论何时对任何模型调用 generate()。如果传给 generate() 的 input 参数里的消息数等于或超过限制,会抛出 LimitExceededError。这是为了避免已经到顶了还再去做一次浪费的生成调用。
  • 无论何时 TaskState.messages 或 AgentState.messages 被改动,但只有计数超过限制时才抛出 LimitExceededError。

作用域消息限制和作用域 token 限制的行为不同:只检查最内层处于活动状态的 message_limit()。

要限制一段代码里的对话长度:

@agent
def myagent() -> Agent:
    async def execute(state: AgentState):

        with message_limit(50):
            # A LimitExceededError will be raised when the limit is exceeded
            ...
            with message_limit(None):
                # The limit of 50 is temporarily removed in this block of code
                ...

重要:

要注意 message_limit() 限制的是对话里的消息总数,不只是智能体新追加的消息。

Token 限制

所有模型都会自动记录 token 用量(用 ModelUsage 的 total_tokens)。token 限制在每次调用 generate() 时检查。默认计量全部 token;用类型为 "output" 的限制只计量输出 token(含推理 token),或用算术公式计量加权混合(见下文)。

要限制一段代码里能用的 token 总数:

@agent
def myagent(tokens: int = (1024*500)) -> Agent:
    async def execute(state: AgentState):

        with token_limit(tokens):
            # a LimitExceededError will be raised if the limit is exceeded
            ...

限制可以叠起来。上下文管理器打开期间用掉的 token,会计入所有打开着的 token 限制。

@agent
def myagent() -> Solver:
    async def execute(state: AgentState):

        with token_limit(1024*500):
            ...
            with token_limit(1024*200):
                # Tokens used here count towards both active limits
                ...

如果只计量输出 token(含推理 token)而不是全部 token,传入 type="output":

with token_limit(1024*500, type="output"):
    # a LimitExceededError will be raised if more than
    # 500K output tokens are used
    ...

type 也可以是变量 input 和 output 上的算术公式(支持 +、-、*、/、括号和一元负号),用来计量不同 token 类型的加权混合:

with token_limit(1024*500, type="(input * 0.1) + output"):
    # meters 10% of input tokens plus all output tokens
    ...

公式里,input 是真实提示大小(含缓存 token),output 含推理 token;计算值向下取整为整数。不同类型的限制可以叠在一起,每条限制按自己的口径计量。

重要:

要注意 token_limit() 管的是上下文管理器打开期间用掉的全部 token。如果想限制单次模型调用能产出多少 token,应该用生成选项 max_tokens。

#### 暂停 Token 限制

要跑一段不应计入任何活动 token 限制的代码,用 suspend_token_limit():

with token_limit(10_000):
    await generate()  # counts against the 10k budget
    with suspend_token_limit():
        # tokens used here are not metered against the 10k limit,
        # and any inner `token_limit()` is also suspended
        await expensive_summary()
    await generate()  # counts again

和 with token_limit(None): 不同,后者只压掉最内层限制的检查;suspend_token_limit() 在这段代码期间,会把所有活动 token 限制的记录和检查都完全关掉。

回合限制

一个“回合”是一次模型生成(一次产生助手消息的模型调用)。回合限制和消息限制不同,消息限制数的是对话里的全部消息(用户、助手和工具消息)。一个回合常常变成好几条消息(例如一条助手消息,加上它的工具调用带来的工具消息)。

每个顶层 generate() 调用记一个回合(重试和回退已经解决之后,包含缓存命中)。经 model.compact() 做的生成不计为回合。回合限制在每次记下回合时检查。

要限制一段代码里能用的回合(模型生成)总数:

@agent
def myagent(turns: int = 300) -> Agent:
    async def execute(state: AgentState):

        with turn_limit(turns):
            # a LimitExceededError will be raised if the limit is exceeded
            ...

和 token 限制一样,回合限制可以叠起来:一次生成会计入所有打开着的回合限制。要跑不应计入任何活动回合限制的生成,用 suspend_turn_limit():

with turn_limit(10):
    await generate()  # counts against the 10 turn budget
    with suspend_turn_limit():
        # generations here are not metered against the 10 turn limit
        await auxiliary_generate()
    await generate()  # counts again

花费限制

花费由 token 用量和模型价格数据算出(见模型价格)。花费限制在每次调用 generate() 时检查。

要限制一段代码里的总花费:

@agent
def myagent(budget: float = 2.00) -> Agent:
    async def execute(state: AgentState):

        with cost_limit(budget):
            # a LimitExceededError will be raised if the limit is exceeded
            ...

花费限制和 token 限制类似,可以叠起来,并跟踪上下文管理器打开期间用掉的花费。

重要:

使用 cost_limit() 要求通过 set_model_cost() 或 --model-cost-config 配好模型价格数据。细节见模型价格。

智能体限制

要用一个或多个限制来跑智能体,把限制对象放进 limits 参数,传给 handoff()、as_tool()、as_solver() 或 run() 这类函数(跑智能体的各种方式见使用智能体)。

这里把一个当作求解器放进来的智能体限制在 500K token:

eval(
    task="research_bench", 
    solver=as_solver(web_surfer(), limits=[token_limit(1024*500)])
)

这里把一次 handoff()(交接,把控制权交给另一个智能体)限制在 500K token:

eval(
    task="research_bench", 
    solver=[
        use_tools(
            addition(),
            handoff(web_surfer(), limits=[token_limit(1024*500)]),
        ),
        generate()
    ]
)

超出限制

智能体执行期间超出限制时,处理方式取决于智能体是怎么跑的:

  • 经 as_solver() 使用的智能体,如果超出限制,样本会终止(这和样本级限制完全一样)。
  • 带着限制被直接 run() 的智能体,它们的限制异常会被接住,放在一个元组里返回。不是传给 run() 的那些限制,会沿栈向上传。
  from inspect_ai.agent import run

  state, limit_error = await run(
      agent=web_surfer(), 
      input="What were the 3 most popular movies of 2020?",
      limits=[token_limit(1024*500)])
  )
  if limit_error:
      ...
  • 基于工具的智能体(handoff() 和 as_tool()),如果超出限制,会把说明这一点的消息交回模型,但样本继续运行。

出处:UK AI Security Institute,Setting Limits,文档页无单独发布日期(检索于 2026-11-03),https://inspect.aisi.org.uk/setting-limits.html ,MIT License(Copyright (c) 2024 UK AI Security Institute)。

Found it useful? Pass it on

WeChat

Scan with WeChat to open it on your phone and forward it.

Subscribe via RSS

Submit a correction