7. LangGraph

为什么需要 LangGraph —— 从 LCEL 的局限说起

LCEL 链本质上是一个 DAG（有向无环图），数据单向流动，每一步都往前走，永远不会回头。但下面这些场景，LCEL 做起来很别扭：

场景 LCEL 的困境 LangGraph 的解法
模型决定要不要调用工具，调用完还要把结果喂回去再问一次 需要手写循环，状态在循环外维护，容易写成"意大利面" 用条件边（conditional edge）直接表达"循环回到模型节点"
任务执行到一半，进程重启了 全部状态丢失，只能从头来 Checkpointer 自动保存每一步状态，重启后从断点恢复
某个工具调用有风险（发邮件、删数据），需要人工确认 没有原生机制可以"暂停等人" interrupt() 原生支持暂停-恢复
多个专精 Agent 需要互相转交任务 链与链之间没有"路由"概念 图的节点天然就是可路由的单元

LangGraph 的关键突破是条件边可以指向之前已经走过的节点，这样就在图里造出了一个"环"（cycle）。它的工作流程可以简化这样表示：

[start] → [node A] → [node B] → [end]
              ↑           │
              └───────────┘  （条件边：回到 A）

和 ai SDK 对照：

ai SDK 的 generateText({ tools, stopWhen }) 内部其实也是一个循环，但它是封装好的黑盒，你只能通过 stopWhen/onStepFinish 介入。LangGraph 把这个循环拆开给你看，你可以完全控制每一步。

安装

其实我们前面学习的时候已经安装过了，这里只是重复记录一遍。

npm install typescript tsx @types/node -D
# 核心依赖
npm install @langchain/langgraph @langchain/core langchain
# DeepSeek 模型供应商
npm install @langchain/deepseek
# 工具相关
npm install zod dotenv

LangGraph 的四个核心概念

  • 状态（State）：在节点间流转的数据（一个贯穿全图的"共享笔记本"，所有节点都从这里读、往这里写）。
  • 节点（Node）：一个函数，输入当前状态，返回"状态的局部更新"（不是覆盖，是合并）。
  • 边（Edge）：节点之间的跳转（节点之间的连接，决定执行顺序）。
  • 条件边（Conditional Edge）：根据当前状态决定下一步去哪个节点——这是"循环"和"分支"的来源。

状态(State)：流程里流转的数据

是什么：在节点之间传来传去的数据。整个图共享这一份数据,每个节点读取它、修改它。

类比：像快递包裹。包裹(状态)在不同分拣站(节点)之间传递,每个站可以往包裹里塞东西(更新状态),下一站能看见前面站塞的内容。

具体到 Agent：状态通常包含：

  • messages：对话历史(用户问题、AI 回复、工具调用、工具结果)
  • 其它你想记录的东西(当前步骤数、中间结果、标志位等)
自定义 State

可以用 Annotation 来定义状态的"形状(结构)"和"合并规则"（reducer）。Annotation.Root() 提供的是“怎么描述一个字段”的语法（类型 + 合并规则），但“这个状态里到底该有哪些字段”完全是我们根据业务建模出来的，没有任何预设的标准结构。比如：

// 定义状态：有两个字段 question 和 answer
const State = Annotation.Root({
    question: Annotation<string>,
    answer: Annotation<string[]>,
})
// 定义状态：有两个字段 fieldA 和 fieldB
const State = Annotation.Root({
    fieldA: Annotation<string>,
    fieldB: Annotation<number>,
})
// 定义状态：有两个字段 userIntent 和 steps
const State = Annotation.Root({
    userIntent: Annotation<string>,  // 普通字段：不写 reducer，默认行为是"后者覆盖前者"
    steps: Annotation<string[]>({
        reducer: (prev, next) => [...prev, ...next], // 显式声明"我要追加，不要覆盖"
        default: () => [],
    }),
})

看到这里，有人可能会有疑问，为什么上面的代码中的第三个定义的时候，steps 字段的写法和其它的不一样。因为咱们前面说过 State 会在节点之间传来传去，当一个节点函数 return 之后，框架立刻拿这个返回值去跟当前的全局状态做合并，生成新的全局状态。这个时候，就需要知道如何合并的了，如果我们不需要关注旧值，那么直接覆盖就行了，如果我们需要自己指定新值和旧值的合并规则，那么我们就需要按照前面 steps 字段那样的写法定义规则了。比如：

