前面我们已经实战过：

• 《AI - 使用 Google ADK 创建你的第一个 AI Agent》LlmAgent：一个模型 + 一段提示词，做一件事
• 《AI - SequentialAgent 实战：用 ADK 把「写代码 → 审查 → 重构」串成一条流水线》SequentialAgent：按顺序跑子 agent
• 《AI - LoopAgent 实战：用 ADK 做一个 “写作-批改-再写” 的迭代写作智能体》LoopAgent：在一组子 agent 上循环
• 《AI - ParallelAgent 实战：用并行工作流做一个「多主题 Web 调研」Agent》ParallelAgent：把子 agent 并行跑
• 《AI - Custom Agent 实战：当 Sequential / Loop / Parallel 都不够用了，该怎么办？》Custom Agent：自定义 agent 执行逻辑

今天我们来了解一下 Multi-Agent Systems 多智能体系统，如何设计自己的 Multi-Agent Systems。

当你的 Agent 越做越复杂时，很容易走向“单体巨兽”：

• 一个 prompt 里塞了所有能力（写作、检索、总结、审核、执行……）
• 工具越来越多，逻辑越来越绕
• 维护困难：改一处，另一处就坏

ADK 的官方建议是：把应用拆成多个职责清晰的小 Agent，让它们协作完成更大目标——这就是 Multi-Agent System（MAS，多智能体系统）。

一、什么是 ADK 的 Multi-Agent System？

MAS 本质上就是：把多个不同的 BaseAgent 实例组合起来，形成一个层级结构，让它们协作/调度去完成一个更大的目标。

把系统拆成多 Agent 的好处包括：

• 模块化：每个 Agent 管一块职责，代码/提示词更小、更好维护
• 专业化：每个 Agent 可以用不同的指令、工具、甚至不同模型
• 可复用：某个“研究员 Agent / 总结 Agent”可以被多个系统复用
• 可维护：问题更容易定位到某个环节/某个 Agent
• 可控的工作流：用 Workflow Agents（Sequential/Loop/Parallel）把执行流程结构化起来

ADK 里你可以混合使用这些 Agent 类型来组成 MAS：

• LLM Agents：由大模型驱动的 Agent
• Workflow Agents：只负责流程调度（Sequential / Parallel / Loop）
• Custom Agents：你自己继承 BaseAgent 写的非 LLM 逻辑 Agent

二、搭 MAS 的三块“积木”

构建 MAS 的关键能力归纳为三类“原语/积木”：

• Agent Hierarchy（层级：parent / sub_agents）
• Workflow Agents（工作流编排器）
• Interaction & Communication（交互与通信机制）

1. Agent Hierarchy：用 sub_agents 搭一棵“Agent 树”

需要记住的 3 条规则
① 用 sub_agents=[…] 建树
当你在初始化一个父 Agent 时传入 sub_agents，ADK 会自动把每个子 Agent 的 parent_agent 指向这个父 Agent。

② 一个 Agent 实例只能有一个 parent（单父规则）
同一个 Agent 实例如果被挂到第二个 parent 上，会抛 ValueError。

③ 这棵树很重要
层级结构会影响：
- Workflow Agent 的作用范围
- LLM 驱动的转交（delegation）能转交给谁
- 你可以用 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")  # 自定义非 LLM agent（示意）

coordinator = LlmAgent(
    name="Coordinator",
    model="gemini-2.0-flash",
    description="I coordinate greetings and tasks.",
    sub_agents=[greeter, task_doer],
)
# ADK 会自动设置：
# greeter.parent_agent == coordinator
# task_doer.parent_agent == coordinator

这段代码想表达的重点：层级关系不是你手动赋值的，是框架根据 sub_agents 自动建立的。

2. Workflow Agents：把它们当成“流程导演”

把 Sequential/Parallel/Loop 这类 Agent 定位为：

自己不干活，只负责 orchestrate（编排）子 agent 的执行流。

• SequentialAgent：排队执行

  • 依次执行 sub_agents
  • 关键点：顺序执行时会传递同一个 InvocationContext，所以特别适合用共享 state 在步骤间传数据。

