1. 项目概述从单兵作战到团队协作的跃迁在AI应用开发的浪潮里LangChain已经从一个热门框架变成了许多开发者构建智能应用的首选工具箱。我们之前聊了链、聊了记忆、聊了如何让大模型调用工具这些都是构建一个“聪明”的单个AI智能体Agent的基石。但不知道你有没有想过当任务复杂到像开发一个完整软件项目、分析一份跨领域市场报告或者处理一个从客服到售后全流程的客户请求时单个Agent再“聪明”也难免会力不从心。它就像一个全能的超人既要会写代码又要懂业务逻辑还得能画UI设计图最后可能每样都懂点但每样都不够精。这就是“多Agent系统”登场的时刻。它不再是打造一个超人而是组建一支各有所长的复仇者联盟。第九章要探讨的正是如何用LangChain来设计和实现这样一支AI团队。简单来说多Agent系统就是让多个具备特定专长和角色的AI智能体协同工作通过彼此间的通信、任务分配与结果整合共同完成一个复杂的、超出单个智能体能力范围的宏大目标。这不仅仅是技术的堆砌更是一种系统设计思维的转变。从单点智能走向群体智能是当前AI应用走向深度和实用化的一个关键分水岭。那么谁需要关注这个呢如果你正在尝试用AI解决业务流程自动化、复杂决策支持、多步骤内容生成比如从大纲到初稿再到润色的全流程写作或者构建一个数字员工团队那么多Agent系统就是你绕不开的课题。它能让你的应用从“玩具级”的对话机器人蜕变为真正能扛事的“生产力引擎”。2. 核心设计思路如何构建一个高效协作的AI团队设计一个多Agent系统远比串联几个工具调用要复杂。它本质上是在设计一个组织的协作流程。你不能简单地把几个Agent扔在一起指望它们自己就能默契配合。这里面的核心设计思路我把它总结为“角色定义、通信协议、控制流”三位一体。2.1 角色定义与能力边界划分这是第一步也是最关键的一步。每个Agent必须有清晰、唯一的职责。模糊的角色定位是团队内耗和效率低下的根源。在LangChain的语境下一个Agent的角色通常由其三个核心要素决定系统提示词System Prompt这是Agent的“岗位说明书”。它明确告诉Agent“你是谁”、“你的职责是什么”、“你的行事风格如何”。例如一个“代码审查Agent”的提示词会强调其专注于发现代码中的安全漏洞、性能问题和风格不一致而不是去重写业务逻辑。工具集Tools这是Agent的“专业技能工具箱”。一个数据分析Agent可能需要pandas、matplotlib等工具一个文档检索Agent则需要向量数据库查询工具。严格限制每个Agent的工具访问权限是实现安全性和专业性的基础。一个拥有所有工具权限的Agent不仅容易“越权”行事也增加了系统的不可控风险。底层大模型LLM这是Agent的“大脑”。虽然理论上所有Agent可以共用一个大模型但根据角色选择不同特性的模型往往有奇效。比如让一个需要严谨逻辑的“架构师Agent”使用GPT-4而让一个负责创意发散的“头脑风暴Agent”使用Claude可能比都用同一个模型效果更好。实操心得在定义角色时我强烈建议采用“名词动词”的格式如“数据分析师”、“API协调员”、“文案润色员”。这能让你和你的代码都时刻明确每个Agent的使命。避免使用“智能助手”、“处理器”这类泛泛的名称。2.2 通信协议Agent之间如何“说话”Agent不能活在各自的孤岛上它们需要交换信息、传递任务、汇报结果。LangChain提供了几种主流的通信“语言”或模式共享全局状态Shared State这是最简单直接的方式类似于一个共享的公告板或工作区。所有Agent都能读取和写入一个公共的字典或对象。例如一个“信息收集Agent”把找到的资料存入共享状态的research_materials字段后续的“报告撰写Agent”直接从这里读取。这种方式实现简单但缺乏结构化管理容易产生数据污染和冲突。消息传递Message Passing这是更模块化、更接近分布式系统的设计。每个Agent有明确的输入/输出接口。Agent A完成任务后生成一个结构化的消息通常包含sender,recipient,content等字段发送给Agent B。LangChain的AgentExecutor本身就可以被视作一个消息处理单元。这种方式职责清晰易于调试和扩展是构建复杂系统的推荐选择。基于发布/订阅Pub/Sub在更动态的系统中你可能不希望Agent之间是固定的点对点通信。发布/订阅模式允许Agent向某个“频道”发布消息而关心这类消息的其他Agent可以订阅该频道。这非常适合事件驱动的场景比如“当用户需求变更时通知所有相关Agent”。在我的项目中对于任务流相对固定的场景如固定的报告生成流程我倾向于使用显式的消息传递因为它逻辑清晰而对于需要动态响应事件的场景如一个实时协作的白板则会考虑引入发布/订阅模式。2.3 控制流谁来决定下一步做什么多个Agent动起来了但整个团队的指挥棒在谁手里这就是控制流问题。LangChain生态中主要有两种范式中心化编排Orchestration存在一个“管理者”或“协调者”Agent有时也可以是一个简单的程序逻辑。它负责解析总任务将其分解为子任务然后像项目经理一样按顺序或并行地指派给各个专业Agent并汇总最终结果。LangGraph这个库就是专门为这种中心化、图结构的流程编排而生的。你可以用它将Agent定义为图的节点用边来定义控制流逻辑顺序、分支、循环。去中心化协作Collaboration没有绝对的中央指挥。Agent之间通过通信协议自主协商和协作。例如Agent A发现自己无法解决某个子问题可以主动广播一个“求助”消息由具备相关能力的Agent B接手。这种方式更灵活、健壮但设计起来更复杂容易陷入“扯皮”或死循环。对于绝大多数应用级项目中心化编排是更务实、更可控的选择。LangGraph提供了非常直观的方式来绘制你的AI团队工作流图你可以清晰地看到任务从“需求输入”开始经过“研究员”、“分析师”、“撰稿人”等节点最终产出“报告”的完整路径。3. 核心组件与架构拆解理解了设计思路我们来看看用LangChain搭建多Agent系统时手上有哪些“积木块”。这里我结合LangGraph来讲解一个典型架构因为它代表了当前最主流和强大的实现方式。3.1 节点Node每个Agent的独立工作间在LangGraph中每个Agent或任何执行单元都被建模为一个节点。一个节点本质上是一个函数它接收系统的当前状态执行一些操作比如调用一个Agent然后更新状态。创建节点非常简单from langgraph.graph import StateGraph, END from your_agent_module import research_agent, analysis_agent, writing_agent # 定义节点函数 def research_node(state): # state 包含了当前所有共享信息如用户问题、中间结果等 result research_agent.invoke({question: state[user_query]}) # 更新状态将研究成果放入共享空间 state[research_data] result return state def analysis_node(state): # 从状态中读取上游节点的产出 data_to_analyze state[research_data] analysis_result analysis_agent.invoke({data: data_to_analyze}) state[analysis_insights] analysis_result return state这里的关键是状态对象。它像一个共享的文件夹流经每个节点节点从中读取输入并将输出写回传递给下一个节点。3.2 边Edge定义工作流的交通规则节点定义了“谁来做”边则定义了“做完之后去哪”。这是控制流的核心。LangGraph提供了几种基础的边条件边Conditional Edge根据当前状态的内容决定下一步走向。这实现了if-else分支逻辑。例如在“分析”节点之后如果分析结果显示数据质量高就流向“撰写报告”节点如果数据质量差则流向“重新收集数据”节点。def should_continue(state): if state[analysis_insights][data_quality] high: return write_report else: return recollect_data固定边最简单的单向连接一个节点完成后无条件进入下一个节点。通过组合节点和边你就能画出一张完整的AI团队工作流程图。3.3 状态State团队共享的工作内存状态是一个贯穿始终的字典或Pydantic模型它是所有Agent之间通信的载体。设计一个好的状态结构至关重要。我建议明确字段为每一类中间产物和最终结果设计清晰的字段名如raw_research,cleaned_data,key_findings,final_report。版本管理对于可能被多个节点修改的字段考虑使用列表来记录历史版本方便回溯和调试。例如analysis_history: List[Dict]。轻量存储避免在状态中存储过大的二进制对象如图片、长视频只存放引用或元数据。真正的重型数据应放在外部存储如数据库、对象存储中。一个定义良好的状态模型能让你的工作流像管道一样清晰数据在其中有序流动、加工。3.4 编排器Graph把一切组装起来最后你需要一个StateGraph实例作为编排器将节点、边和状态模型组装成一个可执行的工作流。from langgraph.graph import StateGraph from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator # 1. 定义状态模型 class AgentState(TypedDict): user_query: str research_data: str analysis_insights: dict final_report: str messages: Annotated[List, add_messages] # 用于消息传递的专用字段 # 2. 创建图 workflow StateGraph(AgentState) # 3. 添加节点 workflow.add_node(researcher, research_node) workflow.add_node(analyst, analysis_node) workflow.add_node(writer, writing_node) # 4. 添加边定义流程 workflow.set_entry_point(researcher) # 入口节点 workflow.add_edge(researcher, analyst) # 研究员完成后交给分析师 workflow.add_edge(analyst, writer) # 分析师完成后交给撰写者 workflow.add_edge(writer, END) # 撰写者完成后工作流结束 # 5. 编译图获得可执行对象 app workflow.compile()现在你只需要向这个app传入初始状态比如{user_query: 分析一下新能源汽车市场的未来趋势}它就会自动按照你定义的流程驱动整个AI团队运转起来。4. 实战构建一个智能报告生成团队理论说得再多不如动手搭一个。我们来构建一个相对完整的“行业分析报告自动生成”多Agent系统。这个系统将包含三个Agent信息检索员、数据分析师、报告撰写员。4.1 第一步定义Agent成员及其工具首先我们为每位成员配置独特的技能。from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun from langchain_openai import ChatOpenAI import pandas as pd import numpy as np # 工具1网络搜索工具给检索员用 search_tool DuckDuckGoSearchRun() # 工具2数据分析工具给分析师用这里用模拟函数代替 def analyze_market_trends(data_description: str) - str: 模拟一个数据分析函数输入描述输出洞察。 # 在实际项目中这里可能是调用pandas进行真实计算 return f基于数据{data_description}的分析预计未来三年复合增长率达15%头部企业市场份额集中度提升。 analysis_tool Tool( nameMarketAnalyzer, funcanalyze_market_trends, description用于分析市场数据和趋势输入一段数据描述返回分析洞察。 ) # 创建大模型实例可以为不同Agent选择不同模型 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) # 构建信息检索员Agent research_agent_prompt 你是一个专业的信息检索员。你的任务是根据用户的问题利用搜索工具查找最新、最相关的公开资料和新闻。 只提供事实性信息的摘要不要进行分析或总结。将找到的信息清晰罗列。 research_agent create_react_agent(llm, tools[search_tool], promptresearch_agent_prompt) research_executor AgentExecutor(agentresearch_agent, tools[search_tool], verboseTrue) # 构建数据分析师Agent analysis_agent_prompt 你是一个资深数据分析师。你会收到检索员提供的原始信息。 你的工作是批判性地审视这些信息识别其中的核心数据点、矛盾之处和潜在趋势并调用分析工具生成结构化洞察。 输出应包含关键指标、趋势判断和风险提示。 analysis_agent create_react_agent(llm, tools[analysis_tool], promptanalysis_agent_prompt) analysis_executor AgentExecutor(agentanalysis_agent, tools[analysis_tool], verboseTrue) # 构建报告撰写员Agent (不需要特殊工具纯靠LLM) writing_agent_prompt 你是一位优秀的商业报告撰写人。你将收到用户的问题、检索到的信息以及数据分析师的洞察。 你的任务是整合所有材料撰写一份结构完整、语言精练、论据充分的商业分析报告大纲。 报告需包含摘要、背景、核心发现、趋势分析、建议与展望等部分。 # 这是一个简单的Chain不是Tool-using Agent from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser write_prompt ChatPromptTemplate.from_messages([ (system, writing_agent_prompt), (human, 用户问题{query}\n检索信息{research}\n数据分析{analysis}) ]) writing_chain write_prompt | llm | StrOutputParser()4.2 第二步用LangGraph组装工作流现在我们将这三个成员组装到一个有秩序的工作流中。from langgraph.graph import StateGraph, END from typing import TypedDict # 定义工作流的状态结构 class ReportState(TypedDict): query: str # 用户原始问题 research_result: str # 检索员的产出 analysis_result: str # 分析师的产出 final_report: str # 撰写员的最终产出 # 1. 定义各个节点函数 def research_node(state: ReportState) - ReportState: 信息检索节点 print(f[Research Agent] 开始检索: {state[query]}) result research_executor.invoke({input: state[query]}) state[research_result] result[output] return state def analysis_node(state: ReportState) - ReportState: 数据分析节点 print(f[Analysis Agent] 开始分析检索结果...) result analysis_executor.invoke({input: f请分析以下信息{state[research_result]}}) state[analysis_result] result[output] return state def writing_node(state: ReportState) - ReportState: 报告撰写节点 print(f[Writing Agent] 开始整合撰写报告...) result writing_chain.invoke({ query: state[query], research: state[research_result], analysis: state[analysis_result] }) state[final_report] result return state # 2. 创建并组装图 workflow StateGraph(ReportState) # 添加节点 workflow.add_node(researcher, research_node) workflow.add_node(analyst, analysis_node) workflow.add_node(writer, writing_node) # 设置流程检索 - 分析 - 撰写 - 结束 workflow.set_entry_point(researcher) workflow.add_edge(researcher, analyst) workflow.add_edge(analyst, writer) workflow.add_edge(writer, END) # 编译图 app workflow.compile()4.3 第三步运行与迭代优化现在让我们运行这个AI团队。# 初始化状态输入任务 initial_state {query: 请分析2024年人工智能在医疗诊断领域的主要应用、市场格局及未来两年的发展趋势。} # 运行工作流 final_state app.invoke(initial_state) print(\n *50) print(最终生成的报告大纲) print(*50) print(final_state[final_report])运行后你会在控制台看到每个Agent被依次激活、执行任务。最终final_state中的final_report字段就包含了整合后的报告大纲。注意事项第一次运行可能不会完美。多Agent系统的调试是一个迭代过程。你需要关注信息衰减检查research_result和analysis_result的质量是否在传递过程中丢失了关键信息可能需要优化Agent的提示词要求其输出更结构化的内容。流程僵化当前的线性流程检索-分析-撰写是否适合所有问题也许有些简单问题不需要分析可以直接撰写。这时就需要引入条件边。错误处理如果检索Agent什么都没找到怎么办如果分析Agent工具调用出错怎么办一个健壮的系统需要在关键节点添加错误处理和备用路径如重试、降级处理。5. 高级模式与设计模式探讨当你掌握了基础的多Agent编排后可以尝试一些更高级的模式来解决复杂问题。5.1 管理者-工作者模式这是最经典的中心化模式。一个“管理者Agent”负责理解和分解任务然后根据子任务的性质动态地调用不同的“工作者Agent”专家来完成。管理者还负责整合各工作者的结果。这类似于一个公司的部门经理。在LangGraph中你可以用一个“管理者”节点来接收任务其内部逻辑是调用一个LLM来判断该任务属于哪一类然后通过条件边将状态路由到不同的工作者节点如“编码工作者”、“写作工作者”、“查询工作者”。工作者完成后再将结果返回给管理者节点进行汇总。5.2 辩论与共识模式当面对没有标准答案的开放性问题时如“设计一个产品logo”可以让多个同类型的Agent如几个不同的“创意设计师Agent”独立工作产生多个方案然后引入一个“评审员Agent”来评估这些方案或者让Agent们进行多轮“辩论”最终达成一个共识或选出最佳方案。这种模式能有效提升输出的多样性和质量。实现上这需要循环。你可以设置一个“创意生成”节点并行或串行调用多个Agent将结果收集到一个列表中。然后进入“评审/辩论”节点该节点可能会多次调用评审Agent更新方案评分或修改意见直到满足某个终止条件如达到最大轮次或共识度阈值。5.3 动态子任务创建对于一些极其复杂、无法预先规划的任务如“为我策划一次跨国旅行”可能需要系统在运行过程中动态创建新的子任务。这要求“管理者Agent”具备更强的规划能力。一种实现方式是让管理者Agent的输出不仅包含结果还包含一个“后续任务列表”。工作流在执行完当前节点后会检查这个列表。如果不为空则动态地将新任务作为新的节点或新的工作流实例加入执行队列。LangGraph的State可以设计一个pending_tasks: List字段来支持这种动态性。6. 避坑指南与性能调优搭建多Agent系统会踩很多坑这里分享几个我亲身经历过的教训和优化技巧。6.1 常见问题与排查Agent陷入循环或卡住症状工作流长时间不结束某个Agent反复执行相同操作。排查首先检查Agent的提示词是否包含了明确的停止条件如“当你获得足够信息时请用Final Answer:开头输出”。其次在LangGraph中为循环结构如辩论模式必须设置最大迭代次数可以使用add_conditional_edges并设置一个计数器在状态中达到上限后强制跳出。工具设计确保工具函数的返回值是确定且格式良好的。一个返回None或抛出异常的工具很容易导致Agent困惑并反复重试。信息在传递中失真或丢失症状下游Agent抱怨拿到的是“空数据”或“错误数据”。排查这是状态设计不佳的典型表现。不要传递冗长的、非结构化的文本。强制上游Agent输出结构化数据如JSON。例如要求检索Agent输出{key_facts: [...], source_links: [...]}而不是一大段话。下游Agent的提示词也要明确指定从状态的哪个字段、以何种格式读取输入。系统延迟高、成本昂贵症状运行一个流程耗时过长API调用费用激增。优化并行化如果多个子任务间没有依赖关系一定要用LangGraph的并行节点能力让其同时执行。缓存对LLM调用和工具调用实施缓存。对于相同输入直接返回历史结果。LangChain提供了InMemoryCache、SQLiteCache等组件。模型分级并非所有环节都需要GPT-4。让“管理者”或核心分析环节用强模型让“格式化”、“简单分类”等任务使用gpt-3.5-turbo甚至更小的开源模型能大幅降低成本。精简上下文避免将整个对话历史或所有中间结果都塞进每个Agent的上下文。只传递必要的信息。6.2 状态管理与可观测性当系统变得复杂调试会变得困难。你必须建立良好的可观测性。日志记录在每个节点的函数中详细记录输入、输出和关键步骤。可以使用Python的logging模块并将日志级别与状态关联。状态快照在LangGraph中app.invoke()返回的最终状态只包含最后结果。为了调试你可以在编译图时设置debugTrue或者手动在状态中增加一个execution_log: List字段每个节点执行后都把关键信息append进去。可视化LangGraph一个很棒的功能是能将你的图可视化出来。使用app.get_graph().draw_mermaid_png()可以生成流程图直观地看到你的团队结构这对于向他人解释系统设计非常有帮助。6.3 安全与稳定性考量工具权限隔离这是最重要的安全原则。绝不要给所有Agent访问所有工具的权限。一个负责外部搜索的Agent不应该有访问数据库或执行系统命令的权限。在创建AgentExecutor时严格限定其tools列表。输入输出验证对流入每个Agent的输入和其产生的输出进行基本的验证和清理防止Prompt注入攻击或异常数据导致系统崩溃。可以使用Pydantic模型对状态进行强类型校验。设置超时与回退为每个Agent执行器或工具调用设置超时时间。当某个环节长时间无响应时应有超时机制和预定义的回退方案如返回一个默认值或错误信息避免整个工作流被卡死。人性化接管在关键决策点如最终报告生成前、执行具有外部影响的操作前设计“人工审核”节点。工作流可以暂停将当前状态和待决策信息发送给人工审核待确认后再继续执行。这在生产环境中至关重要。从单个Agent到多Agent系统你构建的不再是一个工具而是一个数字组织。这其中的挑战从单纯的提示工程、工具调用上升到了系统架构、流程设计、协同逻辑的层面。它要求开发者兼具AI应用开发和软件工程的双重思维。但回报也是巨大的一个设计精良的多Agent系统能够解决那些过去被认为必须由人类团队协作才能完成的复杂任务真正释放出AI的群体智能潜力。 LangChain多智能体系统：从架构设计到实战构建AI协作团队
1. 项目概述从单兵作战到团队协作的跃迁在AI应用开发的浪潮里LangChain已经从一个热门框架变成了许多开发者构建智能应用的首选工具箱。我们之前聊了链、聊了记忆、聊了如何让大模型调用工具这些都是构建一个“聪明”的单个AI智能体Agent的基石。但不知道你有没有想过当任务复杂到像开发一个完整软件项目、分析一份跨领域市场报告或者处理一个从客服到售后全流程的客户请求时单个Agent再“聪明”也难免会力不从心。它就像一个全能的超人既要会写代码又要懂业务逻辑还得能画UI设计图最后可能每样都懂点但每样都不够精。这就是“多Agent系统”登场的时刻。它不再是打造一个超人而是组建一支各有所长的复仇者联盟。第九章要探讨的正是如何用LangChain来设计和实现这样一支AI团队。简单来说多Agent系统就是让多个具备特定专长和角色的AI智能体协同工作通过彼此间的通信、任务分配与结果整合共同完成一个复杂的、超出单个智能体能力范围的宏大目标。这不仅仅是技术的堆砌更是一种系统设计思维的转变。从单点智能走向群体智能是当前AI应用走向深度和实用化的一个关键分水岭。那么谁需要关注这个呢如果你正在尝试用AI解决业务流程自动化、复杂决策支持、多步骤内容生成比如从大纲到初稿再到润色的全流程写作或者构建一个数字员工团队那么多Agent系统就是你绕不开的课题。它能让你的应用从“玩具级”的对话机器人蜕变为真正能扛事的“生产力引擎”。2. 核心设计思路如何构建一个高效协作的AI团队设计一个多Agent系统远比串联几个工具调用要复杂。它本质上是在设计一个组织的协作流程。你不能简单地把几个Agent扔在一起指望它们自己就能默契配合。这里面的核心设计思路我把它总结为“角色定义、通信协议、控制流”三位一体。2.1 角色定义与能力边界划分这是第一步也是最关键的一步。每个Agent必须有清晰、唯一的职责。模糊的角色定位是团队内耗和效率低下的根源。在LangChain的语境下一个Agent的角色通常由其三个核心要素决定系统提示词System Prompt这是Agent的“岗位说明书”。它明确告诉Agent“你是谁”、“你的职责是什么”、“你的行事风格如何”。例如一个“代码审查Agent”的提示词会强调其专注于发现代码中的安全漏洞、性能问题和风格不一致而不是去重写业务逻辑。工具集Tools这是Agent的“专业技能工具箱”。一个数据分析Agent可能需要pandas、matplotlib等工具一个文档检索Agent则需要向量数据库查询工具。严格限制每个Agent的工具访问权限是实现安全性和专业性的基础。一个拥有所有工具权限的Agent不仅容易“越权”行事也增加了系统的不可控风险。底层大模型LLM这是Agent的“大脑”。虽然理论上所有Agent可以共用一个大模型但根据角色选择不同特性的模型往往有奇效。比如让一个需要严谨逻辑的“架构师Agent”使用GPT-4而让一个负责创意发散的“头脑风暴Agent”使用Claude可能比都用同一个模型效果更好。实操心得在定义角色时我强烈建议采用“名词动词”的格式如“数据分析师”、“API协调员”、“文案润色员”。这能让你和你的代码都时刻明确每个Agent的使命。避免使用“智能助手”、“处理器”这类泛泛的名称。2.2 通信协议Agent之间如何“说话”Agent不能活在各自的孤岛上它们需要交换信息、传递任务、汇报结果。LangChain提供了几种主流的通信“语言”或模式共享全局状态Shared State这是最简单直接的方式类似于一个共享的公告板或工作区。所有Agent都能读取和写入一个公共的字典或对象。例如一个“信息收集Agent”把找到的资料存入共享状态的research_materials字段后续的“报告撰写Agent”直接从这里读取。这种方式实现简单但缺乏结构化管理容易产生数据污染和冲突。消息传递Message Passing这是更模块化、更接近分布式系统的设计。每个Agent有明确的输入/输出接口。Agent A完成任务后生成一个结构化的消息通常包含sender,recipient,content等字段发送给Agent B。LangChain的AgentExecutor本身就可以被视作一个消息处理单元。这种方式职责清晰易于调试和扩展是构建复杂系统的推荐选择。基于发布/订阅Pub/Sub在更动态的系统中你可能不希望Agent之间是固定的点对点通信。发布/订阅模式允许Agent向某个“频道”发布消息而关心这类消息的其他Agent可以订阅该频道。这非常适合事件驱动的场景比如“当用户需求变更时通知所有相关Agent”。在我的项目中对于任务流相对固定的场景如固定的报告生成流程我倾向于使用显式的消息传递因为它逻辑清晰而对于需要动态响应事件的场景如一个实时协作的白板则会考虑引入发布/订阅模式。2.3 控制流谁来决定下一步做什么多个Agent动起来了但整个团队的指挥棒在谁手里这就是控制流问题。LangChain生态中主要有两种范式中心化编排Orchestration存在一个“管理者”或“协调者”Agent有时也可以是一个简单的程序逻辑。它负责解析总任务将其分解为子任务然后像项目经理一样按顺序或并行地指派给各个专业Agent并汇总最终结果。LangGraph这个库就是专门为这种中心化、图结构的流程编排而生的。你可以用它将Agent定义为图的节点用边来定义控制流逻辑顺序、分支、循环。去中心化协作Collaboration没有绝对的中央指挥。Agent之间通过通信协议自主协商和协作。例如Agent A发现自己无法解决某个子问题可以主动广播一个“求助”消息由具备相关能力的Agent B接手。这种方式更灵活、健壮但设计起来更复杂容易陷入“扯皮”或死循环。对于绝大多数应用级项目中心化编排是更务实、更可控的选择。LangGraph提供了非常直观的方式来绘制你的AI团队工作流图你可以清晰地看到任务从“需求输入”开始经过“研究员”、“分析师”、“撰稿人”等节点最终产出“报告”的完整路径。3. 核心组件与架构拆解理解了设计思路我们来看看用LangChain搭建多Agent系统时手上有哪些“积木块”。这里我结合LangGraph来讲解一个典型架构因为它代表了当前最主流和强大的实现方式。3.1 节点Node每个Agent的独立工作间在LangGraph中每个Agent或任何执行单元都被建模为一个节点。一个节点本质上是一个函数它接收系统的当前状态执行一些操作比如调用一个Agent然后更新状态。创建节点非常简单from langgraph.graph import StateGraph, END from your_agent_module import research_agent, analysis_agent, writing_agent # 定义节点函数 def research_node(state): # state 包含了当前所有共享信息如用户问题、中间结果等 result research_agent.invoke({question: state[user_query]}) # 更新状态将研究成果放入共享空间 state[research_data] result return state def analysis_node(state): # 从状态中读取上游节点的产出 data_to_analyze state[research_data] analysis_result analysis_agent.invoke({data: data_to_analyze}) state[analysis_insights] analysis_result return state这里的关键是状态对象。它像一个共享的文件夹流经每个节点节点从中读取输入并将输出写回传递给下一个节点。3.2 边Edge定义工作流的交通规则节点定义了“谁来做”边则定义了“做完之后去哪”。这是控制流的核心。LangGraph提供了几种基础的边条件边Conditional Edge根据当前状态的内容决定下一步走向。这实现了if-else分支逻辑。例如在“分析”节点之后如果分析结果显示数据质量高就流向“撰写报告”节点如果数据质量差则流向“重新收集数据”节点。def should_continue(state): if state[analysis_insights][data_quality] high: return write_report else: return recollect_data固定边最简单的单向连接一个节点完成后无条件进入下一个节点。通过组合节点和边你就能画出一张完整的AI团队工作流程图。3.3 状态State团队共享的工作内存状态是一个贯穿始终的字典或Pydantic模型它是所有Agent之间通信的载体。设计一个好的状态结构至关重要。我建议明确字段为每一类中间产物和最终结果设计清晰的字段名如raw_research,cleaned_data,key_findings,final_report。版本管理对于可能被多个节点修改的字段考虑使用列表来记录历史版本方便回溯和调试。例如analysis_history: List[Dict]。轻量存储避免在状态中存储过大的二进制对象如图片、长视频只存放引用或元数据。真正的重型数据应放在外部存储如数据库、对象存储中。一个定义良好的状态模型能让你的工作流像管道一样清晰数据在其中有序流动、加工。3.4 编排器Graph把一切组装起来最后你需要一个StateGraph实例作为编排器将节点、边和状态模型组装成一个可执行的工作流。from langgraph.graph import StateGraph from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator # 1. 定义状态模型 class AgentState(TypedDict): user_query: str research_data: str analysis_insights: dict final_report: str messages: Annotated[List, add_messages] # 用于消息传递的专用字段 # 2. 创建图 workflow StateGraph(AgentState) # 3. 添加节点 workflow.add_node(researcher, research_node) workflow.add_node(analyst, analysis_node) workflow.add_node(writer, writing_node) # 4. 添加边定义流程 workflow.set_entry_point(researcher) # 入口节点 workflow.add_edge(researcher, analyst) # 研究员完成后交给分析师 workflow.add_edge(analyst, writer) # 分析师完成后交给撰写者 workflow.add_edge(writer, END) # 撰写者完成后工作流结束 # 5. 编译图获得可执行对象 app workflow.compile()现在你只需要向这个app传入初始状态比如{user_query: 分析一下新能源汽车市场的未来趋势}它就会自动按照你定义的流程驱动整个AI团队运转起来。4. 实战构建一个智能报告生成团队理论说得再多不如动手搭一个。我们来构建一个相对完整的“行业分析报告自动生成”多Agent系统。这个系统将包含三个Agent信息检索员、数据分析师、报告撰写员。4.1 第一步定义Agent成员及其工具首先我们为每位成员配置独特的技能。from langchain.agents import create_react_agent, AgentExecutor from langchain.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun from langchain_openai import ChatOpenAI import pandas as pd import numpy as np # 工具1网络搜索工具给检索员用 search_tool DuckDuckGoSearchRun() # 工具2数据分析工具给分析师用这里用模拟函数代替 def analyze_market_trends(data_description: str) - str: 模拟一个数据分析函数输入描述输出洞察。 # 在实际项目中这里可能是调用pandas进行真实计算 return f基于数据{data_description}的分析预计未来三年复合增长率达15%头部企业市场份额集中度提升。 analysis_tool Tool( nameMarketAnalyzer, funcanalyze_market_trends, description用于分析市场数据和趋势输入一段数据描述返回分析洞察。 ) # 创建大模型实例可以为不同Agent选择不同模型 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) # 构建信息检索员Agent research_agent_prompt 你是一个专业的信息检索员。你的任务是根据用户的问题利用搜索工具查找最新、最相关的公开资料和新闻。 只提供事实性信息的摘要不要进行分析或总结。将找到的信息清晰罗列。 research_agent create_react_agent(llm, tools[search_tool], promptresearch_agent_prompt) research_executor AgentExecutor(agentresearch_agent, tools[search_tool], verboseTrue) # 构建数据分析师Agent analysis_agent_prompt 你是一个资深数据分析师。你会收到检索员提供的原始信息。 你的工作是批判性地审视这些信息识别其中的核心数据点、矛盾之处和潜在趋势并调用分析工具生成结构化洞察。 输出应包含关键指标、趋势判断和风险提示。 analysis_agent create_react_agent(llm, tools[analysis_tool], promptanalysis_agent_prompt) analysis_executor AgentExecutor(agentanalysis_agent, tools[analysis_tool], verboseTrue) # 构建报告撰写员Agent (不需要特殊工具纯靠LLM) writing_agent_prompt 你是一位优秀的商业报告撰写人。你将收到用户的问题、检索到的信息以及数据分析师的洞察。 你的任务是整合所有材料撰写一份结构完整、语言精练、论据充分的商业分析报告大纲。 报告需包含摘要、背景、核心发现、趋势分析、建议与展望等部分。 # 这是一个简单的Chain不是Tool-using Agent from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser write_prompt ChatPromptTemplate.from_messages([ (system, writing_agent_prompt), (human, 用户问题{query}\n检索信息{research}\n数据分析{analysis}) ]) writing_chain write_prompt | llm | StrOutputParser()4.2 第二步用LangGraph组装工作流现在我们将这三个成员组装到一个有秩序的工作流中。from langgraph.graph import StateGraph, END from typing import TypedDict # 定义工作流的状态结构 class ReportState(TypedDict): query: str # 用户原始问题 research_result: str # 检索员的产出 analysis_result: str # 分析师的产出 final_report: str # 撰写员的最终产出 # 1. 定义各个节点函数 def research_node(state: ReportState) - ReportState: 信息检索节点 print(f[Research Agent] 开始检索: {state[query]}) result research_executor.invoke({input: state[query]}) state[research_result] result[output] return state def analysis_node(state: ReportState) - ReportState: 数据分析节点 print(f[Analysis Agent] 开始分析检索结果...) result analysis_executor.invoke({input: f请分析以下信息{state[research_result]}}) state[analysis_result] result[output] return state def writing_node(state: ReportState) - ReportState: 报告撰写节点 print(f[Writing Agent] 开始整合撰写报告...) result writing_chain.invoke({ query: state[query], research: state[research_result], analysis: state[analysis_result] }) state[final_report] result return state # 2. 创建并组装图 workflow StateGraph(ReportState) # 添加节点 workflow.add_node(researcher, research_node) workflow.add_node(analyst, analysis_node) workflow.add_node(writer, writing_node) # 设置流程检索 - 分析 - 撰写 - 结束 workflow.set_entry_point(researcher) workflow.add_edge(researcher, analyst) workflow.add_edge(analyst, writer) workflow.add_edge(writer, END) # 编译图 app workflow.compile()4.3 第三步运行与迭代优化现在让我们运行这个AI团队。# 初始化状态输入任务 initial_state {query: 请分析2024年人工智能在医疗诊断领域的主要应用、市场格局及未来两年的发展趋势。} # 运行工作流 final_state app.invoke(initial_state) print(\n *50) print(最终生成的报告大纲) print(*50) print(final_state[final_report])运行后你会在控制台看到每个Agent被依次激活、执行任务。最终final_state中的final_report字段就包含了整合后的报告大纲。注意事项第一次运行可能不会完美。多Agent系统的调试是一个迭代过程。你需要关注信息衰减检查research_result和analysis_result的质量是否在传递过程中丢失了关键信息可能需要优化Agent的提示词要求其输出更结构化的内容。流程僵化当前的线性流程检索-分析-撰写是否适合所有问题也许有些简单问题不需要分析可以直接撰写。这时就需要引入条件边。错误处理如果检索Agent什么都没找到怎么办如果分析Agent工具调用出错怎么办一个健壮的系统需要在关键节点添加错误处理和备用路径如重试、降级处理。5. 高级模式与设计模式探讨当你掌握了基础的多Agent编排后可以尝试一些更高级的模式来解决复杂问题。5.1 管理者-工作者模式这是最经典的中心化模式。一个“管理者Agent”负责理解和分解任务然后根据子任务的性质动态地调用不同的“工作者Agent”专家来完成。管理者还负责整合各工作者的结果。这类似于一个公司的部门经理。在LangGraph中你可以用一个“管理者”节点来接收任务其内部逻辑是调用一个LLM来判断该任务属于哪一类然后通过条件边将状态路由到不同的工作者节点如“编码工作者”、“写作工作者”、“查询工作者”。工作者完成后再将结果返回给管理者节点进行汇总。5.2 辩论与共识模式当面对没有标准答案的开放性问题时如“设计一个产品logo”可以让多个同类型的Agent如几个不同的“创意设计师Agent”独立工作产生多个方案然后引入一个“评审员Agent”来评估这些方案或者让Agent们进行多轮“辩论”最终达成一个共识或选出最佳方案。这种模式能有效提升输出的多样性和质量。实现上这需要循环。你可以设置一个“创意生成”节点并行或串行调用多个Agent将结果收集到一个列表中。然后进入“评审/辩论”节点该节点可能会多次调用评审Agent更新方案评分或修改意见直到满足某个终止条件如达到最大轮次或共识度阈值。5.3 动态子任务创建对于一些极其复杂、无法预先规划的任务如“为我策划一次跨国旅行”可能需要系统在运行过程中动态创建新的子任务。这要求“管理者Agent”具备更强的规划能力。一种实现方式是让管理者Agent的输出不仅包含结果还包含一个“后续任务列表”。工作流在执行完当前节点后会检查这个列表。如果不为空则动态地将新任务作为新的节点或新的工作流实例加入执行队列。LangGraph的State可以设计一个pending_tasks: List字段来支持这种动态性。6. 避坑指南与性能调优搭建多Agent系统会踩很多坑这里分享几个我亲身经历过的教训和优化技巧。6.1 常见问题与排查Agent陷入循环或卡住症状工作流长时间不结束某个Agent反复执行相同操作。排查首先检查Agent的提示词是否包含了明确的停止条件如“当你获得足够信息时请用Final Answer:开头输出”。其次在LangGraph中为循环结构如辩论模式必须设置最大迭代次数可以使用add_conditional_edges并设置一个计数器在状态中达到上限后强制跳出。工具设计确保工具函数的返回值是确定且格式良好的。一个返回None或抛出异常的工具很容易导致Agent困惑并反复重试。信息在传递中失真或丢失症状下游Agent抱怨拿到的是“空数据”或“错误数据”。排查这是状态设计不佳的典型表现。不要传递冗长的、非结构化的文本。强制上游Agent输出结构化数据如JSON。例如要求检索Agent输出{key_facts: [...], source_links: [...]}而不是一大段话。下游Agent的提示词也要明确指定从状态的哪个字段、以何种格式读取输入。系统延迟高、成本昂贵症状运行一个流程耗时过长API调用费用激增。优化并行化如果多个子任务间没有依赖关系一定要用LangGraph的并行节点能力让其同时执行。缓存对LLM调用和工具调用实施缓存。对于相同输入直接返回历史结果。LangChain提供了InMemoryCache、SQLiteCache等组件。模型分级并非所有环节都需要GPT-4。让“管理者”或核心分析环节用强模型让“格式化”、“简单分类”等任务使用gpt-3.5-turbo甚至更小的开源模型能大幅降低成本。精简上下文避免将整个对话历史或所有中间结果都塞进每个Agent的上下文。只传递必要的信息。6.2 状态管理与可观测性当系统变得复杂调试会变得困难。你必须建立良好的可观测性。日志记录在每个节点的函数中详细记录输入、输出和关键步骤。可以使用Python的logging模块并将日志级别与状态关联。状态快照在LangGraph中app.invoke()返回的最终状态只包含最后结果。为了调试你可以在编译图时设置debugTrue或者手动在状态中增加一个execution_log: List字段每个节点执行后都把关键信息append进去。可视化LangGraph一个很棒的功能是能将你的图可视化出来。使用app.get_graph().draw_mermaid_png()可以生成流程图直观地看到你的团队结构这对于向他人解释系统设计非常有帮助。6.3 安全与稳定性考量工具权限隔离这是最重要的安全原则。绝不要给所有Agent访问所有工具的权限。一个负责外部搜索的Agent不应该有访问数据库或执行系统命令的权限。在创建AgentExecutor时严格限定其tools列表。输入输出验证对流入每个Agent的输入和其产生的输出进行基本的验证和清理防止Prompt注入攻击或异常数据导致系统崩溃。可以使用Pydantic模型对状态进行强类型校验。设置超时与回退为每个Agent执行器或工具调用设置超时时间。当某个环节长时间无响应时应有超时机制和预定义的回退方案如返回一个默认值或错误信息避免整个工作流被卡死。人性化接管在关键决策点如最终报告生成前、执行具有外部影响的操作前设计“人工审核”节点。工作流可以暂停将当前状态和待决策信息发送给人工审核待确认后再继续执行。这在生产环境中至关重要。从单个Agent到多Agent系统你构建的不再是一个工具而是一个数字组织。这其中的挑战从单纯的提示工程、工具调用上升到了系统架构、流程设计、协同逻辑的层面。它要求开发者兼具AI应用开发和软件工程的双重思维。但回报也是巨大的一个设计精良的多Agent系统能够解决那些过去被认为必须由人类团队协作才能完成的复杂任务真正释放出AI的群体智能潜力。