agent开发学习 三 graph框架语法速查

0. 先记住 LangGraph 到底是什么

LangGraph 本质上是在帮你实现一个：

有状态的任务执行器

核心组成：

State
= 当前任务的数据

Node
= 一个执行步骤

Edge
= Node 执行完以后去哪

Reducer
= 多次 State 更新怎么合并

Checkpoint
= 某个执行阶段的 State 快照

Interrupt
= 执行过程中主动暂停

Command
= 更新 State + 控制下一步

整个运行过程：

State
  ↓
Node
  ↓
返回部分 State 更新
  ↓
Reducer 合并
  ↓
得到新 State
  ↓
Edge 决定下一步
  ↓
下一个 Node
  ↓
……
  ↓
END

LangGraph 在执行过程中按“步骤”调度 Node，并可以在这些执行步骤之间保存 Checkpoint。官方把这种执行阶段称为 super-step（超级步骤）。

---

1. StateSchema：定义整个 Graph 的共享状态

语法

import {
  StateSchema,
} from "@langchain/langgraph";

import { z } from "zod/v4";

const State = new StateSchema({
  question: z.string(),
  answer: z.string().optional(),
  retryCount: z.number().default(0),
});

运行中的 State 可能是：

{
  question: "什么是 Memory？",
  answer: "Memory 是长期记忆……",
  retryCount: 1
}

---

实现原理

你可以把 State 想成：

Graph 的中央共享数据区

每一个 Node：

读取当前 State
↓
做自己的事情
↓
返回需要修改的字段

Node 一般不是自己直接把整个 State 覆盖掉。

例如：

return {
  answer: "新的答案",
};

LangGraph 会把这个更新合并回当前 State。

变成：

旧 State

question = xxx
answer = undefined
retryCount = 0

        +

Node 返回

answer = 新答案

        ↓

新 State

question = xxx
answer = 新答案
retryCount = 0

所以：

State 是任务当前状态；Node 返回的是对 State 的更新。

---

2. Node：真正执行任务的地方

语法

const analyzeNode: typeof State.Node =
  async (state) => {

    const result =
      await analyze(state.question);

    return {
      answer: result,
    };
  };

---

实现原理

Node 本质就是：

State
↓
普通 TypeScript 函数
↓
State Update

所以 Node 并不是 LLM。

Node 里面可以：

调用 LLM
调用 Tool
查数据库
查 Memory
调用 API
普通代码计算
做规则判断

例如：

async function node(state) {

  // 读取 State

  const result =
    await llm.invoke(...);

  // 返回 State 更新

  return {
    result,
  };
}

LangGraph Runtime 负责：

什么时候执行这个 Node
↓
给它哪个 State
↓
拿到它返回的数据
↓
合并 State
↓
决定后面执行谁

---

3. addNode：把函数注册成 Graph 节点

语法

const builder =
  new StateGraph(State);

builder.addNode(
  "analyze",
  analyzeNode
);

---

实现原理

原本你只有一个普通函数：

analyzeNode()

注册：

.addNode(
  "analyze",
  analyzeNode
)

以后 Graph 内部认识的是：

节点名称：
"analyze"

对应执行函数：
analyzeNode

所以：

Node 名字
→ Graph 的路由 ID

Node 函数
→ 真正执行的代码

Edge 后面连接的是节点名字：

.addEdge(
  "analyze",
  "search"
)

---

4. addEdge：固定执行顺序

语法

.addEdge(
  "analyze",
  "search"
)

完整：

const graph =
  new StateGraph(State)

    .addNode(
      "analyze",
      analyzeNode
    )

    .addNode(
      "search",
      searchNode
    )

    .addEdge(
      START,
      "analyze"
    )

    .addEdge(
      "analyze",
      "search"
    )

    .addEdge(
      "search",
      END
    )

    .compile();

对应：

START
  ↓
analyze
  ↓
search
  ↓
END

---

实现原理

Edge 本身不干活。

它只是告诉 Runtime：

A Node 成功执行完成
↓
下一步把 B Node 加入执行队列

所以：

.addEdge("A", "B")

本质就是：

A 完成
↓
调度 B

---

5. START / END

START

.addEdge(
  START,
  "agent"
)

表示：

Graph 开始
↓
第一个执行 agent

START 不是真正业务 Node。

它更像：

入口标记

---

END

.addEdge(
  "answer",
  END
)