const State = Annotation.Root({
    // 情况1：纯覆盖，且不需要"默认初始值"（一开始就是 undefined，没关系）
    question: Annotation<string>,
    // 情况2：需要自定义合并逻辑，同时也需要初始值
    steps: Annotation<string[]>({
        reducer: (prev, next) => [...prev, ...next],
        // default可以没有，不过没有这个，第一次reducer执行时prev会是 undefined，[...undefined, ...next] 直接报错
        default: () => [],  
    })
})

reducer 函数中的两个参数的意义：

  • 第一个参数（通常叫 current 或 prev）：代表当前状态中已经存在的旧值。
  • 第二个参数（通常叫 update 或 next）：代表当前这个节点（Node）刚返回的新值。

而 reducer 函数的返回值，就是经过你自定义逻辑处理后，最终要写回状态中的新值。

内置的 MessagesAnnotation

做对话 Agent 时，状态几乎总是 messages(对话历史)。LangGraph 提供了预定义的 MessagesAnnotation，不用自己手动定义了。

// 使用我们前面自定义的 State
const graph = new StateGraph(State)
    .addNode('agent', agentNode)
    ...
// 使用内置的 MessagesAnnotation
const graph = new StateGraph(MessagesAnnotation)
    .addNode('agent', agentNode)
    ...

MessagesAnnotation 帮你处理了:

  • 消息追加(把新消息加到历史末尾)
  • 消息去重(同一 id 不重复加)
  • 消息删除(支持 RemoveMessage 移除旧消息)

其内部实现大致是这样的(伪代码)：

import { BaseMessage, RemoveMessage } from "@langchain/core/messages";
// （简化版，突出三个行为）
const messagesStateReducer = (currentMessages: BaseMessage[], newMessages: BaseMessage[]) => {
 for (const newMsg of newMessages) {
     // 1. 删除：如果是 RemoveMessage，把对应 id 的旧消息从数组里移除
     if (newMsg instanceof RemoveMessage) {
         currentMessages = currentMessages.filter(m => m.id !== newMsg.id);
         continue;
     }
     // 2. 更新：如果新消息的 id 和某条旧消息相同，替换它（不是追加）
     const existingIndex = currentMessages.findIndex(m => m.id === newMsg.id);
     if (existingIndex !== -1) {
         currentMessages[existingIndex] = newMsg;
         continue;
     }
     // 3. 追加：全新的消息（没有匹配 id），加到末尾
     currentMessages.push(newMsg);
 }
 return currentMessages;
};

完整的 MessagesAnnotation 使用示例：

/**
 * 演示 MessagesAnnotation 内置 reducer 的三种行为：追加、替换、删除
 * 不需要真实调用模型，纯粹操作 state，方便你看清楚 reducer 的合并逻辑。
 *
 * 运行方式：npx tsx demo.ts
 */
import { StateGraph, MessagesAnnotation, START, END } from "@langchain/langgraph";
import { HumanMessage, AIMessage, RemoveMessage } from "@langchain/core/messages";

// ---------- 节点1：追加（最常见的情况） ----------
// 返回全新的消息（没有指定 id，或 id 是新的），reducer 会把它加到数组末尾
async function appendNode(state: typeof MessagesAnnotation.State) {
    console.log(`\n[appendNode] 进入时消息数：${state.messages.length}`);
    return { messages: [new AIMessage("这是新追加的一条回复")] };
}

// ---------- 节点2：替换（id 相同，内容更新） ----------
// 故意拿"上一条消息的 id"，构造一条新内容的消息
// reducer 发现 id 已存在，会原地替换，而不是追加成第二条
async function replaceNode(state: typeof MessagesAnnotation.State) {
    const lastMessage = state.messages[state.messages.length - 1];
    console.log(`\n[replaceNode] 即将替换 id=${lastMessage.id} 的消息内容`);

    return {
        messages: [
            new AIMessage({
                id: lastMessage.id, // 关键：复用相同的 id
                content: "这条内容已经被替换掉了",
            }),
        ],
    };
}

