【LangGraph 学习笔记】告别纯文本拼接：使用 MemorySaver 实现真正的有角色多轮记忆

一、什么是 Checkpointer在 LangGraph 中Checkpointer就像是游戏的“存档系统”。MemorySaver基于内存的存档。适合开发测试程序一关记忆就没了。PostgresSaver基于 PostgreSQL 数据库的存档。适合生产环境永久保存。当你在图中加入了 CheckpointerLangGraph 每执行完一个节点就会自动把当前的 State“拍个快照”存起来。下次你再调用图时它会自动读取上一次的“快照”作为起点。二、完整代码实战下面是通过MemorySaver实现多轮记忆的完整代码。注意观察代码中是如何通过config来区分不同用户的会话的。from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from langchain_openai import ChatOpenAI from dotenv import load_dotenv import os load_dotenv() # # 定义 State # class ChatState(TypedDict): # 使用 Annotated 实现消息追加 messages: Annotated[list[str], operator.add] # 初始化 LLM llm ChatOpenAI( modeldeepseek-chat, api_keyos.getenv(DEEPSEEK_API_KEY), base_urlos.getenv(DEEPSEEK_BASE_URL), temperature0 ) # # 定义节点 # def chatbot(state: ChatState) - dict: # 取出历史消息并拼接 history \n.join(state[messages]) # 调用 LLM 继续对话 response llm.invoke(fContinue the conversation:\n{history}) # 返回 AI 的回复会自动追加到 messages 中 return {messages: [response.content]} # # 构建图 # workflow StateGraph(ChatState) workflow.add_node(chat, chatbot) workflow.set_entry_point(chat) workflow.add_edge(chat, END) # # 知识点 1启用 Checkpointer # # 创建内存存档器 memory MemorySaver() # 编译图时将 checkpointer 传入 app workflow.compile(checkpointermemory) # # 知识点 2 3模拟多轮对话 # if __name__ __main__: # 定义配置字典通过 thread_id 区分不同的用户/会话 # 相当于这个用户的“存档档位名” config {configurable: {thread_id: user-abc}} print(--- 开始多轮对话 ---) # 第一轮传入配置 app.invoke({messages: [你好我叫小明]}, config) # 第二轮传入同样的 configLangGraph 会自动去 memory 里找上次的状态 app.invoke({messages: [我叫什么名字]}, config) # 第三轮 app.invoke({messages: [我喜欢贝多芬的音乐]}, config) # 第四轮 app.invoke({messages: [我喜欢什么]}, config) # # 知识点 4查看完整历史快照 # print(\n 会话历史快照记录 (最新在前):) print(- * 30) # get_state_history 会返回一个包含所有历史版本的生成器 for snapshot in app.get_state_history(config): if snapshot.values[messages]: # 打印每个快照中最后一条消息 print(f- {snapshot.values[messages][-1]}) print(- * 30)三、核心步骤拆解在上面的代码中实现持久化记忆只需要四步1. 创建并绑定 Checkpointer这是最关键的一步。实例化一个MemorySaver并在compile()的时候作为参数传进去。memory MemorySaver() app workflow.compile(checkpointermemory)2. 构建配置字典会话隔离通过config字典定义一个thread_id线程ID。config {configurable: {thread_id: user-abc}}它的作用是什么这是实现“多用户隔离”的关键。如果张三调用时传thread_id: zhangsan李四调用时传thread_id: lisiLangGraph 会在内存中保存两份完全独立的对话记录互不干扰。3. 运行时携带配置在调用invoke时除了传入新的输入状态必须把刚才定义的 config 作为第二个参数传进去。app.invoke({messages: [你好]}, config)如果不传 configLangGraph 就不知道该去哪个“档位”读取历史记录了。4. 回溯历史快照通过app.get_state_history(config)你可以拿到这个thread_id下的所有历史版本记录即每次节点执行后的 State 状态。这对于做“撤回”、“查看日志”非常有用。四、 避坑与进阶为什么叫“有角色”仔细看上面的代码虽然我们实现了多轮记忆但在 State 定义中我们用的依然是messages: list[str]纯字符串列表。真正的“有角色”在 LangChain 生态中是有专门的数据结构的。如果你希望 State 不仅能记住内容还能明确记住“这是 Human 说的那是 AI 说的”标准的做法是结合add_messages使用BaseMessage对象而不是纯字符串# 进阶写法了解即可后续详细讲 from langgraph.graph.message import add_messages from langchain_core.messages import HumanMessage, AIMessage class ChatState(TypedDict): # add_messages 专门用于处理带角色的 Message 对象 messages: Annotated[list, add_messages] # 节点中的返回方式也会变成 # return {messages: [AIMessage(content我是AI回复)]} # 调用时的传入方式会变成 # app.invoke({messages: [HumanMessage(content我是用户提问)]}, config)