AI Agent记忆系统架构：让智能体真正记住你的2026方案
过去一年我们在生产环境里跑了十几个AI Agent项目，有一个问题始终绕不开：用户跟Agent聊了三轮之后，Agent就开始"失忆"。上次说过的偏好忘了，上周交代的任务细节丢了，甚至连用户的名字都记不住。这不是模型能力的问题，而是记忆系统架构的问题。
这篇文章是我对Agent记忆系统工程实践的一次完整梳理。如果你正在做Agent产品，或者在评估现有的记忆方案，希望这些经验能帮到你。
为什么Agent需要记忆系统
先说一个真实场景。我们给一家SaaS公司做客服Agent，第一版没有持久记忆，每次对话都是从零开始。用户反馈很直接："这个机器人像个金鱼，每次都得重新介绍自己。"满意度评分只有2.1（满分5分）。
加了记忆系统之后，Agent能记住用户的历史工单、产品配置、之前的沟通偏好。同一个用户再来咨询时，Agent能直接说"上次您反馈的导出功能问题，我们已经修复了"。满意度涨到4.3。
这不是个例。没有记忆的Agent本质上就是个无状态的API，每次调用都是一次性的。而真正的Agent需要：
- 连续性：跨会话记住用户身份和历史交互
- 一致性：不要前后矛盾，比如上次说喜欢简洁风格，这次就别长篇大论
- 效率：减少用户重复输入，提升任务完成速度
- 个性化：根据用户习惯调整行为模式
三层记忆架构
经过多个项目的迭代，我们总结出一套三层记忆架构。这个架构参考了认知科学里人类记忆的分层模型，但在工程上做了大幅简化。
┌─────────────────────────────────────────────────┐
│ Agent Pipeline │
│ │
│ ┌──────────────┐ ┌──────────┐ ┌───────────┐ │
│ │ 短期上下文 │→│ 工作记忆 │→│ 长期记忆 │ │
│ │ (Context Win) │ │(Scratchpad│ │(Persistent)│ │
│ │ 数秒~数分钟 │ │ 数分钟~ │ │ 天~年 │ │
│ │ 原始对话内容 │ │ 当前任务 │ │ 结构化知识 │ │
│ └──────────────┘ └──────────┘ └───────────┘ │
│ │
│ ┌──────────────────────────┐ │
│ │ Memory Consolidation │ │
│ │ (记忆整合与衰减) │ │
│ └──────────────────────────┘ │
└─────────────────────────────────────────────────┘
第一层：短期上下文窗口
这一层就是模型的原始context window。当前对话的消息列表，直接塞进prompt。实现最简单，但也最受限。
class ShortTermMemory:
def __init__(self, max_tokens=128000):
self.messages = []
self.max_tokens = max_tokens
def add(self, role: str, content: str):
self.messages.append({"role": role, "content": content})
self._trim()
def _trim(self):
# 简单截断：从前面删，保留最近的对话
while self._estimate_tokens() > self.max_tokens * 0.8:
self.messages.pop(0)
def get_context(self) -> list[dict]:
return self.messages
工程上的主要挑战是token管理。即使用128K的窗口，复杂的多轮对话加system prompt加工具调用结果，很快就满了。我们的做法是：
- 工具调用的原始JSON结果做摘要后再存入上下文
- 超过一定轮次的历史对话做压缩（用小模型生成摘要）
- 保留最近N轮的完整对话作为"近端记忆"
第二层：工作记忆（Scratchpad）
工作记忆是当前任务的临时存储区。类似人类做数学题时在草稿纸上写写画画。这一层的数据生命周期是"一次任务"，任务结束后要么被清理，要么被整合进长期记忆。
class WorkingMemory:
def __init__(self):
self.scratchpad = {} # 键值对存储
self.task_stack = [] # 任务栈
self.intermediate_results = [] # 中间结果
def note(self, key: str, value: any):
"""记录当前任务的关键信息"""
self.scratchpad[key] = {
"value": value,
"timestamp": time.time(),
"task_id": self.current_task_id()
}
def recall(self, key: str) -> any:
return self.scratchpad.get(key, {}).get("value")
def build_prompt_section(self) -> str:
"""将工作记忆转成prompt片段"""
lines = ["## 当前任务上下文"]
for k, v in self.scratchpad.items():
lines.append(f"- {k}: {v['value']}")
return "\n".join(lines)
在实际实现中，工作记忆通常和Agent的ReAct循环绑定。每一步推理产生的中间结论、工具调用的结果、子任务的返回值，都放在这里。LangChain的AgentExecutor和LlamaIndex的WorkflowRunner内部都有类似机制。
第三层：长期记忆
这是最关键也最复杂的一层。长期记忆需要跨会话持久化，需要支持高效的检索，还需要处理信息的更新和冲突。
长期记忆通常由多个存储后端组合而成：
长期记忆
├── 向量数据库（语义检索）
│ └── 存储：对话摘要、文档片段、经验记录
├── 知识图谱（关系推理）
│ └── 存储：实体关系、用户画像、领域知识
└── 结构化存储（精确查询）
└── 存储：用户偏好、配置项、历史事件时间线
我们的典型实现：
class LongTermMemory:
def __init__(self, vector_db, graph_db, kv_store):
self.vector = vector_db # 如 Qdrant, Milvus, Chroma
self.graph = graph_db # 如 Neo4j, FalkorDB
self.kv = kv_store # 如 Redis, PostgreSQL
async def store(self, memory: MemoryRecord):
"""存储一条记忆"""
# 1. 生成embedding存入向量库
embedding = await embed(memory.content)
await self.vector.upsert(
id=memory.id,
vector=embedding,
metadata=memory.metadata
)
# 2. 提取实体关系存入图谱
entities = await extract_entities(memory.content)
for e in entities:
await self.graph.merge_entity(e)
# 3. 结构化信息存入KV
if memory.structured_data:
await self.kv.hset(
f"user:{memory.user_id}:prefs",
memory.structured_data
)
async def retrieve(self, query: str, user_id: str, top_k=5):
"""多路召回"""
# 语义搜索
vec_results = await self.vector.search(
query=await embed(query),
filter={"user_id": user_id},
top_k=top_k
)
# 图谱查询（如果query涉及实体关系）
graph_results = []
entities = await extract_entities(query)
if entities:
graph_results = await self.graph.query_relations(entities)
# 结构化查询
prefs = await self.kv.hgetall(f"user:{user_id}:prefs")
# 合并去重排序
return self._merge_and_rank(vec_results, graph_results, prefs)
主流方案对比
2026年市面上的记忆方案已经比较成熟了，但各有侧重。以下是我们在实际项目中评估和使用过的几个方案：
| 特性 | Mem0 | Zep | Hermes Memory | LangGraph Store |
|---|---|---|---|---|
| 核心定位 | 通用记忆层 | 对话记忆+分析 | 全栈Agent记忆 | 图状态存储 |
| 向量检索 | 支持 | 支持 | 支持 | 需外接 |
| 知识图谱 | 有限 | 支持（v2） | 原生集成 | 图结构原生 |
| 记忆衰减 | 基础 | 时间衰减 | 多策略衰减 | 无内置 |
| 多用户隔离 | 支持 | 支持 | Profile级 | 线程级 |
| 自托管 | 支持 | 支持 | 默认自托管 | 自托管 |
| 上下文压缩 | 手动 | 自动 | 自动+可控 | 需自实现 |
| 部署复杂度 | 低 | 中 | 中 | 低（依赖LangGraph） |
| 适合场景 | 快速原型 | 客服/销售 | 复杂Agent | 图工作流 |
Mem0
Mem0的设计哲学是"简单即美"。它提供了一个统一的add/search API，底层可以对接多种向量数据库。上手很快，几行代码就能跑起来：
from mem0 import Memory
m = Memory()
# 存储
m.add("我喜欢用Python写后端", user_id="alice")
# 检索
results = m.search("这个用户喜欢什么编程语言", user_id="alice")
优点是集成简单，缺点是记忆模型比较扁平。它把所有记忆都当成文本片段存在向量库里，缺乏结构化的组织。当记忆量增长到几万条之后，检索质量会明显下降。另外它的记忆衰减策略比较基础，主要靠时间戳排序。
Zep
Zep在v2版本加入了知识图谱支持（基于FalkorDB），这让它在关系推理方面有了明显提升。比如"这个用户的老板是谁"、"上次讨论的项目进展如何"这类需要关系链的查询，Zep处理得比纯向量方案好很多。
from zep_cloud.client import Zep
client = Zep(api_key="your-key")
# 自动从对话中提取记忆
client.memory.add(session_id="s1", messages=messages)
# 检索
memory = client.memory.search_sessions(
text="项目截止日期",
user_id="alice"
)
Zep的短板在于它更偏向"对话分析"场景。如果你的Agent需要执行复杂任务（比如代码生成、数据分析），单纯的记忆检索不够用，还需要工作记忆和任务状态管理。
Hermes Memory
Hermes Agent内置的记忆系统是我们在复杂Agent项目中用得最多的。它的设计覆盖了完整的三层架构，包括短期上下文管理、工作记忆（scratchpad）、长期记忆（含向量+KV+文件存储）。
Hermes Memory的一个特色是Profile机制。不同的Agent profile有独立的记忆空间、技能集和插件配置。这在多Agent系统或者需要人格切换的场景下非常实用。
~/.hermes/
├── profiles/
│ ├── default/
│ │ ├── memories/ # 长期记忆文件
│ │ ├── skills/ # 技能定义
│ │ └── plugins/ # 插件配置
│ └── customer_service/
│ ├── memories/
│ └── ...
记忆的存储格式是结构化的Markdown文件，配合向量索引。这种设计的好处是记忆可读、可编辑、可版本控制，不会被锁死在某个向量数据库里。
LangGraph Store
LangGraph的Store是一个图原生的状态存储。它不是专门为记忆设计的，但天然适合存储Agent的工作状态和上下文。
from langgraph.store import InMemoryStore
store = InMemoryStore()
# 按命名空间存储
store.put(("user", "alice"), "preferences", {
"language": "zh",
"verbosity": "concise"
})
# 检索
prefs = store.search(("user", "alice"))
LangGraph Store的优势是和LangGraph的工作流引擎无缝集成。如果你的Agent已经是LangGraph生态，用它做工作记忆和短期状态存储很方便。但它不是完整记忆方案，缺少语义检索、记忆衰减、自动整合等功能。
记忆整合策略
记忆整合（Memory Consolidation）是记忆系统中最容易被忽略但又最关键的环节。没有整合，记忆库会变成一个垃圾堆积场。
我们采用的整合策略分三个维度：
1. 去重与合并
同一个事实可能在不同对话中被多次提到。比如用户说了三次"我用Mac"，应该合并成一条记忆而不是三条。
async def consolidate_memories(user_id: str):
# 获取最近的记忆
recent = await memory.get_recent(user_id, days=7)
# 聚类相似记忆
clusters = await cluster_by_similarity(recent, threshold=0.85)
for cluster in clusters:
if len(cluster) > 1:
# 用LLM合并同一簇的记忆
merged = await llm.merge_memories(cluster)
# 保留最新的，删除旧的
await memory.replace(cluster, merged)
2. 重要性衰减
不是所有记忆都同等重要。"用户喜欢蓝色"比"用户昨天下午3点登录过"更有长期价值。我们用一个多因子评分模型：
def memory_importance_score(memory):
recency = time_decay(memory.created_at, half_life_days=30)
frequency = min(memory.access_count / 10, 1.0)
explicitness = 1.0 if memory.source == "user_explicit" else 0.6
relevance = memory.entity_count / 5 # 涉及多少实体
return 0.3 * recency + 0.25 * frequency + 0.25 * explicitness + 0.2 * relevance
评分低于阈值的记忆会被自动归档或删除。这个过程每天跑一次，保持记忆库的"干净"。
3. 时间线整合
把零散的记忆事件按时间线组织起来，形成结构化的用户故事。比如：
{
"user_id": "alice",
"timeline": [
{
"date": "2026-05-01",
"event": "首次使用产品",
"details": "通过官网注册，选择了Pro套餐"
},
{
"date": "2026-05-15",
"event": "反馈导出功能问题",
"details": "CSV导出中文乱码，已提交工单#1234"
},
{
"date": "2026-06-10",
"event": "问题已修复",
"details": "确认导出功能正常，满意度评分4/5"
}
]
}
时间线记忆在客服和销售场景中特别有用。Agent可以快速了解用户的完整交互历史，提供连贯的服务体验。
部署实践中的坑
理论说完，讲几个实际部署中踩过的坑。
坑一：记忆检索的延迟
在我们的一个项目中，每次用户发消息，Agent都要检索长期记忆。向量搜索本身很快（几十毫秒），但加上embedding生成、图谱查询、结果排序，整个链路要200-500ms。用户体感就是"机器人反应慢"。
解决方案： - embedding请求做缓存，相似query复用结果 - 检索和LLM推理做并行，不串行 - 只在需要时检索，用一个轻量级的classifier判断当前query是否需要查记忆
async def process_message(user_msg, user_id):
# 并行执行记忆检索和LLM准备
memory_task = asyncio.create_task(
retrieve_if_needed(user_msg, user_id)
)
system_prompt_task = asyncio.create_task(
build_system_prompt(user_id)
)
memory_context, system_prompt = await asyncio.gather(
memory_task, system_prompt_task
)
# 合并到prompt中
if memory_context:
system_prompt += f"\n\n## 用户相关记忆\n{memory_context}"
return await llm.generate(system_prompt, user_msg)
坑二：记忆冲突
用户说"我改用VS Code了"，但记忆里还存着"用户喜欢用PyCharm"。两条记忆都是对的，但时间不同。如果不处理，Agent可能会给出过时的建议。
我们的做法是给每条记忆加版本号和时间戳，检索时优先返回最新的。同时维护一个"事实状态表"，当新信息和旧信息矛盾时，自动标记旧信息为"已过时"。
坑三：隐私与合规
记忆系统天然涉及用户数据的持久化存储。在GDPR和国内个人信息保护法的要求下，需要注意：
- 用户有权要求删除所有记忆，需要提供一键清除功能
- 敏感信息（手机号、地址等）需要脱敏存储
- 记忆的访问和导出需要有审计日志
- 不同用户之间的记忆必须严格隔离
坑四：记忆库膨胀
上线三个月后，一个活跃用户的记忆条目可能达到数千条。全量检索不现实，我们采用了分层检索策略：
class TieredRetrieval:
async def search(self, query, user_id, top_k=5):
# 第一层：从最近7天的记忆中搜索（热数据）
hot = await self.vector.search(
query, filter={"user_id": user_id, "days": 7}, top_k=top_k
)
if len(hot) >= top_k:
return hot
# 第二层：从重要性高的长期记忆中搜索（冷数据）
cold = await self.vector.search(
query,
filter={"user_id": user_id, "importance": {"$gt": 0.7}},
top_k=top_k - len(hot)
)
return self._deduplicate(hot + cold)
一个完整的记忆管道示例
把上面的组件串起来，一个完整的记忆处理流程是这样的：
用户消息进入
│
▼
┌──────────────────┐
│ 意图判断：是否需要 │──── 不需要 ──→ 直接进入LLM推理
│ 查记忆？ │
└──────────────────┘
│ 需要
▼
┌──────────────────┐
│ 多路检索长期记忆 │
│ (向量+图谱+KV) │
└──────────────────┘
│
▼
┌──────────────────┐
│ 记忆排序与过滤 │
│ (相关性+时效性) │
└──────────────────┘
│
▼
┌──────────────────┐
│ 注入到System │
│ Prompt中 │
└──────────────────┘
│
▼
┌──────────────────┐
│ LLM生成回复 │
└──────────────────┘
│
▼
┌──────────────────┐
│ 对话结束后的 │
│ 记忆整合 │
│ (提取新记忆/ │
│ 更新旧记忆/ │
│ 清理过期记忆) │
└──────────────────┘
总结
记忆系统是Agent从"工具"变成"助手"的关键基础设施。2026年的方案已经比较成熟，但没有银弹。选择哪种方案取决于你的具体场景：
- 快速原型/简单场景：Mem0足够，几行代码搞定
- 客服/销售Agent：Zep的知识图谱和对话分析能力更适合
- 复杂多能力Agent：Hermes Memory的三层架构加Profile机制更完整
- 已有LangGraph生态：LangGraph Store做状态管理，外挂向量库做语义记忆
不管选哪个方案，记忆整合策略都是必须认真对待的。没有整合的记忆系统，时间一长就会退化成一个充满噪音的垃圾堆。
最后一个建议：先从最简单的方案开始，跑通端到端的流程，再根据实际瓶颈逐步升级。别一上来就搞一个五脏俱全的记忆平台，大概率会在架构复杂度上翻车。