LangGraph状态管理实战多轮交互中维护Agent状态上一篇讲了LangGraph的核心概念。这一篇深入状态管理。状态管理是LangGraph最有价值的功能。Agent执行复杂任务的时候中间会产生大量信息。搜索结果、中间推理、用户偏好、任务进度。这些信息要在多个节点之间共享和更新。状态管理做不好Agent就会健忘。做了前面忘了后面或者信息在不同步骤之间对不上。状态的设计状态设计是第一步。好的状态设计能让工作流清晰高效。设计状态要考虑几个问题。需要存什么信息。把任务执行过程中需要共享的信息都列出来。对话历史、搜索结果、中间结论、任务进度、用户信息。信息之间有没有层级。简单的用扁平结构就行。复杂的可以嵌套。哪些信息是只读的哪些需要更新。只读的不用管需要更新的要想好更新策略。fromtypingimportTypedDict,List,OptionalclassResearchState(TypedDict):# 用户输入topic:str# 研究主题user_level:str# 用户水平初学者/中级/专家# 搜索相关search_queries:List[str]# 搜索关键词列表search_results:List[str]# 搜索结果search_round:int# 搜索轮次# 分析相关key_findings:List[str]# 关键发现summary:str# 总结# 控制流status:str# 当前状态retry_count:int# 重试次数max_retries:int# 最大重试次数这个状态设计覆盖了一个研究Agent需要的所有信息。用户输入、搜索过程、分析结果、控制流分门别类。状态的更新机制LangGraph的状态更新机制是这样的。每个节点返回一个字典只包含要更新的字段。LangGraph会把这个字典合并到当前状态里。defsearch_node(state:ResearchState)-dict:topicstate[topic]round_numstate.get(search_round,0)1# 执行搜索new_resultf第{round_num}轮搜索{topic}的结果# 只返回要更新的字段return{search_results:state[search_results][new_result],search_round:round_num,status:searched,}对于列表类型的字段更新时要手动拼接旧值和新值。state[search_results] [new_result]把新结果追加到旧列表后面。这样写有点麻烦。LangGraph提供了Annotated类型可以自动处理列表的追加。fromtypingimportAnnotatedfromoperatorimportaddclassResearchState(TypedDict):search_results:Annotated[List[str],add]# 自动追加search_queries:Annotated[List[str],add]# 自动追加topic:strsummary:strstatus:strsearch_round:int用了Annotated和add以后节点返回时直接给新值就行LangGraph会自动追加。defsearch_node(state:ResearchState)-dict:new_resultf搜索{state[topic]}的结果# 直接给新值自动追加到现有列表return{search_results:[new_result],search_round:state.get(search_round,0)1,}消息列表的管理对话型Agent的状态里最常见的是消息列表。用户消息、AI消息、工具调用消息都要存。LangGraph提供了add_messagesreducer专门用来管理消息列表。fromtypingimportAnnotated,TypedDictfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportHumanMessage,AIMessageclassChatState(TypedDict):messages:Annotated[list,add_messages]user_id:strdefchat_node(state:ChatState)-dict:# 读取所有消息messagesstate[messages]# 调用大模型fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-3.5-turbo,temperature0)responsellm.invoke(messages)# 返回AI的回复自动追加到消息列表return{messages:[response]}add_messages会自动处理消息的追加。如果同一条消息被重复添加它还会自动去重。比手动拼接方便多了。多轮对话的状态管理多轮对话是状态管理的典型场景。用户跟Agent来回对话每轮对话都要参考之前的上下文。fromlanggraph.graphimportStateGraph,ENDfromtypingimportAnnotated,TypedDictfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportHumanMessage,AIMessage,SystemMessagefromlangchain_openaiimportChatOpenAIclassConversationState(TypedDict):messages:Annotated[list,add_messages]user_name:strtopic_count:int# 对话节点defchat_node(state:ConversationState)-dict:llmChatOpenAI(modelgpt-3.5-turbo,temperature0.7)messagesstate[messages]# 如果知道用户名字加到系统提示里ifstate.get(user_name):system_msgSystemMessage(contentf你是一个友好的助手。用户叫{state[user_name]}。)messages[system_msg]messages responsellm.invoke(messages)return{messages:[response],topic_count:state.get(topic_count,0)1,}# 判断是否结束defshould_continue(state:ConversationState)-str:ifstate.get(topic_count,0)5:returnENDreturnchat# 构建图workflowStateGraph(ConversationState)workflow.add_node(chat,chat_node)workflow.set_entry_point(chat)workflow.add_conditional_edges(chat,should_continue)appworkflow.compile()# 模拟多轮对话resultapp.invoke({messages:[HumanMessage(content你好我叫张三),HumanMessage(content我想聊聊AI Agent),HumanMessage(contentLangChain和LangGraph有什么区别),],user_name:张三,topic_count:0,})# 查看完整的对话历史formsginresult[messages]:role用户ifisinstance(msg,HumanMessage)else助手print(f{role}:{msg.content[:50]}...)状态持久化默认情况下LangGraph的状态存在内存里。程序结束就没了。要想跨会话保持状态需要用持久化存储。LangGraph支持用SQLite做持久化。fromlanggraph.checkpoint.sqliteimportSqliteSaverimportsqlite3# 创建SQLite持久化存储connsqlite3.connect(agent_state.db,check_same_threadFalse)checkpointerSqliteSaver(conn)# 编译时传入checkpointerappworkflow.compile(checkpointercheckpointer)# 运行时指定thread_id不同对话用不同的idconfig{configurable:{thread_id:user_001_session_1}}# 第一轮对话resultapp.invoke({messages:[HumanMessage(content我叫张三)]},configconfig,)# 第二轮对话同一个thread_id能访问之前的状态resultapp.invoke({messages:[HumanMessage(content我叫什么名字)]},configconfig,)# Agent能回答张三因为状态被持久化了用了持久化以后即使程序重启状态也不会丢。下次用同一个thread_id还能接着之前的对话继续。状态的查看和调试开发的时候经常需要查看中间状态。用stream方法可以看到每一步的状态变化。foreventinapp.stream({messages:[HumanMessage(content什么是RAG)],topic_count:0},configconfig,):fornode_name,state_updateinevent.items():print(f节点{node_name}执行完毕)print(f状态更新:{state_update})print()也可以用get_state方法查看当前状态。# 获取当前状态current_stateapp.get_state(config)print(current_state.values)调试的时候把状态打印出来看比猜来猜去高效得多。状态设计的几个建议状态不要太大。状态在节点之间传递太大会影响性能。特别是消息列表聊久了会很长。定期做摘要压缩。状态字段要有明确的语义。别用data1、data2这种名字。用search_results、key_findings这种一看就懂的名字。控制流字段和业务字段分开。status、retry_count这些是控制流用的。search_results、summary这些是业务数据。分开放逻辑更清晰。给状态加默认值。新建状态时给所有字段初始值。避免节点读到None报错。下一篇讲条件分支与循环。让Agent学会思考和重试处理更复杂的任务流程。
LangGraph状态管理实战多轮交互中维护Agent状态上一篇讲了LangGraph的核心概念。这一篇深入状态管理。状态管理是LangGraph最有价值的功能。Agent执行复杂任务的时候中间会产生大量信息。搜索结果、中间推理、用户偏好、任务进度。这些信息要在多个节点之间共享和更新。状态管理做不好Agent就会健忘。做了前面忘了后面或者信息在不同步骤之间对不上。状态的设计状态设计是第一步。好的状态设计能让工作流清晰高效。设计状态要考虑几个问题。需要存什么信息。把任务执行过程中需要共享的信息都列出来。对话历史、搜索结果、中间结论、任务进度、用户信息。信息之间有没有层级。简单的用扁平结构就行。复杂的可以嵌套。哪些信息是只读的哪些需要更新。只读的不用管需要更新的要想好更新策略。fromtypingimportTypedDict,List,OptionalclassResearchState(TypedDict):# 用户输入topic:str# 研究主题user_level:str# 用户水平初学者/中级/专家# 搜索相关search_queries:List[str]# 搜索关键词列表search_results:List[str]# 搜索结果search_round:int# 搜索轮次# 分析相关key_findings:List[str]# 关键发现summary:str# 总结# 控制流status:str# 当前状态retry_count:int# 重试次数max_retries:int# 最大重试次数这个状态设计覆盖了一个研究Agent需要的所有信息。用户输入、搜索过程、分析结果、控制流分门别类。状态的更新机制LangGraph的状态更新机制是这样的。每个节点返回一个字典只包含要更新的字段。LangGraph会把这个字典合并到当前状态里。defsearch_node(state:ResearchState)-dict:topicstate[topic]round_numstate.get(search_round,0)1# 执行搜索new_resultf第{round_num}轮搜索{topic}的结果# 只返回要更新的字段return{search_results:state[search_results][new_result],search_round:round_num,status:searched,}对于列表类型的字段更新时要手动拼接旧值和新值。state[search_results] [new_result]把新结果追加到旧列表后面。这样写有点麻烦。LangGraph提供了Annotated类型可以自动处理列表的追加。fromtypingimportAnnotatedfromoperatorimportaddclassResearchState(TypedDict):search_results:Annotated[List[str],add]# 自动追加search_queries:Annotated[List[str],add]# 自动追加topic:strsummary:strstatus:strsearch_round:int用了Annotated和add以后节点返回时直接给新值就行LangGraph会自动追加。defsearch_node(state:ResearchState)-dict:new_resultf搜索{state[topic]}的结果# 直接给新值自动追加到现有列表return{search_results:[new_result],search_round:state.get(search_round,0)1,}消息列表的管理对话型Agent的状态里最常见的是消息列表。用户消息、AI消息、工具调用消息都要存。LangGraph提供了add_messagesreducer专门用来管理消息列表。fromtypingimportAnnotated,TypedDictfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportHumanMessage,AIMessageclassChatState(TypedDict):messages:Annotated[list,add_messages]user_id:strdefchat_node(state:ChatState)-dict:# 读取所有消息messagesstate[messages]# 调用大模型fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-3.5-turbo,temperature0)responsellm.invoke(messages)# 返回AI的回复自动追加到消息列表return{messages:[response]}add_messages会自动处理消息的追加。如果同一条消息被重复添加它还会自动去重。比手动拼接方便多了。多轮对话的状态管理多轮对话是状态管理的典型场景。用户跟Agent来回对话每轮对话都要参考之前的上下文。fromlanggraph.graphimportStateGraph,ENDfromtypingimportAnnotated,TypedDictfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportHumanMessage,AIMessage,SystemMessagefromlangchain_openaiimportChatOpenAIclassConversationState(TypedDict):messages:Annotated[list,add_messages]user_name:strtopic_count:int# 对话节点defchat_node(state:ConversationState)-dict:llmChatOpenAI(modelgpt-3.5-turbo,temperature0.7)messagesstate[messages]# 如果知道用户名字加到系统提示里ifstate.get(user_name):system_msgSystemMessage(contentf你是一个友好的助手。用户叫{state[user_name]}。)messages[system_msg]messages responsellm.invoke(messages)return{messages:[response],topic_count:state.get(topic_count,0)1,}# 判断是否结束defshould_continue(state:ConversationState)-str:ifstate.get(topic_count,0)5:returnENDreturnchat# 构建图workflowStateGraph(ConversationState)workflow.add_node(chat,chat_node)workflow.set_entry_point(chat)workflow.add_conditional_edges(chat,should_continue)appworkflow.compile()# 模拟多轮对话resultapp.invoke({messages:[HumanMessage(content你好我叫张三),HumanMessage(content我想聊聊AI Agent),HumanMessage(contentLangChain和LangGraph有什么区别),],user_name:张三,topic_count:0,})# 查看完整的对话历史formsginresult[messages]:role用户ifisinstance(msg,HumanMessage)else助手print(f{role}:{msg.content[:50]}...)状态持久化默认情况下LangGraph的状态存在内存里。程序结束就没了。要想跨会话保持状态需要用持久化存储。LangGraph支持用SQLite做持久化。fromlanggraph.checkpoint.sqliteimportSqliteSaverimportsqlite3# 创建SQLite持久化存储connsqlite3.connect(agent_state.db,check_same_threadFalse)checkpointerSqliteSaver(conn)# 编译时传入checkpointerappworkflow.compile(checkpointercheckpointer)# 运行时指定thread_id不同对话用不同的idconfig{configurable:{thread_id:user_001_session_1}}# 第一轮对话resultapp.invoke({messages:[HumanMessage(content我叫张三)]},configconfig,)# 第二轮对话同一个thread_id能访问之前的状态resultapp.invoke({messages:[HumanMessage(content我叫什么名字)]},configconfig,)# Agent能回答张三因为状态被持久化了用了持久化以后即使程序重启状态也不会丢。下次用同一个thread_id还能接着之前的对话继续。状态的查看和调试开发的时候经常需要查看中间状态。用stream方法可以看到每一步的状态变化。foreventinapp.stream({messages:[HumanMessage(content什么是RAG)],topic_count:0},configconfig,):fornode_name,state_updateinevent.items():print(f节点{node_name}执行完毕)print(f状态更新:{state_update})print()也可以用get_state方法查看当前状态。# 获取当前状态current_stateapp.get_state(config)print(current_state.values)调试的时候把状态打印出来看比猜来猜去高效得多。状态设计的几个建议状态不要太大。状态在节点之间传递太大会影响性能。特别是消息列表聊久了会很长。定期做摘要压缩。状态字段要有明确的语义。别用data1、data2这种名字。用search_results、key_findings这种一看就懂的名字。控制流字段和业务字段分开。status、retry_count这些是控制流用的。search_results、summary这些是业务数据。分开放逻辑更清晰。给状态加默认值。新建状态时给所有字段初始值。避免节点读到None报错。下一篇讲条件分支与循环。让Agent学会思考和重试处理更复杂的任务流程。
