记忆管理

一句话理解

短期记忆 = 会话内的聊天记录（多轮对话） 长期记忆 = 跨会话的用户画像（记住用户偏好）

两种记忆对比

特性	短期记忆	长期记忆
作用域	单个会话（thread_id）	跨会话、跨线程
用途	多轮对话上下文	用户偏好、应用级数据
实现	checkpointer	Store
数据	消息历史 + State 自定义字段	任意结构化数据
持久化	内存/SQL/Redis	内存/SQL

快速上手

短期记忆（多轮对话）

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph

# 关键：传入 checkpointer
checkpointer = InMemorySaver()
graph = StateGraph(...).compile(checkpointer=checkpointer)

# 同一个 thread_id 会共享记忆
graph.invoke({"messages": [...]}, {"configurable": {"thread_id": "1"}})

长期记忆（用户数据）

from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# 存：namespace + key
store.put(("用户偏好",), "user_123", {"喜欢咖啡": True})

# 取
info = store.get(("用户偏好",), "user_123")

核心概念

thread_id

• 类似微信聊天记录，一个 thread_id = 一会话
• 不同 thread_id 之间数据隔离
• 通过 configurable: {"thread_id": "xxx"} 指定

State（状态）

• 短期记忆的载体，包含：
  • 消息历史 messages
  • 自定义字段（如文件、文档、生成的工件）
• 每个步骤开始时自动读取 State

# State 不仅存消息，还可以存自定义数据
class CustomState(TypedDict):
    messages: Messages
    uploaded_files: list      # 上传的文件
    retrieved_docs: list       # 检索的文档
    user_preferences: dict    # 用户偏好

namespace（命名空间）

• 长期记忆的分类方式
• 格式：元组 ("用户",) 或 ("用户", "偏好")
• 同一 namespace 下 key 唯一

checkpointer

• 负责保存/恢复 State
• 自动保存：图被调用或步骤完成时
• 自动恢复：每个步骤开始时读取
• 支持断点续传

Store

• 长期存储，需要主动读写
• 不随图执行自动更新

进阶用法

自定义 State 字段

from langgraph.graph import MessagesState
from typing import TypedDict

class CustomState(MessagesState):
    user_preferences: dict

# 创建带自定义字段的智能体
agent = create_agent(model, tools, state_schema=CustomState)

工具中访问 State

from langchain.tools import tool
from langchain.agents import ToolRuntime

@tool
def 查询天气(location: str, runtime: ToolRuntime) -> str:
    """查询天气"""
    # 访问短期记忆（State）
    history = runtime.messages

    return "晴天"

获取上一步结果

from langgraph.func import entrypoint

@entrypoint(checkpointer=checkpointer)
def workflow(text: str):
    # 获取上一步的返回值
    previous = get_previous_state()  # 或 previous 参数
    return text.upper()

# 同一 thread_id 调用
workflow.invoke("hello")  # 返回 "HELLO"
workflow.invoke("world")  # 返回 "WORLD" + 可获取上一步 "HELLO"

长对话优化

当对话过长超出上下文窗口时：

方案	做法	适用场景
修剪	删除头部/尾部消息	对话主题明确
总结	用 LLM 压缩为摘要	需要保留核心信息
过滤	只保留最近 N 条	简单场景

from langchain.messages import trim_messages

# 只保留最近 10 条消息，总 token 不超过 4000
trim_messages(messages, max_tokens=4000, strategy="last", token_counter=len)

生产环境

存储	场景	说明
InMemory*	开发调试	内存，重启丢失
Postgres*	生产环境	关系型数据库
Redis*	高并发	性能优先

# PostgreSQL 示例
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore

checkpointer = PostgresSaver.from_conn_string("postgresql://...")
checkpointer.setup()  # 建表

store = PostgresStore.from_conn_string("postgresql://...")
store.setup()

记忆流程图

常见问题

Q: 短期记忆和长期记忆可以同时用吗？ A: 可以，两者互补。短期记忆管对话，长期记忆管用户画像。

Q: State 和消息有什么区别？ A: State 是整个短期记忆的载体，messages 是 State 中的一个字段，专门存对话历史。

Q: 内存不够怎么办？ A: 生产环境用 PostgresSaver/PostgresStore 替代 InMemory 实现。

Q: 如何选择 trim 策略？ A: 保留最新消息用 strategy="last"，保留系统提示用 strategy="first"。