在 LangGraph 中**State状态**是整个系统最核心的概念之一。如果说 Graph 定义了执行结构那么 State 定义的就是系统的记忆与数据流动方式。可以用一句话概括LangGraph Stateful Graph Runtime也就是说LangGraph 并不是单纯的 workflow engine而是一个围绕 State 设计的 Agent 执行框架。一、为什么 LangGraph 需要 State在传统 LLM pipeline 中数据通常以函数参数传递output1 step1(input) output2 step2(output1) output3 step3(output2)这种模式的问题是数据流是线性的难以支持循环难以支持并行难以管理复杂上下文而 Agent 系统通常需要reasoning loop tool outputs multi-agent collaboration long-running tasks因此 LangGraph 引入共享状态模型StateGraph 的每个节点都可以读取 state 更新 state整个执行流程变成state → node → state → node二、State 的本质共享数据结构LangGraph 的 State 本质是一个结构化数据对象。通常定义为classState(TypedDict):messages:listtool_results:listplan:strGraph 运行时state { messages: [...], tool_results: [...], plan: ... }每个 Nodeinput: state output: partial state update关键原则Node 不直接修改 state Node 返回 state 更新 Runtime 负责合并这是一种immutable update model。三、State Schema 的定义方式LangGraph 支持多种 State 定义方式。最常见的是fromtyping_extensionsimportTypedDictclassState(TypedDict):messages:list也可以使用dataclass pydantic model但工程实践中最推荐的是TypedDict原因类型提示清晰运行开销低与 Python typing 生态兼容四、Node 如何更新 StateNode 函数通常形如defnode(state:State):return{messages:[hello]}注意返回值不是完整 state 而是 partial updateRuntime 会执行state merge(state, update)例如初始 statemessages [hi]node 返回{messages: [hello]}如果没有 reducermessages [hello]如果有 reducermessages [hi, hello]因此Reducer 决定 State 的合并方式五、Reducer 机制State 的核心Reducer 是 LangGraph State 设计的关键。Reducer 定义多个更新如何合并例如fromtypingimportAnnotatedimportoperatorclassState(TypedDict):messages:Annotated[list,operator.add]含义list list执行[hi] [hello]结果[hi, hello]常见 Reducer1 list 累加Annotated[list,operator.add]适用于messages tool results logs2 数值累加Annotated[int,operator.add]适用于token count step count metrics3 自定义 reducerdefmerge(a,b):returnab然后Annotated[list,merge]六、Reducer 为什么重要Reducer 的设计解决了一个关键问题并行节点如何更新 state假设 Graph 中存在并行节点A ├─ B └─ C执行B(state) C(state)如果 B 和 C 同时更新messages就会产生冲突。Reducer 定义如何 merge B 和 C 的更新例如[msgB] [msgC]最终[msgB, msgC]七、State 在并行执行中的行为LangGraph 支持并行节点。关键机制是State Isolation执行流程state ↓ branch B branch C两个 branchstate_B state_C执行完成后merge(state_B, state_C)合并规则reducer因此 State 模型可以安全支持parallel tasks map-reduce multi-agent execution八、State 与 MessagesStateLangGraph 官方提供了一个常用状态MessagesState定义classMessagesState(TypedDict):messages:Annotated[list[AnyMessage],operator.add]作用自动累加聊天记录适用于chat agent tool calling agent ReAct agent执行效果user message assistant message tool message全部累积在 state 中。九、State 与 Agent Memory很多人会问LangGraph state 和 memory 有什么区别区别如下类型作用StateGraph 运行时状态Memory跨会话长期存储State 生命周期graph invocationMemory 生命周期multiple sessions例如Statecurrent conversationMemoryuser profile十、State 与 CheckpointLangGraph 支持checkpoint。Checkpoint 的核心就是保存 state执行流程node ↓ state update ↓ checkpoint save恢复load checkpoint resume graph适用于long running agents human approval fault recovery十一、State 设计最佳实践在真实项目中State 设计非常关键。推荐原则1 保持 State 扁平避免deep nested objects推荐classState(TypedDict):messages:listplan:strresults:list2 明确 Reducer所有可能并行更新的字段必须定义 reducer否则会产生state overwrite3 区分短期与长期状态state → short-term memory → long-term4 控制 State 大小State 会进入 LLM context。过大可能导致token explosion建议summary truncation十二、State 的架构意义LangGraph 的 State 模型实际上融合了多个系统设计思想Functional Programming Actor Model Event Sourcing特点immutable update state merging concurrent safety因此 LangGraph 可以支持parallel agents multi-agent collaboration long-running workflows十三、总结LangGraph 的 State 设计是整个系统的核心。可以总结为四个关键点1 State 是共享数据结构 2 Node 返回 partial updates 3 Reducer 定义 merge 规则 4 Runtime 负责 state 合并完整模型State ↓ Node execution ↓ Partial update ↓ Reducer merge ↓ Next node这使 LangGraph 成为一个真正的Stateful Agent Runtime而不是简单的 LLM pipeline。from pprint import pprint from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver from langchain_core.runnables import RunnableConfig from typing import Annotated from typing_extensions import TypedDict from operator import add import copy # ----------------------- # 定义状态类型 # ----------------------- class State(TypedDict): foo: str bar: Annotated[list[str], add] # ----------------------- # 定义节点 # ----------------------- def node_a(state: State): return {foo: a, bar: [a]} def node_b(state: State): return {foo: b, bar: [b]} # ----------------------- # 构建图 # ----------------------- workflow StateGraph(State) workflow.add_node(node_a) workflow.add_node(node_b) workflow.add_edge(START, node_a) workflow.add_edge(node_a, node_b) workflow.add_edge(node_b, END) # ----------------------- # 配置 checkpointer # ----------------------- checkpointer InMemorySaver() graph workflow.compile(checkpointercheckpointer) thread_id 1 config: RunnableConfig {configurable: {thread_id: thread_id}} # ----------------------- # 执行完整图 # ----------------------- print( 执行完整图 ) graph.invoke({foo: , bar: []}, config) # 查看新线程保存的 checkpoint print(\n state history ) for state in graph.get_state_history(config): pprint(state) print(----)
LangGraph State 深度解析
在 LangGraph 中**State状态**是整个系统最核心的概念之一。如果说 Graph 定义了执行结构那么 State 定义的就是系统的记忆与数据流动方式。可以用一句话概括LangGraph Stateful Graph Runtime也就是说LangGraph 并不是单纯的 workflow engine而是一个围绕 State 设计的 Agent 执行框架。一、为什么 LangGraph 需要 State在传统 LLM pipeline 中数据通常以函数参数传递output1 step1(input) output2 step2(output1) output3 step3(output2)这种模式的问题是数据流是线性的难以支持循环难以支持并行难以管理复杂上下文而 Agent 系统通常需要reasoning loop tool outputs multi-agent collaboration long-running tasks因此 LangGraph 引入共享状态模型StateGraph 的每个节点都可以读取 state 更新 state整个执行流程变成state → node → state → node二、State 的本质共享数据结构LangGraph 的 State 本质是一个结构化数据对象。通常定义为classState(TypedDict):messages:listtool_results:listplan:strGraph 运行时state { messages: [...], tool_results: [...], plan: ... }每个 Nodeinput: state output: partial state update关键原则Node 不直接修改 state Node 返回 state 更新 Runtime 负责合并这是一种immutable update model。三、State Schema 的定义方式LangGraph 支持多种 State 定义方式。最常见的是fromtyping_extensionsimportTypedDictclassState(TypedDict):messages:list也可以使用dataclass pydantic model但工程实践中最推荐的是TypedDict原因类型提示清晰运行开销低与 Python typing 生态兼容四、Node 如何更新 StateNode 函数通常形如defnode(state:State):return{messages:[hello]}注意返回值不是完整 state 而是 partial updateRuntime 会执行state merge(state, update)例如初始 statemessages [hi]node 返回{messages: [hello]}如果没有 reducermessages [hello]如果有 reducermessages [hi, hello]因此Reducer 决定 State 的合并方式五、Reducer 机制State 的核心Reducer 是 LangGraph State 设计的关键。Reducer 定义多个更新如何合并例如fromtypingimportAnnotatedimportoperatorclassState(TypedDict):messages:Annotated[list,operator.add]含义list list执行[hi] [hello]结果[hi, hello]常见 Reducer1 list 累加Annotated[list,operator.add]适用于messages tool results logs2 数值累加Annotated[int,operator.add]适用于token count step count metrics3 自定义 reducerdefmerge(a,b):returnab然后Annotated[list,merge]六、Reducer 为什么重要Reducer 的设计解决了一个关键问题并行节点如何更新 state假设 Graph 中存在并行节点A ├─ B └─ C执行B(state) C(state)如果 B 和 C 同时更新messages就会产生冲突。Reducer 定义如何 merge B 和 C 的更新例如[msgB] [msgC]最终[msgB, msgC]七、State 在并行执行中的行为LangGraph 支持并行节点。关键机制是State Isolation执行流程state ↓ branch B branch C两个 branchstate_B state_C执行完成后merge(state_B, state_C)合并规则reducer因此 State 模型可以安全支持parallel tasks map-reduce multi-agent execution八、State 与 MessagesStateLangGraph 官方提供了一个常用状态MessagesState定义classMessagesState(TypedDict):messages:Annotated[list[AnyMessage],operator.add]作用自动累加聊天记录适用于chat agent tool calling agent ReAct agent执行效果user message assistant message tool message全部累积在 state 中。九、State 与 Agent Memory很多人会问LangGraph state 和 memory 有什么区别区别如下类型作用StateGraph 运行时状态Memory跨会话长期存储State 生命周期graph invocationMemory 生命周期multiple sessions例如Statecurrent conversationMemoryuser profile十、State 与 CheckpointLangGraph 支持checkpoint。Checkpoint 的核心就是保存 state执行流程node ↓ state update ↓ checkpoint save恢复load checkpoint resume graph适用于long running agents human approval fault recovery十一、State 设计最佳实践在真实项目中State 设计非常关键。推荐原则1 保持 State 扁平避免deep nested objects推荐classState(TypedDict):messages:listplan:strresults:list2 明确 Reducer所有可能并行更新的字段必须定义 reducer否则会产生state overwrite3 区分短期与长期状态state → short-term memory → long-term4 控制 State 大小State 会进入 LLM context。过大可能导致token explosion建议summary truncation十二、State 的架构意义LangGraph 的 State 模型实际上融合了多个系统设计思想Functional Programming Actor Model Event Sourcing特点immutable update state merging concurrent safety因此 LangGraph 可以支持parallel agents multi-agent collaboration long-running workflows十三、总结LangGraph 的 State 设计是整个系统的核心。可以总结为四个关键点1 State 是共享数据结构 2 Node 返回 partial updates 3 Reducer 定义 merge 规则 4 Runtime 负责 state 合并完整模型State ↓ Node execution ↓ Partial update ↓ Reducer merge ↓ Next node这使 LangGraph 成为一个真正的Stateful Agent Runtime而不是简单的 LLM pipeline。from pprint import pprint from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver from langchain_core.runnables import RunnableConfig from typing import Annotated from typing_extensions import TypedDict from operator import add import copy # ----------------------- # 定义状态类型 # ----------------------- class State(TypedDict): foo: str bar: Annotated[list[str], add] # ----------------------- # 定义节点 # ----------------------- def node_a(state: State): return {foo: a, bar: [a]} def node_b(state: State): return {foo: b, bar: [b]} # ----------------------- # 构建图 # ----------------------- workflow StateGraph(State) workflow.add_node(node_a) workflow.add_node(node_b) workflow.add_edge(START, node_a) workflow.add_edge(node_a, node_b) workflow.add_edge(node_b, END) # ----------------------- # 配置 checkpointer # ----------------------- checkpointer InMemorySaver() graph workflow.compile(checkpointercheckpointer) thread_id 1 config: RunnableConfig {configurable: {thread_id: thread_id}} # ----------------------- # 执行完整图 # ----------------------- print( 执行完整图 ) graph.invoke({foo: , bar: []}, config) # 查看新线程保存的 checkpoint print(\n state history ) for state in graph.get_state_history(config): pprint(state) print(----)
