记忆分为短期记忆和长期记忆，对应不同的使用场景：

• 短期记忆（Short-term memory、会话级记忆、thread-scoped memory）：作用范围是单个 对话线程（Thread）内，一旦开启新对话（更换 thread_id ），记忆即消失。
• 长期记忆（Long-term memory，跨会话级记忆 ）：在会话间存储用户特定或应用级数据， 并 在 会话线程间共享 。它可以随时在任何线程中被调用。记忆的范围是任意自定义命名空间，而不 仅仅是单一线程 ID。

在langchain 1.x中我们如何使用记忆呢

在LangChain v0.x版本中，通过专用的xxxMemory类管理记忆。 在LangChain v1.x版本中，Agent是构建在LangGraph图结构之上的，通过上文提到的state和store构建记忆系统。使用更简单、功能更统一。

• state：短期记忆对象，以 会话 为单位组织，包含当前会话的所有消息记录以及自定义信息。
• store：长期记忆对象， 跨会话持久化 的数据，通常需要结合向量数据库或外部存储实现。

1. 短期记忆

LangChain1.x 的短期记忆是三者的组合：

State（会话内部状态） + Checkpointer（持久化机制） + Thread ID（会话作用域）

• State ：默认 存储历史消息列表messages ，通过State 管理历史消息
• Checkpointer ：负责将State 作为检查点持久化保存，检查点是某个时刻的State 快照
• Thread ID ：用于唯一标识State ，LangChain运行时会按照 thread_id 读写State快照

1.1 基于内存的持久化器

python
from langchain_core.messages import HumanMessage
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()
# 1. 创建 Agent 时添加 checkpointer
agent = create_agent(
    model=model,
    checkpointer=checkpointer  # 添加内存管理
)
# 2. 调用时指定 thread_id
config = {
    "configurable": {
        "thread_id": "1"
    }
}
print("\n第一轮对话：")
response1 = agent.invoke({
    "messages": [HumanMessage("我叫张三")]},
    config=config  # 传入 config
)
print(f"Agent: {response1['messages'][-1].content}")
print("\n第二轮对话：")
response2 = agent.invoke({
    "messages": [HumanMessage("我叫什么？")]},
    config=config  # 使用相同的 thread_id
)
print(f"Agent: {response2['messages'][-1].content}")

只需传入 checkpointer 和 config，Agent 就能自然具备连续对话能力。 如果更新线程ID，则会重新开启对话。thread_id 隔离不同会话空间。

注意： InMemorySaver会存在以下问题

1. InMemorySaver 只保存在内存中

    ✅ 同一进程内有效（不支持跨进程共享）
    ❌ 程序重启后丢失（或进程重启后丢失）
    ❌ 不同进程无法共享
    解决方案：持久化（SQLite、PostgreSQL）
   

2. InMemorySaver 会保存所有消息

    消息越来越多（无限增长，需要管理上下文）
    token消耗增加，甚至会超过模型的 token 限制
    响应速度变慢、成本增加
    解决方案：上下文管理（修剪、摘要）
    
   

1.2 基于外部存储介质的持久化器

如果将 状态检查点(checkpointer) 保存在内存， 进程结束 则状态丢失，生产环境不可接受。因此，生产环境要用持久化的外部存储介质，如PostgreSQL。LangGraph提供的checkpointer后端列表如下 https://docs.langchain.com/oss/python/langgraph/persistence#checkpointer-libraries 此处选择PostgreSQL作为持久化器。

python
from langgraph.checkpoint.postgres import PostgresSaver

DB_URL = "postgresql://postgres:123456@127.0.0.1:5432/langchain_db?sslmode=disable"

