第7章：多智能体协作 Multi-Agent Collaboration
虽然单一智能体架构在解决明确定义的问题时可能非常有效，但在面对复杂的多领域任务时，其能力通常受到限制。多智能体协作模式通过将系统结构化为一组合作的独立且专门化的智能体来解决这些局限性。这种方法基于任务分解的原则，即将高层次的目标分解为离散的子问题。然后将每个子问题分配给具备特定工具、数据访问或推理能力的智能体，以便完成该任务。
例如，一个复杂的研究查询可以被分解并分别分配给以下智能体：信息检索的研究智能体、统计处理的数据分析智能体，以及生成最终报告的综合智能体。这种系统的效率不仅仅依赖于劳动分工，还取决于智能体之间的通信机制。这需要一个标准化的通信协议和共享的本体，以便智能体能够交换数据、委派子任务并协调其行动，从而确保最终输出的连贯性。
这种分布式架构具有多种优势，包括增强的模块化、可扩展性和鲁棒性，因为单个智能体的失败不一定导致整个系统的崩溃。协作使得多智能体系统的集体性能能够超越任何单个智能体的潜在能力。
多智能体协作模式概述
多智能体协作模式涉及设计多个独立或半独立的智能体共同工作以实现共同目标的系统。每个智能体通常具有明确的角色、与整体目标一致的具体目标，并可能访问不同的工具或知识库。这种模式的强大之处在于这些智能体之间的互动和协同效应。
协作可以采取多种形式：
- 顺序交接： 一个智能体完成任务后将其输出传递给另一个智能体，后者负责管道中的下一步（类似于规划模式，但明确涉及不同的智能体）。
- 并行处理： 多个智能体同时处理问题的不同部分，随后将它们的结果合并。
- 辩论与共识： 多智能体协作中，具有不同视角和信息来源的智能体进行讨论以评估选项，最终达成共识或做出更明智的决策。
- 层次结构： 一个管理智能体可以根据工具访问或插件能力动态地将任务分配给工作智能体，并综合它们的结果。每个智能体还可以管理相关的工具组，而不是由单个智能体处理所有工具。
专家团队： 在不同领域具有专业知识的智能体（例如，研究员、撰稿人、编辑）协作以生成复杂的输出。
批评-审查者： 智能体生成初步输出，例如计划、草稿或答案。然后，第二组智能体对该输出进行严格评估，以确保其符合政策、安全性、合规性、正确性、质量以及与组织目标的一致性。原始创建者或最终智能体根据反馈对输出进行修订。这种模式在代码生成、研究写作、逻辑检查以及确保伦理对齐方面特别有效。其优势包括提高鲁棒性、改善质量以及降低出现幻觉或错误的可能性。
多智能体系统（见图1）本质上包括智能体角色和职责的划分、智能体之间交换信息的通信渠道的建立，以及指导其协同工作的任务流程或交互协议的制定。
图1：多智能体系统示例
诸如 Crew AI 和 Google ADK 等框架旨在促进这一范式，通过提供智能体、任务及其交互过程的规范结构来实现。这种方法对需要多种专业知识、包含多个离散阶段或利用并行处理和智能体间信息验证优势的挑战尤为有效。
实际应用与使用场景
多智能体协作是一种强大的模式，适用于多个领域：
- 复杂研究与分析： 一组智能体可以协作完成研究项目。一个智能体可能专注于搜索学术数据库，另一个负责总结发现，第三个识别趋势，第四个将信息综合成报告。这类似于人类研究团队的运作方式。
- 软件开发： 想象智能体协作开发软件。一个智能体可以是需求分析师，另一个是代码生成器，第三个是测试人员，第四个是文档编写者。他们可以相互传递输出以构建和验证组件。
- 创意内容生成： 创建营销活动可能涉及市场研究智能体、文案撰写智能体、图形设计智能体（使用图像生成工具）以及社交媒体调度智能体共同协作。
- 金融分析： 多智能体系统可以分析金融市场。智能体可能专注于获取股票数据、分析新闻情绪、进行技术分析以及生成投资建议。
- 客户支持升级： 前线支持智能体可以处理初步查询，将复杂问题升级到专业智能体（例如技术专家或账单专家），根据问题复杂性进行顺序交接。
- 供应链优化： 智能体可以代表供应链中的不同节点（供应商、制造商、分销商），协作优化库存水平、物流和调度，以应对需求变化或中断。
- 网络分析与修复： 自主操作在故障定位方面受益于智能体架构，多个智能体可以协作进行问题分类和修复，建议最佳行动方案。这些智能体还可以与传统机器学习模型和工具集成，在利用现有系统的同时提供生成式人工智能的优势。
能够划分专业化智能体并精心编排其相互关系，使开发者能够构建具有更高模块化、可扩展性以及解决单一集成智能体难以应对的复杂问题能力的系统。
多智能体协作：探索相互关系与通信结构
理解智能体之间复杂的交互和通信方式是设计有效多智能体系统的基础。如图2所示，存在一系列从最简单的单智能体场景到复杂的定制协作框架的相互关系和通信模型。每种模型都具有独特的优势和挑战，影响多智能体系统的整体效率、鲁棒性和适应性。
1. 单一智能体：
在最基本的层面上，“单一智能体”能够自主运行，无需与其他实体直接互动或通信。这种模型易于实现和管理，但其能力受到单个智能体范围和资源的固有限制。它适用于可以分解为独立子问题的任务，每个子问题都可以由一个独立且自给自足的智能体解决。
2. 网络：
“网络”模型向协作迈出了重要一步，其中多个智能体以去中心化的方式直接相互交互。通信通常采用点对点方式，允许信息、资源甚至任务的共享。该模型增强了系统的韧性，因为单个智能体的故障不会必然导致整个系统瘫痪。然而，在大型、非结构化网络中管理通信开销并确保一致的决策可能会面临挑战。
3. 监督者：
在“监督者”模型中，一个专门的智能体，即“监督者”，负责监督和协调一组下属智能体的活动。监督者充当通信、任务分配和冲突解决的中心枢纽。这种层级结构提供了明确的权责分工，可以简化管理和控制。然而，它引入了单点故障（即监督者）的问题，并且当监督者面对大量下属或复杂任务时，可能成为瓶颈。
4. 工具型监督者：
该模型是“监督者”概念的细化扩展，其中监督者的角色更多地侧重于提供资源、指导或分析支持，而不是直接指挥和控制其他智能体。监督者可能提供工具、数据或计算服务，使其他智能体能够更有效地完成任务，而无需严格地控制其每一个行动。这种方法旨在利用监督者的能力，同时避免强制性自上而下的控制。
5. 层级结构：
“层级结构”模型在监督者概念的基础上进一步扩展，创建了一个多层次的组织结构。这包括多个层级的监督者，高层监督者负责监督低层监督者，最终在最低层由一组操作智能体执行任务。这种结构适用于可以分解为子问题的复杂问题，每个子问题由层级中的特定层管理。它为扩展性和复杂性管理提供了一种结构化的方法，允许在定义的边界内进行分布式决策。
图 2：智能体以多种方式进行通信和交互。
6. 定制：
“定制”模型代表了多智能体系统设计中的终极灵活性。它允许根据特定问题或应用的需求，量身定制独特的交互和通信结构。这可能包括结合前述模型元素的混合方法，或根据环境的独特约束和机遇开发全新的设计。定制模型通常源于优化特定性能指标的需求、处理高度动态的环境，或将领域特定知识融入系统架构中。设计和实施定制模型通常需要对多智能体系统原理有深入理解，并仔细考虑通信协议、协调机制以及可能出现的行为。
总结来说，为多智能体系统选择交互和通信模型是一个关键的设计决策。每种模型都有其独特的优点和缺点，最佳选择取决于任务的复杂性、智能体的数量、所需的自主性水平、对鲁棒性的需求以及可接受的通信开销等因素。未来的多智能体系统发展可能会继续探索和优化这些模型，同时开发新的协作智能范式。
实践代码（Crew AI）
以下 Python 代码使用 CrewAI 框架定义了一个 AI 驱动的团队，用于生成关于 AI 趋势的博客文章。代码首先设置环境，从 .env 文件中加载 API 密钥。应用的核心部分包括定义两个智能体：一个研究员用于查找和总结 AI 趋势，另一个写作者根据研究内容创建博客文章。
代码相应地定义了两个任务：一个用于研究趋势，另一个用于撰写博客文章，其中写作任务依赖于研究任务的输出。随后，这些智能体和任务被组装到一个 Crew 中，指定任务按顺序执行的过程。Crew 使用指定的语言模型（特别是 "gemini-2.0-flash" 模型）进行初始化。主函数通过调用 kickoff() 方法执行该 Crew，协调智能体之间的协作以生成所需的输出。最后，代码打印出 Crew 执行的最终结果，即生成的博客文章。
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from langchain_google_genai import ChatGoogleGenerativeAI
def setup_environment():
"""加载环境变量并检查所需的 API 密钥是否存在。"""
load_dotenv()
if not os.getenv("GOOGLE_API_KEY"):
raise ValueError("未找到 GOOGLE_API_KEY。请在 .env 文件中设置该变量。")
def main():
"""
初始化并运行使用最新 Gemini 模型的 AI 团队进行内容创作。
"""
setup_environment()
# 定义要使用的语言模型。
# 更新为 Gemini 2.0 系列中的模型，以获得更好的性能和功能。
# 如果需要前沿（预览）功能，可以使用 "gemini-2.5-flash"。
llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash")
# 定义具有特定角色和目标的智能体
researcher = Agent(
role='高级研究分析师',
goal='发现并总结 AI 的最新趋势。',
backstory="你是一位经验丰富的研究分析师，擅长识别关键趋势并综合信息。",
verbose=True,
allow_delegation=False,
)
writer = Agent(
role='技术内容撰写员',
goal='根据研究结果撰写一篇清晰且引人入胜的博客文章。',
backstory="你是一位技术娴熟的撰稿人，能够将复杂的技术主题转化为易于理解的内容。",
verbose=True,
allow_delegation=False,
)
# 为智能体定义任务
research_task = Task(
description="研究 2024-2025 年人工智能领域的三个新兴趋势，重点关注实际应用和潜在影响。",
expected_output="关于三个主要 AI 趋势的详细总结，包括关键点和来源。",
agent=researcher,
)
writing_task = Task(
description="根据研究结果撰写一篇 500 字的博客文章。文章应具有吸引力，并易于普通读者理解。",
expected_output="一篇关于最新 AI 趋势的完整 500 字博客文章。",
agent=writer,
context=[research_task],
)
# 创建团队
blog_creation_crew = Crew(
agents=[researcher, writer],
tasks=[research_task, writing_task],
process=Process.sequential,
llm=llm,
verbose=2 # 设置详细团队执行日志的级别
)
# 执行团队任务
print("## 使用 Gemini 2.0 Flash 运行博客创建团队... ##")
try:
result = blog_creation_crew.kickoff()
print("\n------------------\n")
print("## 团队最终输出 ##")
print(result)
except Exception as e:
print(f"\n发生意外错误: {e}")
if __name__ == "__main__":
main()
实践代码（Google ADK）
以下代码示例展示了如何通过创建父子关系，在 Google ADK 中建立层级智能体结构。代码定义了两种类型的智能体：LlmAgent 和从 BaseAgent 派生的自定义 TaskExecutor 智能体。TaskExecutor 设计用于特定的非 LLM 任务，在此示例中，它仅生成一个“任务成功完成”的事件。一个名为 greeter 的 LlmAgent 被初始化为指定模型，并指示其充当友好的问候者。自定义 TaskExecutor 被实例化为 task_doer。一个名为 coordinator 的父级 LlmAgent 被创建，同样具有模型和指令。coordinator 的指令指导它将问候任务委派给 greeter，将任务执行委派给 task_doer。greeter 和 task_doer 被添加为 coordinator 的子智能体，从而建立了父子关系。代码随后断言该关系已正确设置。最后，它打印一条消息，指示智能体层级已成功创建。
from google.adk.agents import LlmAgent, BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from typing import AsyncGenerator
## 正确实现通过扩展 BaseAgent 的自定义智能体
class TaskExecutor(BaseAgent):
"""一个具有自定义非 LLM 行为的专用智能体。"""
name: str = "TaskExecutor"
description: str = "执行预定义任务。"
async def _run_async_impl(self, context: InvocationContext) -> AsyncGenerator[Event, None]:
"""任务的自定义实现逻辑。"""
# 这是您可以添加自定义逻辑的地方。
# 在这个示例中，我们仅返回一个简单事件。
yield Event(author=self.name, content="任务成功完成。")
## 定义具有正确初始化的单个智能体
## LlmAgent 需要指定一个模型。
greeter = LlmAgent(
name="Greeter",
model="gemini-2.0-flash-exp",
instruction="你是一个友好的迎宾员。"
)
task_doer = TaskExecutor() # 实例化我们具体的自定义智能体
## 创建一个父智能体并分配其子智能体
## 父智能体的描述和指令应指导其委派逻辑。
coordinator = LlmAgent(
name="Coordinator",
model="gemini-2.0-flash-exp",
description="一个可以迎接用户并执行任务的协调员。",
instruction="当被要求迎接时，委派给 Greeter。当被要求执行任务时，委派给 TaskExecutor。",
sub_agents=[
greeter,
task_doer
]
)
## ADK 框架会自动建立父子智能体关系。
## 如果在初始化后检查，这些断言将通过。
assert greeter.parent_agent == coordinator
assert task_doer.parent_agent == coordinator
print("智能体层级成功创建。")
此代码片段展示了在 Google ADK 框架中使用 LoopAgent 来建立迭代工作流。代码定义了两个智能体：ConditionChecker 和 ProcessingStep。ConditionChecker 是一个自定义智能体，用于检查会话状态中的 "status" 值。如果 "status" 为 "completed"，ConditionChecker 会触发事件以停止循环。否则，它会返回一个事件以继续循环。ProcessingStep 是一个使用 "gemini-2.0-flash-exp" 模型的 LlmAgent。它的指令是执行任务，并在最后一步将会话 "status" 设置为 "completed"。创建了一个名为 StatusPoller 的 LoopAgent。StatusPoller 配置了 max_iterations=10。StatusPoller 包括 ProcessingStep 和一个 ConditionChecker 实例作为子智能体。LoopAgent 将最多执行 10 次迭代，按顺序执行子智能体，如果 ConditionChecker 发现状态为 "completed"，则停止。
import asyncio
from typing import AsyncGenerator
from google.adk.agents import LoopAgent, LlmAgent, BaseAgent
from google.adk.events import Event, EventActions
from google.adk.agents.invocation_context import InvocationContext
## 最佳实践：定义自定义智能体为完整、自描述的类。
class ConditionChecker(BaseAgent):
"""一个自定义智能体，用于检查会话状态中的 'completed' 状态。"""
name: str = "ConditionChecker"
description: str = "检查进程是否完成并向循环发送停止信号。"
async def _run_async_impl(
self, context: InvocationContext
) -> AsyncGenerator[Event, None]:
"""检查状态并返回事件以继续或停止循环。"""
status = context.session.state.get("status", "pending")
is_done = (status == "completed")
if is_done:
# 当条件满足时触发事件以终止循环。
yield Event(author=self.name, actions=EventActions(escalate=True))
else:
# 返回一个简单事件以继续循环。
yield Event(author=self.name, content="条件未满足，继续循环。")
## 修正：LlmAgent 必须具有模型和明确的指令。
process_step = LlmAgent(
name="ProcessingStep",
model="gemini-2.0-flash-exp",
instruction="你是一个较长流程中的步骤。执行你的任务。如果你是最后一步，请通过设置 'status' 为 'completed' 来更新会话状态。"
)
## LoopAgent 负责协调工作流程。
poller = LoopAgent(
name="StatusPoller",
max_iterations=10,
sub_agents=[
process_step,
ConditionChecker() # 实例化定义良好的自定义智能体。
]
)
## 此 poller 将会执行 'process_step'
## 然后重复执行 'ConditionChecker'
## 直到状态为 'completed' 或者达到 10 次迭代。
这段代码阐明了 Google ADK 中的 SequentialAgent 模式，该模式专为构建线性工作流程而设计。代码使用 google.adk.agents 库定义了一个顺序智能体管道。该管道由两个智能体组成：step1 和 step2。step1 被命名为 "Step1_Fetch"，其输出将存储在会话状态的键 "data" 下。step2 被命名为 "Step2_Process"，其指令是分析存储在 session.state["data"] 中的信息并提供摘要。名为 "MyPipeline" 的 SequentialAgent 负责协调这些子智能体的执行。当管道以初始输入运行时，step1 将首先执行。step1 的响应将保存到会话状态的键 "data" 下。随后，step2 将执行，并根据其指令使用 step1 放入状态的信息。此结构允许构建工作流程，其中一个智能体的输出成为下一个智能体的输入。这是创建多步骤 AI 或数据处理管道中的常见模式。
from google.adk.agents import SequentialAgent, Agent
## 此智能体的输出将保存到 session.state["data"]
step1 = Agent(name="Step1_Fetch", output_key="data")
## 此智能体将使用前一步的输出数据。
## 我们通过指令告诉它如何找到并使用这些数据。
step2 = Agent(
name="Step2_Process",
instruction="分析存储在 state['data'] 中的信息并提供摘要。"
)
pipeline = SequentialAgent(
name="MyPipeline",
sub_agents=[step1, step2]
)
## 当管道以初始输入运行时，Step1 将执行，
## 它的响应将存储在 session.state["data"] 中，随后
## Step2 将根据指令使用状态中的信息并执行。
以下代码示例展示了 Google ADK 中的 ParallelAgent 模式，该模式支持多个智能体任务的并发执行。data_gatherer 被设计为同时运行两个子智能体：weather_fetcher 和 news_fetcher。weather_fetcher 智能体的指令是获取给定位置的天气，并将结果存储在 session.state["weather_data"] 中。同样，news_fetcher 智能体的指令是检索给定主题的头条新闻，并将其存储在 session.state["news_data"] 中。每个子智能体都配置为使用 "gemini-2.0-flash-exp" 模型。ParallelAgent 负责协调这些子智能体的执行，使它们能够并行工作。weather_fetcher 和 news_fetcher 的结果将被收集并存储在会话状态中。最后，示例展示了如何在智能体执行完成后从 final_state 中访问收集的天气和新闻数据。
from google.adk.agents import Agent, ParallelAgent
## 最好将获取逻辑定义为智能体的工具
## 为了简化此示例，我们将逻辑嵌入到智能体的指令中。
## 在实际场景中，你可以使用工具。
## 定义并行运行的独立智能体
weather_fetcher = Agent(
name="weather_fetcher",
model="gemini-2.0-flash-exp",
instruction="获取指定位置的天气信息，并仅返回天气报告。",
output_key="weather_data" # 结果将存储在 session.state["weather_data"] 中
)
news_fetcher = Agent(
name="news_fetcher",
model="gemini-2.0-flash-exp",
instruction="获取指定主题的头条新闻，并仅返回该新闻内容。",
output_key="news_data" # 结果将存储在 session.state["news_data"] 中
)
## 创建 ParallelAgent 来协调子智能体
data_gatherer = ParallelAgent(
name="data_gatherer",
sub_agents=[
weather_fetcher,
news_fetcher
]
)
上述代码段展示了 Google ADK 中的“智能体作为工具”范式，该范式使一个智能体能够以类似函数调用的方式利用另一个智能体的能力。具体来说，代码定义了一个使用 Google 的 LlmAgent 和 AgentTool 类的图像生成系统。它包含两个智能体：一个父智能体 artist_agent 和一个子智能体 image_generator_agent。generate_image 函数是一个简单的工具，用于模拟图像生成并返回虚拟的图像数据。image_generator_agent 的职责是根据接收到的文本提示使用此工具。artist_agent 的角色是首先创建一个创意图像提示，然后通过 AgentTool 包装器调用 image_generator_agent。AgentTool 充当桥梁，使一个智能体能够将另一个智能体作为工具使用。当 artist_agent 调用 image_tool 时，AgentTool 使用艺术家的提示调用 image_generator_agent。随后，image_generator_agent 使用该提示调用 generate_image 函数。最终，生成的图像（或虚拟数据）通过智能体返回。这种架构展示了一个分层智能体系统，其中高级智能体协调低级专业智能体完成任务。
from google.adk.agents import LlmAgent
from google.adk.tools import agent_tool
from google.genai import types
## 1. 一个简单的功能工具，用于核心功能。
## 遵循将操作与推理分离的最佳实践。
def generate_image(prompt: str) -> dict:
"""
根据文本提示生成图像。
参数:
prompt: 对要生成的图像的详细描述。
返回:
包含状态和生成的图像字节数据的字典。
"""
print(f"TOOL: 正在为提示生成图像: '{prompt}'")
# 在实际实现中，这里会调用图像生成 API。
# 在此示例中，我们返回虚拟图像数据。
mock_image_bytes = b"mock_image_data_for_a_cat_wearing_a_hat"
return {
"status": "success",
# 工具返回原始字节数据，智能体将处理 Part 的创建。
"image_bytes": mock_image_bytes,
"mime_type": "image/png"
}
## 2. 将 ImageGeneratorAgent 重构为 LlmAgent。
## 它现在可以正确使用传递给它的输入。
image_generator_agent = LlmAgent(
name="ImageGen",
model="gemini-2.0-flash",
description="根据详细的文本提示生成图像。",
instruction=(
"你是一名图像生成专家。你的任务是根据用户的请求，使用 `generate_image` 工具生成图像。"
"用户的完整请求应作为工具的 'prompt' 参数。"
"工具返回图像字节后，你必须输出该图像。"
),
tools=[generate_image]
)
## 3. 使用 AgentTool 包装修正后的智能体。
## 此处的描述是父智能体可见的内容。
image_tool = agent_tool.AgentTool(
agent=image_generator_agent,
description="使用此工具生成图像。输入应为所需图像的描述性提示。"
)
## 4. 父智能体保持不变，其逻辑是正确的。
artist_agent = LlmAgent(
name="Artist",
model="gemini-2.0-flash",
instruction=(
"你是一位富有创意的艺术家。首先，为一幅图像创造一个富有创意且描述性的提示。"
"然后，使用 `ImageGen` 工具根据你的提示生成图像。"
),
tools=[image_tool]
)
概览
定义（What）
复杂问题通常超出单一、单体化的基于LLM的智能体的能力范围。单一智能体可能缺乏解决多方面任务所需的多样化、专业化技能或特定工具的访问权限。这种局限性会导致瓶颈，降低系统的整体效率和可扩展性。因此，处理复杂的多领域目标变得低效，并可能导致不完整或次优的结果。
设计意图（Why）
多智能体协作模式通过创建多个协作智能体的系统提供了一种标准化解决方案。复杂问题被分解为更小、更易管理的子问题。每个子问题被分配给具有解决问题所需精确工具和能力的专业智能体。这些智能体通过定义的通信协议和交互模型（如顺序交接、并行工作流或层级委派）进行协作。这种智能体化的分布式方法创造了协同效应，使团队能够实现单个智能体无法完成的成果。
使用原则（Rule of Thumb）
当任务对单一智能体来说过于复杂且可以分解为需要专业技能或工具的不同子任务时，使用此模式。它特别适用于需要多样化专业知识、并行处理或具有多个阶段的结构化工作流的问题，例如复杂的研究与分析、软件开发或创意内容生成。
图解 （Visual Summary）
图3：多智能体设计模式
关键要点
- 多智能体协作涉及多个智能体共同努力以实现共同目标。
- 此模式利用专业化角色、分布式任务和智能体间通信。
- 协作形式包括顺序交接、并行处理、辩论或层级结构。
- 此模式非常适合需要多样化专业知识或多个不同阶段的复杂问题。
结论
本章探讨了多智能体协作模式，展示了在系统中协调多个专业智能体的优势。我们研究了各种协作模型，强调了该模式在解决复杂、多方面问题中的重要作用。理解智能体协作自然引发了对其与外部环境交互的进一步探讨。