对话一多就失忆？用LangGraph打造有状态Agent，状态持久化与人机协作全搞定

🎁个人主页：我滴老baby
🎉欢迎大家点赞👍评论📝收藏⭐文章
🔍系列专栏：AI

对话一多就失忆？用LangGraph打造有状态Agent，状态持久化与人机协作全搞定

普通聊天机器人对话一多就失忆，有状态Agent却能记住所有关键信息。本文深入LangGraph的状态管理机制，教你构建真正"有记忆"的对话系统。

一、什么是有状态对话Agent？

在之前的文章中，我们构建的Agent大多是"无状态"的——每次对话都是独立的，Agent不会记住之前说过什么。但在实际业务场景中，用户期望的是像跟真人对话一样流畅的体验：Agent需要记住用户的名字、之前的讨论话题、已经完成的任务等等。这就是"有状态"对话Agent要解决的核心问题。

1.1 无状态 vs 有状态

下面这张表从多个维度对比了无状态对话和有状态对话的差异。可以看到，有状态对话在用户体验、复杂任务处理等方面有着显著优势，但实现复杂度也相应更高。选择哪种方式取决于你的具体业务需求——简单的FAQ场景用无状态就够了，但涉及多轮协作的复杂业务流程就必须用有状态方案：

1.2 LangGraph状态管理核心

LangGraph的状态管理遵循一个简洁而强大的模型：用户输入触发状态对象的更新，各个节点（Node）读取和修改状态，同时通过Checkpoint机制将状态持久化到存储中。这样即使对话中断，也可以从上次的状态恢复继续：

用户输入 → State(状态对象) → Node1(更新状态) → Node2(读取状态) → 输出 ↓ Checkpoint(持久化)

二、LangGraph状态定义

状态定义是有状态Agent设计的核心环节。一个好的状态模型需要包含对话所需的所有上下文信息，同时又不能过于臃肿导致性能问题。LangGraph使用Python的TypedDict来定义状态，配合Annotated类型注解实现特殊的合并策略。

2.1 基础状态模型

下面这段代码定义了一个通用的对话状态模型。它包含了五个维度的信息：用户画像、对话上下文、任务状态、会话元数据和工具相关数据。特别值得注意的是messages字段使用了Annotated[list, operator.add]，这意味着每个节点返回的新消息会自动追加而不是覆盖——这是LangGraph实现消息累积的核心机制：

# stateful_agent/state.pyfromtypingimportTypedDict,Annotated,List,Optional,Literalfromdataclassesimportdataclassimportoperatorfromdatetimeimportdatetime@dataclassclassUserProfile:"""用户画像"""name:str=""preferences:dict=Noneexpertise_level:str="beginner"# beginner/intermediate/expertclassConversationState(TypedDict):"""对话状态"""# 用户信息user_id:struser_profile:UserProfile# 对话上下文messages:Annotated[list,operator.add]current_topic:strmentioned_entities:List[str]# 任务状态active_tasks:List[dict]completed_tasks:List[dict]pending_questions:List[str]# 会话元数据session_start:strturn_count:intlast_intent:str# 工具相关tool_results:Annotated[list,operator.add]artifacts:List[dict]# 生成的文件、图表等

2.2 高级状态：条件分支

对于更复杂的业务场景（比如客服系统），状态模型需要支持阶段流转和条件分支。下面这段代码定义了一个客服对话状态，包含了对话阶段（phase）、问题类型、紧急程度、处理状态以及人工介入标志等字段。phase字段使用Literal类型限制了合法的取值范围，确保对话阶段只能在预定义的几个状态之间流转：

# stateful_agent/advanced_state.pyfromtypingimportTypedDict,Annotated,LiteralclassSupportState(TypedDict):"""客服对话状态"""messages:Annotated[list,operator.add]# 对话阶段phase:Literal["greeting","understanding","resolving","confirming","closing"]# 用户问题issue_type:str# refund/billing/technical/complaintissue_description:strurgency:Literal["low","medium","high"]# 处理状态attempted_solutions:List[str]resolution:strsatisfaction_score:Optional[int]# 1-5# 人工介入escalated:boolhuman_agent_id:Optional[str]

三、完整实现

3.1 有状态对话Agent

有了状态定义之后，我们来构建完整的有状态对话Agent。这个Agent包含四个核心节点：greeting（问候）、understanding（理解意图）、response（生成回复），以及基于意图的条件路由。关键的设计点在于route_after_understanding函数——它根据LLM分析出的用户意图，将对话路由到不同的处理节点，实现了智能化的分支控制：

