LangGraph 记忆模块实战：从线程记忆、长期记忆到对话摘要压缩

一、为什么 LangGraph 需要“记忆”？

在普通的 LLM 调用中，模型本身是无状态的。也就是说，用户第一轮说“我叫小明”，第二轮再问“我是谁？”，如果我们没有把历史消息重新传给模型，模型并不知道“小明”是谁。
LangGraph 的记忆机制就是为了解决这个问题。
参考https://docs.langchain.com/oss/python/langgraph/add-memory，
可以把 LangGraph 的记忆分成三类：

• 线程级记忆：同一个 thread_id 下保存完整图状态，适合连续对话。
• 长期记忆：使用 store 保存用户偏好、个人信息等跨会话数据。
• 上下文压缩记忆：当消息过多时，通过裁剪或摘要保留关键信息，避免 token 爆炸。
  这三类记忆组合起来，就可以构建一个既能“记住上下文”，又能“长期了解用户偏好”的智能体。
  本文根据LangChainDocs官方文档写了test几个demo进行实践。

二、线程级记忆：Checkpointer + thread_id

在 test.py 中，代码使用了 PostgresSaver 作为 LangGraph 的 checkpoint 存储：

fromlanggraph.checkpoint.postgresimportPostgresSaverfrompsycopgimportConnection conn=Connection.connect(DB_URI,autocommit=True)checkpointer=PostgresSaver(conn)agent_search=agent_builder.compile(checkpointer=checkpointer)

这里的核心是 checkpointer。它会把每次图执行后的状态保存下来，包括消息列表、节点执行位置、状态字段等。
调用时通过 configurable.thread_id 指定会话线程：

config={"configurable":{"thread_id":"2"}}response=agent_search.invoke({"messages":[{"role":"user","content":"今天成都天气怎么样？"}],"llm_calls":0},config=config)

只要后续继续使用同一个 thread_id，LangGraph 就能从 checkpoint 中恢复这条线程之前的状态。因此第二次可以继续问：

response=agent_search.invoke({"messages":[{"role":"user","content":"刚才天气怎么样？"}],"llm_calls":0},config=config)

模型之所以知道“刚才天气”指什么，不是因为模型真的记住了，而是 LangGraph 根据 thread_id 找回了历史状态。
注意：实际发布代码时，不建议把数据库连接字符串硬编码在代码中，推荐放到 .env 里，例如 POSTGRES_URI。

三、查看历史状态与时间旅行

LangGraph 还有些很有特色的能力：查看历史状态和从某个 checkpoint 重放。

print(agent_search.get_state(config))forstateSnapshotinagent_search.get_state_history(config):print("id :",stateSnapshot.config["configurable"]["checkpoint_id"])print("下一个节点：",stateSnapshot.next)

get_state(config)可以查看当前线程的最新状态。
get_state_history(config)可以查看这个线程下所有历史快照。每个快照都有自己的checkpoint_id。
如果想从某个历史状态继续执行，可以使用：
result = agent_search.invoke(None, config=to_replay.config)
这就类似“时间旅行”：从某个历史节点重新运行图。这个能力在调试 Agent、复现线上问题、观察工具调用链路时非常有用。

四、短期记忆和长期记忆的区别

在 test2.py 中，代码同时使用了：

graph=builder.compile(checkpointer=InMemorySaver(),store=store)

这里要区分两个概念：
类型 对应组件 作用
短期记忆 checkpointer 保存某个 thread_id 下的对话状态
长期记忆 store 保存用户信息、偏好等跨线程数据

checkpointer 更像“当前会话的上下文记录”，它依赖 thread_id。
store 更像“用户档案库”，它可以跨不同 thread_id 读取，只要使用相同的 user_id 或命名空间即可。

五、长期记忆：InMemoryStore 与命名空间

test2.py 中使用 InMemoryStore 创建了长期记忆：

fromlanggraph.store.memoryimportInMemoryStorefromlangchain.embeddingsimportinit_embeddings store=InMemoryStore(index={"embed":init_embeddings("openai:text-embedding-3-small"),"dims":1536,"fields":["$"]})

这里的 index 表示给存储内容建立向量索引，后续可以做语义搜索。
写入记忆时，代码使用了分层命名空间：

user_id1="user_123"namespace_1=(user_id1,"prefs","food")memory_id_1=str(uuid.uuid4())memory_value_1={"food":"披萨"}store.put(namespace_1,memory_id_1,memory_value_1)

这个命名空间(user_id, "prefs", "food")很关键。它可以把不同用户、不同类型的数据隔离开。
例如：

namespace_1=("user_123","prefs","food")namespace_2=("user_456","prefs","music")

这样 user_123 的饮食偏好和 user_456 的音乐偏好就不会混在一起。
读取单条记忆：

