Agent Memory(Agent 记忆架构)
AI Agent 的记忆系统决定了它能”记住”多少信息、多长时间、以何种方式检索。是实现长期任务、个性化和持续学习的核心机制。
记忆类型
Agent Memory
├── 短期记忆(In-Context)
│ └── 当前对话的 context window
└── 长期记忆(External Storage)
├── 语义记忆 → 向量数据库(概念、知识)
├── 情节记忆 → 键值存储(具体事件、对话片段)
├── 程序记忆 → 代码/工具库(技能、操作步骤)
└── 结构记忆 → 图数据库(实体关系网络)
| 类型 | 存储介质 | 内容 | 检索方式 |
|---|---|---|---|
| 短期(Working) | Context Window | 当前会话消息 | 直接读取 |
| 语义(Semantic) | 向量数据库 | 事实、知识、文档 | 相似度检索 |
| 情节(Episodic) | KV 存储 / 数据库 | 历史对话、用户偏好 | 精确查询 |
| 程序(Procedural) | 代码库 / 工具注册表 | 技能、函数、工作流 | 工具调用 |
| 结构(Structural) | 图数据库 | 实体关系、知识图谱 | 图查询 |
Agent + Memory 架构
用户输入
↓
┌─────────────────────────────────────┐
│ Agent Core │
│ │
│ 规划(Plan) → 工具调用(Act) │
│ ↑ ↓ │
│ 记忆读取 记忆写入 │
└────────┬────────────────┬───────────┘
│ │
┌────▼────┐ ┌─────▼──────┐
│ 短期记忆 │ │ 长期记忆 │
│ Context │ │ Vector DB │
│ Window │ │ KV Store │
└─────────┘ │ Graph DB │
└────────────┘
读-推理-写循环
1. 检索(Retrieve) 查询长期记忆 → 召回相关历史
2. 推理(Reason) 当前输入 + 记忆 → LLM 生成回复/行动
3. 存储(Store) 将本轮关键信息写回长期记忆
mem0:记忆中间层
专为 AI Agent 设计的记忆管理库,自动提取、去重、更新用户偏好。
pip install mem0aifrom mem0 import Memory
m = Memory()
# 写入(自动提取关键信息)
m.add("用户喜欢 Python,不喜欢 Java", user_id="user_001")
m.add("项目使用 Redis 做缓存,MySQL 做持久化", user_id="user_001")
# 检索
memories = m.search("用户的技术栈是什么?", user_id="user_001")
for mem in memories:
print(mem["memory"])
# 自动合并矛盾信息
m.add("用户改用 PostgreSQL 替代 MySQL", user_id="user_001")
# → 旧的 MySQL 记忆被更新为 PostgreSQLMemGPT / Letta:分页记忆架构
将 context window 模拟为操作系统内存分页,Agent 主动管理冷热数据:
Main Context(工作区,有限容量)
├── System Prompt
├── 当前对话
└── 活跃记忆片段(从外部换入)
External Storage(无限容量)
├── Archival Memory(归档库,向量检索)
└── Recall Memory(对话历史,时序查询)
Agent 自主调用内置工具换页:
# Agent 调用内置记忆工具
archival_memory_search("用户上次提到的项目需求")
archival_memory_insert("用户要求支持多租户,优先级高")
recall_memory_search("最近5次对话摘要")LangGraph Checkpointing
通过 Checkpointer 在 Graph 节点间持久化 Agent 状态,天然支持跨会话记忆:
from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver
# 内存(进程内持久)
checkpointer = MemorySaver()
# PostgreSQL(跨会话持久)
checkpointer = PostgresSaver.from_conn_string("postgresql://...")
app = graph.compile(checkpointer=checkpointer)
# 同一 thread_id 自动继承上次对话状态
config = {"configurable": {"thread_id": "user_001_session_1"}}
result = app.invoke({"messages": [HumanMessage("继续上次的任务")]}, config)记忆更新策略
| 策略 | 说明 | 适用 |
|---|---|---|
| 追加(Append) | 直接写入,不修改旧记录 | 事件日志、对话历史 |
| 更新(Update) | 覆盖同一实体的旧信息 | 用户偏好、配置项 |
| 摘要(Summarize) | 定期压缩历史,减少存储量 | 长对话 |
| 遗忘(Forget) | TTL 过期或置信度衰减后删除 | 临时任务状态 |
与 RAG 的关系
核心区别
| 维度 | RAG | Agent Memory |
|---|---|---|
| 目的 | 检索外部知识以辅助生成 | 积累交互历史以实现连续性 |
| 内容 | 结构化文档、领域知识库 | 用户偏好、对话历史、任务状态 |
| 更新频率 | 低(文档库定期更新) | 高(每次对话后写入) |
| 检索触发 | 每次回答时按需检索 | 会话开始时加载 + 推理过程中查询 |
| 时间维度 | 无时序(相关性排序) | 有时序(事件日志、版本追踪) |
| 个性化 | 无(对所有用户相同) | 强(用户级隔离) |
| 典型工具 | LangChain、LlamaIndex、Milvus | mem0、MemGPT、LangGraph Checkpointer |
两者的互补定位
RAG 静态知识库检索 → 回答"这是什么"
Memory 动态历史积累 → 回答"你之前说过什么、你喜欢什么"
分工:
RAG 负责 领域知识(Java 文档、公司手册、技术规范)
Memory 负责 交互知识(用户偏好、任务进度、对话上下文)
结合使用架构
用户提问
↓
┌───────────────────────────────────┐
│ 检索层 │
│ Memory 检索(用户画像 + 历史) │
│ + │
│ RAG 检索(领域知识库) │
└───────────────┬───────────────────┘
↓
上下文拼接
┌─────────────────────┐
│ System: 你是... │
│ Memory: 用户偏好 Python,上次任务进行到第3步 │
│ RAG: 检索到的相关文档片段 │
│ Human: 当前问题 │
└─────────────────────┘
↓
LLM
↓
更准确、更个性化的回答
def build_context(question: str, user_id: str) -> str:
# Memory:个人化上下文
memories = memory_client.search(question, user_id=user_id)
memory_ctx = "\n".join([m["memory"] for m in memories[:3]])
# RAG:领域知识
docs = vectorstore.similarity_search(question, k=5)
rag_ctx = "\n\n".join([d.page_content for d in docs])
return f"用户记忆:\n{memory_ctx}\n\n相关知识:\n{rag_ctx}"