ADK 中的多智能体系统¶

随着智能体应用程序变得越来越复杂，将它们构建为单一的、整体式智能体可能会变得难以开发、维护和理解。智能体开发套件 (ADK) 支持通过将多个不同的 BaseAgent 实例组合成多智能体系统 (MAS) 来构建复杂的应用程序。

在 ADK 中，多智能体系统是一个应用程序，其中不同的智能体（通常形成层次结构）协作或协调以实现更大的目标。以这种方式构建你的应用程序提供了显著的优势，包括增强的模块化、专业化、可重用性、可维护性，以及使用专用工作流智能体定义结构化控制流的能力。

你可以组合从 BaseAgent 派生的各种类型的智能体来构建这些系统：

LLM 智能体： 由大型语言模型驱动的智能体。（参见 LLM 智能体）
工作流智能体： 专门设计用于管理其子智能体执行流程的智能体（SequentialAgent、ParallelAgent、LoopAgent）。（参见工作流智能体）
自定义智能体： 你自己的继承自 BaseAgent 并具有特殊非 LLM 逻辑的智能体。（参见自定义智能体）

以下部分详细介绍了使你能够有效构建和管理这些多智能体系统的核心 ADK 原语——如智能体层次结构、工作流智能体和交互机制。

2. 用于智能体组合的 ADK 原语¶

ADK 提供了核心构建块——原语——使你能够在多智能体系统中构建和管理交互。

2.1. 智能体层次结构（`parent_agent`、`sub_agents`）¶

多智能体系统结构的基础是 BaseAgent 中定义的父子关系。

建立层次结构： 你通过在初始化父智能体时将智能体实例列表传递给 sub_agents 参数来创建树结构。ADK 在初始化期间自动在每个子智能体上设置 parent_agent 属性（google.adk.agents.base_agent.py - model_post_init）。
单一父级规则： 一个智能体实例只能被添加为一次子智能体。尝试分配第二个父级将导致 ValueError。
重要性： 这种层次结构为工作流智能体定义了范围，并影响 LLM 驱动委托的潜在目标。你可以使用 agent.parent_agent 导航层次结构或使用 agent.find_agent(name) 查找后代。

# 概念示例：定义层次结构
from google.adk.agents import LlmAgent, BaseAgent

# 定义单个智能体
greeter = LlmAgent(name="Greeter", model="gemini-2.0-flash")
task_doer = BaseAgent(name="TaskExecutor") # Custom non-LLM agent

# 创建父智能体并通过 sub_agents 分配子智能体
coordinator = LlmAgent(
    name="Coordinator",
    model="gemini-2.0-flash",
    description="我协调问候和任务。",
    sub_agents=[ # 在此处分配 sub_agents
        greeter,
        task_doer
    ]
)

# 框架自动设置：
# assert greeter.parent_agent == coordinator
# assert task_doer.parent_agent == coordinator

2.2. 工作流智能体作为编排者¶

ADK 包含从 BaseAgent 派生的专门智能体，它们本身不执行任务，而是编排其 sub_agents 的执行流程。

SequentialAgent： 按照列出的顺序依次执行其 sub_agents。

上下文： 顺序传递相同的 InvocationContext，使智能体能够通过共享状态轻松传递结果。

# 概念示例：顺序管道
from google.adk.agents import SequentialAgent, LlmAgent

step1 = LlmAgent(name="Step1_Fetch", output_key="data") # 将输出保存到 state['data']
step2 = LlmAgent(name="Step2_Process", instruction="处理来自状态键 'data' 的数据。")

pipeline = SequentialAgent(name="MyPipeline", sub_agents=[step1, step2])
# 当 pipeline 运行时，Step2 可以访问 Step1 设置的 state['data']。

ParallelAgent： 并行执行其 sub_agents。子智能体的事件可能会交错。

上下文： 为每个子智能体修改 InvocationContext.branch（例如，ParentBranch.ChildName），提供不同的上下文路径，在某些内存实现中对隔离历史很有用。
状态： 尽管有不同的分支，所有并行子智能体都访问相同的共享 session.state，使它们能够读取初始状态并写入结果（使用不同的键以避免竞争条件）。

# 概念示例：并行执行
from google.adk.agents import ParallelAgent, LlmAgent

fetch_weather = LlmAgent(name="WeatherFetcher", output_key="weather")
fetch_news = LlmAgent(name="NewsFetcher", output_key="news")

gatherer = ParallelAgent(name="InfoGatherer", sub_agents=[fetch_weather, fetch_news])
# 当 gatherer 运行时，WeatherFetcher 和 NewsFetcher 并发运行。
# 后续智能体可以读取 state['weather'] 和 state['news']。