single_item=store.get(namespace_1,memory_id_1)print(single_item.value)

语义搜索记忆：

all_mem=store.search(("user_123","prefs","food"),query="用户喜欢的食物",limit=1)

如果配置了 embedding，store.search 不只是关键词匹配，而是可以按照语义查找相关记忆。

六、PostgresStore：让长期记忆真正持久化

InMemoryStore 只适合本地调试，程序重启后数据就会丢失。如果希望长期记忆落库，可以使用 PostgresStore。
test2.py 如下：

fromlanggraph.store.postgresimportPostgresStorefromlanggraph.checkpoint.postgresimportPostgresSaverwith(PostgresSaver.from_conn_string(DB_URI)ascheckpointer,PostgresStore.from_conn_string(DB_URI)asstore,):store.setup()store.put(namespace=namespace,key=memory_id,value=memory_value)

生产环境中更推荐这样写：

DB_URI=os.environ["POSTGRES_URI"]with(PostgresSaver.from_conn_string(DB_URI)ascheckpointer,PostgresStore.from_conn_string(DB_URI)asstore,):checkpointer.setup()store.setup()

PostgresSaver 负责保存图的 checkpoint。
PostgresStore 负责保存长期记忆。
两者职责不同，但可以使用同一个 PostgreSQL 数据库。

七、跨线程读取用户记忆

test3.py 最接近真实业务示例。
第一轮对话：

config={"configurable":{"thread_id":"1111","user_id":"user_123"}}response=agent_search.invoke({"messages":[HumanMessage(content="我叫小明，我喜欢吃川菜")]},config)

第二轮对话换了新的 thread_id：

config2={"configurable":{"thread_id":"2222","user_id":"user_123"}}response=agent_search.invoke({"messages":[HumanMessage(content="给我推荐下相关餐厅")]},config2)

这里最重要的是：thread_id 变了，但 user_id 没变。
这意味着第二轮对话虽然是一个新会话，但仍然可以通过 store 读取 user_123 的长期记忆，比如姓名和饮食偏好。

八、用结构化输出抽取用户信息

test3.py 中定义了一个 Person 模型：

classPerson(BaseModel):name:Optional[str]=Field(default=None,description="人的姓名")height:Optional[str]=Field(default=None,description="人的身高，以米为单位")favourite_food:Optional[list[str]]=Field(default=None,description="最喜欢的食物的列表")

然后通过：
model_with_structured = model.with_structured_output(Person)
让大模型按照固定结构抽取用户信息。
抽取节点如下：

defget_person_by_llm(state,config,*,store):people_info=model_with_structured.invoke([SystemMessage(content="你是一个提取信息的专家，只从文本中提取用户本人的相关信息，不能提取别人的信息。不知道的值返回null。")]+state["messages"][-3:])user_id=config["configurable"]["user_id"]namespace1=(user_id,"info")namespace2=(user_id,"prefs")store.put(namespace1,"user_base",{"name":people_info.name,"height":people_info.height,})store.put(namespace2,"user_food",{"favourite_food":people_info.favourite_food,})

这里没有使用随机 UUID，而是使用固定 key：
“user_base”
“user_food”
这样做的好处是：同一类信息可以被覆盖更新，不会无限生成重复记忆。
例如用户第一次说“我喜欢吃川菜”，后面又说“我现在更喜欢粤菜”，就可以覆盖更新偏好，而不是存成两条互相冲突的记忆。
九、在 LLM 调用前注入长期记忆
抽取到用户信息后，需要在下一次调用模型前把记忆注入上下文。
test3.py 中的 llm_call 做了这件事：

defllm_call(state,config,*,store):user_id=config["configurable"]["user_id"]namespace1=(user_id,"info")namespace2=(user_id,"prefs")info_result=store.search(namespace1)prefs_result=store.search(namespace2)user_info=info_result[0].valueifinfo_resultelse"暂无用户基础信息"user_prefs=prefs_result[0].valueifprefs_resultelse"暂无用户饮食偏好"

然后把这些记忆拼到模型输入里：

result=model_with_tools.invoke([SystemMessage(content=sys_prompt)]+[HumanMessage(content=f"调用聊天模型前可以参考以下信息：用户基本情况：{user_info}用户偏好情况：{user_prefs}")]+messages)

这样模型在回答“给我推荐下相关餐厅”时，就能参考之前保存的“用户喜欢吃川菜”。
这就是典型的长期记忆工作流：
从用户消息中抽取信息。
写入长期记忆 store。
新会话开始时按 user_id 查找记忆。
把记忆注入模型上下文。
模型生成个性化回答。

十、消息过多怎么办：裁剪与删除

