记忆系统
更新: 4/9/2026 字数: 0 字 时长: 0 分钟
本章详细介绍 LangChain Agent 的记忆系统,包括 Checkpointer、消息历史、向量记忆等。
记忆系统概述
为什么需要记忆?
Without Memory With Memory
═══════════════════════════════════════════════════════════
用户: 你好 用户: 你好
助手: 你好! 助手: 你好!张三
用户: 我叫张三 (Agent 记住了)
用户: 我叫什么? 用户: 我叫什么?
助手: 抱歉,我不知道 助手: 你叫张三!
═══════════════════════════════════════════════════════════记忆类型
Memory System Architecture
────────────────────────────────────────────────────────────
┌─────────────────────────────────────────────────────────────┐
│ Memory System │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Short-term Memory (STM) │ │
│ │ 用途:当前对话的上下文 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Long-term Memory (LTM) │ │
│ │ 用途:跨会话的持久化信息 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘Checkpointer
Checkpointer 是什么?
python
# Checkpointer = 状态持久化器
# 自动保存和恢复 Agent 状态
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
agent = create_agent(
model="...",
tools=[...],
checkpointer=checkpointer
)多种 Checkpointer
python
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.checkpoint.redis import RedisSaver
# 1. 内存存储(开发用)
memory_checkpointer = InMemorySaver()
# 2. SQLite(轻量级)
sqlite_checkpointer = SqliteSaver.from_conn_string("checkpoints.db")
# 3. PostgreSQL(生产用)
postgres_checkpointer = PostgresSaver.from_conn_string(
"postgresql://user:pass@host:5432/db"
)
# 4. Redis(高并发)
redis_checkpointer = RedisSaver.from_conn_string("redis://localhost:6379")会话管理
thread_id 管理
python
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
checkpointer = InMemorySaver()
agent = create_agent(
model="anthropic:claude-sonnet-4-6",
tools=[...],
checkpointer=checkpointer
)
# 单用户单会话
config = {"configurable": {"thread_id": "alice_001"}}
# 多用户隔离
configs = {
"alice": {"configurable": {"thread_id": "alice_session"}},
"bob": {"configurable": {"thread_id": "bob_session"}},
}向量记忆
向量检索记忆
python
from langchain.memory import VectorStoreRetrieverMemory
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
# 1. 创建向量存储
vectorstore = Chroma(
embedding_function=OpenAIEmbeddings(),
persist_directory="memory_db"
)
# 2. 创建向量记忆
memory = VectorStoreRetrieverMemory(
vectorstore=vectorstore,
memory_key="chat_history",
return_messages=True,
k=3
)
# 3. 保存记忆
memory.save_context(
{"input": "我叫张三,喜欢编程"},
{"output": "好的,张三,我记住了"}
)
# 4. 检索记忆
related = memory.load_memory_variables({"input": "我是谁?"})生产环境配置
PostgreSQL Checkpointer
bash
# 安装依赖
pip install psycopg2-binarypython
from langgraph.checkpoint.postgres import PostgresSaver
pg_checkpointer = PostgresSaver.from_conn_string(
"postgresql://user:password@host:5432/langchain"
)
pg_checkpointer.setup()配置对比
| 存储 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| Memory | 开发/测试 | 简单、快速 | 不持久化 |
| SQLite | 小规模/单机 | 零配置 | 并发差 |
| PostgreSQL | 生产环境 | 稳定、可扩展 | 需要运维 |
| Redis | 高并发 | 极快、支持TTL | 内存成本 |
本章小结
记忆系统要点
────────────────────────────────────────────────────────
✅ Checkpointer = 状态持久化
✅ InMemorySaver = 开发用(内存)
✅ PostgresSaver = 生产用(数据库)
✅ thread_id = 会话隔离
记忆层次
────────────────────────────────────────────────────────
1. Working Memory - 当前执行状态
2. Short-term Memory - 消息历史
3. Long-term Memory - Checkpointer
4. Vector Memory - 语义检索