LangGraph 做 Multi-Agent,其实没有你想的那么玄乎,但也不是简单地调几个库就能搞定。最近我把几个真实的 Multi-Agent 项目用 LangGraph 重构了一遍,踩了一圈坑之后,想把这套框架的底层逻辑、典型案例和实操细节完整拆出来。这篇文章不端着,直接以案例为主线,从架构设计到踩坑记录,给你一条可以照抄的落地路径。如果你是第一次接触 LangGraph 和 Multi-Agent,或者已经写过几个 Agent 但总感觉协作逻辑拧巴,那这篇应该能帮到你。
LangGraph 本质上是一个基于图结构的 Agent 编排框架,它把传统的 Agent 循环(调用 LLM -> 执行工具 -> 再调用 LLM)拆成节点、边和状态三个核心要素。多智能体(Multi-Agent)在这个框架里不再是简单地把多个 Agent 塞在一起,而是通过图拓扑去定义它们之间的通信、路由和共享状态。这就解决了一个以前很头疼的问题:多个 Agent 各自为战、互相抢上下文、数据流混乱。LangGraph 的好处在于,它让你把“智能”交给 LLM,但把“流程控制”牢牢握在自己手里,这非常关键。因为生产环境里的多智能体系统,最怕的就是失控。
以下所有案例都能在 LangGraph 最新的 0.2.x 版本上跑通,代码基于 Python,我会标明每段代码的作用和关键参数。别急,先从设计思路讲起。
1. 整体设计:为什么用 LangGraph 来搭 Multi-Agent
1.1 先想清楚 Multi-Agent 不是“多个 Agent 的堆叠”
很多人一听到 Multi-Agent,第一反应是“我有多个任务,所以我有多个 Agent,每个 Agent 负责一个任务”。这个理解不能说错,但很容易把系统做成接口调用,而不是一个有机协作的团队。真正的 Multi-Agent 系统,核心在于 Agent 之间需要共享上下文、需要路由决策、需要授权调用、需要异常回退、需要合并结果。这些需求在 LangGraph 里对应的就是节点、边、状态、条件分支和 checkpointer。
举个我自己的例子。之前我做一个行业研究报告生成系统,最初拆成了三个独立的 Agent:数据采集 Agent、分析 Agent、写作 Agent。每个 Agent 通过 HTTP 调用串联,后一个 Agent 拿前一个 Agent 的输出作为输入。跑起来就发现几个问题:
- 数据采集 Agent 返回的 JSON 结构不稳定,分析 Agent 经常解析失败。
- 分析 Agent 如果需要补查数据,没法自己回头调用采集 Agent,只能抛出异常让上层服务去协调。
- 改写阶段没法看到中间推理过程,出问题很难定位。
换成 LangGraph 之后,我把这三个“模型”变成了三个“节点”,用图结构让分析 Agent 可以动态决定是否重新调用数据采集节点。整个系统的状态由一个统一的 State 对象管理,每个节点读写 State 中的某个字段,传输和解析的契约终于在同一个框架内收敛了。这就是 Multi-Agent 的“多”,不是在数量上多,而是在协作模式上多。
1.2 LangGraph 与主流编排方案的选型对比
现在市面上的 Agent 编排方案很多,我简单拉一张表,不是要否定谁,而是为了让选型有依据。
| 方案 | 运行模型 | 状态管理 | 复杂流程支持 | LangChain 集成 | 学习曲线 | 典型场景 |
|---|---|---|---|---|---|---|
| LangGraph | 图结构 + Pregel 风格 | 显式 State + Reducer | 支持循环、条件分支、并行、多 Agent | 原生集成 | 中 | 复杂业务流、生产级系统 |
| AutoGen | 对话驱动 | 会话历史 | 通过群聊、发言顺序控制 | 需要额外适配 | 中低 | 多角色对话研究 |
| CrewAI | 角色 + 任务列表 | 任务输出聚合 | 任务依赖,Pipeline 为主 | 有但相对轻量 | 低 | 快速搭建、简单协作 |
| 手写 Pipeline | 代码硬编 | 无 | 每次改动要改代码 | 无 | 高(维护) | 简单一次性脚本 |
我当时选 LangGraph,核心就看中两点:第一,它天然支持循环。多智能体系统里最常见的协作模式是“A 想了之后告诉 B,B 缺信息再回头找 A”,这种循环用 Pipeline 写要反复设计回调,但在图里只是一条从 B 回到 A 的边。第二,它的 State 可序列化且支持 Checkpointer,这意味着每个 Agent 都能看到全局上下文,而且能在任意节点断点恢复,排障能力极强。如果你只是做一次性 demo,CrewAI 确实更爽;但要做到生产级可控,LangGraph 的图模型在理论上更干净。
2. 核心概念拆解:StateGraph、State、Node、Edge
2.1 State:所有 Agent 的“共享白板”
LangGraph 里最重要的概念是 State。它不是单纯的内存字典,而是带 reducer 的状态模型。最简单的定义方式是使用 TypedDict:
from typing import Annotated, TypedDict from langgraph.graph import add_messages class AgentState(TypedDict): messages: Annotated[list, add_messages] # 消息列表,自动追加 current_actor: str # 当前执行节点名称 research_results: dict # 数据库采集结果 draft: str # 写作中间产物 final_answer: str # 最终输出注意Annotated[list, add_messages]这种写法。add_messages是 LangGraph 内置的 reducer,它定义了多个节点往同一个字段写值时如何合并。对于消息列表来说,默认是追加,而不是覆盖。这一点特别关键:你想象多个 Agent 在同一块白板上写字,如果每个人写到同一行,最后只会留下最后一个人的字。用add_messages就相当于每个人在下一行接着写,历史记录不丢。
自定义 reducer 也是可以的。比如我需要把多个 Agent 返回的搜索结果合并成一个大列表,就可以定义:
def merge_results(left: list, right: list) -> list: return left + right然后在 State 字段中写results: Annotated[list, merge_results]。实际开发中,状态字段要精心设计,不要什么都往里塞。我见过有人把大段文本直接放在 State 里,导致每次状态更新都要序列化几十 KB 数据,性能下降非常明显。State 里只存关键词级别的中间结果,详细文档建议给引用 ID。
2.2 Node 和 Edge:状态转移的规则
节点就是普通的 Python 函数(也可以异步 async)。输入是 State 的字典,输出是 State 的部分更新(Partial State)。LangGraph 会自动把输出合并到全局状态。边则分为普通边和条件边,用来决定从哪个节点走到哪个节点。
from langgraph.graph import StateGraph, END def agent_a(state: AgentState): # 处理逻辑 return {"draft": "A 生成的内容"} def agent_b(state: AgentState): # 处理逻辑 return {"final_answer": state["draft"] + " B 完善的内容"}构建图:
graph = StateGraph(AgentState) graph.add_node("A", agent_a) graph.add_node("B", agent_b) graph.set_entry_point("A") graph.add_edge("A", "B") graph.add_edge("B", END)是不是很简单?但一旦引入条件路由,整个复杂度才真正体现出来:
def route_after_research(state: AgentState) -> str: if not state["research_results"]: return "research" # 回到采集节点 return "writer" # 进入写作节点这条条件边就是“返回需要跳转到的节点名称”,非常直观。LangGraph 的一个设计哲学是:所有决策都可以由 LLM 或自定义逻辑驱动。说白了,条件边里你可以调用 GPT-4 判断下一步,也可以写 if-else 硬编码,甚至结合工具结果去判断。这就为 Multi-Agent 提供了“不稳定的智能决策”和“稳定的流程骨架”的结合点。
2.3 Checkpointer:让 Agent 拥有“记忆”和断点
多智能体系统如果真的只有图定义,那它只是静态流程。要支持人机交互、手动干预、故障恢复,必须要有 Checkpointer。LangGraph 的 Checkpointer 可以保存每一步的状态快照,并且支持从任意节点恢复执行。
from langgraph.checkpoint.memory import MemorySaver saver = MemorySaver() # 也可以换成 PostgresSaver、SqliteSaver 等 graph = graph.compile(checkpointer=saver)调用时传入thread_id:
config = {"configurable": {"thread_id": "case-41"}} response = graph.invoke({"messages": []}, config)同一thread_id下的执行历史会被保留,下次调用时会从上次中断的地方继续。这个机制在 Multi-Agent 中太有用了。比如,用户在与多智能体系统交互的过程中,写作 Agent 写了一半,用户突然说“换个语气”,就可以从 writing 节点继续而不需要重新跑一遍数据采集。另外,生产级系统强烈建议用持久化 Checkpointer(如PostgresSaver),别用内存版的MemorySaver,否则服务重启什么都丢了。
3. 实操案例:实现一个“研究顾问”多智能体系统
3.1 业务场景与角色定义
案例背景:做一个研究顾问,用户提出一个问题,系统需要先搜索资料、整理关键信息、生成完整回答,同时能够根据回答质量决定是否重写。三个角色分工如下:
- 主管(Supervisor):负责整体调度,接收用户问题,协调研究节点和写作节点,判断结果是否满足要求。
- 研究员(Researcher):通过知识库或搜索工具获取资料,输出结构化要点。
- 写作者(Writer):把研究要点扩展成用户友好的答案。
这里不是简单的一条链路,因为主管要动态判断研究员是否需要补充信息,写作者写完可能要打回重写。这就是典型的多 Agent 协作模式。
3.2 代码实现:从节点定义到完整流程图
我用 LangGraph 的StateGraph来实现。为了便于理解,这里用 mock 的“工具”代替真实的数据库或 API。
先定义状态和节点:
from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver class ResearchState(TypedDict): question: str # 用户问题 context: list[str] # 检索到的上下文 draft: str # 写作草稿 review_score: int # 主管评分 final_answer: str # 最终答案 # 模拟工具:知识检索 def search_knowledge_base(query: str) -> list[str]: # 这里替换为真实的向量检索 / API 调用 return [f"{query} 的相关资料片段 {i}" for i in range(3)] # 研究员节点 def researcher_node(state: ResearchState): query = state["question"] results = search_knowledge_base(query) return {"context": results} # 写作者节点 def writer_node(state: ResearchState): context = "\n".join(state.get("context", [])) draft = f"根据资料撰写答案:\n上下文: {context}" return {"draft": draft} # 主管节点:对草稿评分并决定后续路由 def supervisor_node(state: ResearchState): # 实际场景可以调用 LLM 评分,这里模拟 if "重要客户" in state["question"]: score = 3 else: score = 9 if score >= 8: return {"review_score": score, "final_answer": state["draft"]} else: return {"review_score": score, "draft": state["draft"] + "\n补充细节"} def route_after_supervisor(state: ResearchState) -> str: if state["review_score"] < 8: return "writer" # 回到写作者重写 return END # 完成构建图:
builder = StateGraph(ResearchState) builder.add_node("researcher", researcher_node) builder.add_node("writer", writer_node) builder.add_node("supervisor", supervisor_node) builder.set_entry_point("researcher") builder.add_edge("researcher", "writer") builder.add_edge("writer", "supervisor") builder.add_conditional_edges( "supervisor", route_after_supervisor, {"writer": "writer", END: END} ) graph = builder.compile(checkpointer=MemorySaver())调用:
config = {"configurable": {"thread_id": "research-case-5"}} result = graph.invoke( {"question": "我们下个季度需要重点拓展哪些客户?"}, config ) print(result["final_answer"])你可能会说,这个案例太简单了,每个节点都没用 LLM。是的,为了展示框架结构我没让节点调用 LLM。真正生产环境里,你可以把节点内部的逻辑换成任何 LLM 调用链。LangGraph 不会限制你在一个节点里是调用一次 LLM 还是调用十次工具。它只负责节点之间的流转。
3.3 参数说明与状态流细节
这个案例里几个我实际敲过的地方,值得细说:
route_after_supervisor的条件边返回字符串,必须映射到图中存在的节点名,或END。如果你的返回键不在映射字典里,会直接报错。记得检查add_conditional_edges第三个参数,它就是一个从返回值到实际节点名的映射。- 每次
invoke传的config里面包含thread_id,这个 ID 用来标识一整个会话。不同会话的状态完全隔离,这一点对于多用户场景非常重要。同一个线程内,状态是累积的;如果不同请求用同一线程 ID,上一个请求的 State 会残留下来。一般我们会用用户 ID + 会话 ID 组合一个唯一线程 ID。 - 我用的
MemorySaver只适合本地测试。用 Postgres 的话只需一行替换,LangGraph 官方提供的 saver 接口都是一致的。后面要讲部署时再展开。
3.4 加入 LLM 与工具调用的真实节点
我们做一个升级版,研究员节点内部会调用真实 LLM 并可能迭代多次工具调用,直到收集足够信息。LangGraph 节点的自由度很高,节点内部可以嵌套使用AgentExecutor或langchain.tools。但这里我想强调一种更地道的做法:把工具调用也建模成子图。
为了让文章更贴近实战,我用 LangChain 的langchain_openai来做 LLM 调用:
from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2) def researcher_node(state: ResearchState): prompt = [ SystemMessage(content="你是行业研究员,请根据问题检索相关资料,并总结3条关键要点。"), HumanMessage(content=state["question"]) ] response = llm.invoke(prompt) return {"context": [response.content]}这样虽然节点内部只调用了 LLM,没有调用外部工具,但如果需要工具,你完全可以在节点内部用tool_calls实现循环。LangGraph 并不强制你使用它内置的ToolNode,你可以自己控制调用次数。
不过内置的ToolNode配合create_react_agent非常省事,前提是你愿意让 LangGraph 替你管理工具循环。以我的经验,如果 Agent 内部的工具调用流程相对固定,用内置ToolNode最稳;如果工具调用逻辑特别复杂且有跨 Agent 共享需求,那就自己写节点内部循环。
3.5 多 Agent 之间的消息传递机制
多智能体的本质是消息交换。LangGraph 里,消息通常放在 State 的messages字段,并且用add_messagesreducer 做追加。每个节点的输入是整个 State,输出中如果包含messages字段,就会追加到全局消息历史。这就像一群人开会,每个人都在会议纪要里不断添加自己的发言。
实际开发中,我会定义额外的字段来存储“结构化消息”,而不是全靠messages。因为 LLM 自然语言消息在跨 Agent 传递时会有解析成本。比如,研究员返回给写作者的应该是结构化要点,而不是一大段自然语言。所以我上面的ResearchState里安排了context: list[str],就是结构化中间产物。在节点内部,messages是给 LLM 看的,而context是给下游流程用的。两者共存不冲突。
4. 经典 Multi-Agent 拓扑:Supervisor / Hierarchical / Network
4.1 三种拓扑对比与选择
在 LangGraph 官方文档里,多智能体结构大致分成三类,我结合自己的项目讲讲适用范围。
Supervisor(主管-工人):所有 Agent 都向主管汇报,主管负责调度。适合上游有明确意图分类,下游任务相对独立的场景。比如,一个客服机器人,主管先判断用户想退款还是想查物流,然后分别派给退款专家或物流专家。优点是决策集中,行为可控;缺点是主管容易成为瓶颈。
Hierarchical(分层):主管下面还有子主管,子主管管理具体执行 Agent。适合任务层级很深、决策粒度从粗到细的场景。比如公司内部知识助手:顶层主管判断是技术还是人事问题,技术子主管再判断是后端还是前端问题,最后交给具体代码 Agent。这种结构在 LangGraph 里嵌套子图来实现,顶层图包含子图节点,子图里再建更细的图。
Network(网络):Agent 之间可以任意通信,彼此不依赖主线。适合高度开放、需要动态协作的场景。但 Network 结构在 LangGraph 里实现起来要额外小心,容易形成死循环和状态爆炸。我的建议是生产系统少用纯网络结构,除非你对每个节点的终止条件做了非常严格的状态约束。
下面这张表是我常用的粗略判断依据:
| 结构 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| Supervisor | 意图清晰、子任务独立 | 易控制、易监控 | 主管节点负载高 |
| Hierarchical | 任务复杂度高、层级分明 | 模块化强、职责分明 | 结构深度大、调试成本高 |
| Network | 探索型、需要多路径自由协作 | 灵活性最强 | 状态复杂、难收敛 |
4.2 用 LangGraph 实现一个 Supervisor 多智能体
我来完整展示一个常见 supervisor 实现。这里的关键点是主管节点里用一个 LLM 来决定下一步找哪个 Agent,其他节点分别处理各自任务。
假设系统里有三个执行 Agent:Researcher、Summarizer、Critic。主管执行流程如下:
from typing import Literal from langgraph.graph import StateGraph, END # 角色消息 SYSTEM_PROMPT = """你是一个主管,根据用户需求选择下一步执行者: - researcher: 需要检索资料 - summarizer: 对已有内容进行总结 - critic: 批评和评估答复质量 只输出一个角色名。""" def supervisor_router(state: ResearchState) -> Literal["researcher", "summarizer", "critic"]: # 这里简化:调用 LLM,然后返回决定 response = llm.invoke([SystemMessage(content=SYSTEM_PROMPT), HumanMessage(content=state["question"])]) role = response.content.strip().lower() if role not in ("researcher", "summarizer", "critic"): return "researcher" return role然后图结构:
builder.add_node("supervisor", supervisor_router) # 这里只是演示,路由函数并不负责LLM内部执行 builder.add_node("researcher", researcher_node) builder.add_node("summarizer", summarizer_node) builder.add_node("critic", critic_node) builder.set_entry_point("supervisor") builder.add_conditional_edges( "supervisor", supervisor_router, { "researcher": "researcher", "summarizer": "summarizer", "critic": "critic" } ) builder.add_edge("researcher", "supervisor") # 执行后回到主管 builder.add_edge("summarizer", "supervisor") builder.add_edge("critic", "supervisor")这里的图是无限循环的:主管不断决定下一步执行者,执行完又回到主管。必须在某处加终止条件。一种做法是在主管节点判断“如果回答满意就返回 END”:
def supervisor_router(state: ResearchState) -> Literal["researcher", "summarizer", "critic", END]: if state.get("final_answer"): return END # ... 其余逻辑注意,supervisor_router被条件边当作路由函数调用,但它也可以是一个实际节点。我通常会把“决策 LLM”放在条件边函数里,这样图结构更清晰:主管不是一个实际执行节点,而是一个路由函数。但条件边函数不能修改状态。如果你需要主管既做决策又修改状态,那就把它定义成节点,节点返回后走条件边。
4.3 子图与嵌套:Hierarchical 的落地姿势
如果你想实现层级结构,LangGraph 允许把一个编译好的图作为另一个图的节点。比如,我有一个“技术组”子图,里面包含前后端两个 Agent,然后将这个子图挂到总图的一个节点上:
tech_subgraph = build_tech_subgraph() # 返回 StateGraph builder.add_node("tech_team", tech_subgraph.compile()) builder.add_edge("supervisor", "tech_team")子图和父图之间通过状态传递数据。父图的 State 字段可以被子图读取、更新,前提是字段名一致。这样就把需求层层拆分成嵌套图。这个模式极大提升了组织的复用性。我在实际项目里,会把可复用的 Agent 团队封装成子图,比如“搜索小组”“合同审核小组”,然后父图通过 supervisor 路由到不同子图。
需要留意的是,子图编译后的invoke内部可能有自己的 checkpointer。如果父子图都配置了 checkpointer,内部状态恢复时会比较绕。我的经验是:子图不单独设置 checkpointer,统一由顶层图管理。这样数据结构最简单,避免多层嵌套的 checkpoint 冲突。
5. 常见问题与排查技巧实录
5.1 无限循环就是出不来怎么办
多智能体图最常见的问题就是死循环。原因往往是条件边的路由总是满足回退条件。我遇到过写作者写出来的内容永远被主管评低分,导致重复循环几十上百次。解决办法有三种:
- 加最大迭代次数。在状态里增加
iteration_count字段,路由函数判断超过阈值直接结束或强制降级。 - 用 checkpointer 做时间旅行。如果你开了 checkpoint,可以回溯到循环起始点,检查每个节点的输出变化。
- 把评价逻辑换成更严格的规则。比如限定评分子项,不要仅凭一个 AI 评分就决定打回。
在 LangGraph 里你可以直接给循环边设置add_conditional_edges,在路由函数里用state["iteration_count"] >= 3返回 END。这是最直接有效的。
5.2 状态被覆盖或丢失
如果你发现下游 Agent 拿到的是空值,先确认你有没有在节点里返回对应字段。还有一个非常隐蔽的坑:State 的 TypedDict 字段类型是list,默认策略是整体覆盖。如果两个节点都往同一个 list 字段写值,后写的会覆盖先写的。所以跨 Agent 的共享结果字段,请务必用Annotated[list, add_messages]或自定义 reducer。
另一种“丢失”发生在子图边界。子图里的节点只更新子图的 State,不会自动同步到父图,除非父图 State 和子图 State 定义完全一致的字段名,并且子图的 super step 返回了这些字段。简单来说,子图只能更新它自己 State 里定义的字段,而这些字段必须与父图 State 同名字段兼容。建议把父子图共享的字段单独提取出来,放在基类 TypedDict 里继承,避免出现两边字段定义不一致的奇葩 bug。
5.3 并发节点结果乱序
LangGraph 支持并行执行多个节点。比如同时让研究员和数据分析员各自工作,然后合并结果。默认执行顺序是并发的,State 的 reducer 会按节点完成顺序合并。如果你的系统对顺序敏感,要利用 reducer 或者给结果加入时间戳。
我写并行节点时,都会在 State 里定义results: Annotated[list, merge_results],然后把每个并行节点的结果都返回results,reducer 负责合成。这里需要测试 reducer 的幂等性,避免重复运行同一个并行节点时结果重复堆积。
5.4 图可视化:用画图的心态调试
LangGraph 官方提供了get_graph().draw_mermaid_png()等方法。我强烈建议每个图在开发阶段都画出来看一眼。用 Mermaid 文本也能查看:
print(graph.get_graph().draw_mermaid())图像化能帮你立刻发现漏边、错边、没有回到主管的问题。不过注意,如果图里有条件边,Mermaid 只会显示分支,不会显示分支条件的细节,所以还是需要代码审计。
5.5 日志与追踪:给 Agent 加“探针”
由于多智能体系统里每一步都是 LLM 调用,你不知道黑盒里发生了什么。LangChain 生态有一个可选的 LANGCHAIN_TRACING 方式,但我更喜欢纯手工方式:在每个节点内加上日志,记录 state 的关键字段。比如:
import logging logger = logging.getLogger("agent") def researcher_node(state: ResearchState): logger.info("researcher starts, question=%s", state["question"]) ... logger.info("researcher got %d contexts", len(new_context))生产环境还会在关键路由决策时打印决策原因。LLM 调用时也在 prompt 里让它把决策理由附加到输出中,这样排障时就能看到完整思路。相信我,在多智能体系统里,没有日志的 Agent 就是定时炸弹。
6. 进阶经验:性能优化与部署注意点
6.1 避免输入历史无限膨胀
多智能体协作时,Agent 的 messages 数量会随迭代次数快速增长。参考对话上下文的add_messagesreducer,每次新增一轮对话/协作,都有大量历史积累。LLM 的上下文窗口有限,费用也随 token 上升。
解决办法是定期压缩提示词或裁剪消息。LangGraph 官方有langchain_core.messages里的消息裁剪工具。我实际用的是:在当前节点调用 LLM 之前,把历史消息中无用的工具输出浓缩成摘要。把这个摘要当做人设的一部分塞到 SystemMessage 里。
6.2 选择合适的持久化后端
之前说过 MemorySaver 仅适合测试。LangGraph 提供了PostgresSaver、SqliteSaver、RedisSaver等。生产环境我推荐 PostgresSaver,因为它可以做真正的事务处理,并且支持多进程并发。切换非常方便:
from langgraph.checkpoint.postgres import PostgresSaver saver = PostgresSaver.from_conn_string("postgresql://user:pass@host:port/db")注意,PostgresSaver 使用时要先调用saver.setup()来建表。每个线程状态都会存在这张表里,对多用户服务特别友好。缺点是需要维护数据库连接池,但对一个正经后端系统这不算事。
6.3 断点续跑与人工审核
有些场景必须经过人工确认才放行。LangGraph 支持interrupt_before和interrupt_after参数来插入断点。典型用法是:LLM 生成内容后,在交付给用户前暂停,等待人工编辑。我们可以编译时指定:
graph = builder.compile( checkpointer=saver, interrupt_before=["publish_node"] )然后外部服务在使用时检测graph.invoke返回的状态,若处于中断状态则展示给人类审核。人类同意后再调用graph.invoke(None, config=config)让它继续。这个功能让多智能体系统真正可以部署在“人机协同”的场景里,而不是完全无人监督。
6.4 我踩过的坑:并行模型与同一线程冲突
有一次我图里有两条并行分支,都通过add_edge回到汇总节点。两个分支执行结束后,汇总节点被调度了两次?不对,LangGraph 本身会等待所有并行分支完成后再执行下游节点。但我踩的坑是,两个并行分支同时修改同一个字段而没有 reducer,导致一侧结果被另一侧覆盖。这个问题排查了很久,最后通过加日志发现两个节点输出的字段 key 相同。此后我在设计 State 时严格约束:同一份数据只有一个人去写,其他人只能读。如果必须多写,请上 reducer。
最后分享两个实际心得
先说说选型倾向。LangGraph 的 Multi-Agent 模式不适合特别简单的一次性对话任务,那样纯 Pipeline 就够。但一旦你要构建一个需要团队协作、具备长期记忆、可在任意环节支持人工干预的系统,LangGraph 的图模型确实比手写回调靠谱得多。它并没有把多智能体问题直接变简单,但它把“控制权”放回了开发者手里,这在生产环境里是压倒性的优势。
再说个实在的小技巧。开发阶段,我给每个 Agent 节点都起一个“动作动词 + 对象”的名字,比如research_customer_clues、write_solution_draft、validate_answer_quality。这不仅仅是命名习惯,也是为了让 LangGraph 的可视化图、日志、模块化调试三者共用同一套语义。每次你看到路由结果里跳转到了write_solution_draft,你就知道这一步在干什么。别小看这个习惯,多智能体系统一旦有六个以上节点,命名混乱带来的心理负担比代码本身难缠得多。
最后留一个可扩展的方向。上面所有案例都是“静态图”——节点和边在编译前就定死了。LangGraph 也支持动态增加节点的能力,但我们团队在实践中发现,动态图对状态约束要求极高,容易失控。我目前只在一个实验项目里用过,生产系统还是尽量把流程显式画出来。图一旦动态,排查问题的成本就指数级上升。所以除非你是纯粹做研究,否则推荐先把静态图画得明明白白,再去琢磨那些花活。