子 Agent 协作¶

把独立任务交给另一个 Agent，再使用它的结果完成主任务。实现方式是将一次 runtime.run() 封装成工具，也就是 Agent as Tool。

委派任务并等待结果¶

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

python examples/runtime/multi_agent.py

import asyncio
import os

import bumblehive


READ_ONLY_TOOLS = ("read_file", "list_dir", "find_files", "grep")


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=[],
        workspace=".",
        tool_names=["sub_agent"],
    )

    async with bumblehive.from_config(config) as runtime:
        @runtime.tools.tool(
            name="sub_agent",
            description="把独立的文件阅读任务交给子 Agent，等待并返回它的回答。",
        )
        async def sub_agent(task: str) -> str:
            response = await runtime.run(
                task,
                config={"agent": {"tool_names": READ_ONLY_TOOLS}},
            )
            if response.error:
                raise RuntimeError(response.error.message)
            return response.final_content or ""

        result = await runtime.run(
            "请让子 Agent 阅读 README_zh.md，用三句话概括这个项目的用途。"
        )

    print(result.final_content)


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

主 Agent 调用 sub_agent，子 Agent 阅读 README_zh.md 并概括项目用途，主 Agent 再整理回答。

子调用没有传入历史，因此不会自动继承主 Agent 的对话。任务描述应包含它需要的信息。
子调用只开放文件读取和搜索工具。它仍复用 Runtime 的模型配置与工作目录。
子调用返回的文本就是工具结果，主 Agent 会等待它完成。

后台委派¶

如果主 Agent 还有独立工作，可以让子任务后台执行。下面复用后台工具的示例辅助代码：

python -m examples.runtime.multi_agent_background

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

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=["sub_agent"],
    )
    conversation = Conversation()
    jobs = BackgroundTools(conversation.notify)
    async with bumblehive.from_config(config) as runtime:
        @runtime.tools.tool(description=(
            "把独立任务交给子 Agent。background 默认 true，立即返回 pending，"
            "完成结果随后作为消息送达，不要重复委派。"
            "下一步必须依赖结果时传 false，等待结果再继续。"
        ))
        async def sub_agent(task: str, background: bool = True) -> dict:
            async def work() -> dict:
                result = await runtime.run(
                    task, config={"agent": {
                        "instructions": "你是写作助手。直接给出简短答案，不使用工具。",
                        "tool_names": [],
                        "skill_names": [],
                    }},
                )
                if result.error:
                    raise RuntimeError(result.error.message)
                return {"answer": result.final_content}

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

        await conversation.run(
            runtime,
            "准备一篇 Python 入门文章。把“用 100 字以内设计一道列表练习题，不写参考答案”"
            "交给子 Agent 后台完成。你先列三个提纲要点，收到练习题后简短整合。只委派一次。",
            jobs,
        )


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

运行过程类似：

主 Agent：委派练习题 → 拟文章提纲
子 Agent：设计练习题 → 结果送回主 Agent
主 Agent：将提纲与练习题整合

本例的 sub_agent 默认 background=True。如果下一步依赖结果，传 background=False，工具就等待完成后返回。可以将提示词改为“先让子 Agent 设计练习题，调用时传 background=false，再围绕练习题组织文章”来体验。

sub_agent 和 background 都是示例注册的工具及参数，不是 runtime.run() 的内置选项。等待、后台返回和后续投递的共同机制只在后台工具教程中解释。

子调用若需要人工审批，应像普通调用一样传入 approval_handler，主调用的回调不会自动传给它。