1. 从“健忘”到“有记忆”为什么AI应用需要会话记忆如果你用过早期的聊天机器人或者一些基础的大模型API肯定有过这样的体验你问它“我昨天提到的那个项目进展如何”它大概率会一脸茫然地反问你“什么项目”。这种对话就像和一个金鱼聊天它只有7秒的记忆每次交流都是全新的开始。这极大地限制了AI应用的实用性和用户体验尤其是在构建需要多轮交互、状态保持的智能体或复杂工作流时。这就是“会话记忆”要解决的核心痛点。它不是一个简单的“记住所有对话”的功能而是一个结构化的状态管理机制。想象一下你正在和一个人类专家协作完成一个项目。专家不仅记得你们之前讨论过的所有细节项目目标、技术选型、遇到的坑还能根据当前的对话上下文动态地调整他的建议和行动。这种“记忆”是连贯的、有状态的并且直接影响着后续的决策。在AI应用开发中LangChain等框架提供了基础的记忆模块比如ConversationBufferMemory它能简单地将历史对话拼接起来。但这种方式在面对复杂、分支、循环的工作流时就显得力不从心了。它更像一个被动的记录员而不是一个主动的状态管理者。而LangGraph的出现正是为了解决这种复杂状态的管理问题。它不是一个记忆模块的替代品而是一个更高维度的框架将“记忆”抽象为可编程、可流转、可持久化的状态State。在LangGraph中整个对话的上下文、中间结果、用户偏好、乃至智能体的内部决策逻辑都可以被清晰地定义在一个状态对象里并随着图的节点Node和边Edge的流转而更新。所以当我们说“使用LangGraph实现会话记忆功能”时我们实际上是在学习如何用工程化的、图计算的思想来设计和维护一个AI应用的核心“大脑”。这不仅仅是记住几句话而是构建一个能够理解上下文、管理任务进度、并做出连贯响应的智能系统的基石。对于想从“调用API”进阶到“开发复杂AI应用”的开发者来说掌握LangGraph的状态管理是至关重要的一步。2. LangGraph状态管理超越简单的聊天记录在深入代码之前我们必须先理解LangGraph设计哲学中关于“状态”的核心概念。这能帮你从根本上明白为什么用LangGraph做记忆管理比传统方式更强大。2.1 State一个可扩展的共享数据字典LangGraph中的State不是一个神秘的黑盒它本质上是一个Python字典dict或Pydantic模型。这个字典在整个图Graph的执行过程中是共享的。图中的每个节点Node即一个函数都可以读取和修改这个字典里的值。举个例子假设我们在开发一个旅行规划智能体。我们的State可能会被设计成这样from typing import TypedDict, List, Annotated from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史LangGraph内置的“消息”处理机制 messages: Annotated[List, add_messages] # 用户明确的旅行目的地 destination: str # 用户偏好的出行日期列表 preferred_dates: List[str] # 智能体已查询到的航班信息列表 flights_found: List[dict] # 当前对话轮次用于控制流程 conversation_round: int # 用户预算 budget: float关键点解析Annotated与add_messages这是LangGraph处理对话记忆的“语法糖”。add_messages是一个归约器Reducer。它的作用是定义当新消息产生时如何更新messages这个字段。add_messages的规则是“将新消息追加到列表末尾”。这省去了我们手动append的操作让消息管理变得声明式且线程安全。状态即上下文在这个State里messages字段存储了原始的对话历史而destination、preferred_dates等字段则是从历史中提炼出来的结构化信息。节点函数可以通过分析messages来更新这些结构化字段后续节点则可以直接使用这些提炼后的信息无需再次分析冗长的历史记录。这大大提升了效率。工作流状态conversation_round这样的字段标志着对话所处的阶段。它控制着图的流转逻辑例如第1轮问目的地第2轮问时间第3轮查询航班。2.2 Node与Edge状态流转的驱动者有了State谁来修改它答案是节点Node。每个Node都是一个普通的Python函数或可调用对象它接收一个State字典作为参数并返回一个包含要更新字段的字典。def collect_destination(state: State) - dict: 节点收集用户旅行目的地 # 1. 从最新的用户消息中提取目的地这里简化处理 latest_message state[“messages”][-1] user_input latest_message.content # 假设我们通过一个简单的LLM调用或规则提取目的地 extracted_destination extract_destination_from_text(user_input) # 2. 返回要更新的状态 return {“destination”: extracted_destination, “conversation_round”: 1}边Edge则决定了执行完当前Node后下一步该去哪个Node。边的条件可以基于State的值来判断。from langgraph.graph import END, START from langgraph.graph import StateGraph # 创建图 workflow StateGraph(State) # 添加节点 workflow.add_node(“collect_dest”, collect_destination) workflow.add_node(“collect_dates”, collect_dates) workflow.add_node(“search_flights”, search_flights) # 设置边和路由逻辑 workflow.add_edge(START, “collect_dest”) # 从开始到收集目的地 def route_after_dest(state: State): # 根据是否成功收集到目的地决定下一步 if state.get(“destination”): return “collect_dates” # 有目的地了去收集日期 else: return “collect_dest” # 没收集到继续收集 workflow.add_conditional_edges(“collect_dest”, route_after_dest) workflow.add_edge(“collect_dates”, “search_flights”) workflow.add_edge(“search_flights”, END) # 编译图 app workflow.compile()在这个流程中State就像一份共享的工作清单在每个节点间传递和更新。collect_dest节点在清单上写下了destinationroute_after_dest这个边条件检查了清单然后决定将清单传给collect_dates节点。整个会话的记忆和进度完全由这个State字典和图的结构来定义和维持。2.3 与LangChain Memory的本质区别很多初学者会混淆LangGraph的State和LangChain的Memory。它们有关联但层级不同LangChain Memory主要是一个存储和加载对话历史的抽象。它关心“如何把过去的对话存下来并在下次对话时读出来”。它的核心接口是load_memory_variables和save_context。它更偏向于数据持久化层。LangGraph State是一个应用运行时的工作状态管理器。它当然可以包含对话历史通过messages字段但它更核心的职责是管理应用在完成一个复杂任务过程中的所有中间状态和上下文。它定义了状态的结构、更新规则和流转逻辑。它处于业务逻辑层。可以说在LangGraph构建的应用中你可以使用LangChain的Memory作为持久化State尤其是messages字段到数据库或文件的一种后端实现。但LangGraph State的概念远比Memory宽泛和强大。3. 实战构建一个带记忆的旅行规划智能体理论说得再多不如一行代码。让我们动手实现一个简化但完整的旅行规划智能体它将展示如何利用LangGraph State来管理多轮对话记忆。3.1 环境准备与State定义首先安装必要库并定义我们的State。pip install langgraph langchain-openai# app.py from typing import TypedDict, List, Annotated, Optional from langgraph.graph import StateGraph, START, END from langgraph.graph.message import add_messages from langchain_openai import ChatOpenAI import operator import json # 1. 定义State class TravelAgentState(TypedDict): 旅行智能体的状态定义 messages: Annotated[List, add_messages] # 对话记忆 destination: Optional[str] # 结构化信息目的地 travel_dates: Optional[List[str]] # 结构化信息旅行日期 budget: Optional[float] # 结构化信息预算 search_results: Optional[List[dict]] # 中间结果查询到的航班/酒店 confirmed_booking: Optional[dict] # 最终结果确认的预订 step: str # 控制状态当前步骤 # 2. 初始化LLM llm ChatOpenAI(model“gpt-4o-mini”, temperature0)这里我们定义了一个相对丰富的State。step字段将用来控制我们的对话流程它是一个字符串比如“collect_destination”,“collect_dates”,“search”,“confirm”。3.2 实现核心节点函数节点是智能体的“器官”每个负责一项具体工作。# 3. 节点函数实现 def node_collect_destination(state: TravelAgentState) - dict: 节点从对话历史中提取并确认目的地 print(f“[节点收集目的地] 当前状态步进: {state.get(‘step’)}“) # 获取最新的用户消息 last_message state[“messages”][-1] user_text last_message.content if hasattr(last_message, ‘content’) else str(last_message) # 使用LLM进行信息提取结构化输出 prompt f””” 你是一个旅行助手。请从用户的输入中提取旅行目的地。 用户输入{user_text} 只返回目的地的城市或国家名称如果无法确定返回‘UNKNOWN’。 不要返回任何其他文字。 “”” response llm.invoke(prompt) extracted_dest response.content.strip() # 构建回复并准备更新状态 if extracted_dest and extracted_dest ! “UNKNOWN”: reply f”好的已记录您的目的地是{extracted_dest}。请问您的出行日期是例如2024-10-01 到 2024-10-07” next_step “collect_dates” else: reply “抱歉我没有听清您的目的地。能再告诉我一次您想去哪里旅行吗” next_step “collect_destination” # 保持在本步骤 # 返回要更新的状态部分 update { “destination”: extracted_dest if extracted_dest ! “UNKNOWN” else None, “step”: next_step } # 注意messages字段会由框架通过add_messages归约器自动更新 # 我们只需要在返回的dict中包含AI的回复消息。 # 但为了清晰我们在这里不直接操作messages而是在主流程中通过invoke传入。 # 本节点只负责更新结构化状态。 return update def node_collect_dates(state: TravelAgentState) - dict: 节点收集出行日期 print(f”[节点收集日期] 目的地: {state.get(‘destination’)}“) last_message state[“messages”][-1] user_text last_message.content if hasattr(last_message, ‘content’) else str(last_message) prompt f””” 从用户输入中提取旅行日期范围。用户输入{user_text} 请将日期范围格式化为一个包含两个字符串的列表例如 [‘2024-10-01’ ‘2024-10-07’]。 如果无法提取出两个有效日期返回空列表 []。 只返回JSON格式的列表不要有其他内容。 “”” response llm.invoke(prompt) try: dates json.loads(response.content) if isinstance(dates, list) and len(dates) 2: reply f”已记录出行时间从{dates[0]}到{dates[1]}。接下来您的预算是多少请输入数字” next_step “collect_budget” else: reply “日期格式不太对请重新输入例如‘2024年10月1号到10月7号’。” next_step “collect_dates” dates [] except json.JSONDecodeError: reply “日期解析失败请重新输入。” next_step “collect_dates” dates [] return {“travel_dates”: dates, “step”: next_step} def node_collect_budget(state: TravelAgentState) - dict: 节点收集预算 print(f”[节点收集预算] 日期: {state.get(‘travel_dates’)}“) last_message state[“messages”][-1] user_text last_message.content # 简单提取数字 import re match re.search(r’\d’, user_text) if match: budget float(match.group()) reply f”预算{budget}元已记录。正在为您模拟查询航班和酒店...此为演示无真实查询” next_step “search_and_summarize” else: reply “请输入一个明确的预算数字。” next_step “collect_budget” budget None return {“budget”: budget, “step”: next_step} def node_search_and_summarize(state: TravelAgentState) - dict: 节点模拟搜索并生成总结 print(f”[节点搜索总结] 预算: {state.get(‘budget’)}“) # 模拟搜索结果 mock_flights [{“airline”: “模拟航空”, “price”: state.get(“budget”, 5000) * 0.6}] mock_hotels [{“name”: “模拟酒店”, “price_per_night”: 800}] summary f””” 【旅行规划总结】 目的地{state[‘destination’]} 出行日期{state[‘travel_dates’][0]} 至 {state[‘travel_dates’][1]} 总预算{state[‘budget’]} 元 为您找到以下推荐 航班{mock_flights[0][‘airline’]} 预估价格 {mock_flights[0][‘price’]} 元。 酒店{mock_hotels[0][‘name’]} 每晚约 {mock_hotels[0][‘price_per_night’]} 元。 请确认是否按此规划进行(回复‘是’或‘否’) “”” return { “search_results”: {“flights”: mock_flights, “hotels”: mock_hotels}, “step”: “wait_confirmation” } def node_handle_confirmation(state: TravelAgentState) - dict: 节点处理用户确认 last_message state[“messages”][-1] user_text last_message.content if “是” in user_text or “yes” in user_text.lower(): reply “太好了您的旅行规划已确认。祝您旅途愉快\n演示结束” next_step END # 指向结束 booking {“status”: “confirmed”, “plan”: state.get(“search_results”)} else: reply “好的已取消本次规划。我们可以重新开始。” next_step START # 指向开始重置流程 booking {“status”: “cancelled”} # 注意返回START并不会自动清空State如果需要重置可以在这里返回一个重置的状态字典。 # 更常见的做法是进入一个‘reset’节点来清理状态。 return {“confirmed_booking”: booking, “step”: next_step}3.3 构建图与路由逻辑现在我们将这些节点组装起来并定义它们之间的流转关系。# 4. 构建状态图 workflow StateGraph(TravelAgentState) # 添加所有节点 workflow.add_node(“collect_destination”, node_collect_destination) workflow.add_node(“collect_dates”, node_collect_dates) workflow.add_node(“collect_budget”, node_collect_budget) workflow.add_node(“search_and_summarize”, node_search_and_summarize) workflow.add_node(“handle_confirmation”, node_handle_confirmation) # 设置路由逻辑 def router(state: TravelAgentState): 核心路由函数根据state.step决定下一个节点 current_step state.get(“step”, “collect_destination”) print(f”[路由决策] 当前步骤: {current_step}“) return current_step # LangGraph会将此返回值映射到同名的节点 # 设置入口和条件边 workflow.add_edge(START, “collect_destination”) # 初始入口 workflow.add_conditional_edges( “collect_destination”, router, { # 将router返回的字符串映射到具体的节点 “collect_destination”: “collect_destination”, # 继续收集目的地 “collect_dates”: “collect_dates”, # 其他步骤... } ) workflow.add_conditional_edges(“collect_dates”, router, {“collect_dates”: “collect_dates”, “collect_budget”: “collect_budget”}) workflow.add_conditional_edges(“collect_budget”, router, {“collect_budget”: “collect_budget”, “search_and_summarize”: “search_and_summarize”}) workflow.add_conditional_edges(“search_and_summarize”, router, {“wait_confirmation”: “handle_confirmation”}) workflow.add_conditional_edges(“handle_confirmation”, router, {END: END, START: “collect_destination”}) # 处理结束或重启 # 编译应用 app workflow.compile()3.4 运行与调试观察记忆的流转让我们运行这个智能体并观察State是如何在对话中演变的。# 5. 运行智能体 from langchain_core.messages import HumanMessage, AIMessage def run_conversation(): # 初始化状态 initial_state { “messages”: [HumanMessage(content“我想去上海旅游。”)], “destination”: None, “travel_dates”: None, “budget”: None, “search_results”: None, “confirmed_booking”: None, “step”: “collect_destination” # 初始步骤 } print(“ 对话开始 “) for i in range(10): # 防止无限循环设置上限 # 执行图 result app.invoke(initial_state) print(f”\n[第{i1}轮调用后]“) print(f”State.step: {result[‘step’]}“) print(f”State.destination: {result[‘destination’]}“) print(f”State.budget: {result[‘budget’]}“) print(f”最新AI回复: {result[‘messages’][-1].content if result[‘messages’] else ‘None’}“) # 检查是否结束 if result[‘step’] END: print(“\n 对话正常结束 ) break # 模拟用户输入在实际应用中这里替换为真实的用户输入 if “出行日期” in result[‘messages’][-1].content: user_input “下个月1号到5号” elif “预算” in result[‘messages’][-1].content: user_input “8000” elif “请确认” in result[‘messages’][-1].content: user_input “是” else: user_input “我不知道” # 兜底 print(f”[模拟用户输入]: {user_input}“) # 将用户输入作为新消息添加到状态中驱动下一轮 initial_state result initial_state[“messages”].append(HumanMessage(contentuser_input)) if __name__ “__main__”: run_conversation()运行这段代码你将在控制台看到一个完整的、有状态的对话流程。State中的destination、travel_dates、budget等字段被逐一填充step字段引导着流程前进。这就是LangGraph管理的“会话记忆”——它不是一堆杂乱的聊天记录而是一个结构化的、驱动应用逻辑的工作上下文。4. 进阶记忆的持久化、管理与优化一个生产级的应用不能只把记忆放在内存里。我们需要考虑持久化、长期记忆、以及如何优化记忆的效率和成本。4.1 状态持久化让记忆跨越会话LangGraph的State本身是内存中的对象。要实现跨会话的记忆我们需要将其保存到数据库。常见的做法是使用检查点Checkpointing。# persistence.py from langgraph.checkpoint import MemorySaver from langgraph.checkpoint.sqlite import SqliteSaver import sqlite3 # 方法1使用内存检查点仅用于开发/演示 memory_checkpointer MemorySaver() app_with_memory workflow.compile(checkpointermemory_checkpointer) # 为每个会话线程分配一个唯一的ID thread_id “user_123_session_1” config {“configurable”: {“thread_id”: thread_id}} # 第一次调用会创建检查点 initial_state {“messages”: [HumanMessage(content“Hi”)], …} result1 app_with_memory.invoke(initial_state, configconfig) # 第二次调用可以从上次中断的地方继续即使程序重启如果使用持久化存储 # 我们不需要传入完整的initial_state检查点会恢复之前的状态 user_input HumanMessage(content“My destination is Paris.”) result2 app_with_memory.invoke({“messages”: [user_input]}, configconfig) # 此时result2的状态包含了result1的历史消息和状态。 # 方法2使用SQLite持久化生产环境推荐 conn sqlite3.connect(“checkpoints.db”) sqlite_checkpointer SqliteSaver(conn) app_with_sqlite workflow.compile(checkpointersqlite_checkpointer)关键点检查点机制不仅保存了messages还保存了整个State字典的快照。这意味着用户的旅行目的地、预算、当前步骤等所有结构化信息在用户下次回来时都能完美恢复。这是实现“长期记忆”的基础。4.2 记忆窗口与摘要应对上下文长度限制大模型有上下文窗口限制如128K。如果对话历史messages无限增长最终会超出限制。解决方案是记忆窗口和摘要。滑动窗口只保留最近N轮对话。在LangGraph中可以通过自定义messages字段的归约器来实现。add_messages是追加我们可以写一个keep_last_n的归约器。from typing import Sequence from langchain_core.messages import BaseMessage def keep_last_n(n: int): 一个只保留最后n条消息的归约器 def reducer(old_messages: Sequence[BaseMessage], new_messages: Sequence[BaseMessage]): all_messages list(old_messages) list(new_messages) return all_messages[-n:] # 只返回最后n条 return reducer class StateWithWindow(TypedDict): messages: Annotated[List[BaseMessage], keep_last_n(10)] # 只保留10条 # ... 其他字段对话摘要更高级的做法是定期或当消息积累到一定数量时触发一个“摘要节点”。这个节点调用LLM将冗长的对话历史总结成一段精炼的文字然后替换或补充到messages中同时将关键信息提取到结构化字段如destination。这能极大地节省token并保留核心信息。def node_summarize_conversation(state: StateWithWindow) - dict: 摘要节点 long_history state[“messages”] if len(long_history) 20: # 未达到摘要阈值 return {} prompt f”请将以下对话总结成一段简洁的摘要保留关键决策和事实\n{long_history}” summary llm.invoke(prompt).content # 用摘要替换旧的历史或者作为一条系统消息插入 new_messages [SystemMessage(contentf”对话摘要{summary}”)] long_history[-5:] # 保留最近5条原始消息 return {“messages”: new_messages}4.3 结构化记忆与向量检索实现“长期记忆”对于需要记住大量历史信息如过去一年的所有旅行咨询的应用仅靠上下文窗口和摘要是不够的。这时需要引入向量数据库。记忆写入在对话过程中可以将重要的用户信息如“用户喜欢靠窗的座位”、“用户对花生过敏”转换成向量存入向量数据库如Chroma, Pinecone并与用户ID关联。记忆读取当新对话开始时或对话中提到相关话题时如用户说“我还是想要上次那种座位”可以从向量数据库中检索出相关的历史记忆片段作为上下文插入到本次对话的State中。这相当于为智能体配备了一个外部记忆库。在LangGraph中可以设计专门的retrieve_memories节点和update_memories节点来管理这个过程。# 伪代码示例 def node_retrieve_relevant_memories(state: State) - dict: query state[“messages”][-1].content # 最新用户问题 # 从向量库检索相关记忆 relevant_memories vector_db.similarity_search(query, filter{“user_id”: state[“user_id”]}) # 将检索到的记忆作为上下文添加到prompt或单独的消息中 memory_context “\n”.join([mem.page_content for mem in relevant_memories[:3]]) return {“retrieved_memories”: memory_context} # 在调用LLM的节点中将retrieved_memories加入到提示词中 prompt f””” 相关历史信息 {state[‘retrieved_memories’]} 当前对话 {state[‘messages’][-3:]} # 最近几条对话 请根据以上信息回答。 “””5. 避坑指南与性能调优在实际开发中你会遇到一些预料之外的问题。以下是我从项目中总结的几个关键点。5.1 状态更新冲突与归约器选择问题当多个节点并发或在复杂循环中修改State的同一个字段时可能会发生更新冲突。例如一个节点想清空search_results另一个节点想往里添加新数据。解决方案深入理解并正确使用归约器Reducer。add_messages是LangGraph提供的用于列表的归约器。对于其他类型的字段你需要根据业务逻辑定义自己的归约器。operator.setitem 直接设置值后执行的覆盖先执行的。operator.add 对于数字字段进行相加。自定义归约器对于复杂操作如“合并两个字典”、“列表去重后追加”需要自己写函数。def merge_dict_reducer(old: dict, new: dict): return {**old, **new} # 新字典覆盖旧字典的相同键 class MyState(TypedDict): config: Annotated[dict, merge_dict_reducer] count: Annotated[int, operator.add]注意在设计State时尽量让每个节点更新不同的字段减少冲突。如果必须更新同一字段务必想清楚归约逻辑。5.2 图的可视化与调试问题当图变得复杂有多个条件分支和循环时逻辑流难以跟踪。解决方案利用LangGraph的可视化工具。# 生成图的Mermaid格式定义注意输出的是文本需在支持Mermaid的编辑器中渲染 graph_definition app.get_graph().draw_mermaid() print(graph_definition) # 或者直接保存为图片需要安装pygraphviz可能比较麻烦 # app.get_graph().draw(“my_graph.png”, format“png”)更实用的调试方法是使用LangSmith。将你的应用与LangSmith集成后可以清晰地追踪每一次invoke的完整流程每个节点的输入State、输出State、耗时、LLM调用详情等。这对于排查状态流转错误和性能瓶颈至关重要。5.3 性能优化减少不必要的LLM调用问题在router函数或某些节点中频繁调用LLM来判断下一步或提取信息成本高且慢。优化策略规则优先能用简单规则如关键字匹配、正则表达式判断的就不要用LLM。例如判断用户是否确认可以用if “是” in user_input or “ok” in user_input.lower()。状态标志位在State中设置明确的标志位如step,needs_clarification让路由逻辑基于这些结构化字段进行而不是每次都分析整个对话历史。批量处理如果一个节点需要根据多条历史消息做决策尽量一次性将这些消息组织好发给LLM而不是进行多次串行调用。缓存对于纯信息提取且输入相同的节点例如从固定格式的文本中提取日期可以考虑使用简单的缓存如functools.lru_cache来避免重复调用LLM。5.4 处理用户中断与流程重置问题用户可能在流程中途说“算了重新开始吧”或问一个完全不相关的问题如何优雅处理解决方案在图中设计一个interrupt_handler节点和对应的边。在每个主要节点后或者通过一个全局的“监听”机制检查用户输入是否包含中断指令如“/reset”, “重新开始”。如果检测到中断则路由到interrupt_handler节点。这个节点可以清除当前State中的关键字段如destination,step等。将step设置为START。返回一条友好的提示信息如“好的我们重新开始。您想去哪里旅行”在router函数中对interrupt_handler节点的输出进行特殊处理使其能跳回流程起点。这确保了智能体不会被卡在一个无效的状态中始终保持灵活和健壮。掌握LangGraph的状态管理你就掌握了构建有记忆、有逻辑的复杂AI应用的钥匙。它迫使你以“状态流”的思维来设计应用这种思维对于开发任何复杂的、交互式的软件系统都是极其宝贵的。从简单的对话记忆到复杂的工作流状态机LangGraph提供了一套统一而强大的范式。