LoopAgent： 在循环中顺序执行其 sub_agents。

终止： 如果达到可选的 max_iterations，或者任何子智能体产生带有 actions.escalate=True 的 Event，循环将停止。
上下文和状态： 在每次迭代中传递相同的 InvocationContext，允许状态更改（例如，计数器、标志）在循环中持续存在。

# 概念示例：带条件的循环
from google.adk.agents import LoopAgent, LlmAgent, BaseAgent
from google.adk.events import Event, EventActions
from google.adk.agents.invocation_context import InvocationContext
from typing import AsyncGenerator

class CheckCondition(BaseAgent): # 自定义智能体检查状态
    async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
        status = ctx.session.state.get("status", "pending")
        is_done = (status == "completed")
        yield Event(author=self.name, actions=EventActions(escalate=is_done)) # 如果完成则升级

process_step = LlmAgent(name="ProcessingStep") # 可能更新 state['status'] 的智能体

poller = LoopAgent(
    name="StatusPoller",
    max_iterations=10,
    sub_agents=[process_step, CheckCondition(name="Checker")]
)
# 当 poller 运行时，它反复执行 process_step 然后 Checker，
# 直到 Checker 升级（state['status'] == 'completed'）或经过 10 次迭代。

2.3. 交互与通信机制¶

系统内的智能体通常需要相互交换数据或触发动作。ADK 通过以下方式实现这一点：

a) 共享会话状态（`session.state`）¶

在同一调用中运行的智能体（因此通过 InvocationContext 共享相同的 Session 对象）被动通信的最基本方式。

机制： 一个智能体（或其工具/回调）写入一个值（context.state['data_key'] = processed_data），随后的智能体读取它（data = context.state.get('data_key')）。状态变更通过 CallbackContext 跟踪。
便利性： LlmAgent 上的 output_key 属性自动将智能体的最终响应文本（或结构化输出）保存到指定的状态键。
性质： 异步、被动通信。非常适合由 SequentialAgent 编排的管道或跨 LoopAgent 迭代传递数据。
另见： 状态管理

# 概念示例：使用 output_key 和读取状态
from google.adk.agents import LlmAgent, SequentialAgent

agent_A = LlmAgent(name="AgentA", instruction="找出法国的首都。", output_key="capital_city")
agent_B = LlmAgent(name="AgentB", instruction="告诉我存储在状态键 'capital_city' 中的城市相关信息。")

pipeline = SequentialAgent(name="CityInfo", sub_agents=[agent_A, agent_B])
# AgentA 运行，将 "巴黎" 保存到 state['capital_city']。
# AgentB 运行，其指令处理器从 state['capital_city'] 读取 "巴黎"。

b) LLM 驱动的委托（智能体转移）¶

利用 LlmAgent 的理解能力来动态将任务路由到层次结构中其他合适的智能体。

机制： 智能体的 LLM 生成特定的函数调用：transfer_to_agent(agent_name='target_agent_name')。
处理： 默认情况下，当存在子智能体或未禁止转移时使用的 AutoFlow 会拦截此调用。它使用 root_agent.find_agent() 识别目标智能体并更新 InvocationContext 以切换执行焦点。
要求： 调用的 LlmAgent 需要明确的 instructions，说明何时转移，而潜在的目标智能体需要明确的 description，以便 LLM 做出明智的决定。可以在 LlmAgent 上配置转移范围（父级、子智能体、同级）。
性质： 基于 LLM 解释的动态、灵活路由。

# 概念设置：LLM 转移
from google.adk.agents import LlmAgent

booking_agent = LlmAgent(name="Booker", description="处理航班和酒店预订。")
info_agent = LlmAgent(name="Info", description="提供一般信息并回答问题。")

coordinator = LlmAgent(
    name="Coordinator",
    instruction="你是一个助手。将预订任务委托给 Booker，将信息请求委托给 Info。",
    description="主要协调员。",
    # AutoFlow 通常在这里被隐式使用
    sub_agents=[booking_agent, info_agent]
)
# 如果 coordinator 收到 "预订一个航班"，其 LLM 应该生成：
# FunctionCall(name='transfer_to_agent', args={'agent_name': 'Booker'})
# ADK 框架随后将执行路由到 booking_agent。

c) 显式调用（`AgentTool`）¶

允许 LlmAgent 将另一个 BaseAgent 实例视为可调用函数或工具。