• ParallelAgent：并行执行

  • 同时运行多个子 agent（降低延迟）
  • 并行分支会共享同一个 session.state（所以写 state 时要用不同 key，避免覆盖/竞态）

• LoopAgent：循环执行

  • 重复执行子 agent
  • 通常靠：
    • max_iterations 上限
    • 或者某个“检查条件 agent”通过 EventActions(escalate=True) 告诉 Loop “可以停了”

一个“检查状态决定是否退出”的自定义 agent 概念例子：

class CheckCondition(BaseAgent):
    async def _run_async_impl(self, ctx):
        status = ctx.session.state.get("status", "pending")
        is_done = (status == "completed")
        yield Event(author=self.name, actions=EventActions(escalate=is_done))

思是：Loop 并不会自己知道何时结束，你要设计一个明确的退出信号。

3、Agent 之间怎么“传话”？三种通信机制

多 Agent 协作时最关键的问题就是：它们怎么交换信息、怎么触发彼此动作？
MAS 的交互机制分成三类：

a) Shared Session State：共享 session.state（最常用、最稳）

这是最基础的通信方式：同一次 invocation 里，共享同一个 Session（通过 InvocationContext 拿到），所以可以用 session.state 传数据。
父 agent 调子 agent 时会传同一个 InvocationContext，因此也会共享同一份临时的 temp: state，适合“只在当前轮对话有意义”的数据。

• A Agent 把结果写进 session.state[“key”]
• B Agent 在自己的 prompt 里通过 {key} 或在代码里 state.get(“key”) 读出来
• LlmAgent.output_key 可以自动把最终输出写进 state，省掉手动赋值

适用场景：流水线、汇总、评审、迭代优化——基本 80% 的 multi-agent 都会用到它

概念示例：

from google.adk.agents import LlmAgent, SequentialAgent

agent_A = LlmAgent(
    name="AgentA",
    instruction="Find the capital of France.",
    output_key="capital_city"
)

agent_B = LlmAgent(
    name="AgentB",
    instruction="Tell me about the city stored in {capital_city}."
)

pipeline = SequentialAgent(name="CityInfo", sub_agents=[agent_A, agent_B])
# A 写 state['capital_city']，B 读 {capital_city}

你可以把它理解为：state 就是一块共享白板。上一步写，下一步读。

b) LLM-Driven Delegation：让模型“转交任务”（Agent Transfer）

这就是很多人喜欢的“让模型自己决定交给谁”的路由方式：

• 机制是：LLM 生成一个函数调用：transfer_to_agent(agent_name=‘target_agent_name’)
• 处理方式是：默认情况下（当你存在 sub_agents，且没有禁用 transfer），ADK 的 AutoFlow 会拦截这个调用：
  • 用 root_agent.find_agent() 找到目标 agent
  • 更新 InvocationContext，把执行焦点切换到目标 agent

要用好它，强调两点：

• 需要父 agent 的 instruction 写清楚“什么时候该转交”
• 需要子 agent 的 description 足够清晰区分，否则模型不知道该选谁

适用场景：客服路由、意图分发、动态选择专家（更像“调度台”）。

概念示例：

billing_agent = LlmAgent(name="Billing", description="Handles billing inquiries.")
support_agent = LlmAgent(name="Support", description="Handles technical support requests.")

coordinator = LlmAgent(
    name="HelpDeskCoordinator",
    model="gemini-2.0-flash",
    instruction="Use Billing for payment issues, Support for technical problems.",
    sub_agents=[billing_agent, support_agent],
)
# 用户说 “payment failed” -> 模型应生成 transfer_to_agent(...Billing...)

c) Explicit Invocation：把 Agent 当“工具”显式调用（AgentTool）

如果你不想让模型自由转交，而是希望它像调用工具一样“明确调用某个 agent”，就用 AgentTool。
它的关键点：

• 把一个 Agent（甚至是 custom agent）包装成 AgentTool
• 然后另一个 LlmAgent 把它当工具用（LLM 发起 tool call）
• 框架运行被包装的 agent，把结果作为 tool result 返回给调用者

适用场景：想要强约束、可观测、可测试的调用链（尤其在复杂系统里更稳）。

概念示例（“把 ImageGen Agent 当工具”）：

