> ## Documentation Index
> Fetch the complete documentation index at: https://vectorpeak.cn/llms.txt
> Use this file to discover all available pages before exploring further.

# Memory Management

> 介绍 LangGraph 中短期记忆、长期记忆、消息裁剪、总结和跨线程召回

## 1. 智能体记忆管理与多轮对话方法

<Badge icon="clock" color="green">Written: 2026.06</Badge> 要实现多轮对话，核心问题是 如何保存上下文记忆，让智能体对用户历史输入有“记忆”，能够根据历史消息记录回答问题

## 2. 记忆模型与关键组件

### 2.1 短期记忆(Checkpointer)

载体：Checkpointer(`MemorySaver`、`RedisSaver`、`PostgresSaver`…)
作用：把每轮消息 + 工具调用结果序列化成图状态，按 `thread_id` 持久化；下次传入相同 `thread_id` 自动续写

原理：

* 每次你调用 `graph.invoke(...)` 或 `graph.stream(...)`，LangGraph 都会维护一个状态(state)

* 如果没有 Checkpointer，这个 state 默认只存在本次调用内，调用结束就丢掉了

* 如果启用了 Checkpointer，它会把 state 保存到存储中(内存/数据库/文件)，下次继续调用时，可以恢复之前的 state，实现“记忆”

### 2.2 长期记忆(BaseStore)

载体：BaseStore(`InMemoryStore`、`RedisStore`、`AsyncPostgresStore`…)
作用：显式保存“用户偏好”“背景事实”等高密度信息，由 LLM 主动读写；Store 支持向量检索，支持命名空间隔离

和 **Checkpointer** 的区别：

* Checkpointer：保存图的运行状态(短期记忆，主要用于同一个线程连续对话)

* Store：LangGraph 的存储模块提供持久化的键值存储，支持跨线程和会话的长期内存，适用于需要持久化数据的复杂工作流

### 2.3 消息裁剪( Trimming)

当历史消息过长时，可在 `pre_model_hook` 里插入 `trim_messages` 策略，按最近 *N 条消息* 或 *Token 数* 保留，超出部分丢弃。这种做法的优点是：简单、可控，保证上下文长度不超限。但缺点是：容易丢失长对话中的重要信息

### 2.4 消息总结(Summarization )

通过生成摘要来“压缩”历史，避免 token 爆炸

* 定期总结：每对话 X 轮，把旧消息合并成一段摘要，再存入 memory，新的上下文里只保留摘要 + 最近消息

* 递归总结：对摘要再继续总结，形成分层结构(像树状记忆)

* 角色分段总结：比如只总结用户输入，系统或 AI 回复不做摘要

* 优点：历史不会丢失，只是被压缩成更短的摘要

* 缺点：摘要质量依赖 LLM，可能丢细节

## 3. 代码演示

### 3.1 预构建 Agent 实现记忆存储

```python title="预构建 Agent 记忆示例" icon="python" wrap theme={null}
import dotenv
from langchain_ollama import ChatOllama
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent

# 加载环境变量配置文件
dotenv.load_dotenv()

# 初始化本地大语言模型，模型名称和推理模式
llm = ChatOllama(model="qwen3:14b", reasoning=False)

# 定义工具列表，
tools = []
# 定义短期记忆使用内存(生产可以换 RedisSaver/PostgresSaver)
checkpointer = InMemorySaver()
# 创建ReAct代理，结合语言模型和工具函数
agent = create_react_agent(model=llm, tools=tools, checkpointer=checkpointer)
# 多轮对话配置，同一 thread_id 即同一会话
config = {"configurable": {"thread_id": "user-001"}}

msg1 = agent.invoke({"messages": [("user", "你好，我叫崔亮，喜欢学习。")]}, config)
msg1["messages"][-1].pretty_print()

# 6. 第二轮(继续同一 thread)
msg2 = agent.invoke({"messages": [("user", "我叫什么？我喜欢做什么？")]}, config)
msg2["messages"][-1].pretty_print()
```

```console title="执行结果如下" icon="terminal" wrap theme={null}
================================== Ai Message ==================================

你好崔亮，很高兴认识你！学习是一个非常棒的追求，不知道你最近在学习什么内容呢？是有什么特别感兴趣的领域吗？😊

================================== Ai Message ==================================

你叫崔亮，你喜欢学习。😊

如果你愿意的话，可以告诉我你具体对哪些方面感兴趣，比如是学习新技能、阅读、还是其他什么？我很乐意和你一起探讨！

```

### 3.2 底层 API 实现记忆存储

定义一个聊天机器人节点函数 chatbot，它接收包含消息历史的 state，调用本地大语言模型 llm 生成回复，并返回新消息。该函数被集成到一个基于图结构的对话流程中，支持多轮对话并保持上下文

