多智能体系统这两年听起来特别火但真上手做过的都知道绝大多数问题根本不在模型选择上而在于你拿什么把多个Agent组织起来。我最早用纯Python写调度循环状态机全靠手工维护改一个分支逻辑要牵出一堆全局变量后来转到LangGraph重做了一遍整个思路清晰了不止一个量级。这篇文章就围绕LangGraph多智能体系统设计从需求拆解、状态建模、图编排到上线后的实测踩坑给你一条完整的端到端路径。不管是刚开始接触Agent开发还是已经在用LangChain想升级到带状态管理的编排层都可以参考这里的思路。1. 一切从需求拆解开始没有明确边界的多智能体等于失控很多人一上来就急着画图、写代码结果做到一半发现不知道该分几个Agent、谁调谁。我自己早期吃过这个亏接手一个合同审核项目想当然地拆了六个Agent每个负责一段文本最后发现它们各说各话上下文对不上一个问题来回踢皮球。后来才明白多智能体设计的第一步永远是需求拆解而不是系统设计。1.1 先回答三个问题再决定要不要用多个Agent我建议在画任何架构图之前先逼自己回答三个问题这个任务是否存在明确可划分的子任务比如检索资料和撰写总结就是天然的两个子任务而把用户问题分类通常只需要一个分类器。每个子任务是否需要独立的上下文窗口如果所有阶段必须共享完整对话历史多个Agent反而浪费Token处理不好还会相互污染。子任务之间是严格串行还是可以有并行、分支、条件回退串行场景用LangChain的链就能搞定复杂拓扑才需要LangGraph。这三个问题的答案直接决定你是否真的要上多智能体。我的判断标准很简单如果单Agent加工具调用能解决那就别硬上多智能体复杂度是线性增长的但调试难度是指数增长的。1.2 用角色-输入-输出-失败出口四元组把任务写清楚确定需要多Agent之后我会给每个Agent建一张四元组表这个习惯让我少走了很多弯路角色输入输出失败出口调研Agent用户问题结构化的搜索结果列表与要点摘要若搜索结果为空返回带空态标记的摘要写作Agent调研摘要文章初稿若摘要质量分过低触发重写流程审核Agent文章初稿审核意见或通过标记若连续两次未通过降级为人工提示这个表格看着简单但它逼你把接口定义清楚。多智能体系统本质上是分布式接口设计每个Agent的输入输出就是它和外部世界的协议。协议定不清楚后面用LangGraph做状态Schema时必然手忙脚乱。1.3 需求到流程的映射不要跳过拓扑这一步四元组写好之后下一步是画流程图。这里的流程图不是随便画个草稿而是要把每个节点之间的流转方向、条件分支、回退路径全部标出来。我遇到过最典型的问题就是回退路径被忽略。比如调研Agent返回结果质量很差时到底应该重跑调研还是把带标记的结果直接传给写作Agent这两种选择的拓扑完全不同前者是一条循环边后者是正常的顺序边。如果没有在需求阶段想清楚写代码的时候就会卡住。在LangGraph里循环边和顺序边的表示方法不一样循环还需要额外设置最大迭代次数否则模型偶尔抽风时会把流程带进死循环。这些细节都是需求拆解阶段要考虑的。2. LangGraph的核心心智模型状态图是地图Agent是司机LangGraph的学习曲线其实并不陡真正难的是切换心智模型。它本质上是把整个系统当成一张有向状态图来跑节点是Agent或普通函数边是流转逻辑状态是贯穿全程的共享数据容器。理解了这三个要素LangGraph就理解了一大半。2.1 State、Node、Edge三个要素如何搭出任意拓扑**State状态**是LangGraph里最重要的概念。它不是一个全局变量池而是一个严格定义的数据结构。每个节点执行完后返回部分字段的更新LangGraph会用Reducer合并进全局状态再传给下一个节点。from typing import TypedDict, Annotated from operator import add class AgentState(TypedDict): query: str research_report: str draft: str messages: Annotated[list, add] # 使用add合并器实现追加而非覆盖上面这段代码中messages字段用了Annotated[list, add]意味着每次节点返回新的消息列表时LangGraph会把新消息追加到已有列表而不是整体覆盖。如果不加这个合并器后一个节点返回的messages会把前一个节点写入的messages冲掉这是最常见的State踩坑点后面我会单独展开讲。**Node节点**是一个普通Python函数接收当前的State字典返回一个Dict表示要对State做哪些更新。节点的输入输出永远是State这保证了节点之间有统一的通信协议不同Agent之间不需要直接互相引用。def research_node(state: AgentState) - dict: question state[query] summary search_and_summarize(question) # 这里做实际的检索与摘要 return {research_report: summary, messages: [(assistant, f完成调研{question})]}**Edge边**定义节点之间的流转途径。LangGraph支持普通边和条件边普通边是确定的转移条件边是根据当前State的值动态决定去哪个节点。2.2 串行、分支、条件路由常见协作模式的图表达有了State、Node、Edge各种协作模式都能用图来表达串行流水线A节点做完传给BB做完给C。这是最基础的模式适合检索→分析→生成这类线性处理流程。扇出并行一个节点把任务拆成多个子任务同时交给多个处理节点。LangGraph里用SendAPI实现动态并行适合一篇文章拆成多段并行重写这类场景。条件路由节点执行后根据State的内容决定走哪条边比如审核Agent判断质量分是否合格合格则结束不合格则回到写作节点。带反馈的循环这是多Agent系统和普通链式调用最大的区别。写作Agent写完初稿后审核Agent发现质量不足触发写作Agent重写这种循环回路在LangGraph里不过是一条回到自身的条件边。2.3 为什么选LangGraph而不是自己写while循环我能理解很多人看到LangGraph的第一反应是这不就是一张图加一个循环调度器吗我自己写也不难。我也踩过这个坑——自己实现了一个调度器刚开始很爽但随着状态字段变多、分支变复杂我开始遇到三个棘手问题状态持久化流程跑到一半系统崩溃了怎么恢复自己写要在每个节点手动存快照LangGraph自带Checkpointer只需一行编译参数。可视化与调试LangGraph可以把编译后的图导出成结构图每个节点当前的状态、走过的路径都能看清楚比自己打印日志强不是一点半点。条件边的标准化自己写if-else时间长了就会混入各种业务逻辑LangGraph把图转移和业务逻辑拆开了路由函数只负责返回目标节点名业务逻辑留在节点里。所以说白了LangGraph不是用来解决Agent本身能力问题的它是帮你管理Agent之间的组织关系、状态流转和异常恢复的。这就是状态图的定位一张精确的地图让每个Agent司机知道自己现在在哪、要去哪、走哪条路。3. 从一张图到能跑的代码一个多智能体协作的具体实现理论说了不少现在走一遍完整代码。我用一个典型的调研写作审核三节点流程作为示例这种结构几乎适用于所有内容生产类任务。3.1 定义共享状态先定数据结构再写任何节点我在动手写节点前第一步永远是定State而且会把所有节点可能要共享的数据字段一次性列出来。示例场景中我需要一个用户问题、一份调研报告、一篇文章初稿、一个审核结果标志、以及全程对话记录的Message列表。from typing import TypedDict, Annotated from operator import add from langgraph.graph import StateGraph, START, END class WorkState(TypedDict): query: str research_report: str draft: str passed: bool messages: Annotated[list, add]这里的passed字段很关键审核节点的输出会写到这里条件路由节点靠它决定流程走向。定好State之后所有节点函数的签名就统一了输入是WorkState输出是dict。3.2 研究者与总结者两个节点如何接力第一个节点是调研Agent它的职责是根据用户问题做检索并把结果整理成结构化摘要。这里不限定具体模型我用一个函数占位方便替换成你自己的检索逻辑。from langchain_openai import ChatOpenAI model ChatOpenAI(modelgpt-4o-mini, temperature0) def research_node(state: WorkState) - dict: query state[query] prompt f基于{query}做一份主题调研输出1) 关键结论2) 支持数据3) 待确认风险点。 response model.invoke(prompt) return { research_report: response.content, messages: [(assistant, 调研Agent完成调研任务)], }第二个节点是写作Agent它读取调研报告输出文章初稿。注意这里写作节点只能看到research_report看不到搜索原始过程这正是把多Agent拆开的意义每个节点只需要关心自己那部分上下文。def writing_node(state: WorkState) - dict: report state[research_report] prompt f根据以下调研报告撰写一篇结构清晰的文章初稿\n{report} response model.invoke(prompt) return { draft: response.content, messages: [(assistant, 写作Agent完成初稿撰写)], }3.3 加入质量检查节点让流程带反馈回路写作节点之后接一个审核节点它的输出决定流程是顺利结束还是回炉重造。我用一个简单的质量评分函数模拟审核逻辑你也可以换成调LLM打分或规则校验。def review_node(state: WorkState) - dict: draft state[draft] # 这里做质量评估比如长度检查、关键信息覆盖率、事实一致性 score evaluate_quality(draft, state[research_report]) return {passed: score 0.8}有了passed字段后就可以写条件路由函数了。这个函数读取当前State返回下一步要去的节点名。返回END表示流程结束返回某个节点名则表示回到该节点重新执行。def route_after_review(state: WorkState) - str: if state[passed]: return end return writer # 重写 builder StateGraph(WorkState) builder.add_nodes([(researcher, research_node), (writer, writing_node), (reviewer, review_node)]) builder.add_edge(START, researcher) builder.add_edge(researcher, writer) builder.add_edge(writer, reviewer) builder.add_conditional_edges(reviewer, route_after_review, { end: END, writer: writer, })3.4 编译与调用invoke和stream的区别图定义好之后最后一步是编译并调用。LangGraph的compile()会检查整张图的连通性比如是否存在不可达节点、是否存在缺边的节点这些检查能在第一时间暴露图设计的错误。app builder.compile() result app.invoke({ query: LangGraph状态管理最佳实践, research_report: , draft: , passed: False, messages: [], })这里有两个调用方式需要区分invoke阻塞式调用直到整张图跑完返回最终State。适合离线批处理和一次性任务。stream流式调用可以按节点逐步返回中间结果。适合需要给用户展示当前Agent在做什么的场景比如前端实时显示进度条。for chunk in app.stream({query: ...}, stream_modeupdates): print(chunk) # 打印每个节点名称和它的输出我实际做项目时基本都用stream原因很简单多Agent流程时间很长用户没法等一个黑盒跑完。哪怕内部做得再好用户看不到中间进展就是会觉得卡住了。4. 真正上线后才会暴露的问题实测数据流与踩坑记录写一个能用起来的图不难难的是把图稳定地在生产环境里跑。这一节是我在几个实际任务处理流程中踩过的坑总结出来的每个坑都见过不止一次值得你提前规避。4.1 状态字段的脏写与覆盖第一个坑来自State合并器的理解错误。我加过不少任务给Agent比如让写作Agent直接把draft累加到历史记录里。如果draft字段定义成普通字符串那么每次节点返回draftLangGraph都会用新字符串覆盖旧字符串。这里有两个典型的错误把messages定义成普通list而非Annotated[list, add]导致后一个节点返回的Message列表覆盖了前一个节点的对话记录上下文丢失。在同一个节点里同时修改多个字段其中一个字段由别的节点维护导致本该保留的数据被意外清空。解决办法就一条State里每个字段都要想清楚谁来写、谁能写、是否允许追加。凡是需要跨节点累积的数据用Annotated[list, add]或自定义Reducer凡是只属于单个节点的中间量就不要放进State里放到节点内部或临时目录。4.2 超时、递归限制与重试策略第二个坑和稳定性有关。多Agent流程中一个Agent通常要调用多次大模型接口每次调用都有超时和限流的风险。我在某次并发压测中一个节点调用模型接口连续失败三次整个图直接卡住最后靠任务发版重启才恢复。后来我在两条线上加固给每个LLM调用加timeout参数并包一层重试装饰器重试间隔用指数退避。给LangGraph的递归迭代设置上限。默认情况下recurs_limit控制整个图最多执行多少步我的习惯是设置为流程所需最少步数的3到5倍防止模型决策异常进入死循环。app builder.compile(checkpointerMemorySaver()) config {recursion_limit: 20, configurable: {thread_id: session-1}} result app.invoke(initial_state, configconfig)另外如果你部署在多进程或多副本环境本地内存Checkpointer会失效。需要把状态存到Redis或PostgreSQLLangGraph有对应的持久化实现把MemorySaver换掉即可。4.3 输出质量不稳定路由决策的幻觉陷阱条件路由是最容易出现不可控问题的地方。我用过一次让LLM直接输出JSON格式来评判是否通过质量检查结果模型在完全没有事实依据的情况下给出了一个合理的通过判断把明显有硬伤的初稿放行了出去。这个问题的根源在于让模型做二元判断它很容易倾向性通过。从那之后我改成混合决策硬性规则优先比如初稿字数低于500字直接打回、关键数据未覆盖直接打回。模型只负责硬规则覆盖不到的主观质量评估且必须输出结构化评分而不是自由文本。class ReviewResult(BaseModel): score: int # 0-100 missing_points: list[str] confidence: float review_model ChatOpenAI(modelgpt-4o-mini).with_structured_output(ReviewResult)这样审核节点返回的是结构化字段路由函数按阈值判断即可。模型的自变量空间变小系统的确定性大幅提升。4.4 可观测性把每条路径都打到日志里最后一个坑是我自己之前不太重视直到有一次生产环境的流程突然变慢才发现没有可观测性排查多Agent系统几乎寸步难行。多Agent系统的调试难点在于同一个输入可能走完全不同的节点路径。今天这个输入走了A-B-C三跳明天同一个输入可能走了A-B-A-B-C五跳。没有日志你根本不知道它实际走了哪条路。我的做法是记录三个层面的信息节点级每个节点开始时间、结束时间、输入State的摘要、输出State的摘要。边级条件路由函数收到的状态快照、返回的目标节点名。Token级每个Agent调模型时消耗的输入输出Token数以及单次调用的耗时。LangGraph的stream本身就能给出节点级信息配合在路由函数里加日志基本能覆盖99%的排查场景。可视化调试可以用langgraph自带的图导出功能把整张图导出成图片哪个节点卡住一眼可见。5. 个人实际使用后的体会和建议如果你正准备用LangGraph搭多智能体系统我最想强调的一句话是先让流程跑通再谈优化。多Agent系统是一个变化很快的领域今天的模型能力明天可能就大幅升级你今天花一个月调优的细枝末节可能明天就不重要了。而状态图的骨架设计——哪个节点负责什么、数据怎么流转、失败怎么回退——这些才是真正值得投入时间的地方。另外提醒一点不要迷信Agent数量越多效果越好。一个写作任务用三到五个节点编排是合理的但如果你的流程需要十几个Agent协作建议先反思一下是不是子任务拆得过度了。拆得越多状态迁移的组合数越大调试成本越高最终收益往往不如用工具调用的单Agent方案。在做某次模拟工单系统时我最初设计了八个Agent最后压到四个流程反而更稳定。模型本身已经足够强多Agent的核心理由从来不是人多力量大而是不同角色之间需要清晰的责任边界和可治理的协作协议。这句话如果只能记住一句就记这句。