最近一年多我发现一个现象凡是聊到复杂LLM应用开口闭口都是Multi-Agent但真正能把多个Agent有效编排跑起来、稳定输出结果的案例并不多。LangGraph作为基于图结构的Agent编排框架是目前我在实际项目里用得最多的一套方案。这篇文章不打算重复官方文档里的入门Demo而是用一个可落地的完整案例——一个三Agent协作的研究报告生成系统把LangGraph Multi-Agent从设计、编码到排错的全过程拆开讲清楚。适合已经写过单Agent调用、想升级到多Agent协作的开发者也适合正在做技术选型的读者作个参考。1. 为什么需要Multi-Agent单Agent的瓶颈从哪来1.1 单Agent在高复杂度任务下的典型困境单Agent本质上是一个大而全的执行者给它一个目标它自己去调工具、读上下文、生成结果。在简单场景下这种模式很舒服比如只做翻译、只做摘要、只做单轮问答效果都不差。但一旦任务复杂起来问题就变得很明显上下文互相污染检索、分析、写作全压在一个上下文窗口里工具返回结果多了之后Agent容易忘记最初目标生成的结论偏离用户需求。提示词互相干扰同一个Agent既要做检索规划又要做数据解读还要控制输出格式提示词越长模型越容易顾此失彼。我见过不少项目把单个Prompt堆到三四千字之后Agent反而开始频繁出错。无法并行单Agent天然串行如果任务里有多个可以同时处理的子任务比如同时查多个数据源它也只能一个一个来整体耗时成倍上涨。错误难以定位一个环节出错整个会话就失败你很难判断是检索环节的问题、还是推理环节的问题排错成本很高。1.2 拆成多Agent为什么能缓解这些困境Multi-Agent的思路是拆。把一个大任务按职责拆成若干个子Agent每个Agent只负责一个相对聚焦的环节通过图结构把这些环节组织起来共享一份状态、有序或并行地推进。这样每个Agent的上下文窗口里只装自己关心的内容Prompt短了模型负担小了出错率自然降下来并行分支让耗时也能压下去。这里要先泼一盆冷水多Agent不是银弹拆得不好反而比单Agent更慢更乱。第6节我会专门说什么时候不该用。但当你确实遇到前面这些情况时LangGraph这套编排框架就是我目前最顺手的选择。1.3 为什么选LangGraph而不是手写编排我选择LangGraph而不是纯LangChain顺序链或者手写循环原因有三点它有显式的图结构节点和边是声明式定义的整个执行流程一眼能看懂排错比黑盒循环清晰得多。状态管理是内置的所有Agent共享一份state不需要自己维护一堆全局变量或者反复传参。支持条件分支、并行发送、人机交互中断恢复等高级控制流扩展性比固定链式调用强很多。我实际用下来LangGraph最大的价值在于它把多Agent系统从思路变成了图纸。你先把图设计出来再往节点里填Agent逻辑整个系统的行为是可预期的。2. LangGraph的核心概念StateGraph、节点与共享状态在进入案例之前先把LangGraph里最关键的几个概念用人话说一遍。我见过太多人卡在概念上其实它背后的思想很朴素就是一个有向图在跑一个状态机。2.1 State所有Agent共享的工作台LangGraph里State是贯穿整个图执行过程的数据对象。你可以把它理解成所有Agent共用的工作台每个Agent执行完后把产出写到工作台上下一个Agent从工作台读取自己需要的东西。State一般用一个TypedDict或dataclass定义比如from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class AgentState(TypedDict): user_query: str search_results: list[dict] analysis_result: dict final_report: str messages: Annotated[list, add_messages]里面有两个值得注意的细节。第一Annotated[list, add_messages]是LangGraph里很常用的reducer写法表示当多个节点往messages字段写入内容时不是覆盖而是累加合并。这个机制对多Agent场景至关重要后面讲并行节点时会再碰到。第二除了messages之外其余字段默认是后写覆盖也就是同一轮里多节点都往search_results写的话最后一次写入会覆盖前面的这也是新手最容易踩的坑之一。2.2 Node与Edge谁干活、谁传给谁Node就是图里的节点一个节点通常就是一个函数输入整个State输出State的局部更新。Edge是节点之间的连接关系决定执行顺序。from langgraph.graph import StateGraph, START, END builder StateGraph(AgentState) builder.add_node(search, search_agent) builder.add_node(analyze, analyze_agent) builder.add_node(write, write_agent) builder.add_edge(START, search) builder.add_edge(search, analyze) builder.add_edge(analyze, write) builder.add_edge(write, END) graph builder.compile()这段代码就是一个最基础的三节点流水线。LangGraph执行时会按图的拓扑顺序把整个图跑完当前节点返回后框架根据State的更新结果和边定义自动决定下一个节点。编译出的Graph对象可以直接invoke调用也可以stream逐步观察。2.3 条件边与Send API图不是死板的流水线纯顺序执行其实用普通函数链也能做LangGraph真正的优势在条件边conditional_edges和Send API。条件边解决结果不同走不同支路的问题比如分析后认为数据不足就返回search节点补搜一轮def route_after_analysis(state): if state.get(needs_more_search): return search return write builder.add_conditional_edges( analyze, route_after_analysis, {search: search, write: write} )Send API解决并行下发的问题一份任务列表可以并发派发给多个同类型的节点整体图结构是动态展开的。这一块结合下面的案例讲会更直观。3. 案例拆解一个三Agent研究助手是如何跑通的下面进入正题。我以一个实际做过的项目为蓝本做一个行业研究报告生成助手用户输入一个研究主题系统自动产出结构化的研究报告。这个场景非常适合讲Multi-Agent因为它的子任务边界天然清晰。3.1 需求定义与Agent角色划分原始需求听起来很简单输入2025年AI Agent落地现状这样的主题输出一份报告。但真正拆开之后会发现里面其实包含三个性质完全不同的子任务资料搜集需要基于主题拆出多个子问题然后分别调用搜索工具获取资料。这个环节的关键是广和快越多独立检索结果越好。数据分析把搜集到的原始资料做清洗、提炼、归纳形成论点清单和关键数据。这个环节的关键是准需要过滤掉无关信息。报告撰写基于分析结果组织成段落结构生成最终报告。这个环节的关键是稳输出格式必须统一。所以我把系统拆成三个Agentsearch_agent资料搜集、analyze_agent数据分析、write_agent报告撰写。每个Agent的Prompt都刻意保持短小聚焦只描述自己的职责和输出格式约束不掺杂其他环节的内容。一个重要经验角色划分的粒度不是越细越好而是按上下文需求差异来划分。如果两个环节需要关注的上下文完全不同才应该拆成两个Agent如果只是顺序不同但上下文相似拆开反而增加通信成本。3.2 代码实现从State定义到Graph编排先定义三个Agent。这里我用的是LangGraph提供的create_agent辅助函数实际项目里用这个版本比手写bind_tools的方式少很多样板代码旧版本里这个函数叫create_react_agent用法基本一致from langgraph.prebuilt import create_agent from search_tools import web_search, fetch_content search_agent create_agent( llm, tools[web_search, fetch_content], system_prompt你负责资料搜集。请根据研究主题拆解出3-5个子问题 逐一调用搜索工具获取资料返回结构化参考资料列表。 注意只负责搜集不要做深度分析。, )分析Agent不接外部工具纯做推理analyze_agent create_agent( llm, tools[], system_prompt你是数据分析员。基于参考资料提炼3-5条核心结论 每条结论需要注明来源索引输出格式为JSON。, )写作Agent同样不接工具负责最终润色write_agent create_agent( llm, tools[], system_prompt你是资深报告撰写人。基于分析结论生成结构完整的行业研究报告 包含摘要、正文各节、关键数据表、结论与风险提示。, )接下来定义State和节点。比较关键的是搜集环节需要并行处理多个子问题这里我用Send API实现import operator from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END from langgraph.types import Send class ReportState(TypedDict): topic: str sub_questions: list[str] all_results: Annotated[list, operator.add] # 并行结果累加不互相覆盖 conclusions: list[dict] report: str def plan_questions(state: ReportState) - dict: questions split_topic_to_questions(state[topic]) return {sub_questions: questions} def search_one(state: dict) - dict: # Send分发后这里的state是全局State 本次Send传入payload合并后的结果 question state[question] results run_search(question) return {all_results: [{question: question, results: results}]} def analyze_one(state: ReportState) - dict: conclusions analyze_all(state[all_results]) return {conclusions: conclusions} def write_report(state: ReportState) - dict: report generate_report(state[topic], state[conclusions]) return {report: report}在Graph里完成编排builder StateGraph(ReportState) builder.add_node(plan, plan_questions) builder.add_node(search_task, search_one) builder.add_node(analyze, analyze_one) builder.add_node(write, write_report) builder.add_edge(START, plan) def dispatch_search(state: ReportState): return [ Send(search_task, { topic: state[topic], question: q }) for q in state[sub_questions] ] # 动态并行每个子问题都会创建一份独立的search_task执行 builder.add_conditional_edges(plan, dispatch_search, [search_task]) builder.add_edge(search_task, analyze) builder.add_edge(analyze, write) builder.add_edge(write, END) graph builder.compile()这里Send(search_task, payload)的含义是为payload里的每个元素创建独立的search_task节点实例并行执行。每个实例的返回值通过operator.add这个reducer累加到all_results上不会互相覆盖。这就是LangGraph里并行多Agent的基本写法也是这套框架对比手写并发代码最舒服的地方。3.3 运行结果与状态流转解读拿真实输入2025年AI Agent落地现状跑一遍整个流程是这样的plan节点把主题拆成企业采用率数据主要应用场景技术栈分布落地瓶颈未来趋势五个子问题。五个search_task并行启动每个调用搜索工具并抓取页面内容耗时基本同时完成。所有结果汇总到all_results后analyze节点把所有资料压缩成5条核心结论并标注来源索引。write节点基于结论生成一篇约3000字的完整报告。我把streaming打开做可视化观测能看到典型的先发散再收敛再发散过程plan阶段拆问题是在发散search阶段并行分支达到最大发散度analyze阶段收敛成结论write阶段又为了组织语言而发散。这种动态过程的掌控感正是图编排框架相对传统链式调用的核心优势。LangGraph的streaming能力在调试多Agent时非常有用。graph.stream(initial_state, stream_modeupdates)可以逐步看到每个节点返回的增量状态stream_modemessages则能看到模型逐字生成的过程。我在调试阶段几乎全程开着这两个模式。4. 三种主流协作模式Supervisor、分层与对等上面的案例是流水线式协作属于最基础的一种。实际项目中做技术选型时LangGraph里常见的有三种组织模式我分别说下适用场景和坑。4.1 Supervisor模式一个老板管一堆员工Supervisor模式里有一个主管Agent负责决策把任务分发给多个执行Agent并根据执行结果决定下一步派给谁def supervisor_agent(state): decision llm.invoke(supervisor_prompt.format(statestate)) return {next: decision.action} builder.add_conditional_edges(supervisor, route_based_on_next, { researcher: researcher, coder: coder, writer: writer, FINISH: END })这种模式适合任务类型不固定、执行路径依赖中间结果的场景。坑在于主管Agent本身可能做出错误决策尤其是Agent数量超过四个之后选错分支的概率会明显上升。我的应对方式是给主管Agent提供当前各Agent能力说明和上次执行摘要并在关键路径上做硬编码校验。4.2 分层模式Agent之间还有子Agent分层结构是多层嵌套顶层Agent把任务拆给中层中层再拆给底层。好处是每个层级的上下文窗口更可控适合非常复杂的任务代价是延迟翻倍调试困难而且任何一层出错都会被放大。我在实际中很少直接写多层嵌套除非有明确的上下文隔离需求。一个相对好用的替代方案是先让顶层Agent做任务规划输出一个带依赖序的任务列表然后交给扁平流水线去执行。这样逻辑上是分层的执行上还是扁平图兼顾了清晰与可控。4.3 对等模式一群Agent互相协作对等模式里每个Agent地位相同可以自由地互相发送消息。在手写循环里这种模式经常失控因为没有收敛机制。LangGraph里我会用固定轮次上限统一汇总节点来约束比如最多三轮对话后强制进入汇总节点。三种模式的选择逻辑我用一个表格总结模式适合场景主要风险我的推荐程度流水线子任务顺序明确输入输出边界清晰中间结果不达预期时缺乏回退优先使用Supervisor执行路径依赖中间决策任务类型多样主管决策错误分支发散适合3-5个Agent分层嵌套任务极复杂需要隔离上下文延迟高调试难谨慎使用对等需要多角度讨论或评审容易发散无收敛建议加轮次限制5. 实测中的坑与优化并发、超时与状态污染这部分是我在真实项目里踩过坑之后沉淀下来的常规教程很少会写但直接影响系统能不能稳定上线。5.1 并行节点的状态覆盖问题这是我在第一次用Send API时踩的坑。当时我在State里定义all_results用的不是Annotated[list, operator.add]而是一个普通list字段。结果发现五个并行search_task节点都返回后all_results里只剩最后一个节点写的结果。原因就是LangGraph默认的后写覆盖语义并行节点往同一个普通字段写数据时后执行的覆盖先执行的。解决方案就是给字段加reducer用operator.add做累加合并。这个坑很隐蔽因为串行流程下你怎么写都正常一旦切到并行立刻翻车。5.2 超时与重试多Agent系统最怕一件事某个Agent调外部工具时卡住了。不设置超时整个Graph会一直挂着用户看到一个转圈圈但是没有任何反馈。我通常给每个外部工具调用加上超时控制并在节点层做降级处理import asyncio async def safe_search(question: str, timeout: float 20.0) - dict: try: return await asyncio.wait_for(run_search(question), timeouttimeout) except asyncio.TimeoutError: return {error: timeout, question: question}在节点函数里遇到timeout时返回降级数据而不是让异常直接冒泡到Graph执行器。这样即使某个搜索源挂了顶上来的也是一条该子问题未获取到数据的占位结果后续分析环节仍然可以继续只是结论里会多一条数据有限说明。多Agent系统的容错设计本质上就是任何单点失败都不能拖垮整条链路。5.3 Token消耗的隐性放大多Agent的Token消耗比单Agent高很多这点很多人启动前没算过账。比如3个Agent各维护一段上下文光系统提示词就是三份并行search时每个分支都带一份完整历史如果任务切成5个子问题状态里的消息数据会被重复传递多次。我习惯在State里区分局部上下文和全局上下文两种字段。局部字段只存在于某个Agent执行期间完成后就丢弃只有必要的结论性内容才写入全局State。这个设计能明显降低Token消耗尤其是并行分支多的时候。5.4 调试手段可视化与逐步执行LangGraph编译后的图可以输出为结构表示配合graph.stream的逐步输出基本能替代肉眼Debug。我个人调试多Agent的固定流程是先用stream_modeupdates看节点级的返回确认每个节点产出的State字段是否符合预期。再用stream_modemessages盯具体某一步的模型输出确认Prompt没有跑偏。最后在开发环境打开debugTrue看每一步的函数调用顺序和耗时。这套流程的核心思路是从粗到细先看结构对不对再看内容对不对最后看性能对不对。直接跳到最后一步往往会被大量日志淹没。6. 什么时候不该用Multi-Agent这个章节本来不想写但无论做技术分享还是自己复盘我觉得什么时候不该用比怎么用更重要。这些年看过不少项目多Agent用得飞起最后效果反而不如单Agent原因基本都落在下面几点。6.1 任务边界本身不清晰如果任务本身没有明确可拆分的子目标硬拆多个Agent只会让每个Agent都看着全局却只做局部上下文里堆满了自己用不到的信息。这种情况最好的方案是强化单个Agent的Prompt和工具集让它自己决定何时调用什么工具。6.2 延迟和成本敏感的场景要慎重多Agent必然带来更多的模型调用次数和更大的上下文开销。如果是一个线上低延迟接口比如要求几百毫秒内响应的场景多Agent的链路往往跑不完尤其是多轮Supervisor决策会带来数倍延迟。这种场景我宁愿单Agent加精调Prompt或者用更小的模型做第一道应答。6.3 团队对Agent机制不熟的时候先别急这是很现实的约束。多Agent系统是伴生复杂度很高的系统对状态管理、并发安全、可观测性的要求都高于普通应用。如果团队刚接触LLM应用我建议先从单Agent加工作流编排入手把状态、工具、上下文管理搞扎实再逐步引入多Agent。毕竟拆Agent容易把拆出来的Agent稳定协调起来才是真正的考验。6.4 能用确定性代码解决的就别用Agent很多所谓的多步决策其实用确定性代码写死更稳。比如先查数据库再根据结果走A/B分支这类逻辑本质上是状态机而不是智能决策直接写代码比让Agent绕一圈更快、更便宜、更可控。Agent应该用在决策空间大、规则难以穷举的地方比如开放式研究、复杂任务规划、多源信息综合。我的经验判断法如果一个任务能在五分钟内用普通代码写出确定性的执行流程就不要引入Agent如果这个流程会随着输入变化而变化才值得让Agent做路由和编排。最后再分享一个小技巧在LangGraph里给每个节点都起一个简洁且语义明确的name不要偷懒用默认函数名。因为后面做条件路由、做Supervisor决策、看可视化图的时候这些名字是唯一能快速定位问题的线索。我在项目里吃过亏——初期不重视命名一上Supervisor就懵了最后花了大量时间在图上重新梳理节点关系。多Agent的编排本质上是把一个聪明但容易迷路的执行者变成一群各司其职的执行者加上一个清晰的指挥体系。LangGraph给的是这套体系的骨架真正让它跑得稳的是你怎么切分职责、怎么设计状态、怎么控制并发。希望这篇案例拆解能帮你少走一些我走过的弯路。