最新下载
热门教程
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
LangGraph 工作流实战：从 StateGraph 入门到多 Agent 编排与恢复
时间：2026-09-11 15:46:01 编辑：袖梨 来源：一聚教程网
当单个 Agent 同时承担过多工具、提示词和业务职责时，上下文成本与决策干扰会随之增加。LangGraph 提供了基于状态图的工作流组织方式，让任务能够按节点拆分，并支持分支、循环、持久化和人工确认。下面将从 StateGraph 的基本结构开始，逐步搭建可运行的多 Agent 编排流程。
LangGraph 实战：从 StateGraph 基础到多 Agent 编排，分支循环中断恢复全掌握
。LangChain 的工作流是"线性的"——一条直线走到底；LangGraph 的工作流是"网状的"——可以分支、循环、中断、恢复。本文从"为什么需要多 Agent"讲起（上下文开销 vs 分工协作），然后手写五个实战案例：基础状态图（节点+边+状态）、条件路由（数学题 vs 聊天）、循环重试（失败自动重来）、MemorySaver 持久化（跨会话记忆）、interrupt 中断恢复（确认场景）。全部代码可运行，建议收藏后动手实操。
一、为什么需要多 Agent？
1.1 单 Agent 的上下文开销
单 Agent 架构的问题：
所有工具描述、所有功能 Prompt 都塞进 system prompt
system prompt = {
...工具1的描述
...工具2的描述
...工具3的描述
...功能A的prompt
...功能B的prompt
...功能C的prompt
}
问题一：Token 消耗更高
→ 每次调用都要带上全部描述
→ 实际只执行一个功能
→ 90% 的 prompt 是冗余的
问题二：无关信息干扰
→ 大量无关的 prompt 文本
→ 干扰 LLM 的注意力
→ 思考效率低
→ 更容易出错
类比：
→ 每个部门开会都全员参加
→ 研发会议带销售、市场、客服一起
→ 效率低下，还互相干扰1.2 多 Agent 的拆分思路
如何拆分多个 Agent？ 每个 Agent 只保留自己需要的 Prompt： 编程 Agent → 只需要编程相关的 prompt 测试 Agent → 只需要测试相关的 prompt 验证 Agent → 只需要验证相关的 prompt 优势： → 执行功能时消耗的 token 更少 → 没有无关信息干扰 → 准确率更高 核心公式： Agent = LLM（大脑） + Harness（tool + mcp + rag + skill ...） ┌──────────────────────────────────────────────────────────┐ │ 单 Agent vs 多 Agent │ │ │ │ 单 Agent： │ │ → 只有一个 LLM 大脑 │ │ → 需要一步步思考，调用 tool │ │ → 串行处理，所有能力塞进一个大脑 │ │ → 上下文开销大，容易混乱 │ │ │ │ 多 Agent： │ │ → 多个大脑，并行思考 │ │ → 主 Agent 下发任务 │ │ → 子 Agent 并行处理完成后返回 │ │ → 每个大脑选择适合的模型 │ │ → Agent 组合式，按需加载，动态加载 │ └──────────────────────────────────────────────────────────┘
1.3 多 Agent 分工协作
多 Agent 分工合作的经典场景： 主 Agent（项目经理） │ ├── 编程 Agent：负责写代码 ├── 测试 Agent：编写测试代码（TDD） └── 验证 Agent：验证代码是否符合预期 │ └── 告诉主 Agent："通过了！" 流程： → 主 Agent 拆解任务 → 编程 Agent 写代码 → 测试 Agent 写测试（TDD 先写测试） → 验证 Agent 验证代码 → 全部通过 → 回报主 Agent
1.4 多 Agent 的三个核心理由
为什么复杂 Agent 产品基本都是多 Agent 架构？ ① 决策准确率高，Token 消耗更低 → 每个 Agent 只带必要的最少 Prompt → 没有冗余信息干扰 → 调用 LLM 次数变多，但总 token 更省 → 准确率反而更高 ② 并行思考和任务处理 → 主管分派子任务 → 子 Agent 并行处理 → 整体效率更高 ③ 多角色互相讨论，纠错能力更强 → 不同角色的 Agent 互相审查 → AutoGen 的"法庭"模式 → 多个 Agent 辩论，互相纠错 → 最终结论更可靠 这就是 LangGraph 存在的意义： → 把多个 Agent 组织成"网状工作流" → 分支、循环、并行、中断 → 多 Agent 协作的编排框架
二、LangChain → LangGraph：从线性到网状
2.1 LangChain 的基础模块
LangChain 提供的基础能力： ┌──────────────────────────────────────────────────────────┐ │ llm api → 大模型调用接口 │ │ document loaders → 文档加载器 │ │ splitter → 文本分割器 │ │ embedding → 向量化 │ │ vector store → 向量数据库 │ │ output parser → 输出解析器 │ │ memory → 记忆管理 │ │ ... → 更多基础模块 │ │ │ │ → 这些是"积木" │ │ → 负责单个能力的封装 │ │ → LangChain 是工具库 │ └──────────────────────────────────────────────────────────┘
2.2 线性 vs 网状
LangChain 工作流：线性（一条直线） A → B → C → D │ │ │ │ └───┴───┴───┘ 只能串行，不能回头，不能分支 LangGraph 工作流：网状（灵活多路） ┌─→ B ─┐ A ──→ │ ├──→ D └─→ C ─┘ 可以分支、循环、并行、中断、恢复 ┌──────────────────────────────────────────────────────────┐ │ 线性 vs 网状 │ │ │ │ LangChain LangGraph │ │ ────────────────────────────────────────────────── │ │ 编排方式 线性（一条线） 网状（多路） │ │ 分支 不支持 条件路由 │ │ 循环 不支持 回环重试 │ │ 并行 不支持 并行节点 │ │ 中断 不支持 interrupt │ │ 状态持久化 不支持 MemorySaver │ │ 场景 简单流程 复杂多 Agent │ │ 演进 简单 Agent → 复杂多 Agent 协作 │ │ │ │ LangGraph = 工作节点 + 组织方式（API） │ │ → 简单 Agent 可以用，复杂多 Agent 更要用 │ └──────────────────────────────────────────────────────────┘
2.3 网状工作流编排的四大要素
LangGraph 的核心 API： ┌──────────────────────────────────────────────────────────┐ │ ① 开始节点（START） │ │ → 工作流的入口 │ │ → 初始状态从这里进入 │ │ │ │ ② 工作节点（Node） │ │ → 每个节点有明确的职责 │ │ → 函数就是节点 │ │ → 接收 state，返回新 state │ │ │ │ ③ 边（Edge） │ │ → 连接工作节点 │ │ → 决定流程走向 │ │ → 可以是固定边、条件边 │ │ │ │ ④ 结束节点（END） │ │ → 工作流的出口 │ │ → 最终状态在这里输出 │ │ │ │ 流程： │ │ START → 节点 → 节点 → ... → END │ │ （初始状态）（职责）（边）（最终状态） │ └──────────────────────────────────────────────────────────┘
三、基础图实战：StateGraph 的 Hello World
3.1 完整代码
// basic-graph.mjs — 最基础的状态图
import {
Annotation, // 工作流的状态值的描述（数据 state）
END, // 结束节点
START, // 开始节点
StateGraph // 状态图，流程编排器，节点的组织
} from '@langchain/langgraph';
// Annotation 声明一个字段（数据部分）
const StateAnnotation = Annotation.Root({
text: Annotation({
// _prev: 来到当前节点之前的状态，next: 当前节点得到的状态
// reducer: 怎么处理状态的改变
reducer: (_prev, next) => next, // 类似数组 reduce，text 状态如何变
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(); // 编译工作流，执行
// 可视化：mermaid 文本画图工具
// 简单的 markdown，自动生成流程图
const drawable = await graph.getGraphAsync();
const mermaid = drawable.drawMermaid({ withStyles: true });
console.log(mermaid);
const result = await graph.invoke({ text: 'hello' });
console.log(result);运行流程：
invoke({ text: 'hello' })
│
▼
START（初始状态: text = "hello"）
│
▼
step1 → text: "hello -> step1"
│
▼
step2 → text: "hello -> step1 -> step2"
│
▼
END → 返回 { text: "hello -> step1 -> step2" }
最终输出：
{ text: 'hello -> step1 -> step2' }生成的 mermaid 图： graph TD; __start__ --> step1; step1 --> step2; step2 --> __end__; → 可视化整个节点的流转关系 → mermaid 是文本画图工具 → 简单的 markdown 自动生成流程图 → 方便调试和展示工作流
3.2 Annotation：状态声明
Annotation 的作用：
→ 描述工作流的状态结构
→ 定义有哪些字段
→ 定义字段如何变化（reducer）
→ 定义字段的默认值（default）
const StateAnnotation = Annotation.Root({
text: Annotation({
reducer: (_prev, next) => next,
default: () => '',
})
});
字段名：text
→ 状态里只有一个 text 字段
reducer: (_prev, next) => next
→ 状态如何更新
→ _prev: 旧值，next: 新值
→ (_, next) => next：直接覆盖
→ 类似数组 reduce，所以叫 reducer
default: () => ''
→ 初始默认值
→ 状态未初始化时的兜底3.3 节点：函数就是节点
节点的本质：
const step1 = (state) => ({ text: `${state.text} -> step1` });
→ 节点是一个普通函数
→ 入参：当前状态 state
→ 返回值：要更新的状态片段
→ 返回的对象会经过 reducer 处理合并进状态
节点职责：
→ 接收状态
→ 处理业务逻辑
→ 返回新状态
类比：
→ 状态 = 流水线上的"工件"
→ 节点 = 流水线上的"工位"
→ 每个工位加工后交给下一个工位3.4 图的构建链式调用
StateGraph 的构建：
new StateGraph(StateAnnotation) // 传入状态定义
.addNode('step1', step1) // 注册节点
.addNode('step2', step2)
.addEdge(START, 'step1') // 连接边
.addEdge('step1', 'step2')
.addEdge('step2', END)
.compile(); // 编译
addNode(name, fn)
→ 注册节点
→ name: 节点名（字符串标识）
→ fn: 节点函数
addEdge(from, to)
→ 连接两个节点
→ START / END 是内置的特殊节点
→ 固定走向，没有分支
compile()
→ 编译工作流
→ 校验图的完整性
→ 生成可执行的图四、条件路由：让工作流自己"选路"
4.1 场景设计
场景：一个入口，两个出口 用户输入： → "1+2"（数学表达式）→ 走 math 节点，计算 → "你好"（普通聊天）→ 走 ch@t 节点，回复 流程图： ┌──────────────────────────────────────────────────────────┐ │ │ │ START ──→ router ──→ math ──→ END │ │ │ │ │ └──→ ch@t ──→ END │ │ │ │ router 是"路口"： │ │ → 根据 query 判断走哪条路 │ │ → 含 + - * / 字符 → math │ │ → 否则 → ch@t │ └──────────────────────────────────────────────────────────┘
4.2 完整代码
// conditional-routing.mjs — 条件路由
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: () => 'ch@t'
}),
answer: Annotation({
reducer: (_prev, next) => next,
default: () => ''
})
});
// 路由节点：决定下一步怎么走
const router = (state) => {
// 判断是否是数学表达式（含 + - * / 字符）
const isMath = /[+-*/]/.test(state.query);
// 如果是数学问题走 math，否则走 ch@t
return { route: isMath ? 'math' : 'ch@t' }
};
// 数学计算节点
const mathNode = (state) => {
try {
return { answer: String(eval(state.query)) } // "1+2" → 3 → "3"
} catch {
return { answer: '表达式无法计算' }
}
};
// 聊天节点
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') // 固定走向：先到 router
// 条件跳转：根据 state.route 的值判断，去 math 还是 ch@t
.addConditionalEdges('router', (state) => state.route, {
math: 'math',
ch@t: 'ch@t'
})
.addEdge('math', END)
.addEdge('ch@t', END)
.compile();
console.log('result:', await graph.invoke({ query: '你好' }))
// { query: '你好', route: 'ch@t', answer: '你说的是：你好' }
console.log('result:', await graph.invoke({ query: '1+2' }))
// { query: '1+2', route: 'math', answer: '3' }4.3 addConditionalEdges：条件边
条件边的三个参数：
.addConditionalEdges(
'router', // ① 从哪个节点出发
(state) => state.route, // ② 决定函数：返回路由值
{ // ③ 路由表：值 → 目标节点
math: 'math',
ch@t: 'ch@t'
}
)
执行流程：
router 节点执行完
│
▼
决定函数执行：(state) => state.route
│
├── 返回 "math" → 查路由表 → 去 math 节点
│
└── 返回 "ch@t" → 查路由表 → 去 ch@t 节点
类比：
→ 路由器（router）根据数据包的"目的地"选择转发路径
→ 这就是"路由"（routing）名字的由来
与 addEdge 的区别：
→ addEdge：固定走向（一条路）
→ addConditionalEdges：根据状态动态选路（多条路）4.4 eval()：教学用，生产禁用！
数学节点用到了 eval()：
const mathNode = (state) => {
try {
return { answer: String(eval(state.query)) }
} catch {
return { answer: '表达式无法计算' }
}
};
eval('1+2') → 3
eval('5*3') → 15
eval() 是什么？
→ 把传入的字符串当作 JS 代码执行
→ 返回代码执行结果
→ 非常强大，也非常危险！
⚠️ 安全警告：
┌──────────────────────────────────────────────────────────┐
│ eval() 的危险： │
│ │
│ 输入 "process.exit()" │
│ → eval 直接执行 → 程序退出！ │
│ │
│ 输入 "require('fs').readFileSync('/etc/passwd')" │
│ → eval 直接执行 → 读取系统文件！ │
│ │
│ 输入 "1; 恶意代码" │
│ → eval 直接执行 → 任意代码注入！ │
│ │
│ 这就是代码注入（Code Injection）攻击 │
│ │
│ ✅ 安全替代方案： │
│ → 数学表达式解析库（mathjs 等） │
│ → 只解析数学表达式，不执行代码 │
│ → 生产环境禁止使用 eval │
└──────────────────────────────────────────────────────────┘
教学目的：
→ 演示"分支路由"的核心思想
→ 用 eval 简化代码，聚焦路由逻辑
→ 理解危险，比回避危险更重要五、循环重试：失败自动重来
5.1 场景设计
场景：重试直到成功 → 某些操作可能失败（网络、临时错误） → 需要自动重试 → 达到最大次数才算结束 流程图： ┌──────────────────────────────────────────────────────────┐ │ │ │ START ──→ attempt ──→ 成功？ │ │ ↑ │ │ │ │ ├─ 成功 → END │ │ │ │ │ │ └── 失败（retry）←───┘ │ │ │ │ attempt 节点： │ │ → tries + 1 │ │ → 达到 3 次 → 成功 → 结束 │ │ → 未达到 → 失败 → 回到 attempt 再来 │ └──────────────────────────────────────────────────────────┘
5.2 完整代码
// loop-retry.mjs — 循环重试
import {
Annotation,
END,
START,
StateGraph
} from '@langchain/langgraph';
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: () => ''
})
});
// 尝试节点：每来一次尝试次数 +1
const attempt = (state) => {
const tries = state.tries + 1;
const ok = tries >= 3; // 第 3 次必然成功
return {
tries,
ok,
message: ok ? `第${tries}次成功` : `第${tries}次失败`
}
};
const graph = new StateGraph(StateAnnotation)
.addNode('attempt', attempt)
.addEdge(START, 'attempt')
// 条件边：成功 → 结束，失败 → 回到 attempt（形成循环）
.addConditionalEdges('attempt', (state) => state.ok ? 'done' : 'retry', {
retry: 'attempt',
done: END
})
.compile();
console.log('result:', await graph.invoke({ tries: 0 }))执行过程（invoke({tries: 0})）：
第 1 次进入 attempt：
→ tries: 0 + 1 = 1
→ ok: 1 >= 3 → false
→ message: "第1次失败"
→ 条件边 → 返回 "retry" → 回到 attempt
第 2 次进入 attempt：
→ tries: 1 + 1 = 2
→ ok: 2 >= 3 → false
→ message: "第2次失败"
→ 条件边 → 返回 "retry" → 回到 attempt
第 3 次进入 attempt：
→ tries: 2 + 1 = 3
→ ok: 3 >= 3 → true
→ message: "第3次成功"
→ 条件边 → 返回 "done" → END
最终结果：
{ tries: 3, ok: true, message: '第3次成功' }循环的本质：
条件边指向自己 = 循环
.addConditionalEdges('attempt', ..., {
retry: 'attempt', // 回到自己 → 循环！
done: END
})
→ 条件边不仅可以指向其他节点
→ 还可以指回自己
→ 这就形成了"循环"
→ 循环必须有出口（done → END）
→ 否则死循环
实际应用：
→ API 调用失败重试
→ LLM 输出校验不过重试
→ 指数退避重试（间隔递增）六、MemorySaver：状态持久化
6.1 为什么要持久化？
没有持久化的问题： → 每次 invoke 都是全新开始 → 状态在调用结束后丢失 → 无法跨会话记忆 场景：访问计数 → 第 1 次访问：显示"第1次进入" → 第 2 次访问：应该显示"第2次进入" → 但状态丢了，每次都显示"第1次" 用 MemorySaver 保存状态： → 把 state 保存到内存里 → 下次基于上次的 state 继续执行 → 跨调用保持记忆 更复杂的场景： → Agent 执行中断、失败、暂停、需要授权 → MemorySaver 保存状态 → 之后可以继续运行
6.2 完整代码
// checkpointer-memory.mjs — MemorySaver 状态持久化
import {
Annotation,
END,
START,
MemorySaver, // 内存保存器
StateGraph
} from '@langchain/langgraph';
const StateAnnotation = Annotation.Root({
// session 相关：某人访问次数
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
});
// 多用户：thread_id 区分会话
const user1 = { configurable: { thread_id: '用户-小张' } };
const user2 = { configurable: { thread_id: '用户-小李' } };
const res1 = await app.invoke({}, user1);
console.log(res1); // visitCount: 1, "第1次进入"
const res2 = await app.invoke({}, user1);
console.log(res2); // visitCount: 2, "第2次进入"
const res3 = await app.invoke({}, user1);
console.log(res3); // visitCount: 3, "第3次进入"
const res4 = await app.invoke({}, user2);
console.log(res4); // visitCount: 1, "第1次进入"（新会话）6.3 MemorySaver 与 thread_id
核心机制：
checkpointer（检查点）
→ 在每次节点执行后保存状态快照
→ 相当于游戏里的"存档"
→ 下次可以从存档继续
compile({ checkpointer })
→ 把 checkpointer 挂到图上
→ 图就具备持久化能力
thread_id（线程 ID）
→ 区分不同的会话
→ 每个 thread_id 是独立的记忆空间
→ 小张的访问计数不影响小李
┌──────────────────────────────────────────────────────────┐
│ 执行过程： │
│ │
│ user1 第 1 次 invoke │
│ → 会话 "用户-小张" 状态为空 │
│ → visitCount: 0 + 1 = 1 │
│ → 保存 → "第1次进入" │
│ │
│ user1 第 2 次 invoke │
│ → 从内存读取上次状态 │
│ → visitCount: 1 + 1 = 2 │
│ → 保存 → "第2次进入" │
│ │
│ user1 第 3 次 invoke │
│ → visitCount: 2 + 1 = 3 │
│ → "第3次进入" │
│ │
│ user2 第 1 次 invoke │
│ → 新线程 "用户-小李" │
│ → 状态独立，从 0 开始 │
│ → "第1次进入" │
└──────────────────────────────────────────────────────────┘6.4 持久化的进阶
MemorySaver 的局限： → 保存在内存里 → 进程重启就丢失 → 只适合开发测试 生产级持久化： ┌──────────────────────────────────────────────────────────┐ │ 持久化方案： │ │ │ │ MemorySaver（内存） │ │ → 开发测试用 │ │ → 进程重启丢失 │ │ │ │ SQLite / Redis（数据库） │ │ → 生产环境用 │ │ → 重启不丢失 │ │ → 支持分布式共享 │ │ │ │ 用途： │ │ → Agent 执行中断、失败、暂停 │ │ → 需要授权（human-in-the-loop） │ │ → 保存状态后下次继续运行 │ │ → 这就是 harness 的中断、恢复机制 │ └──────────────────────────────────────────────────────────┘
七、interrupt：中断与恢复
7.1 场景设计
场景：需要人工确认
→ AI 帮你，不能直接转！
→ 必须停下来等用户确认
→ 用户输入确认后才继续
流程图：
┌──────────────────────────────────────────────────────────┐
│ │
│ START ──→ showTransfer ──→ waitConfirm ──→ END │
│ │ │
│ │ interrupt（中断） │
│ ▼ │
│ 等待用户输入确认... │
│ │ │
│ │ Command({resume}) │
│ ▼ │
│ 继续执行 │
│ │
│ 这就是 Human-in-the-Loop（人在回路） │
│ → AI 执行到关键步骤，停下来等人工批准 │
│ → 这是 Agent 工程的安全底线 │
└──────────────────────────────────────────────────────────┘7.2 完整代码
// graph-interrupt.mjs — 中断与恢复
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() });
// 第一次 invoke：会中断在 waitConfirm
const config = { configurable: { thread_id: 'interrupt-demo' } };
const paused = await graph.invoke({}, config);
console.log('待你确认', paused.__interrupt__?.[0]?.value);
// 待你确认 { hint: '...', actionSummary: '向张三 $100' }
// 等用户输入（模拟人工确认）
const rl = createInterface({
input: process.stdin,
output: process.stdout
});
const line = (await rl.question('> ')).trim();
console.log('你输入了', line);
await rl.close();
// 第二次 invoke：用 Command({ resume }) 恢复执行
const done = await graph.invoke(new Command({ resume: line }), config);
console.log('done:', done);7.3 中断机制详解
interrupt 的完整流程：
第一次 invoke：
→ START → showTransfer（展示信息）
→ waitConfirm 执行到 interrupt(...)
→ 图在此"暂停"
→ 返回 paused 对象
→ paused.__interrupt__ 包含中断信息
{ hint: '...', actionSummary: '向张三 $100' }
人工介入：
→ 用户看到信息
→ 在终端输入确认或备注
→ 这是"人在回路"（Human-in-the-Loop）
第二次 invoke：
→ new Command({ resume: line })
→ resume 携带用户输入
→ 图从暂停处"恢复"
→ waitConfirm 收到用户输入
→ 继续执行到 END
┌──────────────────────────────────────────────────────────┐
│ 为什么需要 checkpointer？ │
│ │
│ 中断时： │
│ → 图执行到一半停止 │
│ → 状态必须保存！ │
│ → 否则恢复时不知道执行到哪了 │
│ │
│ MemorySaver 的作用： │
│ → 保存中断时的状态快照 │
│ → 恢复时从快照继续 │
│ → 中断 + 持久化 = 可恢复的 Agent │
│ │
│ 场景： │
│ → 审批流程（AI 起草 → 人工批准 → 执行） │
│ → 支付确认（AI 生成订单 → 用户确认 → 扣款） │
│ → 敏感操作（AI 生成代码 → 人工审核 → 合并） │
└──────────────────────────────────────────────────────────┘7.4 Command：控制图的指令
Command 是 LangGraph 的控制指令：
new Command({ resume: line })
→ resume: 恢复中断时传给 interrupt 的返回值
→ waitConfirm 里 interrupt() 的返回值就是 line
→ 实现了"外部输入注入图中"
Command 的其他用途：
→ Command({ goto: '节点名' })：指定跳转
→ Command({ update: {...} })：更新状态
→ Command({ resume: data })：恢复中断
本质：
→ 图是"状态机"
→ 中断 = 停在某个状态
→ Command = 外部给状态机发指令
→ resume = 告诉它"继续，这是你要的数据"八、总结
8.1 知识体系图
LangGraph 完整实战
│
├── 为什么需要多 Agent
│ ├── 单 Agent 上下文开销（冗余 prompt + 干扰）
│ ├── 每个 Agent 只带必要 prompt → 省 token + 高准确率
│ ├── Agent = LLM（大脑）+ Harness（tool/mcp/rag/skill）
│ ├── 主 Agent 下发任务，子 Agent 并行处理
│ └── 三大理由：省 token / 并行 / 互相纠错
│
├── LangChain → LangGraph
│ ├── LangChain：线性工作流 + 基础模块
│ ├── LangGraph：网状工作流编排
│ ├── 四大要素：START / 节点 / 边 / END
│ └── 场景：简单 Agent → 复杂多 Agent 协作
│
├── 基础图 StateGraph
│ ├── Annotation.Root 状态声明（reducer / default）
│ ├── 函数就是节点（state → new state）
│ ├── addNode / addEdge / compile / invoke
│ └── mermaid 可视化（getGraphAsync + drawMermaid）
│
├── 条件路由
│ ├── router 节点判断意图
│ ├── addConditionalEdges(节点, 决定函数, 路由表)
│ ├── 正则判断数学表达式
│ ├── eval() 教学演示（生产禁用！代码注入风险）
│ └── 分支：math / ch@t 两个出口
│
├── 循环重试
│ ├── 条件边指向自己 = 循环
│ ├── tries + 1 计数
│ ├── 成功 → done → END
│ ├── 失败 → retry → 回到 attempt
│ └── 应用：API 失败重试 / 输出校验重试
│
├── MemorySaver 持久化
│ ├── checkpointer 保存状态快照（存档）
│ ├── compile({ checkpointer })
│ ├── thread_id 区分多用户会话
│ ├── 跨调用保持记忆
│ └── 生产级：SQLite / Redis
│
├── interrupt 中断恢复
│ ├── interrupt({...}) 暂停图
│ ├── __interrupt__ 携带中断信息
│ ├── Command({ resume }) 恢复执行
│ ├── 必须配 checkpointer（保存中断状态）
│ └── 应用：审批 / 支付确认 / Human-in-the-Loop
│
└── 核心认知
├── 图 = 状态机（状态 + 节点 + 边）
├── 条件边 = 分支 + 循环
├── checkpointer = 存档机制
├── interrupt = 人在回路
└── 单 Agent 串行思考 → 多 Agent 网状协作8.2 一句话总结
LangGraph 是 LangChain 的网状升级：LangChain 的线性工作流只能 A→B→C 一条路走到底，LangGraph 的状态图支持分支、循环、中断、持久化——这正是复杂多 Agent 产品需要的编排能力。核心 API 只有四个要素：Annotation 声明状态（reducer 决定状态如何变）、函数即节点（state 进 state 出）、边连接节点（addEdge 固定走向 / addConditionalEdges 条件选路）、START/END 作为出入口。条件边指向自己是循环（失败重试），挂上 MemorySaver 是持久化（checkpointer 存档 + thread_id 分会话），interrupt 是中断（执行到关键步骤停下来等人工确认，Command({ resume }) 恢复）——这三个组合起来，就实现了 Agent 工程最核心的 Human-in-the-Loop 机制。多 Agent 为什么是趋势？单 Agent 把所有工具和 prompt 塞进一个大脑，token 浪费且互相干扰；多 Agent 各司其职，每个大脑只带最少 prompt，省 token、准确率高、还能并行和互相纠错。LangGraph 就是把这些"大脑"组织成网状工作流的编排框架。
相关文章
- tp5660路由器手机怎么设置（tp5660路由器手机设置方法） 09-11
- GoofCord：实践指南 09-11
- Lumina-Layers：实践指南 09-11
- moodleapp：实践指南 09-11
- zhihu-api：实践指南 09-11
- 不用常规写法：让 Agent 工具调用更稳定的 4 个实测技巧 09-11