with PostgresSaver.from_conn_string(DB_URL) as checkpointer:
    # 初始化PostgreSQL数据库
    checkpointer.setup()
    agent = create_agent(
        model=model,
        checkpointer=checkpointer
    )
    config = {"configurable": {"thread_id": "1"}}
    response1 = agent.invoke(
        {"messages": [HumanMessage("你好，我是老王")]},
        config=config
    )
    print("=" * 30, "-> 第一次调用 <-", "=" * 30)
    for msg in response1["messages"]:
        msg.pretty_print()

    response2 = agent.invoke(
        {"messages": [HumanMessage("你好，我是谁？")]},
        config=config
    )
    print("=" * 30, "-> 第二次调用 <-", "=" * 30)
    for msg in response2["messages"]:
        msg.pretty_print()

setup() 用于初始化PostgreSQL数据库，首次运行会创建必要的表，重复执行不会重新建表，底层逻辑是 Create IF Not Exists langchain会初始化4张表：

• checkpoints ：这是主表，存每个 thread 在某个时刻的 checkpoint 快照。
• checkpoint_blobs ：这张表专门存不适合直接内联进 checkpoints.checkpoint 的较复杂 channel 值。
• checkpoint_writes ：这张表存的是中间写入 / pending writes，不是最终完整 checkpoint。
• checkpoint_migrations ：这张表不是业务数据表，而是迁移版本表。

关于这两种策略：

1. InMemorySaver()将状态持久化到内存， 进程结束或重建Saver() 则历史状态丢失
2. 基于外部存储介质（如PostgreSQL）的持久化器，其存储的状态不会随进程终止而丢失，只要 不 显式删除历史状态 ，即可通过 thread_id 加载历史状态。

1.3 记忆治理策略（上下文管理）

随着对话的进行，历史消息不断累积， state会持续增长 ，为模型带来挑战：

1. LLM的 上下文窗口是有限的 ，完整历史可能无法装入LLM的上下文窗口，导致上下文丢失或错 误。
2. 即便模型的上下文窗口够大，多数LLM在长上下文场景仍然表现不佳。模型会 被陈旧或离题的内 容“分散注意力” 。
3. 同时，会带来 高昂的token花费 。 此时需要对上下文进行管理：对历史记录进行压缩、清理、重组等。

1.3.1 消息裁剪

调用模型前裁剪上下文。 目标是控制token用量，通常 保留系统初始消息和最近若干消息 ，或 按token数保留末尾内容 。 适合成本敏感、对旧上下文依赖不强的场景。

python
from langchain_core.messages import HumanMessage
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import InMemorySaver
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langchain.agents import AgentState, create_agent
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from typing import Any


@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    """在调用模型之前，判断是否需要对消息进行裁剪"""
    messages = state["messages"]
    if len(messages) <= 3:
        return None

    # 保留起始消息
    first_msg = messages[0]
    # 如果有偶数条消息，则取最近的3条消息；如果有奇数条消息，则取最近4条消息
    recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
    new_messages = [first_msg] + recent_messages
    return {
        "messages": [
            # 把原有的消息删除
            RemoveMessage(id=REMOVE_ALL_MESSAGES),
            # 再添加要保留的消息
            *new_messages
        ]
    }


agent = create_agent(
    model=model,
    middleware=[trim_messages],
    checkpointer=InMemorySaver(),
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": [HumanMessage("你好，我是老王")]}, config)
agent.invoke({"messages": [HumanMessage("从现在起，你叫小王")]}, config)
agent.invoke({"messages": [HumanMessage("今天天气不错")]}, config)
final_response = agent.invoke({"messages": [HumanMessage("告诉我，你是谁？我是谁？")]}, config)

for msg in final_response["messages"]:
    msg.pretty_print()

1.3.2 消息删除

消息裁剪强调“在 模型调用前裁剪 消息列表，控制模型可以看到的上下文范围”，而消息删除强调 模型调用完成后将某些消息从消息列表中移除 ，永久更改状态。 适合明确要遗忘、清理、重置某些历史。

python
from langchain.agents.middleware import after_model


@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
    messages = state["messages"]
    # 保持最近的 5 条消息
    if len(messages) > 5:
        # 框架中通常使用 RemoveMessage 来标记删除，并返回更新状态。
        to_delete = len(messages) - 5
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:to_delete]]}
    return None


