LangGraph狀態(tài)管理實戰(zhàn)多輪交互中維護Agent狀態(tài)上一篇講了LangGraph的核心概念。這一篇深入狀態(tài)管理。狀態(tài)管理是LangGraph最有價值的功能。Agent執(zhí)行復(fù)雜任務(wù)的時候中間會產(chǎn)生大量信息。搜索結(jié)果、中間推理、用戶偏好、任務(wù)進度。這些信息要在多個節(jié)點之間共享和更新。狀態(tài)管理做不好Agent就會健忘。做了前面忘了后面或者信息在不同步驟之間對不上。狀態(tài)的設(shè)計狀態(tài)設(shè)計是第一步。好的狀態(tài)設(shè)計能讓工作流清晰高效。設(shè)計狀態(tài)要考慮幾個問題。需要存什么信息。把任務(wù)執(zhí)行過程中需要共享的信息都列出來。對話歷史、搜索結(jié)果、中間結(jié)論、任務(wù)進度、用戶信息。信息之間有沒有層級。簡單的用扁平結(jié)構(gòu)就行。復(fù)雜的可以嵌套。哪些信息是只讀的哪些需要更新。只讀的不用管需要更新的要想好更新策略。fromtypingimportTypedDict,List,OptionalclassResearchState(TypedDict):# 用戶輸入topic:str# 研究主題user_level:str# 用戶水平初學(xué)者/中級/專家# 搜索相關(guān)search_queries:List[str]# 搜索關(guān)鍵詞列表search_results:List[str]# 搜索結(jié)果search_round:int# 搜索輪次# 分析相關(guān)key_findings:List[str]# 關(guān)鍵發(fā)現(xiàn)summary:str# 總結(jié)# 控制流status:str# 當(dāng)前狀態(tài)retry_count:int# 重試次數(shù)max_retries:int# 最大重試次數(shù)這個狀態(tài)設(shè)計覆蓋了一個研究Agent需要的所有信息。用戶輸入、搜索過程、分析結(jié)果、控制流分門別類。狀態(tài)的更新機制LangGraph的狀態(tài)更新機制是這樣的。每個節(jié)點返回一個字典只包含要更新的字段。LangGraph會把這個字典合并到當(dāng)前狀態(tài)里。defsearch_node(state:ResearchState)-dict:topicstate[topic]round_numstate.get(search_round,0)1# 執(zhí)行搜索new_resultf第{round_num}輪搜索{topic}的結(jié)果# 只返回要更新的字段return{search_results:state[search_results][new_result],search_round:round_num,status:searched,}對于列表類型的字段更新時要手動拼接舊值和新值。state[search_results] [new_result]把新結(jié)果追加到舊列表后面。這樣寫有點麻煩。LangGraph提供了Annotated類型可以自動處理列表的追加。fromtypingimportAnnotatedfromoperatorimportaddclassResearchState(TypedDict):search_results:Annotated[List[str],add]# 自動追加search_queries:Annotated[List[str],add]# 自動追加topic:strsummary:strstatus:strsearch_round:int用了Annotated和add以后節(jié)點返回時直接給新值就行LangGraph會自動追加。defsearch_node(state:ResearchState)-dict:new_resultf搜索{state[topic]}的結(jié)果# 直接給新值自動追加到現(xiàn)有列表return{search_results:[new_result],search_round:state.get(search_round,0)1,}消息列表的管理對話型Agent的狀態(tài)里最常見的是消息列表。用戶消息、AI消息、工具調(diào)用消息都要存。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]# 調(diào)用大模型fromlangchain_openaiimportChatOpenAI llmChatOpenAI(modelgpt-3.5-turbo,temperature0)responsellm.invoke(messages)# 返回AI的回復(fù)自動追加到消息列表return{messages:[response]}add_messages會自動處理消息的追加。如果同一條消息被重復(fù)添加它還會自動去重。比手動拼接方便多了。多輪對話的狀態(tài)管理多輪對話是狀態(tài)管理的典型場景。用戶跟Agent來回對話每輪對話都要參考之前的上下文。fromlanggraph.graphimportStateGraph,ENDfromtypingimportAnnotated,TypedDictfromlanggraph.graph.messageimportadd_messagesfromlangchain_core.messagesimportHumanMessage,AIMessage,SystemMessagefromlangchain_openaiimportChatOpenAIclassConversationState(TypedDict):messages:Annotated[list,add_messages]user_name:strtopic_count:int# 對話節(jié)點defchat_node(state:ConversationState)-dict:llmChatOpenAI(modelgpt-3.5-turbo,temperature0.7)messagesstate[messages]# 如果知道用戶名字加到系統(tǒng)提示里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,}# 判斷是否結(jié)束defshould_continue(state:ConversationState)-str:ifstate.get(topic_count,0)5:returnENDreturnchat# 構(gòu)建圖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有什么區(qū)別),],user_name:張三,topic_count:0,})# 查看完整的對話歷史formsginresult[messages]:role用戶ifisinstance(msg,HumanMessage)else助手print(f{role}:{msg.content[:50]}...)狀態(tài)持久化默認(rèn)情況下LangGraph的狀態(tài)存在內(nèi)存里。程序結(jié)束就沒了。要想跨會話保持狀態(tài)需要用持久化存儲。LangGraph支持用SQLite做持久化。fromlanggraph.checkpoint.sqliteimportSqliteSaverimportsqlite3# 創(chuàng)建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能訪問之前的狀態(tài)resultapp.invoke({messages:[HumanMessage(content我叫什么名字)]},configconfig,)# Agent能回答張三因為狀態(tài)被持久化了用了持久化以后即使程序重啟狀態(tài)也不會丟。下次用同一個thread_id還能接著之前的對話繼續(xù)。狀態(tài)的查看和調(diào)試開發(fā)的時候經(jīng)常需要查看中間狀態(tài)。用stream方法可以看到每一步的狀態(tài)變化。foreventinapp.stream({messages:[HumanMessage(content什么是RAG)],topic_count:0},configconfig,):fornode_name,state_updateinevent.items():print(f節(jié)點{node_name}執(zhí)行完畢)print(f狀態(tài)更新:{state_update})print()也可以用get_state方法查看當(dāng)前狀態(tài)。# 獲取當(dāng)前狀態(tài)current_stateapp.get_state(config)print(current_state.values)調(diào)試的時候把狀態(tài)打印出來看比猜來猜去高效得多。狀態(tài)設(shè)計的幾個建議狀態(tài)不要太大。狀態(tài)在節(jié)點之間傳遞太大會影響性能。特別是消息列表聊久了會很長。定期做摘要壓縮。狀態(tài)字段要有明確的語義。別用data1、data2這種名字。用search_results、key_findings這種一看就懂的名字。控制流字段和業(yè)務(wù)字段分開。status、retry_count這些是控制流用的。search_results、summary這些是業(yè)務(wù)數(shù)據(jù)。分開放邏輯更清晰。給狀態(tài)加默認(rèn)值。新建狀態(tài)時給所有字段初始值。避免節(jié)點讀到None報錯。下一篇講條件分支與循環(huán)。讓Agent學(xué)會思考和重試處理更復(fù)雜的任務(wù)流程。