Appearance
核心概念总览
上一篇你跑通了第一个程序,对 State / Node / Edge 有了直观印象。本篇用一张大图把 LangGraph 所有核心概念串起来,建立全局地图——后续每一章都是对这张图的深入展开。
一、全景概念图
mermaid
graph TB
subgraph 构图阶段
SG[StateGraph 状态图容器]
S[State 状态]
N1[Node 节点]
N2[Node 节点]
E1[Edge 普通边]
CE[ConditionalEdge 条件边]
R[Reducer 归约器]
end
subgraph 运行阶段
CG[CompiledGraph 编译产物]
CP[Checkpointer 检查点]
CFG[Config 配置 thread_id 等]
ST[Streaming 流式输出]
HITL[Interrupt 人机交互]
SUB[Subgraph 子图]
end
S --> SG
N1 -.注册.-> SG
N2 -.注册.-> SG
R -.修饰字段.-> S
SG --> E1
SG --> CE
SG -->|compile| CG
CG --> CP
CG --> CFG
CG --> ST
CG -.可选.-> HITL
CG -.可选.-> SUB读图方法:上半部分"构图阶段"是你搭图时做的事;下半部分"运行阶段"是图跑起来后能用的能力。
二、核心概念速览
| 概念 | 一句话 | 详见 |
|---|---|---|
| State 状态 | 在节点间共享的数据容器,整个图的"工作内存" | 状态 State |
| Node 节点 | 处理逻辑的单元,本质是普通 Python 函数 | 节点 Node |
| Edge 边 | 节点之间的连线,决定执行顺序 | 边 Edge |
| ConditionalEdge 条件边 | 根据状态动态选下一节点 | 条件边 ConditionalEdge |
| StateGraph 状态图 | 把节点和边装起来的容器 | StateGraph 状态图 |
| Reducer 归约器 | 决定字段如何被多个节点合并更新 | Reducer 归约器 |
| 编译与运行 | compile 生成可执行版本,invoke/stream 跑图 | 编译与运行 |
三、概念逐个讲解(2-3 句话版)
State 状态
图执行过程中在节点之间流动的共享数据。用 TypedDict 或 Pydantic 定义。状态让节点解耦:节点 A 不需要知道节点 B 的存在,只要读/写状态即可。
Node 节点
普通 Python 函数,签名 (state: State) -> dict。返回的 dict 按 reducer 合并进状态。节点内可以调 LLM、调工具、读写文件、做计算——干任何事。
Edge 边
节点之间的连线 add_edge(source, target)。边本身不传数据——数据通过 state 传递,边只决定"执行完 source 后下一步去哪"。
ConditionalEdge 条件边
路由函数 router(state) -> str 根据当前状态返回一个字符串(下一节点名),实现动态分支。这是构建循环(如 ReAct)的关键。
StateGraph 状态图
核心容器 StateGraph(State),三个大方法 add_node / add_edge / add_conditional_edges 把图搭出来。
Reducer 归约器
字段更新策略。默认是"覆盖",但消息列表需要"追加",所以用 Annotated[list, add_messages]。reducer 决定了状态如何进化。
编译与运行
compile() 校验并生成可执行版本;invoke 一次性同步跑,stream 流式逐步返回。配合 config={"thread_id": ...} 和 checkpointer 可以让图"有记忆"。
四、进阶特性(后续模块深入)
mermaid
mindmap
root((进阶能力))
持久化
MemorySaver 内存
SQLite 检查点
Postgres 生产
人机交互 HITL
interrupt 中断
审核后继续
Command 恢复
流式
stream_mode values
stream_mode updates
stream_mode messages
子图
嵌套图
模块化复用
多智能体
Supervisor
Swarm
Send 多路分发
并行扇出这些在 04-进阶特性 模块展开,本篇只做总览。
五、LangGraph(Python) ↔ LangGraph4j(Java) 对照
如果你从 LangGraph4j 迁移过来,下面这张表能帮你快速对号入座:
| 概念 | LangGraph (Python) | LangGraph4j (Java) |
|---|---|---|
| 状态类型 | TypedDict / Pydantic v2 | Record<String,Object> / 自定义类 |
| 字段合并策略 | Annotated[list, add_messages] | @Reducer 注解 + 类 |
| 消息状态 | MessagesState(内置) | MessagesState(内置) |
| 节点 | 普通 Python 函数 | AgentNode 接口 / lambda |
| 节点签名 | (state: State) -> dict | apply(state) -> Map<String,Object> |
| 图容器 | StateGraph(State) | StateGraph.of(State.class) |
| 添加节点 | add_node("name", fn) | addNode("name", fn) |
| 添加边 | add_edge("a", "b") | addEdge("a", "b") |
| 条件边 | add_conditional_edges(src, router, path_map) | addConditionalEdges(src, router, path_map) |
| 入口/出口 | START / END(常量) | START / END(常量) |
| 编译 | graph.compile() | graph.compile() |
| 编译参数 | compile(checkpointer=..., interrupt_before=...) | compile(checkpointer, interrupt_before) |
| 运行 | app.invoke(input) | app.invoke(input) |
| 流式 | app.stream(input) / astream | app.stream(input) |
| 检查点 | MemorySaver / SQLite / Postgres | MemorySaver |
| 预置 ReAct | create_react_agent | createReActAgent |
| 工具 | @tool 装饰器 | @Tool 注解 |
| 中断 | interrupt(value) | interrupt(value) |
| 恢复 | app.invoke(Command(resume=...)) | app.invoke(Command(resume=...)) |
| 多路分发 | Send(node, data) | Send(node, data) |
迁移提示:
- API 几乎一一对应,主要差异是 Python 函数一等公民、Java 要 lambda/接口。
- 类型系统不同:Python 用
TypedDict(鸭子类型),Java 要严格类定义。 - 生态不同:Python 版本因 LangChain 庞大生态,工具/向量库选择更多。
- 执行模型相同:都是基于状态快照 + reducer 合并的图执行。
- 运行时校验:Python 默认无校验(除非用 Pydantic),Java 因静态类型天然更严格。
- 包导入路径:Python 从
langgraph.graph/langgraph.prebuilt/langgraph.checkpoint.memory引入;Java 从对应的org.langgraph4j.*包引入。
六、学习顺序建议
mermaid
graph LR
A[State 状态] --> B[Node 节点]
B --> C[Edge 边]
C --> D[ConditionalEdge 条件边]
D --> E[StateGraph 状态图]
E --> F[Reducer 归约器]
F --> G[编译与运行]
G --> H[进阶: Checkpointer/HITL/Streaming/Subgraph]推荐顺序:
- 先理解 State——其他概念都围绕它展开。
- 再看 Node——它读写 State。
- Edge 把 Node 连起来。
- ConditionalEdge 让图能分叉和循环。
- StateGraph 把上面都装进容器。
- Reducer 决定字段如何进化。
- 编译与运行 让图真正跑起来。
- 进阶特性(持久化、HITL、流式、子图)建立在以上基础上。
提示:如果跳着读,至少先把 State 和 Node 两篇看完,否则后面的代码会看不懂字段从哪来、节点怎么写。
七、一个小例子串所有概念
python
from typing import TypedDict, Annotated
from operator import add
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
# 1. State(带 reducer)
class State(TypedDict):
topic: str
messages: Annotated[list, add_messages] # 消息累加
# 2. Node
def think(state: State) -> dict:
return {"messages": [("user", f"关于 {state['topic']} 的思考")]}
def respond(state: State) -> dict:
return {"messages": [("assistant", "好主意!")]}
# 3. 路由:决定下一节点
def router(state: State) -> str:
if len(state["messages"]) < 4:
return "think" # 消息少就继续思考
return "END" # 够了就结束
# 4. StateGraph 装配 + 5. ConditionalEdge
g = StateGraph(State)
g.add_node("think", think)
g.add_node("respond", respond)
g.add_edge(START, "think")
g.add_conditional_edges("think", router, {"think": "respond", "END": END})
g.add_edge("respond", "think") # 形成循环
# 6. 编译
app = g.compile()
# 7. 运行
for ev in app.stream({"topic": "学 LangGraph", "messages": []}):
print(ev)这个例子虽然简单,但已经涵盖了 State / Node / Edge / ConditionalEdge / StateGraph / Reducer / 编译运行 全部七个核心概念。
八、小结
- LangGraph 七大核心概念:State、Node、Edge、ConditionalEdge、StateGraph、Reducer、编译与运行。
- 上半部分是"构图",下半部分是"运行"。
- 与 LangGraph4j 概念一一对应,迁移友好。
- 建议按 State → Node → Edge → 条件边 → StateGraph → Reducer → 运行 的顺序学。
- 一段 30 行代码就能串起所有核心概念。
接下来进入核心概念详解,从最基础的 状态 State 开始。