1. MemPalace 记忆挖掘流水线架构解析
MemPalace 是一个本地化的 AI Agent 记忆系统,其核心设计理念是"No API key. No internet. Everything local."。整个系统采用"宫殿/翼/房间/抽屉"的隐喻架构,其中抽屉(drawer)是 ChromaDB 中的基本存储单元,每条记录都带有 wing、room、source_file 等元数据字段。
记忆系统的入口是 mining(挖掘)流水线,它负责将原始数据转化为结构化的记忆单元。与大多数 RAG 系统不同,MemPalace 采用了三种独立的挖掘模式,分别针对不同类型的输入数据:
- Projects 模式:处理代码仓库和文档目录
- Convos 模式:处理各种聊天记录导出
- General 模式:处理任意散文文本
这种设计源于一个关键认知:不同类型的内容具有不同的"最小语义单元"。代码文档的最小单元是段落或函数注释,对话的最小单元是问答对(Q+A),而散文的最小单元则是包含完整思想的段落。
需要模型API调用? 免费领10W Token,多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. Projects 模式:代码仓库的结构化处理
2.1 文件发现与过滤机制
miner.py 是 projects 模式的核心实现,其文件发现机制采用三层过滤:
python复制READABLE_EXTENSIONS = {
".txt", ".md", ".py", ".js", ".ts", ".jsx", ".tsx",
".json", ".yaml", ".yml", ".html", ".css",
".java", ".go", ".rs", ".rb", ".sh", ".csv", ".sql", ".toml",
}
SKIP_DIRS = {
".git", "node_modules", "__pycache__",
".venv", "venv", "env", "dist", "build", ".next",
"coverage", ".mempalace", ".ruff_cache", ".mypy_cache",
# ...其他常见构建目录
}
过滤流程遵循严格顺序:
- 先通过
SKIP_DIRS快速排除明显不需要的目录 - 再应用
.gitignore规则进行精确过滤 - 最后检查文件扩展名白名单
这种分层过滤的设计考虑到了性能因素:目录级过滤比逐文件检查快一个数量级。
2.2 内容切分策略
Projects 模式采用滑动窗口算法进行内容切分,关键参数如下:
python复制CHUNK_SIZE = 800 # 字符数
CHUNK_OVERLAP = 100 # 重叠字符数
MIN_CHUNK_SIZE = 50 # 最小有效块大小
切分算法有几个值得注意的特点:
- 优先在段落边界(
\n\n)处切分 - 其次在行边界(
\n)处切分 - 确保每个 chunk 至少有 400 字符的有效内容(通过
CHUNK_SIZE//2检查) - 保留 100 字符的重叠以避免语义断裂
这种设计确保了代码注释和文档段落能保持完整,同时控制了 chunk 的信息密度。
2.3 房间分配逻辑
房间(room)分配采用四级优先级策略:
- 路径匹配:检查文件路径是否包含房间名或关键词
- 文件名匹配:检查文件名是否包含房间名
- 内容关键词计数:统计前2000字符中的关键词出现次数
- 通用房间:以上都不匹配时归入"general"房间
这种设计充分利用了代码仓库的固有结构,使得相关文件能自动归类到合适的房间。
3. Convos 模式:对话记录的智能处理
3.1 对话切分的特殊挑战
与 projects 模式不同,对话记录的最小语义单元是"问答对"(Q+A pair)。convo_miner.py 实现了专门的对话切分逻辑:
python复制def chunk_exchanges(content: str) -> list:
lines = content.split("\n")
quote_lines = sum(1 for line in lines if line.strip().startswith(">"))
if quote_lines >= 3: # 至少有3个引用行才认为是对话
return _chunk_by_exchange(lines)
else:
return _chunk_by_paragraph(content)
这种自适应策略能正确处理各种边界情况,包括格式不规范的对话记录。
3.2 五种聊天格式的归一化处理
normalize.py 模块负责将不同来源的聊天记录转换为统一格式。支持的格式包括:
- Claude Code JSONL:每行一个JSON对象,包含type和message字段
- OpenAI Codex CLI JSONL:区分event_msg和response_item类型
- Claude.ai 导出:支持flat和privacy export两种变体
- ChatGPT mapping树:处理对话分支和编辑历史
- Slack DM:多人对话的智能角色分配
以ChatGPT格式处理为例,它需要特别处理对话分支:
python复制def _try_chatgpt_json(data) -> Optional[str]:
if not isinstance(data, dict) or "mapping" not in data:
return None
# 查找根节点
root_id = None
for node_id, node in mapping.items():
if node.get("parent") is None and node.get("message") is None:
root_id = node_id
break
# 只追踪主分支(children[0])
current_id = root_id
while current_id:
node = mapping.get(current_id, {})
# ...处理消息内容...
current_id = node.get("children", [])[0] if node.get("children") else None
这种设计确保只保留用户最终看到的对话主线,忽略被丢弃的编辑分支。
4. General 模式:任意文本的智能分类
4.1 五类记忆标记系统
general_extractor.py 实现了启发式的文本分类系统,将内容分为五类:
- Decisions:包含明确决策的文本
- Preferences:表达个人偏好的内容
- Milestones:重要进展或突破
- Problems:遇到的问题或挑战
- Emotional:带有情绪色彩的表达
分类基于正则表达式标记和打分系统实现:
python复制MARKER_PATTERNS = {
"decision": r"\b(decided|chose|selected|opted for)\b",
"preference": r"\b(prefer|like|favor|would rather)\b",
# ...其他模式...
}
def classify_segment(text: str) -> Optional[str]:
scores = {}
for category, pattern in MARKER_PATTERNS.items():
matches = re.findall(pattern, text, re.IGNORECASE)
if matches:
scores[category] = len(matches)
if scores:
best = max(scores, key=scores.get)
if scores[best] >= MIN_CONFIDENCE[best]:
return best
return None
4.2 消歧与置信度控制
系统还实现了复杂的消歧逻辑,例如:
python复制def disambiguate(text: str, category: str) -> str:
if category == "problem" and contains_positive_sentiment(text):
return "milestone" # 将"问题+正面情绪"升级为里程碑
# ...其他消歧规则...
return category
这种细粒度的分类使得记忆检索更加精准,能根据不同类型的查询返回最相关的内容。
5. 实体识别与房间检测
5.1 两步式实体识别
entity_detector.py 采用了两阶段实体识别策略:
- 候选提取:通过正则表达式和启发式规则找出可能的人名和项目名
- 评分过滤:根据上下文信号对候选进行评分和过滤
python复制def extract_entities(text: str) -> dict:
candidates = extract_candidates(text)
return {
"persons": [c for c in candidates if score_entity(c, "person") > PERSON_THRESHOLD],
"projects": [c for c in candidates if score_entity(c, "project") > PROJECT_THRESHOLD]
}
5.2 本地房间检测
room_detector_local.py 实现了离线房间发现功能,它分析文本内容并推荐合适的房间分配。算法结合了:
- 关键词频率统计
- 实体共现分析
- 预设的房间-关键词映射
这种设计使得系统能在完全离线的环境下实现智能的房间分配。
6. 性能优化与边界处理
MemPalace 在性能优化方面做了诸多考虑:
- 增量更新:通过文件修改时间(mtime)跳过未变更文件
- 懒加载:只在需要时解析文件内容
- 缓存机制:重复使用的解析器保持缓存状态
- 快速失败:对明显不符合格式的文件快速返回
以增量更新为例,其实现非常高效:
python复制def file_already_mined(collection, source_file: str) -> bool:
results = collection.get(where={"source_file": source_file}, limit=1)
if not results.get("ids"):
return False
stored_mtime = results["metadatas"][0].get("source_mtime")
current_mtime = os.path.getmtime(source_file)
return float(stored_mtime) == current_mtime
这种设计使得大型代码仓库的重复挖掘几乎可以瞬间完成。
7. 设计哲学与实现考量
MemPalace 的挖掘流水线体现了几个核心设计原则:
- 零大模型依赖:全部使用规则和启发式方法,确保完全本地运行
- 语义感知切分:不同类型内容采用不同的切分策略
- 渐进式增强:从简单规则开始,逐步增加复杂情况处理
- 确定性与可重现:相同的输入总是产生相同的输出
这种设计使得系统具有以下优势:
- 完全离线工作,保护隐私
- 运行速度快,适合大规模数据处理
- 行为可预测,便于调试
- 无API调用成本
8. 实践经验与技巧
在实际使用 MemPalace 进行记忆挖掘时,有几个实用技巧:
- 项目初始化:为代码仓库创建
mempalace.yaml文件明确定义房间结构 - 强制重挖:删除
.mempalace目录可以强制重新处理所有文件 - 格式检查:使用
normalize.py的独立函数验证聊天记录格式 - 自定义规则:通过继承和重写关键函数实现特定领域的处理逻辑
对于特别大的代码仓库,建议采用分阶段挖掘策略:
- 先处理核心目录
- 再处理测试和示例代码
- 最后处理文档和辅助文件
这种策略可以优先保证关键记忆的质量和相关性。
