在 LangChain 中,我们通常从
create_agent这样的高层入口开始构建 Agent;
而 LangGraph 更关注底层执行过程:状态保存在哪里、节点什么时候执行、执行完之后走向哪里,以及多个节点的状态如何合并。如果刚开始学习 LangGraph,我认为最值得先弄明白的不是各种 API,而是
State、Node、Edge和Superstep之间到底是什么关系。
1. LangGraph 的三个基本元素
一个最基础的 LangGraph,可以先抽象成下面这条流程:
START -> node_a -> node_b -> END其中真正需要理解的只有三个东西:
| 概念 | 作用 |
|---|---|
State | 保存整个工作流运行过程中需要共享的数据 |
Node | 真正执行任务的节点,通常就是一个 Python 函数 |
Edge | 决定节点执行完成后,下一步去哪里 |
可以把它简单理解成:
State = 数据 Node = 处理数据 Edge = 控制流程例如:
fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassOverallState(TypedDict):username:strgreeting:stroutput:str这里的OverallState就是在声明整个图运行过程中可能使用到的数据结构。
2. 一个节点到底做了什么?
定义两个节点:
defnode_a(state:OverallState):return{"greeting":"Dear "+state["username"]}defnode_b(state:OverallState):return{"output":state["greeting"]+",你好!"}节点返回的不是一个新的完整 State,而是对 State 的局部更新。
例如执行node_a之前:
{"username":"小黄"}node_a返回:
{"greeting":"Dear 小黄"}并不意味着username被删除了。
LangGraph 会把这次返回理解为:
只更新 greeting 字段于是更新后的状态可以理解为:
{"username":"小黄","greeting":"Dear 小黄"}节点只返回需要更新的字段;没有返回的字段保持原来的值。
3. State、Node、Edge 是怎样组合起来的?
定义完状态和节点以后,就可以开始构建图:
builder=StateGraph(state_schema=OverallState)builder.add_node("node_a",node_a)builder.add_node("node_b",node_b)builder.add_edge(START,"node_a")builder.add_edge("node_a","node_b")builder.add_edge("node_b",END)graph=builder.compile()这里可以分成三步来看。
3.1add_node():注册执行节点
builder.add_node("node_a",node_a)表示把 Python 函数node_a注册到图中,并给它一个节点名称node_a。
3.2add_edge():规定执行顺序
builder.add_edge(START,"node_a")builder.add_edge("node_a","node_b")builder.add_edge("node_b",END)表达的就是:
START | v node_a | v node_b | v END3.3compile():把状态图编译成可执行图
前面的builder更像是在描述图结构。
真正执行之前,需要:
graph=builder.compile()然后再调用:
result=graph.invoke({"username":"小黄"})print(result)最终可以得到类似:
{"username":"小黄","greeting":"Dear 小黄","output":"Dear 小黄,你好!"}4. LangGraph 并不是一个节点改完状态,下一个马上读取
但 LangGraph 更准确的运行方式要结合Superstep(超步)来理解。
一个 Superstep 大致分为三个阶段:
1. Plan / Routing ↓ 2. Execution ↓ 3. Update / Commit4.1 Plan / Routing
LangGraph 根据当前 State 和 Edge,判断这一轮需要执行哪些节点。
4.2 Execution
本轮被选中的节点开始执行。
如果有多个节点在同一个 Superstep 中被触发,它们会基于本轮开始时的状态快照分别执行。
也就是说,同一轮中的某个节点刚产生的更新,不会立刻被另一个并行节点读取。
4.3 Update / Commit
等这一轮需要执行的节点全部结束后,LangGraph 才会统一处理它们返回的状态更新,并生成新的 State。
新的 State 再进入下一轮 Superstep。
所以更准确的理解应该是:
读取当前状态快照 | v 执行本轮节点 | v 收集节点返回的局部更新 | v 统一合并到 State | v 进入下一轮5. Edge 不只能表示固定顺序,还可以实现条件分支
前面的边都是固定的:
builder.add_edge("node_a","node_b")但真实 Agent 中经常会出现这样的逻辑:
用户要诗歌 -> poem_node 用户要笑话 -> joke_node这时就可以使用:
add_conditional_edges()例如:
fromtypingimportTypedDict,Literalfromlanggraph.graphimportStateGraph,START,ENDclassContentState(TypedDict):topic:strcontent_type:strresult:strdefpoem_node(state:ContentState):return{"result":f"生成一首关于{state['topic']}的诗"}defjoke_node(state:ContentState):return{"result":f"生成一个关于{state['topic']}的笑话"}defrouter(state:ContentState)->Literal["poem","joke"]:if"诗"instate["content_type"]:return"poem"return"joke"然后定义图:
builder=StateGraph(ContentState)builder.add_node("poem_node",poem_node)builder.add_node("joke_node",joke_node)builder.add_conditional_edges(START,router,path_map={"poem":"poem_node","joke":"joke_node",})builder.add_edge("poem_node",END)builder.add_edge("joke_node",END)graph=builder.compile()这里注意三个参数:
builder.add_conditional_edges(source,path,path_map)source 条件从哪里开始判断 path 使用哪个路由函数判断 path_map 路由函数返回值最终对应哪个真实节点例如:
router(...)->"poem"再通过:
"poem":"poem_node"最终进入poem_node。
为什么要用path_map?
当然也可以让router()直接返回:
"poem_node"但这样路由函数会和图中的节点名称绑定得比较紧。
使用:
"poem""joke"这样的业务标签,再通过path_map映射到真实节点,代码的语义通常会更清晰。
6. 两个节点同时更新一个字段,会发生什么?
假设我们从START同时连接两个节点:
builder.add_edge(START,"node_a")builder.add_edge(START,"node_b")这两个节点就可能在同一个 Superstep 中被调度。
如果它们分别修改不同字段:
node_a->poem node_b->joke状态合并比较直接。
但如果它们都更新:
logs就出现了一个问题:
node_a 返回 ["node_a"] node_b 返回 ["node_b"] 最后 logs 到底应该是什么?这正是Reducer要解决的问题。
例如:
fromtypingimportTypedDict,AnnotatedfromoperatorimportaddclassOverallState(TypedDict):logs:Annotated[list[str],add]其中:
Annotated[list[str],add]可以拆开理解:
list[str] ↓ 这个字段的数据类型 add ↓ 这个字段出现多次更新时采用的合并规则假设原始状态是:
["START"]节点返回:
["node_a"]那么operator.add的效果就是:
["START"]+["node_a"]最终:
["START","node_a"]所以 Reducer 的本质并不神秘:
它就是告诉 LangGraph:同一个状态字段出现新的更新值时,旧值和新值应该怎样合并。
对于并行节点来说,这一点更加重要。多个节点如果在同一个 Superstep 中同时写入同一状态字段,而这个字段又没有合适的合并规则,就可能产生状态更新冲突。
7. 把整个执行过程串起来
到这里,我们可以把 LangGraph 最核心的运行逻辑串起来:
graph.invoke(input) | v 把输入放入 State | v 根据 Edge 判断下一轮执行节点 | v Node 读取当前 State 快照 | v Node 返回局部状态更新 | v Reducer 处理需要合并的字段 | v 提交新的 State | v 根据 Edge 继续路由 | v 直到 END