Agent¶

Agent 为 Eval 提供统一的执行接口。创建 Agent 对象时配置模型、环境变量和原生参数，再将对象传给 eval(agent=...)。

内置 Agent 默认复用准备阶段有校验值的安装包；cache=False 关闭。缓存目录通过当前进程的 ENVWEAVER_CACHE_DIR 设置，默认 ~/.cache/envweaver/downloads，容量为 30 GiB。

已支持的 Agent：Codex CLI、Pi、mini-swe-agent、Terminus 2、自定义 Agent。

CodexAgent¶

Parameters:

Name	Type	Description	Default
`model`	`str \| None`	Codex 模型名称，不填使用 Codex 默认配置。	`None`
`version`	`str \| None`	Codex CLI 版本；不填时使用已安装版本，未安装则安装默认版本。	`None`
`env`	`dict[str, str] \| None`	环境变量字典，传入值覆盖沙箱同名变量；默认 None，不自动读取当前机器的变量。	`None`
`extra_args`	`list[str] \| None`	原生 codex exec 参数列表，如 ["-c", 'model_reasoning_effort="high"']；默认 None。	`None`
`cache`	`bool`	是否复用已缓存的安装包，默认 True。	`True`

from envweaver import CodexAgent

agent = CodexAgent(model=model_name, env=agent_env)

PiAgent¶

Parameters:

Name	Type	Description	Default
`model`	`str \| None`	Pi 模型名称，可用 provider/model 格式；不填使用 Pi 的配置。	`None`
`version`	`str \| None`	安装的 Pi 版本；不填则使用已安装版本，未安装时安装最新版。	`None`
`env`	`dict[str, str] \| None`	环境变量字典，传入值覆盖沙箱同名变量；默认 None，不自动读取当前机器的变量。	`None`
`extra_args`	`list[str] \| None`	原生 pi 参数列表，如 ["--thinking", "high"]；默认 None。	`None`
`cache`	`bool`	是否复用已缓存的安装包，默认 True。	`True`

from envweaver import PiAgent

agent = PiAgent(model=model_name, env=agent_env)

MiniSweAgent¶

Parameters:

Name	Type	Description	Default
`model`	`str \| None`	mini-swe-agent 模型名称，例如 provider/model；不填使用其配置。	`None`
`version`	`str \| None`	安装的 mini-swe-agent 版本，仅支持 2.x；不填则使用已安装版本，未安装时安装最新的 2.x 版本。	`None`
`env`	`dict[str, str] \| None`	环境变量字典，传入值覆盖沙箱同名变量；默认 None，不自动读取当前机器的变量。	`None`
`extra_args`	`list[str] \| None`	原生 mini 参数列表，如 ["-c", "/workspace/agent.yaml"]；配置文件须已在沙箱内。默认 None。	`None`
`cache`	`bool`	是否复用已缓存的安装包，默认 True。	`True`

from envweaver import MiniSweAgent

agent = MiniSweAgent(model=model_name, env=agent_env)

Terminus2Agent¶

Parameters:

Name	Type	Description	Default
`model`	`str`	模型名称，必填，例如 openai/Qwen3.8-27B。	required
`env`	`dict[str, str] \| None`	传给沙箱内 Agent 的环境变量，默认 None；不自动读取当前机器的变量。 TMUX_TMPDIR 由适配器设置为本次独立终端目录。	`None`
`options`	`dict \| None`	Harbor 0.25.0 的 Terminus 2 参数字典，默认 None；例如 api_base、 temperature、max_turns、model_info、llm_call_kwargs。	`None`
`cache`	`bool`	是否复用已缓存的安装包，默认 True。	`True`

from envweaver import Terminus2Agent

agent = Terminus2Agent(
    model="openai/Qwen3.8-27B",
    env={"OPENAI_API_KEY": api_key},
    options={
        "api_base": base_url,
        "model_info": {
            "max_input_tokens": 262144,
            "max_output_tokens": 8192,
            "input_cost_per_token": 0,
            "output_cost_per_token": 0,
        },
    },
)

自定义 Agent

自定义 Agent 需继承 Agent 基类。run() 必须实现；preflight()、prepare() 和 stop() 按需实现。Eval 自动调用这些方法并提供参数：

方法	Eval 调用时机	适配代码需要实现什么
`preflight()`	沙箱启动前，以及 `run()` 执行前	检查输入和配置是否支持
`prepare()`	做题沙箱完成 context 上传和 task.setup 后	安装工具、配置环境或初始化客户端
`run()`	`prepare()` 完成后	执行题目，保存轨迹和产物，返回运行结果
`stop()`	超时、取消或执行失败，需要停止 Agent 时	停止本次请求和进程，保留任务沙箱及题目服务

from envweaver import Agent