agent = create_agent(
    model=model,
    middleware=[delete_old_messages],
    checkpointer=InMemorySaver()
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}
agent.invoke({"messages": "你好，我是老王"}, config)
agent.invoke({"messages": "从现在起，你叫小王"}, config)
agent.invoke({"messages": "今天天气不错"}, config)
final_response = agent.invoke({"messages": "告诉我，你是谁？我是谁？"}, config)
for msg in final_response["messages"]:
    msg.pretty_print()

RemoveMessage到底干了什么？ 当你在中间件里返回 [RemoveMessage(id=m.id)] 时，你实际上是向框架发送了一个 删除指令 。 框架的底层处理逻辑如下：

palaintext
[历史消息池 (内存中持续存在)]
├── Message(id="1", content="你好，我是老王")
├── Message(id="2", content="...")
└── RemoveMessage(id="1") <-- 这是一个新追加进去的“墓碑”标记

1. 追加“墓碑”标记：框架收到 RemoveMessage(id="1") 后，并不会去内存的数组里把 id="1" 的对象删掉，而是把这个 RemoveMessage 作为一条新记录追加到当前线程的状态历史中。这个 RemoveMessage 就像是一个“墓碑”。
2. 运行时过滤合并（Reducer）：当下一次你再次调用 agent.invoke 或者大模型要去读取上下文 时，框架的内置合并器（Reducer）会把“原始消息”和“墓碑标记”放在一起进行计算： 原 始 消 息 墓 碑 标 记 对 外 隐 藏它在丢给大模型之前，会自动把被标记删除的消息过滤掉。

1.3.3 摘要

把早期历史压缩成摘要，再替换原始消息。 消息裁剪和删除都会导致上下文缺失，影响回答质量和用户体验。和它们相比，摘要是更适合长会话的 折中方案：保语义，不保原文。官方推荐内置 SummarizationMiddleware 。

python
agent = create_agent(
    model=model_out,
    tools=[],
    checkpointer=InMemorySaver(),
    middleware=[
        SummarizationMiddleware(
            model=model_in,
            trigger=[
                ("tokens", 100),  # 超过 100 tokens 就摘要
            ],
            keep=("messages", 2),
            summary_prompt="对历史消息摘要，消息列表如下\n{messages}",
        )
    ]
)

具体使用参考上一篇

设置最大token数触发摘要的标准是啥？

建议如下：

trigger_token = 模型总上下文窗口 × 0.70 ~ 0.80

2. 长期记忆

长期记忆的存储是 store -> namespace -> key -> value 的四层架构。

每个namespace存储的都是key-value键值对，通过key可以唯一标识一条value。

python
namespace = ("users", "user_123", "preferences") # 元组类型
key = "profile" # 字符串类型
value = { # 字典类型
    "language": "zh-CN",
    "style": "short_direct",
    "likes": ["python", "rag"]
}
store.put(namespace, key, value)

2.1 基础API

LangChain 1.2.x 的长期记忆基于 store 持久化数据，相关的API有：

• put() ：负责写入
• get() ：负责读取
• search() ：负责检索 我们可以在Agent执行流程之外直接访问长期记忆。

2.1.1 put()/get()：写入/读取API

1. put 参数说明

• namespace: 文档所在的层级路径
• key: 该路径下的唯一键
• value: 要保存的 JSON-like 字典
• index: 控制语义检索索引
  • None(默认选项): 使用 store 初始化时配置的索引配置，如果初始化时没有指定索引策略，则
  • index参数将会被忽略
  • False: 不为该 item 建立语义索引
  • list[str]: 只对指定字段路径建索引
• ttl: 可选，过期时间；是否支持取决于具体 store 实现

2. get 参数说明

• namespace: 文档所在的层级路径
• key: 该路径下的唯一键
• refresh_ttl：是否刷新当前item的ttl（time-to-live，存活时间）
  • 默认为None：表示采用创建store对象时指定的同名配置
  • 如果没有配置TTL，该参数被忽略。