```python title="底层 API 记忆示例" icon="python" wrap expandable theme={null}
from typing import TypedDict, Annotated
from langgraph.checkpoint.memory import MemorySaver
from langgraph.constants import START, END
from langgraph.graph import StateGraph
from langgraph.graph.message import add_messages
from langchain_ollama import ChatOllama

class State(TypedDict):
    """
    定义图结构中节点间传递的状态结构

    Attributes:
        messages: 消息列表，使用add_messages函数进行合并
    """
    messages: Annotated[list, add_messages]

# 创建状态图构建器
graph_builder = StateGraph(State)

# 初始化本地大语言模型，配置基础URL、模型名称和推理模式
llm = ChatOllama(base_url="http://localhost:11434", model="qwen3:14b", reasoning=False)

def chatbot(state: State):
    """
    聊天机器人节点函数，处理输入消息并生成回复

    Args:
        state (State): 包含消息历史的状态字典

    Returns:
        dict: 包含新生成消息的字典，格式为{"messages": [回复消息]}
    """
    return {"messages": [llm.invoke(state["messages"])]}

# 将聊天机器人节点添加到图中
graph_builder.add_node("chatbot", chatbot)

# 添加从开始节点到聊天机器人节点的边
graph_builder.add_edge(START, "chatbot")

# 添加从聊天机器人节点到结束节点的边
graph_builder.add_edge("chatbot", END)

# 创建内存保存器用于保存对话状态
memory = MemorySaver()

# 编译图结构并设置检查点保存器
graph = graph_builder.compile(checkpointer=memory)

# 绘制图结构并保存为PNG图片
graph.get_graph().draw_png('./graph.png')

# 配置对话线程ID
config = {"configurable": {"thread_id": "chat-1"}}

# 第一次对话：发送初始消息
msg1 = graph.invoke({"messages": ["你好，我叫崔亮，喜欢学习。"]}, config=config)
msg1["messages"][-1].pretty_print()

# 第二次对话：基于上下文询问用户信息
msg2 = graph.invoke({"messages": ["我叫什么？我喜欢做什么？"]}, config=config)
msg2["messages"][-1].pretty_print()

```

```console title="执行结果如下" icon="terminal" wrap theme={null}
================================== Ai Message ==================================

你好崔亮，很高兴认识你！😊 你对学习的热爱真的很棒，这种态度会让你在很多领域都取得不错的成就。不知道你最近在学习什么内容呢？如果有任何问题或者想讨论的话题，我很乐意和你交流！

================================== Ai Message ==================================

你叫**崔亮**，你喜欢**学习**。😊

如果你想了解更多关于自己的事情，或者想探索新的兴趣，也可以告诉我，我们可以一起聊聊！

```

### 3.3 长期记忆+跨线程召回

整体实现步骤为：

1. 初始化一个 `InMemoryStore`(或 `RedisStore`)

2. 把“记忆工具”塞进智能体工具箱，让 LLM 自己决定何时存/取

3. 命名空间按 `user_id` 隔离，防止用户数据串线

