02 状态（State）

tags:/n - langgraph

• 状态管理
• TypedDict
• Pydantic
• 面试
  aliases:/n - State
• 状态定义

02 状态（State）

State 是 LangGraph 最核心的概念，代表了整个图的数据目标和节点的修改目标。

一、状态的定义

1.1 三种定义方式

方式一：TypedDict（推荐）

fromtypingimportTypedDictclassMyState(TypedDict):rag_result:strweb_search_result:strfinal_answer:strquery:str

方式二：Pydantic BaseModel

frompydanticimportBaseModelclassMyState(BaseModel):rag_result:strweb_search_result:strfinal_answer:strquery:str

方式三：Dataclass

fromdataclassesimportdataclass@dataclassclassMyState:rag_result:strweb_search_result:strfinal_answer:strquery:str

二、输入输出数据隔离

通过StateGraph的三个参数实现精细控制：

使用场景：

• input_schema：限制用户只能传入特定字段（如query）
• output_schema：限制返回给用户的结果（如只返回final_answer）
• state_schema：定义图内部所有可能用到的字段

""" 输入输出数据隔离演示 运行方式：python 02_输入输出隔离.py """fromtypingimportTypedDictfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphclassMyStateFull(TypedDict):rag_result:strweb_search_result:strfinal_answer:strquery:stra_new_key:strclassInputSchema(TypedDict):query:strclassOutputSchema(TypedDict):final_answer:strgraph=StateGraph(state_schema=MyStateFull,input_schema=InputSchema,output_schema=OutputSchema)defrag_search_node(state:MyStateFull):query=state["query"]return{"rag_result":f"关于{query}的rag_result","a_new_key":"value"}deffinal_answer_node(state:MyStateFull):rag=state["rag_result"]return{"final_answer":f"LLM基于{rag}的回复"}graph.add_node(rag_search_node)graph.add_node(final_answer_node)graph.add_edge(START,"rag_search_node")graph.add_edge("rag_search_node","final_answer_node")graph.add_edge("final_answer_node",END)compiled=graph.compile()# 输入只能是 InputSchema 中的字段result=compiled.invoke({"query":"如何使用LangGraph"})print(result)# 只返回 output_schema 中的字段

三、Reducer 函数

Reducer 解决多节点并发写入同一字段时的合并问题。

3.1 默认行为：覆盖更新

""" 默认 Reducer 演示（覆盖更新） 运行方式：python 02_默认Reducer.py """fromtypingimportListfromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassDefaultReducerState(TypedDict):foo:intbar:List[str]defnode_1(state):return{"foo":2}defnode_2(state):return{"bar":["bye"]}builder=StateGraph(DefaultReducerState)builder.add_node("node1",node_1)builder.add_node("node2",node_2)builder.add_edge(START,"node1")builder.add_edge("node1","node2")builder.add_edge("node2",END)graph=builder.compile()result=graph.invoke({"foo":1,"bar":["hi"]})print(result)# {'foo': 2, 'bar': ['bye']} ← 覆盖更新

3.2 自定义 Reducer

""" 自定义 Reducer 演示 运行方式：python 02_自定义Reducer.py """fromlanggraph.graphimportStateGraphfromlangchain_core.messagesimportHumanMessage,AIMessage,BaseMessagefromtypingimportAnnotated,TypedDict,Listfromlanggraph.constantsimportSTARTdefadd_message(left:list,right:list):returnleft+rightclassMyAgent(TypedDict):messages:Annotated[List[BaseMessage],add_message]deftool_node(state):return{"messages":[HumanMessage(content="来自tool的内容")]}defllm_node(state):return{"messages":[AIMessage(content="来自llm的内容")]}builder=StateGraph(state_schema=MyAgent)builder.add_node(tool_node)builder.add_node(llm_node)builder.add_edge(START,"tool_node")builder.add_edge("tool_node","llm_node")graph=builder.compile()result=graph.invoke({"messages":[HumanMessage(content="你好")]})print(result['messages'])

四、状态的存储

4.1 Checkpointer

""" 状态存储演示 运行方式：python 02_状态存储.py """fromlanggraph.checkpoint.memoryimportInMemorySaverfromlangchain.agentsimportcreate_agentfromlangchain.toolsimporttoolfromlangchain_openaiimportChatOpenAI# 1. 创建 checkpointercheckpointer=InMemorySaver()llm=ChatOpenAI(model="gpt-4o-mini")@tooldefweather_tool(city:str,date:str)->str:"""查天气的工具"""returnf'{city}在{date}的天气是晴朗的'# 2. 构建 Agentagent=create_agent(model=llm,tools=[weather_tool],checkpointer=checkpointer)# 3. 第一次调用config={"configurable":{"thread_id":"user_1"}}result1=agent.invoke({"messages":"北京天气怎么样"},config=config)print(f"第一次:{result1['messages'][-1]}")# 4. 第二次调用（相同 thread_id，上下文自动延续）result2=agent.invoke({"messages":"适合出去玩吗"},config=config)print(f"第二次:{result2['messages'][-1]}")

4.2 故障恢复

""" 故障恢复演示 运行方式：python 02_故障恢复.py """importsqlite3fromlanggraph.checkpoint.sqliteimportSqliteSaverfromlanggraph.graphimportStateGraph,START# 使用 SQLite 持久化conn=sqlite3.connect("checkpointer.db",check_same_thread=False)memory=SqliteSaver(conn)# ... 构建图 ...graph=builder.compile(checkpointer=memory)# 第一次调用（可能在某节点报错）config={"configurable":{"thread_id":"123"}}try:graph.invoke({},config=config)exceptExceptionase:print(f"出错:{e}")# 恢复执行（传入 None + 相同 config）result=graph.invoke(None,config=config)

五、获取历史状态

""" 获取历史状态演示 运行方式：python 02_历史状态.py """# 获取所有历史状态history=list(graph.get_state_history(config))forstateinhistory:print(state)# 获取最近一次状态last_state=graph.get_state(config)print(last_state)

相关笔记

[[01-LangGraph概述与快速入门]] · [[03-节点]] · [[04-边]]