from google.adk.tools import agent_tool

image_tool = agent_tool.AgentTool(agent=image_agent)  # 把 agent 包成工具

artist_agent = LlmAgent(
    name="Artist",
    model="gemini-2.0-flash",
    instruction="Create a prompt and use the ImageGen tool to generate the image.",
    tools=[image_tool],
)
# LLM 会像调用工具一样：FunctionCall(name='ImageGen', args={...})

你可以把它理解成：

Transfer 更像“把对话控制权交给另一个 agent”， AgentTool 更像“我（父 agent）调用你（子
agent）给我一个结果”。

三、8 种常见 Multi-Agent 模式（Patterns）

这一部分非常实用，你可以直接把它当“架构模板库”。

1 Coordinator/Dispatcher：总控路由到专家

• 结构：一个中央 Coordinator（LlmAgent）+ 多个专家 sub_agents
• 目标：把请求路由给正确专家
• 常用交互：LLM-driven delegation 或 AgentTool（显式调用）

适用：Helpdesk、IT 支持、企业内部知识问答分发。

# Conceptual Code: Coordinator using LLM Transfer
from google.adk.agents import LlmAgent

billing_agent = LlmAgent(name="Billing", description="Handles billing inquiries.")
support_agent = LlmAgent(name="Support", description="Handles technical support requests.")

coordinator = LlmAgent(
    name="HelpDeskCoordinator",
    model="gemini-2.0-flash",
    instruction="Route user requests: Use Billing agent for payment issues, Support agent for technical problems.",
    description="Main help desk router.",
    # allow_transfer=True is often implicit with sub_agents in AutoFlow
    sub_agents=[billing_agent, support_agent]
)
# User asks "My payment failed" -> Coordinator's LLM should call transfer_to_agent(agent_name='Billing')
# User asks "I can't log in" -> Coordinator's LLM should call transfer_to_agent(agent_name='Support')

2 Sequential Pipeline：顺序流水线

• 结构：SequentialAgent 固定顺序跑 sub_agents
• 目标：多步骤处理，前一步输出喂给下一步
• 通信：主要靠共享 state（尤其 output_key）

适用：数据提取 → 清洗 → 总结、写初稿 → 审稿 → 改稿。

