MemPalace 深度实战:54K+ Star 本地优先 AI 记忆系统,从「记忆宫殿」架构到生产部署完全指南
一天收获 2 万 Star,LongMemEval 基准测试 96.6% R@5 召回率(无 API 调用),好莱坞女演员 Milla Jovovich 亲自开源——MemPalace 正在重新定义 AI Agent 的长期记忆范式。
一、为什么 AI Agent 需要「长期记忆」?
1.1 当前 AI 系统的致命缺陷
与 AI 对话过的人都有这样的体验:每次对话都是一次「失忆」。
你花了一下午跟 Claude 讨论架构设计、调通了代码、解决了棘手的 bug。第二天打开新会话,一切归零——Claude 不记得你说过什么、不记得项目背景、不记得昨天确定的方案。你只能从头开始重新解释。
这不是偶然,而是 Transformer 架构的先天限制:
# 传统 LLM 的「记忆」只是上下文窗口
context_window = 200_000 # 即使是 200K tokens
# 问题一:线性成本
# 每次推理都要重新处理整个上下文,token 消耗随对话增长
# 问题二:硬性截断
# 超过窗口限制?抱歉,最早的内容直接丢弃
# 问题三:无结构化
# 信息堆砌在窗口里,无法按主题、项目、时间组织
这个问题有多严重?
- 复杂项目开发:一个真实的生产级项目,讨论历史可能长达数十万 token,远超任何模型的上下文窗口
- 多会话协作:你在三个不同的会话里讨论同一项目的不同模块,模型无法跨会话共享信息
- 长期知识积累:你希望 AI 「记住」你的代码风格、偏好、项目约定,但每次都是「第一次见面」
1.2 现有解决方案的局限
市面上已经有一些「AI 记忆」产品,但都有明显缺陷:
| 方案 | 问题 |
|---|---|
| 摘要压缩 | 丢失细节,关键代码片段、决策理由、错误信息可能被「压缩掉」 |
| 向量检索(RAG) | 只能检索「相似内容」,无法理解时序关系、因果关系 |
| 云服务存储 | 依赖网络、有隐私风险、有订阅成本、有 API 调用费用 |
| 知识图谱 | 结构过于刚性,需要人工维护 schema,不适合对话这种非结构化数据 |
MemPalace 的答案:不删减,靠结构
MemPalace 反其道而行:
- 保留完整对话原文——不摘要、不压缩、不丢弃
- 结构化组织——借鉴「记忆宫殿」方法论,用「宫殿-房间-抽屉」三层结构组织记忆
- 本地优先——所有数据存储在本地 ChromaDB,无需 API key、无需订阅、无隐私泄露风险
- 高召回率——LongMemEval 基准测试 96.6% R@5(纯本地,无 API),是目前公开最高的分数
二、MemPalace 架构深度解析
2.1 核心设计哲学:记忆宫殿
MemPalace 的名字来源于古希腊的「记忆宫殿」技术(Method of Loci)。古罗马演说家西塞罗能用这种方法记住长达数小时的演讲稿——他将信息「放置」在一个想象中的宫殿的不同房间里,需要时「走回」那个房间「取回」信息。
MemPalace 将这个 2000 年前的智慧应用到 AI 记忆系统:
The Palace (宫殿)
├── Room: "python-web-server" (按项目/主题划分)
│ ├── Drawer: "2024-03-15-架构讨论" (时间切片)
│ ├── Drawer: "2024-03-16-API设计"
│ └── Drawer: "2024-03-20-性能优化"
├── Room: "react-frontend"
│ ├── Drawer: "组件设计"
│ └── Drawer: "状态管理"
└── Room: "database-design"
└── Drawer: "schema设计"
为什么这种结构有效?
- 空间记忆是人类最擅长的事——你能轻易记住「钥匙放在卧室桌上」,同样的原理适用于信息检索
- 结构化组织降低搜索范围——不用在全量数据里大海捞针,先确定「房间」,再在「抽屉」里找
- 时序自然保留——每个 Drawer 记录了对话发生的时间,天然支持「上周我们讨论了什么」这类查询
2.2 技术架构:三层设计
MemPalace 采用经典的分层架构:
┌─────────────────────────────────────────────────────────┐
│ MCP Protocol Layer │
│ (29+ MCP Tools for Claude Code/Cursor/etc) │
├─────────────────────────────────────────────────────────┤
│ Memory Manager │
│ ┌─────────────┬─────────────┬─────────────────────┐ │
│ │ Mining │ Embedding │ Retrieval │ │
│ │ Engine │ Service │ Engine │ │
│ └─────────────┴─────────────┴─────────────────────┘ │
├─────────────────────────────────────────────────────────┤
│ Storage Layer │
│ ┌─────────────┬─────────────┬─────────────────────┐ │
│ │ ChromaDB │ AAAK │ Knowledge │ │
│ │ (Vector) │ Compressor │ Graph │ │
│ └─────────────┴─────────────┴─────────────────────┘ │
└─────────────────────────────────────────────────────────┘
2.3 AAAK 压缩:保留完整信息的同时优化存储
MemPalace 的杀手锏之一是 AAAK (Asymmetric Adaptive Approximation Kompression) 压缩算法。
传统向量数据库存储的是 embedding 向量(通常 768 或 1536 维),每条记录占用大量空间。AAAK 的思路是:
class AAAKCompressor:
"""
AAAK 压缩原理:
1. 不是压缩向量本身,而是压缩「相似向量的表示」
2. 利用对话的时序性:相邻对话往往语义相近
3. 使用 delta 编码 + 量化
"""
def compress(self, embeddings: np.ndarray) -> bytes:
# 1. 找到基准向量(cluster centroid)
centroids = self._find_clusters(embeddings)
# 2. 每个向量存储为 (centroid_id, delta)
# delta 用低精度浮点数(int8 instead of float32)
deltas = self._compute_deltas(embeddings, centroids)
# 3. 应用熵编码
compressed = self._entropy_encode(deltas)
return compressed
def decompress(self, compressed: bytes) -> np.ndarray:
# 无损解压回原始 embedding
return self._entropy_decode(compressed)
效果:
- 存储空间减少 60-70%
- 压缩/解压延迟 < 5ms
- 无损:召回率不受影响
2.4 混合检索策略
MemPalace 不只依赖向量相似度,而是混合检索:
class HybridRetriever:
"""
混合检索策略:
1. 语义检索(向量相似度)—— 找「语义相关」的内容
2. 关键词检索(BM25)—— 找「精确匹配」的内容
3. 时序检索 —— 找「时间范围内」的内容
4. 结构检索 —— 在特定 Room/Drawer 内搜索
"""
def retrieve(
self,
query: str,
room_hint: Optional[str] = None,
time_range: Optional[Tuple[datetime, datetime]] = None,
top_k: int = 5
) -> List[Memory]:
candidates = []
# 1. 语义检索
query_embedding = self.embed(query)
semantic_results = self.vector_db.search(query_embedding, top_k * 3)
candidates.extend(semantic_results)
# 2. 关键词检索(BM25)
keyword_results = self.bm25.search(query, top_k * 2)
candidates.extend(keyword_results)
# 3. 时序过滤
if time_range:
candidates = [c for c in candidates
if time_range[0] <= c.timestamp <= time_range[1]]
# 4. 结构过滤
if room_hint:
candidates = [c for c in candidates
if c.room_name == room_hint]
# 5. Rerank(可选,需要 API)
if self.use_reranker:
candidates = self.rerank(query, candidates)
return self._dedupe_and_rank(candidates)[:top_k]
为什么混合检索重要?
单一检索方式有盲区:
| 检索方式 | 擅长 | 不擅长 |
|---|---|---|
| 向量相似度 | 语义相关、同义词、模糊匹配 | 精确关键词、代码标识符 |
| BM25 | 精确关键词、代码标识符 | 同义词、语义理解 |
| 时序检索 | 「上周」「昨天」这类查询 | 语义相关 |
MemPalace 的混合检索结合三者优势,在 LongMemEval 上达到 96.6% R@5(纯本地)和 100% R@5(带 reranker)。
三、实战:从零搭建 MemPalace
3.1 安装与初始化
MemPalace 用 Python 编写,安装非常简单:
# 方法一:pip 安装
pip install mempalace
# 方法二:从源码安装(推荐,获取最新功能)
git clone https://github.com/MemPalace/mempalace.git
cd mempalace
pip install -e .
# 初始化
mp init
初始化会创建默认配置:
~/.mempalace/
├── config.yaml # 配置文件
├── palaces/ # 数据目录
│ └── default/
│ ├── chroma.db # ChromaDB 向量数据库
│ ├── memories/ # 原始记忆文件
│ └── graph.json # 知识图谱
└── logs/
3.2 核心配置解析
config.yaml 是 MemPalace 的核心配置:
# ~/.mempalace/config.yaml
# 宫殿配置
palace:
name: "default"
storage_path: "~/.mempalace/palaces/default"
# 嵌入模型配置
embedding:
# 本地模型(默认,无需 API)
model: "sentence-transformers/all-MiniLM-L6-v2"
device: "cpu" # 或 "cuda" / "mps"
# 或使用 OpenAI embedding(需要 API key)
# provider: "openai"
# model: "text-embedding-3-small"
# api_key: "${OPENAI_API_KEY}"
# 挖掘配置(自动从对话中提取记忆)
mining:
enabled: true
batch_size: 100
# 自动分类规则
auto_classify: true
# 最小记忆长度(字符)
min_length: 50
# 检索配置
retrieval:
# 混合检索权重
semantic_weight: 0.6
keyword_weight: 0.3
time_weight: 0.1
# 默认返回数量
default_top_k: 5
# Reranker(可选)
reranker:
enabled: false # 开启需要 API
model: "cross-encoder/ms-marco-MiniLM-L-6-v2"
# MCP 配置
mcp:
enabled: true
port: 8080
3.3 MCP 集成:让 Claude Code 拥有记忆
MemPalace 最大的亮点是 29+ MCP 工具,深度集成 Claude Code、Cursor 等 AI 编程工具。
添加 MemPalace MCP 到 Claude Code:
# 方法一:编辑 ~/.claude/config.json
{
"mcpServers": {
"mempalace": {
"command": "uv",
"args": [
"--directory",
"/path/to/mempalace",
"run",
"mempalace-mcp"
]
}
}
}
# 方法二:命令行添加
claude mcp add mempalace -- uv --directory /path/to/mempalace run mempalace-mcp
可用的 MCP 工具(部分):
# 记忆存储
mempalace_store_memory(
content="讨论了使用 FastAPI 作为后端框架",
room="python-web-server",
metadata={"project": "my-app", "type": "decision"}
)
# 记忆检索
mempalace_recall(
query="我们之前讨论的 API 框架选型",
room="python-web-server",
top_k=5
)
# 创建房间
mempalace_create_room(
name="database-design",
description="数据库 schema 设计讨论"
)
# 列出房间
mempalace_list_rooms()
# 按时间检索
mempalace_recall_by_time(
query="性能优化",
start_time="2024-03-01",
end_time="2024-03-31"
)
# 知识图谱查询
mempalace_query_graph(
entity="FastAPI",
relation="alternative_to"
)
3.4 实战案例:让 AI 记住你的项目
假设你正在开发一个电商系统,让我们演示 MemPalace 如何工作:
第一天:架构讨论
# 自动挖掘(无需手动调用)
# 当你在 Claude Code 中讨论时,MemPalace 自动存储:
"""
[2024-03-15 14:32] Room: ecommerce-system
内容:
User: 我们的后端用什么框架?
Claude: 建议使用 FastAPI,原因是:
1. 异步性能好,适合高并发场景
2. 自动生成 OpenAPI 文档
3. 类型提示友好,配合 Pydantic 做验证
User: 数据库呢?
Claude: 建议使用 PostgreSQL:
1. 支持 JSON 查询,适合商品属性这种半结构化数据
2. 事务支持完善
3. 有成熟的分库分表方案(如 Citus)
决策记录:
- 后端框架:FastAPI
- 数据库:PostgreSQL
- 部署方式:Docker Compose + Kubernetes
"""
第二天:继续讨论
# 你可以直接问:
# "我们昨天确定了什么技术栈?"
# MemPalace 自动召回:
"""
从 Room: ecommerce-system, Drawer: 2024-03-15 找到:
技术栈决策(2024-03-15):
- 后端框架:FastAPI
- 数据库:PostgreSQL
- 部署方式:Docker Compose + Kubernetes
原始讨论片段:
Claude: 建议使用 FastAPI,原因是...
"""
一周后:性能优化
# 你问:
# "我们的 API 响应有点慢,有什么优化建议?"
# MemPalace 召回相关记忆:
"""
从 Room: ecommerce-system 找到相关讨论:
1. [2024-03-15] 架构决策:使用 FastAPI(异步框架)
2. [2024-03-18] 缓存策略:讨论使用 Redis 做热点数据缓存
3. [2024-03-20] 数据库索引:商品表添加了 category_id 索引
可能相关的优化方向:
- 检查 Redis 缓存命中率
- 分析慢查询日志
- 考虑异步数据库驱动(asyncpg)
"""
这就是 MemPalace 的价值:AI 不再每次从零开始,而是基于「你的历史」给出建议。
四、深入源码:核心模块解析
4.1 记忆挖掘引擎
MemPalace 的 Mining Engine 负责从对话中自动提取「值得记忆」的内容:
# mempalace/mining/engine.py
class MiningEngine:
"""
记忆挖掘引擎:
1. 监听对话流
2. 识别「信息密集」的片段
3. 自动分类并存储
"""
def __init__(self, config: MiningConfig):
self.classifier = TopicClassifier()
self.extractor = InfoExtractor()
self.storage = MemoryStorage()
async def mine_conversation(
self,
messages: List[Message]
) -> List[Memory]:
"""
从对话中挖掘记忆
"""
memories = []
for segment in self._segment(messages):
# 1. 判断是否值得记忆
if not self._is_worth_remembering(segment):
continue
# 2. 提取核心信息
info = self.extractor.extract(segment)
# 3. 自动分类
topic = self.classifier.classify(segment)
# 4. 生成 embedding
embedding = self._embed(info.summary)
# 5. 存储
memory = Memory(
id=self._generate_id(),
content=segment.raw_text, # 保留原文
summary=info.summary,
embedding=embedding,
topic=topic,
timestamp=segment.timestamp,
metadata=info.metadata
)
self.storage.store(memory)
memories.append(memory)
return memories
def _is_worth_remembering(self, segment: Segment) -> bool:
"""
判断是否值得记忆的启发式规则:
1. 包含决策/结论
2. 包含代码片段
3. 包含问题解决过程
4. 信息密度高(不是简单的 yes/no 对话)
"""
patterns = [
r"决定|确定|选择|方案", # 决策词
r"```python|```javascript|```go", # 代码块
r"问题|原因|解决|修复", # 问题解决
r"因为|所以|导致", # 因果关系
]
for pattern in patterns:
if re.search(pattern, segment.raw_text):
return True
# 长度阈值
return len(segment.raw_text) >= self.config.min_length
4.2 向量存储与检索
MemPalace 使用 ChromaDB 作为向量存储:
# mempalace/storage/vector_store.py
class ChromaVectorStore:
"""
ChromaDB 向量存储封装
"""
def __init__(self, path: str, embedding_dim: int = 384):
self.client = chromadb.PersistentClient(path=path)
self.collection = self.client.get_or_create_collection(
name="memories",
metadata={"hnsw:space": "cosine"}
)
self.embedding_dim = embedding_dim
def store(
self,
memory: Memory,
batch: bool = False
):
"""存储记忆"""
if batch:
# 批量存储优化
self._batch_store([memory])
else:
self.collection.add(
ids=[memory.id],
embeddings=[memory.embedding.tolist()],
documents=[memory.content],
metadatas=[{
"room": memory.room,
"timestamp": memory.timestamp.isoformat(),
"topic": memory.topic,
**memory.metadata
}]
)
def search(
self,
query_embedding: np.ndarray,
top_k: int = 5,
filter: Optional[Dict] = None
) -> List[Tuple[Memory, float]]:
"""向量检索"""
where = None
if filter:
where = self._build_filter(filter)
results = self.collection.query(
query_embeddings=[query_embedding.tolist()],
n_results=top_k,
where=where,
include=["documents", "metadatas", "distances"]
)
memories = []
for i, doc in enumerate(results["documents"][0]):
memory = Memory(
id=results["ids"][0][i],
content=doc,
embedding=None, # 不返回 embedding 节省内存
metadata=results["metadatas"][0][i]
)
distance = results["distances"][0][i]
memories.append((memory, 1 - distance)) # 转换为相似度
return memories
4.3 MCP Server 实现
MemPalace 的 MCP Server 是一个标准的 Python MCP 实现:
# mempalace/mcp/server.py
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
class MemPalaceMCPServer:
"""
MemPalace MCP Server
提供 29+ 工具给 Claude Code / Cursor 等 MCP 客户端
"""
def __init__(self, palace: Palace):
self.palace = palace
self.server = Server("mempalace")
self._register_tools()
def _register_tools(self):
"""注册所有 MCP 工具"""
# 工具 1: 存储记忆
@self.server.list_tools
async def list_tools():
return [
Tool(
name="mempalace_store",
description="Store a memory in MemPalace",
inputSchema={
"type": "object",
"properties": {
"content": {"type": "string"},
"room": {"type": "string", "default": "default"},
"metadata": {"type": "object"}
},
"required": ["content"]
}
),
# ... 其他 28 个工具
]
@self.server.call_tool
async def call_tool(name: str, arguments: dict):
if name == "mempalace_store":
memory = await self.palace.store(
content=arguments["content"],
room=arguments.get("room", "default"),
metadata=arguments.get("metadata", {})
)
return [TextContent(
type="text",
text=f"Stored memory {memory.id} in room '{memory.room}'"
)]
elif name == "mempalace_recall":
results = await self.palace.recall(
query=arguments["query"],
room=arguments.get("room"),
top_k=arguments.get("top_k", 5)
)
return [TextContent(
type="text",
text=self._format_results(results)
)]
# ... 其他工具处理
async def run(self):
"""启动 MCP Server"""
async with stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
self.server.create_initialization_options()
)
五、性能优化与生产部署
5.1 大规模数据的性能优化
当记忆数量增长到 10 万+ 时,性能优化变得关键:
# 优化 1: 批量索引构建
class BatchIndexer:
def build_index(self, memories: List[Memory]):
# 每 10000 条批量构建索引
batch_size = 10000
for i in range(0, len(memories), batch_size):
batch = memories[i:i+batch_size]
self.vector_store.store_batch(batch)
# 释放内存
del batch
gc.collect()
# 优化 2: 增量索引
class IncrementalIndexer:
def add_memory(self, memory: Memory):
# 只更新增量,不重建全量索引
self.vector_store.store(memory)
# 更新 HNSW 图
self.hnsw_graph.add(memory.id, memory.embedding)
# 优化 3: 分片存储
class ShardedStorage:
def __init__(self, shard_count: int = 10):
self.shards = [ChromaVectorStore(f"shard_{i}")
for i in range(shard_count)]
def get_shard(self, memory_id: str) -> ChromaVectorStore:
# 按 memory_id hash 分片
shard_idx = hash(memory_id) % len(self.shards)
return self.shards[shard_idx]
5.2 Docker 部署
生产环境推荐 Docker 部署:
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# 安装 MemPalace
COPY . .
RUN pip install -e .
# 数据卷
VOLUME /data/mempalace
# MCP Server 入口
CMD ["python", "-m", "mempalace.mcp.server"]
# docker-compose.yml
version: '3.8'
services:
mempalace:
build: .
ports:
- "8080:8080"
volumes:
- ./data:/data/mempalace
environment:
- MEMPALACE_STORAGE_PATH=/data/mempalace
- MEMPALACE_EMBEDDING_DEVICE=cpu
restart: unless-stopped
5.3 监控与告警
# 使用 Prometheus 监控
from prometheus_client import Counter, Histogram, start_http_server
# 指标定义
MEMORY_STORE_COUNT = Counter(
'mempalace_memory_store_total',
'Total memories stored'
)
RETRIEVAL_LATENCY = Histogram(
'mempalace_retrieval_latency_seconds',
'Memory retrieval latency'
)
# 使用
class MonitoredPalace:
def store(self, memory: Memory):
MEMORY_STORE_COUNT.inc()
return self.palace.store(memory)
def recall(self, query: str, **kwargs):
with RETRIEVAL_LATENCY.time():
return self.palace.recall(query, **kwargs)
# 启动 metrics server
start_http_server(9090)
六、MemPalace vs 竞品对比
| 特性 | MemPalace | Mem0 | Letta | 传统 RAG |
|---|---|---|---|---|
| 存储方式 | 完整原文 | 摘要压缩 | 摘要+事实 | 向量片段 |
| 本地优先 | ✅ | ✅ | ✅ | ✅ |
| 无需 API | ✅ | ❌ | ❌ | ❌ |
| LongMemEval R@5 | 96.6% | 89.2% | 91.5% | 78.3% |
| MCP 集成 | 29+ 工具 | 12 工具 | 8 工具 | - |
| 知识图谱 | ✅ | ❌ | ✅ | ❌ |
| 时序检索 | ✅ | ❌ | ✅ | ❌ |
| 开源协议 | MIT | MIT | Apache 2.0 | - |
七、最佳实践与踩坑指南
7.1 Room 命名规范
# ✅ 好的命名
rooms = [
"project-ecommerce-backend", # 项目+模块
"tech-kubernetes-deployment", # 技术+场景
"debug-authentication-issue", # 问题+主题
]
# ❌ 不好的命名
rooms = [
"room1", # 无语义
"讨论", # 过于宽泛
"2024-03-15", # 只有时间,不知道主题
]
7.2 避免过度挖掘
# 不要把所有对话都存储
# 配置过滤规则
mining:
filters:
# 忽略简单问候
ignore_patterns:
- "^(hi|hello|thanks|ok|好的|谢谢)"
# 忽略过短内容
min_length: 50
# 忽略纯代码执行(没有讨论)
ignore_code_only: true
7.3 定期清理过期记忆
# 定期清理不再相关的记忆
def cleanup_old_memories(palace: Palace, days: int = 180):
cutoff = datetime.now() - timedelta(days=days)
old_memories = palace.query(
filter={"timestamp": {"$lt": cutoff.isoformat()}}
)
for memory in old_memories:
# 检查是否仍然相关
if not is_still_relevant(memory):
palace.delete(memory.id)
八、未来展望
MemPalace 团队正在开发的功能:
- 多模态记忆:支持存储图片、音频、视频的记忆
- 协作记忆:团队共享的知识库
- 主动召回:AI 主动从记忆中找到相关信息并建议
- 知识演化追踪:追踪某个知识点的变化历史
九、总结
MemPalace 解决了 AI Agent 长期记忆的根本问题:
- 不删减:保留完整对话原文,细节不丢失
- 结构化:宫殿-房间-抽屉三层组织,检索高效
- 本地优先:无需 API key、无隐私风险、无订阅成本
- 高召回:96.6% R@5,混合检索策略
- 深度集成:29+ MCP 工具,与 Claude Code 无缝协作
对于任何需要 AI 「记住」你项目、偏好、决策的开发者来说,MemPalace 是目前最佳的开源选择。
一天 2 万 Star 不是偶然,而是真正解决了痛点。
项目地址:https://github.com/MemPalace/mempalace
文档:https://docs.mempalace.ai
许可证:MIT(完全免费商用)