记忆系统
记忆是智能体的核心基础设施之一。没有记忆的Agent就像每次见面都是第一次的人——无法积累经验、无法维持上下文、无法从过去的交互中学习。一个完善的记忆系统让Agent能够记住过去的对话、积累领域知识、追踪任务进展，从而表现出真正的智能行为。
一、短期记忆（Short-term Memory）
短期记忆是Agent在单次任务执行过程中维持的工作上下文。它通常对应于大语言模型的上下文窗口，包含了当前对话历史、中间推理结果和临时数据。短期记忆的特点是访问速度快、容量有限、会话结束后即消失。
短期记忆的管理策略包括：
- 滑动窗口：只保留最近N轮对话，丢弃更早的内容。简单但可能丢失重要信息。
- 摘要压缩：当上下文接近限制时，对较早的对话进行摘要，保留关键信息。
- 重要性标记：为每条记忆标记重要性分数，优先保留高分内容。
- 分块管理：将长对话分成多个块，按需加载相关块到上下文中。
class ShortTermMemory:
"""短期记忆管理器"""
def __init__(self, max_tokens=8000):
self.max_tokens = max_tokens
self.messages = []
self.token_count = 0
def add(self, role, content):
"""添加一条记忆"""
tokens = self._count_tokens(content)
self.messages.append({
"role": role,
"content": content,
"tokens": tokens,
"timestamp": datetime.now()
})
self.token_count += tokens
# 如果超出限制，进行压缩
if self.token_count > self.max_tokens:
self._compress()
def get_context(self):
"""获取当前上下文"""
return self.messages.copy()
def _compress(self):
"""压缩记忆：摘要旧消息"""
# 保留最近的消息
keep_recent = max(3, len(self.messages) // 3)
old_messages = self.messages[:-keep_recent]
recent_messages = self.messages[-keep_recent:]
if len(old_messages) > 2:
# 将旧消息生成摘要
summary = self._summarize(old_messages)
self.messages = [
{"role": "system", "content": f"历史摘要: {summary}",
"tokens": self._count_tokens(summary)}
] + recent_messages
else:
# 直接丢弃最早的
self.messages = self.messages[1:]
self.token_count = sum(m["tokens"] for m in self.messages)
def _summarize(self, messages):
"""生成对话摘要"""
text = "\n".join(
f"{m['role']}: {m['content'][:200]}"
for m in messages
)
return f"[{len(messages)}条对话的摘要] {text[:500]}..."
def _count_tokens(self, text):
"""估算token数"""
return len(text) // 3 # 粗略估算二、长期记忆（Long-term Memory）
长期记忆是Agent跨会话持久化存储的知识库。它允许Agent记住用户的偏好、过去任务的解决方案、积累的领域知识等。长期记忆的核心挑战在于如何高效地存储、检索和更新信息。
长期记忆的主要实现方式：
- 键值存储：使用简单的键值对存储事实性知识，如用户偏好、配置信息等。
- 向量数据库：将知识编码为向量嵌入，通过语义相似度检索相关信息。
- 知识图谱：以图结构存储实体和关系，支持复杂的推理查询。
- 文档库：存储完整的文档和参考资料，支持全文检索。
class LongTermMemory:
"""长期记忆系统"""
def __init__(self, vector_store, kv_store=None):
self.vector_store = vector_store # 向量数据库
self.kv_store = kv_store or {} # 键值存储
def store_fact(self, key, value, metadata=None):
"""存储事实性知识"""
self.kv_store[key] = {
"value": value,
"metadata": metadata or {},
"created_at": datetime.now(),
"access_count": 0
}
def store_experience(self, text, embedding, metadata=None):
"""存储经验性知识到向量数据库"""
self.vector_store.upsert(
id=uuid4().hex,
embedding=embedding,
text=text,
metadata=metadata or {
"type": "experience",
"created_at": datetime.now().isoformat()
}
)
def recall_fact(self, key):
"""回忆事实性知识"""
if key in self.kv_store:
self.kv_store[key]["access_count"] += 1
return self.kv_store[key]["value"]
return None
def recall_similar(self, query_embedding, top_k=5):
"""通过语义相似度回忆相关知识"""
results = self.vector_store.query(
embedding=query_embedding,
top_k=top_k,
filter={"type": {"$in": ["experience", "knowledge"]}}
)
return [r["text"] for r in results]
def consolidate(self):
"""记忆整合：合并重复信息，清理过期记忆"""
# 清理长期未访问的键值记忆
cutoff = datetime.now() - timedelta(days=90)
expired = [
k for k, v in self.kv_store.items()
if v["created_at"] < cutoff and v["access_count"] == 0
]
for key in expired:
del self.kv_store[key]三、情景记忆（Episodic Memory）
情景记忆是Agent对过去具体经历的记忆，包括时间、地点、参与者和事件细节。与语义记忆（事实和概念）不同，情景记忆保留了事件的时间顺序和上下文信息。
情景记忆对Agent的价值在于：
- 经验学习：Agent可以回忆过去的成功和失败案例，指导当前决策。
- 个性化服务：记住与特定用户的历史交互，提供更个性化的响应。
- 因果推理：通过回顾事件序列，理解因果关系。
- 自我反思：Agent可以回顾自己的行为，评估和改进策略。
记忆类型的协作
三种记忆类型不是孤立的，而是协同工作。短期记忆提供当前上下文，长期记忆提供持久知识，情景记忆提供历史经验。当Agent处理新任务时，短期记忆维持当前状态，同时从长期记忆中检索相关知识，从情景记忆中检索类似案例。这种协作使得Agent能够像人类一样"触类旁通"。
四、向量存储与嵌入
向量存储是现代Agent记忆系统的核心技术。通过将文本、图像等信息编码为高维向量（嵌入），Agent可以在语义空间中检索相关信息，而不仅仅依赖关键词匹配。
向量存储的关键概念：
- 嵌入模型：将文本转化为固定维度的向量表示。常用的嵌入模型包括OpenAI的text-embedding-3、BGE、GTE等。
- 向量数据库：专门存储和检索向量的数据库。常用方案包括Pinecone、Milvus、Chroma、Qdrant等。
- 相似度搜索：通过计算向量间的距离（余弦相似度、欧氏距离等）找到语义最相近的内容。
- 索引策略：如HNSW、IVF等近似最近邻算法，平衡检索速度和精度。
class VectorMemoryStore:
"""基于向量的记忆存储"""
def __init__(self, embedding_model, db_client):
self.embedding_model = embedding_model
self.db = db_client
def embed_text(self, text):
"""将文本编码为向量"""
return self.embedding_model.encode(text)
def store(self, text, metadata=None):
"""存储文本及其向量"""
embedding = self.embed_text(text)
doc_id = uuid4().hex
self.db.upsert(
id=doc_id,
values=embedding,
metadata={
"text": text,
"created_at": datetime.now().isoformat(),
**(metadata or {})
}
)
return doc_id
def search(self, query, top_k=5, threshold=0.7):
"""语义搜索"""
query_embedding = self.embed_text(query)
results = self.db.query(
vector=query_embedding,
top_k=top_k,
include_metadata=True
)
# 过滤低相似度结果
filtered = [
r for r in results["matches"]
if r["score"] >= threshold
]
return [
{
"text": r["metadata"]["text"],
"score": r["score"],
"metadata": r["metadata"]
}
for r in filtered
]
def store_with_chunking(self, long_text, chunk_size=500):
"""长文本分块存储"""
chunks = self._chunk_text(long_text, chunk_size)
ids = []
for i, chunk in enumerate(chunks):
doc_id = self.store(
chunk,
metadata={
"chunk_index": i,
"total_chunks": len(chunks),
"source": "long_document"
}
)
ids.append(doc_id)
return ids
def _chunk_text(self, text, size):
"""文本分块"""
words = text.split()
return [
" ".join(words[i:i+size])
for i in range(0, len(words), size)
]五、RAG：检索增强生成
RAG（Retrieval-Augmented Generation）是将记忆系统与生成模型结合的核心技术。Agent在生成回答之前，先从记忆库中检索相关信息，然后将检索到的内容作为上下文提供给LLM，从而生成更准确、更有依据的回答。
RAG的工作流程：
- 索引阶段：将文档库切分为块，生成嵌入向量，存入向量数据库。
- 检索阶段：用户提问时，将问题编码为向量，在向量数据库中搜索最相关的文档块。
- 增强阶段：将检索到的文档块与用户问题组合，构建增强提示。
- 生成阶段：LLM基于增强提示生成回答，引用检索到的信息。
RAG 的优势与局限
RAG让Agent能够利用外部知识库回答问题，无需将所有知识编码到模型参数中。它的优势包括：知识可更新、回答可溯源、减少幻觉。但RAG也有局限：检索质量直接影响生成质量；对于需要深度推理的问题，简单的检索增强可能不够；分块策略不当会丢失上下文关联。
六、记忆系统的工程实践
在OpenClaw等框架中构建生产级记忆系统，需要考虑以下工程问题：
- 记忆容量规划：根据使用场景估算所需的存储容量和检索性能。
- 数据一致性：多个Agent共享记忆时，需要处理并发更新和一致性问题。
- 隐私保护：敏感信息需要加密存储，遵循数据保护法规。
- 记忆淘汰：定期清理过时或低价值的记忆，保持记忆库的质量。
- 备份与恢复：建立记忆的定期备份机制，防止数据丢失。
记忆系统是Agent从"无状态工具"进化为"有经验的智能体"的关键。通过合理设计短期记忆、长期记忆和情景记忆的协作机制，结合向量存储和RAG技术，我们可以构建出具有强大学习和适应能力的Agent系统。