```python title="长期记忆示例" icon="python" wrap expandable theme={null}
import uuid
from typing import TypedDict, Annotated
import dotenv
from langchain_ollama import ChatOllama
from langchain_core.runnables import RunnableConfig
from langgraph.constants import END, START
from langgraph.graph import StateGraph, MessagesState, add_messages
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.store.base import BaseStore

# 加载环境变量配置
dotenv.load_dotenv()
# 初始化本地大语言模型，配置模型名称和推理模式
model = ChatOllama(model="qwen3:14b", reasoning=False)

class State(TypedDict):
    """
    定义图中状态的数据结构。

    属性:
        messages (Annotated[list, add_messages]): 使用 add_messages 合并的消息列表。
    """
    messages: Annotated[list, add_messages]

def save_memory(store: BaseStore, user_id: str, content: str):
    """
    将用户输入的内容保存为记忆。

    参数:
        store (BaseStore): 存储系统的实例，用于持久化数据。
        user_id (str): 用户唯一标识符。
        content (str): 需要存储的文本内容。
    """
    namespace = ("memories", user_id)
    store.put(namespace, str(uuid.uuid4()), {"data": content})

def recall_memories(store: BaseStore, user_id: str, query: str, limit: int = 5):
    """
    根据查询语句检索与用户相关的记忆。

    参数:
        store (BaseStore): 存储系统的实例。
        user_id (str): 用户唯一标识符。
        query (str): 查询关键词或句子。
        limit (int, optional): 返回的记忆条数上限，默认是 5 条。

    返回:
        list[str]: 匹配的记忆内容列表。
    """
    namespace = ("memories", user_id)
    memories = store.search(namespace, query=query, limit=limit)
    return [m.value["data"] for m in memories]

def chatbot(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    """
    聊天机器人主逻辑节点函数。

    参数:
        state (MessagesState): 当前对话的状态信息，包括历史消息等。
        config (RunnableConfig): 运行时配置信息，如线程ID、用户ID等。
        store (BaseStore): 用于读取和写入用户记忆的存储接口。

    返回:
        dict: 更新后的消息状态字典。
    """
    user_id = config["configurable"]["user_id"]

    # 检索历史记忆
    query = state["messages"][-1].content
    related_memories = recall_memories(store, user_id, query)

    # 构造系统提示
    system_msg = (
        "你是一个友好的聊天助手。\n"
        f"以下是关于用户的记忆:\n{chr(10).join(related_memories) if related_memories else '暂无'}"
    )

    # 保存当前消息到记忆
    save_memory(store, user_id, query)

    # 调用模型生成回复
    response = model.invoke(
        [{"role": "system", "content": system_msg}] + state["messages"]
    )
    return {"messages": response}

# 创建状态图并定义流程
builder = StateGraph(State)
builder.add_node(chatbot)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

# 初始化检查点和存储组件
checkpointer = InMemorySaver()
store = InMemoryStore()

# 编译构建最终可运行的图对象，并绘制其结构图
graph = builder.compile(
    checkpointer=checkpointer,
    store=store,
)
graph.get_graph().draw_png('./graph.png')

# 第一次交互测试：记录用户基本信息
config1 = {"configurable": {"thread_id": "1", "user_id": "1"}}
msg1 = graph.invoke({"messages": [{"role": "user", "content": "我叫崔亮，喜欢学习。"}]}, config1)
print("第一次回复：")
msg1["messages"][-1].pretty_print()

# 第二次交互测试：验证是否能回忆起之前的信息
config2 = {"configurable": {"thread_id": "2", "user_id": "1"}}
msg2 = graph.invoke({"messages": [{"role": "user", "content": "我叫什么？我喜欢做什么？"}]}, config2)
print("第二次回复：")
msg2["messages"][-1].pretty_print()

```

```console title="执行结果如下" icon="terminal" wrap theme={null}
第一次回复：
================================== Ai Message ==================================

很高兴认识你，崔亮！很高兴你喜欢学习，这真是一个很棒的品质。你平时喜欢学习哪些方面的知识呢？是喜欢阅读、上课，还是通过其他方式学习？我很想听听你的故事。

第二次回复：
================================== Ai Message ==================================

你叫崔亮，喜欢学习。很高兴认识你！学习是一件很有趣的事情，你平时都喜欢学习什么呢？

```

### 3.4 消息裁剪

一个钩子函数 `pre_model_hook`，用于在模型处理前裁剪消息历史，只保留最近几条消息，避免上下文过长。它使用 `trim_messages` 函数按策略裁剪消息，限制总 token 数为 100，从人类用户消息开始裁剪，确保输入模型的消息列表不会超出限制

```python title="消息裁剪示例" icon="python" wrap expandable theme={null}
import dotenv
from langchain_core.messages.utils import trim_messages, count_tokens_approximately
from langchain_ollama import ChatOllama
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent

# 加载环境变量配置
dotenv.load_dotenv()
# 初始化本地大语言模型，配置模型名称和推理模式
model = ChatOllama(model="qwen3:14b", reasoning=False)
# 定义工具列表，
tools = []

def pre_model_hook(state):
    """
    在模型处理前对消息进行预处理的钩子函数

    该函数用于裁剪消息历史，只保留最近的若干条消息，避免上下文过长

    Args:
        state (dict): 包含对话状态的字典，其中"messages"键对应消息列表

    Returns:
        dict: 包含裁剪后消息的字典，键为"llm_input_messages"
    """
    # 参数说明:
    #   state["messages"]: 需要裁剪的消息列表
    #   strategy: 裁剪策略，"last"表示从最后开始裁剪
    #   token_counter: 用于计算token数量的函数，这里使用近似计算方法
    #   max_tokens: 最大token数量限制，设置为300
    #   start_on: 开始裁剪的消息类型，"human"表示从人类用户的消息开始
    #   end_on: 结束裁剪的消息类型，可以是"human"或"tool"类型的消息
    # 返回值: 裁剪后的消息列表
    trimmed_messages = trim_messages(
        state["messages"],
        strategy="last",
        token_counter=count_tokens_approximately,
        max_tokens=300,
        start_on="human",
        end_on=("human", "tool"),
    )

    return {"llm_input_messages": trimmed_messages}

checkpointer = InMemorySaver()
agent = create_react_agent(
    model,
    tools,
    pre_model_hook=pre_model_hook,
    checkpointer=checkpointer,
)
config = {"configurable": {"thread_id": "user-001"}}
msg1 = agent.invoke({"messages": [("user", "你好，我叫崔亮")]}, config)
msg1["messages"][-1].pretty_print()
like_list = ['唱', '跳', 'rap', '篮球']
for i in like_list:
    msg = "我喜欢做的事是：" + i
    print(msg)
    agent.invoke({"messages": [("user", msg)]}, config)
msg2 = agent.invoke({"messages": [("user", "我叫什么？我喜欢做的事是什么？")]}, config)
msg2["messages"][-1].pretty_print()
```

