# 第05章：多智能体系统

## 📌 本章目标

- 理解多智能体系统的设计理念
- 掌握 Agent 层级结构的构建方法
- 学习三种工作流 Agent：Sequential、Parallel、Loop
- 掌握常见多 Agent 协作模式
- 了解 Agent 间的通信机制

---

## 5.1 多智能体系统概述

### 5.1.1 为什么需要多智能体？

当应用变得复杂时，使用单个"巨型"Agent 会面临以下问题：

| 问题 | 说明 |
|------|------|
| **指令膨胀** | 系统指令过长，LLM 难以同时遵循所有规则 |
| **工具过多** | 工具列表过长，LLM 选择工具的准确率下降 |
| **难以维护** | 修改一个功能可能影响其他功能 |
| **难以调试** | 出错时难以定位问题来源 |

**多智能体系统**通过将复杂任务分解为多个专门的 Agent 来解决这些问题。

### 5.1.2 ADK 多智能体架构

```
                    root_agent（根智能体）
                         │
           ┌─────────────┼─────────────┐
           │             │             │
      LlmAgent     SequentialAgent  ParallelAgent
      （协调器）     （顺序流水线）    （并行执行器）
           │             │             │
      ┌────┼────┐   ┌────┼────┐   ┌───┴───┐
      │    │    │   │    │    │   │       │
    Agent Agent Agent Agent Agent Agent Agent
    (子1) (子2) (子3) (步骤1)(步骤2)(任务A) (任务B)
```

---

## 5.2 Agent 层级结构

### 5.2.1 父子关系

```python
"""
Agent 层级结构基础
通过 sub_agents 参数建立父子关系
"""

from google.adk.agents import LlmAgent      # 导入 LlmAgent 类

# ========================================
# 第一步：定义叶子节点 Agent（最底层的 Agent）
# ========================================

# 定义一个专门处理问候的 Agent
greeter_agent = LlmAgent(
    name="Greeter",                         # Agent 名称
    model="gemini-2.0-flash",               # 使用 Gemini 模型
    description="负责向用户打招呼和寒暄。",  # 描述：用于路由决策
    instruction=(                           # 系统指令
        "你是一个友好的问候助手。\n"
        "热情地向用户打招呼，并询问有什么可以帮助的。"
    ),
)

# 定义一个专门处理任务执行的 Agent
task_agent = LlmAgent(
    name="TaskExecutor",                    # Agent 名称
    model="gemini-2.0-flash",               # 使用 Gemini 模型
    description="负责执行用户的具体任务。",  # 描述：用于路由决策
    instruction=(                           # 系统指令
        "你是一个任务执行助手。\n"
        "认真完成用户交给你的任务。"
    ),
)

# ========================================
# 第二步：定义父 Agent，通过 sub_agents 建立层级
# ========================================

coordinator = LlmAgent(
    name="Coordinator",                     # 父 Agent 名称
    model="gemini-2.0-flash",               # 使用 Gemini 模型
    description="主协调器，负责分配任务。",  # 描述
    instruction=(                           # 系统指令
        "你是一个协调器。\n"
        "根据用户的请求，将任务分配给合适的子 Agent。\n"
        "问候和寒暄类请求交给 Greeter。\n"
        "具体任务交给 TaskExecutor。\n"
        "使用 transfer_to_agent 工具来委派任务。"
    ),
    sub_agents=[                            # 🔑 注册子 Agent 列表
        greeter_agent,                      # 问候 Agent
        task_agent,                         # 任务执行 Agent
    ],
)

# ========================================
# 层级关系说明
# ========================================
# coordinator（父）
#   ├── greeter_agent（子）
#   └── task_agent（子）
#
# ADK 会自动设置：
#   greeter_agent.parent_agent == coordinator
#   task_agent.parent_agent == coordinator
#
# 注意：一个 Agent 只能有一个父 Agent
# 尝试将同一个 Agent 添加到多个父 Agent 会抛出 ValueError
```

### 5.2.2 多层级嵌套

