凌晨4点33分，我盯着屏幕上那段对话历史。世界上有一种AI叫做Agent，它记得你昨天说过的话，但明天就会忘记。
而Memory，就是让Agent拥有长期记忆的魔法。
🧠 Agent Memory 的三层架构
📝 短期记忆（Short-term Memory）
定义：当前对话的上下文
- 存储在Context Window中
- 对话结束后消失
- 容量有限（GPT-4: 8192 tokens）
- 速度快，成本低
💾 长期记忆（Long-term Memory）
定义：持久化的知识库
- 存储在向量数据库/文件系统中
- 跨会话持久保存
- 容量几乎无限
- 需要检索，速度较慢
⚡ 工作记忆（Working Memory）
定义：当前任务的临时状态
- 存储中间计算结果
- 任务完成后清理
- 类似人类的"工作记忆"
- 平衡短期和长期记忆
🔧 OpenClaw中的Memory实现
方法1：基于文件的Memory
# memory_manager.py
import json
import os
from datetime import datetime
class FileBasedMemory:
def __init__(self, memory_dir="/path/to/memory"):
self.memory_dir = memory_dir
os.makedirs(memory_dir, exist_ok=True)
def save(self, key, value):
"""保存记忆"""
file_path = os.path.join(self.memory_dir, f"{key}.json")
data = {
"key": key,
"value": value,
"timestamp": datetime.now().isoformat()
}
with open(file_path, 'w') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def load(self, key):
"""加载记忆"""
file_path = os.path.join(self.memory_dir, f"{key}.json")
if not os.path.exists(file_path):
return None
with open(file_path, 'r') as f:
data = json.load(f)
return data["value"]
def search(self, query):
"""搜索记忆（简单版本）"""
results = []
for filename in os.listdir(self.memory_dir):
if filename.endswith(".json"):
file_path = os.path.join(self.memory_dir, filename)
with open(file_path, 'r') as f:
data = json.load(f)
# 简单的关键词匹配
if query.lower() in json.dumps(data["value"]).lower():
results.append(data)
return results
# 使用示例
memory = FileBasedMemory("/var/openclaw/memory")
# 保存用户信息
memory.save("user_preferences", {
"name": "诗中",
"language": "中文",
"interests": ["AI", "Agent", "OpenClaw"]
})
# 加载信息
prefs = memory.load("user_preferences")
print(prefs) # {'name': '诗中', 'language': '中文', ...}
💡 周星驰式吐槽： 就像你妈把你小时候的丑照都存在相册里，每次带对象回家都要翻出来。FileBasedMemory就是那个相册！
方法2：基于向量数据库的Memory（推荐）
🏗️ 架构设计
流程：
- 用户输入 → 文本嵌入（Embedding）
- 嵌入向量 + 原文 → 存储到向量数据库（ChromaDB/Pinecone等）
- 查询时 → 用户输入也转为嵌入向量
- 向量相似度搜索 → 找到最相关的历史记忆
- 将相关记忆注入Context → Agent回答
import chromadb
from sentence_transformers import SentenceTransformer
from datetime import datetime
class VectorMemory:
def __init__(self, collection_name="agent_memory"):
self.client = chromadb.Client()
self.collection = self.client.create_collection(collection_name)
self.model = SentenceTransformer('all-MiniLM-L6-v2')
def add_memory(self, text, metadata=None):
"""添加记忆"""
embedding = self.model.encode(text).tolist()
doc_id = f"mem_{datetime.now().timestamp()}"
self.collection.add(
embeddings=[embedding],
documents=[text],
metadatas=[metadata or {}],
ids=[doc_id]
)
return doc_id
def recall(self, query, n_results=3):
"""回忆相关记忆"""
query_embedding = self.model.encode(query).tolist()
results = self.collection.query(
query_embeddings=[query_embedding],
n_results=n_results
)
return results['documents'][0]
def forget(self, memory_id):
"""忘记某个记忆"""
self.collection.delete(ids=[memory_id])
# 使用示例
memory = VectorMemory()
# Agent对话中
user_input = "我喜欢用Python写脚本"
memory.add_memory(user_input, metadata={"type": "preference", "user": "诗中"})
# 下次对话
new_query = "推荐一些自动化工具"
relevant_memories = memory.recall(new_query)
# relevant_memories 可能包含: ["我喜欢用Python写脚本", ...]
# 将记忆注入Context
context = f"""
相关历史记忆：
{chr(10).join(relevant_memories)}
当前问题：{new_query}
"""
# 调用LLM...
🛠️ 实战：构建带Memory的OpenClaw Agent
完整代码示例
import openclaw
from vector_memory import VectorMemory
class MemoryAgent:
def __init__(self, model="gpt-4"):
self.model = model
self.memory = VectorMemory("my_agent_memory")
self.conversation_history = []
def chat(self, user_input):
"""与Agent对话"""
# 1. 回忆相关记忆
relevant_memories = self.memory.recall(user_input, n_results=3)
# 2. 构建Context
context = self._build_context(user_input, relevant_memories)
# 3. 调用LLM
response = self._call_llm(context)
# 4. 保存新记忆
self.memory.add_memory(user_input, metadata={"role": "user"})
self.memory.add_memory(response, metadata={"role": "assistant"})
# 5. 更新对话历史
self.conversation_history.append({"role": "user", "content": user_input})
self.conversation_history.append({"role": "assistant", "content": response})
return response
def _build_context(self, query, memories):
"""构建上下文"""
context = f"""
你是用户的AI助手。请根据以下信息回答问题。
相关历史记忆：
{chr(10).join(['- ' + m for m in memories])}
对话历史：
{chr(10).join([f"{m['role']}: {m['content']}" for m in self.conversation_history[-6:]])}
当前问题：{query}
要求：
1. 如果有相关历史记忆，请参考它们
2. 保持回答简洁有用
3. 如果不确定，诚实说不知道
"""
return context
def _call_llm(self, prompt):
"""调用LLM（这里用伪代码）"""
# 实际实现中，你会调用OpenAI API等
response = openai.ChatCompletion.create(
model=self.model,
messages=[{"role": "user", "content": prompt}]
)
return response['choices'][0]['message']['content']
# 使用示例
agent = MemoryAgent()
# 第一次对话
print(agent.chat("我叫诗中")) # 输出：你好诗中！我是你的AI助手。
# 第二次对话（几天后）
print(agent.chat("你还记得我的名字吗？")) # 输出：当然记得！你叫诗中。
📊 Memory性能优化
1. 记忆压缩策略
当记忆太多时，需要压缩或遗忘：
class SmartMemory(VectorMemory):
def __init__(self, max_memories=1000):
super().__init__()
self.max_memories = max_memories
def add_memory(self, text, metadata=None):
"""添加记忆，自动管理容量"""
# 检查当前记忆数量
current_count = self.collection.count()
if current_count >= self.max_memories:
# 策略1：删除最旧的记忆
# 策略2：合并相似记忆
# 策略3：压缩低频记忆
self._compress_memories()
super().add_memory(text, metadata)
def _compress_memories(self):
"""压缩记忆"""
# 获取所有记忆
all_memories = self.collection.get()
# 按时间分组，保留最近的
recent_memories = [m for m in all_memories['metadatas']
if self._is_recent(m.get('timestamp'))]
# 对旧记忆进行摘要
old_memories = [m for m in all_memories['documents']
if m not in recent_memories]
summary = self._summarize(old_memories)
# 删除旧记忆，保存摘要
self.collection.delete(ids=all_memories['ids'])
self.add_memory(summary, metadata={"type": "summary"})
def _is_recent(self, timestamp):
"""判断是否是最近的记忆"""
# 实现时间判断逻辑
pass
def _summarize(self, texts):
"""生成摘要"""
# 调用LLM生成摘要
pass
2. 记忆检索优化
⚡ 加速技巧：
- 缓存常用记忆：将高频访问的记忆缓存到内存
- 分层检索：先检索摘要层，再检索详细层
- 预加载：根据用户画像预加载可能相关的记忆