一篇讲透 LangGraph 记忆管理：30 分钟让 AI Agent 从“金鱼脑”变成“记忆大师”

很多新手在开发 AI 应用时都会卡在一个地方我的 AI 怎么老是“失忆”明明上一句刚说过名字下一句就忘了。别急LangGraph 早就帮你把“记忆”这件事安排得明明白白。今天这篇干货我用最通俗的大白话 每一行都能直接跑的代码带你彻底搞懂短期记忆、长期记忆、消息修剪、对话总结。只要你跟着敲一遍代码以后任何 AI 记忆问题都难不倒你。一、AI 的“记忆”到底是什么一个生活例子你就懂了想象你开了一家奶茶店雇了一个机器人店员。短期记忆就是机器人和同一个顾客在当前这一轮对话里能记住顾客刚才说“我要一杯少冰的杨枝甘露”。只要对话不中断它不会反复问你“要冰的吗”长期记忆哪怕顾客过了一个星期再次进店机器人还记得“这位顾客上次说他喜欢少冰、多糖”。在 LangGraph 里短期记忆→ Checkpointer检查点按对话线程 ID保存状态。长期记忆→ Store存储按用户 ID或其他命名空间跨会话保存信息。下面我们一个知识点一个知识点地手写代码 二、短期记忆一用内存实现“对话存档”——10 行核心代码搞定先掌握最简单的 InMemorySaver你就能明白短期记忆的底层原理。2.1 代码前先讲人话不着急看代码你定义一个状态比如 messages 消息列表 user_name 用户名。你画一张图先问候节点 → 回复节点 → 结束。你编译图的时候塞给它一个checkpointer检查点工具同时给每个对话分配一个thread_id。之后每次调用 graph.invoke只要 thread_id 一样LangGraph 就会从数据库中把之前的状态读出来自动合并新消息。就像是打游戏时自动存档你每次玩到某个进度系统自动保存下次你只要说“还是那个存档”游戏就从那里继续。2.2 完整代码复制粘贴即可运行注释多到“发指”# ---------- 1. 导入所需的工具 ---------- from typing import Annotated from typing_extensions import TypedDict import operator # InMemorySaver 是短期记忆的“内存版”适合开发调试 from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import StateGraph, START, END # ---------- 2. 定义状态就是 AI 要记住的内容结构 ---------- class ChatState(TypedDict): # messages 用了 operator.add意思是每次新消息不是覆盖而是追加到列表后面 messages: Annotated[list, operator.add] user_name: str # 简单字段直接覆盖保存 # ---------- 3. 定义图里的节点每个节点就是一个执行步骤 ---------- def greeting_node(state: ChatState) - dict: 第一步打招呼 user_name state.get(user_name, 访客) print(f[问候节点] 当前用户{user_name}) # 返回一条助手消息会被追加到 messages 里 return {messages: [(assistant, f你好{user_name}我是你的 AI 助手。)]} def respond_node(state: ChatState) - dict: 第二步根据用户最新消息回应 # 从已有的消息中找出所有用户发的消息 user_msgs [msg for msg in state[messages] if msg[0] user] if user_msgs: last_msg user_msgs[-1][1] # 最近一条用户消息的内容 user_name state.get(user_name, 访客) if 名字 in last_msg or 我叫 in last_msg: reply f我知道你叫 {user_name}很高兴认识你 elif 天气 in last_msg: reply f抱歉 {user_name}我查不到实时天气。 else: reply f我明白了{user_name}。能多说一点吗 else: reply 我没有收到你的消息。 return {messages: [(assistant, reply)]} # ---------- 4. 构建图 ---------- builder StateGraph(ChatState) builder.add_node(greeting, greeting_node) builder.add_node(respond, respond_node) builder.add_edge(START, greeting) builder.add_edge(greeting, respond) builder.add_edge(respond, END) # ---------- 5. 关键编译时加入 checkpointer ---------- memory InMemorySaver() # 创建内存检查点 graph builder.compile(checkpointermemory) # ---------- 6. 配置线程 ID —— 这是短期记忆的“身份证” ---------- config {configurable: {thread_id: chat_user_xiao_ming}} # ---------- 7. 第一轮对话用户告诉 AI 名字 ---------- print( 第一轮对话 ) result1 graph.invoke( {messages: [(user, 你好我叫小明)], user_name: 小明}, config ) for role, msg in result1[messages]: print(f{role}: {msg}) print(\n 第二轮对话同一个线程AI 还记得名字) result2 graph.invoke( {messages: [(user, 你还记得我叫什么吗)], user_name: 小明}, config ) for role, msg in result2[messages]: print(f{role}: {msg}) # 试试用不同的 thread_id —— 那就是全新的对话 new_config {configurable: {thread_id: another_chat}} print(\n 新线程不同 thread_idAI 完全不认识 ) result3 graph.invoke( {messages: [(user, 你还记得我叫什么吗)], user_name: 路人}, new_config ) for role, msg in result3[messages]: print(f{role}: {msg})运行上面代码你会看到同一 thread_id 下第二轮 AI 会回答“我知道你叫小明”。换一个 thread_idAI 又是从头开始。这就是短期记忆最核心的模式。生产环境里你不能只用内存服务重启就丢了所以下面我们换成SQLite数据库版。三、短期记忆二用 SQLite 数据库——生产环境首选SQLite 会把状态存成一个文件你甚至可以把它放到服务器上随时读取。3.1 改动只有两处导入 SqliteSaver 而不是 InMemorySaver创建数据库连接然后把 checkpointer 传给编译函数完整代码依然是复制即用import sqlite3 from typing import Annotated from typing_extensions import TypedDict import operator # 注意SqliteSaver 在 langgraph.checkpoint.sqlite 里 from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph, START, END class ChatState(TypedDict): messages: Annotated[list, operator.add] user_name: str def greeting_node(state: ChatState): user_name state.get(user_name, 访客) return {messages: [(assistant, f嗨{user_name}我是你的数据库记忆助手。)]} def respond_node(state: ChatState): user_msgs [m for m in state[messages] if m[0] user] if not user_msgs: return {messages: [(assistant, 你说啥)]} last_msg user_msgs[-1][1] user_name state.get(user_name, 访客) if 名字 in last_msg or 我叫 in last_msg: reply f好嘞我记住你叫 {user_name} 了 else: reply f收到{user_name}。继续聊 return {messages: [(assistant, reply)]} # 1️⃣ 创建或连接到 SQLite 数据库文件 conn sqlite3.connect(chat_memory.db, check_same_threadFalse) # 2️⃣ 创建 SqliteSaver 实例 saver SqliteSaver(conn) builder StateGraph(ChatState) builder.add_node(greeting, greeting_node) builder.add_node(respond, respond_node) builder.add_edge(START, greeting) builder.add_edge(greeting, respond) builder.add_edge(respond, END) # 3️⃣ 编译时把 SQLite Saver 作为 checkpointer graph builder.compile(checkpointersaver) # 同一个 thread_id数据会保留在 chat_memory.db 里即使你关掉 Python 再重启 config {configurable: {thread_id: zhang_san}} print(第1次调用) resp graph.invoke({messages: [(user, 我叫张三)], user_name: 张三}, config) print(resp[messages]) print(\n第2次调用同线程) resp2 graph.invoke({messages: [(user, 我叫什么)], user_name: 张三}, config) print(resp2[messages]) # 记得关闭连接 conn.close()好处即使程序重启数据库里的 chat_memory.db 还在所有对话历史不会丢。你可以把 SQLite 换成 PostgreSQL生产更推荐但思路完全一样。四、长期记忆跨会话记住用户——比女朋友记性还好短期记忆限制在一个 thread_id 里。如果用户今天用手机聊明天用电脑聊你想让他不用重新输入“我喜欢喝冰美式”就需要长期记忆。LangGraph 的 Store 机制可以存任何键值对数据并且支持命名空间就像文件夹一样。4.1 用InMemoryStore给你看原理下面的例子中AI 会从存储里读取该用户的历史偏好即使换了 thread_id也能记住用户的名字。from typing import Annotated, List from langchain_core.messages import HumanMessage, AIMessage from langgraph.graph import StateGraph, START from langgraph.store.memory import InMemoryStore # 定义状态这里直接用消息列表 class ChatState(TypedDict): messages: Annotated[List, lambda x, y: x y] def chat_node(state: ChatState, *, store): 聊天节点可以从 store 读取/写入长期记忆 # 假设当前用户ID正常应该从 config 里取这里演示写固定值 user_id user_888 # ---------- 读取长期记忆 ---------- # store.get((命名空间,), key) - 返回一个 item没有则 None user_item store.get((users,), user_id) if user_item: user_info user_item.value print(f[记忆] 读取到历史信息{user_info}) else: user_info {} print([记忆] 没有找到历史信息这是第一次对话。) # 获取用户最新消息 last_msg state[messages][-1].content if state[messages] else if 我叫 in last_msg: # 提取简单名字这里示意生产请用正则 name last_msg.split(我叫)[-1].strip()[:10] user_info[name] name print(f[记忆] 新记住名字{name}) # 保存回长期记忆 store.put((users,), user_id, user_info) # 生成回复 known_name user_info.get(name, 朋友) reply f你好{known_name}有什么想聊的 return {messages: [AIMessage(contentreply)]} # ---------- 创建记忆存储 ---------- memory_store InMemoryStore() builder StateGraph(ChatState) builder.add_node(chat, chat_node) builder.add_edge(START, chat) # 关键编译时传入 storememory_store graph builder.compile(storememory_store) # 第一次对话用户说名字 print( 第一次对话thread_1) resp1 graph.invoke({messages: [HumanMessage(content你好我叫李雷)]}) print(AI:, resp1[messages][-1].content) # 第二次对话换一个全新的 thread_id但长期记忆依然有效 # 注意store 是全局的不依赖于 config print(\n 第二次对话新 thread但我们能通过 store 找回记忆 ) resp2 graph.invoke({messages: [HumanMessage(content你还记得我叫什么吗)]}) print(AI:, resp2[messages][-1].content) # 查看 store 里保存的内容 user_data memory_store.get((users,), user_888) print(f\n[验证] 长期记忆存储{user_data.value if user_data else 无})输出示例第二次对话 AI 会说“你好李雷有什么想聊的”——证明它跨会话记住了你。4.2 生产环境SQLite 版长期记忆与检查点分开LangGraph 也提供了 SqliteStore可以和 SqliteSaver 一起用共享同一个数据库文件。但写法稍复杂官方推荐用上下文管理器。下面给出可以直接复制运行的生产级例子同时使用了短期记忆checkpointer和长期记忆store让你看到两者如何协同工作import sqlite3 import uuid from langgraph.graph import StateGraph, START, MessagesState from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.store.sqlite import SqliteStore from langchain_core.messages import HumanMessage, AIMessage DB_PATH my_agent_memory.db def call_model(state: MessagesState, config, *, store): 一个既能短期记忆又能长期记忆的节点 user_id config[configurable][user_id] namespace (user_memories, user_id) # 1. 长期记忆检索之前的记忆 memories store.search(namespace, querystr(state[messages][-1].content)) mem_text \n.join([m.value[data] for m in memories]) if memories else system_hint f关于用户的历史信息{mem_text} if mem_text else # 2. 检测“记住”指令并写入长期记忆 last_msg state[messages][-1].content if 记住 in last_msg: memory_content f用户说过{last_msg} store.put(namespace, str(uuid.uuid4()), {data: memory_content}) reply f好的我记住了{last_msg} else: reply f收到消息{last_msg}。{system_hint}当前仅演示记忆能力 return {messages: [AIMessage(contentreply)]} # 同时初始化 checkpointer 和 store with SqliteStore.from_conn_string(DB_PATH) as store, \ SqliteSaver.from_conn_string(DB_PATH) as checkpointer: builder StateGraph(MessagesState) builder.add_node(call_model) builder.add_edge(START, call_model) graph builder.compile(checkpointercheckpointer, storestore) # 第一次对话要求 AI 记住信息 config1 {configurable: {thread_id: 1, user_id: alice}} res graph.invoke({messages: [HumanMessage(content记住我家里养了一只猫叫胖丁)]}, config1) print(AI:, res[messages][-1].content) # 第二次对话新线程但 user_id 相同AI 能通过长期记忆召回 config2 {configurable: {thread_id: 2, user_id: alice}} res2 graph.invoke({messages: [HumanMessage(content我家猫叫什么)]}, config2) print(AI:, res2[messages][-1].content)这样你的 AI 就拥有了双重记忆系统既不会丢掉当前对话的上下文也能跨会话记住用户的重要信息。五、记忆管理进阶不让 AI “撑爆”上下文窗口即便是最好的大模型上下文长度也有限。如果你的短期记忆里攒了上千条消息迟早会把模型“撑死”。你需要学会修剪消息、删除消息和总结消息。5.1 修剪消息保留最后 N 个 tokenLangChain 提供了一个 trim_messages 函数你可以设定最多保留多少个 token它会自动去掉前面的旧消息。from langchain_core.messages import HumanMessage, AIMessage from langchain_core.messages.utils import trim_messages, count_tokens_approximately # 假设我们有很长的一段对话历史 long_history [ HumanMessage(content第1条很长很长的消息... * 20), AIMessage(content回复1 * 20), HumanMessage(content第2条也很长... * 20), AIMessage(content回复2 * 20), HumanMessage(content最后一条我是小明), ] # 修剪只保留最后 100 个 token 左右并且尽量以 human 消息结尾 trimmed trim_messages( long_history, strategylast, token_countercount_tokens_approximately, max_tokens100, start_onhuman, # 从 human 消息开始 end_on(human, tool) # 尽量结束在 human 或 tool 消息 ) print(f原始消息数{len(long_history)}修剪后{len(trimmed)}) for msg in trimmed: print(msg.content[:50])使用场景在调用大模型之前的节点里调用 trim_messages再传给模型。5.2 删除消息用RemoveMessage如果你的状态里用了 add_messages reducer那么你可以返回一个 RemoveMessage 对象来删除特定消息。from langchain_core.messages import RemoveMessage, HumanMessage # 假设当前 state[messages] 包含很多消息 # 你想删除第一条消息 to_remove RemoveMessage(idstate[messages][0].id) # 在节点中返回 {messages: [to_remove]} 即可删除5.3 总结消息最优雅的长上下文策略当你觉得对话太长时可以写一个总结节点用一个更便宜的小模型把前面几十条消息压缩成一段摘要然后用摘要替换掉原始消息。def summarize_conversation(state): messages state[messages] if len(messages) 10: # 消息太少就不要总结了 return {} # 假设你有一个 summarize_model summary_text 用户问了关于天气和名字的问题以及…… # 实际调用模型生成 # 删除旧消息保留一个 SystemMessage 包含摘要 new_messages [SystemMessage(contentf之前的对话摘要{summary_text})] messages[-4:] return {messages: new_messages}小技巧把总结节点放在对话进行到一定长度后自动触发既节省 token 又不丢失关键信息。六、文章总结看完你可以带走什么今天我们基本把 LangGraph 的记忆管理从头到尾过了一遍记忆类型用途核心组件关键点短期记忆多轮对话、单次任务的上下文Checkpointerthread_id状态自动存档消息自动追加长期记忆跨会话的用户属性、知识库Store 命名空间独立于线程按用户 ID 存取消息修剪防止上下文爆炸trim_messages保留最后若干个 token消息删除精确控制状态大小RemoveMessage依赖add_messagesreducer对话总结优雅压缩历史自定义 summarize 节点用低成本模型生成摘要你现在能做什么直接复制文章里的代码改一改就能用到你自己的客服机器人、个人助理、RAG Agent 里。再也不用怕用户骂 AI “你怎么又忘了”。面试官问你 LangGraph 记忆怎么实现你可以自信地画出 checkpointer 和 store 两张牌。记住一句话短期记忆保证“当前聊得顺”长期记忆保证“下次记得你”两者结合才是真智能。如果这篇文章帮到了你不妨收藏下来以后写 Agent 记忆功能时直接翻出代码示例修改。也欢迎你在评论区留言交流你在记忆管理上遇到的奇葩问题 → 我们一起踩坑一起填坑。