```python
"""
多层级嵌套的 Agent 结构
构建更复杂的多 Agent 系统
"""

from google.adk.agents import LlmAgent, SequentialAgent  # 导入 Agent 类型

# ========================================
# 第三层：具体执行 Agent
# ========================================

code_reviewer = LlmAgent(
    name="CodeReviewer",                    # 代码审查 Agent
    model="gemini-2.0-flash",               # 模型
    description="审查代码质量，提出改进建议。",  # 描述
    instruction="你是代码审查专家，审查代码并给出改进建议。",  # 指令
)

test_writer = LlmAgent(
    name="TestWriter",                      # 测试编写 Agent
    model="gemini-2.0-flash",               # 模型
    description="为代码编写单元测试。",      # 描述
    instruction="你是测试工程师，为给定的代码编写单元测试。",  # 指令
)

# ========================================
# 第二层：开发流水线（顺序执行）
# ========================================

dev_pipeline = SequentialAgent(
    name="DevPipeline",                     # 流水线名称
    sub_agents=[                            # 按顺序执行子 Agent
        code_reviewer,                      # 先审查代码
        test_writer,                        # 再编写测试
    ],
)

# ========================================
# 第二层：其他专门 Agent
# ========================================

doc_writer = LlmAgent(
    name="DocWriter",                       # 文档编写 Agent
    model="gemini-2.0-flash",               # 模型
    description="编写技术文档。",            # 描述
    instruction="你是技术文档工程师，编写清晰的技术文档。",  # 指令
)

# ========================================
# 第一层：根 Agent（协调器）
# ========================================

root_agent = LlmAgent(
    name="RootCoordinator",                 # 根协调器
    model="gemini-2.0-flash",               # 模型
    description="开发团队协调器。",          # 描述
    instruction=(                           # 指令
        "你是一个开发团队协调器。\n"
        "代码审查和测试任务交给 DevPipeline。\n"
        "文档编写任务交给 DocWriter。"
    ),
    sub_agents=[                            # 子 Agent 列表
        dev_pipeline,                       # 开发流水线
        doc_writer,                         # 文档编写
    ],
)

# 完整层级结构：
# root_agent（根协调器）
#   ├── dev_pipeline（开发流水线 - SequentialAgent）
#   │     ├── code_reviewer（代码审查）
#   │     └── test_writer（测试编写）
#   └── doc_writer（文档编写）
```

---

## 5.3 工作流 Agent

### 5.3.1 SequentialAgent（顺序执行）

```python
"""
SequentialAgent — 顺序执行工作流
子 Agent 按照列表顺序依次执行
适用于数据处理流水线等场景
"""

from google.adk.agents import LlmAgent, SequentialAgent  # 导入 Agent 类型

# ========================================
# 定义流水线的各个步骤
# ========================================

# 步骤一：信息提取 Agent
extractor = LlmAgent(
    name="Extractor",                       # 信息提取 Agent
    model="gemini-2.0-flash",               # 模型
    instruction="从用户输入中提取关键信息。只输出提取的关键信息。",  # 指令
    output_key="extracted_info",            # 🔑 将输出保存到 state['extracted_info']
)

# 步骤二：信息分析 Agent
analyzer = LlmAgent(
    name="Analyzer",                        # 信息分析 Agent
    model="gemini-2.0-flash",               # 模型
    instruction=(                           # 指令：引用上一步的输出
        "分析以下提取的信息，给出深度分析报告。\n"
        "提取的信息：{extracted_info}"       # 通过 {key} 引用 state 中的值
    ),
    output_key="analysis_result",           # 将分析结果保存到 state
)

# 步骤三：报告生成 Agent
reporter = LlmAgent(
    name="Reporter",                        # 报告生成 Agent
    model="gemini-2.0-flash",               # 模型
    instruction=(                           # 指令：引用前两步的输出
        "根据分析结果生成一份简洁的报告。\n"
        "原始信息：{extracted_info}\n"       # 引用提取的信息
        "分析结果：{analysis_result}"        # 引用分析结果
    ),
)

# ========================================
# 创建顺序流水线
# ========================================

pipeline = SequentialAgent(
    name="InfoPipeline",                    # 流水线名称
    sub_agents=[                            # 🔑 按顺序执行
        extractor,                          # 第一步：提取信息
        analyzer,                           # 第二步：分析信息
        reporter,                           # 第三步：生成报告
    ],
)

# 执行流程：
# 1. Extractor 运行 → 输出保存到 state['extracted_info']
# 2. Analyzer 运行 → 读取 state['extracted_info'] → 输出保存到 state['analysis_result']
# 3. Reporter 运行 → 读取两个 state 值 → 生成最终报告
```