# stateful_agent/agent.pyfromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpoint.sqliteimportSqliteSaverfromlangchain_openaiimportChatOpenAIfromstateful_agent.stateimportConversationStateimportjson llm=ChatOpenAI(model="gpt-4o",temperature=0.7)defgreeting_node(state:ConversationState)->dict:"""问候节点"""ifstate["turn_count"]==0:response=f"你好！我是你的AI助手。请问有什么可以帮你的？"else:name=state["user_profile"].nameifname:response=f"欢迎回来{name}！我们继续上次的对话。"else:response="你好！请问有什么可以帮你的？"return{"messages":[{"role":"assistant","content":response}],"phase":"understanding","turn_count":state["turn_count"]+1}defunderstanding_node(state:ConversationState)->dict:"""理解用户意图"""last_user_msg=""formsginreversed(state["messages"]):ifmsg.get("role")=="user":last_user_msg=msg["content"]breakprompt=f"""分析用户输入，提取关键信息： 用户输入:{last_user_msg}已知上下文: - 当前话题:{state.get('current_topic','无')}- 已提及实体:{state.get('mentioned_entities',[])}- 用户等级:{state.get('user_profile',UserProfile()).expertise_level}返回JSON: {{ "intent": "意图", "topic": "话题", "entities": ["实体列表"], "needs_tool": true/false, "tool_name": "工具名(如需要)" }}"""result=llm.invoke(prompt)try:analysis=json.loads(result.content)except:analysis={"intent":"general","topic":"unknown","entities":[],"needs_tool":False}new_entities=list(set(state.get("mentioned_entities",[])+analysis.get("entities",[])))return{"current_topic":analysis.get("topic",state.get("current_topic","")),"mentioned_entities":new_entities,"last_intent":analysis.get("intent","general"),"messages":[]}defresponse_node(state:ConversationState)->dict:"""生成回复"""context_summary=f"""当前状态: - 话题:{state.get('current_topic')}- 意图:{state.get('last_intent')}- 用户等级:{state.get('user_profile',UserProfile()).expertise_level}- 已提及:{state.get('mentioned_entities',[])}- 回合数:{state.get('turn_count',0)}"""messages=[{"role":"system","content":f"你是一个有帮助的AI助手。\n{context_summary}"},*state["messages"][-10:]]response=llm.invoke(messages)return{"messages":[{"role":"assistant","content":response.content}],"turn_count":state.get("turn_count",0)+1}defroute_after_understanding(state:ConversationState)->str:"""根据理解结果路由"""intent=state.get("last_intent","general")ifintent=="task_request":return"task_handler"elifintent=="question":return"qa_handler"else:return"response"# 构建图workflow=StateGraph(ConversationState)workflow.add_node("greeting",greeting_node)workflow.add_node("understanding",understanding_node)workflow.add_node("response",response_node)workflow.add_node("task_handler",lambdas:{"messages":[{"role":"assistant","content":"我来帮你处理这个任务..."}]})workflow.add_node("qa_handler",lambdas:{"messages":[{"role":"assistant","content":"让我来回答你的问题..."}]})workflow.set_entry_point("greeting")workflow.add_edge("greeting","understanding")workflow.add_conditional_edges("understanding",route_after_understanding,{"task_handler":"task_handler","qa_handler":"qa_handler","response":"response"})workflow.add_edge("task_handler",END)workflow.add_edge("qa_handler",END)workflow.add_edge("response",END)# 启用检查点持久化memory=SqliteSaver.from_conn_string("./stateful_conversation.db")app=workflow.compile(checkpointer=memory)

3.2 多轮对话运行

状态持久化最直接的好处就是支持多轮连贯对话。通过指定thread_id，LangGraph会自动将同一会话的状态持久化到数据库中，每次调用时从上次的状态继续。下面这段代码展示了三轮对话的效果：Agent在第二和第三轮中能够记住之前讨论的话题，并基于上下文做出连贯的回答：

# stateful_agent/run.pydefchat(user_input:str,thread_id:str="user-001"):config={"configurable":{"thread_id":thread_id}}result=app.invoke({"messages":[{"role":"user","content":user_input}],"turn_count":0,"user_profile":UserProfile(),"mentioned_entities":[],"active_tasks":[],"completed_tasks":[],"pending_questions":[],"session_start":datetime.now().isoformat(),"tool_results":[],"artifacts":[]},config=config)# 获取最后一条助手消息formsginreversed(result["messages"]):ifmsg.get("role")=="assistant":returnmsg["content"]return""# 多轮对话print(chat("我想学习Python的装饰器","thread-1"))# → 我来帮你学习Python装饰器...print(chat("能给我一个实际的例子吗？","thread-1"))# → Agent记住了之前的话题是"装饰器"，给出相关示例print(chat("这和我之前问的生成器有什么关系？","thread-1"))# → Agent知道用户之前问过装饰器，对比装饰器和生成器

