首页
/ Agent Zero 的 wait 工具深入解析:受干预流程约束的托管等待机制

Agent Zero 的 wait 工具深入解析:受干预流程约束的托管等待机制

2026-09-14 10:24:27作者:魏献源Searcher

本指南围绕 Agent Zero 框架中的 wait 工具展开,聚焦 tools/wait.py 及其配套 DOX 文档 tools/wait.py.dox.md,完整讲解该工具的参数契约、执行流程、底层 managed_wait 调度循环、干预(intervention)处理、日志输出与验证方式。读者读完可掌握 wait 工具的完整工作原理,并能在自己的 Agent 会话中正确使用时长等待与时间戳等待两种模式。

一、工具定位与职责边界

在 Agent Zero 中,wait 是一个由 Agent 自主调用的工具,用于在任务流程中"暂停",等待一段受控时长或等待到达某个目标时间点。其核心定位(见 tools/wait.py.dox.md 的 Purpose 一节)包括三点:

  • 负责 wait.py 这个 agent 工具的实现与维护;
  • 让 Agent 在尊重干预流程(intervention flow)的前提下完成受管等待;
  • 由于 tools/ 目录刻意保持扁平结构,wait.pywait.py.dox.md 必须保持同步,DOX 文档负责记录实现的责任、契约、副作用与验证方式。

从职责划分看,wait.py 拥有运行时实现,而 wait.py.dox.md 持有关于该实现的持久化说明。这种"代码 + DOX"的双文件模式贯穿整个 tools/ 目录,任何工具参数、输出形态、break_loop 行为、干预处理、提示词指令或副作用的变化,都要求同步更新 DOX 文档。

二、工具契约:Tool 子类与 Response 返回

2.1 运行时契约

DOX 文档明确列出了工具的运行时契约:

  • 工具模块必须定义 helpers.tool.Tool 的子类;
  • execute(...) 必须返回 helpers.tool.Response

查看 helpers/tool.py 可以看到基类定义:Tool 是一个抽象基类,execute 是其抽象方法,Response 是一个 dataclass,包含 message: strbreak_loop: bool 和可选的 additional: dictbreak_loop 字段控制工具执行后是否终止 Agent 主循环,wait 工具的所有返回路径都显式设置为 break_loop=False,即等待结束后 Agent 继续正常执行。

2.2 类结构

根据 DOX 文档,WaitTool 继承自 Tool,定义了三个公开成员:

  • async execute(self, **kwargs) -> Response:核心执行入口;
  • get_log_object(self):创建执行日志对象;
  • get_heading(self, text: str = ..., done: bool = ...):生成日志标题(含图标前缀)。

WaitTool 的导入依赖包括 asynciodatetimehelpers.localizationhelpers.print_stylehelpers.toolhelpers.wait,从源码 tools/wait.py 可以逐行确认这些依赖。

三、参数契约与两种等待模式

wait 工具的提示词契约位于 prompts/agent.system.tool.wait.md

pause until a duration or timestamp,args 为 secondsminuteshoursdays 中任意组合,或 until ISO 时间戳;仅当等待确实是任务的一部分时才使用

这条提示词约束值得注意:wait 不是用来"拖时间"的,Agent 只有在任务本身需要等待(例如等待外部服务就绪、等待定时任务触发)时才应调用它。

execute 中,参数解析逻辑如下(tools/wait.py):

seconds = self.args.get("seconds", 0)
minutes = self.args.get("minutes", 0)
hours = self.args.get("hours", 0)
days = self.args.get("days", 0)
until_timestamp_str = self.args.get("until")

四种时长参数默认均为 0,until 为可选的时间戳字符串。关键判断是 is_duration_wait = not bool(until_timestamp_str)——只要提供了 until,就走时间戳模式;否则走时长模式。

3.1 时长模式(duration wait)

时长模式使用 timedelta 将四个参数累加为总时长:

wait_duration = timedelta(
    days=int(days),
    hours=int(hours),
    minutes=int(minutes),
    seconds=int(seconds),
)

随后校验 wait_duration.total_seconds() <= 0,若总时长不为正,则直接返回 Response(message="Wait duration must be positive.", break_loop=False)。也就是说,全部参数缺省(均为 0)时会触发该校验,wait 不会空转。目标时间点计算为 target_time = now + wait_duration

3.2 时间戳模式(until 模式)

时间戳模式调用 Localization.get().localtime_str_to_utc_dt(until_timestamp_str) 将本地时间字符串转换为 UTC datetime。查看 helpers/localization.py 可知该转换函数的行为:

  • 输入为 None 或无效字符串时返回 None
  • 字符串会先 strip() 并将末尾 Z 替换为 +00:00
  • 优先用 datetime.fromisoformat 解析;若解析出的对象无时区信息,则按用户配置的时区(localize_naive_datetime)处理;
  • 若 ISO 解析失败,会尝试 %Y-%m-%d %H:%M:%S%Y-%m-%d %H:%M%Y-%m-%d 等常见本地格式;
  • 最终统一 astimezone(UTC) 返回。

