Memory 记忆
Memory（记忆）是 LangChain 中为 Agent 和链（Chain）提供状态持久化和上下文记忆的组件。它让 AI 应用能够记住之前的对话内容，实现连续、有上下文的交互体验。
概述
LangChain 的记忆系统主要分为两类：
- 短期记忆（Checkpointer）：保存对话历史和工作流状态
- 长期记忆（Store）：跨会话持久化知识和信息
短期记忆（checkpointer）
├── 同一个 thread_id 内的消息历史
├── 工作流节点状态
└── 对话上下文
长期记忆（store）
├── 跨会话的用户偏好
├── 累积的知识信息
└── 用户画像数据短期记忆：Checkpointer
Checkpointer（检查点）用于保存 Agent 的执行状态和消息历史。它通过 thread_id 将不同会话隔离。
InMemorySaver（内存保存器）
最简单的内存级 checkpointer：
python
from langgraph.checkpoint import InMemorySaver
from langgraph.prebuilt import create_agent
from langchain_openai import ChatOpenAI
# 创建 checkpointer
checkpointer = InMemorySaver()
# 创建带 checkpointer 的 agent
agent = create_agent(
model=ChatOpenAI(model="gpt-4o-mini"),
tools=[],
checkpointer=checkpointer
)
# 会话 1：第一次对话
result = agent.invoke(
{"messages": [("human", "你好，我叫小明")]},
config={"configurable": {"thread_id": "session-1"}}
)
print(result["messages"][-1].content)
# 助手：你好小明！很高兴认识你。
# 会话 1：第二次对话（记住名字）
result = agent.invoke(
{"messages": [("human", "我叫什么名字？")]},
config={"configurable": {"thread_id": "session-1"}}
)
print(result["messages"][-1].content)
# 助手：你叫小明！
# 会话 2：新会话（不记得）
result = agent.invoke(
{"messages": [("human", "我叫什么名字？")]},
config={"configurable": {"thread_id": "session-2"}}
)
print(result["messages"][-1].content)
# 助手：抱歉，我不知道你的名字。自定义 Checkpointer
python
from langgraph.checkpoint import BaseCheckpointSaver
from typing import Any, Dict, Optional, Tuple
class MyCheckpointer(BaseCheckpointSaver):
"""自定义 checkpointer（示例：文件存储）"""
def __init__(self):
self.checkpoints: Dict[str, Any] = {}
def get(self, thread_id: str) -> Optional[Any]:
return self.checkpoints.get(thread_id)
def put(self, thread_id: str, checkpoint: Any) -> None:
self.checkpoints[thread_id] = checkpoint
def list(self, limit: int = 10) -> list:
return list(self.checkpoints.values())[:limit]
# 使用自定义 checkpointer
my_checkpointer = MyCheckpointer()使用 thread_id 管理多会话
python
# 为每个用户创建独立的会话
def get_user_agent(user_id: str):
return create_agent(
model=ChatOpenAI(model="gpt-4o-mini"),
tools=tools,
checkpointer=InMemorySaver(),
system_prompt=f"你正在帮助用户 {user_id}，请提供个性化服务。"
)
# 不同用户的独立对话
user_1_result = agent.invoke(
{"messages": [("human", "我喜欢的颜色是蓝色")]},
config={"configurable": {"thread_id": "user-001"}}
)
user_2_result = agent.invoke(
{"messages": [("human", "我喜欢的颜色是红色")]},
config={"configurable": {"thread_id": "user-002"}}
)长期记忆：Store
Store 用于跨会话持久化信息，即使用户开始新的对话也能记住。
InMemoryStore（内存存储）
python
from langgraph.store import InMemoryStore
from langgraph.prebuilt import create_agent
# 创建长期记忆存储
store = InMemoryStore()
# 存储用户信息
store.put(
("users", "user-001"),
"preferences",
{
"name": "小明",
"age": 28,
"interests": ["编程", "AI"],
"language": "zh-CN"
}
)
# 读取存储的信息
prefs = store.get(("users", "user-001"), "preferences")
print(prefs.value) # {'name': '小明', ...}带长期记忆的 Agent
python
from langgraph.store import InMemoryStore
from langgraph.prebuilt import create_agent
from langchain_openai import ChatOpenAI
store = InMemoryStore()
# 存储用户知识
store.put(
("users", "alice"),
"knowledge",
{"company": "ABC 科技", "role": "AI 工程师"}
)
agent = create_agent(
model=ChatOpenAI(model="gpt-4o-mini"),
tools=tools,
checkpointer=InMemorySaver(),
store=store, # 同时使用 checkpointer 和 store
system_prompt="你是一个个性化助手。"
)
# 用户查询（Agent 可以访问 store 中的信息）
result = agent.invoke(
{"messages": [("human", "我在哪家公司工作？")]},
config={
"configurable": {"thread_id": "session-1", "user_id": "alice"}
}
)基于存储的个性化
python
from langgraph.store import InMemoryStore
from langgraph.prebuilt import create_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
store = InMemoryStore()
@tool
def get_user_profile() -> str:
"""获取当前用户的个人信息"""
# 在实际应用中，这里会从 store 中读取用户信息
user_id = "current_user" # 从上下文获取
profile = store.get(("users", user_id), "profile")
if profile:
return str(profile.value)
return "暂无个人信息"
@tool
def save_user_preference(key: str, value: str) -> str:
"""保存用户偏好设置"""
user_id = "current_user"
current = store.get(("users", user_id), "preferences")
prefs = current.value if current else {}
prefs[key] = value
store.put(("users", user_id), "preferences", prefs)
return f"已保存偏好: {key} = {value}"
tools = [get_user_profile, save_user_preference]
agent = create_agent(
model=ChatOpenAI(model="gpt-4o-mini"),
tools=tools,
checkpointer=InMemorySaver(),
store=store
)
# 用户保存偏好
result = agent.invoke({
"messages": [("human", "请帮我记住我的名字叫小明")]
})
print(result["messages"][-1].content)消息历史管理
有限消息历史
控制保留的消息数量，避免上下文过长：
python
from langgraph.prebuilt import create_agent
from langgraph.checkpoint import InMemorySaver
# 消息历史通过系统 prompt 和 agent 配置控制
agent = create_agent(
model=ChatOpenAI(model="gpt-4o-mini"),
tools=tools,
checkpointer=InMemorySaver(),
system_prompt="保留最近 10 轮对话。"
)清除消息历史
python
# 清除特定会话的历史
checkpointer = InMemorySaver()
# 实现清除方法（如有需要）
def clear_history(thread_id: str):
checkpointer.checkpoints.pop(thread_id, None)记忆系统架构
Agent 执行流程
═══════════════════════════════════
┌──────────────┐
│ 用户输入 │
└──────┬───────┘
│
┌─────────▼─────────┐
│ Checkpointer │
│ (加载历史状态) │
└─────────┬─────────┘
│
┌─────────▼─────────┐
│ Agent 执行 │
│ (LLM + Tools) │
└─────────┬─────────┘
│
┌─────────▼─────────┐
│ Checkpointer │
│ (保存新状态) │
└─────────┬─────────┘
│
┌─────────▼─────────┐
│ Store │
│ (跨会话持久化) │
└─────────┬─────────┘
│
┌──────▼──────┐
│ 用户输出 │
└─────────────┘完整示例：带记忆的个人助手
python
from langgraph.checkpoint import InMemorySaver
from langgraph.store import InMemoryStore
from langgraph.prebuilt import create_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
# 创建记忆系统
checkpointer = InMemorySaver()
store = InMemoryStore()
# 定义工具
@tool
def remember_fact(fact: str) -> str:
"""记住一个关于用户的事实"""
# 存储记忆
memories = store.get(("user", "default"), "facts")
facts = memories.value if memories else []
facts.append(fact)
store.put(("user", "default"), "facts", {"facts": facts})
return f"已记住: {fact}"
@tool
def recall_facts() -> str:
"""回忆记住的用户信息"""
memories = store.get(("user", "default"), "facts")
if memories and memories.value.get("facts"):
facts = memories.value["facts"]
return "我记得关于你的事情:\n" + "\n".join(f"- {f}" for f in facts)
return "我还没有关于你的记忆。"
tools = [remember_fact, recall_facts]
# 创建 Agent
agent = create_agent(
model=ChatOpenAI(model="gpt-4o-mini", temperature=0.7),
tools=tools,
checkpointer=checkpointer,
store=store,
system_prompt="你是一个贴心的个人助手。你可以记住用户告诉你的信息，"
"并在后续对话中使用这些信息来提供个性化帮助。"
)
# 交互示例
def chat(message: str, thread_id: str = "main"):
result = agent.invoke(
{"messages": [("human", message)]},
config={"configurable": {"thread_id": thread_id}}
)
return result["messages"][-1].content
# 场景演示
print("=== 第一次对话 ===")
print(f"用户: 我叫小明，我喜欢编程")
print(f"助手: {chat('我叫小明，我喜欢编程')}")
print(f"\n用户: 请记住我喜欢蓝色")
print(f"助手: {chat('请记住我喜欢蓝色')}")
print("\n=== 第二次对话（同一会话）===")
print(f"用户: 你知道我喜欢什么吗？")
print(f"助手: {chat('你知道我喜欢什么吗？')}")
print("\n=== 第三次对话（新会话，但长期记忆还在）===")
print(f"用户: 你还记得我吗？")
print(f"助手: {chat('你还记得我吗？', 'new-session')}")记忆策略选择
| 记忆类型 | 组件 | 作用范围 | 用途 |
|---|---|---|---|
| 短期记忆 | Checkpointer | 同一 thread_id | 对话历史、工作流状态 |
| 长期记忆 | Store | 跨会话 | 用户偏好、知识积累 |
| 无记忆 | 无 | 单次请求 | 简单问答、无状态 API |
最佳实践
- 使用 thread_id 隔离会话：每个用户或对话使用独立的 thread_id
- 合理配置 checkpointer：生产环境可配置数据库持久化（如 PostgreSQL）
- 区分短期和长期记忆：对话历史用 checkpointer，用户知识用 store
- 控制记忆大小：定期清理过期的对话历史
- 隐私保护：不要在 store 中存储敏感信息
下一步
- Streaming 流式输出：结合记忆实现流式对话
- Callbacks 与追踪：追踪记忆操作和 Agent 执行
- RAG 应用设计：将记忆集成到 RAG 系统中
- 测试与评估：测试记忆功能的正确性