记忆（Memory）是一个可选模块。如非必要，你无需向智能体添加 Memory 模块。因为 StateGraph 本身就含有历史消息列表 messages，足以满足最基础的“记忆”需求。
需要添加 Memory 模块的情况包括：
历史消息太多，需要用外部工具存储记忆
触发人工干预（interrupt），需要临时保存 Agent 的状态
跨对话提取用户偏好 等等
LangGraph 将记忆分为：
此外，还有一个 LangMem 也提供记忆存取功能。
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）在工作流中使用短期记忆¶
如果为工作流配置了检查点，下次调用该工作流时，会接着上一次对话内容继续聊下去。如果没有配置，将不会保留历史对话。
# 创建短期记忆
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 ==================================
你好，派大星！很高兴见到你！今天过得怎么样？有什么我可以帮助你的吗？🌟
# 让智能体说出我的名字
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）在智能体中使用短期记忆¶
在智能体中使用短期记忆的效果，和工作流中类似。
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 ==================================
你好，章鱼哥！很高兴认识你！今天过得怎么样？有什么我可以帮助你的吗？😊
# 让智能体说出我的名字
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 检查点：
pip install langgraph-checkpoint-sqlite# 删除SQLite数据库
if os.path.exists("short-memory.db"):
os.remove("short-memory.db")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 中读取关于我名字的记忆。
# 创建一个新的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。
# 嵌入维度
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 中写入两条数据。
# 创建 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，可以直接读取长期记忆。
item = store.get(namespace, "user_2")
itemItem(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')也可以通过向量检索召回。
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）使用工具读取长期记忆¶
@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）使用工具写入长期记忆¶
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'}