表示：

answer 执行完成
↓
不再调度其他 Node
↓
Graph 完成

---

6. addConditionalEdges：动态决定下一步

这是 Agent 最核心的语法之一。

语法

function route(
  state: typeof State.State
) {

  if (state.needTool) {
    return "tools";
  }

  return END;
}

然后：

.addConditionalEdges(
  "agent",
  route
)

对应：

            ┌→ tools
            │
agent → 判断
            │
            └→ END

---

实现原理

普通 Edge：

A
↓
永远去 B

条件 Edge：

A
↓
先运行 router 函数
↓
读取当前 State
↓
router 返回一个目标
↓
Runtime 根据结果调度对应 Node

也就是：

State
↓
route(state)
↓
"tools"
↓
Runtime
↓
执行 tools Node

注意：

Router 通常只负责决定去哪，不负责真正处理业务。

---

7. 条件标签映射

也可以：

function route(state) {

  if (state.needTool) {
    return "use_tool";
  }

  return "finish";
}

然后：

.addConditionalEdges(
  "agent",
  route,
  {
    use_tool: "tools",
    finish: END,
  }
)

这里：

use_tool
finish

只是“路由结果”。

不是 Node。

真正关系：

use_tool
↓
tools


finish
↓
END

---

8. Reducer：State 更新到底是覆盖还是合并

这是 Graph 很重要的底层概念。

普通字段：

answer: z.string()

第一次：

answer = A

后来 Node 返回：

answer = B

通常变成：

B

也就是：

覆盖

---

如果希望累积

例如：

logs：

第一次
["A"]

第二次
["B"]

你希望最终：

["A", "B"]

这时候需要 Reducer。

语法

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

const State =
  new StateSchema({

    logs:
      new ReducedValue(
        z.array(
          z.string()
        ).default(() => []),

        {
          reducer:
            (
              oldValue,
              newValue
            ) => [
              ...oldValue,
              ...newValue
            ]
        }
      )
  });

---

实现原理

Node 不直接修改最终 State。

例如：

当前 logs

["A"]

Node 返回

["B"]

LangGraph 发现：

logs 有 Reducer

于是执行：

reducer(
  ["A"],
  ["B"]
)

得到：

["A", "B"]

所以：

Reducer 就是 State 字段的“更新规则”。

官方 ReducedValue 就是用来定义这种累计式状态更新。

---

9. MessagesValue：messages 专用 Reducer

聊天 Agent 经常有：

messages

不要简单：

messages:
  z.array(...)

可以直接：

import {
  MessagesValue,
} from "@langchain/langgraph";

const State =
  new StateSchema({
    messages: MessagesValue,
  });

Node：

return {
  messages: [
    response
  ]
};

---

实现原理

MessagesValue 本质上是：

为聊天消息专门做好的 Reducer

它不是简单：

旧数组 + 新数组

还会根据消息 ID 处理：

新消息
→ 追加

已有同 ID 消息
→ 更新

并负责把输入反序列化成 LangChain Message 对象。

所以：

MessagesValue
=
官方帮你写好的 messages Reducer

---

10. compile()：为什么 Graph 要编译

语法

const graph =
  builder.compile();

---

实现原理

在 compile() 之前：

你只是定义了：

有哪些 Node
有哪些 Edge
State 怎么定义
哪里开始
哪里结束

相当于：

Graph 设计图

compile() 后：

设计图
↓
生成可运行 Graph Runtime

之后才能：

graph.invoke(...)

所以可以理解：

StateGraph
=
Graph 配置 / 蓝图


compile()
=
生成执行器


CompiledGraph
=
真正可以运行

如果配置 Checkpointer，也是在 compile 时：

builder.compile({
  checkpointer
});

---

11. invoke()：启动一次 Graph 运行

语法

const result =
  await graph.invoke({
    question:
      "Memory 是什么？"
  });

---

实现原理

invoke() 大概发生：

输入数据
↓
生成初始 State
↓
进入 START
↓
调度第一个 Node
↓
Node 返回 State Update
↓
Reducer 合并
↓
得到新 State
↓
根据 Edge 调度
↓
继续执行
↓
到 END
↓
返回最终 State

所以：

invoke 不是“调用一个函数”，而是启动一次完整 Graph 执行。

---

12. stream()：边运行边得到事件

语法

for await (
  const event
  of await graph.stream(input)
) {
  console.log(event);
}