3. 基于InMemoryStore

python
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
namespace = ("users",)
user_id = 'user-1'
username = "小蓝"
store.put(namespace, user_id, {"name": username})
print(store.get(namespace, user_id))
# Item(namespace=['users'], key='user-1', value={'name': '小蓝'}, created_at='2026-08-12T09:05:13.505569+00:00', updated_at='2026-08-12T09:05:13.505573+00:00')

注意到，Item对象新增了 created_at 和 updated_at 字段，分别为数据新增和更改时间。 注意：对于当前版本，InMemoryStore每次put都会创建一个新的Item对象，无论namespace和key是否相同，所以 created_at 和 updated_at 始终是一致的。

更新 对同一条数据进行更改后查询。

python
store.put(namespace, user_id, {"name": '小红'})
print(store.get(namespace, user_id))
# Item(namespace=['users'], key='user-1', value={'name': '小红'}, created_at='2026-08-12T09:07:25.450238+00:00', updated_at='2026-08-12T09:07:25.450242+00:00')

我们发现当namespace，key完全相同时，value被完全覆盖了，生成一条新的数据。

4. 基于PostgresStore PostgresStore不同于InMemoryStore，更改数据的逻辑是update而非覆盖，因此 created_at 固定为Item 创建时间，而 updated_at 为更新时间，二者可能不同，符合直觉。

python
from langgraph.store.postgres import PostgresStore

namespace = ("users",)
user_id = "user-11"
username = "小蓝"
DB_URL = "postgresql://postgres:123456@127.0.0.1:5432/langchain_db?sslmode=disable"
with PostgresStore.from_conn_string(DB_URL) as store:
    store.setup()
    store.put(namespace, user_id, {"name": username})
    print(store.get(namespace, user_id))
    # Item(namespace=['users'], key='user-11', value={'name': '小蓝'}, created_at='2026-08-12T09:12:12.388146+00:00', updated_at='2026-08-12T09:12:12.388146+00:00')

当对一条数据更新时

python
with PostgresStore.from_conn_string(DB_URI) as store:
    store.setup()
    store.put(namespace, user_id, {"name": "小红"})
    print(store.get(namespace, user_id))

created_at 不会变，只会更新updated_at的值。

2.1.2 search()：检索API

参数说明：

• namespace_prefix：命名空间前缀，在该前缀下搜索。
• query：语义检索时用于查询的自然语言。
• filter：过滤条件，value中的键值对组合，见下文举例。
• limit：可以返回item的最大条数，效果等同于SQL中的limit。
• offset：返回结果之前跳过的item数量。
• refresh_ttl：同上。

它支持两种检索方式（对应上面的参数2、3）：

• 按 filter 做结构化过滤 ，即用 value 中的 键值 筛选符合条件的记录。
• 按 query 做语义相似度检索 ，需要将输入转换为向量。

返回值： 返回匹配的 SearchItem 列表，并额外携带匹配分数等检索元信息。

举例1：按照namespace前缀搜索

python
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
namespace1 = ("users", "Alice", "memories")
key1 = 'preferences'
value1 = {
    "course": "计算机组成原理",
    "sports": "跑步",
    "food": "紫光园奶皮子酸奶"
}
namespace2 = ("users", "Bob", "memories")
key2 = 'preferences'
value2 = {
    "course": "数字电路与模拟电路",
    "sports": "跑步",
    "food": "奶皮子糖葫芦"
}
namespace3 = ("users", "Black", "memories")
key3 = 'preferences'
value3 = {
    "course": "数字电路与模拟电路",
    "sports": "羽毛球",
    "food": "紫光园奶皮子酸奶"
}
store.put(namespace1, key1, value1)
store.put(namespace2, key2, value2)
store.put(namespace3, key3, value3)

print('=' * 30, '-> (users, Alice) <-', "=" * 30)
for item in store.search(("users", "Alice")):
    print(item)

检索结果如下：