# 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 {validation_status} is 'valid'.", output_key="result")
reporter = LlmAgent(name="ReportResult", instruction="Report the result from {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']

3 Parallel Fan-Out/Gather：并行扇出再汇总

• 结构：ParallelAgent 并行跑多个 sub_agents，常见做法是外面再套 SequentialAgent 做“汇总/聚合”
• 目标：并发执行独立任务降低延迟，然后合并输出
• 通信：每个并行子任务写入不同的 state key，后续 gather agent 读取多个 key

适用：多主题调研、多数据源检索、多份文档并行摘要。

# Conceptual Code: Parallel Information Gathering
from google.adk.agents import SequentialAgent, ParallelAgent, LlmAgent

fetch_api1 = LlmAgent(name="API1Fetcher", instruction="Fetch data from API 1.", output_key="api1_data")
fetch_api2 = LlmAgent(name="API2Fetcher", instruction="Fetch data from API 2.", output_key="api2_data")

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

synthesizer = LlmAgent(
    name="Synthesizer",
    instruction="Combine results from {api1_data} and {api2_data}."
)

overall_workflow = SequentialAgent(
    name="FetchAndSynthesize",
    sub_agents=[gather_concurrently, synthesizer] # Run parallel fetch, then synthesize
)
# fetch_api1 and fetch_api2 run concurrently, saving to state.
# synthesizer runs afterwards, reading state['api1_data'] and state['api2_data'].

4 Hierarchical Task Decomposition：分层任务分解

• 结构：多层 agent 树（上层拆解任务，下层执行子任务）
• 目标：把复杂目标递归拆成可执行步骤
• 交互：常用 transfer 或 AgentTool；结果向上返回（tool response 或 state）

“研究-写报告”例子很典型：

• ReportWriter（高层）调用 ResearchAssistant（中层）
• ResearchAssistant 再调用 WebSearch / Summarizer（底层）
• 结果逐层回传

适用：需要“组织结构”的复杂任务，比如研究、规划、报告生成。

# Conceptual Code: Hierarchical Research Task
from google.adk.agents import LlmAgent
from google.adk.tools import agent_tool

# Low-level tool-like agents
web_searcher = LlmAgent(name="WebSearch", description="Performs web searches for facts.")
summarizer = LlmAgent(name="Summarizer", description="Summarizes text.")

# Mid-level agent combining tools
research_assistant = LlmAgent(
    name="ResearchAssistant",
    model="gemini-2.0-flash",
    description="Finds and summarizes information on a topic.",
    tools=[agent_tool.AgentTool(agent=web_searcher), agent_tool.AgentTool(agent=summarizer)]
)

# High-level agent delegating research
report_writer = LlmAgent(
    name="ReportWriter",
    model="gemini-2.0-flash",
    instruction="Write a report on topic X. Use the ResearchAssistant to gather information.",
    tools=[agent_tool.AgentTool(agent=research_assistant)]
    # Alternatively, could use LLM Transfer if research_assistant is a sub_agent
)
# User interacts with ReportWriter.
# ReportWriter calls ResearchAssistant tool.
# ResearchAssistant calls WebSearch and Summarizer tools.
# Results flow back up.

5 Review/Critique：生成 + 审核（Generator-Critic）

• 结构：通常在 SequentialAgent 中：先生成，再审查/评审
• 目标：提高生成质量或有效性
• 通信：生成结果写 state，审查 agent 读取并给出状态/反馈（也可写入新的 state key）

适用：写代码/文案/方案都很通用。

# Conceptual Code: Generator-Critic
from google.adk.agents import SequentialAgent, LlmAgent

generator = LlmAgent(
    name="DraftWriter",
    instruction="Write a short paragraph about subject X.",
    output_key="draft_text"
)

reviewer = LlmAgent(
    name="FactChecker",
    instruction="Review the text in {draft_text} for factual accuracy. Output 'valid' or 'invalid' with reasons.",
    output_key="review_status"
)

# Optional: Further steps based on review_status

review_pipeline = SequentialAgent(
    name="WriteAndReview",
    sub_agents=[generator, reviewer]
)
# generator runs -> saves draft to state['draft_text']
# reviewer runs -> reads state['draft_text'], saves status to state['review_status']

6 Iterative Refinement：迭代优化（Loop + 质量门禁）

• 结构：LoopAgent 包含若干 agent，在多轮迭代里改进结果
• 目标：不断提升 state 中的产物（代码/文本/计划），直到达标或达到最大迭代次数
• 终止方式：max_iterations，或一个检查 agent 在达标时设置 escalate=True

一个非常“工程味”的例子：

• CodeRefiner：生成/改代码（写入 current_code）
• QualityChecker：评估 pass/fail（写入 quality_status）
• StopChecker（custom agent）：读 quality_status，如果 pass 就 EventActions(escalate=True) 让 loop 停止

适用：直到满足标准才结束的场景（测试通过、质量达标、格式正确等）。

# Conceptual Code: Iterative Code Refinement
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

# Agent to generate/refine code based on state['current_code'] and state['requirements']
code_refiner = LlmAgent(
    name="CodeRefiner",
    instruction="Read state['current_code'] (if exists) and state['requirements']. Generate/refine Python code to meet requirements. Save to state['current_code'].",
    output_key="current_code" # Overwrites previous code in state
)

# Agent to check if the code meets quality standards
quality_checker = LlmAgent(
    name="QualityChecker",
    instruction="Evaluate the code in state['current_code'] against state['requirements']. Output 'pass' or 'fail'.",
    output_key="quality_status"
)

# Custom agent to check the status and escalate if '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")]
)
# Loop runs: Refiner -> Checker -> StopChecker
# State['current_code'] is updated each iteration.
# Loop stops if QualityChecker outputs 'pass' (leading to StopChecker escalating) or after 5 iterations.

7 Human-in-the-Loop：把“人”放进工作流

• 结构：在流程中插入人工介入点（审批、纠错、确认）
• 目标：人类监督、批准或处理 AI 无法完成的任务
• 实现方式：通常用自定义 tool 把请求发送到外部系统（UI/工单系统），等待人类输入，再把结果返回给 agent