### 5.3.2 ParallelAgent（并行执行）

```python
"""
ParallelAgent — 并行执行工作流
子 Agent 同时执行，互不阻塞
适用于多任务同时处理的场景
"""

from google.adk.agents import LlmAgent, ParallelAgent  # 导入 Agent 类型

# ========================================
# 定义并行执行的任务
# ========================================

# 任务一：天气查询 Agent
weather_agent = LlmAgent(
    name="WeatherFetcher",                  # 天气查询 Agent
    model="gemini-2.0-flash",               # 模型
    instruction="查询并返回用户指定城市的天气信息。只返回天气数据。",  # 指令
    output_key="weather_data",              # 输出保存到 state['weather_data']
)

# 任务二：新闻查询 Agent
news_agent = LlmAgent(
    name="NewsFetcher",                     # 新闻查询 Agent
    model="gemini-2.0-flash",               # 模型
    instruction="查询并返回用户指定城市的最新新闻。只返回新闻摘要。",  # 指令
    output_key="news_data",                 # 输出保存到 state['news_data']
)

# 任务三：交通查询 Agent
traffic_agent = LlmAgent(
    name="TrafficFetcher",                  # 交通查询 Agent
    model="gemini-2.0-flash",               # 模型
    instruction="查询并返回用户指定城市的交通状况。只返回交通信息。",  # 指令
    output_key="traffic_data",              # 输出保存到 state['traffic_data']
)

# ========================================
# 创建并行执行器
# ========================================

gatherer = ParallelAgent(
    name="InfoGatherer",                    # 并行执行器名称
    sub_agents=[                            # 🔑 这些 Agent 将同时执行
        weather_agent,                      # 天气查询（并行）
        news_agent,                         # 新闻查询（并行）
        traffic_agent,                      # 交通查询（并行）
    ],
)

# 执行流程：
# 1. WeatherFetcher、NewsFetcher、TrafficFetcher 同时开始执行
# 2. 各自独立运行，互不影响
# 3. 所有 Agent 完成后，后续 Agent 可以读取三个 state 值
#
# 注意事项：
# - 并行 Agent 共享同一个 session.state
# - 使用不同的 output_key 避免数据覆盖
# - 事件（Event）可能交错输出
```

### 5.3.3 LoopAgent（循环执行）

```python
"""
LoopAgent — 循环执行工作流
子 Agent 按顺序循环执行，直到满足退出条件
适用于轮询、迭代优化等场景
"""

from google.adk.agents import (                # 导入所需类型
    LlmAgent,                                 # LLM 智能体
    LoopAgent,                                # 循环工作流
    BaseAgent,                                # 基础智能体（用于自定义）
)
from google.adk.events import Event, EventActions  # 导入事件和事件动作
from google.adk.agents.invocation_context import InvocationContext  # 导入调用上下文
from typing import AsyncGenerator              # 导入异步生成器类型


# ========================================
# 自定义条件检查 Agent
# ========================================

class QualityChecker(BaseAgent):
    """
    质量检查 Agent
    检查内容质量是否达标，决定是否继续循环
    """

    async def _run_async_impl(
        self,
        ctx: InvocationContext,              # 调用上下文
    ) -> AsyncGenerator[Event, None]:        # 返回异步事件生成器
        """异步执行方法"""

        # 从会话状态中获取当前质量分数
        quality_score = ctx.session.state.get(  # 读取状态
            "quality_score",                     # 状态键名
            0,                                   # 默认值
        )

        # 判断质量是否达标（分数 >= 80 则通过）
        is_passed = quality_score >= 80      # 质量阈值

        # 生成事件，设置 escalate 标志
        # escalate=True 表示退出循环
        # escalate=False 表示继续循环
        yield Event(
            author=self.name,                # 事件作者
            actions=EventActions(
                escalate=is_passed,          # 🔑 达标则退出循环
            ),
            content=f"质量分数: {quality_score}, {'通过' if is_passed else '未通过'}",
        )


# ========================================
# 内容优化 Agent
# ========================================

content_improver = LlmAgent(
    name="ContentImprover",                 # 内容优化 Agent
    model="gemini-2.0-flash",               # 模型
    instruction=(                           # 指令
        "你是一个内容优化专家。\n"
        "根据当前内容的质量分数进行优化。\n"
        "优化后评估质量分数（0-100），并更新到 state['quality_score']。"
    ),
)

# ========================================
# 创建循环执行器
# ========================================

optimizer_loop = LoopAgent(
    name="ContentOptimizer",                # 循环执行器名称
    max_iterations=5,                       # 🔑 最大循环次数（防止无限循环）
    sub_agents=[                            # 每次循环依次执行
        content_improver,                   # 第一步：优化内容
        QualityChecker(name="Checker"),     # 第二步：检查质量
    ],
)

# 执行流程：
# 迭代1: ContentImprover → Checker(分数<80, 继续)
# 迭代2: ContentImprover → Checker(分数<80, 继续)
# 迭代3: ContentImprover → Checker(分数>=80, 退出)
#
# 退出条件（满足任一即退出）：
# 1. QualityChecker 设置 escalate=True
# 2. 达到 max_iterations 上限
```