// ---------- 节点3：删除 ----------
// 用 RemoveMessage 指定要删除的消息 id，reducer 会把它从数组里移除
async function removeNode(state: typeof MessagesAnnotation.State) {
    const firstMessage = state.messages[0];
    console.log(`\n[removeNode] 即将删除 id=${firstMessage.id} 的消息（最早的一条）`);

    return { messages: [new RemoveMessage({ id: firstMessage.id })] };
}

// ---------- 组装图：依次走 append → replace → remove ----------
const graph = new StateGraph(MessagesAnnotation)
    .addNode("append", appendNode)
    .addNode("replace", replaceNode)
    .addNode("remove", removeNode)
    .addEdge(START, "append")
    .addEdge("append", "replace")
    .addEdge("replace", "remove")
    .addEdge("remove", END);

const app = graph.compile();

// ---------- 打印工具：每一步状态长什么样 ----------
function printMessages(label: string, messages: typeof MessagesAnnotation.State["messages"]) {
    console.log(`\n===== ${label}（共 ${messages.length} 条） =====`);
    messages.forEach((m, i) => {
        console.log(`  [${i}] id=${m.id} type=${m._getType()} content="${m.content}"`);
    });
}

async function main() {
    const initial = {
        messages: [new HumanMessage({ id: "msg-1", content: "你好，第一条消息" })],
    };
    printMessages("初始状态", initial.messages);
    const result = await app.invoke(initial);
    printMessages("最终状态", result.messages);
}

main();

附：混合用法(消息 + 自定义字段)：用 MessagesAnnotation 展开，再加字段：

import { Annotation, MessagesAnnotation } from '@langchain/langgraph'

const State = Annotation.Root({
    ...MessagesAnnotation.spec,    // 展开内置的 messages 字段
    userName: Annotation<string>,  // 再加自己的字段
})

节点(Node)：干一件事的函数

是什么：图里的一个步骤，本质上就是一个普通函数。

它干什么：

  1. 接收当前的状态
  2. 干一件事(调 LLM、执行工具、算个东西、打个日志...)
  3. 返回状态的更新(只返回要改的部分)

关键：节点返回的不是"新状态"，而是"要更新的字段"。框架会拿你的返回值，去更新状态。

最基本的形式：
async function myNode(state: MyState): Promise<Partial<MyState>> {
  // 读 state，做点事
  return { someField: "新的值" }; // 返回"局部更新"
}
graph.addNode("my_node", myNode);

记住三条规矩：

  • 输入：当前全局状态（已经合并好的最新版本）
  • 输出：只返回你这个节点"负责"的字段，不用返回整个状态
  • 节点名 ≠ 函数名：addNode("my_node", myNode) 里第一个参数是图里引用这个节点用的字符串 id，边（addEdge、条件边）都靠这个字符串定位，跟函数本身叫什么名字没关系

节点函数的第二个参数：config

async function myNode(state: MyState, config: RunnableConfig) {
const threadId = config.configurable?.thread_id;
// config 里还能拿到 callbacks、tags 等，常用于往下传递追踪信息
}

实际开发中第二个参数用得不算多，但在需要拿 thread_id、做自定义 store 读写、或者要把 callback 传给内部调用的 LLM/工具时会用到。

addNode 的配置项：

addNode(name, fn, options?) 的第三个参数能配置不少生产级特性：

  • retryPolicy：自动重试

    import type { RetryPolicy } from "@langchain/langgraph";
workflow.addNode(
  "searchDocumentation",
  searchDocumentation,
  //   
  { retryPolicy: { maxAttempts: 3, initialInterval: 1.0 } }
);

  • timeout：超时控制

  • error_handler：重试耗尽后的兜底

    .addNode("call_llm", callLlm, {
  retryPolicy: { maxAttempts: 4, backoffFactor: 2.0 },
  timeout: { runTimeout: 30, idleTimeout: 5 },
  errorHandler: handleModelFailure,
})

起点和终点不是真节点，是常量

import { START, END } from "@langchain/langgraph";

START、END 是框架内置的特殊标记，不是你定义的函数，专门用来表示"图的入口/出口"。

异步节点(调 LLM、调 API)

节点可以是 async 函数：

const agentNode = async (state: typeof MessagesAnnotation.State) => {
    // 调 LLM
    const response = await model.invoke(state.messages)
    // 把 LLM 的回复加进消息历史
    return { messages: [response] }
}
用 Runnable 当节点

除了函数,任何 LangChain Runnable(model、prompt 链等)都能当节点:

