LangGraph 从入门到进阶（Node.js + TypeScript）

2026/11/13 · by ZHANG XING YUN · 阅读 2 次 · 2455 字 · 预计阅读 7 分钟

LangGraph 解决的是 LangChain 不擅长的那类问题：需要循环、分支、持久化状态、人工审批的长时运行 Agent。LangChain 的 createAgent 底层就跑在 LangGraph 上，但当你想自己掌控控制流时，需要直接使用 LangGraph 的图 API。

适用场景：多步骤 Agent、人在回路、故障恢复、多智能体编排
技术栈：@langchain/langgraph（TypeScript 端），Node.js 22+
目标：工作流可控制、可恢复、可观测、可生产

一、先分清：LangChain 和 LangGraph 各管什么

这是最容易混淆的地方。官方定位很明确：

工具	角色	解决什么
LangChain	Agent 框架	模型、工具、提示词的统一抽象，`createAgent` 快速构建 Agent
LangGraph	编排运行时	持久化执行、流式输出、人在回路、状态管理
LangSmith	观测平台	追踪、评估、部署

LangGraph 不规定你用什么提示词、什么 Agent 架构。它只提供运行时原语：共享状态 + 节点（函数）+ 边（路由）。你可以在同一个图里混合确定性步骤（解析、检索）和 LLM 驱动的步骤（决策、生成）。

关键认知：LangChain 的 createAgent 底层就是 LangGraph。你不需要在两者之间“二选一”，而是根据控制粒度选择抽象层级。需要细粒度控制状态流转时，直接用 LangGraph。

二、核心概念：状态、节点、边

2.1 共享状态（State）

State 是图的“内存”，所有节点读写同一份数据。用 Zod schema 定义：

import { StateSchema } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  messages: z.array(z.any()).default(() => []),
  nextStep: z.string().optional(),
  retryCount: z.number().default(0),
});

如果多个节点写同一个字段，需要指定 reducer 决定合并方式。默认是覆盖，消息列表通常需要追加：

import { ReducedValue } from "@langchain/langgraph";

const State = new StateSchema({
  messages: new ReducedValue(
    z.array(z.any()).default(() => []),
    { reducer: (left, right) => left.concat(right) }
  ),
});

2.2 节点（Node）

节点是一个函数：接收当前 State，返回部分更新。LangGraph 会自动合并到全局状态。

const callModel = async (state: typeof State.State) => {
  const response = await model.invoke(state.messages);
  return { messages: [response] };  // 只返回要更新的字段
};

节点是检查点和重试的边界。带 checkpointer 时，LangGraph 在节点边界保存状态，中断或失败后从该节点重新执行。这意味着：节点内的副作用必须幂等，因为恢复时节点会从头重跑。

2.3 边（Edge）

边决定控制流向：

类型	作用	用法
普通边	A 之后必到 B	`addEdge("A", "B")`
条件边	根据状态选择下一个节点	`addConditionalEdges("A", router)`
入口	START → 首个节点	`addEdge(START, "A")`
终止	节点 → END	`addEdge("A", END)`

一个节点不能同时使用静态边和动态路由（条件边/Command），混用会导致两条路径都执行。

三、入门：最小图

npm install @langchain/langgraph @langchain/openai zod

一个“判断奇偶并格式化”的确定性图，不涉及 LLM，先理解结构：

import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import * as z from "zod";

const State = new StateSchema({
  number: z.number(),
  isEven: z.boolean().optional(),
  message: z.string().optional(),
});

const checkEven = (state: typeof State.State) => ({
  isEven: state.number % 2 === 0,
});

const formatMessage = (state: typeof State.State) => ({
  message: state.isEven ? "偶数" : "奇数",
});

const graph = new StateGraph(State)
  .addNode("checkEven", checkEven)
  .addNode("formatMessage", formatMessage)
  .addEdge(START, "checkEven")
  .addEdge("checkEven", "formatMessage")
  .addEdge("formatMessage", END)
  .compile();

const result = await graph.invoke({ number: 7 });
console.log(result.message); // "奇数"

这就是 LangGraph 的全部骨架：定义状态 → 添加节点 → 连边 → 编译 → 调用。

四、控制流：分支、循环、并行

4.1 条件分支

根据状态决定走哪条路：

const router = (state: typeof State.State) => {
  return state.needsTool ? "useTool" : "directAnswer";
};

graph.addConditionalEdges("agent", router, {
  useTool: "toolNode",
  directAnswer: END,
});

4.2 循环

Agent 的典型模式：模型调用工具 → 回到模型 → 再决定是否继续。用条件边指回上游节点即可形成循环：

graph.addConditionalEdges("agent", shouldContinue, {
  continue: "tools",
  end: END,
});
graph.addEdge("tools", "agent");  // 循环回 agent

循环必须有终止条件，否则会无限执行。可以设置递归上限，或在路由函数中检查计数。

4.3 并行

一个节点有多条出边时，目标节点在下一个超级步并行执行。适合“同时检索多个数据源”这类场景。并行的结果需要 reducer 正确合并。

4.4 Send：动态并行

当并行数量运行时才知道（比如对列表每一项启动一个 worker），用 Send：

import { Send } from "@langchain/langgraph";

graph.addConditionalEdges("split", (state) => {
  return state.items.map(item => new Send("processItem", { item }));
});

五、持久化：Checkpointer 与 Store

LangGraph 提供两套互补的持久化机制：

	Checkpointer	Store
存什么	图的线程状态快照	应用自定义的键值数据
作用域	单个线程（thread）	跨线程
用途	对话连续性、人在回路、时间旅行、故障恢复	用户偏好、长期记忆、共享知识
使用方式	编译时传入，调用时带 `thread_id`	在节点内读写

