从线性 Agent 到流程图：用 LangGraph 编排分岔、循环与暂停

当 Agent 从单步问答扩展到路由判断、失败重试和人工审批时，线性调用链很快会被大量条件与回调淹没。LangGraph 将执行过程表示为由节点、边和共享状态组成的图，让流程控制变得清晰。下面从最小示例入手，逐步实现分岔、循环、暂停恢复与状态持久化。

单 Agent 是直线代码，多 Agent 是一张会分岔、循环、暂停的图：LangGraph 工作流编排

如果把复杂 Agent 比作一家公司，LangChain 是"流水线"，LangGraph 是"组织架构图"。 本文用最小可运行的代码，把 LangGraph 最值钱的分岔、循环、暂停恢复三件事讲透。

---

1. 先说人话：为什么需要"图"而不是"一条线"

上一个项目里，我用 LangChain 写 Agent，本质是线性工作流：A 做完 → B 做 → C 做，一条直线走到底。

可真实的 Agent 不是流水线：

• 用户问"1+2 等于几"，和问"你好"，走的根本不是同一条路——该分岔；
• 调模型失败、校验不过，要重试——该循环；
• 要用户确认才能继续——该暂停，等批准再走。

直线写得出来这些，但代码会变成一坨 if/else 套循环套回调，自己都嫌乱。

LangGraph 换个思路：把流程本身画成一张图。

开始
  │
  ▼
 step1 ──► step2 ──► 结束

• 节点 = 一个函数（做一件事）
• 边 = 节点怎么连（路由怎么走）
• state = 整张图的"全局血液"，节点读它、改它、传给下一个

我最开始理解 LangGraph，就是这句话：节点是函数，边决定走向，State 是血液。 抓准这三个词，后面全通了。

---

2. 最小图：State = 图的血液，reducer 决定"改"的方式

先看最基础的一张图，两个节点串起来：

import { Annotation, END, START, StateGraph } from '@langchain/langgraph';

// ① 声明状态的"形状"：只有 text 一个字段
const StateAnnotation = Annotation.Root({
  text: Annotation({
    // 关键：reducer 决定"这个字段更新时怎么变"，而不是无脑覆盖
    reducer: (_prev, next) => next,
    default: () => "",
  })
});

// ② 节点就是函数：返回"给下一个节点的新状态"
const step1 = (state) => ({ text: `${state.text} -> step1` });
const step2 = (state) => ({ text: `${state.text} -> step2` });

// ③ 建图：申明节点、连线、编译
const graph = new StateGraph(StateAnnotation)
  .addNode("step1", step1)
  .addNode("step2", step2)
  .addEdge(START, "step1")
  .addEdge("step1", "step2")
  .addEdge("step2", END)
  .compile();

const result = await graph.invoke({ text: "hello" });
console.log(result); // { text: "hello -> step1 -> step2" }

全图只有三件事：

节点 = 函数。 addNode("step1", step1)，函数入参是当前 state，返回的是更新后的状态。

边 = 走向。 addEdge(START, "step1") 从入口进 step1，最后 addEdge("step2", END) 收在结束。

reducer 决定状态怎么变。 这是最容易忽略、也最值钱的一句。reducer: (_prev, next) => next 意思是：新来的覆盖旧的。但如果你想让字段追加而不是覆盖，把 reducer 改成数组拼接即可——状态怎么合并，是你说了算，而不是框架一刀切。

用 console.log(state.result.text) 能看到图里所有中间 state 流转，调试必备。

---

3. 分岔：让图自己挑路走（条件边）

同一个入口，问题不同走不同节点——这就是 addConditionalEdges。

const StateAnnotation = Annotation.Root({
  query:  Annotation({ reducer: (_prev, next) => next, default: () => "" }),
  route:  Annotation({ reducer: (_prev, next) => next, default: () => "ch@t" }),
  answer: Annotation({ reducer: (_prev, next) => next, default: () => "" })
});

// 路由节点：看 query 决定 next 走哪
const router = (state) => {
  const isMath = /[+-*/]/.test(state.query);
  return { route: isMath ? "math" : "ch@t" };
};
const mathNode = (state) => ({ answer: String(eval(state.query)) });
const ch@tNode = (state) => ({ answer: `你说的是：${state.query}` });

const graph = new StateGraph(StateAnnotation)
  .addNode("router", router)
  .addNode("math", mathNode)
  .addNode("ch@t", ch@tNode)
  .addEdge(START, "router")
  // 关键：条件边！返回值 "math" / "ch@t" 决定走哪条
  .addConditionalEdges("router", (state) => state.route, {
    math: "math",
    ch@t: "ch@t"
  })
  .addEdge("math", END)
  .addEdge("ch@t", END)
  .compile();

console.log(await graph.invoke({ query: "你好" })); // { route:'ch@t', answer:'你说的是：你好' }
console.log(await graph.invoke({ query: "1+2" }));  // { route:'math', answer:'3' }

            ┌─ math ─ 算结果 ─┐
开始 → router                END
            └─ ch@t ─ 回一句 ─┘

addConditionalEdges("router", fn, map) 的机制：fn(state) 返回一个 key，map 里 key → 节点名，图就跳到对应节点。把"决策"从节点里抽到边上，是整个框架最灵活的地方。