// 用 LCEL 链当节点
const prompt = ChatPromptTemplate.fromTemplate('回答：{question}')
const chain = prompt.pipe(model)

const graph = new StateGraph(State)
    .addNode('ask', chain)   // ← 直接传 Runnable
    ...

框架会自动调 chain.invoke(state),把返回值合并到状态。

边(Edge)：节点之间的连接

是什么：告诉框架"做完这个节点，去做哪个节点"。

普通边：固定的。nodeA 跑完，无条件地去 nodeB。没有判断逻辑，没有分支，就是一条单向线。

graph.addEdge("nodeA", "nodeB");

graph
  .addEdge(START, "nodeA")
  .addEdge("nodeA", "nodeB")
  .addEdge("nodeB", END);

适合用在"流程是固定的、没有岔路"的环节，比如"读取邮件 → 分类意图"这种铁定要按顺序走的步骤。

条件边(Conditional Edge)：根据状态决定去哪

是什么：最强大的边。"做完这个节点,根据当前状态决定去哪个节点"。

怎么决定：你提供一个"路由函数",它接收状态,返回一个节点名(去哪)。

graph.addConditionalEdges("nodeA", routingFunction, {
    optionA: "nodeB",
    optionB: "nodeC",
    [END]: END,
});

function routingFunction(state: MyState): "optionA" | "optionB" | typeof END {
    if (state.someField === "条件1") return "optionA";
    if (state.someField === "条件2") return "optionB";
    return END;
}

routingFunction 是一个函数，接收当前状态，返回一个字符串（或字符串数组），告诉框架"接下来走哪条路"。第三个参数（映射表）是可选的——如果 routingFunction 直接返回的就是真实节点名，可以省略这个映射；但写出来能让图的可视化更清楚，也方便你在不改路由函数的情况下调整目标节点名。

比如：

graph.addConditionalEdges(
    'agent',                    // 从哪个节点出发
    (state) => {                // 路由函数：看 state，返回"去哪个节点"
        const lastMsg = state.messages[state.messages.length - 1]
        if (lastMsg.tool_calls && lastMsg.tool_calls.length > 0) {
            return 'tools'      // LLM 想调工具 → 去 tools 节点
        }
        return END              // LLM 回答完了 → 结束
    }
)

这就是循环的秘密：条件边可以返回"上一个节点"，形成循环。比如 agent → (条件边) → tools → (固定边) → agent → ...。

// agent → 条件边 → tools（固定边）→ agent（回到上游，循环）
graph
    .addEdge(START, 'agent')
    .addConditionalEdges('agent', routeFn)   // routeFn 可能返回 'tools'
    .addEdge('tools', 'agent')               // ← tools 完了回 agent，形成循环

这就是 Agent 工具循环的本质:条件边 + 回环。

并行边：一个节点同时连多个下游

graph
.addEdge("nodeA", "nodeB")
.addEdge("nodeA", "nodeC"); // nodeA 同时指向 nodeB 和 nodeC

nodeA 跑完之后，nodeB 和 nodeC 会在同一个 superstep 里并行执行（前面讲节点时提到的概念）。如果你想让它们跑完之后汇合到同一个节点，再加一条边：

graph
.addEdge("nodeB", "nodeD")
.addEdge("nodeC", "nodeD"); // nodeB、nodeC 都跑完后才会触发 nodeD

注意：nodeD 会等 nodeB 和 nodeC 都执行完毕才会被触发——这是图结构里天然的"汇合点"（join），不需要你手写等待逻辑。

StateGraph：图的骨架

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

const graph = new StateGraph(StateAnnotation)
  .addNode("nodeA", nodeAFunction)
  .addNode("nodeB", nodeBFunction)
  .addEdge(START, "nodeA")
  .addEdge("nodeA", "nodeB")
  .addEdge("nodeB", END);

const app = graph.compile(); // 编译后才能 invoke

compile() 这一步会做图结构校验（节点是否都被连接、是否有死路等），编译失败会直接报错——这是为什么你在 LangGraph 里很少遇到“跑到一半才发现某个节点没人连”的问题。

一个完整的最小例子(串起来)：

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

// 1. 定义状态
const State = Annotation.Root({
    number: Annotation<number>,      // 当前数字
    log: Annotation<string[]>({      // 执行日志(累加)
        reducer: (prev, next) => [...prev, ...next],
        default: () => [],
    }),
})

