ag2-subagent-delegation

子 Agent 委派

适用场景

• “协调者 + 专家”——父 Agent 应将任务的部分内容交给研究 Agent、数学 Agent 等。
• “并行分派，然后汇总”——多部分问题中的各个部分彼此独立，并行执行可以缩短耗时。
• “自我委派”——一个 Agent 将复杂工作拆分成多个专注的子任务，供自己处理。

两种模式

模式	适用场景	API
自动注入 run_subtask / run_subtasks	轻量级自我委派、动态并行分派、并行处理子问题	在父 Agent 上设置 tasks=TaskConfig(...)
Agent.as_tool()	需要 LLM 推理并调用具有明确名称的委派 Agent（“调用研究员”“调用作者”）	将子 Agent 封装为父 Agent 上的工具

两种模式可以组合使用，协调者可以同时拥有两者。

模式 1——自动注入 run_subtasks

子任务工具默认关闭（tasks=False）。通过设置 tasks=TaskConfig(...) 启用后，Agent 将获得：

• run_subtask(task: str)——一个独立的子任务 Agent。
• run_subtasks(tasks: list[str], parallel: bool = True)——在一次工具调用中并行分派多个子任务（默认并发执行）。

from ag2 import Agent, TaskConfig
from ag2.config import GeminiConfig

config = GeminiConfig(model="gemini-3-flash-preview")

coordinator = Agent(
    "coordinator",
    prompt=(
        "You answer multi-part questions by dispatching run_subtasks "
        "with parallel=True. Use one tool call with every sub-question "
        "packed into the 'tasks' list."
    ),
    config=config,
    tasks=TaskConfig(),  # opt in
)

reply = await coordinator.ask(
    "In one run_subtasks call, answer: "
    "(a) tallest waterfall, (b) Eiffel Tower year, (c) boiling point of nitrogen."
)

TaskConfig 用于控制子任务 Agent 的构建方式：

@dataclass
class TaskConfig:
    config: ModelConfig | None = None    # falls back to parent's config
    prompt: str = "You are a task agent..."
    include_tools: Iterable[str] | None = None   # None = inherit all parent tools
    exclude_tools: Iterable[str] = ()
    extra_tools: Iterable[Callable | Tool] = ()

常见配置——子任务使用成本更低的模型，并限制可用工具范围：

TaskConfig(
    config=worker_config,                  # smaller model
    prompt="You are a focused worker; one step only.",
    include_tools=["search", "fetch_url"], # don't expose `summarize` to children
)

子任务 Agent 构建时会设置 tasks=False——它们自身不会获得 run_subtask 工具。因此，递归委派在结构上不可能发生，无需设置深度限制。

模式 2——Agent.as_tool()

将整个 Agent 暴露为 LLM 可以命名并调用的工具：

from ag2 import Agent
from ag2.config import AnthropicConfig

config = AnthropicConfig(model="claude-sonnet-4-6")

researcher = Agent("researcher", prompt="Provide concise factual findings.", config=config, tools=[search_tool])
writer     = Agent("writer", prompt="Turn research into clear prose.", config=config)

coordinator = Agent(
    "coordinator",
    prompt="First delegate research, then pass findings to the writer.",
    config=config,
    tools=[
        researcher.as_tool(description="Research a topic and return findings."),
        writer.as_tool(description="Write an article. Pass research notes in the context parameter."),
    ],
)

协调者的 LLM 会看到 task_researcher 和 task_writer。每次调用有两个参数：

• objective（必填）——子任务要完成的工作。
• context（可选）——父 Agent 希望共享的相关信息。

as_tool() 接受以下参数：

参数	说明
description	展示给 LLM 的工具描述（必填）
name	覆盖默认名称 task_{agent.name}
stream	用于自定义子任务流的 StreamFactory（见下文）
middleware	ToolMiddleware 可调用对象（例如 approval_required）

如需更精细的控制，可直接使用 subagent_tool()：

from ag2.tools.subagents import subagent_tool

coordinator = Agent("coordinator", config=config, tools=[
    subagent_tool(researcher, description="Research a topic."),
])

通过 as_tool() 实现自我委派

如果你想使用有明确名称的自我委派工具（sub_task，而非通用的 run_subtask），可以让 Agent 拥有自己的工具：

