从 LangChain 到 LangGraph：多 Agent 不是玄学，是 token 账本和干扰问题

本文目录

一个决策冲突：所有功能塞给一个 Agent，还是拆开？
问题 0：一张图最少长什么样？
问题 1：分支怎么写？
问题 2：循环怎么写？
问题 3：第二次调用怎么记得上一次？
问题 4：高危操作执行前怎么让人确认？
收藏资产：机制—写法对照表 + 迁移自检清单
结尾：一个可迁移的判断

从 LangChain 到 LangGraph：多 Agent 不是玄学，是 token 账本和干扰问题 ----------------------------------------------------

摘要：这篇文章回答一个具体问题——为什么会从 LangChain 走到 LangGraph，图编排、状态持久化、人工中断这些关键词落到代码上长什么样。核心判断：LangGraph 的本质是把工作流组织方式从线性链升级为网状图，三件套是 State + 节点 + 边。文中 5 个最小示例均不调用 LLM API，为静态阅读整理，运行未验证。

一个决策冲突：所有功能塞给一个 Agent，还是拆开？

先把冲突摆出来。做 Agent 应用时有一个看似省事的选择：单 Agent，把所有 tool 描述、所有功能的 prompt 全部写进 system prompt。代价在仓库笔记里写得很直白（readme.md#L7-12）：每次调用全量携带，token 消耗高；无关信息干扰模型，准确率下降。

拆成多个 Agent 后，每个 Agent 只带最少 prompt——省 token、无干扰、准确率高；主 Agent 下发任务，子 Agent 还能并行处理、互相讨论纠错（readme.md#L17-31）。所以本文的第一个判断是：多 Agent 拆分不是架构玄学，是 token 开销与信息干扰的工程权衡。

而组织多个 Agent（以及任何带分支、循环、暂停的工作流），LangChain 的线性链不够用了。LangChain 与 LangGraph 的分工（readme.md#L33-38）：LangChain = 线性工作流编排 + 基础模块（LLM API/loaders/splitter/embedding/vector store/output parser/memory）；LangGraph = 网状工作流编排，负责"工作节点 + 组织方式"。两者是配合关系，不是替代关系。

下面按"遇到什么问题 → LangGraph 给什么答案"的顺序，过一遍 5 个最小示例。

问题 0：一张图最少长什么样？

答案：State + 节点 + 边三件套。这是全文最重要的一段代码（basic-graph.mjs）：

<span>import</span> { <span>Annotation</span>, <span>StateGraph</span>, <span>START</span>, <span>END</span> } <span>from</span> <span>'@langchain/langgraph'</span>

<span>// 1. 声明 State：reducer 决定"怎么更新"，default 是初始值</span>
<span>const</span> <span>StateAnnotation</span> = <span>Annotation</span>.<span>Root</span>({
  <span>text</span>: <span>Annotation</span>({
    <span>reducer</span>: <span>(<span>_prev, next</span>) =></span> next, <span>// 直接用新值覆盖旧值</span>
    <span>default</span>: <span>() =></span> <span>""</span>,
  }),
})

<span>// 2. 节点就是普通函数，返回值是"对状态的更新"（部分状态）</span>
<span>const</span> <span>step1</span> = (<span>state</span>) => ({ <span>text</span>: <span>`<span>${state.text}</span> -> step1`</span> })
<span>const</span> <span>step2</span> = (<span>state</span>) => ({ <span>text</span>: <span>`<span>${state.text}</span> -> step2`</span> })

<span>// 3. 编排：加节点 → 连边 → 编译</span>
<span>const</span> graph = <span>new</span> <span>StateGraph</span>(<span>StateAnnotation</span>)
  .<span>addNode</span>(<span>"step1"</span>, step1)
  .<span>addNode</span>(<span>"step2"</span>, step2)
  .<span>addEdge</span>(<span>START</span>, <span>"step1"</span>)     <span>// START 是特殊节点：图的入口</span>
  .<span>addEdge</span>(<span>"step1"</span>, <span>"step2"</span>)
  .<span>addEdge</span>(<span>"step2"</span>, <span>END</span>)       <span>// END 是特殊节点：图的出口</span>
  .<span>compile</span>()