四、状态持久化方案对比

选择合适的持久化方案是Agent生产化的重要决策。不同的方案在性能、持久性和适用场景上各有侧重。

下面这张表对比了五种主流的状态持久化方案。对于开发阶段，SQLite是最简单的选择——零配置、单文件、性能足够。对于生产环境，PostgreSQL和Redis是更稳妥的选择，它们支持高并发和集群部署：

状态管理最佳实践

在实际项目中，状态管理有一些经过验证的最佳实践值得遵循。下面这张表总结了五个关键实践，每一条都能帮你避免常见的状态管理陷阱：

五、进阶模式

5.1 人工接管节点

在很多业务场景中（比如金融交易、客服升级），Agent不能完全自主决策，需要在关键节点引入人工审批。LangGraph通过interrupt_before参数优雅地支持了这种"人工在环"（Human-in-the-Loop）模式——在指定节点之前自动暂停执行，等待人工确认后再继续：

# stateful_agent/human_in_loop.pyfromlanggraph.graphimportStateGraph,END workflow=StateGraph(ConversationState)# ... 添加其他节点 ...workflow.add_node("approval_gate",lambdas:{"messages":[{"role":"assistant","content":"此操作需要确认，等待人工审批..."}]})# 编译时设置中断点app=workflow.compile(checkpointer=memory,interrupt_before=["approval_gate"]# 在此节点前暂停)# 恢复执行defresume_after_approval(thread_id:str,approved:bool):config={"configurable":{"thread_id":thread_id}}state=app.get_state(config)state.values["approved"]=approved app.update_state(config,state.values)app.invoke(None,config)

5.2 状态快照与回滚

在生产环境中，状态快照和回滚能力至关重要。当Agent执行了错误的操作或者用户想要"撤销"上一步时，能够快速恢复到之前的状态是保障用户体验的关键。下面这段代码实现了一个简单的状态快照管理器，支持创建、列举和恢复状态快照：

# stateful_agent/snapshot.pyclassStateManager:"""状态快照管理"""def__init__(self,app,thread_id:str):self.app=app self.config={"configurable":{"thread_id":thread_id}}self.snapshots=[]defsave_snapshot(self,label:str):state=self.app.get_state(self.config)self.snapshots.append({"label":label,"state":state.values,"timestamp":datetime.now().isoformat()})defrestore_snapshot(self,label:str):forsnapinreversed(self.snapshots):ifsnap["label"]==label:self.app.update_state(self.config,snap["state"])returnTruereturnFalsedeflist_snapshots(self):return[{"label":s["label"],"time":s["timestamp"]}forsinself.snapshots]

六、状态设计模式对比

不同的业务场景适合不同的状态设计模式。下面这张表对比了五种常见模式的特点和适用场景。对于大多数项目来说，从"扁平状态"或"嵌套状态"开始是最务实的选择，随着业务复杂度的增长再逐步演进到更高级的模式：

总结

有状态对话是Agent从"问答工具"进化为"协作伙伴"的关键能力。通过本文的学习，你应该已经掌握了构建有状态Agent的核心技能，以下是关键要点回顾：

1. TypedDict定义状态——类型安全、结构清晰，配合Annotated注解实现灵活的合并策略。建议在项目初期就花时间设计好状态模型，因为后期的状态结构调整成本很高。

2. LangGraph状态图——灵活编排状态流转，支持条件分支和循环。通过add_conditional_edges实现智能路由，让Agent根据意图自动选择处理路径。

3. Checkpoint持久化——状态可恢复、可回滚，是Agent生产化的基础保障。根据项目规模选择合适的存储方案，开发用SQLite，生产用PostgreSQL或Redis。

4. 人工接管——关键节点引入人工审批，平衡自动化效率和安全性。LangGraph的interrupt_before机制让Human-in-the-Loop变得非常优雅。

5. 实战建议：建议从最简单的扁平状态开始，先实现基本的多轮记忆能力，再逐步添加意图识别、任务管理、快照回滚等高级功能。不要一开始就设计过于复杂的状态模型，先跑通核心流程再迭代优化。

下一篇预告：《Python实现智能体的规划与任务分解能力》