---

## 5.4 常见多 Agent 协作模式

### 5.4.1 协调器/调度器模式

```python
"""
协调器/调度器模式（Coordinator/Dispatcher Pattern）
最常用的多 Agent 模式
父 Agent 根据用户请求动态分配任务给子 Agent
"""

from google.adk.agents import LlmAgent      # 导入 LlmAgent

# 定义专门的子 Agent
math_agent = LlmAgent(
    name="MathExpert",                      # 数学专家
    model="gemini-2.0-flash",               # 模型
    description="擅长数学计算和统计分析。",  # 描述
    instruction="你是数学专家，解决数学问题。",  # 指令
)

writing_agent = LlmAgent(
    name="WritingExpert",                   # 写作专家
    model="gemini-2.0-flash",               # 模型
    description="擅长文案撰写和内容创作。",  # 描述
    instruction="你是写作专家，帮助用户撰写内容。",  # 指令
)

coding_agent = LlmAgent(
    name="CodingExpert",                    # 编程专家
    model="gemini-2.0-flash",               # 模型
    description="擅长编程和代码调试。",      # 描述
    instruction="你是编程专家，帮助用户解决编程问题。",  # 指令
)

# 定义协调器
coordinator = LlmAgent(
    name="Coordinator",                     # 协调器
    model="gemini-2.0-flash",               # 模型
    description="根据问题类型分配给合适的专家。",  # 描述
    instruction=(                           # 指令
        "你是一个任务协调器。\n"
        "根据用户的问题类型，将任务分配给合适的专家：\n"
        "- 数学相关 → MathExpert\n"
        "- 写作相关 → WritingExpert\n"
        "- 编程相关 → CodingExpert\n"
        "使用 transfer_to_agent 工具进行任务分配。"
    ),
    sub_agents=[                            # 注册所有专家 Agent
        math_agent,                         # 数学专家
        writing_agent,                      # 写作专家
        coding_agent,                       # 编程专家
    ],
)
```

### 5.4.2 评审/批评模式（Generator-Critic）