⚠️ 踩坑（readme 里的原话，最常见的坑）：eval() 会把字符串当 JS 代码执行。这里 eval("1+2") 是 3，但如果 query 是恶意字符串（比如 process.exit()），你的进程就没了。demo 里图方便用它，生产环境绝不能用 eval 跑用户输入，代码里 dogfood 时尤其要警惕。

---

4. 循环：条件边指向自己 = 重试

分岔是边指到另一个节点；如果条件边指回自己，就是循环重试。

const attempt = (state) => {
  const tries = state.tries + 1;
  const ok = tries >= 3;
  return { tries, ok, message: ok ? `第${tries}次成功` : `第${tries}次失败` };
};

const graph = new StateGraph(StateAnnotation)
  .addNode("attempt", attempt)
  .addEdge(START, "attempt")
  .addConditionalEdges("attempt", (state) => state.ok ? "done" : "retry", {
    retry: "attempt", // ← 指回自己，形成循环
    done: END
  })
  .compile();

             ┌────────── retry ─────────┐
             ▼                          │
开始 → attempt（tries=1,2,3） ── done ──► END

第 1、2 次 ok=false → 走 retry 回到自己，tries 累加；第 3 次 ok=true → 走 done 结束。重试 3 次的逻辑，在 LangGraph 里就是"一条边指回自己"，循环越界全靠 state 里的 tries 条件收敛，优雅到有点不真实。

---

5. 暂停与恢复：把"人工确认"做进图里

前面三招已经能画公司架构图了，但还缺最关键的一环——机器算得再好，要不要真的打？ 得人点头。

LangGraph 用 interrupt + Command({ resume }) + MemorySaver 三件套解决"图走一半停下来等人"。

const StateAnnotation = Annotation.Root({
  actionSummary: Annotation({ reducer: (_prev, n) => n, default: () => "" }),
  userInput:     Annotation({ reducer: (_prev, n) => n, default: () => "" }),
});

const showTransfer = () => ({ actionSummary: "向张三 $100" });

const waitConfirm = (state) => {
  const text = interrupt({ // 在这里打断：图暂停，把控制权交回人
    hint: "终端输入[确认]或备注后回车，图继续",
    actionSummary: state.actionSummary,
  });
  return { userInput: String(text) };
};

const graph = new StateGraph(StateAnnotation)
  .addNode("showTransfer", showTransfer)
  .addNode("waitConfirm", waitConfirm)
  .addEdge(START, "showTransfer")
  .addEdge("showTransfer", "waitConfirm")
  .addEdge("waitConfirm", END)
  .compile({ checkpointer: new MemorySaver() }); // ⚠️ 必须有 checkpointer 才能暂停

const config = { configurable: { thread_id: "interrupt-demo" } };
const paused = await graph.invoke({}, config);
console.log("待你确认:", paused.__interrupt__?.[0]?.value); // → 提示确认

// …你（人）确认了，把结果喂回去，图从打断处继续
const done = await graph.invoke(new Command({ resume: "确认, 金额正确" }), config);
console.log("done:", done); // → userInput: "确认, 金额正确"

开始 → 显示概要 → 【interrupt 停下来，等人确认】
                                │  用户输入"确认"
                                ▼
                         继续 → 结束（拿到 userInput）

这下"前必须人工确认"这种需求，不再是堆锁和状态机，而是图上一个会暂停的节点。interrupt 暂停、Command.resume 续走、thread_id 记住是哪一段中断，天然支持审批流、人工兜底、失败恢复。

---

6. 持久化 + 多用户隔离：thread_id 是"谁的记忆"

上面 interrupt 用到的 MemorySaver，同时解决一个大问题：默认每次 invoke 状态都从零开始，重启全丢。 checkpointer 把 state 存起来，同 thread_id 下次继续。

const checkpointer = new MemorySaver();
const app = graph.compile({ checkpointer });

const user1 = { configurable: { thread_id: "用户-小张" } };
const user2 = { configurable: { thread_id: "用户-小李" } };

await app.invoke({}, user1); // 小张 visitCount=1
await app.invoke({}, user1); // 小张 visitCount=2（记住了！）
await app.invoke({}, user2); // 小李 visitCount=1（隔离，不串）

• 同一个 thread_id → 状态在上一次基础上继续（记忆）
• 不同 thread_id → 互不干扰，天然多用户隔离

到项目里，把 MemorySaver 换成 SQLite / Redis / Postgres 的 checkpointer，state 就能跨进程、跨重启持久化——Agent 从"每次失忆"变成"有记忆的个体"。

---

7. 一句话收尾

LangChain 教会我串（线性流水线）；LangGraph 教会我画（一张状态图）。

真正的分水岭不是"多一个库"，而是想法的转变：

别再把流程写成代码里的一串命令，把它画成一幅图——节点是函数、边是走向、State 是血液。 这样，决策放边上（分岔）、失败指回自己（循环）、拿不准就停在半路等批准（暂停恢复），全都成了图的"艺术"，而不再是你手工维护的 if/else 地狱。

多 Agent 的协作结构（主管下发、子 Agent 并行、评审纠错），本质就是一张更大的、会分岔会循环的图。先学会画这张图，多 Agent 才真正落地。