下面我们来深入拆解这段 LangGraph 代码中出现的 四种 State(状态)形式,并结合执行流程说明它们各自的作用、约束与数据流转方式。
1. 四种 State 的定义与角色
① InputState(输入状态)
pythonclassInputState(TypedDict):user_input:str- 角色:定义外部调用者向图(Graph)发起请求时必须传入的数据结构。
- 使用位置:在创建 StateGraph时通过 input_schema=InputState指定。
- 特点:
仅描述“入口数据”,不包含图内部产生的中间或最终结果。
调用 graph.invoke({“user_input”: “My”})时,参数必须兼容 InputState。 - 与本例关系:初始全局状态中只有 user_input: “My”。
② OutputState(输出状态)
pythonclassOutputState(TypedDict):graph_output:str- 角色:定义图执行完毕后返回给外部的结果结构。
- 使用位置:在创建 StateGraph时通过 output_schema=OutputState指定。
- 特点:
对全局状态进行“裁剪”,最终只保留此处声明的字段。
示例中返回 {‘graph_output’: ‘My name is Lance’}。 - 与本例关系:即使全局状态还包含 foo、bar、user_input,输出时也被过滤得只剩 graph_output。
③ OverallState(总体状态)
pythonclassOverallState(TypedDict):foo:struser_input:strgraph_output:str- 角色:定义整个图内部维护的完整全局状态(即“总状态”)。
- 使用位置:作为 StateGraph的第一个类型参数:StateGraph(OverallState, …)。
- 特点:
理论上应囊括所有节点需要读写的状态字段(本例为了演示故意没放 bar)。
各节点函数的输入/输出可以只是它的子集,LangGraph 负责合并。
初始时由 InputState注入(user_input),随后被各个节点逐步填充。 - 与本例关系:
初始:{“user_input”: “My”}(foo、graph_output暂缺)
node_1 写入 foo→ {“user_input”: “My”, “foo”: “My name”}
node_3 写入 graph_output→ 最终全局状态含三个键(外加临时的 bar)
④ PrivateState(私有/局部状态)
pythonclassPrivateState(TypedDict):bar:str- 角色:定义某两个(或多个)相邻节点之间传递的临时数据结构,不属于全局状态的正式 schema。
- 使用位置:仅作为某些节点的输入输出类型标注(非图级配置)。
特点:
用于封装中间计算结果,避免污染 OverallState。
LangGraph 允许节点返回未在 OverallState中声明的字段,并传递给下一节点(只要下一节点标注接收该结构)。
最终不会被 OutputState输出。 - 与本例关系:
node_2返回 {“bar”: “My name is”}(写入私有状态)
node_3接收 PrivateState,读取 bar并生成最终输出。
2. 数据流转全过程(结合代码)
python graph.invoke({"user_input":"My"})| 步骤 | 节点 | 接收的状态(类型标注) | 实际可访问的字段 | 返回的状态 | 全局状态变化 |
|---|---|---|---|---|---|
| START→node_1 | node_1 | InputState | user_input=“My” | {“foo”: “My name”} | + foo |
| node_1→node_2 | node_2 | OverallState | user_input, foo | {“bar”: “My name is”} | + bar(私有) |
| node_2→node_3 | node_3 | PrivateState | bar | {“graph_output”: “My name is Lance”} | + graph_output |
| node_3→END | 输出 | OutputState | — | — | 仅保留 graph_output |
最终输出:
python{'graph_output':'My name is Lance'}3. 核心设计思想总结
关注点分离:
InputState:对外契约(调用方知道传什么)OutputState:对外契约(调用方知道收什么)OverallState:对内全景(图作者管理全量状态)PrivateState:节点间临时契约(解耦中间逻辑)
类型驱动:LangGraph 利用 Python 类型标注(TypedDict)在编译期/运行期做状态校验与自动补全提示。
渐进式状态构建:全局状态从输入开始,经各节点返回的部分状态不断合并,最后按输出 schema 剪裁。
代码
此代码取自官方给的demo
fromtypingimportTypedDictfromlanggraph.graphimportEND,START,StateGraphclassInputState(TypedDict):user_input:strclassOutputState(TypedDict):graph_output:strclassOverallState(TypedDict):foo:struser_input:strgraph_output:strclassPrivateState(TypedDict):bar:strdefnode_1(state:InputState)->OverallState:# Write to OverallStatereturn{"foo":state["user_input"]+" name"}defnode_2(state:OverallState)->PrivateState:# Read from OverallState, write to PrivateStatereturn{"bar":state["foo"]+" is"}defnode_3(state:PrivateState)->OutputState:# Read from PrivateState, write to OutputStatereturn{"graph_output":state["bar"]+" Lance"}builder=StateGraph(OverallState,input_schema=InputState,output_schema=OutputState)builder.add_node("node_1",node_1)builder.add_node("node_2",node_2)builder.add_node("node_3",node_3)builder.add_edge(START,"node_1")builder.add_edge("node_1","node_2")builder.add_edge("node_2","node_3")builder.add_edge("node_3",END)graph=builder.compile()graph.invoke({"user_input":"My"})# {'graph_output': 'My name is Lance'}💡 实践中,为了避免混淆,通常建议把所有会被多个节点使用的字段都放进
OverallState,而将PrivateState视为可选的精细化工具。本例特意将bar放在PrivateState是为了演示四种形态的共存与区别。