analyst = Agent(
    "analyst",
    prompt=(
        "You have search and sub_task tools. "
        "Only use sub_task when the task has clearly independent parts."
    ),
    config=config,
    tools=[search_tool],
)

analyst.add_tool(
    analyst.as_tool(
        description="Break work into a focused sub-task for independent analysis.",
        name="sub_task",
    )
)

递归安全

通过 as_tool() 实现的自我委派可能会递归——子 Agent 拥有相同的 sub_task 工具，因此如果没有防护，LLM 可能会无限链式调用。

最简单安全的做法是，优先使用自动注入的 run_subtask / run_subtasks 路径实现自我委派。通过这种方式生成的子任务会在构建时设置 tasks=False，因此它们没有 run_subtask 工具，从结构上杜绝了递归。

如果确实需要通过 as_tool() 实现递归自我委派，请编写一个工具中间件，在 context.dependencies 中递增深度计数器，并在超过阈值时提前终止调用。subagents 模块会从 ag2.tools.subagents 导出 subagent_tool、background_agent_tool、persistent_stream 和 StreamFactory——在依赖内置的深度限制辅助工具之前，请先确认该模块当前公开的接口。

子任务流

默认情况下，每个子任务都会获得一个全新的 MemoryStream——其历史记录相互隔离，并从空白状态开始。上下文传递方式如下：

内容	行为	原因
依赖项	复制（仅顶层浅拷贝）	实现隔离；在子任务中应将依赖项视为只读
变量	复制；不会同步回父 Agent	保障并发安全——通过 asyncio.gather 并行运行同级任务时，“最后写入者获胜”会静默覆盖值，因此子级修改按设计仅作用于子级
历史记录	使用全新流	保持上下文干净；相关信息通过 context 工具参数传递
工具	从父 Agent 继承（由 TaskConfig 筛选）	子任务需要实际能力来完成工作

persistent_stream()

当子 Agent 可以通过查看之前的调用受益时（例如避免重复搜索），为它提供一个在父上下文中的多次调用之间持续存在的流：

from ag2.tools.subagents import persistent_stream

researcher.as_tool(
    description="Research a topic",
    stream=persistent_stream(),
)

将流 ID 存储在 context.dependencies 中，键为 f"ag:{agent.name}:stream"，并复用父流的存储后端。

自定义工厂

from ag2 import Agent, Context
from ag2.streams.redis import RedisStream

def make_redis_stream(agent: Agent, ctx: Context) -> RedisStream:
    return RedisStream(MY_REDIS_URL, prefix=f"ag2:sub:{agent.name}")

researcher.as_tool(description="Research a topic", stream=make_redis_stream)

深入了解

• 可运行的入门示例：assets/research_squad.py（与 code_examples/05 相对应）——涵盖 run_subtasks(parallel=True) 和 Agent.as_tool()，以及 TaskStarted / TaskCompleted 生命周期事件。
• 完整参考：website/docs/user-guide/subagents.mdx。
• tasks= 构造函数参数（以及 KnowledgeConfig 等）：website/docs/user-guide/agent_harness.mdx。

常见问题

• 忘记启用——默认值为 tasks=False。没有 TaskConfig，也就没有 run_subtask 工具。
• 以为子任务可以通过 run_subtask 递归——它们不能。子任务自身的 tasks=False。如果需要更深层级的树，请使用 Agent.as_tool() 自我委派，并通过手动的深度计数器中间件（见上文“递归安全”）控制递归。
• 期望共享可变变量合并——每个子任务都会复制父级的变量，修改永远不会同步回父级（即使成功也不会）。兄弟任务之间的修改也不会传播。需要传回的结果应通过子任务的返回值传递，而不是通过共享变量。
• 将 dependencies 视为每个子任务独有的变量——只有顶层字典会被复制。其中的可变值仍按引用共享。在子任务中应将依赖项视为只读。
• 在 as_tool() 中未提供 description=——大模型不知道何时调用它。此参数为必填项。
• 在工作需要并发时使用 run_subtasks(parallel=False)——默认值为 True 是有原因的；只有后续任务依赖前序结果时才设为 False。
• 混淆 task_{agent.name} 冲突——如果想使用更短的名称，或为同一 Agent 指定不同的委派对象，请传入 name= 覆盖默认名称。