// 2. 定义节点
const double = (state: typeof State.State) => {
    const next = state.number * 2
    return {
        number: next,
        log: [`double: ${state.number} → ${next}`],
    }
}

const addTen = (state: typeof State.State) => {
    const next = state.number + 10
    return {
        number: next,
        log: [`addTen: ${state.number} → ${next}`],
    }
}

// 3. 构建图（带条件边）
const graph = new StateGraph(State)
    .addNode('double', double)
    .addNode('addTen', addTen)
    .addEdge(START, 'double')
    // double 完了，根据数字大小决定走哪
    .addConditionalEdges('double', (state) => {
        return state.number < 50 ? 'double' : 'addTen'
        // ↑ 小于50 继续翻倍（循环），否则去 addTen
    })
    .addEdge('addTen', END)
    .compile()

async function main() {
    // 4. 执行
    const result = await graph.invoke({ number: 3 })
    // 执行轨迹：3 → double → 6（<50继续） → double → 12 → double → 24 → double → 48（<50继续）→ double → 96（>=50）→ addTen → 106
    console.log(result.number)   // 106
    console.log(result.log)      // ['double: 3 → 6', 'double: 6 → 12', ...]
}

main()

这个例子展示了:

  • 状态定义(reducer 累加 log)
  • 节点(普通函数)
  • 固定边(START→double,addTen→END)
  • 条件边(double 完了根据状态决定 double 还是 addTen)
  • 循环(条件边返回自己形成循环)

第一个相对完整的 ReAct 循环

这是最经典的 Agent 模式：模型判断要不要调用工具 → 调用 → 把结果喂回模型 → 再判断 → 直到不再需要工具为止。手写一遍能帮你彻底理解 LangGraph 的循环机制，之后再用高层封装就知道它内部在干什么。

import { StateGraph, MessagesAnnotation, START, END } from "@langchain/langgraph";
import { ToolNode } from "@langchain/langgraph/prebuilt";
import { ChatDeepSeek } from "@langchain/deepseek";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
import { AIMessage } from "@langchain/core/messages";

// 1. 定义工具（你已经会这部分）
const getWeather = tool(
  async ({ city }) => {
    return city.toLowerCase().includes("osaka")
      ? "郑州今天 28°C，晴"
      : "未知城市的天气数据";
  },
  {
    name: "get_weather",
    description: "查询城市当前天气",
    schema: z.object({ city: z.string() }),
  }
);

const tools = [getWeather];
// 注意：用 deepseek-v4-flash 或 deepseek-v4-pro，不要用 reasoner/thinking 模式
// （thinking 模式不支持工具调用，bindTools 会形同虚设）
const model = new ChatDeepSeek({ model: "deepseek-v4-flash" }).bindTools(tools);

// 2. 定义节点：调用模型
async function callModel(state: typeof MessagesAnnotation.State) {
  const response = await model.invoke(state.messages);
  return { messages: [response] };
}

// 3. 定义条件边：模型回复里有没有工具调用？
function shouldContinue(state: typeof MessagesAnnotation.State) {
  const lastMessage = state.messages[state.messages.length - 1] as AIMessage;
  return lastMessage.tool_calls && lastMessage.tool_calls.length > 0
    ? "tools"
    : END;
}

// 4. 组装图
const workflow = new StateGraph(MessagesAnnotation)
  .addNode("agent", callModel)
  .addNode("tools", new ToolNode(tools))
  .addEdge(START, "agent")
  .addConditionalEdges("agent", shouldContinue, { tools: "tools", [END]: END })
  .addEdge("tools", "agent"); // 工具跑完，回到 agent 节点，形成循环

const app = workflow.compile();

const result = await app.invoke({
  messages: [{ role: "user", content: "郑州今天天气怎么样？" }],
});
console.log(result.messages.at(-1)?.content);

这张图里发生的事：agent 节点调模型 → 模型想调工具 → 条件边判断"有工具调用"，路由到 tools 节点 → ToolNode 实际执行工具 → 边把结果送回 agent 节点 → 模型看到工具结果，决定不再需要工具 → 条件边路由到 END。这就是循环的全部秘密：一条边把执行流指回之前的节点。

内置工具(直接拿来用)

