ag2-subagent-delegation

用途

ag2-subagent-delegation 用于在单个 AG2 Agent 内实现子智能体委派，主要覆盖三类场景：

协调者 + 专家：父智能体把任务的一部分交给研究者、数学智能体等。
扇出后收集：多部分问题中，各部分相互独立，并行执行可节省实际耗时。
自我委派：一个智能体把复杂工作拆成聚焦的子任务交给自己。

主要能力与两种模式

模式 1：自动注入的 run_subtask / run_subtasks

子任务工具默认关闭（tasks=False）。通过父智能体上的 tasks=TaskConfig(...) 选择启用后，智能体获得：

run_subtask(task: str)：一个隔离的子任务智能体。
run_subtasks(tasks: list[str], parallel: bool = True)：在一次工具调用中扇出多个任务（默认并发）。

TaskConfig 控制子任务智能体的构建方式，字段包括：config（模型配置，为 None 时回退到父配置）、prompt（默认 "You are a task agent..."）、include_tools（None 表示继承父智能体全部工具）、exclude_tools、extra_tools。

常见用法是为子任务使用更便宜的模型并收窄工具范围，例如 TaskConfig(config=worker_config, prompt="You are a focused worker; one step only.", include_tools=["search", "fetch_url"])。

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

模式 2：Agent.as_tool()

把一个完整智能体暴露为 LLM 可以具名调用的工具。协调者的 LLM 会看到 task_researcher、task_writer 之类的工具。每次调用有两个参数：objective（必填，子任务要做什么）和 context（可选，父智能体想共享的相关信息）。

as_tool() 接受的参数：description（展示给 LLM 的工具描述，必填）、name（覆盖默认的 task_{agent.name}）、stream（用于自定义子任务流的 StreamFactory）、middleware（ToolMiddleware 可调用对象，例如 approval_required）。

如需更多控制，可直接使用 subagent_tool()：from ag2.tools.subagents import subagent_tool，然后 subagent_tool(researcher, description="Research a topic.")。

通过 as_tool() 实现自我委派

若想要一个具名的自委派（用 sub_task 代替通用的 run_subtask），可以给智能体添加它自己的工具，通过 as_tool(description=..., name="sub_task") 并用 add_tool 添加。

递归安全

通过 as_tool() 的自我委派可能递归——子智能体拥有相同的 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，其历史隔离且从空开始。上下文流转行为：

Dependencies：复制（顶层浅拷贝），隔离；在子任务内应视为只读。
Variables：复制，不会同步回父智能体；并发安全——当兄弟任务通过 asyncio.gather 运行时，last-writer-wins 会静默覆盖值，因此子任务的修改按设计保留在子任务内。
History：全新流，上下文干净；相关信息通过 context 工具参数传递。
Tools：从父智能体继承（由 TaskConfig 过滤），子任务需要真实能力来完成工作。

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)

使用方式

模式 1 示例：

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."
)

模式 2 示例：

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 具名调用特定子智能体。

常见陷阱：

忘记选择启用——tasks=False 是默认值，没有 TaskConfig 就没有 run_subtask 工具。
期望子任务通过 run_subtask 递归——它们不能。子任务自身 tasks=False。若需要更深的树，请用带手动深度计数器中间件的 Agent.as_tool() 自我委派。
期望共享可变变量会合并——每个子任务复制父变量，修改永不同步回父（即使成功也不），兄弟任务的修改也不传播。需要返回的结果应通过子任务返回值传递，而非共享变量。
把 dependencies 当作每个子任务独立的——只复制顶层字典，其中的可变值仍按引用共享。在子任务内应把 dependencies 视为只读。
as_tool() 没有 description=——LLM 不知道何时调用它。该参数必填。
在并发工作时用 run_subtasks(parallel=False)——默认 True 是有原因的；仅当后续任务依赖先前结果时才设为 False。
task_{agent.name} 命名冲突——若想要更短的名字或同一智能体的不同委托，传递 name= 覆盖。

深入参考

可运行的入门示例：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。