<span>// 可视化：生成 mermaid 流程图文本</span>
<span>const</span> drawable = <span>await</span> graph.<span>getGraphAsync</span>()
<span>console</span>.<span>log</span>(drawable.<span>drawMermaid</span>({ <span>withStyles</span>: <span>true</span> }))

<span>// 运行：状态沿节点链路依次更新</span>
<span>console</span>.<span>log</span>(<span>await</span> graph.<span>invoke</span>({ <span>text</span>: <span>"hello"</span> }))

读这段代码抓三个点：

Annotation.Root({...}) 声明状态 schema（basic-graph.mjs#L9-16）：字段级 reducer: (_prev, next) => next 决定状态怎么更新（这里新值覆盖旧值），default 给初始值。
节点即函数（basic-graph.mjs#L18-21）：返回值不是完整状态，是"本节点的更新"，交给 reducer 合并。
invoke({text:"hello"}) 沿 START → step1 → step2 → END 流转，最终 text 为 hello -> step1 -> step2（reducer 逻辑推导，运行未验证）。

对应图 API 四要素心智模型（readme.md#L40-48）：开始节点（初始状态）、工作节点（职责 + state）、边（连接）、结束节点（最终状态）。drawMermaid({withStyles:true}) 可以导出 mermaid 文本，随时检查图的形状。

问题 1：分支怎么写？

答案：决策写在节点里，路由交给条件边。示例是"算式走计算、否则走聊天"（conditional-routing.mjs）：

<span>// router 节点：只负责"判断"，把决策写进 state</span>
<span>const</span> <span>router</span> = (<span>state</span>) => ({
  <span>route</span>: <span>/[+\-*]/</span>.<span>test</span>(state.<span>query</span>) ? <span>"math"</span> : <span>"chat"</span>,
})

<span>const</span> <span>math</span> = (<span>state</span>) => {
  <span>try</span> {
    <span>// eval 把字符串当 JS 代码执行并返回结果（如 "1+2" → 3）</span>
    <span>return</span> { <span>answer</span>: <span>String</span>(<span>eval</span>(state.<span>query</span>)) }
  } <span>catch</span> {
    <span>return</span> { <span>answer</span>: <span>"数学公式有误"</span> } <span>// 节点内兜底错误</span>
  }
}

<span>const</span> <span>chat</span> = (<span>state</span>) => ({ <span>answer</span>: <span>`聊天模式：<span>${state.query}</span>`</span> })

<span>// 条件边：条件函数返回 key，映射表把 key 翻译成下一个节点</span>
graph.<span>addConditionalEdges</span>(<span>"router"</span>, <span>(<span>state</span>) =></span> state.<span>route</span>, {
  <span>math</span>: <span>"math"</span>,
  <span>chat</span>: <span>"chat"</span>,
})

注意关注点分离：router 节点只做正则判断 /[+\-*]/.test(query)，返回 {route:"math"|"chat"}（#L23-28）；addConditionalEdges 的条件函数返回 key，映射表把 key 翻译成下一个节点（#L51）。eval 包在 try/catch 里，失败兜底"数学公式有误"（#L31-39）；eval("1+2") 返回 3 可单独验证（test.mjs）。

问题 2：循环怎么写？

答案：没有专门的循环 API，条件边指向自己就是循环（loop-retry.mjs）：

<span>// attempt 节点：每次把 tries + 1，第 3 次算成功</span>
<span>const</span> <span>attempt</span> = (<span>state</span>) => {
  <span>const</span> tries = state.<span>tries</span> + <span>1</span>
  <span>const</span> ok = tries >= <span>3</span>
  <span>return</span> { tries, ok, <span>message</span>: <span>`第<span>${tries}</span>次<span>${ok ? <span>"成功"</span> : <span>"失败"</span>}</span>`</span> }
}