LangGraph 提供了几个"预构建"组件,省得自己写。

MessagesAnnotation(状态)

前面讲过,预定义的 messages 状态。Agent 几乎必用。

ToolNode(节点)

之前我们手写 ReAct 循环时，tools 节点本质上要做的事是：拿到模型返回的 tool_calls，一个个去调用对应的工具函数，再把结果包装成 ToolMessage 塞回去。ToolNode 就是把这套逻辑封装好的现成节点。它自动：

  • 读最后一条消息里的 tool_calls
  • 执行对应工具
  • 把工具结果包成 ToolMessage 加进消息历史
import { ToolNode } from '@langchain/langgraph/prebuilt'
import { tool } from '@langchain/core/tools'

const getWeather = tool(async ({ city }) => `晴天`, {
    name: 'get_weather',
    description: '查天气',
    schema: z.object({ city: z.string() }),
})

// 一行创建"执行工具"节点
const toolNode = new ToolNode([getWeather])

const graph = new StateGraph(MessagesAnnotation)
    .addNode('agent', agentNode)
    .addNode('tools', toolNode)         // ← 直接用 ToolNode
    .addEdge(START, 'agent')
    .addConditionalEdges('agent', (state) => {
        const last = state.messages[state.messages.length - 1]
        return last.tool_calls?.length ? 'tools' : END
    })
    .addEdge('tools', 'agent')          // tools 完了回 agent
    .compile()

不用 ToolNode 的话，你得自己写:解析 tool_calls → 遍历 → 执行 → 包成 ToolMessage → 返回。ToolNode 帮你全包了。

错误处理：handleToolErrors

这是 ToolNode 比较关键的一个配置项，默认值是 true，工具抛出异常会被自动捕获并转换成一条 ToolMessage，让 Agent 能继续往下走而不是直接崩溃：

// 默认：宽容模式，出错也转成 ToolMessage 让模型自己看着办
const forgiving = new ToolNode(tools);

// 严格模式：工具出错直接抛出原始异常
const strict = new ToolNode(tools, { handleToolErrors: false });

// 自定义：针对特定错误类型做特殊处理
const dynamic = new ToolNode(tools, {
  handleToolErrors: (error, toolCall) => {
    if (error instanceof Error && error.message.includes("Fetch Failed")) {
      return new ToolMessage({
        content: "请求失败，请重试。",
        tool_call_id: toolCall.id!,
      });
    }
    throw error; // 其它类型的错误照样抛出去
  },
});

createAgent 封装的 Agent 构造器

  • 它在做什么

    前面我们手写的"模型节点 + 条件边 + ToolNode + 循环回模型"，本质上是一个固定模板——绝大多数单 Agent 任务都是这个结构。createAgent 把这个模板预先搭好，一次调用就能拿到一个编译好的 StateGraph：

    import { createAgent, tool } from "langchain";
import { ChatDeepSeek } from "@langchain/deepseek";
import { z } from "zod";

const getWeather = tool(
  async ({ city }) => `${city} 今天晴，28°C`,
  { name: "get_weather", description: "查询天气", schema: z.object({ city: z.string() }) }
);

const agent = createAgent({
  model: new ChatDeepSeek({ model: "deepseek-v4-flash" }),
  // 或者直接传字符串："deepseek:deepseek-v4-flash"（需要装好对应供应商包）
  tools: [getWeather],
  systemPrompt: "你是一个友善的助手。",
});

const result = await agent.invoke({
  messages: [{ role: "user", content: "郑州天气如何？" }],
});

    agent 这个返回值，本质上跟我们自己用 new StateGraph(...).compile() 拿到的东西是同一类对象——支持 .invoke()、.stream()，能传 config，能配 checkpointer。区别只是"图结构内部怎么搭"这一步被帮你做掉了。

  • 核心参数一览

    createAgent({
  model,           // 必填：模型实例，或 "provider:model-name" 字符串
  tools,           // 工具数组
  systemPrompt,    // 系统提示词，定义 Agent 的角色和行为边界
  middleware,      // 中间件数组，定制 Agent 循环的各个钩子
  checkpointer,    // 持久化（本文重点之一）
  responseFormat,  // 结构化输出（你已经学过这部分，可以直接复用）
})

  • 中间件：定制 Agent 行为的标准方式

    这是 createAgent 相比旧版 createReactAgent 最大的能力提升。中间件可以在 Agent 循环的关键节点插入逻辑：

    import { createMiddleware } from "langchain";