仅靠 checkpoint 保存完整消息，会遇到一个问题：对话越长，消息越多，最终 prompt 可能超过模型上下文窗口。
test4.py 演示了两种处理方式。
第一种是调用模型前进行消息裁剪：
messages = trim_messages(
state[“messages”],
strategy=“last”,
max_tokens=128,
token_counter=deepseek_token_counter,
start_on=“human”,
end_on=(“human”, “tool”)
)
这表示只保留最后一部分消息，并控制在指定 token 范围内。
第二种是从状态中删除旧消息：

iflen(state["messages"])>6:return{"messages":[RemoveMessage(id=msg.id)formsginstate["messages"]][:6]}

RemoveMessage 是 LangGraph 中管理消息状态的常见方式。它不是简单地在 Python list 里 pop，而是通过状态更新告诉 LangGraph 删除指定消息。
需要注意：test4.py 中出现了两个同名 call_model 函数，后一个会覆盖前一个。实际项目里建议把它们拆成两个不同函数，例如call_model_with_trim和call_model_with_remove。

十一、摘要压缩：删除前先总结

直接删除历史消息虽然能节省 token，但可能丢失关键信息。
因此 test5.py 使用了更合理的方式：先生成摘要，再删除旧消息。
状态中增加一个 summary 字段：

classState(MessagesState):summary:str

模型调用时，如果已有摘要，就把摘要加入上下文：

defcall_model(state:State):summary=state.get("summary","")all_messages=[]ifsummary:all_messages.append(HumanMessage(content=summary))all_messages.extend(state["messages"])resp=model.invoke(all_messages)return{"messages":[resp]}

当消息数量超过 6 条时，进入摘要节点：

defshould_summarize(state:State):iflen(state["messages"])>6:return"summarize_conversation"returnEND

摘要节点会基于历史消息生成新的 summary：

defsummarize_conversation(state:State):summary=state.get("summary","")ifsummary:summary_message=(f"这是目前为止的对话摘要：{summary}\n""基于上面的消息来扩展摘要:")else:summary_message="创建关于上面对话的完整摘要"result=model.invoke(state["messages"]+[HumanMessage(content=summary_message)])return{"summary":result.content,"messages":[RemoveMessage(id=m.id)forminstate["messages"][:-1]]}

这样既删除了大部分历史消息，又把关键信息保存到了 summary 字段中。
最终效果是：
短期内保留最近几轮原始对话。
长期保留历史摘要。
避免 prompt 越来越长。
用户问“what’s my name?” 时，仍然可以从摘要里找回 “bob”。

十二、完整记忆架构总结

结合这几个示例，可以把 LangGraph 的记忆模块理解成下面这张图：

其中：
StateGraph 负责编排智能体流程。
checkpointer 保存线程级上下文。
thread_id 区分不同会话。
store 保存长期记忆。
user_id 用来跨线程关联同一个用户。
RemoveMessage 和 summary 用来控制上下文长度。
PostgresSaver 和 PostgresStore 用来实现生产级持久化。

十三、开发LangGraph 记忆模块中的建议

本地调试用 InMemorySaver 和 InMemoryStore 即可，简单方便。
生产环境使用 PostgresSaver 和 PostgresStore，否则程序重启后记忆会丢失。
thread_id 用来保存单次会话上下文，user_id 用来关联长期用户画像。
用户偏好、姓名、身高等稳定信息适合放到 store。
最近几轮聊天适合交给 checkpointer 管理。
对话变长后，不要无限追加历史消息，要做裁剪或摘要。
抽取长期记忆时，推荐使用结构化输出，减少脏数据。
存储记忆时，固定 key 适合覆盖更新，UUID key 适合保存多条独立事实。
不要把数据库账号、API Key 写死在代码中。

十四、结语

LangGraph 的记忆模块不是单一组件，而是一套组合能力。
如果只需要让同一轮会话连续起来，用 checkpointer + thread_id 就够了。
如果希望智能体跨会话记住用户偏好，就需要引入 store + user_id。
如果对话会持续很久，还要加入 trim_messages、RemoveMessage 或摘要压缩。
参考文档写的demo的 test.py 到 test5.py 正好展示了这个递进过程：
test.py：Postgres checkpoint、状态恢复、历史快照、重放。
test2.py：长期记忆 store、命名空间、语义搜索、Postgres 持久化。
test3.py：结构化抽取用户信息，并跨线程读取长期记忆。
test4.py：消息裁剪和消息删除。
test5.py：摘要压缩，删除旧消息前先保留关键信息。
掌握这几种模式后，就可以把 LangGraph Agent 从“单轮问答机器人”升级成真正具备上下文记忆和用户画像能力的智能体。

参考文档

官方文档：LangChainDocs
实践demo: gitte仓库