做过多智能体编排的人应该都有体会:单 Agent 写起来很爽,但一旦要让系统学会“找队友帮忙”“把活拆下去”“被打断之后还能接着干”,代码立刻就会变成一团乱麻。LangGraph 多智能体开发的核心,就是用“图”的思维把这团乱麻理顺:状态是团队共享的白板,节点是干活的员工,边是任务交接流程,条件路由则是“看情况让谁上”的调度规则。这篇文章把我从零构建多智能体核心架构的完整过程梳理出来,适合已经写过单个 Agent、想往上走一层做编排系统的开发者。你可以把它当成一份能直接抄作业的骨架,也可以当成排查架构问题时的手边参考。
1. 多智能体到底是怎么回事,以及为什么选择 LangGraph
1.1 多智能体不是“多个模型排队调用”
很多刚接触这个方向的人,对多智能体的理解就是“循环调用好几次大模型”。比如先让模型写方案,再把方案喂给模型写代码,最后把代码喂给模型做测试。这本质上只是单 Agent 的串行流水线,模型的每一步都在消费上一步的文本输出,中间没有真正的“分工”和“决策”。一旦某个环节的输出格式变了,后面所有环节都会跟着崩,排查起来极其痛苦。
真正的多智能体系统更像一个小团队:有一个调度主管负责判断当前任务该交给谁,有若干执行者各自负责一类具体工作,任务在执行者之间可以来回流转,而不是一条直线走到黑。比如一次对话进来,主管判断需要查资料,于是先交给检索类 Agent;检索结果回来后,主管再判断“资料够了,可以写代码了”,于是再交给代码类 Agent;写完之后如果发现还要补充数据,代码 Agent 可以再把问题抛回给检索 Agent。这种“有来有回”的协作形态,才是多智能体区别于普通多步调用的关键。
从工程落地的角度看,这种形态带来三个硬需求:第一,所有 Agent 要能访问一份共享的会话状态,否则互相之间只靠文本拼接,信息必然丢失;第二,流程必须支持循环和条件跳转,因为下一步做什么要由模型根据当前上下文动态决定;第三,每一步都要可观测、可中断、可恢复,否则一个错误发生在第三次循环时,你根本不知道从哪里开始排查。这三个需求,恰好就是 LangGraph 这类图编排框架要解决的基本盘。
1.2 LangGraph 解决的核心问题,一句话讲明白
LangGraph 的核心抽象其实只有四样东西:State、Node、Edge 和 Checkpointer。State 是全局共享的数据结构,描述当前对话“已经发生了什么”;Node 是一个普通函数,接收 State,返回 State 的部分更新;Edge 定义节点之间的流转关系,分为普通边和条件边;Checkpointer 负责把每一步的 State 快照保存下来,让系统具备长对话记忆和断点恢复能力。
如果你过去用过传统的链式编排,对比会更明显。链式结构是线性的,A 到 B 到 C,写完就完;图结构允许 B 执行完回到 A,再根据新状态决定是去 C 还是循环回 B。这就是多智能体最需要的“循环调度”能力。另外,LangGraph 内置了断点续跑和人工确认的机制,生产环境里非常实用的“关键操作先问人”这个需求,不用自己造轮子。
从选型角度说,市面上也有其他 Agent 编排框架,但 LangGraph 的优势在于它离底层足够近:没有把你封死在一套固定的“主管-执行者”模板里,节点怎么写、路由怎么定、状态怎么拆,全部由你掌控。这意味着你可以先搭一个最小的图跑通,再逐步往里面加节点、加工具、加记忆,架构始终是可控的。对于要上生产、要做长期维护的项目来说,这种“自由度”比开箱即用的产品更重要。下面我从环境搭建开始,把整个构建过程完整过一遍。
2. 从零搭环境:依赖安装、模型接入与目录规划
2.1 最小依赖清单,少装一个都会后悔
多智能体项目本质上是一个 Python 工程,环境准备没有太多花活,但依赖版本最好一步到位。我建议用虚拟环境隔离项目,避免和本机其他工程互相污染。新建一个项目目录后,安装下面这两个核心包就够了:
pip install langgraph langchainlanggraph 是这个项目的主角,负责状态管理和图执行;langchain 提供消息模型、工具封装和各类生态组件。装完之后可以顺手验证一下版本:
import langgraph print(langgraph.__version__)我个人的建议是 LangGraph 版本不要低于 0.2,因为较新版本里的interrupt机制、Command语义和断点 API 都更稳定,网上大多数教程也基于这个版本线。如果你是从旧项目升级上来的,尤其要注意版本变更带来的 API 差异,后面第 6 章会专门讲这些坑。
装包之外还有一件事要提前做:确认你的运行环境能正常访问需要用到的模型服务。这里不涉及具体厂商,只说原则——建议在环境变量或配置文件里统一管理密钥和模型地址,不要硬编码在代码里。工程上这叫“配置与代码分离”,换环境、换服务商时你只需要改配置,不用动业务逻辑。
2.2 模型接入:把核心逻辑和具体厂商解耦
多智能体里每个节点都要跟模型打交道,如果每个节点都直接去初始化自己的模型客户端,后期替换模型厂商时会改到怀疑人生。我的做法是写一个统一的模型加载函数,全工程只有这一个地方知道“模型是谁”:
def load_llm(): # 在这里初始化你的 ChatModel 实例。 # 无论哪家服务商,LangChain 框架都提供了对应的接入类, # 接口统一为:接收消息列表,返回一条 AIMessage。 # 示例:chat_model = SomeProviderChatModel(model="your-model-id") return chat_model这里刻意不写死某个厂商的类名,是因为不同服务的接入类名、初始化参数差异很大,而你真正需要的只是一个“输入消息列表、输出模型回复”的标准对象。只要你的代码里所有节点都依赖这个load_llm()函数,将来换模型就只改这一个函数,图结构和业务逻辑完全不用动。
实际项目中我还会做一层更细的区分:调度主管用一个模型,执行者用另一个模型。主管节点的输出是结构化的调度决策,对格式稳定性要求高;执行者节点要调用工具、写长文本,对推理能力要求更高。两个角色分开配置,既能按角色调优,也能更精确地控制成本。这个技巧等到第 3 章讲主管-执行者模式时还会再展开。
2.3 项目目录规划,先把家底列清楚
多智能体项目一旦节点变多,代码分布就会变得很重要。我推荐把状态定义、主管、执行者、工具、图装配分开管理,目录结构大致如下:
multi_agent_demo/ ├── agent_state.py # 状态定义,整个图共享的数据结构 ├── supervisor.py # 主管节点,负责路由决策 ├── workers.py # 执行者节点,负责具体任务 ├── tools.py # 工具注册与权限配置 ├── graph_builder.py # 图的装配与编译入口 └── main.py # 命令行入口或服务入口这个结构的好处是职责清晰:改状态定义时只动agent_state.py,加一个执行者时只动workers.py和graph_builder.py,不用在几百行的文件里找东找西。等系统再变大,还可以把每个 Agent 单独拆成一个包,每个包有自己的 tools 和 prompts。对于刚开始上手的人,先按这个五文件结构走,足够应对绝大多数场景。
3. 核心架构拆解:状态、节点、边,从零构建骨架
3.1 状态(State)是团队共用的白板
多智能体协作时,所有节点看到的必须是同一份“会议纪要”,这就是 State 存在的意义。LangGraph 里 State 就是一个 TypedDict,但字段的定义方式有讲究。最经典的字段是保存对话消息的messages,它必须用add_messages这个 reducer 标注:
from typing import TypedDict, Annotated from langgraph.graph.message import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] next_agent: str这里需要理解一个关键点:节点函数的返回值只是“状态的增量更新”,不是整体覆盖。如果没有add_messages,每次节点返回messages都会把整个消息列表替换掉,前面的对话记录就丢了;有了它,新消息会被自动“追加”到已有消息列表里。这正是多智能体最需要的协作语义——每个节点都像在白板上添加新内容,而不是擦掉别人写的东西。
Annotation这种标注方式第一次接触会觉得绕,但你只要记住一个经验:所有需要“累加、追加、合并更新”的字段,都要声明对应的 reducer;而像next_agent这种“每轮被重新决策”的字段,反而应该直接覆盖,不需要 reducer。状态字段的类型也会影响后续断点、记忆的可靠性,能用字符串、数字、消息对象这类可序列化类型时,就不要往 State 里塞不可序列化的对象,否则保存检查点时大概率会报错。
3.2 节点就是干活的员工,一个节点一个职责
在 LangGraph 里,节点就是一个普通 Python 函数:输入是当前 State,输出是一个字典,字典里的键是要更新的字段。下面是一个最简单的执行者节点:
def researcher(state: AgentState): # 执行者根据当前消息历史生成回复 resp = research_llm.invoke( [ {"role": "system", "content": "你是资料检索 Agent,负责查找事实、核对数据。"}, *state["messages"], ] ) # 只返回新增的消息,由 add_messages 自动追加到总状态 return {"messages": [resp]}很多人写节点时会犯一个错误:把“生成回复”和“图的流转逻辑”混在一起。节点内部动不动就自己判断要不要调用下一个节点,结果图的调度关系被藏在代码深处,改起来非常费劲。正确做法是节点只做一件事——基于当前状态干活,然后把结果写回状态;至于下一步去哪个节点,交给后面要讲的边和路由函数去决定。
工程上还有一个非常实用的习惯:每个节点都应该能脱离图单独运行。写节点时顺便写一个测试,手工构造一段 State 调用这个函数,看返回值是否符合预期。这样做的价值在第 5 章讲调试时会体现得特别明显——绝大多数多智能体问题,都能靠“单测节点 + 走查路由”来定位,而不是对着整张图瞎猜。
3.3 边与条件路由:谁能干谁上
普通边是固定的交接线,比如 A 干完一定去 B。条件边则是一段自定义逻辑,根据当前 State 动态决定下一步去哪个节点。这是多智能体调度最核心的机制。路由函数本身并不复杂:
def route(state: AgentState): next_agent = state.get("next_agent", "") if next_agent == "coder": return "coder" if next_agent == "researcher": return "researcher" return "done"这个函数的返回值会传给add_conditional_edges的映射表,映射表里每个 key 对应一个目标节点或END。这里有一个小细节很容易被忽略:路由函数必须有“兜底分支”。因为模型输出的内容是不确定的,哪怕你提示它只能返回三个选项,它偶尔也会给你冒出第四个值。没有兜底分支,运行时就会报出让人摸不着头脑的匹配错误。我的习惯是任何路由都默认返回"done"或END,宁可让流程提前结束,也不能让整个图崩溃。
3.4 主管-执行者模式的完整代码骨架
把前面的状态、节点、边组装起来,就是一套经典的“主管-执行者”多智能体架构。主管负责看当前对话,决定下一步交给谁;执行者干完活,把结果写回状态,流程再回到主管,由主管判断是否继续。
from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langgraph.checkpoint.memory import MemorySaver class AgentState(TypedDict): messages: Annotated[list, add_messages] next_agent: str def supervisor(state: AgentState): prompt = ( "你是调度主管。请判断当前任务需要哪种角色协助:" "coder(写代码)、researcher(查资料)," "如果任务已经完成或无需协助,返回 done。" "只输出一个JSON:{\"next\": \"coder\"}" ) resp = supervisor_llm.invoke([{"role": "system", "content": prompt}, *state["messages"]]) decision = extract_json_field(resp.content, "next") return {"next_agent": decision if decision in ("coder", "researcher") else "done"} def coder(state: AgentState): resp = coder_llm.invoke( [ {"role": "system", "content": "你是代码实现 Agent,负责编写和修改代码。"}, *state["messages"], ] ) return {"messages": [resp]} def researcher(state: AgentState): resp = research_llm.invoke( [ {"role": "system", "content": "你是资料检索 Agent,负责查证资料、整理数据。"}, *state["messages"], ] ) return {"messages": [resp]} def route(state: AgentState) -> Literal["coder", "researcher", "done"]: value = state.get("next_agent", "") if value in ("coder", "researcher"): return value return "done" builder = StateGraph(AgentState) builder.add_node("supervisor", supervisor) builder.add_node("coder", coder) builder.add_node("researcher", researcher) builder.add_edge(START, "supervisor") builder.add_conditional_edges( "supervisor", route, {"coder": "coder", "researcher": "researcher", "done": END}, ) builder.add_edge("coder", "supervisor") builder.add_edge("researcher", "supervisor") graph = builder.compile()执行一次完整对话时,流程是这样的:任务从 START 进入主管节点,主管用模型分析意图,把next_agent写进状态;条件边看到next_agent是coder,就把控制权交给代码 Agent;代码 Agent 干完活,通过普通边回到主管;主管再看一眼新状态,如果认为任务完成,就返回done,图在执行到 END 时收尾。
这 50 来行代码就是多智能体最核心的骨架。你不需要一开始就搞几十个节点,先把这套“主管-执行者”跑通,理解状态如何流转,再慢慢加执行者、加工具、加记忆,架构的底座不会变。
4. 关键功能实现:工具调用、长对话记忆与人工确认
4.1 让 Agent 真正“会动手”:工具注册与权限控制
一个只能“说话”的多智能体价值有限,真正生产可用的系统里,执行者必须能查库、调接口、写文档。LangChain 生态里注册工具的方式非常轻量,用装饰器把一个普通函数变成工具:
from langchain_core.tools import tool @tool def query_project_status(project_id: str) -> str: """查询某个项目的最新状态,返回一句话摘要。""" # 实际项目里这里会接业务数据库或内部接口 return f"项目 {project_id} 当前处于联调阶段,阻塞在接口字段对齐。" @tool def write_doc(title: str, content: str) -> str: """把指定内容写入团队文档。""" # 实际项目里这里会接文档服务 return f"已写入《{title}》"工具定义好之后,在执行者初始化时通过bind_tools挂到模型上:
coder_llm = load_llm().bind_tools([query_project_status, write_doc])这样模型在回答过程中如果发现需要查项目状态,就会在回复里携带一个“工具调用请求”,你的节点代码负责解析这个请求、执行对应函数、把结果作为新的消息放回状态里。
这里我要重点强调一个经验:多智能体场景下,工具权限一定要按角色分配,不要把所有工具一股脑绑给所有 Agent。比如write_doc这种有副作用、会改数据的工具,只应该绑定在少数被授权的执行者上;主管节点原则上不应该持有任何业务工具,它的职责是调度,不是干活。原因很简单——模型是会“自作主张”的,如果不做权限控制,主管可能在某个边缘 case 里顺手调用了一个写操作工具,后果很难收拾。权限最小化,是安全的多智能体系统与玩具 Demo 的分水岭。
4.2 长对话记忆:Checkpointer 的配置方法
第 3.4 节的骨架能跑,但没有记忆:每次invoke之后状态就丢了,下一次调用又是全新的对话。要让系统记住多轮上下文,关键是把 Checkpointer 接进图里:
from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() graph = builder.compile(checkpointer=memory)调用时,通过thread_id来区分不同的对话线程:
config = {"configurable": {"thread_id": "conversation-007"}} graph.invoke( {"messages": [{"role": "user", "content": "先帮我查一下项目 A 的进度。"}], "next_agent": ""}, config=config, ) # 第二轮继续问,同一个 thread_id 下系统记得上一轮的内容 graph.invoke( {"messages": [{"role": "user", "content": "既然还在联调,帮我把阻塞问题列出来。"}], "config": config} )理解 Checkpointer 的关键在于“快照”这个词:它不是简单缓存一段文本,而是把每一步执行后的整个 State(包括消息列表、路由结果、中间变量)保存下来。因此同一个thread_id继续对话,主管能“想起”之前做过什么决策,执行者也能看到自己上一轮写了什么。换一个thread_id,就是完全独立的新会话。
我见过不少人在这一步踩坑:代码里明明配了 Checkpointer,但多轮对话还是“失忆”。排查下来发现是每次调用都重新生成一个随机thread_id,或者压根没传 config。记住一个口诀:配置了 Checkpointer 不等于有了记忆,thread_id稳定才是记忆生效的前提。另外,MemorySaver只是进程内存方案,适合开发和单机 Demo;生产环境建议换成基于数据库的持久化方案,避免服务重启后所有对话全部归零。
4.3 人工确认机制:关键步骤要踩一脚刹车
多智能体系统最大的风险就是把一个“有点能力但偶尔犯浑”的模型放进了自动化流程里。尤其涉及写文件、发消息、扣费这类操作,绝对不能让它全自动执行。LangGraph 提供了中断机制,可以在关键节点前把控制权交还给人类确认。
from langgraph.types import interrupt def before_write(state: AgentState): # 执行到这里时,图会暂停并等待外部输入 decision = interrupt( {"doc_title": state["next_doc_title"], "doc_content": state["next_doc_content"]} ) if not decision["confirmed"]: return {"messages": [{"role": "assistant", "content": "用户取消了写入文档操作。"}]} return {"messages": [{"role": "assistant", "content": "用户已确认,继续执行。"}]}配合编译时的interrupt_before参数,可以指定在哪个节点之前暂停:
graph = builder.compile( checkpointer=memory, interrupt_before=["before_write"], )当流程走到before_write时,graph.invoke会返回一个待确认状态,你的服务端可以把待确认内容推给前端,等用户点“同意”或“拒绝”之后,再带着结果继续调用。这个机制对生产系统的价值是决定性的。没有它,多智能体只是“自动化的模型玩具”;有了它,系统才敢把真实业务动作交给 Agent 去推进,同时也保留了人在回路里的最终控制权。
5. 调试与可视化:看到图结构,才能定位坏链路
5.1 先看拓扑,再谈优化
多智能体项目一旦节点超过五六个,光靠脑子记住谁连谁已经不现实了。LangGraph 编译出来的对象自带图结构信息,最简单的方式是打印一棵 ASCII 结构树:
graph.get_graph().print_ascii()输出会以缩进树的形式展示所有节点和边的连接关系,我每次加完新节点都会立刻打印一次,确认拓扑符合预期。如果你配好了图片渲染环境,还可以用get_graph()自带的绘制方法把图导出成 PNG 文件,贴在方案文档里做评审。重点看两件事:一是START正确指向了入口节点,二是条件路由的所有目标都在映射表里。
不过要提醒一句:静态图能看到的是“可能的分支”,看不到运行时模型实际选择了哪条路。所以可视化只是第一步,真正的调试要靠下面的动态追踪。
5.2 流式输出与逐步追踪:看清每一步发生了什么
多智能体出错时,最常见的问题是“不知道哪一步开始跑偏”。解决办法是开启逐步流式输出,把每一步执行到的节点名称和状态更新打出来:
for event in graph.stream(input_dict, config=config, stream_mode="updates"): print(event)stream_mode="updates"返回的是每个节点的增量更新,你能清楚地看到:主管先输出了什么决策,然后哪个执行者被触发,执行者返回了怎样的消息,流程又回到了哪里。这就像给一次多智能体协作拍了慢动作录像,跑偏发生在哪个节点一目了然。
如果嫌流式输出信息太碎,也可以在节点函数里手动加日志。我的习惯是每个节点入口和出口各打一行关键信息,比如“进入 coder”“coder 返回,耗时 3.2 秒,消息长度 400 字”。这些日志在单体调试时看着多余,但在多智能体协作出现问题时,就是保命的线索。
5.3 用“单点走查法”定位坏链路
这是我在实际项目里总结出来的一套排查心法:一旦发现多智能体的最终回复不对,不要拿整条链路去猜,而是从最终结果反推,逐个节点单独验证。先把执行者节点抽出来,手工构造输入 State 调用一次,看它的输出是不是对的;再单独调主管节点,看在给定上下文时它做出的调度决策是否符合预期。哪个节点的输出与预期不符,问题就定位在哪个节点。
这套方法还有个额外好处,就是你顺手给每个节点积累了一批测试用例。改动 prompt 或模型版本后,把这些用例跑一遍,能快速发现“是不是这次改动引入了回归”。我一直认为,多智能体项目的工程质量不取决于框架多高级,而取决于你把关键节点测试得有多细。主管节点的路由测试、执行者的工具调用测试、异常输入的兜底测试,这三类测试是底线。
6. 常见问题与排查实录
6.1 高频报错速查表
下面这些问题,是我和身边同事在多智能体开发中反复遇到的高频坑。整理成一张速查表,排查时先对照一遍,往往能省下大量时间:
| 报错或异常现象 | 几乎可以锁定的原因 | 处理方式 |
|---|---|---|
| 多轮对话“失忆” | Checkpointer 没接,或 thread_id 每次不同 | 检查 compile 是否传了 checkpointer,调用时是否传稳定 thread_id |
| 状态里消息被覆盖、历史丢失 | messages 字段没用 add_messages reducer | 给字段补上 Annotated[list, add_messages] |
| 节点返回了未定义的字段报错 | 返回字典的 key 不在 State 定义中 | 先扩展 State 定义,再在节点里返回 |
| 图永不结束、卡在循环里 | 路由函数缺少兜底分支,或主管反复决策同一执行者 | 增加循环次数上限,路由默认走 done,必要时加“已完成标志”字段 |
| 模型没执行工具调用 | 工具没 bind 到对应模型的实例上 | 确认执行者用的是 bind_tools 之后的模型句柄 |
| interrupt 后无法恢复 | 高版本 API 用法不兼容旧教程 | 确认 LangGraph 版本,按当前版本文档写 resume 流程 |
| 断点保存失败 | State 里有不可序列化对象 | 把状态字段收敛为字符串、数字、字典等可序列化类型 |
这张表只是入口,真正的排查还得配合第 5 章的追踪手段来做。但 80% 的“疑难杂症”,根因都在状态定义、路由兜底和 thread_id 这三件事上。
6.2 我踩过的几个值得说说的坑
第一个坑是把所有工具挂给所有 Agent。早期做一个模拟项目时,图里的主管、代码 Agent、检索 Agent 共用同一份工具列表,结果某次主管在判断“要不要写文档”的场景里,真的自己调用了写文档工具,直接把内容的初稿写进了线上文档。那次之后我把工具权限彻底按角色隔离,并加了一条铁律:主管不持有任何业务工具,只能持有“查询类”的只读工具,且必须有日志。
第二个坑是状态里塞了不可序列化的大对象。为了让执行者能拿到一个复杂配置,我直接把配置对象放进了 State,结果接 Checkpointer 之后频繁报序列化错误,而且报错信息藏在深层依赖里,非常难找。后来统一改成在节点内部按 key 加载配置,State 里只存配置 ID,问题立刻消失。多智能体的状态一定要“瘦”,只保存必要的信息,其余的东西让节点自己去取。
第三个坑是路由函数写得太“自信”。早期路由逻辑只处理了预期的两个分支,其余一律交给模型输出的原始值去匹配映射表,结果某次模型输出里带了个换行符,匹配直接失败,整张图报错。从那以后每个路由函数我都强制加done兜底,并且对模型输出做一次清理和校验。这个习惯看起来不起眼,但正是它把系统的偶发崩溃率降到了接近零。
第四个坑和调试相关:我一度过度依赖可视化面板,忽略了最原始的打点日志。有一次系统表现时好时坏,静态图看不出任何问题,我对着可视化结果分析半天也没有头绪。最后逼着自己给每个节点加了入口出口日志,跑了二十轮对话,才发现问题出在一个工具在特定输入下会返回空字符串,导致下游节点拿到了空上下文。这个教训让我明白:可视化解决的是“结构对不对”,日志解决的是“执行时发生了什么”,两者不能互相替代。
收尾之前,再分享一个我个人的小习惯:每往图里加一个新节点,先单独测试它,再连图跑集成;每改一次主管的 prompt,先手工构造 5 到 10 条典型对话,验证路由决策是否稳定。这套流程看起来慢,但长期坚持下来,多智能体项目最头疼的不稳定性问题,会被提前拦掉一大半。架构的骨架只是起点,真正让系统可用的,是这些发生在每一处细节上的克制与检查。