class CustomAgent(Agent):
    def preflight(self, goal):
        # 检查 Agent 特有的输入要求。
        ...

    def prepare(self, *, runtime, instance, output, cwd: str):
        # 安装工具或创建客户端；参数由 Eval 提供。
        ...

    def run(self, *, goal, runtime, instance, output, cwd: str):
        # 执行 goal，将最终回复保存为文字和图片列表。
        reply = ...
        return output.finish(output=reply)

    def stop(self, timeout=30):
        # 停止自己创建的请求或进程，保留做题沙箱和题目服务。
        ...

# 自定义参数通过 self.config["参数名"] 读取；修改返回的字典不会改变 Agent 配置。
agent = CustomAgent(model=model_name, endpoint=base_url, temperature=0.7)

Agent.preflight：启动前检查¶

检查 Agent 特有的输入要求；不需要额外检查时可省略。

Parameters:

Name	Type	Description	Default
`goal`	`list[dict]`	题目的文字和图片列表。	required

Returns:

Type	Description
`None`	None。

Agent.prepare：准备 Agent¶

Parameters:

Name	Type	Description	Default
`runtime`	`Runtime`	用来操作沙箱的 Runtime。	required
`instance`	`Instance`	已完成 context 上传和 task.setup 的做题实例。	required
`output`	`AgentOutput`	本次产物写入器，output.directory 是当前机器上的产物目录。	required
`cwd`	`str`	沙箱内已存在的工作目录绝对路径。	required

Returns:

Type	Description
`None`	None。

Agent.run：执行任务¶

Parameters:

Name	Type	Description	Default
`goal`	`list[dict]`	题目的文字和图片列表；图片 path 已转换为当前机器上的路径。	required
`runtime`	`Runtime`	用来操作沙箱的 Runtime。	required
`instance`	`Instance`	做题实例。	required
`output`	`AgentOutput`	与 prepare 相同的产物写入器。	required
`cwd`	`str`	沙箱内已存在的工作目录绝对路径。	required

Returns:

Type	Description
`dict`	包含 output、artifacts、metadata 的 JSON 对象；可通过 output.finish(回复列表) 保存并返回。

返回结果

字段	含义
`output`	最终回复的文字和图片列表，无回复时为 []；使用与 goal 相同的字段，图片 path 相对于 output.directory
`artifacts.trajectory`	已保存的轨迹文件路径，无轨迹时为 null
`artifacts.logs`	已保存的日志文件路径列表，无日志时为 []
`artifacts.files`	已保存的附件路径列表，无附件时为 []
`metadata`	运行信息字典，包含以下固定字段，也可添加耗时、用量等自定义字段；未知数值使用 null
↳ `agent`	Agent 名称
↳ `agent_version`	Agent 版本，未知时为 null
↳ `model`	模型名称，未知时为 null

产物格式：

agent/
├── result.json       # Agent 运行结果快照
├── trajectory.json   # ATIF 轨迹
├── logs/             # 原始事件、原生记录和日志
└── files/            # 图片和模型上下文快照

轨迹使用 Harbor ATIF v1.8，schema_version 为 ATIF-v1.8；可通过 envweaver.ATIF_VERSION 读取当前版本。

模型步骤的 extra.context.path 指向该次调用的上下文快照；source 标明采集位置，complete 和 missing 标明是否存在已知缺失。图片、上下文和子轨迹文件路径均相对于 Agent 的产物目录。

Codex 子 Agent 使用独立子轨迹；Terminus 的压缩和续接沿用 Harbor 的轨迹引用。加密内容保存在原生记录中，不还原为推理文字。未采集到原生模型调用时，轨迹中的 extra.capture 会注明缺失。

Agent.stop：停止 Agent¶

Parameters:

Name	Type	Description	Default
`timeout`	`float`	停止本次 Agent 的最长等待秒数。可与 prepare 或 run 并发调用；保留任务沙箱和题目服务。	`30`

Returns:

Type	Description
`None`	None。

适配建议¶

1. 强烈建议使用 output 保存产物。

Eval 将 output 传给 prepare() 和 run()。新增 Agent 适配需要将原生轨迹转换为 ATIF，并保存每次模型调用的上下文；原始记录单独保存为日志。

output.directory 是当前机器上的产物目录，结果中的产物路径相对于该目录。以下写入方法默认会同步产物；直接写入目录的文件可通过 sync() 同步。

AgentOutput.save_trajectory：保存 ATIF 轨迹

写入 directory/trajectory.json，并更新 artifacts.trajectory。

Parameters:

Name	Type	Description	Default
`trajectory`	`dict`	ATIF 字典；附件和上下文文件路径相对于 directory。	required
`related`	`dict[str, dict] \| None`	子轨迹或续接轨迹的文件名到 ATIF 字典的映射，默认 None。文件名填写 trajectory.<名称>.json，文件保存到 directory 并登记到 artifacts.files。	`None`

Returns:

Type	Description
`None`	None。