机制： 将目标智能体实例包装在 AgentTool 中并将其包含在父 LlmAgent 的 tools 列表中。AgentTool 为 LLM 生成相应的函数声明。
处理： 当父 LLM 生成针对 AgentTool 的函数调用时，框架执行 AgentTool.run_async。此方法运行目标智能体，捕获其最终响应，将任何状态/制品更改转发回父级的上下文，并将响应作为工具的结果返回。
性质： 同步（在父级的流程内），明确、受控的调用，就像任何其他工具一样。
（注意： 需要显式导入和使用 AgentTool)。

# 概念设置：智能体作为工具
from google.adk.agents import LlmAgent, BaseAgent
from google.adk.tools import agent_tool
from pydantic import BaseModel

# 定义目标智能体（可以是 LlmAgent 或自定义 BaseAgent）
class ImageGeneratorAgent(BaseAgent): # 示例自定义智能体
    name: str = "ImageGen"
    description: str = "根据提示生成图像。"
    # ... 内部逻辑 ...
    async def _run_async_impl(self, ctx): # 简化的运行逻辑
        prompt = ctx.session.state.get("image_prompt", "默认提示")
        # ... 生成图像字节 ...
        image_bytes = b"..."
        yield Event(author=self.name, content=types.Content(parts=[types.Part.from_bytes(image_bytes, "image/png")]))

image_agent = ImageGeneratorAgent()
image_tool = agent_tool.AgentTool(agent=image_agent) # 包装智能体

# 父智能体使用 AgentTool
artist_agent = LlmAgent(
    name="Artist",
    model="gemini-2.0-flash",
    instruction="创建一个提示并使用 ImageGen 工具生成图像。",
    tools=[image_tool] # 包含 AgentTool
)
# Artist LLM 生成一个提示，然后调用：
# FunctionCall(name='ImageGen', args={'image_prompt': '一只戴帽子的猫'})
# 框架调用 image_tool.run_async(...)，它运行 ImageGeneratorAgent。
# 生成的图像 Part 作为工具结果返回给 Artist 智能体。

这些原语提供了设计多智能体交互的灵活性，范围从紧密耦合的顺序工作流到动态、LLM 驱动的委托网络。

3. 使用 ADK 原语的常见多智能体模式¶

通过组合 ADK 的组合原语，你可以实现多智能体协作的各种已建立模式。

协调员/调度员模式¶

结构： 一个中央 LlmAgent（协调员）管理几个专业 sub_agents。
目标： 将传入请求路由到适当的专家智能体。
使用的 ADK 原语：
- 层次结构： 协调员在 sub_agents 中列出专家。
- 交互： 主要使用 LLM 驱动的委托（要求子智能体有明确的 description 和协调员有适当的 instruction）或 显式调用（AgentTool）（协调员在其 tools 中包含 AgentTool 包装的专家）。

# 概念代码：使用 LLM 转移的协调员
from google.adk.agents import LlmAgent

billing_agent = LlmAgent(name="Billing", description="处理账单查询。")
support_agent = LlmAgent(name="Support", description="处理技术支持请求。")

coordinator = LlmAgent(
    name="HelpDeskCoordinator",
    model="gemini-2.0-flash",
    instruction="Route 用户请求：使用 Billing 智能体处理支付问题，使用 Support 智能体处理技术问题。",
    description="Main help desk router.",
    # 在 AutoFlow 中，使用 sub_agents 时，allow_transfer=True 通常是隐式的
    sub_agents=[billing_agent, support_agent]
)
# 用户询问 "我的付款失败了" -> Coordinator 的 LLM 应该调用 transfer_to_agent(agent_name='Billing')
# 用户询问 "我无法登录" -> Coordinator 的 LLM 应该调用 transfer_to_agent(agent_name='Support')

顺序管道模式¶

结构： 一个 SequentialAgent 包含按固定顺序执行的 sub_agents。
目标： 实现一个多步骤流程，其中一个步骤的输出作为下一个步骤的输入。
使用的 ADK 原语：
- 工作流： SequentialAgent 定义顺序。
- 通信： 主要使用共享会话状态。较早的智能体写入结果（通常通过 output_key），较晚的智能体从 context.state 读取这些结果。

# Conceptual Code: Sequential Data Pipeline
from google.adk.agents import SequentialAgent, LlmAgent

validator = LlmAgent(name="ValidateInput", instruction="Validate the input.", output_key="validation_status")
processor = LlmAgent(name="ProcessData", instruction="Process data if state key 'validation_status' is 'valid'.", output_key="result")
reporter = LlmAgent(name="ReportResult", instruction="Report the result from state key 'result'.")