<span>// 条件边自环：未达标回到自己（retry → attempt），达标去 END</span>
graph.<span>addConditionalEdges</span>(
  <span>"attempt"</span>,
  <span>(<span>state</span>) =></span> (state.<span>ok</span> ? <span>"done"</span> : <span>"retry"</span>),
  { <span>retry</span>: <span>"attempt"</span>, <span>done</span>: <span>END</span> }
)

tries 计数 + ok = tries >= 3 判定在节点里（#L23-31），条件边在未达标时把路由指回 attempt 自身，达标走 END（#L36-39）。自环 + 计数 + 终止条件三要素，缺终止条件就是死循环。

问题 3：第二次调用怎么记得上一次？

答案：compile({checkpointer}) + thread_id（checkpointer-memory.mjs）：

<span>import</span> { <span>MemorySaver</span> } <span>from</span> <span>'@langchain/langgraph'</span>

<span>const</span> <span>StateAnnotation</span> = <span>Annotation</span>.<span>Root</span>({
  <span>visitCount</span>: <span>Annotation</span>({
    <span>reducer</span>: <span>(<span>_prev, next</span>) =></span> next,
    <span>default</span>: <span>() =></span> <span>0</span>,
  }),
})

<span>const</span> <span>visit</span> = (<span>state</span>) => ({ <span>visitCount</span>: state.<span>visitCount</span> + <span>1</span> })

<span>const</span> checkpointer = <span>new</span> <span>MemorySaver</span>()          <span>// 内存检查点</span>
<span>const</span> graph = <span>new</span> <span>StateGraph</span>(<span>StateAnnotation</span>)
  .<span>addNode</span>(<span>"visit"</span>, visit)
  .<span>addEdge</span>(<span>START</span>, <span>"visit"</span>)
  .<span>addEdge</span>(<span>"visit"</span>, <span>END</span>)
  .<span>compile</span>({ checkpointer })                    <span>// 编译时挂上</span>

<span>const</span> config = { <span>configurable</span>: { <span>thread_id</span>: <span>"用户_小张"</span> } }

<span>await</span> graph.<span>invoke</span>({}, config)                  <span>// 第 1 次：visitCount = 1</span>
<span>await</span> graph.<span>invoke</span>({}, config)                  <span>// 第 2 次：基于上次状态 = 2</span>

<span>// 换一个 thread_id，状态从初始值重新开始</span>
<span>await</span> graph.<span>invoke</span>({}, { <span>configurable</span>: { <span>thread_id</span>: <span>"用户_小李"</span> } }) <span>// = 1</span>

关键行为（#L36-45）：同一 thread_id 连续两次 invoke，第 1 次 visitCount = 1，第 2 次基于上次状态累加为 2；换 thread_id 从初始值重新计数，会话相互隔离。MemorySaver 只存内存，中断/暂停/失败后可继续；真正持久化按材料指引可换 sqlite、redis（readme.md#L56-63）。

问题 4：高危操作执行前怎么让人确认？

答案：interrupt() 暂停 + Command({resume}) 恢复。场景是转账确认——图执行"向张三转账 $100"前停下来等人工审批（graph-interrupt.mjs）：

<span>const</span> <span>showTransfer</span> = (<span>state</span>) => ({
  <span>actionSummary</span>: <span>"向张三转账$100"</span>, <span>// 高危操作的摘要，供人审阅</span>
})

<span>const</span> <span>waitConfirm</span> = (<span>state</span>) => {
  <span>const</span> text = <span>interrupt</span>({          <span>// 图在这里暂停，payload 抛给调用方</span>
    <span>hint</span>: <span>"中断里输入[确认]或者备注后回车，图才会继续执行"</span>,
    <span>actionSummary</span>: state.<span>actionSummary</span>,
  })
  <span>return</span> { <span>useInput</span>: <span>String</span>(text) } <span>// resume 后从这里继续</span>
}

