在 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 是怎样组合起来的定义完状态和节点以后就可以开始构建图builderStateGraph(state_schemaOverallState)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)graphbuilder.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更像是在描述图结构。真正执行之前需要graphbuilder.compile()然后再调用resultgraph.invoke({username:小黄})print(result)最终可以得到类似{username:小黄,greeting:Dear 小黄,output:Dear 小黄你好}4. LangGraph 并不是一个节点改完状态下一个马上读取但 LangGraph 更准确的运行方式要结合Superstep超步来理解。一个 Superstep 大致分为三个阶段1. Plan / Routing ↓ 2. Execution ↓ 3. Update / Commit4.1 Plan / RoutingLangGraph 根据当前 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]:returnpoemreturnjoke然后定义图builderStateGraph(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)graphbuilder.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但这样路由函数会和图中的节点名称绑定得比较紧。使用poemjoke这样的业务标签再通过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