1. 从单兵作战到团队协作为什么需要多 Agent 系统如果你已经跟着 LangChain 的教程一路走来从基础的链Chain到工具Tool再到单个 Agent你可能会觉得已经掌握了让大模型“干活”的秘诀。单个 Agent 就像一个全能的超级员工你给它一个任务它自己规划、调用工具、思考、最终给出答案。处理一些明确、线性的任务比如查天气、写摘要、分析单个文档确实得心应手。但现实世界的问题尤其是稍微复杂一点的业务场景很少是线性的。想象一下你要开发一个智能数据分析平台用户上传一份销售报表希望得到一份包含市场趋势分析、竞品对比和未来销售预测的综合性报告。这个任务至少涉及几个环节1数据清洗与提取2趋势分析与可视化3外部市场信息检索4报告撰写与润色。让一个 Agent 从头到尾包办就像让一个程序员同时兼任产品经理、UI设计师、后端开发和测试工程师——不是完全不行但效率低下且每个环节都难以做到专业。这就是多 Agent 系统Multi-Agent System登场的时刻。它的核心思想是“专业的人做专业的事”。我们将一个复杂的宏观任务分解成多个子任务然后为每个子任务设计一个或多个专门的 Agent。这些 Agent 各司其职有的擅长数据解析有的精通代码生成有的负责对外沟通。它们之间通过一套明确的协作机制比如共享工作区、消息传递、任务委派来协同工作共同完成最终目标。我最初接触这个概念时觉得它有点“杀鸡用牛刀”。但真正在一个需要处理多轮对话、动态工具选择和复杂状态管理的项目里踩过坑后我才明白当业务逻辑的复杂度超过某个阈值时多 Agent 架构不是可选项而是必选项。它能带来几个显著优势模块化与可维护性每个 Agent 功能单一职责清晰。当需要修改数据分析逻辑时你只需要改动对应的“数据分析师”Agent而不会影响到“报告撰写员”Agent。专业化与性能提升可以为不同的子任务配备最合适的模型或工具。例如让一个较小的、速度快的模型处理简单的信息归类而让一个更大的、能力更强的模型负责需要深度推理的环节。鲁棒性与容错单个 Agent 的失败比如工具调用超时不意味着整个系统崩溃。通过设计监督者Supervisor或路由逻辑可以将任务重新分配或进入错误处理流程。模拟复杂社会行为这在游戏、仿真、辩论等场景中非常有用。你可以创建具有不同性格、目标和知识背景的 Agent让它们互动产生意想不到的、接近真实世界的动态。所以当你发现你的单个 Agent 的提示词Prompt变得无比冗长复杂或者任务状态管理让你头疼不已时就该认真考虑引入多 Agent 系统了。接下来我们就深入 LangChain 提供的多 Agent 实现核心LangGraph。2. LangGraph为多 Agent 协作绘制“工作流程图”在 LangChain 的生态中实现多 Agent 系统的核心库是LangGraph。你可以把它理解为一个专门为基于大模型的智能体工作流而设计的“流程图绘制与执行引擎”。如果说 LangChain 的核心Chain和AgentExecutor描述的是“一步接一步”的线性或有限循环流程那么 LangGraph 描述的就是任意复杂的、有向的、可能带环的图Graph结构。为什么是“图”因为多 Agent 协作很少是简单的流水线。它可能包含分支根据条件选择下一个执行的 Agent、循环某个环节需要反复迭代直至满足条件、并行多个 Agent 同时处理不同任务以及汇聚等待多个并行任务完成后再继续。用代码里的if-else和for循环硬编码这些逻辑会非常痛苦且难以维护。LangGraph 通过“状态State”和“节点Node”的概念让你能够直观地设计和运行这样的工作流。2.1 核心概念拆解理解 LangGraph需要先掌握几个核心概念我结合一个“技术客服工单处理系统”的例子来解释状态State 这是一个贯穿整个工作流的共享数据容器通常是一个字典Dict或 Pydantic 模型。它存储了工作流执行过程中的所有信息。比如在我们的客服系统中状态可能包含{ “messages”: [用户提问 Agent回复…], # 对话历史 “customer_query”: “我的订单号12345为什么还没发货”, # 原始问题 “parsed_intent”: “物流查询”, # 意图分析结果 “order_info”: {“status”: “shipped”, “tracking_number”: “XYZ789”}, # 查询到的订单信息 “current_agent”: “router”, # 当前正在执行或刚执行完的Agent名称 “final_answer”: None # 最终给用户的答复 }每个节点Agent读取状态的一部分执行操作然后更新状态。状态是 Agent 之间通信的唯一媒介。节点Node 节点是工作流中的一个执行单元本质上是一个函数。这个函数接收当前的“状态”作为输入执行一些操作比如调用一个大模型运行一段代码查询数据库然后返回更新后的“状态”。在 LangGraph 中一个节点通常就封装了一个 Agent 或一个特定的工具调用逻辑。例如我们可以有intent_classifier_node一个节点其函数读取state[“customer_query”]调用一个 LLM 进行意图分类然后将结果写入state[“parsed_intent”]。order_lookup_node一个节点其函数读取state[“parsed_intent”]和state[“customer_query”]中的订单号调用内部订单查询 API将结果写入state[“order_info”]。边Edge 边定义了节点之间的流转条件。它决定了在当前节点执行完毕后下一个应该执行哪个节点。边可以是条件边Conditional Edge根据状态的某个值来决定下一步。例如如果state[“parsed_intent”] “物流查询”则流向order_lookup_node如果是“产品咨询”则流向product_kb_node。固定边Fixed Edge无条件地流向指定的下一个节点。图Graph 图就是由节点和边组成的网络。LangGraph 让你通过代码“绘制”出这个网络然后它提供一个运行时来按图索骥地执行。2.2 与普通 LangChain Agent 的关键区别很多初学者会混淆觉得用AgentExecutor也能实现多步调用为什么还要用 LangGraph这里的关键区别在于“控制权”和“显式性”。普通 Agent你给 Agent 一个任务和一堆工具然后说“你去干吧”。Agent 内部有一个循环思考 - 决定调用哪个工具或直接结束 - 执行工具 - 观察结果 - 再思考… 这个循环的控制逻辑是隐含在 Agent 内部的由 LLM 的每次输出包含AgentAction或AgentFinish来驱动。你很难从外部干预这个循环的路径也很难清晰地定义“如果工具A失败了应该去尝试工具B”这样的复杂逻辑。LangGraph 多 Agent控制逻辑被外化、显式地定义在了“图”的结构中。你作为架构师清晰地画出了工作流的蓝图“先由分类器 Agent 判断意图如果是A类问题交给专家 Agent A 处理处理完后必须由审核 Agent 检查如果是B类问题直接并行调用 Agent B1 和 B2 收集信息然后汇总给报告 Agent”。每个 Agent节点只关心自己的“一亩三分地”做完自己的事更新状态然后交给“图”来决定下一步谁上。这带来了无与伦比的清晰度、可调试性和可控性。简单说LangChain Agent 是让一个“聪明人”自己想办法完成任务而 LangGraph 是让你作为一个“项目经理”组建一个团队并亲自设计好每个人的工作交接流程。3. 实战构建一个多 Agent 协作的代码审查助手理论讲得再多不如动手建一个。我们来构建一个相对实用且能体现多 Agent 协作价值的系统一个智能代码审查助手。它的工作流程是接收用户提交一段代码。分析由不同的“专家”Agent 从不同角度语法、安全、性能、风格并行审查代码。汇总由一个“主审”Agent 收集所有专家的意见生成一份综合性的、优先级排序的审查报告。优化可选根据报告自动生成修改建议或修复代码。这个流程天然适合多 Agent每个审查维度是独立的可以并行汇总需要综合判断。我们用 LangGraph 来实现它。3.1 定义共享状态与审查专家 Agent首先我们定义整个工作流要共享的状态。我们使用TypedDict来获得更好的类型提示。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END from langchain_core.messages import HumanMessage from langchain_openai import ChatOpenAI from langchain.agents import create_tool_calling_agent, AgentExecutor from langchain.tools import tool import asyncio # 1. 定义状态结构 class CodeReviewState(TypedDict): # 输入 original_code: str # 并行审查的结果 syntax_issues: List[str] security_issues: List[str] performance_issues: List[str] style_issues: List[str] # 汇总结果 summary_report: str # 用于跟踪流程 reviews_completed: Annotated[list, operator.add] # 这是一个特殊注解用于LangGraph的并行节点结果合并 # 2. 实例化LLM这里用OpenAI GPT-4你可以替换为任何ChatModel llm ChatOpenAI(model“gpt-4-turbo-preview”, temperature0) # 3. 创建“专家”Agent的函数节点 # 注意这里为了简化我们没有为每个专家创建完整的ToolAgent结构而是直接让LLM扮演专家角色。 # 在实际复杂场景中每个专家可以有自己的工具集如调用ESLint、Bandit等静态分析工具。 async def syntax_review_node(state: CodeReviewState) - dict: “”“语法与基础错误审查专家”“” print(“[Syntax Agent] 开始工作...”) prompt f“”” 你是一个资深的编程语言语法专家。请严格审查以下代码片段找出所有语法错误、未定义的变量、错误的方法调用等基础问题。 只返回找到的问题列表每个问题用‘- ’开头。如果没有问题返回‘无’。 代码 “{state[‘original_code’]}” “”” message [HumanMessage(contentprompt)] response await llm.ainvoke(message) # 使用异步调用 issues response.content.split(‘\n’) # 简单清理空行 issues [i.strip() for i in issues if i.strip() and i.strip() ! ‘无’] return {“syntax_issues”: issues, “reviews_completed”: [“syntax”]} async def security_review_node(state: CodeReviewState) - dict: “”“安全漏洞审查专家”“” print(“[Security Agent] 开始工作...”) prompt f“”” 你是一个网络安全专家。请审查以下代码找出潜在的安全漏洞例如SQL注入、命令注入、路径遍历、硬编码密钥、不安全的反序列化等。 只返回找到的安全问题列表每个问题用‘- ’开头并简要说明风险。如果没有问题返回‘无’。 代码 “{state[‘original_code’]}” “”” message [HumanMessage(contentprompt)] response await llm.ainvoke(message) issues response.content.split(‘\n’) issues [i.strip() for i in issues if i.strip() and i.strip() ! ‘无’] return {“security_issues”: issues, “reviews_completed”: [“security”]} # 类似地可以定义 performance_review_node 和 style_review_node async def performance_review_node(state: CodeReviewState) - dict: “”“性能问题审查专家”“” print(“[Performance Agent] 开始工作...”) prompt f“”” 你是一个性能优化专家。请审查以下代码找出可能存在的性能瓶颈例如低效的循环、重复计算、未使用索引的数据查询、大对象的不必要拷贝等。 只返回找到的性能问题列表每个问题用‘- ’开头。如果没有问题返回‘无’。 代码 “{state[‘original_code’]}” “”” message [HumanMessage(contentprompt)] response await llm.ainvoke(message) issues response.content.split(‘\n’) issues [i.strip() for i in issues if i.strip() and i.strip() ! ‘无’] return {“performance_issues”: issues, “reviews_completed”: [“performance”]} async def style_review_node(state: CodeReviewState) - dict: “”“代码风格审查专家”“” print(“[Style Agent] 开始工作...”) # 假设我们关注Python代码提示词可以更具体 prompt f“”” 你是一个代码风格警察专注于Python PEP 8。请审查以下代码的风格问题例如命名不规范、行过长、缺少空格/空行、不推荐的写法等。 只返回找到的风格问题列表每个问题用‘- ’开头。如果没有问题返回‘无’。 代码 “{state[‘original_code’]}” “”” message [HumanMessage(contentprompt)] response await llm.ainvoke(message) issues response.content.split(‘\n’) issues [i.strip() for i in issues if i.strip() and i.strip() ! ‘无’] return {“style_issues”: issues, “reviews_completed”: [“style”]}这里有几个关键点我们定义了四个专家节点函数它们都接收state返回要更新到状态中的字典。我们使用了async/await进行异步调用这对于并行执行多个 Agent 至关重要能极大提升效率。Annotated[list, operator.add]是 LangGraph 的一个妙用。它告诉框架reviews_completed这个字段在多个节点并行执行时应该使用operator.add即列表的操作来合并结果。这样当四个专家都完成后这个列表就会是[“syntax”, “security”, “performance”, “style”]。3.2 构建图并实现并行与汇总接下来我们创建图添加节点并定义它们之间的流转关系。# 4. 创建图构建器 workflow StateGraph(CodeReviewState) # 5. 添加节点 workflow.add_node(“syntax_reviewer”, syntax_review_node) workflow.add_node(“security_reviewer”, security_review_node) workflow.add_node(“performance_reviewer”, performance_review_node) workflow.add_node(“style_reviewer”, style_review_node) # 6. 设置入口点我们希望四个专家并行工作。 # LangGraph 通过 add_edge 来连接节点但我们需要一个“开始”节点来触发并行。 # 通常我们会设置一个虚拟的入口节点或者直接设置图的入口。 # 这里我们采用更清晰的方式定义一个“路由”节点它决定下一步是并行执行专家审查。 async def parallel_review_entry_node(state: CodeReviewState) - dict: “”“入口节点不做具体事只是触发并行分支”“” print(“[Entry] 开始并行代码审查...”) return {} # 不修改状态 workflow.add_node(“entry”, parallel_review_entry_node) # 设置图的入口 workflow.set_entry_point(“entry”) # 7. 从入口节点同时指向四个专家节点实现并行分支 workflow.add_conditional_edges( “entry” # 下一个节点由这个函数决定。这里我们直接返回一个包含所有专家节点名的列表。 # LangGraph 会并行执行这些节点。 lambda state: [“syntax_reviewer”, “security_reviewer”, “performance_reviewer”, “style_reviewer”] # 注意add_conditional_edges 通常用于条件路由。对于真正的并行我们需要用 add_edge 到每个节点但标准模式是使用条件边返回列表。 # 更准确的做法是使用 add_edge(“entry”, “syntax_reviewer”) 等四条边但那样不是声明式的并行。 # 实际上LangGraph 通过让条件函数返回多个目标来实现“分支”。这些分支的执行顺序默认可能是并发的如果运行时支持。 ) # 更稳妥且清晰的并行控制是在所有专家节点完成后汇聚到一个汇总节点。这需要用到“边”的汇聚机制。 # 我们先为每个专家节点添加指向汇总节点的边。 workflow.add_node(“report_summarizer”, report_summarizer_node) # 先声明汇总节点函数见下文 # 添加从各专家到汇总节点的边 workflow.add_edge(“syntax_reviewer”, “report_summarizer”) workflow.add_edge(“security_reviewer”, “report_summarizer”) workflow.add_edge(“performance_reviewer”, “report_summarizer”) workflow.add_edge(“style_reviewer”, “report_summarizer”) # 8. 定义汇总节点 async def report_summarizer_node(state: CodeReviewState) - dict: “”“主审Agent汇总所有专家意见生成最终报告”“” print(“[Summarizer] 正在汇总所有审查意见...”) # 检查是否所有专家都已完成通过reviews_completed列表长度 # 在实际中LangGraph有一种“归约Reducer”机制来等待所有并行节点完成这里我们简化处理。 # 更严谨的做法是使用LangGraph的“屏障Barrier”或“归约”功能确保本节点在所有前置节点完成后才执行。 # 为了示例清晰我们假设此节点被调用时状态已包含所有专家的结果。 prompt f“”” 你是一个高级技术主管。以下是针对同一段代码四位不同领域专家语法、安全、性能、风格的独立审查意见 原始代码 “{state[‘original_code’]}” **语法专家意见** {‘\n’.join(state.get(‘syntax_issues’, [‘无’]))} **安全专家意见** {‘\n’.join(state.get(‘security_issues’, [‘无’]))} **性能专家意见** {‘\n’.join(state.get(‘performance_issues’, [‘无’]))} **风格专家意见** {‘\n’.join(state.get(‘style_issues’, [‘无’]))} 你的任务是 1. **整合与去重**将相关或重复的意见合并。 2. **分级与排序**按照问题的严重性阻断性错误 安全漏洞 性能瓶颈 风格问题和紧急程度进行排序。 3. **生成报告**输出一份清晰、专业的代码审查报告格式如下 - 概述简要总结代码整体质量。 - 关键问题高优先级列出必须修复的问题。 - 建议改进中优先级列出推荐改进的问题。 - 风格建议低优先级列出代码风格方面的建议。 请直接输出报告内容。 “”” message [HumanMessage(contentprompt)] response await llm.ainvoke(message) return {“summary_report”: response.content} # 9. 从汇总节点指向结束 workflow.add_edge(“report_summarizer”, END) # 10. 编译图 app workflow.compile()注意上面的代码在并行控制上做了简化。在真实的 LangGraph 应用中要实现“等待所有并行节点完成再进入汇总节点”通常需要使用StateGraph的add_conditional_edges配合状态检查或者使用更高级的构造器。一个更标准的模式是定义一个“归约”节点或者利用state[‘reviews_completed’]列表的长度来判断是否所有分支都已完成。为了保持示例的清晰度我们假设了简化流程。在实际项目中你需要仔细设计节点间的依赖关系。3.3 运行与测试现在让我们用一段有问题的 Python 代码来测试我们的多 Agent 审查系统。# 11. 测试运行 test_code “”” import os import subprocess def fetch_data(user_input): query f“SELECT * FROM users WHERE name ‘{user_input}’;” # 潜在SQL注入 # 连接数据库并执行查询假设 # result db.execute(query) return “data” def process_items(items): result [] for i in range(len(items)): # 性能使用enumerate更好 for j in range(len(items)): # 性能双重循环复杂度高 if items[i] items[j]: result.append(items[i]*2) # 风格操作符两侧应有空格 return result api_key “1234567890abcdef” # 安全硬编码密钥 “”” # 初始化状态 initial_state: CodeReviewState { “original_code”: test_code, “syntax_issues”: [], “security_issues”: [], “performance_issues”: [], “style_issues”: [], “summary_report”: “”, “reviews_completed”: [] } print(“ 开始多 Agent 代码审查 \n”) # 运行图 final_state app.invoke(initial_state) print(“\n 审查完成 \n”) print(“【最终汇总报告】”) print(final_state[“summary_report”]) print(“\n 各专家原始意见 ) print(f“语法问题 {final_state[‘syntax_issues’]}”) print(f“安全问题 {final_state[‘security_issues’]}”) print(f“性能问题 {final_state[‘performance_issues’]}”) print(f“风格问题 {final_state[‘style_issues’]}”)运行这段代码需要配置好 OpenAI API 密钥你会看到控制台依次打印出各个 Agent 开始工作的信息最后得到一份整合了四位专家意见的优先级排序报告。通过这个例子你可以直观地感受到多 Agent 系统如何将复杂任务分解并由专门的“角色”并行处理最后智能汇总。这比让一个全能 Agent 去完成所有步骤要清晰、高效、且结果更专业。4. 高级模式与避坑指南让多 Agent 系统稳定可靠构建了基础的多 Agent 系统后我们会面临更实际的挑战如何管理复杂的工作流如何确保 Agent 间协作的稳定性如何调试下面分享一些进阶模式和我在实践中踩过的坑。4.1 监督者Supervisor与层级式编排在更复杂的场景中Agent 之间可能需要动态的任务分配和调度。例如一个“客服主管”Agent 先接待用户根据问题类型动态地将任务分配给“技术专员”、“账单专员”或“物流专员”等子 Agent。子 Agent 处理完后可能需要将结果返回给主管由主管整合后回复用户或者主管判断子 Agent 处理不当要求其重试。这种模式被称为“监督者模式”或“层级式编排”。在 LangGraph 中可以通过一个“路由”节点即监督者来实现。这个路由节点本身也是一个 Agent它的工具就是“调用其他子 Agent”。监督者根据当前状态和对话历史决定下一步调用哪个子 Agent或者是否结束对话。实现要点为每个子 Agent 定义一个工具这个工具的内部逻辑就是调用该子 Agent 的工作流。监督者 Agent 拥有这些工具。在图中监督者节点循环运行调用 LLM 决定行动 - 执行工具即调用子Agent- 更新状态 - 再次调用 LLM 决定下一步…需要设置终止条件比如监督者 Agent 决定输出AgentFinish。这种方式给了系统极大的灵活性监督者可以像人类项目经理一样动态地规划任务路径。4.2 错误处理与状态管理多 Agent 系统出错是常态。网络超时、工具调用失败、LLM 输出格式不符合预期等等。节点级别的错误处理在每个节点函数内部使用try...except包裹核心逻辑。发生错误时可以选择重试、记录错误信息到状态中如state[“errors”].append(“某Agent失败XXX”)并返回一个标志如state[“node_x_failed”] True让后续的边Edge逻辑能够感知到这个错误从而路由到错误处理节点。图级别的错误处理LangGraph 允许你为图设置interrupts中断和triggers触发器。你可以定义一个“错误处理”节点当任何节点抛出特定异常时图可以中断当前流程跳转到这个错误处理节点进行日志记录、通知或尝试恢复操作。状态的版本化与快照对于长时间运行或关键的工作流考虑定期将状态持久化如存入数据库。这样在系统崩溃重启后可以从最近的一个状态快照恢复执行而不是从头开始。LangGraph 的状态本身是可序列化的这为持久化提供了便利。4.3 调试与可视化调试一个由多个 Agent 和复杂边组成的工作流是挑战。LangGraph 提供了强大的支持。状态快照在app.invoke()或app.astream()过程中你可以访问到每个步骤执行前后的完整状态。通过打印或记录这些状态你可以清晰地看到数据是如何在 Agent 间流动和变化的。流式输出使用app.astream()可以实时观察到每个节点的开始、结束以及产生的中间信息对于监控长流程非常有用。可视化图这是 LangGraph 的一大亮点。你可以使用get_graph().draw_mermaid_png()需要安装pygraphviz或将图导出为 Mermaid 格式从而生成一张可视化的流程图。这张图能让你一目了然地看清整个系统的架构、节点和边是理解和沟通设计的最佳工具。在团队评审时一张清晰的图比千行代码更有说服力。4.4 常见陷阱与性能优化状态爆炸不要在状态中存储过大的中间数据如整个文档内容、大型文件。尽量只存储引用如文件路径、数据库ID或摘要。对于需要传递的大数据考虑使用外部存储如对象存储、缓存在状态中只存键。LLM 调用成本与延迟每个 Agent 节点通常意味着一次或多次 LLM 调用。在设计工作流时要权衡“专业化带来的精度提升”和“调用次数增加带来的成本与延迟”。对于轻量级判断可以考虑使用小模型或规则引擎。充分利用异步Async来并行化独立的 LLM 调用这是降低整体延迟的关键。提示词Prompt设计冲突不同 Agent 的提示词如果设计不当可能导致行为冲突或输出格式不一致给汇总 Agent 带来困难。建立团队的“提示词规范”对于共用的部分如输出格式要求进行标准化。循环与停滞如果图中存在循环例如Agent A 的结果需要 Agent B 确认B 又可能将任务打回给 A必须设置最大迭代次数或超时机制避免陷入死循环。可以在状态中设置一个iteration_count字段每次循环递增并在边条件中检查它是否超过阈值。测试困难多 Agent 系统的测试比单函数复杂。建议采用分层测试先独立测试每个节点函数再测试小的子图如两个Agent的协作最后进行集成测试。使用固定的随机种子和模拟MockLLM 响应来保证测试的确定性。构建多 Agent 系统就像组建并管理一个团队初期设计沟通成本较高但一旦流程跑顺其处理复杂问题的能力和系统的可扩展性是单 Agent 架构难以比拟的。从 LangChain 的单体 Agent 迈入 LangGraph 的多 Agent 世界是你构建真正强大、可靠 AI 应用的关键一步。