★ FEATURED ARTICLE 做多智能体开发最难的其实不是让单个智能体变聪明而是让一群智能体在同一张全局状态下不打架、不空转、不互相覆盖结果。LangGraph 多智能体开发这两年之所以火正是因为它把“一堆智能体如何协作”这个问题变成了“如何在一张有向图里编排节点和边”的问题。我试过用自定义循环、消息队列、异步任务去拼多智能体最后发现都不如直接在图上做状态流转来得清晰。本文会从核心架构开始一步步带着你用 LangGraph 从零构建一个“主编调研写作”的多智能体系统讲清楚状态、节点、条件边、路由这些关键概念并把我踩过的坑也一并列出来。这篇内容适合两类人一类是已经跑通过单 Agent 对话想进一步做多角色协作的开发者另一类是刚接触 Agent 编排希望搞清楚 LangGraph 到底怎么组织代码的新手。无论你是哪种我建议先把“状态是全局共享的”这句话刻在脑子里后面所有设计都围绕它展开。1. 做多智能体之前先把 LangGraph 的边界搞清楚1.1 单智能体的瓶颈在哪里单个智能体本质上是一个反复循环感知输入、调用模型、决定是否调用工具、再把工具结果交给模型继续推理。这种 ReAct 模式在任务单一的时候很好用但一旦任务被拆成“先调研、再规划、再写作、最后校对”让同一个智能体干完所有事就会出现三个问题。第一是 Prompt 越来越长。为了让同一个智能体既懂搜索又懂代码生成还懂文案优化你会把各种指令、示例、边界条件塞进同一个系统提示词里结果模型经常在小事上纠结主任务反而被稀释。第二是上下文窗口浪费严重。一个环节产生的中间结果会被原封不动地带到下一个环节token 翻倍涨而真正有价值的信息可能只有几句话。第三是错误定位困难。整个链路过长一旦某个步骤出错你不知道是工具问题、模型问题还是 Prompt 问题因为所有逻辑都混在一段循环里。在实践中单智能体还有一个容易被忽略的问题它没有“分工意识”。一个负责严谨调研的智能体通常不适合同时承担“快速生成吸睛标题”的角色。你要让它切换风格只能靠改变 Prompt但这种切换是不可控的。与其这样不如把不同能力拆成不同的智能体让每个智能体只专注一件事再通过一个编排者决定当前应该派谁上场。这就是多智能体架构出现的最直接动机。1.2 LangGraph 到底解决了什么LangGraph 是 LangChain 生态里的一个编排框架它把智能体的运行过程建模成一张有向图。图中的节点是具体的计算单元比如“调研智能体”“写作智能体”“工具调用函数”边是控制权的转移路线代表“下一步该轮到谁”。整个图的运行有一个全局共享的状态对象每个节点执行时读取状态、写入状态然后由条件边决定下一步走向。这套设计和我之前用代码手写 while 循环调智能体的做法有很大区别。手写循环时状态靠全局变量或字典传来传去节点之间的调用关系是硬编码在函数内部的每新增一个智能体就要改一次主流程。用 LangGraph 后节点与节点之间是解耦的你只需要把节点注册到图里再声明哪些边需要条件路由运行时框架会按图结构自动推进。加新节点不用动老节点这是它最实用的价值。更关键的是LangGraph 内置了状态管理和持久化机制。它可以记录每一步的执行历史支持把图的状态保存到数据库也支持让执行从某个断点恢复。这些能力看似基础但做多智能体系统时几乎是刚需。你需要知道“上一次运行到哪个智能体”“某个智能体当时看到了什么输入”否则整个系统就像一个没有日志的多线程程序出了问题完全无从排查。1.3 编排框架的选型思考我之前也给多智能体系统选过型确实不需要一上来就上 LangGraph。如果只是两个固定步骤的先后调用用普通代码顺序执行反而更快如果只有少量智能体、想快速验证直接用高层 Agent API 也能跑。但一旦你需要“多个智能体动态竞争执行顺序、共享状态、支持断点续跑”图编排的优势就会非常明显。我的建议是先画一张草图把“有哪些节点”“每个节点输入输出什么”“条件分支有哪些”“起点和终点在哪”画明白再去决定用不用 LangGraph。如果这张图里存在循环、分支、回退LangGraph 就很合适如果只是一条直线流水线那用普通的链式调用就足够了。避免为了技术而技术这是我在几个项目里踩过很多次坑后的真实体会。2. 核心架构设计把“多智能体”翻译成一张图2.1 状态就是所有智能体共用的那块黑板LangGraph 里的状态是一个 TypedDict 或 Pydantic 模型它在整张图运行期间全局共享。你可以把它想象成几个工程师共用一块黑板每个人完成自己的工作后把结果写到黑板上下一个人读取黑板上的内容继续干活。黑板上常见的内容包括消息列表、当前任务描述、某个阶段的产出草稿、最终结果等。值得特别强调的是 reducer 的作用。比如你定义了一个字段messages它接收的是一个消息列表正常情况下新节点返回值会直接覆盖旧值但你会希望新旧消息是拼接在一起的而不是互相覆盖。这时就要用 LangGraph 提供的add_messagesreducer它告诉框架“把新消息追加到旧消息后面”。不理解这一点很容易写出一个“每跑一步就把上一步结果清空”的图。设计状态时要克制。我见过不少同学把几十个字段一股脑塞进状态最后根本分不清某个字段是哪个节点负责更新的。更合理的方式是只放“节点之间必须传递的信息”需要跨节点共享的任务描述、阶段产物、执行标记、消息历史。纯粹只在某个节点内部用的临时变量不要放进全局状态否则只会增加调试复杂度。2.2 节点和边一个节点只干一件事一个节点在 LangGraph 里就是一个普通函数入参是当前状态出参是你要更新的状态片段。它内部可以调用模型、调工具、做计算但最好保持职责单一。一个节点如果能拆成“调研”和“整合调研结果”两个节点就不要硬塞在一个节点里这样既方便复用也方便出问题时单独测试某一段逻辑。边的类型分两种静态边和条件边。静态边表示无条件地从前一个节点走到下一个节点比如“调研完成后回到主编节点”条件边则由一个路由函数根据当前状态动态决定下一跳。条件边是多智能体系统的灵魂因为它负责回答“现在轮到谁”的问题。实现条件边时你需要保证路由函数返回的值和构建图时传入的路由表里的键完全一致否则 LangGraph 会直接报错。我还习惯给每个节点写清晰的 docstring并在节点函数内部用日志把关键决策打出来。因为多智能体系统一旦跑起来执行路径是动态的如果节点自己不记录“我为什么做了这个决定”事后只能靠猜。哪怕只是简单的print对调试的帮助都比我之前预想的大得多。2.3 三种主流协作拓扑的取舍多智能体的协作拓扑没有银弹我把它分成三类来讨论。第一种是 Supervisor 模式也是最常见的一个主编智能体负责接收任务、拆解步骤、分发子任务给 Worker 智能体再从 Worker 的返回结果里总结收敛。这种模式适合“有一个明确负责人”的场景比如内容创作、数据分析报告生成。优点是流程清晰、容易控制缺点是主编节点会成为瓶颈任务一多它就要频繁做路由决策。第二种是 Handoff 模式也叫移交模式智能体之间可以直接把控制权移交给对方而不是每次都要经过主编。比如客服智能体发现用户问的是技术问题直接主动切换给技术支持智能体。这种模式更适合“智能体之间平级协作”的场景但需要每个智能体都具备判断“该不该移交”的能力模型调用次数和编写复杂度都会上升。第三种是层级式或混合式相当于把 Supervisor 叠多层或者把多个 Supervisor 节点组成子图。适合业务分工很重的场景比如“安全审核团队”和“内容生成团队”各有一个主编两个团队之间通过汇总节点沟通。不过层级越多状态管理和调试成本也越高我建议先从单层 Supervisor 模式入手跑通后再往复杂拓扑演进。2.4 路由逻辑conditional edge 才是灵魂条件边表面上是一个简单的 if-else 函数但真正难的是“路由决策的依据从哪里来”。我见过最粗糙的写法是主编节点用大模型输出一个 JSON里面写{next_agent: writer}然后路由函数直接解读这个字段。这个方案能用但大模型偶尔会输出无效 JSON 或非法值导致整条边判断失败。更稳的方案是让大模型做结构化输出。你可以用 Pydantic 定义一个枚举类约束模型只返回research、write、finish这几个值这样解析基本不会出错。也就是说路由函数本身只做机械判断真正聪明的路由决策逻辑放在主编节点的 Prompt 和输出约束里。把“智能”留在模型里把“纪律”留在代码里是 LangGraph 开发里一条非常重要的原则。此外路由函数里一定要有兜底逻辑。比如模型输出异常或字段缺失时默认走research再试一次或者直接走finish结束流程。宁可让结果质量稍微打折也不能让整个图因为一个非法值卡死在中间。3. 从零构建一套“主编调研写作”的多智能体项目3.1 环境和依赖准备我采用的示例是一个模拟项目 X系统里有一个主编智能体Supervisor、一个调研智能体Researcher、一个写作智能体Writer。主编负责拆解任务并分发给调研或写作调研智能体负责查询一些“结构化资料”写作智能体根据资料生成最终文案。整个流程是典型的 Supervisor 模式。环境方面建议用 Python 3.10 以上版本并单独建一个虚拟环境。需要装的库很简单langgraph负责图的运行langchain-core提供消息和大模型抽象另外根据你实际使用的 LLM 服务商 SDK 来装对应的适配包。我通常会把依赖锁定到一个 requirements.txt 里因为 LangGraph 版本迭代很快接口变动也多不锁版本很容易出现“教程是这么写的我的环境却报错了”的情况。下面的代码都会围绕一个get_llm()函数来获取大模型实例你可以把它替换成你熟悉的服务商实现。这里用一个占位函数避免把密钥和厂商信息写死在代码里from langchain_core.language_models import BaseChatModel def get_llm() - BaseChatModel: # 替换为你自己的 LLM 实例构造逻辑 # 例如 langchain_xxx 里的 ChatXXX(modelyour-model) raise NotImplementedError(请在这里接入你的大模型服务)3.2 定义全局状态这个示例的状态需要保存四样东西完整消息历史、当前任务描述、调研结果、写作草稿。我用Annotated和add_messages来确保消息字段是追加而不是覆盖from typing import Annotated, TypedDict from langgraph.graph.message import add_messages from langchain_core.messages import AIMessage, HumanMessage, SystemMessage class AgentState(TypedDict): task: str messages: Annotated[list, add_messages] research_result: str draft: str next: str也许你会问为什么有了messages字段还要单独放research_result和draft这是因为messages是给模型看的对话历史而research_result和draft是给图里不同节点直接读取的强类型字段。如果让写作节点从消息历史里翻找调研结果既浪费 token 又容易翻错把它放在独立字段里节点就能“一眼看到”自己要的输入。多智能体系统越复杂这种“结构化传递”就越重要。3.3 实现两个 Worker 节点调研节点的职责是产出结构化资料。为了不依赖外部服务我这里用一个模拟搜索函数它返回一段固定的内容。你在真实项目里可以把这段逻辑替换为调用搜索 API、读取数据库或请求内部文档系统def mock_research(task: str) - str: # 真实项目中替换为联网搜索或知识库查询 return f关于“{task}”的调研结论核心用户关注易用性、稳定性与部署成本竞品差异点在可视化能力和生态集成。调研节点本身很简单读取task字段调用模拟搜索把结果写到research_result同时追加一条 AIMessage 到消息历史方便主编看到它确实干完活了def researcher_node(state: AgentState) - dict: print( Researcher 开始调研) result mock_research(state[task]) return { research_result: result, messages: [AIMessage(contentf调研完成{result})] }写作节点则负责根据调研结果生成最终草稿。它不关心主编怎么分配任务只关心全局状态里有没有research_result。有就用没有就提示缺少材料。这样设计的好处是职责清晰如果草稿质量不佳你不需要怀疑调研环节的问题只需要单独调试写作节点。def writer_node(state: AgentState) - dict: print( Writer 开始写作) material state.get(research_result) or 暂无调研材料请先补充。 prompt ( 你是一名文案写作助手。请根据以下调研材料写一篇逻辑清晰、面向产品经理的介绍文案。\n\n f调研材料{material}\n\n目标输出正文草稿。 ) response get_llm().invoke([SystemMessage(contentprompt)]) return { draft: response.content, messages: [AIMessage(contentresponse.content)] }3.4 实现 Supervisor 路由节点主编节点是整个图的核心它负责读取当前进展决定下一步是把任务交给调研、交给写作还是结束流程。为了避免模型乱说话我让它输出一个固定枚举值。同时如果draft已经存在就直接结束防止主编反复把任务分给写作节点导致死循环。def supervisor_node(state: AgentState) - dict: print( Supervisor 收到消息准备分发) # 已经写完了直接结束 if state.get(draft): return {next: finish} prompt ( 你是主编负责调度成员。当前任务{task}\n 环节状态\n - 是否已有调研结果{has_research}\n - 是否已有写作草稿{has_draft}\n 请从以下动作中选择一个只能输出一个词\n - research需要先做调研\n - write可以直接开始写作\n - finish全部完成可以结束\n ) llm get_llm() response llm.invoke([ SystemMessage(contentprompt.format( taskstate[task], has_research是 if state.get(research_result) else 否, has_draft是 if state.get(draft) else 否 )), HumanMessage(content下一步动作是什么) ]) decision response.content.strip().lower() # 简单兜底 if decision not in {research, write, finish}: decision research print(f Supervisor 决定{decision}) return {next: decision}这里有个小细节我把“是否有调研结果”作为上下文传给主编模型而不是让主编自己去历史消息里找。这样模型的判断负担小输出也更稳定。毕竟主编的核心能力是调度而不是每次都要做全文阅读理解。3.5 组装图并跑起来现在把节点注册到StateGraph里并声明边。从START进入主编主编通过条件边分给调研或写作调研节点完成后无条件回到主编写作节点完成后也无条件回到主编主编看到草稿后走finish分支到达END。from langgraph.graph import StateGraph, START, END from typing import Literal def router_after_supervisor(state: AgentState) - Literal[researcher, writer, finish]: decision state.get(next, research) if decision write: return writer if decision finish: return finish return researcher builder StateGraph(AgentState) builder.add_node(supervisor, supervisor_node) builder.add_node(researcher, researcher_node) builder.add_node(writer, writer_node) builder.add_edge(START, supervisor) builder.add_conditional_edges( supervisor, router_after_supervisor, { researcher: researcher, writer: writer, finish: END, } ) builder.add_edge(researcher, supervisor) builder.add_edge(writer, supervisor) graph builder.compile()跑起来只要一行invoke入参是初始状态。如果你担心极端情况下死循环可以在配置里设置recursion_limit我示例里给了一个较大的值result graph.invoke( {task: 调研竞品A并写一篇产品介绍, messages: []}, config{recursion_limit: 20} ) print( 最终草稿 ) print(result[draft])这一段看似简单其实已经把 LangGraph 多智能体开发的核心架构全走了一遍状态定义、节点实现、条件边、图编译、运行时调用。如果你之前没有接触过图编排我建议先认真读一遍router_after_supervisor这个函数它就是“由谁决定下一个执行者”这个问题的具象化。3.6 流式观察执行过程invoke适合拿最终结果但开发阶段我更推荐用stream它可以逐步吐出每一个节点的输入输出。用流式输出你能清楚看到主编先选了research调研节点跑了多久回来后主编又选了write最后主编看到draft选择finish。这对排查执行路径帮助巨大。for event in graph.stream( {task: 调研竞品A并写一篇产品介绍, messages: []}, config{recursion_limit: 20} ): print(event)每次流式返回的事件里键是节点名值是该节点返回的状态片段。你不需要去猜“系统现在跑到哪一步了”因为图上每个节点的执行顺序都被完整记录了下来。配合 LangGraph 自带的日志和追踪面板多智能体系统的透明度和可观测性比手写循环高一个量级。4. 进阶状态版本、持久化与并行派发4.1 Checkpointer 与时间旅行LangGraph 允许你在编译时传入一个 Checkpointer它会把每一步执行后的状态保存下来。这个能力有什么用最直接的好处是断点续跑某一次运行中途出错你不用从头再来可以从最后一步继续。另一个好处是“时间旅行”你可以把状态恢复到之前的某个 checkpoint改变输入后再分出一个新分支运行。我自己的经验是Checkpointer 对多智能体系统尤其重要因为多智能体一次执行涉及多个节点中间任何一步失败都有可能导致整张图返回空结果。如果有了 checkpoint至少能把“已经完成的调研结果”留着只重跑后面失败的写作环节。选择持久化方案时本地开发用内存版本就够了生产环境则建议用数据库方案。配置的时候注意一旦启用 Checkpointergraph.invoke的每一次调用就必须带一个线程级配置用来区分这是哪一次会话的执行记录。这也是不少老版本迁移时最容易踩的坑。4.2 Reducer 的更多玩法前面提到的add_messages只是 reducer 最基础的一种。实际上你还可以自定义 reducer 实现更复杂的合并策略。比如你想统计每个智能体的调用次数可以定义一个Counter状态字段reducer 里做数字相加你想记录本轮执行过的节点列表也可以自定义 reducer 做列表合并。写自定义 reducer 时要记住一个原则reducer 是注册在状态字段上的它决定了“新返回值如何合并进旧状态”。因此不要期待 reducer 能感知整个图的上下文它只负责字段级别的合并。更复杂的状态一致性逻辑应该放到节点函数里自己处理。有一个反直觉的点字段越少用 reducer整体行为越可控。因为 reducer 一旦存在就表示多个节点可以并发或先后更新同一个字段这会给调试增加认知负担。如果只是某两个节点顺序写同一个字段直接用覆盖型字段反而更简单。不要为了炫技而过度使用 reducer。4.3 并行派发一个主编带多个同角色 WorkerSupervisor 模式把任务串行跑没问题但如果一次要调研 5 个不同方向串行执行会很慢。LangGraph 里有一个专门用于并行分支的机制可以在一个步骤里同时派发多个任务副本。它比直接开多个线程更干净因为每个分支都有独立的内部状态同时共享全局状态的上游输入。实际实现时主编节点可以选择生成一个“待派发任务列表”然后通过专门的并行分发方式把每个任务作为一个子运行单元发给同一个 Worker 节点。每个子运行单元会执行相同的节点逻辑只是输入不同。等所有子运行单元都返回后再进入汇总节点做整合。并行派发看起来高级但要注意公共资源的并发安全。如果多个 Worker 同时往同一个外部系统写入数据就可能出现并发覆盖或限流。我建议先在状态里设计好“每个 Worker 的结果要放进独立字段”汇总节点再统一合并不要在 Worker 内部直接修改全局黑板上的共享结构。4.4 上下文管理不要让所有智能体共享完整历史多智能体最容易犯的错是想省事把完整对话历史一股脑传给每个节点。对于单个 Agent 这没问题但多智能体里不同节点关注的历史完全不同。主编关心的不是调研结果的全文而是进度和下一步写作节点关心的则是调研结果的正文。如果每次路由都把全量历史塞给模型token 开销会迅速失控而且模型容易被无关信息干扰。我的做法是给每个节点定制“视图”。状态里保存完整历史但节点函数在调用大模型之前先裁剪出自己关注的部分。写作节点只需要读research_result主编节点只需要读“是否已有某某字段”的标签。这样既保留了全局状态的完整性又控制住了每个节点的输入规模。这条经验非常实际。我在一个模拟项目中把上下文缩短后同样的任务在成本和响应速度上都有很明显的改善而且模型输出质量反而更稳定了。多智能体不是信息越多越好而是信息适得其所才好。5. 开发中的高频问题与排查经验5.1 跑着跑着进入死循环多智能体死循环几乎是每个新手都会遇到的第一个大问题。表现是图在几个节点之间无限来回直到触发recursion_limit报错。根本原因一般是路由条件写得不够收敛比如主编节点拿到 Worker 的输出后不知道该不该结束于是反复把同一个 Worker 派出去。要解决这个问题首先要让节点能明确感知“里程碑进展”。状态里要有类似draft这样的字段节点决策时检查它其次要在路由函数里加最大轮数兜底最后也是最有效的就是给主线画一个“终点”只有在指定字段满足条件时才允许返回finish。宁可让流程提前结束也不能让它无限循环。我在调试时还会定期打印每个节点访问次数。如果某个节点访问次数超过预期基本可以断定路由逻辑存在非收敛路径。这种问题在单线程代码里不容易看出来但在多智能体图里因为执行路径动态生成必要的时候要把“访问计数”加入状态字段作为显式判断条件。5.2 Worker 节点不按预期返回Worker 节点不按预期返回常见原因有三个模型输出格式不对、节点抛异常后整条图中断、节点更新了错误的状态字段。针对模型输出问题我强烈建议给 Worker 的返回做结构化约束而不是让它自由发挥。比如写作节点要返回正文你可以让它只返回文本不要顺便写一堆解释性的话。节点内部的异常处理同样重要。多智能体系统里一个节点如果抛出异常默认会中断整张图的运行。我建议在节点函数内部用try/except捕获可预期的异常并把错误信息转成状态字段里的错误标记而不是直接抛出去。这样主编节点就可以根据错误标记决定是重试还是终止。更新状态字段时最常见的坑是类型不匹配。比如状态里声明了research_result: str节点返回了None或一个字典都会导致校验失败。开发阶段我会给状态字段做严格的类型注解让 LangGraph 在写入时就能拦截大部分低级错误。5.3 工具调用和状态写入的边界多智能体里的工具调用不止是“模型调一个外部函数”它往往发生在某个 Worker 节点内部。因此工具调用后返回的结果究竟应该直接作为节点输出写进状态还是先经过一轮模型总结再写进状态会直接影响后续节点的输入质量。我的经验是原始工具结果不要原封不动地塞给下一个模型。比如搜索接口会返回一堆带标签和 URL 的原始内容写作节点拿到这些内容很可能被噪声干扰。更好的方式是先由当前 Worker 的模型对工具结果做一次摘要只把摘要写进状态。虽然这次多模型调用会多花一点时间但后续节点的输出质量会明显提升。同时工具调用必须考虑失败重试。网络抖动、接口限流都是家常便饭。建议封装一层带重试和超时的工具函数确保节点在调用工具时不会因为一次临时错误导致整张图崩溃。这个细节在单 Agent 里影响有限在多智能体里就会因为节点多、调用次数多而被放大。5.4 关于版本锁定和环境可复现LangGraph 这门技术迭代速度非常快我遇到过两次升级之后 API 变了原来还能跑的项目突然报错。所以我会在项目根目录放一份依赖锁定文件把langgraph及其相关包的版本固定到具体版本号。这样无论过多久按同一份文件安装至少能跑出一个确定性结果。除此之外最好把大模型的参数也固化下来比如温度、最大 token、模型名称。我发现多智能体系统对稳定性要求远高于单 Agent因为一个节点的不确定性会被后续节点放大。你可以每次手动调参但团队协作时最好把模型参数和 Prompt 一起纳入版本管理每次改动都过一遍评审。最后分享一个我个人的使用习惯在接手任何一个 LangGraph 项目时我第一件事不是看代码而是先运行一遍所有节点函数用已知输入验证单个节点行为。节点测试通过再跑整张图。这样我才能把“节点自身的 bug”和“图编排的 bug”分离开。多智能体开发踩坑并不丢人丢人的是不知道坑在哪一层。
阅读完成 · 觉得有帮助？