tools/wait.py 中,若转换结果为 None 或解析抛出 ValueError,工具会返回带错误信息的 Responsebreak_loop=False

3.3 目标时间在过去

无论哪种模式,计算完 target_time 后都会执行统一校验 if target_time <= now,若目标时间已过,返回 Response(message=f"Target time ... is in the past.", break_loop=False)。这一步防止 Agent 为过去的时间点发起无意义的等待。

四、干预流程:handle_intervention 的接入点

DOX 文档强调 wait 工具"尊重干预流程",这是该工具与普通 asyncio.sleep 的根本区别。execute 的第一步就是 await self.agent.handle_intervention()tools/wait.py)。

查看 agent.pyhandle_intervention 的实现:它先调用 wait_if_paused()(若上下文处于暂停状态则循环休眠等待恢复),然后检查是否存在待处理的干预消息;若有,则保存当前工具的进度、将干预消息追加为 hist_add_user_message(msg, intervention=True) 并抛出 InterventionException。这意味着:用户在等待过程中发起的暂停、中止或干预指令,wait 工具会及时响应,而不是死等到底。

更重要的是,handle_intervention等待循环内部也会被周期性调用(详见下一节),因此干预流程在整个等待期间始终是"活的"。

五、底层实现:helpers/wait.py 的 managed_wait 调度循环

DOX 文档记录的核心被调用函数是 managed_wait,位于 helpers/wait.py。它的签名是:

async def managed_wait(agent, target_time, is_duration_wait, log, get_heading_callback)

5.1 循环主体

循环条件是 Localization.get().now() < target_time,即当前时间未到达目标时间就持续等待。每次循环执行以下步骤:

  1. 记录 before_intervention = now()
  2. await agent.handle_intervention()——等待期间周期性检查干预;
  3. 记录 after_intervention = now()
  4. 时长模式的暂停补偿:若为时长等待且干预耗时超过 1.5 秒(大于一个休眠周期),则 target_time += pause_duration 并打印扩展日志。这是设计上的精妙之处:干预处理(如用户暂停)占用的时间不计入等待时长,保证"等 10 秒"就是实打实的 10 秒,而不是被干预中断侵蚀。
  5. 重新检查当前时间是否已到目标时间,若是则跳出循环;
  6. 计算剩余秒数,调用 log.update(heading=get_heading_callback(format_remaining_time(remaining_seconds))) 更新日志标题;
  7. sleep_duration = min(1.0, remaining_seconds)await asyncio.sleep(sleep_duration)——以不超过 1 秒的粒度休眠,兼顾响应性与资源占用。

5.2 剩余时间格式化

format_remaining_timehelpers/wait.py)将剩余秒数格式化为人类可读字符串:天/小时/分钟/秒逐级分解,粒度随剩余量自适应——大于小时级别时秒数取整,分钟级别时保留 0.1 秒精度,纯秒级则保留 0.1 秒精度。典型输出如 1d 2h 3m remaining12.5s remaining

5.3 返回值

managed_wait 返回(可能被扩展后的)target_time。调用方在 tools/wait.py 中将其用于最终消息模板。

六、执行完成:日志、提示词模板与 Response

等待结束后,execute 依次完成三件事:

  1. 若存在日志对象,更新日志标题为 self.get_heading("Done", done=True)tools/wait.py);
  2. 通过 self.agent.read_prompt("fw.wait_complete.md", target_time=...) 加载完成消息模板(tools/wait.py)。模板位于 prompts/fw.wait_complete.md,内容为:Wait complete. Reached {{target_time}}.——target_time 会替换为序列化后的实际目标时间;
  3. 返回 Response(message=message, break_loop=False),将模板渲染结果作为工具输出写入历史。

read_prompt 的定义在 agent.py,用于按文件名加载 prompts 目录下的提示词模板并进行变量替换。

七、日志对象与标题约定

get_log_objecttools/wait.py)创建一条类型为 progress 的日志:

return self.agent.context.log.log(
    type="progress",
    heading=self.get_heading(),
    content="",
    kvps=self.args,
)

注意它与基类 Tool.get_log_objecthelpers/tool.py,类型为 tool、标题含 icon://construction)不同——wait 使用 progress 类型和 icon://timer 图标,这符合其"进度展示"语义。DOX 文档将"设置/状态持久化"列为 wait 工具的可观测副作用区域,即上述日志对象会进入 Agent 的上下文日志系统,等待期间不断更新的标题(含剩余时间)也会随历史持久化。

get_headingtools/wait.py)的标题格式为:

icon://timer Wait: {text}{done_icon}

其中 done_icondone=True 时为 icon://done_all,默认文本为 Waiting...。等待过程中标题依次呈现 Wait: Waiting...Wait: 12.5s remainingWait: Done icon://done_all,用户界面(WebUI)据此渲染实时的进度语义。

八、验证与测试关注点

DOX 文档的 Verification 一节给出验证指引:行为变更时应运行针对性的工具与提示词契约测试,无聚焦测试时进行 Agent 执行的冒烟测试。

虽然 DOX 列出的关联测试清单(如 tests/test_api_chat_lifetime.pytests/test_chat_compaction.py 等)多为跨模块回归项,但 tests/test_parallel_tool.py 中可以直接看到 wait 语义被并行工具测试所引用的证据:FakeWaitTool 模拟了 wait 工具的 get_log_object 契约——type="progress"heading="icon://timer Wait: Waiting..."kvps=self.args,与真实实现完全一致。这说明 wait 工具的日志契约(progress 类型 + timer 图标 + 参数快照)是被其他子系统依赖的稳定接口。

从源码结构可以推断,针对 wait 行为本身(如 format_remaining_time 的边界输出、managed_wait 的暂停补偿逻辑)属于高价值单测点,开发者新增或修改行为时应优先补充此类聚焦测试。

九、使用场景与实操要点

综合以上分析,wait 工具在 Agent Zero 中的典型用法与注意事项可归纳如下:

维度 说明
时长等待 seconds/minutes/hours/days 的任意组合,总时长必须为正
时间戳等待 until,值为 ISO 格式本地时间字符串(如 2026-09-14T02:00:00),无时区信息时按用户配置时区解释
干预响应 等待开始前与等待循环内都会调用 handle_intervention,用户暂停/中止可随时生效
时长补偿 时长模式下,干预导致的中断超过 1.5 秒会自动顺延目标时间,保证净等待时长
提示词约束 仅当等待确实是任务的一部分时才使用,避免无意义空转
日志契约 progress 类型日志,icon://timer 标题,参数快照进 kvps,剩余时间实时更新
完成反馈 完成后加载 fw.wait_complete.md 模板,报告实际到达的目标时间,break_loop=False 继续主循环

wait 工具的设计展示了 Agent Zero 工具开发的两个核心原则(DOX 文档 Work Guidance 一节):输出保持简洁、模型可读、安全地持久化到历史长时间运行或用户可见的操作必须尊重干预流程。理解 wait 的实现,也就理解了整个 tools/ 目录下所有长耗时工具(如并行、调度、搜索类工具)共同遵循的框架契约。

十、扩展阅读

登录后查看全文
热门项目推荐
相关项目推荐

项目优选

收起
kernelkernel
deepin linux kernel
C
34
18
ops-transformerops-transformer
本项目是CANN提供的transformer类大模型算子库,实现网络在NPU上加速计算。
C++
1.21 K
2.81 K
ops-nnops-nn
本项目是CANN提供的神经网络类计算算子库,实现网络在NPU上加速计算。
C++
948
1.86 K
docsdocs
暂无描述
Markdown
906
5.84 K
kernelkernel
openEuler内核是openEuler操作系统的核心,既是系统性能与稳定性的基石,也是连接处理器、设备与服务的桥梁。
C
537
610
pytorchpytorch
作为 Ascend for PyTorch 社区的核心组件,TorchNPU 是昇腾专为 PyTorch 打造的深度学习适配插件,使 PyTorch 框架能够直接调用昇腾 NPU,为开发者提供昇腾 AI 处理器的超强算力。
Python
864
1.36 K
jiuwenswarmjiuwenswarm
JiuwenSwarm 是一款基于openJiuwen开发的智能AI Agent,它能够将大语言模型的强大能力,通过你日常使用的各类通讯应用,直接延伸至你的指尖。
Python
4.28 K
1.04 K
ops-mathops-math
本项目是CANN提供的数学类基础计算算子库,实现网络在NPU上加速计算。
C++
1.39 K
1.48 K
AscendNPU-IRAscendNPU-IR
AscendNPU-IR是基于MLIR(Multi-Level Intermediate Representation)构建的,面向昇腾亲和算子编译时使用的中间表示,提供昇腾完备表达能力,通过编译优化提升昇腾AI处理器计算效率,支持通过生态框架使能昇腾AI处理器与深度调优
C++
550
401
flutter_flutterflutter_flutter
本仓库是 Flutter SDK 与 Flutter Engine 的 OpenHarmony 适配版本,由 CPF-Flutter 团队维护。开发者可使用熟悉的 Flutter 技术栈开发 OpenHarmony 应用,3.35.7 及以后的适配版本可基于本仓库源码构建支持 OpenHarmony 的 Flutter Engine。
Dart
1.19 K
348