注意：ADK 没有内置的“Human Agent”类型，需要你做自定义集成
适用：高风险动作、需要合规审批、或模型不可靠时的兜底。

# Conceptual Code: Using a Tool for Human Approval
from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.tools import FunctionTool

# --- Assume external_approval_tool exists ---
# This tool would:
# 1. Take details (e.g., request_id, amount, reason).
# 2. Send these details to a human review system (e.g., via API).
# 3. Poll or wait for the human response (approved/rejected).
# 4. Return the human's decision.
# async def external_approval_tool(amount: float, reason: str) -> str: ...
approval_tool = FunctionTool(func=external_approval_tool)

# Agent that prepares the request
prepare_request = LlmAgent(
    name="PrepareApproval",
    instruction="Prepare the approval request details based on user input. Store amount and reason in state.",
    # ... likely sets state['approval_amount'] and state['approval_reason'] ...
)

# Agent that calls the human approval tool
request_approval = LlmAgent(
    name="RequestHumanApproval",
    instruction="Use the external_approval_tool with amount from state['approval_amount'] and reason from state['approval_reason'].",
    tools=[approval_tool],
    output_key="human_decision"
)

# Agent that proceeds based on human decision
process_decision = LlmAgent(
    name="ProcessDecision",
    instruction="Check {human_decision}. If 'approved', proceed. If 'rejected', inform user."
)

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

8 Human-in-the-Loop with Policy：用 PolicyEngine 做“工具调用前的确认”

文档给了更工程化的 HITL 方式：

• 使用 PolicyEngine 定义策略
• SecurityPlugin 拦截 tool call
• 如果策略要求，就自动向用户请求确认

文档称这种方式更适合治理与安全规则；并且指出 TypeScript 里这是推荐模式，其他语言的支持计划在未来版本完善

const rootAgent = new LlmAgent({
  name: 'weather_time_agent',
  model: 'gemini-2.5-flash',
  description:
      'Agent to answer questions about the time and weather in a city.',
  instruction:
      'You are a helpful agent who can answer user questions about the time and weather in a city.',
  tools: [getWeatherTool],
});

class CustomPolicyEngine implements BasePolicyEngine {
  async evaluate(_context: ToolCallPolicyContext): Promise<PolicyCheckResult> {
    // Default permissive implementation
    return Promise.resolve({
      outcome: PolicyOutcome.CONFIRM,
      reason: 'Needs confirmation for tool call',
    });
  }
}

const runner = new InMemoryRunner({
    agent: rootAgent,
    appName,
    plugins: [new SecurityPlugin({policyEngine: new CustomPolicyEngine()})]
});

9 Combining Patterns：组合拳

这些模式可以自由组合，不是互斥的。
一个非常常见的组合就是：
并行调研（Parallel fan-out）→ 顺序汇总（Sequential gather）→ 循环打磨（Loop refinement）→ 关键步骤人工审批（HITL）。

四、怎么从 0 设计自己的 MAS？

浓缩成一个实战流程：

1. 先画 Agent 树：谁是 root？谁是专家？（sub_agents）
2. 再选控制流：顺序？并行？循环？（Workflow agents）
3. 再选通信方式：
   • 稳定流水线优先 session.state（output_key）
   • 动态路由用 transfer_to_agent（AutoFlow）
   • 强约束/可观测调用用 AgentTool
4. 关键节点加“质量门禁”：Critic、Checker、Stopper（Loop 的退出信号用 escalate）
5. 高风险动作加 HITL / Policy：工具执行前确认、审批流接入

五、multi-agents 最该带走的三句话

1. 多智能体不是“越多越强”，而是“结构更清晰”：层级 + 流程编排 + 通信机制。
2. 三种通信方式各有性格：
   • session.state：最稳、最适合流水线
   • transfer_to_agent：最灵活、最像“智能路由”
   • AgentTool：最工程化、像工具调用一样显式可控
3. 模式可以组合：Coordinator / Pipeline / Fan-out / Decomposition / Critique / Refinement / HITL 都是积木，可以按你的业务拼装。