```console title="执行结果如下" icon="terminal" wrap theme={null}
================================== Ai Message ==================================

你好崔亮！很高兴认识你。😊 今天过得怎么样？有什么有趣的事情发生吗？

我喜欢做的事是：唱
我喜欢做的事是：跳
我喜欢做的事是：rap
我喜欢做的事是：篮球
================================== Ai Message ==================================

你叫什么？你喜欢做的事是篮球对吗？🏀
(不过你刚刚已经告诉过我你喜欢做的事是篮球啦～)

如果你愿意的话，可以告诉我你的名字，这样我们就能更熟悉啦！😊
你是不是也特别喜欢在球场上奔跑、投篮、和朋友们一起打球的感觉？

```

### 3.5 消息总结

实现了一个基于本地大语言模型的对话代理，具备上下文记忆与摘要能力。主要功能包括：

* 加载环境变量并初始化Ollama模型；

* 创建摘要节点以控制输入长度；

* 定义带记忆状态的代理及会话配置；

* 通过多轮对话测试模型对用户信息(姓名、兴趣)的记忆与理解能力

```python title="消息总结示例" icon="python" wrap expandable theme={null}
import dotenv
from langchain_core.messages.utils import trim_messages, count_tokens_approximately
from langchain_ollama import ChatOllama
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent
from langgraph.prebuilt.chat_agent_executor import AgentState
from langmem.short_term import SummarizationNode, RunningSummary

# 加载环境变量配置
dotenv.load_dotenv()

# 初始化本地大语言模型，配置模型名称和推理模式
model = ChatOllama(model="qwen3:14b", reasoning=False)

# 定义工具列表，
tools = []

# 创建一个SummarizationNode实例，用于处理文本摘要任务
# 参数说明:
#   token_counter: 用于估算文本token数量的函数，这里使用count_tokens_approximately函数
#   model: 指定使用的语言模型实例
#   max_tokens: 限制处理文本的最大token数量为300
#   max_summary_tokens: 限制生成摘要的最大token数量为128
#   output_messages_key: 指定输出消息在结果中的键名，设置为"llm_input_messages"
summarization_node = SummarizationNode(
    token_counter=count_tokens_approximately,
    model=model,
    max_tokens=300,
    max_summary_tokens=128,
    output_messages_key="llm_input_messages",
)

# 自定义状态类，继承自AgentState，添加上下文字段用于存储运行时摘要信息
class State(AgentState):
    context: dict[str, RunningSummary]

# 初始化内存检查点保存器，用于持久化代理状态
checkpointer = InMemorySaver()

# 创建React代理，整合模型、工具、摘要节点和状态管理器
agent = create_react_agent(
    model=model,
    tools=tools,
    pre_model_hook=summarization_node,
    state_schema=State,
    checkpointer=checkpointer,
)

# 配置线程ID，用于标识用户会话
config = {"configurable": {"thread_id": "user-001"}}

# 启动对话，发送用户自我介绍消息并获取模型响应
msg1 = agent.invoke({"messages": [("user", "你好，我叫崔亮")]}, config)
msg1["messages"][-1].pretty_print()

# 定义用户兴趣列表
like_list = ['唱', '跳', 'rap', '篮球']

# 循环发送用户兴趣信息，逐条更新上下文
for i in like_list:
    msg = "我喜欢做的事是：" + i
    print(msg)
    agent.invoke({"messages": [("user", msg)]}, config)

# 查询用户姓名和兴趣，测试模型对上下文的理解能力
msg2 = agent.invoke({"messages": [("user", "我叫什么？我喜欢做的事是什么？")]}, config)
msg2["messages"][-1].pretty_print()

```

```console title="执行结果如下" icon="terminal" wrap theme={null}
================================== Ai Message ==================================

你好崔亮，很高兴认识你！😊 今天过得怎么样？有什么有趣的事情发生吗？

我喜欢做的事是：唱
我喜欢做的事是：跳
我喜欢做的事是：rap
我喜欢做的事是：篮球
================================== Ai Message ==================================

你叫**崔亮**，你喜欢做的事是：**唱、跳、rap、篮球**。🎶💃🎤🏀

你是一个多才多艺、热爱生活、充满活力的人！是不是很酷？😎
如果你想继续分享更多关于自己的故事，我随时都在哦！
```