多智能体协作Multi-Agent Collaboration单一 LLM 智能体难以应对复杂问题缺乏多样化专长或工具成为系统瓶颈降低效率与可扩展性难以完成多领域目标。多智能体协作模式通过将复杂问题拆分为可管理子问题分配给具备专长的智能体智能体通过顺序交接、并行处理或层级委托等协议协作实现单一智能体无法完成的目标。核心思想专长分工 协作通信—— 多个专精智能体各司其职通过协议协同完成复杂任务。适用场景当任务过于复杂需拆分为需专长或工具的子任务时适合采用该模式。适用于多领域、并行处理或多阶段结构化流程如复杂研究、软件开发、创意内容生成等。场景协作方式客户支持升级前线支持智能体处理初步问题复杂问题升级至专家智能体如技术或账单专家体现基于问题复杂度的顺序交接供应链优化智能体代表供应链各节点供应商、制造商、分销商协作优化库存、物流与排程应对需求变化或突发事件网络分析与修复自治运维场景下多智能体协作进行故障定位、分级处理与修复可集成传统机器学习模型与工具通过智能体专长划分与关系精细编排开发者可构建具备更强模块化、可扩展性和复杂问题处理能力的系统。智能体关系与通信结构理解智能体间的交互与通信方式是设计高效多智能体系统的基础。从最简单的单智能体到复杂的定制协作结构呈现多样化选择。每种模型有独特优势与挑战影响系统整体效率、健壮性与适应性。1. 单智能体最基础模型智能体独立运行无需与其他实体交互适合可拆分为独立子问题的场景但能力受限。2. 网络型多个智能体以去中心化方式直接交互点对点通信信息、资源和任务共享具备弹性但通信管理和决策一致性较难。[A] ←→ [B] ↕ ╲ ↕ [C] ←→ [D]3. 监督者专门智能体监督者协调下属智能体负责通信、任务分配和冲突解决层级结构清晰易于管理但存在单点故障和瓶颈风险。[监督者] / | \ [A] [B] [C]4. 工具型监督者监督者不直接指挥而是为其他智能体提供资源、指导或分析支持赋能而非强制控制提升灵活性。5. 层级型多层监督者结构高层监督者管理低层监督者底层为操作智能体适合复杂问题分层管理便于扩展和分布式决策。[总监督者] / \ [子监督者A] [子监督者B] / \ / \ [A1] [A2] [B1] [B2]6. 定制型最灵活模型针对具体问题或应用定制独特关系与通信结构可混合前述模型或创新设计适合优化特定性能、动态环境或领域知识集成。定制模型需深入理解多智能体原理慎重设计通信协议、协调机制与涌现行为。综上多智能体系统的关系与通信模型选择至关重要应结合任务复杂度、智能体数量、自治需求、健壮性和通信开销等因素权衡。代码示例示例 1CrewAI 实现importosfromdotenvimportload_dotenvfromcrewaiimportAgent,Task,Crew,Processfromlangchain_openaiimportChatOpenAIdefsetup_environment():加载环境变量并检查 API 密钥。load_dotenv()ifnotos.getenv(OPENAI_API_KEY):raiseValueError(OPENAI_API_KEY 未设置请在 .env 文件中配置。)defmain(): 初始化并运行内容创作 AI 团队。 setup_environment()# 指定语言模型llmChatOpenAI(modelgpt-4o-mini)# 定义 Agent 角色与目标researcherAgent(role高级研究分析师,goal查找并总结 AI 最新趋势。,backstory你是一名经验丰富的研究分析师擅长发现关键趋势并整合信息。,verboseTrue,allow_delegationFalse,)writerAgent(role技术内容写作者,goal根据研究结果撰写清晰易懂的博客。,backstory你是一名技术写作高手能将复杂技术转化为通俗内容。,verboseTrue,allow_delegationFalse,)# 定义任务research_taskTask(description调研 2024-2025 年 AI 三大新兴趋势关注实际应用与影响。,expected_output详细总结三大 AI 趋势包括要点与来源。,agentresearcher,)writing_taskTask(description根据研究结果撰写一篇 500 字博客内容通俗易懂。,expected_output完整的 500 字 AI 趋势博客。,agentwriter,context[research_task],)# 创建团队blog_creation_crewCrew(agents[researcher,writer],tasks[research_task,writing_task],processProcess.sequential,llmllm,verboseTrue)# 执行团队任务print(## 运行博客创作团队... ##)try:resultblog_creation_crew.kickoff()print(\n------------------\n)print(## 团队最终输出 ##)print(result)exceptExceptionase:print(f\n发生异常{e})if__name____main__:main()CrewAI 是专为多智能体协作设计的框架。示例展示如何定义一个 AI 协作团队生成 AI 趋势博客[研究员] → research_task → [写作者] → writing_task → 博客输出研究员Agent查找并总结 AI 最新趋势写作者Agent根据研究结果撰写 500 字博客任务依赖writing_task的context[research_task]确保写作基于研究结果顺序执行Process.sequential指定研究完成后再写作团队执行通过crew.kickoff()启动协作流程示例 2ADK 层级结构fromgoogle.adk.agentsimportLlmAgent,BaseAgentfromgoogle.adk.agents.invocation_contextimportInvocationContextfromgoogle.adk.eventsimportEventfromtypingimportAsyncGenerator GEMINI_MODELgemini-2.0-flash-expclassTaskExecutor(BaseAgent):自定义非 LLM 行为 Agent。name:strTaskExecutordescription:str执行预定义任务。asyncdef_run_async_impl(self,context:InvocationContext)-AsyncGenerator[Event,None]:yieldEvent(authorself.name,content任务成功完成。)greeterLlmAgent(nameGreeter,modelGEMINI_MODEL,instruction你是一名友好的问候者。)task_doerTaskExecutor()coordinatorLlmAgent(nameCoordinator,modelGEMINI_MODEL,description协调问候与任务执行。,instruction问候时委托 Greeter执行任务时委托 TaskExecutor。,sub_agents[greeter,task_doer])assertgreeter.parent_agentcoordinatorasserttask_doer.parent_agentcoordinatorprint(Agent 层级关系创建成功。)演示 Google ADK 中通过父子关系实现层级协作[Coordinator] / \ [Greeter] [TaskExecutor]LlmAgentGreeter负责问候BaseAgentTaskExecutor自定义非 LLM 智能体执行预定义任务返回任务成功完成事件Coordinator父智能体通过sub_agents参数建立父子关系根据请求类型委托给对应子智能体示例 3ADK LoopAgentimportasynciofromtypingimportAsyncGeneratorfromgoogle.adk.agentsimportLoopAgent,LlmAgent,BaseAgentfromgoogle.adk.eventsimportEvent,EventActionsfromgoogle.adk.agents.invocation_contextimportInvocationContext GEMINI_MODELgemini-2.0-flash-expclassConditionChecker(BaseAgent):检查流程是否完成并控制循环。name:strConditionCheckerdescription:str检查流程完成状态并通知循环终止。asyncdef_run_async_impl(self,context:InvocationContext)-AsyncGenerator[Event,None]:statuscontext.session.state.get(status,pending)is_done(statuscompleted)ifis_done:yieldEvent(authorself.name,actionsEventActions(escalateTrue))else:yieldEvent(authorself.name,content条件未满足继续循环。)process_stepLlmAgent(nameProcessingStep,modelGEMINI_MODEL,instruction你是流程中的一步完成任务。若为最后一步将 session 状态设为 completed。)pollerLoopAgent(nameStatusPoller,max_iterations10,sub_agents[process_step,ConditionChecker()])演示LoopAgent实现迭代流程LoopAgent (max10) ├→ [ProcessingStep] → 处理任务最后一步设 statuscompleted └→ [ConditionChecker] → statuscompleted? → 是: escalate 终止 / 否: 继续循环ConditionChecker自定义BaseAgent检查 session 状态若status为completed则通过EventActions(escalateTrue)终止循环ProcessingStepLLM 智能体负责处理任务并在最后一步设置状态LoopAgent配置max_iterations10循环执行直到条件满足或达到最大次数示例 4ADK 顺序与并行fromgoogle.adk.agentsimportAgent,SequentialAgent,ParallelAgent GEMINI_MODELgemini-2.0-flash-exp# # 示例 1SequentialAgent 顺序执行# step1 输出存入 session.state[data]step2 分析该数据# step1Agent(nameStep1_Fetch,output_keydata)step2Agent(nameStep2_Process,instruction分析 state[data] 信息并给出总结。)pipelineSequentialAgent(nameMyPipeline,sub_agents[step1,step2])# # 示例 2ParallelAgent 并行执行# 两个 Agent 同时获取天气和新闻结果存入各自的 state key# weather_fetcherAgent(nameweather_fetcher,modelGEMINI_MODEL,instruction获取指定地点天气仅返回天气报告。,output_keyweather_data)news_fetcherAgent(namenews_fetcher,modelGEMINI_MODEL,instruction获取指定主题头条新闻仅返回新闻内容。,output_keynews_data)data_gathererParallelAgent(namedata_gatherer,sub_agents[weather_fetcher,news_fetcher])SequentialAgent 顺序执行[Step1_Fetch] → state[data] → [Step2_Process] → 总结step1输出存入session.state[data]step2分析该数据并给出总结。ParallelAgent 并行执行ParallelAgent ├→ [weather_fetcher] → state[weather_data] └→ [news_fetcher] → state[news_data]两个 Agent 同时获取天气和新闻结果存入各自的 state key。示例 5ADK 智能体即工具fromgoogle.adk.agentsimportLlmAgentfromgoogle.adk.toolsimportagent_toolfromgoogle.genaiimporttypes GEMINI_MODELgemini-2.0-flashdefgenerate_image(prompt:str)-dict:模拟图片生成工具。print(fTOOL: 正在为提示 {prompt} 生成图片)mock_image_bytesbmock_image_data_for_a_cat_wearing_a_hatreturn{status:success,image_bytes:mock_image_bytes,mime_type:image/png}# 子智能体负责调用图片生成工具image_generator_agentLlmAgent(nameImageGen,modelGEMINI_MODEL,description根据详细文本提示生成图片。,instruction(你是图片生成专家使用 generate_image 工具根据用户请求生成图片。用户请求作为工具的 prompt 参数。工具返回图片字节后必须输出图片。),tools[generate_image])# 将子智能体包装为工具image_toolagent_tool.AgentTool(agentimage_generator_agent,description用于生成图片输入为描述性提示。)# 父智能体先创造提示再通过工具调用子智能体生成图片artist_agentLlmAgent(nameArtist,modelGEMINI_MODEL,instruction(你是一名创意艺术家先创造图片提示再用 ImageGen 工具生成图片。),tools[image_tool])演示智能体即工具模式父智能体通过AgentTool调用子智能体[Artist 父智能体] → 生成创意提示 → 调用 ImageGen 工具 ↓ [ImageGen 子智能体] → generate_image() → 返回图片image_generator_agent子智能体配备generate_image工具函数AgentTool将子智能体包装为工具供父智能体调用artist_agent父智能体先创造提示再通过工具生成图片实现分层协作关键要点多智能体协作即多个智能体共同实现目标该模式利用专长分工、任务分布与智能体间通信协作形式包括顺序交接、并行处理、辩论或层级结构适用于需多领域专长或多阶段复杂问题通信模型选择网络型/监督者/层级型/定制型应结合任务复杂度、智能体数量、健壮性等因素权衡总结本章介绍了多智能体协作模式阐述了多专长智能体协同系统的优势。通过多种协作模型顺序、并行、层级、智能体即工具强调该模式在解决复杂、多元问题中的关键作用。理解智能体协作后下一步将探讨其与外部环境的交互。
【Agent 设计模式】多智能体协作
发布时间：2026/7/30 23:02:56
多智能体协作Multi-Agent Collaboration单一 LLM 智能体难以应对复杂问题缺乏多样化专长或工具成为系统瓶颈降低效率与可扩展性难以完成多领域目标。多智能体协作模式通过将复杂问题拆分为可管理子问题分配给具备专长的智能体智能体通过顺序交接、并行处理或层级委托等协议协作实现单一智能体无法完成的目标。核心思想专长分工 协作通信—— 多个专精智能体各司其职通过协议协同完成复杂任务。适用场景当任务过于复杂需拆分为需专长或工具的子任务时适合采用该模式。适用于多领域、并行处理或多阶段结构化流程如复杂研究、软件开发、创意内容生成等。场景协作方式客户支持升级前线支持智能体处理初步问题复杂问题升级至专家智能体如技术或账单专家体现基于问题复杂度的顺序交接供应链优化智能体代表供应链各节点供应商、制造商、分销商协作优化库存、物流与排程应对需求变化或突发事件网络分析与修复自治运维场景下多智能体协作进行故障定位、分级处理与修复可集成传统机器学习模型与工具通过智能体专长划分与关系精细编排开发者可构建具备更强模块化、可扩展性和复杂问题处理能力的系统。智能体关系与通信结构理解智能体间的交互与通信方式是设计高效多智能体系统的基础。从最简单的单智能体到复杂的定制协作结构呈现多样化选择。每种模型有独特优势与挑战影响系统整体效率、健壮性与适应性。1. 单智能体最基础模型智能体独立运行无需与其他实体交互适合可拆分为独立子问题的场景但能力受限。2. 网络型多个智能体以去中心化方式直接交互点对点通信信息、资源和任务共享具备弹性但通信管理和决策一致性较难。[A] ←→ [B] ↕ ╲ ↕ [C] ←→ [D]3. 监督者专门智能体监督者协调下属智能体负责通信、任务分配和冲突解决层级结构清晰易于管理但存在单点故障和瓶颈风险。[监督者] / | \ [A] [B] [C]4. 工具型监督者监督者不直接指挥而是为其他智能体提供资源、指导或分析支持赋能而非强制控制提升灵活性。5. 层级型多层监督者结构高层监督者管理低层监督者底层为操作智能体适合复杂问题分层管理便于扩展和分布式决策。[总监督者] / \ [子监督者A] [子监督者B] / \ / \ [A1] [A2] [B1] [B2]6. 定制型最灵活模型针对具体问题或应用定制独特关系与通信结构可混合前述模型或创新设计适合优化特定性能、动态环境或领域知识集成。定制模型需深入理解多智能体原理慎重设计通信协议、协调机制与涌现行为。综上多智能体系统的关系与通信模型选择至关重要应结合任务复杂度、智能体数量、自治需求、健壮性和通信开销等因素权衡。代码示例示例 1CrewAI 实现importosfromdotenvimportload_dotenvfromcrewaiimportAgent,Task,Crew,Processfromlangchain_openaiimportChatOpenAIdefsetup_environment():加载环境变量并检查 API 密钥。load_dotenv()ifnotos.getenv(OPENAI_API_KEY):raiseValueError(OPENAI_API_KEY 未设置请在 .env 文件中配置。)defmain(): 初始化并运行内容创作 AI 团队。 setup_environment()# 指定语言模型llmChatOpenAI(modelgpt-4o-mini)# 定义 Agent 角色与目标researcherAgent(role高级研究分析师,goal查找并总结 AI 最新趋势。,backstory你是一名经验丰富的研究分析师擅长发现关键趋势并整合信息。,verboseTrue,allow_delegationFalse,)writerAgent(role技术内容写作者,goal根据研究结果撰写清晰易懂的博客。,backstory你是一名技术写作高手能将复杂技术转化为通俗内容。,verboseTrue,allow_delegationFalse,)# 定义任务research_taskTask(description调研 2024-2025 年 AI 三大新兴趋势关注实际应用与影响。,expected_output详细总结三大 AI 趋势包括要点与来源。,agentresearcher,)writing_taskTask(description根据研究结果撰写一篇 500 字博客内容通俗易懂。,expected_output完整的 500 字 AI 趋势博客。,agentwriter,context[research_task],)# 创建团队blog_creation_crewCrew(agents[researcher,writer],tasks[research_task,writing_task],processProcess.sequential,llmllm,verboseTrue)# 执行团队任务print(## 运行博客创作团队... ##)try:resultblog_creation_crew.kickoff()print(\n------------------\n)print(## 团队最终输出 ##)print(result)exceptExceptionase:print(f\n发生异常{e})if__name____main__:main()CrewAI 是专为多智能体协作设计的框架。示例展示如何定义一个 AI 协作团队生成 AI 趋势博客[研究员] → research_task → [写作者] → writing_task → 博客输出研究员Agent查找并总结 AI 最新趋势写作者Agent根据研究结果撰写 500 字博客任务依赖writing_task的context[research_task]确保写作基于研究结果顺序执行Process.sequential指定研究完成后再写作团队执行通过crew.kickoff()启动协作流程示例 2ADK 层级结构fromgoogle.adk.agentsimportLlmAgent,BaseAgentfromgoogle.adk.agents.invocation_contextimportInvocationContextfromgoogle.adk.eventsimportEventfromtypingimportAsyncGenerator GEMINI_MODELgemini-2.0-flash-expclassTaskExecutor(BaseAgent):自定义非 LLM 行为 Agent。name:strTaskExecutordescription:str执行预定义任务。asyncdef_run_async_impl(self,context:InvocationContext)-AsyncGenerator[Event,None]:yieldEvent(authorself.name,content任务成功完成。)greeterLlmAgent(nameGreeter,modelGEMINI_MODEL,instruction你是一名友好的问候者。)task_doerTaskExecutor()coordinatorLlmAgent(nameCoordinator,modelGEMINI_MODEL,description协调问候与任务执行。,instruction问候时委托 Greeter执行任务时委托 TaskExecutor。,sub_agents[greeter,task_doer])assertgreeter.parent_agentcoordinatorasserttask_doer.parent_agentcoordinatorprint(Agent 层级关系创建成功。)演示 Google ADK 中通过父子关系实现层级协作[Coordinator] / \ [Greeter] [TaskExecutor]LlmAgentGreeter负责问候BaseAgentTaskExecutor自定义非 LLM 智能体执行预定义任务返回任务成功完成事件Coordinator父智能体通过sub_agents参数建立父子关系根据请求类型委托给对应子智能体示例 3ADK LoopAgentimportasynciofromtypingimportAsyncGeneratorfromgoogle.adk.agentsimportLoopAgent,LlmAgent,BaseAgentfromgoogle.adk.eventsimportEvent,EventActionsfromgoogle.adk.agents.invocation_contextimportInvocationContext GEMINI_MODELgemini-2.0-flash-expclassConditionChecker(BaseAgent):检查流程是否完成并控制循环。name:strConditionCheckerdescription:str检查流程完成状态并通知循环终止。asyncdef_run_async_impl(self,context:InvocationContext)-AsyncGenerator[Event,None]:statuscontext.session.state.get(status,pending)is_done(statuscompleted)ifis_done:yieldEvent(authorself.name,actionsEventActions(escalateTrue))else:yieldEvent(authorself.name,content条件未满足继续循环。)process_stepLlmAgent(nameProcessingStep,modelGEMINI_MODEL,instruction你是流程中的一步完成任务。若为最后一步将 session 状态设为 completed。)pollerLoopAgent(nameStatusPoller,max_iterations10,sub_agents[process_step,ConditionChecker()])演示LoopAgent实现迭代流程LoopAgent (max10) ├→ [ProcessingStep] → 处理任务最后一步设 statuscompleted └→ [ConditionChecker] → statuscompleted? → 是: escalate 终止 / 否: 继续循环ConditionChecker自定义BaseAgent检查 session 状态若status为completed则通过EventActions(escalateTrue)终止循环ProcessingStepLLM 智能体负责处理任务并在最后一步设置状态LoopAgent配置max_iterations10循环执行直到条件满足或达到最大次数示例 4ADK 顺序与并行fromgoogle.adk.agentsimportAgent,SequentialAgent,ParallelAgent GEMINI_MODELgemini-2.0-flash-exp# # 示例 1SequentialAgent 顺序执行# step1 输出存入 session.state[data]step2 分析该数据# step1Agent(nameStep1_Fetch,output_keydata)step2Agent(nameStep2_Process,instruction分析 state[data] 信息并给出总结。)pipelineSequentialAgent(nameMyPipeline,sub_agents[step1,step2])# # 示例 2ParallelAgent 并行执行# 两个 Agent 同时获取天气和新闻结果存入各自的 state key# weather_fetcherAgent(nameweather_fetcher,modelGEMINI_MODEL,instruction获取指定地点天气仅返回天气报告。,output_keyweather_data)news_fetcherAgent(namenews_fetcher,modelGEMINI_MODEL,instruction获取指定主题头条新闻仅返回新闻内容。,output_keynews_data)data_gathererParallelAgent(namedata_gatherer,sub_agents[weather_fetcher,news_fetcher])SequentialAgent 顺序执行[Step1_Fetch] → state[data] → [Step2_Process] → 总结step1输出存入session.state[data]step2分析该数据并给出总结。ParallelAgent 并行执行ParallelAgent ├→ [weather_fetcher] → state[weather_data] └→ [news_fetcher] → state[news_data]两个 Agent 同时获取天气和新闻结果存入各自的 state key。示例 5ADK 智能体即工具fromgoogle.adk.agentsimportLlmAgentfromgoogle.adk.toolsimportagent_toolfromgoogle.genaiimporttypes GEMINI_MODELgemini-2.0-flashdefgenerate_image(prompt:str)-dict:模拟图片生成工具。print(fTOOL: 正在为提示 {prompt} 生成图片)mock_image_bytesbmock_image_data_for_a_cat_wearing_a_hatreturn{status:success,image_bytes:mock_image_bytes,mime_type:image/png}# 子智能体负责调用图片生成工具image_generator_agentLlmAgent(nameImageGen,modelGEMINI_MODEL,description根据详细文本提示生成图片。,instruction(你是图片生成专家使用 generate_image 工具根据用户请求生成图片。用户请求作为工具的 prompt 参数。工具返回图片字节后必须输出图片。),tools[generate_image])# 将子智能体包装为工具image_toolagent_tool.AgentTool(agentimage_generator_agent,description用于生成图片输入为描述性提示。)# 父智能体先创造提示再通过工具调用子智能体生成图片artist_agentLlmAgent(nameArtist,modelGEMINI_MODEL,instruction(你是一名创意艺术家先创造图片提示再用 ImageGen 工具生成图片。),tools[image_tool])演示智能体即工具模式父智能体通过AgentTool调用子智能体[Artist 父智能体] → 生成创意提示 → 调用 ImageGen 工具 ↓ [ImageGen 子智能体] → generate_image() → 返回图片image_generator_agent子智能体配备generate_image工具函数AgentTool将子智能体包装为工具供父智能体调用artist_agent父智能体先创造提示再通过工具生成图片实现分层协作关键要点多智能体协作即多个智能体共同实现目标该模式利用专长分工、任务分布与智能体间通信协作形式包括顺序交接、并行处理、辩论或层级结构适用于需多领域专长或多阶段复杂问题通信模型选择网络型/监督者/层级型/定制型应结合任务复杂度、智能体数量、健壮性等因素权衡总结本章介绍了多智能体协作模式阐述了多专长智能体协同系统的优势。通过多种协作模型顺序、并行、层级、智能体即工具强调该模式在解决复杂、多元问题中的关键作用。理解智能体协作后下一步将探讨其与外部环境的交互。
