等待工具

WaitTool 让智能体在再次检查长时间运行的任务之前暂停。

概述

WaitTool 会按给定的秒数暂停执行。它之所以存在，是因为那些启动长时间运行工作的智能体 ——沙箱构建、部署、批量导入、异步 API 任务—— 否则没有办法让时间流逝。没有它，智能体要么在紧密循环中轮询，要么 在工作完成之前就放弃。

该工具不需要 API key，除标准库之外没有任何依赖。

何时使用

该工具的描述会告诉模型，当带外工作确实需要真实时间来推进时，应当使用它：

• 仍在执行的沙箱构建、测试运行或脚本
• 仍在滚动推进的部署或资源配置步骤
• 批量导入、导出或训练任务
• 返回了 job id、需要稍后轮询的异步 API
• 需要在重试前冷却的限流或退避

引导模型采用的模式是：启动任务、等待、检查状态，如果仍在运行就再次 等待。描述还告诉模型不要为了控制对话节奏而等待，也不要在 所需信息已经可用时等待——等待只是让时钟时间流逝，并不会 推进或检查任务。

安装

该工具随 crewai-tools 一起提供：

uv add crewai-tools

示例

from crewai import Agent, Crew, Task
from crewai.tools import tool
from crewai_tools import WaitTool

wait_tool = WaitTool()

@tool("Check build status")
def check_build_status_tool(build_id: str) -> str:
    """Return the current status of a build: queued, running, passed, or failed."""
    # Replace this with a call to your own build system.
    return my_ci_client.get_build(build_id).status

build_agent = Agent(
    role="Build Monitor",
    goal="Start the build and report its final status",
    backstory="An engineer who knows that builds take time.",
    tools=[wait_tool, check_build_status_tool],
    verbose=True,
)

monitor_task = Task(
    description=(
        "Start the build, then wait and re-check its status until it finishes."
    ),
    expected_output="The final build status.",
    agent=build_agent,
)

crew = Crew(agents=[build_agent], tasks=[monitor_task])
result = crew.kickoff()

参数

参数	类型	是否必填	描述
seconds	float	✅	等待多少秒。必须为零或更大。
reason	str	❌	可选，说明在等待什么。会回显在工具结果中。

初始化参数

参数	类型	默认值	描述
max_seconds	float	300	单次等待的上限。更长的请求会被截断到该值，而不是被拒绝。

限制过长的等待

单次调用最多等待 max_seconds。如果智能体要求的更久，工具会等待 最大值并在结果中说明，这样智能体可以再次调用它，而不必失败：

wait_tool = WaitTool()
wait_tool.run(seconds=3600)
# 'Waited 300 seconds. Requested 3600 seconds, capped at 300 seconds per call -
#  call this tool again if more waiting is needed.'

当工作流确实需要更长的单次暂停时，可以提高上限：

wait_tool = WaitTool(max_seconds=1800)

异步支持

该工具同时实现了同步和异步执行，因此在被 await 时不会阻塞事件循环：

import asyncio

async def main():
    result = await wait_tool.arun(seconds=30, reason="waiting for the sandbox build")
    print(result)

asyncio.run(main())