---

实现原理

invoke()：

整个 Graph 完成
↓
一次性返回

stream()：

Node A 完成
↓
产生事件

Node B 执行
↓
产生事件

LLM 生成 token
↓
产生事件

所以 UI 如果希望：

正在分析……
正在调用工具……
搜索完成……
正在生成答案……

一般更适合用 Stream。

---

13. Agent Loop 是怎么形成的

经典 Graph：

.addEdge(
  START,
  "agent"
)

.addConditionalEdges(
  "agent",
  route
)

.addEdge(
  "tools",
  "agent"
)

图：

  ┌─────────────┐
  │             │
  ↓             │
agent → tools ──┘
  │
  ↓
 END

---

实现原理

第一次：

agent
↓
LLM 返回 tool_call
↓
route 判断
↓
tools

tools：

执行 Tool
↓
Tool Result 写回 messages
↓
Edge 再回 agent

第二次 agent：

LLM 现在能看到：

用户问题
+
之前 AI Tool Call
+
Tool Result

然后判断：

还需要 Tool
→ 再循环

不需要
→ END

所以：

Agent Loop 不是 LangGraph 内部特殊魔法，本质就是 Graph 中存在“返回之前 Node 的 Edge”。

---

14. Checkpoint：Graph 为什么能恢复

语法

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

const checkpointer =
  new MemorySaver();

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

调用：

const config = {
  configurable: {
    thread_id:
      "thread-001"
  }
};

await graph.invoke(
  input,
  config
);

---

实现原理

启用 Checkpointer 后，LangGraph 会在执行步骤之间保存：

State 当前值
+
下一步要执行谁
+
Graph 执行元信息
+
checkpoint_id
+
thread_id

例如：

START
↓
Checkpoint 0
↓
A
↓
Checkpoint 1
↓
B
↓
Checkpoint 2

Checkpoint 是 Graph State 在某个执行时刻的快照。官方会在各个 super-step 边界保存 StateSnapshot。

---

15. thread_id：它不是 Memory ID

{
  configurable: {
    thread_id:
      "thread-001"
  }
}

可以把它理解成：

这条 Graph 运行链 / 会话的身份证

Checkpointer 保存：

thread-001

checkpoint 1
checkpoint 2
checkpoint 3
...

以后：

thread_id
↓
找到这一整条执行历史

官方也说明 Checkpointer 会使用 thread_id 来保存和重新加载对应线程的 Checkpoint。

所以：

thread_id
=
找到哪条任务执行线


checkpoint_id
=
找到这条线上的某一个历史状态

---

16. Interrupt：人工暂停

语法

import {
  interrupt,
} from "@langchain/langgraph";

const reviewNode =
  async (state) => {

    const approved =
      interrupt({
        question:
          "是否批准发送？",

        email:
          state.email,
      });

    return {
      approved
    };
  };

---

17. Interrupt 真正的实现原理

这一块一定要记。

很多人会误以为：

JavaScript 真的停在
interrupt() 那一行

并不是。

第一次运行：

reviewNode 从头开始
↓
运行 interrupt 之前的代码
↓
interrupt(...)
↓
抛出一个特殊暂停异常 / 信号
↓
LangGraph Runtime 捕获
↓
Checkpointer 保存状态
↓
__interrupt__ 返回外部
↓
本次运行暂停

官方说明 interrupt() 底层通过特殊异常让 Runtime 暂停，并保存 Graph 状态。

---

18. Interrupt 怎么恢复

人工点击：

批准

调用：

await graph.invoke(

  new Command({
    resume: true
  }),

  {
    configurable: {
      thread_id:
        "thread-001"
    }
  }
);

内部：

thread_id
↓
找到之前 Checkpoint
↓
找到 Interrupt
↓
拿到 resume = true
↓
重新执行发生 Interrupt 的 Node

注意：

Node 是从头重新执行，不是从 interrupt 下一行继续。

执行到：

const approved =
  interrupt(...)

这一次 Runtime 发现：

这个 interrupt 已经有 resume value

于是：

interrupt(...)

直接返回：

true

所以逻辑上：

const approved = true;

然后继续往下执行。官方文档明确说明，恢复时发生 Interrupt 的 Node 会从头重新运行，因此 Interrupt 前面的代码也会再次执行。

