langgraph在节点中访问State

文章目录节点中访问State图节点中读取State图节点更新StateOverwrite绕过Reducer总结节点中访问State图节点中读取State在 LangGraph 中节点本质上是一个可调用对象通常定义为普通 Python 函数。节点函数被执行时LangGraph 会自动将当前图运行到该节点时的State传入节点函数。节点函数的第一个参数通常是当前运行图的状态对象也就是State。def node(state: StateSchema): ...其中state表示当前节点执行时可以访问到的全局状态快照。节点可以通过读取state中的字段获取上游节点写入的数据并基于这些数据完成当前节点的业务逻辑。示例代码如下fromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDict,AnnotatedfromoperatorimportaddclassOverAllState(TypedDict):logs:Annotated[list[str],add]id:strdefnode_a(state:OverAllState):fork,vinstate.items():print(fk:{k}, v:{v})builderStateGraph(state_schemaOverAllState)builder.add_node(node_a,node_a)builder.add_edge(START,node_a)builder.add_edge(node_a,END)graphbuilder.compile()resultgraph.invoke({logs:[START],id:start})输出如下k: logs, v: [START] k: id, v: start图节点更新State在 LangGraph 中节点函数通常不需要返回更新后的完整状态只需要返回本节点对状态的局部更新。也就是说节点的返回值可以只包含需要修改的状态字段。对于节点没有返回的字段LangGraph 会保留其原有状态值对于节点返回的字段LangGraph 会根据该字段是否配置了Reducer来决定如何合并更新值。如果字段配置了Reducer则使用对应的Reducer函数将旧值和新值合并如果字段没有配置Reducer则按照默认规则使用节点返回的新值覆盖原值。LangGraph运行时会按照状态字段的Reducer函数将其与当前的最新状态合并。示例代码如下fromlanggraph.graphimportStateGraph,START,ENDfromtypingimportTypedDict,AnnotatedfromoperatorimportaddclassOverAllState(TypedDict):logs:Annotated[list[str],add]id:strdefnode_a(state:OverAllState):fork,vinstate.items():print(fk:{k}, v:{v})return{logs:[node_a 更新状态]}builderStateGraph(state_schemaOverAllState)builder.add_node(node_a,node_a)builder.add_edge(START,node_a)builder.add_edge(node_a,END)graphbuilder.compile()resultgraph.invoke({logs:[START],id:start})print(*30,- result -,*30)print(result)输出如下k: logs, v: [START] k: id, v: start - result - {logs: [START, node_a 更新状态], id: start}在上述示例中logs字段通过Annotated[list[str], add]绑定了 Reducer 函数operator.add因此 LangGraph 会将原有的logs值和node_a返回的新logs值进行列表拼接[START] [node_a 更新状态]最终得到[START, node_a 更新状态]id字段没有出现在node_a的返回值中因此该字段不会被更新。图运行结束后输出状态中的id仍然保持输入时的值id: start因此LangGraph 节点更新 State 的核心规则可以概括为节点只返回需要更新的字段未返回的字段保持不变返回的字段根据是否配置 Reducer 决定是合并还是覆盖。Overwrite绕过Reducer在前面的示例中如果某个状态字段定义了Reducer那么节点返回该字段的更新值时LangGraph 默认会通过对应的Reducer将新值与已有状态值进行合并。在某些场景下我们可能并不希望继续执行Reducer的聚合逻辑而是希望本次更新直接覆盖旧值。这时可以使用Overwrite。Overwrite的作用是告诉 LangGraph 本次状态更新不走该字段原本定义的Reducer而是直接用新值覆盖状态中的旧值。需要注意的是Overwrite只影响当前这一次更新并不会修改状态字段本身的Reducer定义。后续节点如果继续正常返回该字段的更新值仍然会按照原来的Reducer逻辑进行合并。示例代码如下fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportOverwritefromtypingimportTypedDict,AnnotatedfromoperatorimportaddclassOverAllState(TypedDict):logs:Annotated[list[str],add]id:strdefnode_a(state:OverAllState):return{logs:[node_a],id:node_a}defnode_b(state:OverAllState):return{logs:Overwrite([node_b]),id:node_b}defnode_c(state:OverAllState):return{logs:[node_c],id:node_c}builderStateGraph(state_schemaOverAllState)builder.add_node(node_a,node_a)builder.add_node(node_b,node_b)builder.add_node(node_c,node_c)builder.add_edge(START,node_a)builder.add_edge(node_a,node_b)builder.add_edge(node_b,node_c)builder.add_edge(node_c,END)graphbuilder.compile()resultgraph.invoke({logs:[START],id:start})print(*30,- result -,*30)print(result)输出如下 - result - {logs: [node_b, node_c], id: node_c}上述代码中logs字段绑定了operator.add()函数如果没有Overwrite则最终输出的logs的值应为[START, node_a, node_b, node_c]node_b返回更新时用Overwrite包裹了logs字段的值那么当前状态的logs会被[node_b]覆盖因此最终输出的logs字段值变成了[node_b, node_c]id字段按照默认行为保留最后一次更新的值。总结注意 from langgraph.types import Overwrite 的使用