5.1 Checkpointer

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

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

const config = { configurable: { thread_id: "user-123" } };
await graph.invoke({ messages: [{ role: "user", content: "你好" }] }, config);

MemorySaver 只存在内存中，进程重启就丢失。生产环境用 PostgresSaver 或 SqliteSaver。

5.2 查看与修改状态

带 checkpointer 时，可以读取线程历史和当前状态：

const currentState = await graph.getState(config);
const history = graph.getStateHistory(config);  // 时间旅行

人在回路中，用 updateState 修改状态后再恢复执行。

六、人在回路：中断与恢复

LangGraph 的核心能力之一：暂停执行、人工审批、再继续。

6.1 静态中断（编译时指定）

const graph = builder.compile({
  checkpointer,
  interruptBefore: ["dangerousNode"],  // 执行该节点前暂停
});

6.2 动态中断（节点内触发）

import { NodeInterrupt } from "@langchain/langgraph";

const sensitiveAction = (state: typeof State.State) => {
  if (!state.approved) {
    throw new NodeInterrupt("需要人工审批");
  }
  return { result: "已执行" };
};

6.3 恢复执行

// 首次执行，在中断处暂停
await graph.invoke(input, config);

// 审批后修改状态
await graph.updateState(config, { approved: true });

// 恢复执行，传 null 表示从检查点继续
await graph.stream(null, config);

关键点：thread_id 必须一致。中断状态保存在对应线程的检查点中。

七、多智能体编排模式

当单个 Agent 的提示词和工具过多，选择精度会下降。多智能体把任务拆给多个专家。

7.1 Supervisor 模式

中央 Supervisor 接收任务、分派给专家、汇总结果。控制流清晰，但 Supervisor 可能成为瓶颈。

// Supervisor 节点根据任务路由到不同专家
const supervisor = (state) => {
  if (state.task.includes("天气")) return { next: "weatherAgent" };
  if (state.task.includes("计算")) return { next: "mathAgent" };
  return { next: END };
};

7.2 Swarm 模式

没有中央协调者，Agent 之间通过 handoff（交接）传递控制权。适合客服、分诊这类对话走向不可预测的场景。需要设置最大交接深度防止循环。

7.3 选型建议

模式	适用	风险
Supervisor	子任务边界清晰	分解错误会级联
Swarm	对话式、路由场景	可能无限交接
层级型	大规模团队协作	复杂度高

从 Supervisor 开始，大多数场景够用。

八、生产化要点

8.1 持久化选型

环境	Checkpointer
开发/测试	`MemorySaver`
单机生产	`SqliteSaver`
分布式生产	`PostgresSaver`

thread_id 长度控制在 255 字符以内，否则 Postgres 会报错。

8.2 流式输出

graph.stream() 支持多种模式：

for await (const chunk of await graph.stream(input, {
  ...config,
  streamMode: "updates",  // 只发状态变更
})) {
  console.log(chunk);
}

模式包括 values（完整状态）、updates（增量）、messages（LLM token 流）、debug 等。

8.3 重试与幂等

节点级重试策略：

graph.addNode("flakyNode", fn, {
  retryPolicy: { maxAttempts: 3 },
});

副作用必须幂等：节点恢复时会重跑，外部 API 调用可能重复。用幂等键或 upsert 策略。

8.4 可视化

const drawable = await graph.getGraphAsync();
console.log(drawable.drawMermaid());

用于审查拓扑和调试路由。

九、常见问题

现象	原因	处理
中断后无法恢复	未传 `thread_id` 或 checkpointer 未配置	编译时传 checkpointer，调用时带一致的 `thread_id`
节点副作用重复执行	恢复时节点从头重跑	副作用做幂等处理
图无限循环	循环无终止条件	加条件边终止分支或递归上限
静态边和条件边混用	同一节点两种路由	只选一种
`MemorySaver` 重启后状态丢失	内存存储	生产用 `PostgresSaver`
并行节点结果覆盖	reducer 未配置	对并发写入的字段定义 reducer
`thread_id` 过长报错	Postgres 列长度限制	控制在 255 字符以内

十、FAQ

Q1：LangGraph 和 LangChain 的 createAgent 是什么关系？

createAgent 底层就是 LangGraph。用 createAgent 是“开箱即用”，用 LangGraph 是“自己搭”。需要自定义状态流转、多智能体、复杂中断逻辑时，直接用 LangGraph。

Q2：什么时候该用 LangGraph 而不是 LCEL？

LCEL 适合线性管道（prompt → model → parser）。一旦需要根据中间结果决定下一步、循环、人工审批、故障后从中间恢复，就该用 LangGraph。

Q3：节点和 Task 有什么区别？

节点是检查点边界，恢复时从头重跑。Task 是节点内部的细粒度检查点：一个节点里多个操作，每个包成 Task，恢复时可以跳过已完成的部分。有副作用或昂贵的操作应该用 Task。

Q4：如何实现“人工审批后继续”？

用 interruptBefore 或 NodeInterrupt 暂停，updateState 修改审批状态，stream(null, config) 恢复执行。

Q5：多智能体一定要用 LangGraph 吗？

不一定，但 LangGraph 的图结构和持久化让多智能体的状态管理、故障恢复、人工介入更容易实现。LangChain 也提供了 langgraph-supervisor 等包来简化。

Q6：Checkpointer 和 Store 都要用吗？

大多数应用两个都用：Checkpointer 管当前对话线程的状态，Store 管跨会话的长期记忆（用户偏好、知识）。