Interrupt = “按顺序读取 resume 值的函数；读不到就通过特殊异常暂停”。Command resume = “给等待中的 interrupt 提供值，并让 Runtime 从 Checkpoint 重新调度被打断的 Node

---

19. 为什么 Interrupt 前面不能随便做副作用

错误：

await sendEmail();

const approved =
  interrupt(
    "批准吗？"
  );

第一次：

sendEmail()
↓
发送一次
↓
Interrupt

恢复：

Node 从头执行
↓
sendEmail()
↓
又发送一次

于是重复发送。

所以：

Interrupt 前面的操作
最好可重复执行

也就是：

幂等

更合理：

生成 Email
↓
Interrupt
↓
人工审批
↓
批准
↓
发送 Email

官方也特别要求 Interrupt 前面的副作用应设计为幂等操作。

---

20. Command：不仅能恢复 Interrupt

Command 还有另一个用途：

更新 State
+
决定下一步去哪

语法

return new Command({

  update: {
    status: "done"
  },

  goto: "nextNode"
});

---

实现原理

普通 Node：

Node
↓
返回 State
↓
Reducer
↓
Edge
↓
决定下一步

Command：

Node
↓
Command
├── State Update
└── goto
↓
Runtime 同时处理

也就是：

更新 State
+
动态路由

在一个结果里完成。

官方建议：如果既需要更新状态又需要动态跳转，可以使用 Command；如果只是路由，则条件 Edge 更直接。

---

21. Command.goto 和 ends

例如：

return new Command({
  goto:
    state.ok
      ? "success"
      : "failed"
});

Node 注册：

.addNode(
  "check",
  checkNode,
  {
    ends: [
      "success",
      "failed"
    ]
  }
)

ends 的意思：

告诉 Graph：

check 这个 Node
可能动态跳到：

success
failed

这样 Graph 在编译时知道可能的控制流。

---

22. Conditional Edge 和 Command 的区别

记这个就够：

只需要决定去哪

→ Conditional Edge


既修改 State
又决定去哪

→ Command

例如：

判断有没有 Tool Call

→ Conditional Edge

而：

人工批准
↓
status = approved
↓
goto send

→ Command

---

23. Checkpoint 和 Memory 不要混

前面刚学 Memory，这里特别容易混。

Checkpoint
=
保存当前 Graph State

解决：
任务暂停
任务恢复
异常恢复
Human-in-the-loop

而：

长期 Memory
=
跨任务保存长期有价值的信息

例如：
用户偏好
项目事实
历史经验

所以：

Checkpoint

“这个任务做到哪了？”


Memory

“过去有哪些东西值得以后记住？”

LangGraph 官方也将 Checkpointer 的线程级持久化和跨线程长期 Store 区分开。

---

24. 一个完整 Agent Graph 的执行原理

假设：

START
↓
agent
↓
tools
↓
agent
↓
review
↓
send
↓
END

实际执行：

① invoke()

输入
↓
初始化 State


② START

调度 agent


③ agent Node

读取 State.messages
↓
调用 LLM
↓
LLM 返回 Tool Call
↓
写入 messages


④ Reducer

旧 messages
+
新 AIMessage
↓
新 messages


⑤ Conditional Edge

检查最后一条消息
↓
发现 Tool Call
↓
调度 tools


⑥ tools Node

调用外部 Tool
↓
得到 Tool Result
↓
写入 messages


⑦ Edge

tools
↓
agent


⑧ agent 再执行

LLM 看到 Tool Result
↓
生成结果


⑨ review Node

interrupt()
↓
Runtime 暂停
↓
Checkpoint 保存


⑩ 人工批准

Command({
  resume: true
})


⑪ thread_id

定位之前 Checkpoint
↓
重新执行 review


⑫ interrupt()

返回 true
↓
继续


⑬ send Node

真正执行发送


⑭ END

Graph 完成
↓
返回最终 State

这就是 LangGraph 核心执行原理。

---

25. 最小完整模板

import {
  StateGraph,
  StateSchema,
  MessagesValue,
  START,
  END,
  MemorySaver,
  interrupt,
  Command,
} from "@langchain/langgraph";


// ----------------
// State
// ----------------

const State =
  new StateSchema({

    messages:
      MessagesValue,

  });


// ----------------
// Agent Node
// ----------------

const agentNode:
  typeof State.Node =
  async (state) => {

    const response =
      await model.invoke(
        state.messages
      );

    return {
      messages: [
        response
      ]
    };
  };