```python
"""
评审/批评模式（Generator-Critic Pattern）
生成器创建内容，评审器评估并给出改进建议
通过 LoopAgent 实现迭代优化
"""

from google.adk.agents import LlmAgent, LoopAgent, BaseAgent  # 导入类型
from google.adk.events import Event, EventActions  # 导入事件类型
from google.adk.agents.invocation_context import InvocationContext  # 导入上下文
from typing import AsyncGenerator              # 导入异步生成器


# 生成器 Agent：创建初稿
generator = LlmAgent(
    name="Generator",                       # 生成器
    model="gemini-2.0-flash",               # 模型
    instruction=(                           # 指令
        "你是一个创意写手。\n"
        "根据用户需求创作内容。\n"
        "将创作的内容保存到 state['draft']。"
    ),
    output_key="draft",                     # 输出保存到 state['draft']
)

# 评审器 Agent：评估内容质量
critic = LlmAgent(
    name="Critic",                          # 评审器
    model="gemini-2.0-flash",               # 模型
    instruction=(                           # 指令
        "你是一个严格的内容评审。\n"
        "评审以下内容，给出评分（1-10）和改进建议。\n"
        "当前草稿：{draft}\n"
        "如果评分 >= 8，将 state['approved'] 设为 True。\n"
        "否则，将改进建议保存到 state['feedback']。"
    ),
)

# 条件检查 Agent：决定是否继续循环
class ApprovalChecker(BaseAgent):
    """检查内容是否通过审批"""

    async def _run_async_impl(
        self,
        ctx: InvocationContext,              # 调用上下文
    ) -> AsyncGenerator[Event, None]:        # 返回事件生成器
        """执行检查"""

        approved = ctx.session.state.get("approved", False)  # 获取审批状态

        yield Event(                         # 生成事件
            author=self.name,                # 事件作者
            actions=EventActions(
                escalate=approved,           # 通过则退出循环
            ),
        )

# 创建迭代优化循环
review_loop = LoopAgent(
    name="ReviewLoop",                      # 循环名称
    max_iterations=3,                       # 最多迭代3次
    sub_agents=[                            # 循环体
        critic,                             # 第一步：评审
        ApprovalChecker(name="Checker"),    # 第二步：检查是否通过
        generator,                          # 第三步：如果未通过，重新生成
    ],
)

# 执行流程：
# 1. Generator 创建初稿 → state['draft']
# 2. Critic 评审 → 评分 < 8 → state['feedback']
# 3. Checker → approved=False → 继续循环
# 4. Generator 根据反馈修改 → state['draft']
# 5. Critic 再次评审 → 评分 >= 8 → state['approved']=True
# 6. Checker → approved=True → 退出循环
```

---

## 5.5 Agent 间通信机制

### 5.5.1 共享状态（Shared State）

```python
"""
Agent 间通过共享状态通信
最基础、最常用的通信方式
"""

from google.adk.agents import LlmAgent, SequentialAgent  # 导入类型

# Agent A：收集信息
collector = LlmAgent(
    name="Collector",                       # 信息收集 Agent
    model="gemini-2.0-flash",               # 模型
    instruction="收集用户需求信息。",        # 指令
    output_key="user_requirements",         # 输出保存到 state
)

# Agent B：基于 Agent A 的输出进行设计
designer = LlmAgent(
    name="Designer",                        # 设计 Agent
    model="gemini-2.0-flash",               # 模型
    instruction=(                           # 指令：引用 Agent A 的输出
        "根据用户需求进行系统设计。\n"
        "用户需求：{user_requirements}"      # 通过 {key} 引用
    ),
    output_key="design_doc",                # 输出保存到 state
)

# Agent C：基于 Agent B 的输出进行评审
reviewer = LlmAgent(
    name="Reviewer",                        # 评审 Agent
    model="gemini-2.0-flash",               # 模型
    instruction=(                           # 指令：引用 Agent A 和 B 的输出
        "评审系统设计文档。\n"
        "用户需求：{user_requirements}\n"    # 引用需求
        "设计文档：{design_doc}"             # 引用设计
    ),
)

# 创建流水线
pipeline = SequentialAgent(
    name="DesignPipeline",                  # 流水线名称
    sub_agents=[collector, designer, reviewer],  # 顺序执行
)
```

### 5.5.2 通信机制对比

| 机制 | 说明 | 适用场景 |
|------|------|----------|
| **共享 State** | 通过 `session.state` 读写数据 | 顺序流水线、数据传递 |
| **output_key** | 自动将输出保存到 State | 简单的数据传递 |
| **LLM Transfer** | LLM 动态决定委派给哪个 Agent | 灵活的任务路由 |
| **AgentTool** | 将 Agent 作为工具调用 | Agent 复用 |
| **temp: State** | 临时状态，当前调用有效 | 工具间数据传递 |

---

## 📌 本章小结

- 多智能体系统通过 `sub_agents` 参数建立层级关系
- **SequentialAgent**：子 Agent 按顺序执行，适合流水线
- **ParallelAgent**：子 Agent 并行执行，适合多任务同时处理
- **LoopAgent**：子 Agent 循环执行，适合迭代优化
- Agent 间通信主要通过共享 State 和 LLM Transfer
- `output_key` 是最简单的数据传递方式

**下一章**：[第06章 - 会话与状态管理](./06-session-and-state.md) → 深入了解会话和状态管理机制。