1. 项目概述:构建AI Agent的三层记忆系统
在构建长期运行的AI助手时,记忆缺失是最令人头疼的问题之一。想象一下:你精心调教的助手昨天刚帮你解决了技术难题,今天却对同样的问题一脸茫然——这种体验就像每天都要重新培训新员工。本文介绍的三层记忆架构,正是为了解决这个痛点而生。
这个系统通过会话层(短期记忆)、文件层(身份记忆)和语义层(长期经验)的有机组合,让AI Agent能够像人类一样拥有持续的记忆能力。整套方案采用纯Python实现,核心代码不到200行,依赖仅有chromadb和sentence-transformers两个库,却能支撑起一个可以长期运行的智能助手。
关键优势:相比直接将所有历史对话塞进prompt或单纯依赖向量数据库的方案,这种分层设计在资源消耗、记忆准确性和维护成本之间取得了更好的平衡。
2. 核心架构设计解析
2.1 为什么需要分层记忆
现有方案存在明显局限:
- 全量prompt方案:随着对话增长,token消耗呈线性上升,很快就会触及context window上限(即使是128K的GPT-4-turbo)
- 纯向量检索方案:对结构化信息(如配置参数、精确数据)的召回效果不稳定,且无法保证关键信息的持久可用
我们的三层架构将记忆按生命周期和用途分类处理:
code复制┌───────────────────────┐
│ 会话层 (SessionMemory) │
│ • 当前对话上下文 │
│ • 临时状态 │
│ • 生命周期:单次会话 │
├───────────────────────┤
│ 文件层 (FileMemory) │
│ • 身份定义 │
│ • 工具配置 │
│ • 精选记忆 │
│ • 生命周期:永久 │
├───────────────────────┤
│ 语义层 (SemanticMemory)│
│ • 日常日志 │
│ • 经验片段 │
│ • 生命周期:长期 │
└───────────────────────┘
2.2 各层技术选型考量
会话层选用deque数据结构实现:
- 固定长度队列自动淘汰最早消息(maxlen参数)
- 字符数估算采用len(text)//3的简化方法(生产环境建议使用tiktoken)
- 保留最后两条消息的机制防止误删system prompt
文件层采用Markdown格式存储:
- 人类可读可编辑,便于调试和维护
- 文件系统本身就是天然的KV存储(文件名作为key)
- 按日期分割日志避免单个文件过大
语义层选用Chromadb+MiniLM:
- 完全本地运行,无需API调用
- 多语言支持(paraphrase-multilingual模型)
- 400MB的小模型尺寸适合个人使用
3. 核心实现细节
3.1 会话记忆管理
SessionMemory类的关键实现技巧:
python复制def _trim(self):
"""动态修剪消息队列"""
total = sum(len(m["content"]) // 3 for m in self.messages)
while total > self.max_tokens and len(self.messages) > 2:
removed = self.messages.popleft()
total -= len(removed["content"]) // 3
踩坑记录:初期版本没有保留最后两条消息的限制,导致system prompt被意外清除,造成Agent行为异常。现在的实现确保至少保留两条最新消息。
3.2 文件系统记忆
FileMemory类的设计亮点:
-
身份文件分离存储:
- IDENTITY.md:核心身份定义
- MEMORY.md:精选长期记忆
- TOOLS.md:工具配置和环境变量
-
智能更新机制:
python复制def update_memory(self, key: str, value: str):
"""更新MEMORY.md的指定章节"""
if marker in text: # 已有该章节
# 定位并替换内容
else:
# 追加新章节
- 日志自动分割:
每天生成独立的YYYY-MM-DD.md文件,避免单个文件过大影响读取效率
3.3 语义记忆实现
SemanticMemory类的关键技术点:
- 本地化嵌入模型:
python复制self.ef = embedding_functions.SentenceTransformerEmbeddingFunction(
model_name="paraphrase-multilingual-MiniLM-L12-v2"
)
- 日志自动索引:
python复制def index_daily_logs(self):
for f in Path(log_dir).glob("2*.md"): # 匹配日期格式文件
paragraphs = [p.strip() for p in text.split("\n\n") if len(p.strip()) > 20]
for para in paragraphs:
self.store(para, {"source": f.name})
- 混合检索策略:
- 同时考虑语义相似度(向量距离)和元数据过滤
- 设置0.5的相似度阈值过滤低质量结果
4. 系统集成与使用示例
4.1 记忆管理器整合
AgentMemory类作为统一接口:
python复制class AgentMemory:
def boot(self) -> str:
"""启动时加载文件层和最近日志"""
identity = self.files.load_identity()
recent = self.files.load_recent_logs(days=2)
return f"{identity}\n\n{recent}"
def remember(self, text: str, persist: bool = False):
"""记录信息到会话层和语义层"""
self.session.add("assistant", f"[记忆] {text}")
self.semantic.store(text)
if persist:
self.files.append_log(text)
4.2 典型使用场景
初始化Agent:
python复制memory = AgentMemory("./my_agent/memory")
context = memory.boot() # 加载初始上下文
日常交互:
python复制# 记录需要长期保存的信息
memory.remember("用户偏好:喜欢用Markdown格式回复", persist=True)
# 查询相关记忆
results = memory.recall("用户喜欢什么格式")
定期维护:
python复制# 让Agent自己整理记忆(类似人类睡眠时的记忆巩固)
def consolidate_memory():
recent = memory.files.load_recent_logs(days=7)
# 提取重要信息更新到MEMORY.md
# 删除过时记录
5. 生产环境优化建议
5.1 性能调优技巧
- 懒加载向量库:
python复制def recall(self, query: str) -> str:
if not hasattr(self, '_semantic_loaded'):
self.semantic = SemanticMemory() # 首次调用时初始化
self._semantic_loaded = True
...
-
日志分块策略:
- 每个日志文件不超过100条记录
- 达到限制后自动创建新文件(如2024-03-15_part1.md)
-
嵌入模型选择:
- 英文为主场景:all-MiniLM-L6-v2(更小更快)
- 需要最新技术:选用最近开源的embedding模型
5.2 常见问题解决方案
问题1:向量检索返回无关结果
- 解决方案:
- 在结果中附加来源和日期信息
- 设置更高的相似度阈值(如0.6)
- 添加元数据过滤(如只检索特定日期范围)
问题2:记忆冲突
- 解决策略:
- 文件层优先级高于语义层
- 每周执行记忆整理(类似"碎片整理")
- 对关键配置项添加版本管理
问题3:启动时间过长
- 优化方法:
- 并行加载各层记忆
- 将部分初始化工作移至后台线程
- 使用更轻量的embedding模型
6. 进阶扩展方向
6.1 记忆压缩与摘要
实现定期自动摘要功能:
python复制def summarize_logs():
# 调用LLM对一周日志生成摘要
summary = llm.generate("请总结本周重要事件和技术发现")
memory.files.update_memory("本周摘要", summary)
6.2 记忆重要性评估
添加记忆权重机制:
python复制def remember(self, text: str, importance: int = 1):
"""根据重要性决定存储策略"""
if importance > 3:
self.files.update_memory("重要事件", text)
self.semantic.store(text, {"importance": importance})
6.3 多模态记忆扩展
支持图像等非文本记忆:
python复制def store_image(self, image_path: str, description: str):
"""存储图像及其描述"""
embedding = self.image_model.encode(image_path)
self.collection.upsert(
embeddings=[embedding],
documents=[description],
ids=[hashlib.md5(image_path.encode()).hexdigest()]
)
这套三层记忆系统经过两个月的实际运行测试,成功将一个简单的聊天机器人转变为可以持续学习的个人助手。它现在能准确回忆起三周前的技术讨论细节,并在类似问题出现时主动提供参考方案。记忆系统的加入,真正让AI Agent从"高级玩具"变成了"有用工具"。
