02 状态（State）

tags:/n - langgraph状态管理TypedDictPydantic面试aliases:/n - State状态定义02 状态StateState 是 LangGraph 最核心的概念代表了整个图的数据目标和节点的修改目标。一、状态的定义1.1 三种定义方式方式一TypedDict推荐fromtypingimportTypedDictclassMyState(TypedDict):rag_result:strweb_search_result:strfinal_answer:strquery:str方式二Pydantic BaseModelfrompydanticimportBaseModelclassMyState(BaseModel):rag_result:strweb_search_result:strfinal_answer:strquery:str方式三DataclassfromdataclassesimportdataclassdataclassclassMyState:rag_result:strweb_search_result:strfinal_answer:strquery:str二、输入输出数据隔离通过StateGraph的三个参数实现精细控制参数作用是否必须state_schema图的完整内部状态必须input_schema图接受的输入可选output_schema图返回的输出可选使用场景input_schema限制用户只能传入特定字段如queryoutput_schema限制返回给用户的结果如只返回final_answerstate_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:strgraphStateGraph(state_schemaMyStateFull,input_schemaInputSchema,output_schemaOutputSchema)defrag_search_node(state:MyStateFull):querystate[query]return{rag_result:f关于{query}的rag_result,a_new_key:value}deffinal_answer_node(state:MyStateFull):ragstate[rag_result]return{final_answer:fLLM基于{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)compiledgraph.compile()# 输入只能是 InputSchema 中的字段resultcompiled.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]}builderStateGraph(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)graphbuilder.compile()resultgraph.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):returnleftrightclassMyAgent(TypedDict):messages:Annotated[List[BaseMessage],add_message]deftool_node(state):return{messages:[HumanMessage(content来自tool的内容)]}defllm_node(state):return{messages:[AIMessage(content来自llm的内容)]}builderStateGraph(state_schemaMyAgent)builder.add_node(tool_node)builder.add_node(llm_node)builder.add_edge(START,tool_node)builder.add_edge(tool_node,llm_node)graphbuilder.compile()resultgraph.invoke({messages:[HumanMessage(content你好)]})print(result[messages])四、状态的存储4.1 Checkpointer 状态存储演示 运行方式python 02_状态存储.py fromlanggraph.checkpoint.memoryimportInMemorySaverfromlangchain.agentsimportcreate_agentfromlangchain.toolsimporttoolfromlangchain_openaiimportChatOpenAI# 1. 创建 checkpointercheckpointerInMemorySaver()llmChatOpenAI(modelgpt-4o-mini)tooldefweather_tool(city:str,date:str)-str:查天气的工具returnf{city}在{date}的天气是晴朗的# 2. 构建 Agentagentcreate_agent(modelllm,tools[weather_tool],checkpointercheckpointer)# 3. 第一次调用config{configurable:{thread_id:user_1}}result1agent.invoke({messages:北京天气怎么样},configconfig)print(f第一次:{result1[messages][-1]})# 4. 第二次调用相同 thread_id上下文自动延续result2agent.invoke({messages:适合出去玩吗},configconfig)print(f第二次:{result2[messages][-1]})4.2 故障恢复 故障恢复演示 运行方式python 02_故障恢复.py importsqlite3fromlanggraph.checkpoint.sqliteimportSqliteSaverfromlanggraph.graphimportStateGraph,START# 使用 SQLite 持久化connsqlite3.connect(checkpointer.db,check_same_threadFalse)memorySqliteSaver(conn)# ... 构建图 ...graphbuilder.compile(checkpointermemory)# 第一次调用可能在某节点报错config{configurable:{thread_id:123}}try:graph.invoke({},configconfig)exceptExceptionase:print(f出错:{e})# 恢复执行传入 None 相同 configresultgraph.invoke(None,configconfig)五、获取历史状态 获取历史状态演示 运行方式python 02_历史状态.py # 获取所有历史状态historylist(graph.get_state_history(config))forstateinhistory:print(state)# 获取最近一次状态last_stategraph.get_state(config)print(last_state)相关笔记[[01-LangGraph概述与快速入门]] · [[03-节点]] · [[04-边]]