const loggingMiddleware = createMiddleware({
  beforeModel: (state, runtime) => {
    console.log(`即将调用模型，当前消息数：${state.messages.length}`);
  },
  afterModel: (state, runtime) => {
    console.log("模型已响应");
  },
});

const agent = createAgent({
  model: "deepseek:deepseek-v4-flash",
  tools: [getWeather],
  middleware: [loggingMiddleware],
});

    常见钩子：beforeAgent、beforeModel、wrapModelCall、wrapToolCall、afterModel、afterAgent。人工审核（humanInTheLoopMiddleware）、消息裁剪（trimMessages）都是基于这套机制实现的——可以把中间件理解成"往 Agent 的固定循环里打补丁"的标准接口，不需要你重写整个图。

  • 什么时候不该用 createAgent

    如果你的流程不是"模型反复调用工具直到完成"这种标准 ReAct 模式，而是有自定义的多步骤业务逻辑（比如"先查资料 → 再人工审核 → 再生成报告 → 再发邮件"这种带明确阶段划分的流水线），手写 StateGraph 会比硬塞进 createAgent 更清晰。createAgent 的本质优势是"标准场景下省力"，不是"万能容器"。

START / END(常量)

图的起点和终点。所有图都从 START 开始,到 END 结束。

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

graph.addEdge(START, 'firstNode')
graph.addEdge('lastNode', END)

MemorySaver(持久化，学习阶段不用管)

  • 它解决什么问题

    没有 checkpointer，每次 invoke 都是一张白纸——图执行完，状态就跟着这次调用一起消失了，下一次 invoke 完全不知道上一次发生过什么。MemorySaver 的作用是把每一步的状态快照存进进程内存，按 thread_id 分类保存。

    import { MemorySaver } from "@langchain/langgraph";
const checkpointer = new MemorySaver();
const agent = createAgent({
  model: "deepseek:deepseek-v4-flash",
  tools: [getWeather],
  checkpointer,
});

  • thread_id：区分"对话"的钥匙

    const config = { configurable: { thread_id: "user-123" } };
await agent.invoke({ messages: [{ role: "user", content: "我叫小明" }] }, config);
await agent.invoke({ messages: [{ role: "user", content: "我刚说我叫什么？" }] }, config);
// 第二轮能正确回答"小明"，因为 thread_id 相同，历史自动续上
await agent.invoke({ messages: [{ role: "user", content: "我刚说我叫什么？" }] }, { configurable: { thread_id: "user-456" } });
// 换了 thread_id，等于换了一个全新的对话，不会记得"小明"

    记忆是按 thread_id 隔离的，不是全局共享的——这也是为什么如果你做多用户应用，每个用户（或每个会话）都要分配独立的 thread_id，不然不同用户的对话历史会串在一起。

  • 它具体存了什么

    不只是 messages 字段——checkpointer 存的是完整的状态快照，包括你自定义的所有字段。如果你的 State 里有 userIntent、stepCount 之类的字段，这些也会随着每一步被存下来，下次恢复时一并带回来。

    每一次节点执行完、状态合并完毕，都会生成一个新的"检查点"（checkpoint）。所以一次多轮对话，thread_id 底下实际上是一串检查点的历史记录，不只是"最新状态"这一个快照——这也是为什么 LangGraph 能支持"回溯到某一步重新执行"这类操作（getState、updateState 这些 API 就是建立在这套机制上）。

  • 只活在内存里

    const checkpointer = new MemorySaver(); // 进程一重启，所有 thread 的历史全部消失

    这是它唯一、但很关键的限制。生产环境必须换成持久化实现：

    import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres";
const checkpointer = await PostgresSaver.fromConnString(process.env.DATABASE_URL!);

    好消息是接口完全一致——两者都实现了同一个 BaseCheckpointSaver 接口，所以从 MemorySaver 换成 PostgresSaver，对 createAgent 这一层代码毫无影响，只需要换这一行实例化的代码。这也是为什么"开发用 MemorySaver，上线前再换"是官方的标准建议流程——不用在写业务逻辑的阶段就纠结持久化细节。

  • 一个相对完整的示例

    import { createAgent, humanInTheLoopMiddleware, tool } from "langchain";