data_pipeline = SequentialAgent(
    name="DataPipeline",
    sub_agents=[validator, processor, reporter]
)
# validator runs -> saves to state['validation_status']
# processor runs -> reads state['validation_status'], saves to state['result']
# reporter runs -> reads state['result']

并行扇出/聚合模式¶

结构： 一个ParallelAgent并发运行多个sub_agents，通常后面跟着一个 (在SequentialAgent中的) 聚合结果的智能体。
目标： 同时执行独立任务以减少延迟，然后合并它们的输出。
使用的 ADK 原语：
- 工作流： 使用ParallelAgent进行并发执行 (扇出)。通常嵌套在SequentialAgent中以处理后续的聚合步骤 (聚合)。
- 通信： 子智能体将结果写入共享会话状态中的不同键。随后的"聚合"智能体读取多个状态键。

# 概念代码：并行信息收集
from google.adk.agents import SequentialAgent, ParallelAgent, LlmAgent

fetch_api1 = LlmAgent(name="API1Fetcher", instruction="从 API 1 获取数据。", output_key="api1_data")
fetch_api2 = LlmAgent(name="API2Fetcher", instruction="从 API 2 获取数据。", output_key="api2_data")

gather_concurrently = ParallelAgent(
    name="ConcurrentFetch",
    sub_agents=[fetch_api1, fetch_api2]
)

synthesizer = LlmAgent(
    name="Synthesizer",
    instruction="合并来自状态键'api1_data'和'api2_data'的结果。"
)

overall_workflow = SequentialAgent(
    name="FetchAndSynthesize",
    sub_agents=[gather_concurrently, synthesizer] # 运行并行获取，然后合成
)
# fetch_api1 和 fetch_api2 并发运行，将结果保存到状态中。
# synthesizer 随后运行，读取 state['api1_data'] 和 state['api2_data']。

层次任务分解¶

结构： 一个多层级智能体树，其中高层级智能体分解复杂目标并将子任务委派给低层级智能体。
目标： 通过递归地将复杂问题分解为更简单、可执行的步骤来解决问题。
使用的 ADK 原语：
- 层次结构： 多层级的parent_agent/sub_agents结构。
- 交互： 主要使用LLM 驱动的委托或显式调用 (AgentTool)由父智能体分配任务给子智能体。结果通过工具响应或状态向上返回层次结构。

# 概念代码：层次化研究任务
from google.adk.agents import LlmAgent
from google.adk.tools import agent_tool

# 低层级类工具智能体
web_searcher = LlmAgent(name="WebSearch", description="执行网络搜索以获取事实。")
summarizer = LlmAgent(name="Summarizer", description="对文本进行摘要。")

# 中层级智能体组合工具
research_assistant = LlmAgent(
    name="ResearchAssistant",
    model="gemini-2.0-flash",
    description="查找并总结关于某个主题的信息。",
    tools=[agent_tool.AgentTool(agent=web_searcher), agent_tool.AgentTool(agent=summarizer)]
)

# 高层级智能体委派研究
report_writer = LlmAgent(
    name="ReportWriter",
    model="gemini-2.0-flash",
    instruction="撰写关于主题 X 的报告。使用 ResearchAssistant 收集信息。",
    tools=[agent_tool.AgentTool(agent=research_assistant)]
    # 或者，如果 research_assistant 是 sub_agent，可以使用 LLM Transfer
)
# 用户与 ReportWriter 交互。
# ReportWriter 调用 ResearchAssistant 工具。
# ResearchAssistant 调用 WebSearch 和 Summarizer 工具。
# 结果向上流回。

评审/批评模式（生成 - 评论者）¶

结构： 通常在SequentialAgent中包含两个智能体：一个生成者和一个评论者/审核者。
目标： 通过专门的智能体审核来提高生成输出的质量或有效性。
使用的 ADK 原语：
- 工作流： SequentialAgent确保生成在审核之前发生。
- 通信： 共享会话状态（生成者使用output_key保存输出；审核者读取该状态键）。审核者可能将其反馈保存到另一个状态键，供后续步骤使用。

# 概念代码：生成 - 评论者
from google.adk.agents import SequentialAgent, LlmAgent

generator = LlmAgent(
    name="DraftWriter",
    instruction="写一个关于主题 X 的简短段落。",
    output_key="draft_text"
)

reviewer = LlmAgent(
    name="FactChecker",
    instruction="检查状态键'draft_text'中的文本是否事实准确。输出'valid'或'invalid'以及理由。",
    output_key="review_status"
)

# 可选：基于 review_status 的进一步步骤

