AI News
中文资讯 掘金 AI Sun, 13 Sep 2026 正文 ok 1245 words

一文搞懂 LangGraph 分支、循环与状态持久化

英文标题：一文搞懂 LangGraph 分支、循环与状态持久化

本文系统讲解 LangGraph 的条件分支、循环重试、状态持久化，以及基于 interrupt 的中断与恢复机制。`addConditionalEdges` 既可用于运行时路由，也可通过回指节点实现重试循环；`checkpointer` 与 `thread_id` 则为多轮会话和用户隔离提供状态支持。文章通过 MemorySaver、SqliteSaver、Command 等示例说明，开发者可将普通工作流升级为支持断点续传和人工审批的有状态 Agent 流程。对生产环境而言，内存级 MemorySaver 适合调试，数据库型 Checkpointer 更适合持久化与可审计场景。
打开原文相关 5 · 机会 1
media3.2 KB media589 B media6.2 KB

中文

上一篇我们搞定了 LangGraph 的线性流程编排和多 Agent 协作基础。这一篇深入 LangGraph 最核心的三个能力：条件分支、循环重试、状态持久化，最后再来一个硬核的 中断/恢复 实战。

一、条件分支：让工作流“会思考”

为什么需要分支？

线性流程只能 A → B → C 一路走到黑。真实场景中，Agent 需要根据输入决定走哪条路——用户问数学题走计算器，闲聊走对话，代码走 Code Interpreter。

LangGraph 用 addConditionalEdges 实现分支。

完整代码

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


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


const router = (state) => {
  const isMath = /[+\-*/]/.test(state.query);
  return { route: isMath ? "math" : "chat" };
};


const mathNode = (state) => {
  try {
    return { answer: String(eval(state.query)) };
  } catch {
    return { answer: "表达式无法计算" };
  }
};


const chatNode = (state) => ({ answer: `你说的是：${state.query}` });


const graph = new StateGraph(StateAnnotation)
  .addNode("router", router)
  .addNode("math", mathNode)
  .addNode("chat", chatNode)
  .addEdge(START, "router")
  .addConditionalEdges("router", (state) => state.route, {
    math: "math",
    chat: "chat"
  })
  .addEdge("math", END)
  .addEdge("chat", END)
  .compile();

流程图

graph TD;
    __start__([__start__]) --> router{router}
    router -->|route=math| math[math]
    router -->|route=chat| chat[chat]
    math --> __end__([__end__])
    chat --> __end__

执行过程

await graph.invoke({ query: "1+2" });




await graph.invoke({ query: "你好" });

面试高频问：addConditionalEdges 和 addEdge 有什么区别？

API 作用 决定时机
addEdge(from, to) 固定连线，永远走 to 设计时确定
addConditionalEdges(from, fn, map) 运行时根据状态动态选路 运行时决定

一句话：addEdge 是“写死的路”，addConditionalEdges 是“到路口再决定”。

二、循环重试：让工作流“会坚持”

为什么需要循环？

Agent 执行不是一次就成功的——API 调用超时、LLM 输出格式不对、用户确认不通过……都需要重试机制。

LangGraph 用 addConditionalEdges 让节点回到自己，形成循环。

完整代码

const StateAnnotation = Annotation.Root({
  tries:   Annotation({ reducer: (_prev, next) => next, default: () => 0 }),
  ok:      Annotation({ reducer: (_prev, next) => next, default: () => false }),
  message: Annotation({ reducer: (_prev, next) => next, default: () => "" })
});

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();

流程图

graph TD;
    __start__([__start__]) --> attempt[attempt]
    attempt -->|ok=false| attempt
    attempt -->|ok=true| __end__([__end__])

执行过程

await graph.invoke({ tries: 0 });

面试高频问：分支和循环用的是同一个 API？

是的，都是 addConditionalEdges。

区别在于 map 里映射的目标：

{ math: "math", chat: "chat" }


{ retry: "attempt", done: END }

一句话：分支是“走哪条路”，循环是“要不要重来”，底层同一个 API，映射目标不同。

三、状态持久化：让工作流“有记忆”

没有持久化的问题

每次 graph.invoke() 状态都从默认值开始，相当于每次都失忆。

第1次调用 → visitCount=0 → 结果1
第2次调用 → visitCount=0 → 结果1   ← 又从头开始了
第3次调用 → visitCount=0 → 结果1

MemorySaver：一行代码加记忆

import { MemorySaver } from '@langchain/langgraph';

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

