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學會思考和重試處理更復雜的任務流程。