plaintext
============================== -> (users, Alice) <- ==============================
Item(namespace=['users', 'Alice', 'memories'], key='preferences', value={'course': '计算机组成原理', 'sports': '跑步', 'food': '紫光园奶皮子酸奶'}, created_at='2026-08-12T09:18:51.595739+00:00', updated_at='2026-08-12T09:18:51.595741+00:00', score=None)

举例2：按照filter过滤

python
print("=" * 30, '-> (users, ), filter=food <-", "=" * 30)')
for item in store.search(("users", ), filter={"food": "紫光园奶皮子酸奶"}):
    print(item)

检索结果如下：

plaintext
============================== -> (users, ), filter=food <-", "=" * 30)
Item(namespace=['users', 'Alice', 'memories'], key='preferences', value={'course': '计算机组成原理', 'sports': '跑步', 'food': '紫光园奶皮子酸奶'}, created_at='2026-08-12T09:18:51.595739+00:00', updated_at='2026-08-12T09:18:51.595741+00:00', score=None)
Item(namespace=['users', 'Black', 'memories'], key='preferences', value={'course': '数字电路与模拟电路', 'sports': '羽毛球', 'food': '紫光园奶皮子酸奶'}, created_at='2026-08-12T09:18:51.595784+00:00', updated_at='2026-08-12T09:18:51.595785+00:00', score=None)

举例3：按照语义搜索 自定义嵌入函数，目的是查看嵌入向量

python
# 自定义嵌入函数
def embed(text: list[str]) -> list[list[float]]:
    return [[1.0] * 6 for _ in range(len(text))]
    
index_config = {
    "embed": embed,
    "dims": 6,
    "fields": ["$", "course"]
}
store = InMemoryStore(
    index = index_config
)

初始化Store时通过index指定索引方式

• embeds ：将输入文本转换为向量的嵌入函数，可以是 自定义函数 ，也可以是 嵌入模型对象 ， 本例传递的是自定义嵌入函数。
• dims ：输出向量维度
• fields ：用于计算向量的属性列表，这里的属性都是指value中的key，value是一个JSON-like字典，可取值如下:
  • ["$"] ：将value作为整体嵌入
  • ["fileds1", "fields2"] ：单独指定某个一级字段
  • ["parent.child"] ：从内部的嵌套JSON对象中获取子字段的值
  • ["array[*].field"] ：从JSON数组的每个JSON对象中获取子字段的值
  • 注意：上述四种形式可以同时出现，fields列表的每个元素都会生成一个嵌入向量。

查看嵌入向量

python
from pprint import pprint
pprint(store._vectors)

还可以通过指定namespace、key、index_config的fields中指定的字段名查看特定向量。

python
from pprint import pprint
pprint(store._vectors[('users', 'Alice', 'memories')]['preferences']['$'])

使用嵌入模型 本例要通过CloseAI平台调用OpenAI的嵌入模型 openai

其嵌入维度为3072，所以此处的 dims 应设置为3072

python
from langgraph.store.memory import InMemoryStore
from langchain.embeddings import init_embeddings

embedding_model = init_embeddings(
    model="openai:text-embedding-3-large",
    api_key=os.getenv("CLOSEAI_API_KEY"),
    base_url=os.getenv("CLOSEAI_BASE_URL"),
)
index_config = {
    "embed": embedding_model,
    "dims": 3072,
    "fields": ["$"]
}
store = InMemoryStore(
    index=index_config
)

for item in store.search(("users", ), query="数电模电"):
    print(item)

如果只是指定了 query ，返回的是所有 namespace 前缀满足要求的 item 。底层会按照向量相 似度计算查询和候选的 score ，输出结果按照score 降序排列 。可以结合 limit 或 filter 限制数据条数。

2.2 在Agent运行图中访问长期记忆

我们可以在工具或中间件中访问长期记忆。

2.2.1 在工具中访问长期记忆

基于内存模式

python
from langchain_core.messages import HumanMessage
from typing import NotRequired
from langchain.agents import AgentState, create_agent
from langchain.tools import tool, ToolRuntime
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()