加上 checkpointer 后，每次 invoke 结束自动保存状态快照，下次同一会话调用时自动恢复。

完整代码

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

const StateAnnotation = Annotation.Root({
  visitCount: Annotation({ reducer: (_prev, next) => next, default: () => 0 }),
  message:    Annotation({ reducer: (_prev, next) => next, default: () => "" })
});

function recordVisit(state) {
  const visitCount = state.visitCount + 1;
  const message = visitCount === 1
    ? "这是你在本会话里第1次进入。"
    : `这是你在本会话里${visitCount}次进入`;
  return { visitCount, message };
}

const graph = new StateGraph(StateAnnotation)
  .addNode("recordVisit", recordVisit)
  .addEdge(START, "recordVisit")
  .addEdge("recordVisit", END);

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

多用户会话隔离

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

await app.invoke({}, user1);  
await app.invoke({}, user1);  
await app.invoke({}, user1);  
await app.invoke({}, user2);

thread_id 就是会话标识——不同用户、不同对话，状态完全隔离。

面试高频问：MemorySaver 和数据库存储有什么区别？

| MemorySaver| 数据库存储
---|---|---
存储位置| 进程内存| SQLite / Redis / PostgreSQL
重启后| 丢失| 保留
适用场景| 开发调试、无状态要求| 生产环境、需要持久化
实现方式| LangGraph 内置| 需要换 Checkpointer 实现

生产环境把 MemorySaver 换成 SqliteSaver 或 RedisSaver，API 完全一样，只需改一行 import。

四、中断与恢复：让工作流“会等待”

这是 LangGraph 最强的能力之一——让图暂停，等人类输入后再继续。

场景：转账确认

import {
  Annotation, END, START, StateGraph,
  Command,        
  interrupt,      
  MemorySaver
} from '@langchain/langgraph';
import { createInterface } from "node:readline/promises";