from envweaver import ATIF_VERSION

trajectory = {
    "schema_version": ATIF_VERSION,
    "agent": {"name": "my-agent", "version": "1"},
    "steps": [
        {"step_id": 1, "source": "user", "message": prompt},
        {"step_id": 2, "source": "agent", "message": reply},
    ],
}
output.save_trajectory(trajectory=trajectory)

# 有续接轨迹时一并保存。
trajectory["continued_trajectory_ref"] = "trajectory.cont-1.json"
output.save_trajectory(
    trajectory=trajectory,
    related={"trajectory.cont-1.json": continuation},
)

AgentOutput.record：记录原始事件

向 logs/events.jsonl 追加原始事件，自动填写连续序号 seq 和 UTC 时间 time。

event_type 保存为 type，data 保存为事件内容；文件登记到 artifacts.logs。

Parameters:

Name	Type	Description	Default
`event_type`	`str`	事件名称，如 message、tool_call 或 Agent 原生事件名称。	required
`data`	`Any`	可保存为 JSON 的事件内容。	required

Returns:

Type	Description
`None`	None。

output.record(event_type="message", data={"role": "assistant", "text": text})

AgentOutput.write_log：写入日志

向 output.directory/logs 下的日志文件追加原始文字，并将路径登记到 artifacts.logs。

文件不存在时创建；不会自动添加时间、事件字段或换行。

Parameters:

Name	Type	Description	Default
`name`	`str`	日志文件名，如 stdout.log，保存在 directory/logs 下。	required
`text`	`str`	追加写入的日志内容。	required

Returns:

Type	Description
`None`	None。

output.write_log(name="stdout.log", text=text)

AgentOutput.save_file：保存附件

将附件内容保存到 output.directory/files，并将路径登记到 artifacts.files。

文件名根据内容生成；相同内容和扩展名复用已有文件。

Parameters:

Name	Type	Description	Default
`data`	`bytes`	附件内容。	required
`suffix`	`str`	文件扩展名，如 .png；默认空字符串。	`''`

Returns:

Type	Description
`str`	附件相对于 output.directory 的路径，可用于最终回复的图片 path 或轨迹中的附件引用。

path = output.save_file(data=image_bytes, suffix=".png")

AgentOutput.update：更新结果

更新当前运行结果并保存到 result.json；未传入的字段保持原值。

artifacts 只登记已保存文件的路径，不创建或复制这些文件。

Parameters:

Name	Type	Description	Default
`output`	`list[dict] \| None`	回复列表，填写时替换当前回复；默认 None。	`None`
`metadata`	`dict \| None`	补充或覆盖运行信息的同名字段；默认 None。	`None`
`artifacts`	`dict \| None`	补充或覆盖 trajectory、logs、files 字段；路径相对于 directory，文件须已存在。默认 None。	`None`
`sync`	`bool`	是否立即同步到产物存储目录，默认 True。	`True`

Returns:

Type	Description
`None`	None。

output.update(metadata={"usage": usage})

AgentOutput.finish：保存最终回复

将最终回复保存到 result.json，并返回包含回复、产物路径和运行信息的完整结果。

等价于 update(output=...) 后读取 result；不负责停止 Agent。

Parameters:

Name	Type	Description	Default
`output`	`list[dict]`	最终回复的文字和图片列表；无回复时填写 []。	required

Returns:

Type	Description
`dict`	已保存的完整 Agent 结果，可作为 Agent.run 的返回值。

reply = [{"type": "text", "text": "任务已完成"}]
return output.finish(output=reply)

AgentOutput.result：读取结果

读取当前结果，不写入文件；修改返回的字典不会改变已记录的结果。

Returns:

Type	Description
`dict`	当前结果的独立字典副本，包含 output、artifacts、metadata。

result = output.result

AgentOutput.sync：同步直接写入的文件

将 output.directory 中新增或修改的文件同步到产物存储目录。

直接写入 output.directory，或调用 update(sync=False) 后，可用此方法立即同步。此方法不从任务沙箱下载文件，也不将文件路径自动登记到运行结果中。

Returns:

Type	Description
`None`	None。

# 写入自行生成的文件，再同步到产物存储目录。
(output.directory / "report.txt").write_text(report_text, encoding="utf-8")
output.sync()

2. 强烈建议通过 AgentError 报告问题。

type 使用 ErrorType。

Parameters:

Name	Type	Description	Default
`type`	`ErrorType`	ErrorType 问题类型，必填；无法识别时使用 UNKNOWN。	required
`message`	`str \| None`	具体说明，默认使用类型的字符串值；不要包含密钥。	`None`

from envweaver import ErrorType
from envweaver.errors import AgentError

raise AgentError(type=ErrorType.STEP_LIMIT, message="已达到最大步数")

Keys	Action
`?`	Open this help
`n`	Next page
`p`	Previous page
`s`	Search