class CustomState(AgentState):
    user_id: NotRequired[str]


@tool(parse_docstring=True)
def save_user_info(name: str, runtime: ToolRuntime) -> str:
    """
    将用户信息保存在长期记忆中

    Args:
        name: 用户名

    Returns:
        str: 保存状态
    """

    runtime.store.put(("users",), runtime.state["user_id"], {"name": name})
    return "saved"


@tool(parse_docstring=True)
def get_user_info(runtime: ToolRuntime) -> str:
    """
    从长期记忆中读取用户信息

    Returns:
        str: 用户信息
    """
    item = runtime.store.get(("users",), runtime.state["user_id"])
    return str(item.value) if item else "unknown"


agent = create_agent(
    model=model,
    tools=[save_user_info, get_user_info],
    store=store,
    system_prompt="用户提及个人信息时及时记录，用户询问个人信息时尝试用工具检索",
    state_schema=CustomState,
)

print("=" * 30, '-> 第一个会话（线程） <-', "=" * 30)
response1 = agent.invoke({
    "messages": [HumanMessage("你好，很高兴认识你，我是小花")],
    "user_id": "user-1"
})
for msg in response1["messages"]:
    msg.pretty_print()
print("=" * 30, '-> 第二个会话（线程） <-', "=" * 30)
response2 = agent.invoke({
    "messages": [HumanMessage("我是谁")],
    "user_id": "user-1"
})
for msg in response2["messages"]:
    msg.pretty_print()

其中，CustomState扩展了 Agent 的标准状态。除了默认的 messages （历史消息列表）之外，还额外增加了一个 user_id 字段。这样，Agent 在运行过程中随时随地都能知道当前和它说话的用户 ID 是什么。

基于PostgresStore

python
from langgraph.store.postgres import PostgresStore

DB_URI = "postgresql://postgres:123456@127.0.0.1:5432/langchain_db?sslmode=disable"


class CustomState(AgentState):
   user_id: NotRequired[str]


@tool(parse_docstring=True)
def save_user_info(name: str, runtime: ToolRuntime) -> str:
   """
   将用户信息保存在长期记忆中

   Args:
       name: 用户名

   Returns:
       str: 保存状态
   """
   runtime.store.put(("users",), runtime.state["user_id"], {"name": name})
   return "saved"


@tool(parse_docstring=True)
def get_user_info(runtime: ToolRuntime) -> str:
   """
   从长期记忆中读取用户信息

   Returns:
       str: 用户信息
   """
   item = runtime.store.get(("users",), runtime.state["user_id"])
   return str(item.value) if item else "unknown"


with PostgresStore.from_conn_string(DB_URI) as store:
   store.setup()
   agent = create_agent(
       model=model,
       tools=[save_user_info, get_user_info],
       store=store,
       system_prompt="用户提及个人信息时及时记录，用户询问个人信息时尝试用工具检索",
       state_schema=CustomState,
   )
   print("=" * 30, '-> 第一个会话（线程） <-', "=" * 30)
   response1 = agent.invoke({
       "messages": "你好，很高兴认识你，我是小花",
       "user_id": "user-1"
   })
   for msg in response1["messages"]:
       msg.pretty_print()
   print("=" * 30, '-> 第二个会话（线程） <-', "=" * 30)
   response2 = agent.invoke({
       "messages": "我是谁",
       "user_id": "user-1"
   })
   for msg in response2["messages"]:
       msg.pretty_print()

2.2.2 在中间件中访问长期记忆

① Node-style hooks中访问 以 before_model 为例

python
@before_model
def before_model_middleware(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
   item = runtime.store.get(("users",), runtime.state["user_id"])
   return None

② Wrap-style hooks中访问

python
@wrap_tool_call
def wrap_tool_call_middleware(
       request: ToolCallRequest,
       handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command:
   request.runtime.store.get(("users",), runtime.state["user_id"])
   
   return None