<span>const</span> graph = <span>new</span> <span>StateGraph</span>(<span>StateAnnotation</span>)
  .<span>addNode</span>(<span>"showTransfer"</span>, showTransfer)
  .<span>addNode</span>(<span>"waitConfirm"</span>, waitConfirm)
  .<span>addEdge</span>(<span>START</span>, <span>"showTransfer"</span>)
  .<span>addEdge</span>(<span>"showTransfer"</span>, <span>"waitConfirm"</span>)
  .<span>addEdge</span>(<span>"waitConfirm"</span>, <span>END</span>)
  .<span>compile</span>({ <span>checkpointer</span>: <span>new</span> <span>MemorySaver</span>() }) <span>// interrupt 必须有 checkpointer</span>

<span>const</span> config = { <span>configurable</span>: { <span>thread_id</span>: <span>"interrupt-demo"</span> } }

<span>// 第一次 invoke：跑到 waitConfirm 暂停</span>
<span>const</span> paused = <span>await</span> graph.<span>invoke</span>({}, config)
<span>console</span>.<span>log</span>(<span>"待你确认"</span>, paused.<span>__interrupt__</span>?.[<span>0</span>]?.<span>value</span>)

<span>// ……用 node:readline/promises 读取命令行人工输入（略）</span>

<span>// 同一 thread_id + Command resume：从暂停点继续</span>
<span>const</span> done = <span>await</span> graph.<span>invoke</span>(<span>new</span> <span>Command</span>({ <span>resume</span>: line }), config)
<span>console</span>.<span>log</span>(<span>"done"</span>, done)

闭环四步：节点内 interrupt({hint, actionSummary}) 暂停并抛出 payload（#L27-33）→ 调用方从 paused.__interrupt__?.[0]?.value 读中断信息（#L55-56）→ 示例用 node:readline/promises 读命令行输入模拟人工确认（#L10、L59-66）→ new Command({resume: line}) 配合同一个 thread_id 再次 invoke，从暂停点继续（#L68）。这张图 drawMermaid 导出的形状：__start__ → showTransfer → waitConfirm → __end__（g.md）。一个推导判断：interrupt 依赖 checkpointer，暂停后状态无处保存就无法恢复。

收藏资产：机制—写法对照表 + 迁移自检清单

机制对照表（对应上面 5 个示例）：

机制	解决什么	关键写法	出处
State	状态怎么声明与合并	`Annotation.Root` + 字段级 reducer/default	basic-graph.mjs#L9-16
条件边	分支路由	`addConditionalEdges(节点, 条件函数, 映射表)`	conditional-routing.mjs#L51
条件边自环	循环重试	条件函数未达标时指回自身 + 终止条件	loop-retry.mjs#L36-39
checkpointer + thread\_id	多会话持久化与隔离	`compile({checkpointer})` + `configurable.thread_id`	checkpointer-memory.mjs#L36-45
interrupt + Command	高危操作人工确认	`interrupt(payload)` → `__interrupt__[0].value` → `new Command({resume})`	graph-interrupt.mjs

迁移自检清单（把 LangChain 链改造成图时逐项过）：

State 每个字段都有 reducer 和 default？
节点返回"部分状态"而非整个 state？
条件函数返回值与映射表 key 一一对应？
自环有终止条件，不会死循环？
需要 remember 的图挂了 checkpointer，invoke 带 thread_id？
interrupt 恢复用同一 thread_id + new Command({resume})？

结尾：一个可迁移的判断

编排方式从线性升级到网状，不改变"节点是函数、状态是数据"的本质——先写最小图建立心智模型，再叠加分支、循环、持久化、中断四个机制。本文 5 个示例静态阅读整理、运行未验证，且刻意不调用 LLM API 以聚焦图机制。可以立即执行的一步：跑通 basic-graph，把 drawMermaid 的输出贴进任意 mermaid 渲染器，对照本文核对每张图的形状。

核验说明：平台规则基线 last_verified 2026-07-13；文中技术事实来自本地材料，运行结果未验证。