Eval¶

import json
import os
import warnings
from copy import copy, deepcopy
from dataclasses import replace
from envweaver.agents.execution import AgentExecution
from envweaver.evaluation.store import inspect_task
from envweaver.evaluation.failure import stop_reason
from envweaver.evaluation.runner import _error_record
from envweaver.errors import AgentStopReason

# 下文“阶段时限”到期时抛出 StageTimeout；底层操作的 TimeoutError 不视为阶段到期。

def evaluate(task, agent, config, output_dir):
    execution = AgentExecution(agent.new())  # 为本次尝试创建独立 Agent，并管理其执行流程。
    # 此处 output_dir 是当前机器上的工作目录。
    # 使用 Verifier 已绑定的 Runtime，用来操作沙箱。
    runtime = task.verifier.runtime

    # 先建立结果记录，便于中途失败时仍能保存任务信息和错误。
    result = {
        "name": task.name,           # 任务名称。
        "task_ref": task.task_ref,   # 任务在 Task Pool 中的地址，用于追溯和重新加载。
        "metadata": {},  # 本次 rollout 的自定义信息。
        "agent_stop_reason": None,  # Agent 退出原因；尚未运行时为 None。
        "agent_result": None,       # agent.run 返回的结果，或失败后读取的已有结果快照。
        "verification_result": None, # 验证程序返回的 JSON，包含 score 及其他评测信息。
        "score_valid": False,       # 该次得分是否可以计入统计。
        "errors": [],   # 评测过程中的问题，按发生顺序记录。
    }

    task_instance = None      # Agent 执行任务的沙箱实例。
    verifier_instance = None  # 运行验证程序的沙箱实例。
    stop_attempted = False  # 是否已请求停止 Agent。

    stage = "preflight"  # 当前评测阶段：preflight（启动前检查）、setup、agent 或 verify。
    output_dir.mkdir(parents=True, exist_ok=False)  # 每次尝试使用新目录；目录已存在时在创建资源前报错。

    try:
        execution.preflight(goal=task.goal)
        _, input_error = inspect_task(task)  # 检查并计算输入材料指纹；批量执行时复用已计算的结果。
        if input_error is not None:
            raise input_error

        # 任务的 Verifier 对象
        verifier = task.verifier

        # 从 os.environ 读取必需变量，config.verifier_env 补充或覆盖。
        env = dict(config.verifier_env)
        for name in verifier.required_env:
            if name not in env and (value := os.environ.get(name)) is not None:
                env[name] = value
            if name not in env:
                raise ValueError(f"Missing verifier environment variable: {name}")

        setup_timeout = config.setup_timeout
        if task.setup is not None and setup_timeout < task.setup["timeout"]:
            warnings.warn("setup_timeout is less than task.setup.timeout; Setup may time out before the setup command finishes")

        # Setup
        stage = "setup"
        with 阶段时限(setup_timeout):
            # 递归合并本次任务环境配置，得到启动任务沙箱的引用。
            task_ref = task.task_runtime_ref
            if config.task_runtime_config:
                task_ref = replace(
                    task_ref, config=merge_config(task_ref.config, config.task_runtime_config)
                )

            # 递归合并本次验证环境配置，得到启动验证沙箱的引用。
            verifier_ref = task.verifier_runtime_ref
            if config.verifier_runtime_config:
                verifier_ref = replace(
                    verifier_ref, config=merge_config(verifier_ref.config, config.verifier_runtime_config)
                )

            # 准备做题实例和验证实例
            task_instance = runtime.start(
                ref=task_ref,
                context=task.context,
                setup=task.setup,
                connection=config.task_connection,
            )
            verifier_instance = runtime.start(
                ref=verifier_ref,
                connection=config.verifier_connection,
            )

            # 准备 Agent 执行任务所需的环境和工具，具体操作由各 Agent 适配代码实现。
            execution.prepare(
                runtime=runtime,
                instance=task_instance,
                output_dir=output_dir / "agent",
                cwd=task.cwd,
            )

        # Agent Run
        stage = "agent"
        agent_timeout = config.agent_timeout
        try:
            with 阶段时限(agent_timeout):
                result["agent_result"] = execution.run(
                    goal=task.goal,
                    runtime=runtime,
                    instance=task_instance,
                    output_dir=output_dir / "agent",
                )
        except BaseException as error:
            reason = stop_reason(error)
            result["agent_stop_reason"] = reason.value
            if not reason.should_verify:
                raise
            result["errors"].append(_error_record("agent", error))
            stop_attempted = True
            try:
                execution.stop(timeout=30)
            except Exception as stop_error:
                result["errors"].append(_error_record("cleanup", stop_error, operation="stop"))
                return result  # finally 仍会保存结果并关闭沙箱。
        else:
            result["agent_stop_reason"] = AgentStopReason.COMPLETED.value

        # Verify：正常完成或达到执行限制，均对已有成果评分。
        stage = "verify"
        # 在副本上设置本次验证超时，不修改 Task。
        if config.verify_timeout is not None:
            if config.verify_timeout < verifier.timeout:
                warnings.warn("verify_timeout is less than task.verifier.timeout")
            verifier = copy(verifier)
            verifier.timeout = config.verify_timeout
        result["verification_result"] = verifier.run(
            verifier_instance=verifier_instance,
            task_instance=(
                task_instance if verifier.requires_task_instance else None
            ),
            env=env,
        )

    except BaseException as error:
        result["errors"].append(_error_record(stage, error))
        if not isinstance(error, Exception):
            raise

    finally:
        # 停止本次 Agent
        if not stop_attempted:
            try:
                execution.stop(timeout=30)
            except Exception as error:
                result["errors"].append(_error_record("cleanup", error, operation="stop"))

        # run 未返回结果时，读取已保存的 Agent 结果快照。
        if result["agent_result"] is None:
            try:
                with (output_dir / "agent" / "result.json").open("r", encoding="utf-8") as file:
                    result["agent_result"] = json.load(file)
            except FileNotFoundError:
                pass
            except Exception as error:
                result["errors"].append(_error_record("cleanup", error, operation="snapshot"))

        # 关闭验证实例和做题实例
        for instance in (verifier_instance, task_instance):
            if instance is not None:
                try:
                    runtime.close(instance=instance)
                except Exception as error:
                    result["errors"].append(_error_record("cleanup", error, operation="close"))

        result["score_valid"] = (
            result["verification_result"] is not None
            and result["agent_stop_reason"] is not None
            and AgentStopReason(result["agent_stop_reason"]).should_verify
        )

        # 保存本次评测结果
        with (output_dir / "result.json").open("w", encoding="utf-8") as file:
            json.dump(result, file, ensure_ascii=False, indent=2, allow_nan=False)

    return result


def merge_config(base, overrides):
    merged = deepcopy(base)
    for key, value in overrides.items():
        if isinstance(merged.get(key), dict) and isinstance(value, dict):
            merged[key] = merge_config(merged[key], value)
        else:
            merged[key] = deepcopy(value)
    return merged


task = pool.load(task_ref=task_ref)
result = evaluate(task=task, agent=agent, config=config, output_dir=output_dir)

Eval 结果的 agent_stop_reason 保存 AgentStopReason 的字符串值；尚未进入 Agent Run 时为 null。退出原因如下：

ErrorType：问题类型