// ----------------
// Tool Node
// ----------------

const toolsNode:
  typeof State.Node =
  async (state) => {

    const toolResult =
      await executeTool(
        state.messages
      );

    return {
      messages: [
        toolResult
      ]
    };
  };


// ----------------
// Router
// ----------------

function route(
  state: typeof State.State
) {

  const lastMessage =
    state.messages.at(-1);

  if (
    lastMessage
      ?.tool_calls
      ?.length
  ) {
    return "tools";
  }

  return "review";
}


// ----------------
// Review
// ----------------

const reviewNode:
  typeof State.Node =
  async () => {

    const approved =
      interrupt({
        question:
          "是否批准？"
      });

    return new Command({
      goto:
        approved
          ? "send"
          : END
    });
  };


// ----------------
// Send
// ----------------

const sendNode:
  typeof State.Node =
  async () => {

    await send();

    return {};
  };


// ----------------
// Graph
// ----------------

const checkpointer =
  new MemorySaver();

const graph =
  new StateGraph(State)

    .addNode(
      "agent",
      agentNode
    )

    .addNode(
      "tools",
      toolsNode
    )

    .addNode(
      "review",
      reviewNode,
      {
        ends: [
          "send",
          END
        ]
      }
    )

    .addNode(
      "send",
      sendNode
    )

    .addEdge(
      START,
      "agent"
    )

    .addConditionalEdges(
      "agent",
      route
    )

    .addEdge(
      "tools",
      "agent"
    )

    .addEdge(
      "send",
      END
    )

    .compile({
      checkpointer
    });

运行：

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

const result =
  await graph.invoke(
    {
      messages: [
        {
          role: "user",
          content:
            "帮我执行任务"
        }
      ]
    },

    config
  );

如果 Interrupt：

await graph.invoke(

  new Command({
    resume: true
  }),

  config
);

---

26. 最终速记表

语法	是什么	内部原理
StateSchema	定义 State	定义 Graph 共享数据字段
State.Node	Node 类型	State → 函数 → State Update
addNode()	注册 Node	建立 Node 名称和执行函数映射
addEdge()	固定路线	前 Node 完成后调度下一个
addConditionalEdges()	条件路线	Router 读取 State 后决定调度谁
ReducedValue	Reducer	决定旧值和新值怎么合并
MessagesValue	消息 Reducer	专门处理聊天消息追加/更新
START	Graph 入口	第一次调度从这里开始
END	Graph 结束	不再调度新 Node
compile()	编译 Graph	Graph 蓝图 → 可执行 Runtime
invoke()	执行 Graph	从输入一直调度到 END/Interrupt
stream()	流式执行	执行过程中持续暴露事件
Checkpointer	保存状态	每个执行步骤保存 StateSnapshot
thread_id	任务线程 ID	用来找到对应 Checkpoint 链
interrupt()	暂停	抛特殊暂停信号 + 保存 Checkpoint
Command({resume})	恢复	让 Interrupt 得到外部返回值
Command({update,goto})	更新+跳转	一次完成 State 更新和动态路由
ends	声明动态目标	告诉 Graph Node 可能 goto 哪些节点

---

27. 最值得背下来的 7 句话

1.
State 保存：
“任务现在是什么状态。”


2.
Node 负责：
“这一小步具体做什么。”


3.
Reducer 负责：
“Node 返回的数据怎么合并进 State。”


4.
Edge 负责：
“这一小步做完以后去哪。”


5.
Checkpoint 负责：
“任务当前做到哪里，可以恢复。”


6.
Interrupt 负责：
“这里先暂停，等外部输入再继续。”


7.
Agent Loop 本质：
“Edge 又把执行流程指回之前的 Node。”

最后把整个 LangGraph 压缩成：

              State
                ↓
               Node
                ↓
          State Update
                ↓
             Reducer
                ↓
            新 State
                ↓
               Edge
                ↓
        ┌───────┴───────┐
        ↓               ↓
     下个 Node        END
        │
        └────→ 也可以循环


执行过程中：

Node
↓
Checkpoint
↓
Node
↓
Checkpoint
↓
Interrupt
↓
暂停
↓
Command(resume)
↓
恢复
↓
继续 Graph

如果这张图能在脑子里建立起来，那么以后即使忘了具体 API，基本也可以很快把 LangGraph 代码重新写出来。