# 记忆

[记忆](https://docs.langchain.com/oss/python/langgraph/add-memory)（Memory）是一个可选模块。如非必要，你无需向智能体添加 Memory 模块。因为 StateGraph 本身就含有历史消息列表 `messages`，足以满足最基础的“记忆”需求。

需要添加 Memory 模块的情况包括：

1. 历史消息太多，需要用外部工具存储记忆
1. 触发人工干预（[interrupt](https://docs.langchain.com/oss/python/langgraph/interrupts)），需要临时保存 Agent 的状态
1. 跨对话提取用户偏好 等等

LangGraph 将记忆分为：

- [短期记忆](https://docs.langchain.com/oss/python/langchain/short-term-memory)（MemorySaver）
- [长期记忆](https://docs.langchain.com/oss/python/langchain/long-term-memory)（MemoryStore）

此外，还有一个 [LangMem](https://langchain-ai.github.io/langmem/) 也提供记忆存取功能。

```python
import os
import sqlite3

from dotenv import load_dotenv
from dataclasses import dataclass
from typing_extensions import TypedDict
from openai import OpenAI
from langchain_openai import ChatOpenAI
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

# 加载模型配置
_ = load_dotenv()

# 加载模型
model = ChatOpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
    model="qwen3-coder-plus",
    temperature=0.7,
)

# 创建助手节点
def assistant(state: MessagesState):
    return {'messages': [model.invoke(state['messages'])]}
```

## 一、短期记忆

短期记忆（工作记忆）一般用于临时存储 Agent 或 工作流 的状态，以便在失败或重试后恢复。

### 1）在工作流中使用短期记忆

如果为工作流配置了检查点，下次调用该工作流时，会接着上一次对话内容继续聊下去。如果没有配置，将不会保留历史对话。

```python
# 创建短期记忆
checkpointer = InMemorySaver()

# 创建图
builder = StateGraph(MessagesState)

# 添加节点
builder.add_node('assistant', assistant)

# 添加边
builder.add_edge(START, 'assistant')
builder.add_edge('assistant', END)

# 使用检查点
graph = builder.compile(checkpointer=checkpointer)

## 如果不使用检查点，看看会发生什么？ 
# graph = builder.compile()

# 告诉智能体我是谁
result = graph.invoke(
    {'messages': ['你好！我是派大星']},
    {"configurable": {"thread_id": "1"}},
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

你好！我是派大星
================================== Ai Message ==================================

你好，派大星！很高兴见到你！今天过得怎么样？有什么我可以帮助你的吗？🌟
```

```python
# 让智能体说出我的名字
result = graph.invoke(
    {"messages": [{"role": "user", "content": "请问我是谁？"}]},
    {"configurable": {"thread_id": "1"}},  
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

你好！我是派大星
================================== Ai Message ==================================

你好，派大星！很高兴见到你！今天过得怎么样？有什么我可以帮助你的吗？🌟
================================ Human Message =================================

请问我是谁？
================================== Ai Message ==================================

你好！根据你刚才告诉我的信息，你是派大星！不过我注意到你用了"请问我是谁？"这样的问法，让我有点好奇 - 你是想确认自己的身份，还是在和我开玩笑呢？😊

如果你真的是派大星，那真是太有趣了！我很好奇今天在比奇堡发生了什么有趣的事情？
```

### 2）在智能体中使用短期记忆

在智能体中使用短期记忆的效果，和工作流中类似。

```python
from langchain.agents import create_agent

# 创建短期记忆
checkpointer = InMemorySaver()

agent = create_agent(
    model=model,
    checkpointer=checkpointer
)

# 告诉智能体我是章鱼哥
result = agent.invoke(
    {'messages': ['哈喽！我是章鱼哥']},
    {"configurable": {"thread_id": "2"}},
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

哈喽！我是章鱼哥
================================== Ai Message ==================================

你好，章鱼哥！很高兴认识你！今天过得怎么样？有什么我可以帮助你的吗？😊
```

```python
# 让智能体说出我的名字
result = agent.invoke(
    {"messages": [{"role": "user", "content": "我是谁？"}]},
    {"configurable": {"thread_id": "2"}},  
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

哈喽！我是章鱼哥
================================== Ai Message ==================================

你好，章鱼哥！很高兴认识你！今天过得怎么样？有什么我可以帮助你的吗？😊
================================ Human Message =================================

我是谁？
================================== Ai Message ==================================

根据我们刚才的对话，你是章鱼哥！不过如果你是想让我猜猜你的真实身份或者有什么特别的故事，我可能需要更多线索才能知道哦。你有什么想告诉我的吗？
```

为了验证 InMemorySaver 是否真的有效，可以将 checkpointer 注释后，再观察智能体的行为。

### 3）使用数据库保存短期记忆

如果用 SQLite 保存工作状态，即使退出程序，应该也能恢复退出之前的状态。下面我们来验证这一点。在此之前，需要安装一个 Python 包以支持 SqliteSaver 检查点：

```bash
pip install langgraph-checkpoint-sqlite
```

```python
# 删除SQLite数据库
if os.path.exists("short-memory.db"):
    os.remove("short-memory.db")
```

```python
from langgraph.checkpoint.sqlite import SqliteSaver

# 创建sqlite支持的短期记忆
checkpointer = SqliteSaver(
    sqlite3.connect("short-memory.db", check_same_thread=False)
)

# 创建Agent
agent = create_agent(
    model=model,
    checkpointer=checkpointer,
)

# 告诉智能体我是沙悟净
result = agent.invoke(
    {'messages': ['嗨！我是沙悟净']},
    {"configurable": {"thread_id": "3"}},
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

嗨！我是沙悟净
================================== Ai Message ==================================

你好，沙悟净！很高兴见到你。你是《西游记》中的重要角色，唐僧的第三个徒弟。听说你在保护唐僧西天取经的路上一直都很忠诚可靠。

有什么我可以帮助你的吗？或者你想聊聊关于取经路上的故事？
```

创建一个新的 Agent，并为它配置 SQLite 检查点。看看智能体能否从 SQLite 中读取关于我名字的记忆。

```python
# 创建一个新的Agent
new_agent = create_agent(
    model=model,
    checkpointer=checkpointer,
)

# 让智能体回忆我的名字
result = new_agent.invoke(
    {'messages': ['我是谁？']},
    {"configurable": {"thread_id": "3"}},
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

嗨！我是沙悟净
================================== Ai Message ==================================

你好，沙悟净！很高兴见到你。你是《西游记》中的重要角色，唐僧的第三个徒弟。听说你在保护唐僧西天取经的路上一直都很忠诚可靠。

有什么我可以帮助你的吗？或者你想聊聊关于取经路上的故事？
================================ Human Message =================================

我是谁？
================================== Ai Message ==================================

根据你之前的介绍，你是沙悟净，也就是《西游记》中的沙僧。你是唐僧收的第三个徒弟，在取经团队中以忠诚、稳重著称，经常挑着行李跟随师父和师兄们一起前往西天求取真经。

不过如果你是想让我通过其他方式来"猜"你是谁，那我可能需要更多线索呢！😊
```

## 二、长期记忆

长期记忆一般用于保存与业务相关的重要信息，比如用户属性、流量参数等。

### 1）创建 Embedding 生成函数

长期记忆支持使用 Embedding 检索语义相近的内容。下面创建一个 Embedding 生成函数，该函数可生成检索所需的 Embedding。

```python
# 嵌入维度
EMBED_DIM = 1024

# 获取text embedding的接口
client = OpenAI(
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_BASE_URL"),
)

# embedding生成函数
def embed(texts: list[str]) -> list[list[float]]:
    response = client.embeddings.create(
        model="text-embedding-v4",
        input=texts,
        dimensions=EMBED_DIM,
    )
    return [item.embedding for item in response.data]

# 测试能否正常生成text embedding
texts = [
    "LangGraph的中间件非常强大",
    "LangGraph的MCP也很好用",
]
vectors = embed(texts)

len(vectors), len(vectors[0])
```

```
(2, 1024)
```

### 2）读写长期记忆

先向 InMemoryStore 中写入两条数据。

```python
# 创建 InMemoryStore 内存存储
store = InMemoryStore(index={"embed": embed, "dims": EMBED_DIM})

# 添加两条用户数据 user_1 user_2
namespace = ("users", )

store.put(
    namespace,  # Namespace to group related data together
    "user_1",  # Key within the namespace
    {
        "rules": [
            "User likes short, direct language",
            "User only speaks English & python",
        ],
        "rule_id": "3",
    },
)

store.put(
    ("users",),
    "user_2",
    {
        "name": "John Smith",
        "language": "English",
    }
)
```

通过 namespace 和 key，可以直接读取长期记忆。

```python
item = store.get(namespace, "user_2")
item
```

```
Item(namespace=['users'], key='user_2', value={'name': 'John Smith', 'language': 'English'}, created_at='2025-12-18T17:43:50.154135+00:00', updated_at='2025-12-18T17:43:50.154138+00:00')
```

也可以通过向量检索召回。

```python
items = store.search( 
    namespace,
    query="language preferences",
    filter={"rule_id": "3"},
)
items
```

```
[Item(namespace=['users'], key='user_1', value={'rules': ['User likes short, direct language', 'User only speaks English & python'], 'rule_id': '3'}, created_at='2025-12-18T17:43:50.047616+00:00', updated_at='2025-12-18T17:43:50.047617+00:00', score=0.4085710154661828)]
```

### 3）使用工具读取长期记忆

```python
@dataclass
class Context:
    user_id: str

@tool
def get_user_info(runtime: ToolRuntime[Context]) -> str:
    """用于查询用户信息"""
    user_id = runtime.context.user_id
    user_info = runtime.store.get(("users",), user_id) 
    return str(user_info.value) if user_info else "未知用户"

# 创建Agent
agent = create_agent(
    model=model,
    tools=[get_user_info],
    store=store, 
    context_schema=Context
)

# 运行Agent
result = agent.invoke(
    {"messages": [{"role": "user", "content": "查阅用户信息"}]},
    context=Context(user_id="user_2") 
)

for message in result['messages']:
    message.pretty_print()
```

```
================================ Human Message =================================

查阅用户信息
================================== Ai Message ==================================
Tool Calls:
  get_user_info (call_f54a533c43f84fadab9f20b3)
 Call ID: call_f54a533c43f84fadab9f20b3
  Args:
================================= Tool Message =================================
Name: get_user_info

{'name': 'John Smith', 'language': 'English'}
================================== Ai Message ==================================

用户的姓名是 John Smith，使用的语言是 English。
```

### 4）使用工具写入长期记忆

```python
class UserInfo(TypedDict):
    name: str

@tool
def save_user_info(user_info: UserInfo, runtime: ToolRuntime[Context]) -> str:
    """用于保存/更新用户信息"""
    user_id = runtime.context.user_id
    runtime.store.put(("users",), user_id, user_info) 
    return "成功保存用户信息"

# 创建gent
agent = create_agent(
    model=model,
    tools=[save_user_info],
    store=store,
    context_schema=Context
)

# 运行Agent
agent.invoke(
    {"messages": [{"role": "user", "content": "My name is John Smith"}]},
    context=Context(user_id="user_123") 
)

store.get(("users",), "user_123").value
```

```
{'name': 'John Smith'}
```