如果你正在尝试用 AI Agent 解决一个复杂任务比如写一份完整的商业计划书你可能会发现让一个 Agent 从头写到尾结果往往不尽人意。它可能擅长市场分析却在财务预测上漏洞百出或者能写出漂亮的文案却无法生成结构严谨的图表。你得到的可能是一份“半成品”或“四不像”。这正是当前 AI 应用从“单兵作战”迈向“团队协作”的关键转折点。一个 Agent 的能力是线性的、有限的而现实世界的复杂问题往往是多维的、非线性的。“多 Agent 协作”的核心价值就在于通过分工、协商与调度让一群各有所长的 AI 智能体协同工作解决单个 Agent 无法独立完成的复杂任务。这篇文章不会空谈“多智能体系统”的学术概念而是聚焦于一个更实际的问题作为一个开发者或技术决策者你该如何从零开始设计并实现一个有效的多 Agent 协作系统我们将从最基础的协作模式讲起拆解“顺序流水线”、“辩论协商”、“分层调度”这三种核心架构并用具体的代码示例带你一步步搭建一个能真正跑起来的协作系统。你会看到从“一个 Agent 单干”到“一群 Agent 协作”不仅仅是数量的增加更是架构思维和工程实践的全面升级。1. 这篇文章真正要解决的问题为什么我们需要关注多 Agent 协作根本原因在于“能力边界”与“任务复杂度”之间的不匹配。一个训练有素的文本生成 Agent可能对代码一窍不通一个精通数据分析的 Agent可能无法理解业务逻辑。当你把一项包含市场调研、竞品分析、财务建模、文档撰写和 PPT 制作的任务丢给一个“全能型”Agent 时它大概率会顾此失彼产出质量难以保证。这就像让一个程序员同时负责前端、后端、运维和产品设计结果可想而知。多 Agent 协作要解决的正是这种“专业的事交给专业的人Agent”的问题。它的目标不是创造一个“超级大脑”而是构建一个高效的“项目团队”。这个团队里有“产品经理”Agent 负责拆解需求和制定计划有“市场分析师”Agent 负责搜集数据有“程序员”Agent 负责生成代码或处理结构化数据有“设计师”Agent 负责美化和排版最后还有一个“项目经理”Agent 负责协调进度、整合成果。对于开发者而言实现多 Agent 协作的挑战在于架构设计Agent 之间如何通信任务如何流转状态如何同步角色定义如何为每个 Agent 设定清晰、不重叠的职责和上下文协作逻辑是简单的“流水线”传递还是需要复杂的“辩论”与“投票”工程实现如何用现有的框架如 LangChain、AutoGen快速搭建如何管理成本、监控性能和排查错误本文将围绕这些实际问题提供从概念到落地的完整路径。如果你正在评估是否要将 AI 能力深度集成到你的产品中或者你已经被单个 Agent 的能力天花板所困扰那么这篇文章正是为你准备的。2. 基础概念与核心原理在深入实践之前我们需要统一几个关键术语的理解这些概念是多 Agent 系统的基石。Agent智能体在本文语境下特指一个具备一定自主性的软件实体。它通常由三部分组成大脑LLM大型语言模型负责理解、推理和决策。工具ToolsAgent 可以调用的外部能力如搜索网络、执行代码、查询数据库、调用 API。记忆Memory短期或长期的记忆存储用于保存对话历史、任务上下文、执行结果等。记忆决定了 Agent 的“经验”和“连贯性”。多 Agent 协作系统由多个上述 Agent 组成的系统它们通过特定的通信协议和协作机制共同完成一个总目标。系统的智能体现在 Agent 之间的交互与协同上而非单个 Agent 的能力。三种核心协作模式对应热搜词顺序流水线Sequential Pipeline最简单、最常见的模式。任务被分解为多个步骤每个步骤由一个特定的 Agent 处理其输出作为下一个 Agent 的输入。就像工厂的装配线适合流程清晰、步骤依赖强的任务如“数据抓取 - 数据清洗 - 数据分析 - 报告生成”。辩论协商Debate Negotiation用于解决存在不确定性或需要多角度评估的任务。多个 Agent 从不同立场或专长出发对同一问题提出方案并进行“辩论”最终通过协商或由一个“裁判”Agent 综合各方意见得出最优解。例如在评估一个商业方案的风险时可以让“风控 Agent”、“市场 Agent”、“技术 Agent”分别发表意见。分层调度Hierarchical Scheduling引入管理层级。通常有一个“管理者Manager”Agent 或“编排器Orchestrator”。管理者负责接收总任务将其分解为子任务然后根据子任务的性质调度Dispatch给下属的“工作者Worker”Agent 去执行并最终汇总结果。这种模式扩展性好适合复杂、动态的任务规划。为了更直观地理解它们的区别和适用场景可以参考下表协作模式核心思想通信方式适用场景优点缺点顺序流水线流水线作业环环相扣单向链式传递步骤固定、依赖明确的流程化任务文档处理、数据ETL结构简单易于实现和调试缺乏灵活性一个环节失败会导致整个流程中断辩论协商多角度评估寻求共识多向对话式创意生成、方案评估、复杂决策产品设计、投资分析能产生更全面、更稳健的解决方案计算成本高多次LLM调用流程可能冗长分层调度中央指挥分工作业星型管理者与工作者间通信复杂项目规划、动态任务分配智能客服、研发项目管理灵活性强易于扩展和监控对“管理者”Agent的规划能力要求高是单点瓶颈理解了这些模式我们就可以根据具体任务来选择架构。接下来我们将进入实战环节从环境准备开始。3. 环境准备与前置条件我们将使用 Python 作为主要开发语言并借助LangChain和LangGraph这两个目前生态最成熟、文档最丰富的框架来构建我们的多 Agent 系统。LangChain 提供了构建 Agent 的基础模块而 LangGraph 则专门用于描述和运行基于图的、多参与者的复杂工作流完美契合多 Agent 协作的场景。基础环境要求操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04 推荐)。本文示例在 Linux/macOS 的终端环境下测试。Python版本 3.8 或更高。建议使用 3.9 或 3.10 以获得最佳兼容性。包管理工具pip(Python 自带) 或conda(如果你使用 Anaconda)。核心依赖安装打开你的终端或命令提示符创建一个新的虚拟环境强烈推荐以避免包冲突然后安装以下包# 创建并激活虚拟环境 (以 venv 为例) python -m venv multi_agent_env source multi_agent_env/bin/activate # Linux/macOS # multi_agent_env\Scripts\activate # Windows # 升级 pip pip install --upgrade pip # 安装核心框架 pip install langchain langchain-community langgraph # 安装用于 Agent 工具链的常用包 pip install wikipedia # 示例中可能用到的工具 pip install duckduckgo-search # 用于网络搜索LLM 服务配置Agent 的“大脑”需要一个大语言模型。你可以使用 OpenAI 的 API也可以使用开源的本地模型通过 Ollama、vLLM 等部署。为了演示的通用性我们以 OpenAI API 为例。你需要准备一个有效的 OpenAI API Key。# 设置环境变量 (临时方式更安全的方式是使用 .env 文件) export OPENAI_API_KEYyour-api-key-here # Linux/macOS # set OPENAI_API_KEYyour-api-key-here # Windows代码编辑器任何你熟悉的 IDE 或编辑器即可如 VS Code、PyCharm。环境准备好后我们就可以开始设计并实现第一个协作模式了。4. 核心流程拆解以顺序流水线为例让我们从一个最直观的场景开始自动生成一份行业分析简报。这个任务可以清晰地分解为1) 搜索最新行业动态2) 总结核心观点3) 生成一份格式优美的 Markdown 报告。我们将创建三个 Agent 来组成一个流水线研究员 (Researcher)负责搜索和获取原始信息。分析师 (Analyst)负责消化信息提炼出核心要点。编辑 (Editor)负责将要点组织成结构化的报告。整个系统的流程如下图所示概念图用户输入“生成关于电动汽车的简报” | v [研究员 Agent] --(搜索到的原始文本)-- [分析师 Agent] --(提炼出的要点)-- [编辑 Agent] | | | v (调用搜索工具) (生成最终 Markdown 报告)关键步骤拆解定义每个 Agent 的角色与工具明确每个 Agent 的职责和它能使用的“技能”Tools。研究员需要搜索工具分析师和编辑主要依靠 LLM 的推理和生成能力。构建 Agent 实例使用 LangChain 的create_react_agent或其他 Agent 构造函数为每个角色创建独立的 Agent 对象并绑定相应的工具和提示词Prompt。设计工作流图使用 LangGraph 的StateGraph来定义工作流。图中的每个节点Node对应一个 Agent 的执行函数边Edge定义了执行顺序和数据流向。编译并运行图将定义好的图编译成可执行的对象然后传入初始状态用户问题启动整个协作流程。处理与输出结果从最终节点获取处理结果并返回给用户。这个流程是构建任何多 Agent 系统的基础模板。接下来我们用代码把它实现出来。5. 完整示例与代码实现我们将实现上面描述的“行业分析简报流水线”。请确保你已经完成了环境准备并设置了OPENAI_API_KEY。首先创建一个名为sequential_pipeline.py的 Python 文件。5.1 导入依赖与初始化 LLM# sequential_pipeline.py import os from typing import TypedDict, Annotated, List import operator from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import HumanMessage, AIMessage from langgraph.graph import StateGraph, END # 1. 初始化 LLM # 使用 gpt-3.5-turbo 以控制成本可根据需要换成 gpt-4 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 2. 定义搜索工具 search_tool DuckDuckGoSearchRun()5.2 定义系统状态在 LangGraph 中工作流的状态State是一个在所有节点间共享和传递的数据结构。我们定义一个State类来明确需要传递哪些信息。# 定义图的状态结构 class AgentState(TypedDict): 所有 Agent 共享和更新的状态 # 用户最初的问题 original_query: str # 研究员 Agent 搜索到的原始信息 raw_information: str # 分析师 Agent 提炼出的核心要点 key_points: str # 编辑 Agent 生成的最终报告 final_report: str # 用于记录中间步骤的消息历史可选用于复杂交互 messages: Annotated[List, operator.add]5.3 创建各个 Agent我们为三个角色分别创建 Agent。注意我们为“研究员”赋予了搜索工具而“分析师”和“编辑”则没有额外工具专注于文本处理。# 3. 创建研究员 Agent (拥有搜索能力) researcher_prompt ChatPromptTemplate.from_messages([ (system, 你是一位专业的研究员。你的任务是利用搜索工具针对用户的问题尽可能全面地搜集最新、最相关的公开信息。 请直接返回搜索到的原始信息摘要不要进行加工分析。如果一次搜索不够可以多次搜索。), MessagesPlaceholder(variable_namechat_history), (human, {input}), ]) researcher_agent create_react_agent(llm, tools[search_tool], promptresearcher_prompt) researcher_executor AgentExecutor(agentresearcher_agent, tools[search_tool], verboseFalse) # 4. 创建分析师 Agent (无工具专注分析) analyst_prompt ChatPromptTemplate.from_messages([ (system, 你是一位资深行业分析师。你的任务是从研究员提供的大段原始信息中提炼出3-5个最核心的要点。 要点需要简洁、清晰、有洞察力避免直接复制原文。请用分点列表的形式输出。), (human, 请分析以下信息提炼核心要点\n\n{raw_info}), ]) # 分析师不需要工具直接使用 LLM analyst_chain analyst_prompt | llm # 5. 创建编辑 Agent (无工具专注报告撰写) editor_prompt ChatPromptTemplate.from_messages([ (system, 你是一位专业的科技编辑。你的任务是将分析师提炼的要点整理成一份结构完整、语言流畅、适合高管阅读的 Markdown 格式简报。 简报应包含标题、概述、核心要点分点阐述、总结与展望。请确保格式美观。), (human, 请根据以下核心要点撰写一份行业简报\n\n{key_points}), ]) editor_chain editor_prompt | llm5.4 定义图节点Node函数每个节点函数接收当前的State执行特定 Agent 的任务并更新State中的相应字段。# 6. 定义图节点函数 def research_node(state: AgentState) - AgentState: 研究员节点执行搜索更新 raw_information print( 研究员开始工作 ) result researcher_executor.invoke({input: state[original_query], chat_history: []}) # 更新状态 return {raw_information: result[output]} def analyze_node(state: AgentState) - AgentState: 分析师节点分析原始信息更新 key_points print( 分析师开始工作 ) result analyst_chain.invoke({raw_info: state[raw_information]}) return {key_points: result.content} def edit_node(state: AgentState) - AgentState: 编辑节点生成最终报告更新 final_report print( 编辑开始工作 ) result editor_chain.invoke({key_points: state[key_points]}) return {final_report: result.content}5.5 构建并运行工作流图现在我们将节点和边组合起来形成完整的工作流。# 7. 构建工作流图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(researcher, research_node) workflow.add_node(analyst, analyze_node) workflow.add_node(editor, edit_node) # 设置边的连接关系 (顺序流水线) workflow.set_entry_point(researcher) # 从研究员开始 workflow.add_edge(researcher, analyst) workflow.add_edge(analyst, editor) workflow.add_edge(editor, END) # 结束 # 编译图 app workflow.compile() # 8. 运行工作流 if __name__ __main__: # 初始化状态 initial_state: AgentState { original_query: 请生成一份关于2024年人工智能大模型在金融领域应用最新趋势的简报。, raw_information: , key_points: , final_report: , messages: [] } print(开始执行多Agent协作流水线...\n) # 运行图从 entry_point 开始 final_state app.invoke(initial_state) print(\n *50) print(【最终报告生成完毕】) print(*50) print(final_state[final_report])6. 运行结果与效果验证保存sequential_pipeline.py文件并在终端中运行它python sequential_pipeline.py预期输出流程终端会依次打印 研究员开始工作 、 分析师开始工作 、 编辑开始工作 表明三个 Agent 正在按顺序执行。由于DuckDuckGoSearchRun工具会进行真实的网络搜索研究员阶段可能需要几秒到十几秒时间。最终你会看到一份格式完整的 Markdown 报告输出在终端上。报告内容示例片段# 2024年人工智能大模型在金融领域应用趋势简报 ## 概述 2024年生成式人工智能与大模型技术正加速渗透至金融行业的核心业务环节从风险管控、投资研究到客户服务展现出从“辅助工具”向“核心生产力”演进的明确趋势。 ## 核心要点 1. **风险管理的深度智能化**大模型开始用于解读非结构化数据如财报电话会议记录、新闻舆情实现更动态、前瞻性的信用风险和操作风险评估。 2. **投资研究的效率革命**... 3. **合规与监管科技的升级**... 4. **个性化财富管理的普及**... ## 总结与展望 ...如何验证成功流程验证观察控制台输出确认三个节点都按顺序执行完毕没有报错。内容验证检查raw_information字段可以在代码中打印出来是否包含了从网络搜索到的相关文本。检查key_points字段是否是从原始信息中提炼出的分点列表。检查final_report是否是一份结构清晰有标题、概述、要点、总结、语言通顺的 Markdown 文档。质量验证人工阅读最终报告判断其是否基本回答了初始问题信息是否连贯格式是否正确。如果运行失败请首先检查API KeyOPENAI_API_KEY环境变量是否设置正确且有效。网络连接是否能正常访问 OpenAI API 和 DuckDuckGo 搜索。依赖包是否安装了langchain-openai如果使用最新 LangChain可能需要单独安装pip install langchain-openai。错误信息仔细阅读控制台输出的错误堆栈信息通常能定位到具体问题。7. 常见问题与排查思路在多 Agent 系统开发中你会遇到一些典型问题。下表列出了常见现象、原因及解决方法问题现象可能原因排查方式解决方案Agent 执行超时或无响应1. LLM API 调用缓慢或失败。2. 某个工具如搜索耗时过长。3. 工作流陷入循环。1. 查看 API 返回的错误信息。2. 在节点函数内添加日志打印开始/结束时间。3. 检查图的结构是否有循环边未正确终止。1. 增加 API 调用的超时设置。2. 为耗时工具设置独立的超时限制。3. 使用interrupt_before/after在 LangGraph 中设置检查点或引入最大循环次数限制。最终结果质量差1. Agent 的提示词Prompt不清晰。2. 上游 Agent 输出质量差导致垃圾进、垃圾出。3. Agent 之间传递的信息丢失了关键上下文。1. 分别测试每个 Agent 在孤立任务上的表现。2. 打印并检查每个节点输出 (raw_information,key_points)。3. 审查 State 中传递的数据是否完整。1. 迭代优化每个 Agent 的 System Prompt明确其角色和输出格式要求。2. 在上游 Agent 的输出中强制加入结构化标记如##SUMMARY_START## ... ##SUMMARY_END##便于下游解析。3. 在 State 中传递更丰富的上下文或让 Manager Agent 负责补充上下文。成本过高或 Token 消耗巨大1. 工作流中 LLM 调用次数过多。2. 传递的上下文如原始网页内容过长。3. 使用了更昂贵的模型如 GPT-4。1. 统计每个节点的调用次数和输入/输出的 token 数。2. 使用模型的get_num_tokens方法估算。1. 优化流程避免不必要的 LLM 调用例如缓存中间结果。2. 在上游节点对长文本进行摘要或过滤再传递给下游。3. 在非关键节点使用成本更低的模型如 GPT-3.5-Turbo。4. 设置预算和用量监控。“辩论协商”模式陷入僵局多个 Agent 争论不休无法达成一致。记录辩论过程中的所有消息。1. 引入“裁判”Agent其 Prompt 明确要求综合各方意见并做出最终决策。2. 设置辩论轮次上限超时后由裁判强制裁决。3. 定义清晰的投票或评分机制。工具调用失败1. 工具所需的 API Key 未配置。2. 工具输入参数格式错误。3. 外部服务不可用。1. 检查工具初始化时的配置。2. 在 Agent 之外单独测试工具调用。3. 查看工具返回的具体错误信息。1. 确保所有必要的环境变量已设置。2. 在 Agent 的 Prompt 中更详细地描述工具的使用方法和参数格式。3. 为工具调用添加重试和降级逻辑如返回模拟数据。8. 最佳实践与工程建议构建可用于生产环境的多 Agent 系统远不止让流程跑通那么简单。以下是一些关键的工程实践1. 提示词工程是核心角色扮演要彻底每个 Agent 的 System Prompt 必须清晰定义其角色、职责、输出格式和边界。例如“你是一位严谨的风险评估师只关注技术可行性和潜在漏洞不要提供商业建议。”上下文管理明确告诉 Agent 哪些信息是来自上游的输入哪些是它需要产出的。使用明确的占位符和格式要求。迭代优化将 Prompt 作为代码一样管理进行版本控制和 A/B 测试。2. 状态设计要精简而充分只传递必要信息State 中不要包含所有中间过程只存放下游节点必需的数据。过大的 State 会增加 Token 消耗和传输开销。使用结构化数据尽量使用字典、列表等结构化数据类型在 State 中传递信息避免冗长的自然语言段落除非下游 LLM 必须使用它。考虑记忆外挂对于需要长期记忆或知识库的场景不要依赖 State而是为 Agent 配备向量数据库等外部记忆体。3. 引入健壮性与可观测性错误处理与回退在每个节点函数中使用try...except包裹核心逻辑。当某个 Agent 失败时应有备选方案如使用简化流程、返回默认值、通知人类。日志与监控记录每个 Agent 的输入、输出、耗时和 Token 使用情况。这有助于性能优化、成本分析和问题排查。可以考虑使用 LangSmith 等专门的可观测性平台。设置超时与循环限制在 LangGraph 中对于可能循环的路径如辩论模式必须使用add_conditional_edges并设置最大迭代次数防止无限循环消耗资源。4. 成本与性能优化模型分级使用在流水线中对创造性要求不高的环节如信息提取、格式转换使用廉价快速的模型如 GPT-3.5-Turbo只在核心创意或决策环节使用强大但昂贵的模型如 GPT-4。缓存与去重如果多个 Agent 可能查询相同的外部信息如股票价格引入缓存层避免重复调用和费用。异步执行当 Agent 之间没有严格的先后依赖时可以利用 LangGraph 的异步支持或asyncio并发执行缩短整体流程时间。5. 安全与边界控制工具权限隔离不是所有 Agent 都需要所有工具。为每个 Agent 精确分配最小必要权限的工具集。例如只有“数据清洗”Agent 能访问数据库写操作。输入输出审查在系统的入口和出口设置审查环节对用户输入进行过滤对 Agent 的最终输出进行内容安全校验防止生成有害或不适当的内容。人工审核环节对于关键业务或高风险决策将多 Agent 系统的输出作为“建议”最终决策需加入人工审核确认步骤。从“顺序流水线”到“辩论协商”和“分层调度”架构会变得更复杂但核心思想不变定义清晰的 Agent 角色设计高效可靠的通信机制并用工作流图将它们有机地组织起来。掌握了流水线模式你就已经拿到了打开多 Agent 协作世界大门的钥匙。接下来你可以尝试让“研究员”和“分析师”就某个数据点进行“辩论”或者引入一个“项目经理”Agent 来动态分配任务探索更强大的协作模式。 从零构建多Agent协作系统：架构、实现与工程实践
如果你正在尝试用 AI Agent 解决一个复杂任务比如写一份完整的商业计划书你可能会发现让一个 Agent 从头写到尾结果往往不尽人意。它可能擅长市场分析却在财务预测上漏洞百出或者能写出漂亮的文案却无法生成结构严谨的图表。你得到的可能是一份“半成品”或“四不像”。这正是当前 AI 应用从“单兵作战”迈向“团队协作”的关键转折点。一个 Agent 的能力是线性的、有限的而现实世界的复杂问题往往是多维的、非线性的。“多 Agent 协作”的核心价值就在于通过分工、协商与调度让一群各有所长的 AI 智能体协同工作解决单个 Agent 无法独立完成的复杂任务。这篇文章不会空谈“多智能体系统”的学术概念而是聚焦于一个更实际的问题作为一个开发者或技术决策者你该如何从零开始设计并实现一个有效的多 Agent 协作系统我们将从最基础的协作模式讲起拆解“顺序流水线”、“辩论协商”、“分层调度”这三种核心架构并用具体的代码示例带你一步步搭建一个能真正跑起来的协作系统。你会看到从“一个 Agent 单干”到“一群 Agent 协作”不仅仅是数量的增加更是架构思维和工程实践的全面升级。1. 这篇文章真正要解决的问题为什么我们需要关注多 Agent 协作根本原因在于“能力边界”与“任务复杂度”之间的不匹配。一个训练有素的文本生成 Agent可能对代码一窍不通一个精通数据分析的 Agent可能无法理解业务逻辑。当你把一项包含市场调研、竞品分析、财务建模、文档撰写和 PPT 制作的任务丢给一个“全能型”Agent 时它大概率会顾此失彼产出质量难以保证。这就像让一个程序员同时负责前端、后端、运维和产品设计结果可想而知。多 Agent 协作要解决的正是这种“专业的事交给专业的人Agent”的问题。它的目标不是创造一个“超级大脑”而是构建一个高效的“项目团队”。这个团队里有“产品经理”Agent 负责拆解需求和制定计划有“市场分析师”Agent 负责搜集数据有“程序员”Agent 负责生成代码或处理结构化数据有“设计师”Agent 负责美化和排版最后还有一个“项目经理”Agent 负责协调进度、整合成果。对于开发者而言实现多 Agent 协作的挑战在于架构设计Agent 之间如何通信任务如何流转状态如何同步角色定义如何为每个 Agent 设定清晰、不重叠的职责和上下文协作逻辑是简单的“流水线”传递还是需要复杂的“辩论”与“投票”工程实现如何用现有的框架如 LangChain、AutoGen快速搭建如何管理成本、监控性能和排查错误本文将围绕这些实际问题提供从概念到落地的完整路径。如果你正在评估是否要将 AI 能力深度集成到你的产品中或者你已经被单个 Agent 的能力天花板所困扰那么这篇文章正是为你准备的。2. 基础概念与核心原理在深入实践之前我们需要统一几个关键术语的理解这些概念是多 Agent 系统的基石。Agent智能体在本文语境下特指一个具备一定自主性的软件实体。它通常由三部分组成大脑LLM大型语言模型负责理解、推理和决策。工具ToolsAgent 可以调用的外部能力如搜索网络、执行代码、查询数据库、调用 API。记忆Memory短期或长期的记忆存储用于保存对话历史、任务上下文、执行结果等。记忆决定了 Agent 的“经验”和“连贯性”。多 Agent 协作系统由多个上述 Agent 组成的系统它们通过特定的通信协议和协作机制共同完成一个总目标。系统的智能体现在 Agent 之间的交互与协同上而非单个 Agent 的能力。三种核心协作模式对应热搜词顺序流水线Sequential Pipeline最简单、最常见的模式。任务被分解为多个步骤每个步骤由一个特定的 Agent 处理其输出作为下一个 Agent 的输入。就像工厂的装配线适合流程清晰、步骤依赖强的任务如“数据抓取 - 数据清洗 - 数据分析 - 报告生成”。辩论协商Debate Negotiation用于解决存在不确定性或需要多角度评估的任务。多个 Agent 从不同立场或专长出发对同一问题提出方案并进行“辩论”最终通过协商或由一个“裁判”Agent 综合各方意见得出最优解。例如在评估一个商业方案的风险时可以让“风控 Agent”、“市场 Agent”、“技术 Agent”分别发表意见。分层调度Hierarchical Scheduling引入管理层级。通常有一个“管理者Manager”Agent 或“编排器Orchestrator”。管理者负责接收总任务将其分解为子任务然后根据子任务的性质调度Dispatch给下属的“工作者Worker”Agent 去执行并最终汇总结果。这种模式扩展性好适合复杂、动态的任务规划。为了更直观地理解它们的区别和适用场景可以参考下表协作模式核心思想通信方式适用场景优点缺点顺序流水线流水线作业环环相扣单向链式传递步骤固定、依赖明确的流程化任务文档处理、数据ETL结构简单易于实现和调试缺乏灵活性一个环节失败会导致整个流程中断辩论协商多角度评估寻求共识多向对话式创意生成、方案评估、复杂决策产品设计、投资分析能产生更全面、更稳健的解决方案计算成本高多次LLM调用流程可能冗长分层调度中央指挥分工作业星型管理者与工作者间通信复杂项目规划、动态任务分配智能客服、研发项目管理灵活性强易于扩展和监控对“管理者”Agent的规划能力要求高是单点瓶颈理解了这些模式我们就可以根据具体任务来选择架构。接下来我们将进入实战环节从环境准备开始。3. 环境准备与前置条件我们将使用 Python 作为主要开发语言并借助LangChain和LangGraph这两个目前生态最成熟、文档最丰富的框架来构建我们的多 Agent 系统。LangChain 提供了构建 Agent 的基础模块而 LangGraph 则专门用于描述和运行基于图的、多参与者的复杂工作流完美契合多 Agent 协作的场景。基础环境要求操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04 推荐)。本文示例在 Linux/macOS 的终端环境下测试。Python版本 3.8 或更高。建议使用 3.9 或 3.10 以获得最佳兼容性。包管理工具pip(Python 自带) 或conda(如果你使用 Anaconda)。核心依赖安装打开你的终端或命令提示符创建一个新的虚拟环境强烈推荐以避免包冲突然后安装以下包# 创建并激活虚拟环境 (以 venv 为例) python -m venv multi_agent_env source multi_agent_env/bin/activate # Linux/macOS # multi_agent_env\Scripts\activate # Windows # 升级 pip pip install --upgrade pip # 安装核心框架 pip install langchain langchain-community langgraph # 安装用于 Agent 工具链的常用包 pip install wikipedia # 示例中可能用到的工具 pip install duckduckgo-search # 用于网络搜索LLM 服务配置Agent 的“大脑”需要一个大语言模型。你可以使用 OpenAI 的 API也可以使用开源的本地模型通过 Ollama、vLLM 等部署。为了演示的通用性我们以 OpenAI API 为例。你需要准备一个有效的 OpenAI API Key。# 设置环境变量 (临时方式更安全的方式是使用 .env 文件) export OPENAI_API_KEYyour-api-key-here # Linux/macOS # set OPENAI_API_KEYyour-api-key-here # Windows代码编辑器任何你熟悉的 IDE 或编辑器即可如 VS Code、PyCharm。环境准备好后我们就可以开始设计并实现第一个协作模式了。4. 核心流程拆解以顺序流水线为例让我们从一个最直观的场景开始自动生成一份行业分析简报。这个任务可以清晰地分解为1) 搜索最新行业动态2) 总结核心观点3) 生成一份格式优美的 Markdown 报告。我们将创建三个 Agent 来组成一个流水线研究员 (Researcher)负责搜索和获取原始信息。分析师 (Analyst)负责消化信息提炼出核心要点。编辑 (Editor)负责将要点组织成结构化的报告。整个系统的流程如下图所示概念图用户输入“生成关于电动汽车的简报” | v [研究员 Agent] --(搜索到的原始文本)-- [分析师 Agent] --(提炼出的要点)-- [编辑 Agent] | | | v (调用搜索工具) (生成最终 Markdown 报告)关键步骤拆解定义每个 Agent 的角色与工具明确每个 Agent 的职责和它能使用的“技能”Tools。研究员需要搜索工具分析师和编辑主要依靠 LLM 的推理和生成能力。构建 Agent 实例使用 LangChain 的create_react_agent或其他 Agent 构造函数为每个角色创建独立的 Agent 对象并绑定相应的工具和提示词Prompt。设计工作流图使用 LangGraph 的StateGraph来定义工作流。图中的每个节点Node对应一个 Agent 的执行函数边Edge定义了执行顺序和数据流向。编译并运行图将定义好的图编译成可执行的对象然后传入初始状态用户问题启动整个协作流程。处理与输出结果从最终节点获取处理结果并返回给用户。这个流程是构建任何多 Agent 系统的基础模板。接下来我们用代码把它实现出来。5. 完整示例与代码实现我们将实现上面描述的“行业分析简报流水线”。请确保你已经完成了环境准备并设置了OPENAI_API_KEY。首先创建一个名为sequential_pipeline.py的 Python 文件。5.1 导入依赖与初始化 LLM# sequential_pipeline.py import os from typing import TypedDict, Annotated, List import operator from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import HumanMessage, AIMessage from langgraph.graph import StateGraph, END # 1. 初始化 LLM # 使用 gpt-3.5-turbo 以控制成本可根据需要换成 gpt-4 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 2. 定义搜索工具 search_tool DuckDuckGoSearchRun()5.2 定义系统状态在 LangGraph 中工作流的状态State是一个在所有节点间共享和传递的数据结构。我们定义一个State类来明确需要传递哪些信息。# 定义图的状态结构 class AgentState(TypedDict): 所有 Agent 共享和更新的状态 # 用户最初的问题 original_query: str # 研究员 Agent 搜索到的原始信息 raw_information: str # 分析师 Agent 提炼出的核心要点 key_points: str # 编辑 Agent 生成的最终报告 final_report: str # 用于记录中间步骤的消息历史可选用于复杂交互 messages: Annotated[List, operator.add]5.3 创建各个 Agent我们为三个角色分别创建 Agent。注意我们为“研究员”赋予了搜索工具而“分析师”和“编辑”则没有额外工具专注于文本处理。# 3. 创建研究员 Agent (拥有搜索能力) researcher_prompt ChatPromptTemplate.from_messages([ (system, 你是一位专业的研究员。你的任务是利用搜索工具针对用户的问题尽可能全面地搜集最新、最相关的公开信息。 请直接返回搜索到的原始信息摘要不要进行加工分析。如果一次搜索不够可以多次搜索。), MessagesPlaceholder(variable_namechat_history), (human, {input}), ]) researcher_agent create_react_agent(llm, tools[search_tool], promptresearcher_prompt) researcher_executor AgentExecutor(agentresearcher_agent, tools[search_tool], verboseFalse) # 4. 创建分析师 Agent (无工具专注分析) analyst_prompt ChatPromptTemplate.from_messages([ (system, 你是一位资深行业分析师。你的任务是从研究员提供的大段原始信息中提炼出3-5个最核心的要点。 要点需要简洁、清晰、有洞察力避免直接复制原文。请用分点列表的形式输出。), (human, 请分析以下信息提炼核心要点\n\n{raw_info}), ]) # 分析师不需要工具直接使用 LLM analyst_chain analyst_prompt | llm # 5. 创建编辑 Agent (无工具专注报告撰写) editor_prompt ChatPromptTemplate.from_messages([ (system, 你是一位专业的科技编辑。你的任务是将分析师提炼的要点整理成一份结构完整、语言流畅、适合高管阅读的 Markdown 格式简报。 简报应包含标题、概述、核心要点分点阐述、总结与展望。请确保格式美观。), (human, 请根据以下核心要点撰写一份行业简报\n\n{key_points}), ]) editor_chain editor_prompt | llm5.4 定义图节点Node函数每个节点函数接收当前的State执行特定 Agent 的任务并更新State中的相应字段。# 6. 定义图节点函数 def research_node(state: AgentState) - AgentState: 研究员节点执行搜索更新 raw_information print( 研究员开始工作 ) result researcher_executor.invoke({input: state[original_query], chat_history: []}) # 更新状态 return {raw_information: result[output]} def analyze_node(state: AgentState) - AgentState: 分析师节点分析原始信息更新 key_points print( 分析师开始工作 ) result analyst_chain.invoke({raw_info: state[raw_information]}) return {key_points: result.content} def edit_node(state: AgentState) - AgentState: 编辑节点生成最终报告更新 final_report print( 编辑开始工作 ) result editor_chain.invoke({key_points: state[key_points]}) return {final_report: result.content}5.5 构建并运行工作流图现在我们将节点和边组合起来形成完整的工作流。# 7. 构建工作流图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(researcher, research_node) workflow.add_node(analyst, analyze_node) workflow.add_node(editor, edit_node) # 设置边的连接关系 (顺序流水线) workflow.set_entry_point(researcher) # 从研究员开始 workflow.add_edge(researcher, analyst) workflow.add_edge(analyst, editor) workflow.add_edge(editor, END) # 结束 # 编译图 app workflow.compile() # 8. 运行工作流 if __name__ __main__: # 初始化状态 initial_state: AgentState { original_query: 请生成一份关于2024年人工智能大模型在金融领域应用最新趋势的简报。, raw_information: , key_points: , final_report: , messages: [] } print(开始执行多Agent协作流水线...\n) # 运行图从 entry_point 开始 final_state app.invoke(initial_state) print(\n *50) print(【最终报告生成完毕】) print(*50) print(final_state[final_report])6. 运行结果与效果验证保存sequential_pipeline.py文件并在终端中运行它python sequential_pipeline.py预期输出流程终端会依次打印 研究员开始工作 、 分析师开始工作 、 编辑开始工作 表明三个 Agent 正在按顺序执行。由于DuckDuckGoSearchRun工具会进行真实的网络搜索研究员阶段可能需要几秒到十几秒时间。最终你会看到一份格式完整的 Markdown 报告输出在终端上。报告内容示例片段# 2024年人工智能大模型在金融领域应用趋势简报 ## 概述 2024年生成式人工智能与大模型技术正加速渗透至金融行业的核心业务环节从风险管控、投资研究到客户服务展现出从“辅助工具”向“核心生产力”演进的明确趋势。 ## 核心要点 1. **风险管理的深度智能化**大模型开始用于解读非结构化数据如财报电话会议记录、新闻舆情实现更动态、前瞻性的信用风险和操作风险评估。 2. **投资研究的效率革命**... 3. **合规与监管科技的升级**... 4. **个性化财富管理的普及**... ## 总结与展望 ...如何验证成功流程验证观察控制台输出确认三个节点都按顺序执行完毕没有报错。内容验证检查raw_information字段可以在代码中打印出来是否包含了从网络搜索到的相关文本。检查key_points字段是否是从原始信息中提炼出的分点列表。检查final_report是否是一份结构清晰有标题、概述、要点、总结、语言通顺的 Markdown 文档。质量验证人工阅读最终报告判断其是否基本回答了初始问题信息是否连贯格式是否正确。如果运行失败请首先检查API KeyOPENAI_API_KEY环境变量是否设置正确且有效。网络连接是否能正常访问 OpenAI API 和 DuckDuckGo 搜索。依赖包是否安装了langchain-openai如果使用最新 LangChain可能需要单独安装pip install langchain-openai。错误信息仔细阅读控制台输出的错误堆栈信息通常能定位到具体问题。7. 常见问题与排查思路在多 Agent 系统开发中你会遇到一些典型问题。下表列出了常见现象、原因及解决方法问题现象可能原因排查方式解决方案Agent 执行超时或无响应1. LLM API 调用缓慢或失败。2. 某个工具如搜索耗时过长。3. 工作流陷入循环。1. 查看 API 返回的错误信息。2. 在节点函数内添加日志打印开始/结束时间。3. 检查图的结构是否有循环边未正确终止。1. 增加 API 调用的超时设置。2. 为耗时工具设置独立的超时限制。3. 使用interrupt_before/after在 LangGraph 中设置检查点或引入最大循环次数限制。最终结果质量差1. Agent 的提示词Prompt不清晰。2. 上游 Agent 输出质量差导致垃圾进、垃圾出。3. Agent 之间传递的信息丢失了关键上下文。1. 分别测试每个 Agent 在孤立任务上的表现。2. 打印并检查每个节点输出 (raw_information,key_points)。3. 审查 State 中传递的数据是否完整。1. 迭代优化每个 Agent 的 System Prompt明确其角色和输出格式要求。2. 在上游 Agent 的输出中强制加入结构化标记如##SUMMARY_START## ... ##SUMMARY_END##便于下游解析。3. 在 State 中传递更丰富的上下文或让 Manager Agent 负责补充上下文。成本过高或 Token 消耗巨大1. 工作流中 LLM 调用次数过多。2. 传递的上下文如原始网页内容过长。3. 使用了更昂贵的模型如 GPT-4。1. 统计每个节点的调用次数和输入/输出的 token 数。2. 使用模型的get_num_tokens方法估算。1. 优化流程避免不必要的 LLM 调用例如缓存中间结果。2. 在上游节点对长文本进行摘要或过滤再传递给下游。3. 在非关键节点使用成本更低的模型如 GPT-3.5-Turbo。4. 设置预算和用量监控。“辩论协商”模式陷入僵局多个 Agent 争论不休无法达成一致。记录辩论过程中的所有消息。1. 引入“裁判”Agent其 Prompt 明确要求综合各方意见并做出最终决策。2. 设置辩论轮次上限超时后由裁判强制裁决。3. 定义清晰的投票或评分机制。工具调用失败1. 工具所需的 API Key 未配置。2. 工具输入参数格式错误。3. 外部服务不可用。1. 检查工具初始化时的配置。2. 在 Agent 之外单独测试工具调用。3. 查看工具返回的具体错误信息。1. 确保所有必要的环境变量已设置。2. 在 Agent 的 Prompt 中更详细地描述工具的使用方法和参数格式。3. 为工具调用添加重试和降级逻辑如返回模拟数据。8. 最佳实践与工程建议构建可用于生产环境的多 Agent 系统远不止让流程跑通那么简单。以下是一些关键的工程实践1. 提示词工程是核心角色扮演要彻底每个 Agent 的 System Prompt 必须清晰定义其角色、职责、输出格式和边界。例如“你是一位严谨的风险评估师只关注技术可行性和潜在漏洞不要提供商业建议。”上下文管理明确告诉 Agent 哪些信息是来自上游的输入哪些是它需要产出的。使用明确的占位符和格式要求。迭代优化将 Prompt 作为代码一样管理进行版本控制和 A/B 测试。2. 状态设计要精简而充分只传递必要信息State 中不要包含所有中间过程只存放下游节点必需的数据。过大的 State 会增加 Token 消耗和传输开销。使用结构化数据尽量使用字典、列表等结构化数据类型在 State 中传递信息避免冗长的自然语言段落除非下游 LLM 必须使用它。考虑记忆外挂对于需要长期记忆或知识库的场景不要依赖 State而是为 Agent 配备向量数据库等外部记忆体。3. 引入健壮性与可观测性错误处理与回退在每个节点函数中使用try...except包裹核心逻辑。当某个 Agent 失败时应有备选方案如使用简化流程、返回默认值、通知人类。日志与监控记录每个 Agent 的输入、输出、耗时和 Token 使用情况。这有助于性能优化、成本分析和问题排查。可以考虑使用 LangSmith 等专门的可观测性平台。设置超时与循环限制在 LangGraph 中对于可能循环的路径如辩论模式必须使用add_conditional_edges并设置最大迭代次数防止无限循环消耗资源。4. 成本与性能优化模型分级使用在流水线中对创造性要求不高的环节如信息提取、格式转换使用廉价快速的模型如 GPT-3.5-Turbo只在核心创意或决策环节使用强大但昂贵的模型如 GPT-4。缓存与去重如果多个 Agent 可能查询相同的外部信息如股票价格引入缓存层避免重复调用和费用。异步执行当 Agent 之间没有严格的先后依赖时可以利用 LangGraph 的异步支持或asyncio并发执行缩短整体流程时间。5. 安全与边界控制工具权限隔离不是所有 Agent 都需要所有工具。为每个 Agent 精确分配最小必要权限的工具集。例如只有“数据清洗”Agent 能访问数据库写操作。输入输出审查在系统的入口和出口设置审查环节对用户输入进行过滤对 Agent 的最终输出进行内容安全校验防止生成有害或不适当的内容。人工审核环节对于关键业务或高风险决策将多 Agent 系统的输出作为“建议”最终决策需加入人工审核确认步骤。从“顺序流水线”到“辩论协商”和“分层调度”架构会变得更复杂但核心思想不变定义清晰的 Agent 角色设计高效可靠的通信机制并用工作流图将它们有机地组织起来。掌握了流水线模式你就已经拿到了打开多 Agent 协作世界大门的钥匙。接下来你可以尝试让“研究员”和“分析师”就某个数据点进行“辩论”或者引入一个“项目经理”Agent 来动态分配任务探索更强大的协作模式。