Verifier

Verifier 在验证沙箱中执行 Verify,评估任务完成情况并返回得分,可用于 Data Pipeline、Training、Evaluation。

Verifier:配置验证

Parameters:

Name Type Description Default
entry str | Path

run.py 的路径。run.py 所在目录是验证材料包,保存 Task 和执行 Verify 时会复制整个目录。

required
runtime Runtime

用于操作验证沙箱和任务沙箱的沙箱管理器。

required
timeout float

run.py 的最长运行秒数,须大于 0。

required
required_env list[str] | None

验证程序需要的环境变量名称。

None
requires_task_instance bool

是否需要任务沙箱,默认 True;不需要时显式设为 False。

True
from envweaver import Verifier

checker = Verifier(
    entry="./verify/run.py",
    runtime=runtime,
    timeout=300,
    required_env=["LLM_BASE_URL", "LLM_API_KEY"],
    requires_task_instance=True,
)

Verifier.run:执行 Verify

Parameters:

Name Type Description Default
verifier_instance Instance

验证沙箱。

required
task_instance Instance | None

任务沙箱;requires_task_instance 为 True 时必填,为 False 时不能传入。

None
files list[dict[str, str | Path]] | None

要复制的文件或目录列表。source 填来源路径;target 是验证沙箱内的路径,相对路径以 run.py 所在目录为基准。

None
env dict[str, str] | None

传给验证程序的环境变量,须包含 required_env 中的全部变量。

None

Returns:

Type Description
dict

run.py 输出的 JSON 对象。

result = checker.run(
    verifier_instance=verifier_instance,
    task_instance=task_instance,
    files=[
        {"source": "./answer.json", "target": "answer.json"},
        {"source": "./screenshots", "target": "/data/screenshots"},
    ],
    env={"LLM_BASE_URL": base_url, "LLM_API_KEY": api_key},
)
run.py 入口
  • 文件名为 run.py,作为 Python 脚本在验证沙箱内执行,不要求定义特定函数或类。
  • 标准输出只写一个 JSON 对象,必须包含数值 score,范围为 0~1;可以包含其他 JSON 字段。日志写标准错误。
  • 验证正常完成时退出码为 0,包括任务得零分的情况。非零退出、超过 timeout 或输出格式错误属于执行失败。
  • 需要任务沙箱时,脚本接收 --instance <连接文件路径>,用 Runtime.connect 连接。不需要时,无须处理这个参数。

示例:在任务沙箱中运行测试

requires_task_instance 默认为 True。tests/ 与 run.py 放在一起,测试在任务沙箱的 /work/repo 中运行。

import argparse
import json
from envweaver import Runtime

parser = argparse.ArgumentParser()
parser.add_argument("--instance", required=True)
runtime = Runtime()
task_instance = runtime.connect(path=parser.parse_args().instance)

runtime.upload(
    instance=task_instance,
    source="./tests",
    target="/work/repo/tests",
)
code, stdout, stderr = runtime.exec(
    instance=task_instance,
    command="python3 -m pytest -q",
    cwd="/work/repo",
)
if code not in (0, 1):
    raise RuntimeError(stderr)
print(json.dumps({"score": float(code == 0)}))

示例:在验证沙箱中运行测试

设置 requires_task_instance=False。repo/ 与 run.py 放在一起,submission.patch 通过 files 传入。

import json
import subprocess
import sys
from pathlib import Path

subprocess.run(
    ["git", "apply", str(Path("submission.patch").resolve())],
    cwd="repo", check=True, capture_output=True,
)
result = subprocess.run(
    [sys.executable, "-m", "pytest", "-q"],
    cwd="repo", capture_output=True,
)
if result.returncode not in (0, 1):
    raise RuntimeError(result.stderr.decode())
print(json.dumps({"score": float(result.returncode == 0)}))