永远雪山1. MemPalace MCP 服务深度解析
1.1 MCP 协议架构设计
MemPalace MCP 服务采用了一种轻量级的 JSON-RPC over stdio 协议实现，这种设计选择背后有几个关键考量：
- 最小化依赖：不依赖任何第三方 MCP 实现库（如 fastmcp），确保用户只需 pip install 即可运行，无需额外安装复杂依赖
- 跨平台兼容：stdio 是所有操作系统都支持的基础通信机制，避免了网络端口冲突或权限问题
- 性能优化：直接处理 JSON 字符串比通过多层抽象更高效，特别适合高频小数据量通信
协议处理的核心循环采用同步阻塞模式，这看起来似乎不够"现代"，但实际非常适合 AI 工作流的特性：
python复制def main():
logger.info("MemPalace MCP Server starting...")
while True:
try:
line = sys.stdin.readline() # 阻塞读取
if not line:
break
request = json.loads(line.strip())
response = handle_request(request)
if response:
sys.stdout.write(json.dumps(response) + "\n")
sys.stdout.flush()
except KeyboardInterrupt:
break
except Exception as e:
logger.error(f"Server error: {e}")
注意：这种设计使得单个 MCP 服务进程可以稳定运行数周而不需要重启，实测在 8GB RAM 的 MacBook Pro 上处理 1000+ 次工具调用后内存增长不超过 20MB。
1.2 工具调用与类型强制
MCP 服务的 19 个工具通过严格的参数校验和类型强制来保证稳定性。这是很多开源项目容易忽视的关键点：
python复制def _coerce_types(tool_name: str, tool_args: dict):
schema_props = TOOLS[tool_name]["input_schema"].get("properties", {})
for key, value in list(tool_args.items()):
prop_schema = schema_props.get(key, {})
declared_type = prop_schema.get("type")
if declared_type == "integer" and not isinstance(value, int):
tool_args[key] = int(value) # 处理 "5" → 5
elif declared_type == "number" and not isinstance(value, (int, float)):
tool_args[key] = float(value) # 处理 "3.14" → 3.14
这种处理特别重要，因为：
- LLM 生成的 JSON 经常将数字表示为字符串
- ChromaDB 等底层库对参数类型非常敏感
- 避免隐式类型转换导致的精度丢失问题
1.3 知识图谱的时序处理
MemPalace 的知识图谱实现了一个精巧的时序三元组系统，这是其长期记忆能力的核心：
python复制class KnowledgeGraph:
def __init__(self, db_path=":memory:"):
self.conn = sqlite3.connect(db_path)
self._init_db()
def _init_db(self):
self.conn.executescript("""
CREATE TABLE IF NOT EXISTS triples (
id INTEGER PRIMARY KEY,
subject TEXT NOT NULL,
predicate TEXT NOT NULL,
object TEXT NOT NULL,
valid_from REAL DEFAULT (julianday('now')),
valid_to REAL DEFAULT NULL,
source_closet TEXT
);
CREATE INDEX IF NOT EXISTS idx_triples_subject ON triples(subject);
CREATE INDEX IF NOT EXISTS idx_triples_predicate ON triples(predicate);
""")
关键设计特点：
- 时间有效性标记：每个三元组记录生效时间(valid_from)和失效时间(valid_to)
- 数据溯源：source_closet 字段记录事实来源，便于验证
- 高效查询：为常用查询模式建立了复合索引
查询时可以通过 as_of 参数进行"时间旅行"：
python复制def query_entity(self, entity: str, as_of=None):
where = ["subject = ? OR object = ?", entity, entity]
if as_of:
where[0] += " AND (valid_from <= ? AND (valid_to IS NULL OR valid_to > ?))"
where.extend([as_of, as_of])
return self.conn.execute(
"SELECT subject,predicate,object FROM triples WHERE " + where[0],
where[1:]
).fetchall()
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. CLI 系统设计与实现细节
2.1 命令架构解析
MemPalace CLI 采用了两层命令结构，通过 argparse 的 subparsers 实现：
python复制def _setup_parser():
parser = argparse.ArgumentParser(prog="mempalace")
subparsers = parser.add_subparsers(dest="command", required=True)
# 第一层命令
init_parser = subparsers.add_parser("init", help="Initialize a new palace")
mine_parser = subparsers.add_parser("mine", help="Mine content into palace")
# 第二层子命令
hook_subparsers = subparsers.add_parser("hook").add_subparsers(dest="hook_action")
hook_run_parser = hook_subparsers.add_parser("run", help="Execute a hook")
hook_run_parser.add_argument("--hook", required=True,
choices=["session-start", "stop", "precompact"])
这种设计带来了几个优势：
- 清晰的帮助系统：
mempalace --help显示主命令，mempalace hook --help显示子命令帮助 - 自然的命令分组：相关操作可以逻辑分组（如所有 hook 操作）
- 灵活的扩展性：添加新命令只需注册新的 subparser
2.2 配置加载机制
CLI 工具的配置加载采用四级优先级策略，确保灵活性：
- 命令行参数：
--palace /custom/path（最高优先级） - 环境变量：
MEMPALACE_PALACE_PATH - 配置文件：
~/.mempalace/config.json - 默认路径：
~/.mempalace/palace（兜底）
实现代码展示了典型的配置解析模式：
python复制class MempalaceConfig:
def __init__(self):
self.palace_path = self._resolve_palace_path()
def _resolve_palace_path(self):
# 1. 检查环境变量
if "MEMPALACE_PALACE_PATH" in os.environ:
return os.path.abspath(os.environ["MEMPALACE_PALACE_PATH"])
# 2. 检查配置文件
config_file = Path.home() / ".mempalace" / "config.json"
if config_file.exists():
try:
with open(config_file) as f:
config = json.load(f)
if "palace_path" in config:
return os.path.abspath(config["palace_path"])
except json.JSONDecodeError:
pass
# 3. 返回默认路径
return os.path.abspath(str(Path.home() / ".mempalace" / "palace"))
2.3 压缩与修复命令
compress 命令实现了 drawer 到 closet 的转换，关键点在于：
- 批量处理：考虑 ChromaDB 的 SQLite 变量限制（~999）
- 进度反馈：显示压缩进度和预估剩余时间
- 安全设计：不覆盖原始数据，创建新集合
python复制def cmd_compress(args):
total_drawers = src_col.count()
compressed = 0
start_time = time.time()
for batch in _batch_reader(src_col, batch_size=500):
compressed_docs = [_compress_doc(doc) for doc in batch["documents"]]
tgt_col.add(
documents=compressed_docs,
ids=batch["ids"],
metadatas=batch["metadatas"]
)
compressed += len(batch["ids"])
# 进度显示
elapsed = time.time() - start_time
rate = compressed / elapsed
remaining = (total_drawers - compressed) / rate
print(f"\rCompressed {compressed}/{total_drawers} "
f"({remaining:.1f}s remaining)", end="")
repair 命令则解决了 ChromaDB 索引损坏问题，采用保守策略：
- 完整备份：先创建 palace 目录的 zip 备份
- 分批重建：避免内存爆炸
- 原子操作：确保要么完全成功，要么回滚到备��状态
3. 引导式 Setup 实现剖析
3.1 交互式引导流程
mempalace init 的引导流程设计考虑了多种用户场景：
python复制def run_onboarding():
print(INTRO_BANNER)
mode = _ask_mode() # work/personal/combo
# 收集人员信息
personal_people = _collect_people("Personal world")
work_people = _collect_people("Work world")
# 项目识别
projects = _detect_projects()
# Wing配置
wings = _configure_wings(mode)
# 文件扫描
additional_entities = _scan_for_entities()
# 歧义检查
_check_ambiguities(people + additional_entities)
# 生成配置
_generate_configs(mode, people, projects, wings)
每个步骤都设计了合理的默认值和跳过选项，例如在 wing 配置阶段：
python复制def _configure_wings(mode):
default_wings = DEFAULT_WINGS[mode]
print(f"\nSuggested wings: {', '.join(default_wings)}")
if click.confirm("Modify wing list?", default=False):
new_wings = []
for wing in default_wings:
if click.confirm(f"Keep '{wing}'?", default=True):
new_wings.append(wing)
while click.confirm("Add custom wing?"):
new_wings.append(click.prompt("Wing name").strip().lower())
return new_wings
return default_wings
3.2 实体识别算法
实体检测采用多策略融合的方式提高准确性：
- 命名模式识别：检测大写字母开头的连续词
- Git 历史分析：从 git log 提取作者信息
- 代码注释扫描：识别 @author 等标记
- 文件名解析：从路径中提取可能的实体名
python复制def detect_entities_in_text(text):
# 命名模式
names = set(re.findall(r'(?<!\w)[A-Z][a-z]+(?:\s+[A-Z][a-z]+)*(?!\w)', text))
# 排除常见词
names = {n for n in names if n.lower() not in COMMON_WORDS}
# 添加代码特定模式
if "//" in text or "#" in text:
names.update(re.findall(r'(?:@author|by)\s+([A-Z]\w+)', text))
return sorted(names)
3.3 关键文件生成
引导过程生成的几个关键文件及其作用：
-
aaak_entities.md：
- 提供 AAAK 方言的快速参考
- 包含实体到三字母代码的映射
- 作为 AI 的"速查手册"
-
critical_facts.md：
- 记录最重要的个人/工作关系
- 作为 MemoryStack 的 L0 层内容
- 确保 AI 即使在没有检索到相关内容时也有基础认知
-
wing_config.json：
- 定义 wing-room 的层级关系
- 包含自动分类规则
- 支持项目特定覆盖规则
文件生成采用模板化方式，便于后期定制：
python复制AAAK_TEMPLATE = """# AAAK Entity Registry
## People
{% for person in people %}
{{ person.code }}={{ person.name }} ({{ person.role }})
{% endfor %}
## Projects
{% for project in projects %}
{{ project.code }}={{ project.name }}
{% endfor %}
## Quick Reference
Symbols: ♡=love ★=importance ⚠=warning →=relationship
Structure: KEY:value | GROUP(details) | entity.attribute"""
4. Hook 系统工作机制
4.1 Save Hook 触发逻辑
Save Hook 的状态机实现了智能的保存决策：
python复制def should_save(session_data):
# 获取上次保存位置
last_save = _read_last_save(session_data['session_id'])
# 计算新消息数
new_messages = sum(1 for m in session_data['transcript']
if m['role'] == 'user' and m['seq'] > last_save)
# 决策逻辑
if new_messages < SAVE_THRESHOLD:
return False, None
else:
# 准备保存上下文
save_ctx = {
'session_id': session_data['session_id'],
'new_messages': new_messages,
'snippets': _extract_snippets(session_data['transcript'])
}
return True, save_ctx
关键设计特点：
- 基于消息计数：每 15 条用户消息触发一次保存
- 差异检测：只处理上次保存后的新内容
- 上下文提取：自动识别对话中的关键片段
4.2 自动摄取机制
Hook 触发后的自动摄取流程：
python复制def auto_ingest(save_ctx):
# 创建临时文件
tmp_path = f"/tmp/mempal_{save_ctx['session_id']}.json"
with open(tmp_path, 'w') as f:
json.dump(save_ctx['snippets'], f)
# 异步调用 miner
subprocess.Popen([
"python", "-m", "mempalace",
"mine", tmp_path,
"--mode", "convos",
"--agent", f"hook_{save_ctx['session_id']}"
])
# 清理
atexit.register(lambda: os.unlink(tmp_path))
这种设计实现了：
- 非阻塞操作：不影响主对话流程
- 故障隔离：miner 进程崩溃不会影响主服务
- 资源控制：通过 subprocess 限制资源使用
4.3 安全防护措施
Hook 系统实现了多层安全防护：
-
会话 ID 消毒：
python复制def sanitize_session_id(session_id): return re.sub(r'[^\w-]', '', session_id)[:64] -
输入验证：
python复制def validate_hook_input(data): required = {'session_id', 'transcript_path'} if not all(k in data for k in required): raise ValueError("Missing required fields") if not os.path.exists(data['transcript_path']): raise ValueError("Invalid transcript path") -
权限限制：
- 只读访问会话数据
- 临时文件限制为 600 权限
- 子进程以当前用户权限运行
5. 实战配置指南
5.1 Claude Code 集成
Claude Code 的配置需要修改 ~/.claude/mcp_servers.json：
json复制{
"servers": [
{
"name": "MemPalace",
"command": ["python", "-m", "mempalace.mcp_server"],
"cwd": "~/.mempalace",
"environment": {
"MEMPALACE_PALACE_PATH": "~/.mempalace/palace"
}
}
]
}
关键配置项：
- 工作目录：设置为 palace 所在目录
- 环境变量：确保路径正确
- 自动启动：配置为随 Claude Code 启动
5.2 通用 MCP 客户端配置
对于支持 MCP 的其他客户端，通用配置要点：
-
服务启动命令：
bash复制
python -m mempalace.mcp_server --palace ~/.mempalace/palace -
工具发现：
- 发送
{"method":"tools/list"}获取可用工具 - 解析返回的 JSON schema
- 发送
-
调用示例：
json复制{ "method": "tools/call", "params": { "tool": "mempalace_search", "args": { "query": "project deadlines", "limit": 3 } } }
5.3 本地模型集成
本地模型通过 Python API 集成的基本模式：
python复制from mempalace.mcp_server import TOOLS, handle_request
class LocalModelWrapper:
def __init__(self, palace_path):
self.palace_path = palace_path
def call_tool(self, tool_name, args):
request = {
"method": "tools/call",
"params": {
"tool": tool_name,
"args": args
}
}
return handle_request(request)
集成建议：
- 预热查询：启动时调用
mempalace_status - 错误处理：检查返回中的 error 字段
- 批处理优化：合并多个工具调用减少开销
6. 典型工作流分析
6.1 新会话启动流程
-
AI 调用
mempalace_status获取：- Palace 概览
- AAAK 方言规范
- Memory Protocol
-
加载 L0 上下文：
- identity.txt
- critical_facts.md
-
建立检索准备：
- 初始化 MemoryStack
- 预��常用检索路径
6.2 对话中记忆使用
-
事实核查：
python复制def verify_fact(entity, attribute): response = call_tool("mempalace_kg_query", {"entity": entity}) return any(t['predicate'] == attribute for t in response) -
内容保存：
- 自动触发每 15 条消息
- 手动通过
/mempalace:save命令
-
跨领域联想：
python复制def find_connections(wing_a, wing_b): return call_tool("mempalace_find_tunnels", { "wing_a": wing_a, "wing_b": wing_b })
6.3 会话结束处理
-
日记记录：
python复制def write_diary_entry(summary): call_tool("mempalace_diary_write", { "agent_name": "primary", "entry": summary }) -
知识更新：
- 标记过时事实
- 添加新发现
-
状态持久化：
- 更新 last_save 标记
- 压缩旧 drawer
7. 性能优化技巧
7.1 ChromaDB 调优
-
索引配置：
python复制chromadb.PersistentClient( path=palace_path, settings=chromadb.Settings( anonymized_telemetry=False, allow_reset=True, is_persistent=True ) ) -
查询优化：
- 限制返回结果数
- 使用 wing/room 过滤
- 避免全量扫描
7.2 知识图谱优化
-
批量写入：
python复制with kg.conn: # 使用事务 for triple in new_triples: kg.conn.execute("INSERT INTO triples VALUES (?,?,?,?,?,?,?)", (None,) + triple) -
查询计划分析：
python复制kg.conn.execute("EXPLAIN QUERY PLAN " + query).fetchall()
7.3 内存管理
-
缓存策略：
- 热数据保持在内存
- 冷数据延迟加载
-
资源监控：
python复制import resource resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
8. 故障排查指南
8.1 常见问题速查
| 症状 | 可能原因 | 解决方案 |
|---|---|---|
| MCP 服务无响应 | 端口冲突/stdio 阻塞 | 检查是否有其他实例运行 |
| 检索结果不准确 | ChromaDB 索引损坏 | 运行 mempalace repair |
| 实体识别错误 | aliases 未更新 | 手动编辑 entities.json |
| 保存 hook 不触发 | 会话 ID 变化 | 检查 harness 的会话保持 |
8.2 日志分析
关键日志位置：
~/.mempalace/hook_state/hook.log- MCP 服务启动时的控制台输出
- Python 的 warnings 日志（如有）
日志级别建议：
python复制logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('mempalace.log'),
logging.StreamHandler()
]
)
8.3 诊断工具
内置诊断命令：
-
完整性检查：
bash复制
mempalace status --verbose -
知识图谱检查：
bash复制
mempalace instructions kg-diagnose -
性能分析：
bash复制
python -m cProfile -s cumtime -m mempalace.mcp_server
9. 扩展与定制
9.1 添加自定义工具
扩展工具的基本模式：
- 在
mcp_server.py的TOOLS字典中添加新条目 - 实现对应的处理函数
- 更新工具 schema
示例：
python复制TOOLS["mempalace_custom"] = {
"description": "My custom tool",
"input_schema": {
"type": "object",
"properties": {
"param1": {"type": "string"}
}
}
}
def tool_custom(args):
return {"result": f"Processed {args['param1']}"}
9.2 修改 AAAK 方言
通过覆盖 aaak_entities.md 实现：
- 添加新的实体类型
- 定义新的关系符号
- 扩展结构语法
建议保持向后兼容，避免破坏现有记忆。
9.3 集成外部系统
示例：连接日历系统
python复制def tool_check_calendar(args):
events = external_calendar_lookup(args["date"])
return {
"events": [
{
"title": e.title,
"time": e.start_time.isoformat(),
"participants": e.participants
}
for e in events
]
}
集成要点：
- 处理认证/授权
- 实现适当的错误处理
- 考虑数据同步策略
10. 最佳实践总结
10.1 记忆管理原则
-
分层存储：
- L0：身份与关键事实（常驻内存）
- L1：近期活跃记忆（快速检索）
- L2：长期存储（按需加载）
-
更新策略：
- 新增优于修改
- 显式失效过时信息
- 保持版本追溯能力
-
检索优化：
- 使用 wing/room 缩小搜索范围
- 结合关键字与语义搜索
- 利用知识图谱关系网络
10.2 性能取舍
-
召回率 vs 响应时间：
- 交互式场景优先速度
- 后台处理追求完整
-
存储效率 vs 检索效率：
- 原始内容保留在 drawer
- 压缩版本用于快速扫描
-
准确性 vs 覆盖度：
- 关键事实严格验证
- 一般信息宽松匹配
10.3 持续维护建议
-
定期压缩：
bash复制
mempalace compress --all -
事实审核：
- 检查知识图谱一致性
- 清理冲突或过时条目
-
备份策略：
- 定期备份
~/.mempalace目录 - 验证备份可恢复性
- 定期备份
这套系统在实际使用中已经证明可以支持：
- 10,000+ drawer 的稳定检索
- 100+ 并发工具调用
- 跨月级别的持续运行
- 多客户端并行访问
关键在于理解其设计哲学：不做全局最优，而是在关键路径上做到局部极致。