一、先记住整个 Agent 的核心结构
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
| 用户
↓
LLM（大语言模型）
↓
需要 Tool（工具）？
├─ 不需要 → 最终回答
│
└─ 需要
↓
Tool Call（工具调用请求）
↓
Runtime（运行时）
↓
校验参数
↓
执行 Tool
↓
Tool Result（工具结果）
↓
重新交给 LLM
↓
继续判断
|
这就是：
三个角色：
1
2
3
4
5
6
7
8
| LLM
= 决策者
Runtime
= 调度 / 执行控制者
Tool
= 真正干活的人
|
LLM 不会真的执行工具。
LLM 只是返回类似：
1
2
3
4
5
6
| {
"name": "read_file",
"arguments": {
"path": "src/index.ts"
}
}
|
Runtime 才真正执行：
1
| await readFile("src/index.ts");
|
一个 Tool 可以理解成：
1
2
3
4
5
| Tool
├── name 工具名称
├── description 工具说明
├── schema 参数格式
└── execute 真正执行函数
|
例如：
1
2
3
4
5
6
7
8
9
10
11
12
| const weatherTool = tool(
async ({ city }) => {
return `${city} 25度`;
},
{
name: "get_weather",
description: "查询城市天气",
schema: z.object({
city: z.string(),
}),
}
);
|
其中：
一般使用：
Zod 的作用：
1
2
3
4
5
| z.string()
z.number()
z.boolean()
z.array(z.string())
z.object({...})
|
核心区别：
1
2
3
4
5
| TypeScript 类型
= 写代码时检查
Zod
= 程序真正运行时检查
|
LangGraph 已经帮我们封装好了 Tool Executor（工具执行器）：
1
| const toolNode = new ToolNode(tools);
|
它内部大概做：
1
2
3
4
5
6
7
8
9
10
11
12
13
| 读取最后一个 AIMessage
↓
查看 tool_calls
↓
根据 name 找 Tool
↓
执行工具
↓
得到 Tool Result
↓
生成 ToolMessage
↓
返回 messages 更新
|
工具数组会同时给两边：
这是：
而：
这是：
四、State（状态）
State 是：
当前业务流程正在变化的数据。
例如：
1
2
3
4
5
6
| {
messages: [],
task: "...",
status: "...",
result: "...",
}
|
Node 不应该自己随便修改 State。
一般：
1
2
3
4
5
| async function node(state) {
return {
status: "done"
};
}
|
Node 返回的是：
1
2
| Partial State Update
部分状态更新
|
然后 Runtime：
1
2
3
4
5
6
7
| 旧 State
↓
Node 返回 Update
↓
根据更新规则合并
↓
新 State
|
五、StateSchema（状态模板）
StateSchema 定义：
1
2
3
4
5
| State 有哪些字段
+
字段是什么类型
+
必要时字段怎么更新
|
例如：
1
2
3
4
| const State = new StateSchema({
status: z.string(),
count: z.number(),
});
|
普通字段通常：
六、Reducer（归并函数）
如果不希望覆盖，而是希望：
就定义 Reducer。
例如：
1
2
3
4
5
6
7
8
| results: new ReducedValue(
z.array(z.string()).default(() => []),
{
reducer: (oldValue, newValue) => {
return [...oldValue, ...newValue];
},
}
)
|
理解：
1
2
3
4
5
| ReducedValue
=
字段类型
+
Reducer 更新规则
|
所以：
1
2
3
4
5
| z.array(z.string())
= 这个字段是什么
reducer(...)
= 这个字段怎么更新
|
七、MessagesValue
1
2
3
| const State = new StateSchema({
messages: MessagesValue,
});
|
你可以把 MessagesValue 理解成框架预制好的：
1
2
3
| Message[] 的类型
+
messages 的合并规则
|
所以 Node：
1
2
3
| return {
messages: [response]
};
|
不会简单覆盖之前所有消息。
而会形成：
1
2
3
4
5
6
7
8
| UserMessage
↓
AIMessage
↓
ToolMessage
↓
AIMessage
...
|
八、Node（节点）
Node 本质：
一个接受 State，执行逻辑，然后返回 State Update 的函数。
1
2
3
4
5
6
7
| async function myNode(state) {
return {
result: "done"
};
}
|
LLM Node（模型节点）没有特殊语法。
只是普通 Node 里面调用了模型：
1
2
3
4
5
6
7
8
| async function llmNode(state) {
const response =
await model.invoke(state.messages);
return {
messages: [response]
};
}
|
九、Runtime（运行时对象）
Node 可以：
1
2
| async function node(state, runtime) {
}
|
区别：
1
2
3
4
5
| State
= 业务现在是什么状态
Runtime
= 当前业务在什么运行环境中执行
|
例如：
1
2
3
4
5
6
7
8
9
10
11
| state
├── task
├── messages
├── result
└── status
runtime
├── context
├── executionInfo
├── store
└── streamWriter
|
常见：
1
| runtime.executionInfo.nodeAttempt
|
表示：
runtime 一般不是你自己手动创建并传给 Node。
而是：
1
2
3
4
5
6
| graph.invoke(input, config)
↓
LangGraph Runtime
↓
自动调用
node(state, runtime)
|
十、Edge（边）
普通 Edge：
意思：
固定路线。
十一、Conditional Edge（条件边）
1
2
3
4
| .addConditionalEdges(
"A",
router
)
|
就是：
1
2
3
4
5
6
7
| A 执行完成
↓
State 更新完成
↓
router(最新 State)
↓
根据返回值决定去哪
|
例如：
1
2
3
4
5
6
7
| function router(state) {
if (state.success) {
return "B";
}
return "C";
}
|
对应：
所以：
1
2
3
4
5
| addEdge(A, B)
= A 后固定 B
addConditionalEdges(A, fn)
= A 后去哪看 fn(state) 返回值
|
LangGraph 帮我们预制好的条件函数：
1
2
3
4
| .addConditionalEdges(
"llm",
toolsCondition
)
|
作用：
1
2
3
4
5
| 最后一个 AIMessage
↓
有 tool_calls？
├─ 有 → tools
└─ 没有 → END
|
因此标准 Tool Agent：
1
2
3
4
5
6
7
8
9
10
| const graph = new StateGraph(State)
.addNode("llm", llmNode)
.addNode("tools", toolNode)
.addEdge(START, "llm")
.addConditionalEdges(
"llm",
toolsCondition
)
.addEdge("tools", "llm")
.compile();
|
脑图：
1
2
3
4
5
6
7
8
9
10
11
| ┌──────────────┐
↓ │
START → LLM │
↓ │
有 Tool Call？ │
/ \ │
有 无 │
↓ ↓ │
ToolNode END │
↓ │
└──────────────────┘
|
十三、Node Retry（节点重试）
Node 可以配置重试策略：
1
2
3
4
5
6
7
8
9
| .addNode(
"api",
apiNode,
{
retryPolicy: {
maxAttempts: 3,
}
}
)
|
需要区分：
1
2
3
4
5
6
| Node 抛异常
→ Retry（重试）
Node 正常执行成功
但业务结果不满意
→ Conditional Edge（条件边）
|
例如：
但是：
1
2
3
| 测试程序正常执行
但测试失败
→ test → coding → test
|
这是 Graph Loop（工作流循环），不是 Retry。
十四、Checkpoint（检查点）
Checkpoint 是：
保存某个执行时刻的状态和执行位置。
配置：
1
2
3
4
5
6
| const checkpointer =
new MemorySaver();
const graph = builder.compile({
checkpointer,
});
|
MemorySaver：
生产环境可以换数据库存储。
十五、thread_id（线程 ID）
执行：
1
2
3
4
5
| const config = {
configurable: {
thread_id: "task-123",
},
};
|
这是 LangGraph 约定好的字段。
作用：
概念上：
1
2
3
4
5
| task-123
├── checkpoint 1
├── checkpoint 2
├── checkpoint 3
└── checkpoint 4
|
同一个：
可以找到之前状态继续。
新的：
就是另一条独立状态线。
十六、Persistence（持久化）
Checkpoint 和 Persistence 区分：
1
2
3
4
5
| Checkpoint
= 保存一个恢复点
Persistence
= 这个恢复点到底保存在哪里、怎么读取
|
例如：
1
2
3
4
5
| MemorySaver
= 内存 Persistence
Postgres Saver
= 数据库 Persistence
|
十七、Interrupt（中断）
Node：
1
2
| const approved =
interrupt("是否批准？");
|
执行到这里：
1
2
3
4
5
6
7
8
9
| Node
↓
interrupt
↓
保存 Checkpoint
↓
暂停 Graph
↓
等外部输入
|
十八、Resume（恢复）
恢复：
1
2
3
4
5
6
| await graph.invoke(
new Command({
resume: true,
}),
config
);
|
注意：
这个：
会成为：
1
2
| const approved =
interrupt(...);
|
的返回值。
所以：
1
2
3
4
5
6
7
8
9
10
11
12
13
| 第一次：
approved = interrupt(...)
↓
暂停
第二次：
Command({ resume: true })
↓
恢复
↓
approved = true
|
非常重要：
含 interrupt() 的 Node 恢复时会从 Node 开头重新执行。
所以 interrupt() 前面的副作用必须小心。
十九、Human-in-the-loop（人工介入）
典型：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| Agent
↓
准备危险操作
↓
interrupt
↓
Checkpoint
↓
Persistence
↓
等待用户
↓
Approve / Reject
↓
Command.resume
↓
继续执行
|
二十、Command（命令）
Command 可以：
语法：
1
2
3
4
5
6
7
| return new Command({
update: {
status: "done",
},
goto: "nextNode",
});
|
对比：
1
2
3
4
5
6
7
8
| 普通 Node return
= State Update
Conditional Edge
= 决定下一节点
Command
= State Update + 决定下一节点
|
二十一、ends
如果 Node 用：
注册时可以声明：
1
2
3
4
5
6
7
8
9
10
| .addNode(
"test",
testNode,
{
ends: [
"review",
"coding"
]
}
)
|
记忆：
1
2
3
4
5
| ends
= 这个 Node 最多允许跳到哪些 Node
goto
= 这一次实际跳哪个 Node
|
二十二、Parallel（并行）
固定并行：
1
2
| .addEdge("A", "B")
.addEdge("A", "C")
|
就是：
1
2
3
| ┌→ B ─┐
A ─────┤ ├→ D
└→ C ─┘
|
B、C 之间没有依赖，就可以同时执行。
类似 JS：
1
2
3
4
| await Promise.all([
nodeB(),
nodeC()
]);
|
如果 B、C 同时更新一个 State 字段，就需要 Reducer。
二十三、Send（动态并行）
如果运行时才知道任务数量：
1
2
3
4
5
| state.files = [
"a.ts",
"b.ts",
"c.ts"
];
|
可以：
1
2
3
4
5
6
7
| return state.files.map(
file =>
new Send(
"analyzeFile",
{ file }
)
);
|
结果：
1
2
3
| ┌→ analyzeFile(a.ts)
prepare ──┼→ analyzeFile(b.ts)
└→ analyzeFile(c.ts)
|
一句话：
Send = 根据运行时数据动态创建多份任务，并给每份任务不同输入。
固定并行：
Send：
二十四、Subgraph（子图）
Subgraph：
一个 Graph 可以作为另一个 Graph 里的一个 Node。
例如：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
| Parent Graph（父图）
prepare
↓
┌─────────────────┐
│ Coding Subgraph │
│ │
│ analyze │
│ ↓ │
│ edit │
│ ↓ │
│ test │
└─────────────────┘
↓
review
|
如果父子 State 兼容：
1
2
3
4
| parentGraph.addNode(
"coding",
compiledSubgraph
);
|
二十五、Wrapper Node（包装节点）
如果父图和子图 State 不一样：
1
2
3
4
5
6
7
8
9
10
| Parent State
{
task
}
Subgraph State
{
query,
answer
}
|
就写包装 Node：
1
2
3
4
5
6
7
8
9
10
11
| const researchNode = async (state) => {
const subResult =
await subgraph.invoke({
query: state.task,
});
return {
result: subResult.answer,
};
};
|
这里一定记住：
1
2
3
4
5
| subgraph.invoke(...)
→ 返回子图自己的最终 State
wrapperNode return {...}
→ 返回给父 Graph 的 State Update
|
所以 Wrapper Node 本质就是：
1
2
3
4
5
6
7
8
9
10
11
| 父图格式
↓
转换
↓
子图格式
↓
执行子图
↓
转换
↓
父图 State Update
|
最终一张总图
你现在对 LangGraph 可以形成这个整体认知：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
| Graph Runtime
│
↓
State
│
┌──────────┴──────────┐
↓ ↓
Node Runtime
│ │
│ ├─ context
│ ├─ executionInfo
│ └─ store
↓
State Update
↓
Reducer / Merge
↓
New State
↓
Edge
┌──────┼─────────┐
↓ ↓ ↓
普通 条件 Command
Edge Edge goto
↓
下一个 Node
Node 内部又可以：
LLM
↕
ToolNode
↕
LLM
形成 Agent Loop
|
再加上长期执行：
1
2
3
4
5
6
7
8
9
10
11
12
13
| Graph
↓
Checkpoint
↓
Persistence
↓
Interrupt
↓
等待人工
↓
Command.resume
↓
继续执行
|
再加复杂控制：
1
2
3
4
5
6
7
8
| Parallel
= 固定并行
Send
= 动态并行
Subgraph
= 把复杂 Graph 模块化
|
微信
支付宝