工具的等待与后台执行

生成报告或委派子任务时，工具可能需要一段时间才能完成。如果 Agent 还有独立工作，可以让工具先返回“处理中”，完成后再把结果交给 Agent。

async def 只表示 Python 异步函数，不会自动让 Agent 越过工具继续。下面用 SDK 的工具注册、steer() 和对话历史组合实现后台执行。

选择等待方式

方式	适合什么情况	示例辅助代码的参数
等待完成	下一步必须依赖结果	wait_seconds=None
直接后台执行	可以先做其他工作	wait_seconds=0
等待一段时间后转后台	希望尽快取得结果，又不想一直等待	wait_seconds=5

这些参数属于下面的示例辅助代码，不是 SDK 内置配置。超时只结束本次等待，原任务继续运行。

生成一份后台报告

完成模型配置，从仓库根目录运行：

python -m examples.runtime.background_tools

示例包含两部分：本页的工具代码，以及共用的 examples/runtime/background_support.py。从仓库运行时无需额外安装辅助包。

"""从仓库根目录运行：python -m examples.runtime.background_tools"""

import asyncio
import os

import bumblehive
from examples.runtime.background_support import BackgroundTools, Conversation


async def main() -> None:
    config = bumblehive.RuntimeArguments(
        model=os.environ["BUMBLEHIVE_MODEL"],
        api_key=os.environ["BUMBLEHIVE_API_KEY"],
        base_url=os.environ["BUMBLEHIVE_BASE_URL"],
        skill_names=[],
        tool_names=["make_report"],
    )
    conversation = Conversation()
    jobs = BackgroundTools(conversation.notify)
    async with bumblehive.from_config(config) as runtime:
        @runtime.tools.tool(description=(
            "生成统计报告。background=true 立即后台执行；false 等待完成。"
            "pending 只是处理中，结果随后会作为消息送达；不要重复发起。"
        ))
        async def make_report(background: bool = True) -> dict:
            async def work() -> dict:
                await asyncio.sleep(3)  # 模拟耗时计算。
                return {"orders": 12, "total_yuan": 360}

            return await jobs.submit(
                work(), wait_seconds=0 if background else None,
            )

        await conversation.run(
            runtime,
            "请调用 make_report 后台生成报告，先说明报告可用于什么，"
            "收到实际结果后再总结数字。不要重复调用。",
            jobs,
        )


if __name__ == "__main__":
    asyncio.run(main())

make_report 用三秒延迟模拟计算。background=True 时立即返回 pending 和任务标识；False 时直接等待报告结果。

运行过程类似：

调用 make_report → 工具返回 pending
Agent 先说明报告的用途
后台计算完成 → 发送包含任务标识的结果
Agent 总结：12 笔订单，总金额 360

如果报告很快完成，结果也可能在主 Agent 给出首次回答前到达。

设置等待时间

将工具中的 submit() 改为：

return await jobs.submit(work(), wait_seconds=1)

三秒任务超过一秒等待，会返回 pending。改为 5 则直接返回报告。关键实现只有：

task = asyncio.ensure_future(work)
done, _ = await asyncio.wait({task}, timeout=wait_seconds)
if done:
    return task.result()
# task 仍在运行，登记完成后的投递，再返回 pending。

这里使用 asyncio.wait()，超时不会取消 task。工具只返回一次；晚到结果通过新消息送达，不会修改之前的工具结果。

结果如何回到 Agent

共用辅助代码做两件事：

• BackgroundTools 管理等待及完成通知，失败也会作为结果通知。
• Conversation 在运行中调用 steer()；本轮结束后则更新历史，收到晚到结果再开始下一轮。

查看完整辅助代码

"""教学示例共用的等待与结果投递逻辑；不属于 SDK 公共 API。"""

import asyncio
import json
from collections.abc import Awaitable, Callable
from typing import Any
from uuid import uuid4

import bumblehive


class BackgroundTools:
    """共用等待逻辑；调用方决定怎样投递晚到结果。"""

    def __init__(self, notify: Callable[[str, Any], None]) -> None:
        self.notify = notify
        self.tasks: list[asyncio.Task] = []
        self.deliveries: list[asyncio.Task] = []

    async def submit(self, work: Awaitable, *, wait_seconds: float | None) -> Any:
        task = asyncio.ensure_future(work)
        self.tasks.append(task)
        if wait_seconds is None:
            return await task
        if wait_seconds > 0:
            # 只结束等待，不取消仍在执行的工作。
            done, _ = await asyncio.wait({task}, timeout=wait_seconds)
            if done:
                return task.result()
        task_id = uuid4().hex
        self.deliveries.append(asyncio.create_task(self._deliver(task_id, task)))
        return {"status": "pending", "task_id": task_id}

    async def _deliver(self, task_id: str, task: asyncio.Future) -> None:
        try:
            value = await task
        except Exception as exc:
            value = {"status": "error", "message": str(exc)}
        self.notify(task_id, value)

    async def drain(self) -> None:
        await asyncio.gather(*self.deliveries)

    async def close(self) -> None:
        for task in self.tasks + self.deliveries:
            if not task.done():
                task.cancel()
        await asyncio.gather(*self.tasks, *self.deliveries, return_exceptions=True)


class Conversation:
    """运行中用 steer 投递；运行结束后用同一份历史继续。"""

    def __init__(self) -> None:
        self.current: bumblehive.AgentRun | None = None
        self.inbox: list[str] = []

    def notify(self, task_id: str, value: Any) -> None:
        content = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, default=str)
        message = f"[Background tool result]\nTask ID: {task_id}\n\n{content}"
        if self.current is not None:
            try:
                self.current.steer(message)
                return
            except RuntimeError:
                pass  # 当前运行已结束，交给下一轮。
        self.inbox.append(message)

    async def run(self, runtime, prompt: str, background: BackgroundTools):
        history = bumblehive.MessageHistory()
        try:
            while True:
                self.current = runtime.run(prompt, history=history)
                result = await self.current
                history.replace_run_messages(result.messages)
                self.inbox.extend(
                    message["content"] for message in self.current.take_pending()
                )
                print(result.error.message if result.error else result.final_content)
                if result.error:
                    return result
                # 保持应用存活，等待后台结果；Agent 的这一轮已经结束。
                await background.drain()
                if not self.inbox:
                    return result
                prompt = "\n\n".join(self.inbox)
                self.inbox.clear()
        finally:
            await background.close()

应用需要保持运行才能接收结果。本例等待所有后台任务完成后退出，任务保存在当前进程中，不提供重启恢复。

同一机制可以用于子 Agent 协作和主动追问。工具描述应明确告诉模型：pending 不是最终答案，不要重复发起同一任务；还有独立工作就继续，必须依赖结果时就等待后续消息。