const StateAnnotation = Annotation.Root({
  actionSummary: Annotation({ reducer: (_prev, next) => next, default: () => "" }),
  userInput:     Annotation({ reducer: (_prev, next) => next, 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() });

流程图

graph TD;
    __start__([__start__]) --> showTransfer[showTransfer]
    showTransfer --> waitConfirm[waitConfirm]
    waitConfirm -.->|interrupt: 等待输入| waitConfirm
    waitConfirm --> __end__([__end__])

执行过程：两阶段调用

const config = { configurable: { thread_id: "interrupt-demo" } };


const paused = await graph.invoke({}, config);
console.log("待你确认:", paused.__interrupt__?.[0]?.value);




const rl = createInterface({ input: process.stdin, output: process.stdout });
const line = (await rl.question("> ")).trim();
await rl.close();

const done = await graph.invoke(new Command({ resume: line }), config);
console.log("done:", done);

面试高频问：interrupt 和普通暂停有什么区别？

| interrupt| 普通暂停
---|---|---
状态保存| 自动保存到 checkpointer| 丢失
恢复方式| new Command({ resume })| 无法恢复
多次中断| 支持，状态链式保存| 不支持
生产可用| 是，支持数据库持久化| 仅开发调试

一句话：interrupt 是 LangGraph 的“人工审批节点”——暂停时状态自动存档，恢复时用 Command 传入人类决策，整个过程可审计、可持久化。

五、核心概念总结

API 速查表

API 作用 类比
Annotation.Root({}) 定义状态结构 数据库建表
StateGraph 创建流程编排器 画流程图
.addNode(name, fn) 添加处理节点 流程图里的方框
.addEdge(from, to) 固定连线 “做完 A 做 B”
.addConditionalEdges(from, fn, map) 条件分支/循环 “如果……就……” / “重来”
.compile({ checkpointer }) 编译 + 注入持久化 编译 + 配置运行时
interrupt({}) 中断等待人类输入 审批暂停点
new Command({ resume }) 恢复中断的图 传入审批结果
thread_id 会话隔离标识 用户 session ID

三种模式对比

模式 核心 API 典型场景
分支 addConditionalEdges → 不同节点 输入分类、意图路由
循环 addConditionalEdges → 回到自己 重试、轮询、等待条件满足
持久化 MemorySaver + thread_id 多轮对话、断点续传、多用户
中断 interrupt + Command 人工审批、确认操作

面试万能回答

LangGraph 的核心是有状态的图编排。用 Annotation 定义共享状态，用 StateGraph 组织节点和边，用 addConditionalEdges 实现分支和循环，用 checkpointer 实现状态持久化和多会话隔离，用 interrupt 实现人类审批节点。它和 LangChain 共享底层基础设施，但解决了 LangChain 解决不了的复杂流程控制问题。

六、给你的代码加记忆，就一行的事

const app = graph.compile();


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


import { SqliteSaver } from '@langgraph/checkpoint-sqlite';
const app = graph.compile({ checkpointer: new SqliteSaver() });

记住这个模式，所有 LangGraph 工作流都能一键升级为有记忆的版本。

English / 原文

上一篇我们搞定了 LangGraph 的线性流程编排和多 Agent 协作基础。这一篇深入 LangGraph 最核心的三个能力：条件分支 、循环重试 、状态持久化 ，最后再来一个硬核的 中断/恢复 实战。

一、条件分支：让工作流"会思考"

为什么需要分支？

线性流程只能 A → B → C 一路走到黑。真实场景中，Agent 需要根据输入决定走哪条路 ——用户问数学题走计算器，闲聊走对话，代码走 Code Interpreter。

LangGraph 用 addConditionalEdges 实现分支。

完整代码

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


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


const router = (state) => {
  const isMath = /[+\-*/]/.test(state.query);
  return { route: isMath ? "math" : "chat" };
};


const mathNode = (state) => {
  try {
    return { answer: String(eval(state.query)) };
  } catch {
    return { answer: "表达式无法计算" };
  }
};


const chatNode = (state) => ({ answer: `你说的是：${state.query}` });


const graph = new StateGraph(StateAnnotation)
  .addNode("router", router)
  .addNode("math", mathNode)
  .addNode("chat", chatNode)
  .addEdge(START, "router")
  .addConditionalEdges("router", (state) => state.route, {
    math: "math",
    chat: "chat"
  })
  .addEdge("math", END)
  .addEdge("chat", END)
  .compile();

流程图

graph TD;
    __start__([__start__]) --> router{router}
    router -->|route=math| math[math]
    router -->|route=chat| chat[chat]
    math --> __end__([__end__])
    chat --> __end__

执行过程

await graph.invoke({ query: "1+2" });





await graph.invoke({ query: "你好" });

面试高频问：addConditionalEdges 和 addEdge 有什么区别？

API 作用 决定时机
addEdge(from, to) 固定连线，永远走 to 设计时 确定
addConditionalEdges(from, fn, map) 运行时根据状态动态选路 运行时 决定

一句话：addEdge 是"写死的路"，addConditionalEdges 是"到路口再决定"。

二、循环重试：让工作流"会坚持"

为什么需要循环？

Agent 执行不是一次就成功的——API 调用超时、LLM 输出格式不对、用户确认不通过……都需要重试机制 。

LangGraph 用 addConditionalEdges 让节点回到自己 ，形成循环。

完整代码

const StateAnnotation = Annotation.Root({
  tries:   Annotation({ reducer: (_prev, next) => next, default: () => 0 }),
  ok:      Annotation({ reducer: (_prev, next) => next, default: () => false }),
  message: Annotation({ reducer: (_prev, next) => next, default: () => "" })
});

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();

流程图

graph TD;
    __start__([__start__]) --> attempt[attempt]
    attempt -->|ok=false| attempt
    attempt -->|ok=true| __end__([__end__])

执行过程

await graph.invoke({ tries: 0 });

面试高频问：分支和循环用的是同一个 API？

是的，都是addConditionalEdges。

区别在于 map 里映射的目标：

{ math: "math", chat: "chat" }


{ retry: "attempt", done: END }

一句话：分支是"走哪条路"，循环是"要不要重来"，底层同一个 API，映射目标不同。

三、状态持久化：让工作流"有记忆"

没有持久化的问题

每次 graph.invoke() 状态都从默认值开始，相当于每次都失忆 。

第1次调用 → visitCount=0 → 结果1
第2次调用 → visitCount=0 → 结果1   ← 又从头开始了
第3次调用 → visitCount=0 → 结果1

MemorySaver：一行代码加记忆

import { MemorySaver } from '@langchain/langgraph';

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

加上 checkpointer 后，每次 invoke 结束自动保存状态快照，下次同一会话调用时自动恢复。

完整代码

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

const StateAnnotation = Annotation.Root({
  visitCount: Annotation({ reducer: (_prev, next) => next, default: () => 0 }),
  message:    Annotation({ reducer: (_prev, next) => next, default: () => "" })
});

function recordVisit(state) {
  const visitCount = state.visitCount + 1;
  const message = visitCount === 1
    ? "这是你在本会话里第1次进入。"
    : `这是你在本会话里${visitCount}次进入`;
  return { visitCount, message };
}

const graph = new StateGraph(StateAnnotation)
  .addNode("recordVisit", recordVisit)
  .addEdge(START, "recordVisit")
  .addEdge("recordVisit", END);

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

多用户会话隔离

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

await app.invoke({}, user1);  
await app.invoke({}, user1);  
await app.invoke({}, user1);  
await app.invoke({}, user2);

thread_id 就是会话标识 ——不同用户、不同对话，状态完全隔离。

面试高频问：MemorySaver 和数据库存储有什么区别？

| MemorySaver| 数据库存储
---|---|---
存储位置| 进程内存| SQLite / Redis / PostgreSQL
重启后| 丢失| 保留
适用场景| 开发调试、无状态要求| 生产环境、需要持久化
实现方式| LangGraph 内置| 需要换 Checkpointer 实现

生产环境把 MemorySaver 换成 SqliteSaver 或 RedisSaver，API 完全一样，只需改一行 import。

四、中断与恢复：让工作流"会等待"

这是 LangGraph 最强的能力之一——让图暂停，等人类输入后再继续 。

场景：转账确认

import {
  Annotation, END, START, StateGraph,
  Command,        
  interrupt,      
  MemorySaver
} from '@langchain/langgraph';
import { createInterface } from "node:readline/promises";

const StateAnnotation = Annotation.Root({
  actionSummary: Annotation({ reducer: (_prev, next) => next, default: () => "" }),
  userInput:     Annotation({ reducer: (_prev, next) => next, 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() });

流程图

graph TD;
    __start__([__start__]) --> showTransfer[showTransfer]
    showTransfer --> waitConfirm[waitConfirm]
    waitConfirm -.->|interrupt: 等待输入| waitConfirm
    waitConfirm --> __end__([__end__])

执行过程：两阶段调用

const config = { configurable: { thread_id: "interrupt-demo" } };


const paused = await graph.invoke({}, config);
console.log("待你确认:", paused.__interrupt__?.[0]?.value);





const rl = createInterface({ input: process.stdin, output: process.stdout });
const line = (await rl.question("> ")).trim();
await rl.close();

const done = await graph.invoke(new Command({ resume: line }), config);
console.log("done:", done);

面试高频问：interrupt 和普通暂停有什么区别？

| interrupt| 普通暂停
---|---|---
状态保存| 自动保存到 checkpointer| 丢失
恢复方式| new Command({ resume })| 无法恢复
多次中断| 支持，状态链式保存| 不支持
生产可用| 是，支持数据库持久化| 仅开发调试

一句话：interrupt 是 LangGraph 的"人工审批节点"——暂停时状态自动存档，恢复时用 Command 传入人类决策，整个过程可审计、可持久化。

五、核心概念总结

API 速查表

API 作用 类比
Annotation.Root({}) 定义状态结构 数据库建表
StateGraph 创建流程编排器 画流程图
.addNode(name, fn) 添加处理节点 流程图里的方框
.addEdge(from, to) 固定连线 "做完 A 做 B"
.addConditionalEdges(from, fn, map) 条件分支/循环 "如果...就..." / "重来"
.compile({ checkpointer }) 编译 + 注入持久化 编译 + 配置运行时
interrupt({}) 中断等待人类输入 审批暂停点
new Command({ resume }) 恢复中断的图 传入审批结果
thread_id 会话隔离标识 用户 session ID

三种模式对比

模式 核心 API 典型场景
分支 addConditionalEdges → 不同节点 输入分类、意图路由
循环 addConditionalEdges → 回到自己 重试、轮询、等待条件满足
持久化 MemorySaver + thread_id 多轮对话、断点续传、多用户
中断 interrupt + Command 人工审批、确认操作

面试万能回答

LangGraph 的核心是有状态的图编排 。用 Annotation 定义共享状态，用 StateGraph 组织节点和边，用 addConditionalEdges 实现分支和循环，用 checkpointer 实现状态持久化和多会话隔离，用 interrupt 实现人类审批节点。它和 LangChain 共享底层基础设施，但解决了 LangChain 解决不了的复杂流程控制 问题。

六、给你的代码加记忆，就一行的事

const app = graph.compile();


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


import { SqliteSaver } from '@langgraph/checkpoint-sqlite';
const app = graph.compile({ checkpointer: new SqliteSaver() });

记住这个模式，所有 LangGraph 工作流都能一键升级为有记忆的版本。

觉得有帮助的话，点个赞👍支持一下～