review_pipeline = SequentialAgent(
    name="WriteAndReview",
    sub_agents=[generator, reviewer]
)
# generator 运行 -> 将草稿保存到 state['draft_text']
# reviewer 运行 -> 读取 state['draft_text']，将状态保存到 state['review_status']

迭代改进模式¶

结构： 使用LoopAgent包含一个或多个在多次迭代中处理任务的智能体。
目标： 逐步改进存储在会话状态中的结果（例如代码、文本、计划），直到达到质量阈值或达到最大迭代次数。
使用的 ADK 原语：
- 工作流： LoopAgent管理重复过程。
- 通信： 共享会话状态对于智能体读取前一次迭代的输出和保存改进版本至关重要。
- 终止： 循环通常基于max_iterations或专门的检查智能体在结果令人满意时设置actions.escalate=True而结束。

# 概念代码：迭代代码改进
from google.adk.agents import LoopAgent, LlmAgent, BaseAgent
from google.adk.events import Event, EventActions
from google.adk.agents.invocation_context import InvocationContext
from typing import AsyncGenerator

# 基于 state['current_code'] 和 state['requirements'] 生成/改进代码的智能体
code_refiner = LlmAgent(
    name="CodeRefiner",
    instruction="读取 state['current_code']（如果存在）和 state['requirements']。生成/改进 Python 代码以满足需求。保存到 state['current_code']。",
    output_key="current_code" # 覆盖状态中的前一个代码
)

# 检查代码是否满足质量标准的智能体
quality_checker = LlmAgent(
    name="QualityChecker",
    instruction="根据 state['requirements'] 评估 state['current_code'] 中的代码。输出'pass'或'fail'。",
    output_key="quality_status"
)

# 自定义智能体检查状态并在'pass'时升级
class CheckStatusAndEscalate(BaseAgent):
    async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
        status = ctx.session.state.get("quality_status", "fail")
        should_stop = (status == "pass")
        yield Event(author=self.name, actions=EventActions(escalate=should_stop))

refinement_loop = LoopAgent(
    name="CodeRefinementLoop",
    max_iterations=5,
    sub_agents=[code_refiner, quality_checker, CheckStatusAndEscalate(name="StopChecker")]
)
# 循环运行：Refiner -> Checker -> StopChecker
# State['current_code'] 在每次迭代中更新。
# 如果 QualityChecker 输出'pass'（导致 StopChecker 升级）或 5 次迭代后，循环停止。

人机协作模式¶

结构： 在智能体工作流中集成人类干预点。
目标： 允许人类监督、批准、纠正或执行 AI 无法完成的任务。
使用的 ADK 原语（概念）：
- 交互： 可以通过自定义工具实现，该工具暂停执行并向外部系统（例如 UI、工单系统）发送请求，等待人类输入。然后该工具将人类的响应返回给智能体。
- 工作流： 可以使用LLM 驱动的委托（transfer_to_agent）指向概念性的"人类智能体"触发外部工作流，或在LlmAgent中使用自定义工具。
- 状态/回调： 状态可以保存人类任务详情；回调可以管理交互流程。
- 注意： ADK 没有内置的"人类智能体"类型，所以这需要自定义集成。

# 概念代码：使用工具进行人类审批
from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.tools import FunctionTool

# --- 假设 external_approval_tool 存在 ---
# 此工具将：
# 1. 接收详细信息（例如 request_id、amount、reason）。
# 2. 通过 API 将这些详情发送到人类审核系统。
# 3. 轮询或等待人类响应（approved/rejected）。
# 4. 返回人类的决定。
# async def external_approval_tool(amount: float, reason: str) -> str: ...
approval_tool = FunctionTool(func=external_approval_tool)

# 准备请求的智能体
prepare_request = LlmAgent(
    name="PrepareApproval",
    instruction="根据用户输入准备审批请求详情。将金额和理由存储在状态中。",
    # ... 可能设置 state['approval_amount'] 和 state['approval_reason'] ...
)

# 调用人类审批工具的智能体
request_approval = LlmAgent(
    name="RequestHumanApproval",
    instruction="使用 external_approval_tool，参数来自 state['approval_amount'] 和 state['approval_reason']。",
    tools=[approval_tool],
    output_key="human_decision"
)

# 基于人类决定处理的智能体
process_decision = LlmAgent(
    name="ProcessDecision",
    instruction="检查状态键'human_decision'。如果'approved'，继续。如果'rejected'，通知用户。"
)

approval_workflow = SequentialAgent(
    name="HumanApprovalWorkflow",
    sub_agents=[prepare_request, request_approval, process_decision]
)

这些模式为构建多智能体系统提供了起点。你可以根据需要混合和匹配它们，以创建最适合你特定应用的架构。