import { ChatDeepSeek } from "@langchain/deepseek";
import { MemorySaver, Command } from "@langchain/langgraph";
import { z } from "zod";

const sendEmail = tool(
  async ({ to, subject, body }) => `邮件已发送给 ${to}`,
  {
    name: "send_email",
    description: "发送邮件",
    schema: z.object({ to: z.string(), subject: z.string(), body: z.string() }),
  }
);

const agent = createAgent({
  model: new ChatDeepSeek({ model: "deepseek-v4-flash" }),
  tools: [sendEmail],
  checkpointer: new MemorySaver(), // ← 没有它，下面的 interrupt/Command 恢复机制完全无法工作
  middleware: [
    humanInTheLoopMiddleware({
      interruptOn: {
        send_email: { allowedDecisions: ["approve", "edit", "reject"] },
      },
    }),
  ],
});

const config = { configurable: { thread_id: "task-1" } };

// 第一次调用：模型决定要发邮件，触发中断，暂停在这里
const result = await agent.invoke(
  { messages: [{ role: "user", content: "给 boss@example.com 发请假邮件" }] },
  config
);

// 人工审核通过，恢复执行——靠的就是 checkpointer 记住了暂停那一刻的完整状态
await agent.invoke(
  new Command({ resume: { decisions: [{ type: "approve" }] } }),
  config // thread_id 必须跟暂停时一致，否则找不到对应的检查点
);

这里能清楚看到分工：createAgent 负责"这个 Agent 该怎么循环、该在哪一步暂停"（通过 middleware 声明）；MemorySaver 负责"暂停的那一刻的状态存在哪、恢复的时候从哪儿取"。少了 checkpointer，interrupt 机制完全无法运作——因为框架压根没有地方存"暂停时的快照"，自然也谈不上"恢复"。

使用内置工具重写“第一个相对完整的 ReAct 循环”

前面我们手动编写过“第一个相对完整的 ReAct 循环”，如果使用内置的工具重写他的话，代码可以省很多。

import { createAgent, tool } from "langchain";
import { ChatDeepSeek } from "@langchain/deepseek";
import { z } from "zod";

const getWeather = tool(
  async ({ city }) => {
    return city.toLowerCase().includes("osaka")
      ? "郑州今天 28°C，晴"
      : "未知城市的天气数据";
  },
  {
    name: "get_weather",
    description: "查询城市当前天气",
    schema: z.object({ city: z.string() }),
  }
);

const agent = createAgent({
  model: new ChatDeepSeek({ model: "deepseek-v4-flash" }),
  tools: [getWeather],
});

const result = await agent.invoke({
  messages: [{ role: "user", content: "郑州今天天气怎么样？" }],
});
console.log(result.messages.at(-1)?.content);
手写版的内容 createAgent 内部帮你做了什么
bindTools(tools) 自动绑定，你只需要传 tools 数组
callModel 节点 内置一个等价的模型调用节点
shouldContinue 条件边 内置等价的判断逻辑（看最后一条消息有没有 tool_calls）
new ToolNode(tools) 内部依然用的是 ToolNode，只是不需要你自己实例化、自己 addNode
addEdge(START, "agent") / addConditionalEdges(...) / addEdge("tools", "agent") 这一整套图结构（包括循环边）已经预先搭好并 compile() 过

常见模式速查

模式 1:线性流程(A→B→C)

graph
    .addEdge(START, 'A')
    .addEdge('A', 'B')
    .addEdge('B', 'C')
    .addEdge('C', END)

模式 2:条件分支(if/else)

graph
    .addEdge(START, 'decide')
    .addConditionalEdges('decide', (state) => {
        if (state.type === 'A') return 'pathA'
        return 'pathB'
    })
    .addEdge('pathA', END)
    .addEdge('pathB', END)

模式 3:循环(ReAct Agent 的核心)

graph
    .addEdge(START, 'agent')
    .addConditionalEdges('agent', (state) => {
        return needMoreTools(state) ? 'tools' : END
    })
    .addEdge('tools', 'agent')   // ← 回环，循环

模式 4:并行(一个节点连两条边)

// 从 A 出发，同时去 B 和 C
graph
    .addEdge('A', 'B')
    .addEdge('A', 'C')
// 注意：A 完了 B 和 C 会并行执行，都完了才继续