说个我最近的折腾成果:我自己写了一个开源小工具,叫MemTether。名字是我拼出来的——Memory(记忆)+ Tether(拴绳),核心就干一件事:把多个 AI 客户端的记忆统一拴在一个地方,让它们共享同一份长期记忆。
起因很简单:我电脑上装了 Claude 桌面版、自建的 Open WebUI、随手试验的 AnythingLLM,还有命令行里跑的各种 AI 助手。每个工具各有各的会话窗口、各有各的历史记录。同一个项目背景、同一份会议纪要、同一套我调教好的规则和偏好,我在每个客户端里都要从头讲一遍。换客户端等于换一个新助手,那种感觉非常割裂。
MemTether 不是又一个聊天界面,而是一个本地优先的记忆服务。它把长期记忆抽象成一种独立于任何 AI 产品的基础设施,通过统一的 HTTP API 和 MCP(Model Context Protocol,模型上下文协议)暴露给不同客户端。这篇文章我会把它的架构思路、数据模型、核心实现、部署接入方式以及我踩过的一些坑完整写出来。如果你也在为“AI 客户端记忆割裂”头疼,或者想自己搭一个统一记忆层,这篇文章应该能给你不少参考。
1. 这个工具到底解决什么问题
1.1 多客户端记忆割裂的日常痛点
先说说我真实的使用场景。我同时维护两三个开源项目的文档和代码,日常还要处理大量技术调研和写作素材。过去我的操作习惯是:正经写东西用桌面端,比较零碎的问询用命令行助手,团队协作时又在另一个团队工具里聊。
问题在哪呢?同样是“我倾向于用 FastAPI 写后端接口”这种个人偏好,我在每个工具里都要重新说一遍。更麻烦的是,有些信息是有时间线的,比如“上周三已经决定用 SQLite 做默认存储”,如果新打开的客户端不知道这个结论,它可能又会建议我引入 PostgreSQL,然后我们重新讨论一遍,纯浪费时间。
我还遇到过更尴尬的情况:同一个问题,上午在客户端 A 里问到一半,下午换到客户端 B 里继续问,结果 B 完全不知道上午的结论。你要说这问题很大也不至于,但次数多了之后,我会下意识地避免切换客户端——这显然违背了“用合适的工具做合适的事”的初衷。作为一个喜欢折腾工具链的人,我受不了这种体验。
1.2 现有“记忆方案”为什么不够用
可能有人会说,各家 AI 产品不是都有“长期记忆”功能吗?我在部署了几个主流的自建前端和大模型平台之后发现,这里面的“长期记忆”大多有很强的产品锁定效应。
有的记忆只是服务商账号体系内的对话历史,换个客户端就没了;有的是通过“项目知识库”或“自定义指令”来实现的,知识库归知识库,聊天记录归聊天记录,两套东西还是割裂的;还有一些工具提供了“全局记忆模型”,但只能存偏好,没法存带有时间线的工作结论。如果完全靠人工,那就是复制粘贴历史上最长的提示词,让每个客户端都带上“人设设定”,这种做法不但臃肿而且极其容易过时。
我也试过直接塞一个 RAG(检索增强生成)知识库。但知识库通常要配合特定前端或特定框架才能用,换个客户端就得重写接入层。而且知识库强调的是“文档检索”,对“聊天中产生的结论性记忆”处理得很弱。我需要的是更通用、更轻量、和客户端解耦的东西。
1.3 MemTether 的设计定位
所以我在设计 MemTether 时定了几条原则。
第一,本地优先。数据留在自己机器上,不依赖任何云端账号,甚至不依赖任何一家大模型厂商。第二,客户端无关。它不试图给任何 AI 工具做私有插件,而是提供一个标准化的记忆读写接口,谁都能来对接。第三,工程上足够薄。我不想为了一个记忆工具去维护一套 PostgreSQL plus pgvector,简单轻量才是个人工具能长期活下去的关键。
它的工作方式像一个小型“记忆中枢”:所有客户端把要记得内容写过来,需要时再通过一条查询语句把相关记忆拿出来。客户端仍然负责各自的对话、推理和输出,但“长期记忆”这个职责被单独抽出来了。这就是 MemTether 的核心定位——做 AI 客户端的记忆基础设施,而不是另一个聊天机器人。
2. 整体架构与关键设计思路
2.1 三层架构:存储、服务、接入
MemTether 整体分三层。
存储层用 SQLite 单文件数据库,表结构围绕“记忆条目”设计,每条记忆包含内容、类型、命名空间、标签、来源客户端、时间戳等信息。向量索引这块我用了 sqlite-vec 扩展,直接在 SQLite 里做相似度检索,避免额外引入向量数据库。
服务层是一个 FastAPI 应用,对外提供 REST API,端口默认监听 8765。核心接口就几个:写入记忆、检索记忆、删除记忆、列命名空间。所有接口通过 Bearer Token 做简单鉴权。服务层还负责调用嵌入模型来生成向量。
接入层是面向不同 AI 客户端的那一层。我默认实现了一个 MCP Server,因为现在许多主流桌面客户端已经开始支持 MCP,一份协议能通吃很多端。另外也保留了一份 Python SDK 和几个适配示例,比如 Open WebUI 的 Function 接入、普通 HTTP 接入等。
这三层职责划分得很清楚:存储层只管存取,服务层只做业务逻辑和接口,接入层负责把能力暴露给不同客户端。哪层不行就换哪层,互不牵扯。
2.2 为什么选 SQLite + 向量检索组合
我在最初调研时也想过要不要上 PostgreSQL。后来权衡再三放弃了,原因很现实:个人级工具的首要约束不是性能,而是维护成本。为了一个记忆功能长期跑一个数据库服务,不管是内存占用还是日常升级备份,都是负担。
SQLite 虽然看起来“轻”,但完全够用。几个关键点:
- 单文件存储,备份就是把文件复制走,非常方便。
- WAL(Write-Ahead Logging)模式开启后,读写并发能力明显改善。
- 事务可靠,不会因为断电丢数据。
- 向量检索用 sqlite-vec 扩展解决,不需要单独部署向量库。
要说明的是,如果你的记忆量真的到了几百万条且并发很高,SQLite 可能就顶不住了。但就我的个人使用量级——几千到几万条记忆,每秒钟几次查询——SQLite 绰绰有余。真到了那一天,再考虑迁移到 PostgreSQL 也不迟,因为 API 层已经抽象好了,底层替换并不困难。
2.3 数据模型和命名空间设计
这是整个工具最核心的部分。我建了一张很简单的memories表,没有过度设计。
CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, namespace TEXT NOT NULL DEFAULT 'default', content TEXT NOT NULL, content_hash TEXT UNIQUE NOT NULL, memory_type TEXT NOT NULL DEFAULT 'note', tags TEXT NOT NULL DEFAULT '[]', source_client TEXT, embedding BLOB, hit_count INTEGER NOT NULL DEFAULT 0, last_access_at TIMESTAMP, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, expire_at TIMESTAMP ); CREATE INDEX idx_memories_namespace ON memories(namespace); CREATE INDEX idx_memories_content_hash ON memories(content_hash);字段含义都很直白。namespace是命名空间,我强烈建议多用它,后面细说。content_hash是内容的 SHA-256 哈希,做唯一约束,避免同一个结论被不同客户端重复写入。embedding是二进制的向量数据,我用 numpy 把 float32 数组序列化后存进去。expire_at是过期时间,到达这个时间后记忆会被清理线程标记为失效,这条设计帮了我大忙。
命名空间解决的是“串味”问题。不同客户端的闲聊记录、项目结论、个人偏好混在一起,检索时很容易互相干扰。我规定:
default:全局通用偏好,如“回答时尽量给代码示例”。project:xx:具体某个项目的共享记忆,所有客户端在讨论该项目时只读写这个命名空间。client:claude:仅供某个客户端私有使用的记忆。
这样一来,即使用同一个服务,也能做到“项目共享但客户端私隐”,不会出现 Claude 聊的东西污染到 Open WebUI 里。
2.4 服务协议:为什么不自己造一个 SDK
在接入层,我选择优先支持 MCP 而不是只提供一个 Python SDK,是因为 SDK 只能覆盖会写代码的开发者,而 MCP 现在已经成为不少客户端原生支持的标准。那意味着用户不需要写一行代码,在客户端的配置界面里加一个 MCP Server 地址就能接入。
MCP 本质上是一个 JSON-RPC 2.0 协议,客户端会去发现一个tools/list接口,拿到服务端声明的工具列表,再通过tools/call来调用具体工具。我把记忆能力拆成了几个工具:
save_memory:写入一条记忆。recall_memory:检索相关记忆。forget_memory:删除指定记忆。list_namespaces:列出所有命名空间。
这种设计的好处是,模型自己会根据用户对话内容判断“现在该不该调用 recall_memory 或 save_memory”,中间不需要人手工介入。我只需要通过工具描述把用途写清楚,剩下的交给模型的 function calling 能力。
3. 核心实现拆解(代码级)
3.1 存储层:SQLite 的读写封装
省略掉所有依赖注入和配置管理的细节,核心的存储层封装就是一个基于sqlite3的连接池包一层,并且强制开启 WAL 模式。
import sqlite3 from contextlib import contextmanager DB_PATH = "~/.memtether/memory.db" def get_connection(): conn = sqlite3.connect(DB_PATH, timeout=30) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL;") conn.execute("PRAGMA busy_timeout=5000;") return conn @contextmanager def db_session(): conn = get_connection() try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close()这里有几个细节值得注意。timeout=30和busy_timeout=5000是配合使用的,前者是 Python 层面的等待上限,后者是 SQLite 内部锁等待的上限。如果同时多个客户端并发写入,这个配置能显著降低“database is locked”的出现频率。另外我所有写操作都不使用长事务,能单条完成就单条完成——这是我踩过并发坑之后总结出来的经验。
3.2 记忆写入链路:去重、向量化、入库
写入链路分为三步:先做内容去重,再调用嵌入模型生成向量,最后把数据写入 SQLite。
import hashlib import numpy as np from sentence_transformers import SentenceTransformer model = SentenceTransformer("BAAI/bge-small-zh-v1.5") def save_memory(namespace: str, content: str, source_client: str, tags: list[str]): content_hash = hashlib.sha256(content.encode("utf-8")).hexdigest() with db_session() as conn: row = conn.execute( "SELECT id FROM memories WHERE content_hash = ?", (content_hash,), ).fetchone() if row: return {"duplicated": True, "id": row["id"]} vector = model.encode(content, normalize_embeddings=True) embedding_blob = np.asarray(vector, dtype=np.float32).tobytes() conn.execute( """ INSERT INTO memories (namespace, content, content_hash, memory_type, tags, source_client, embedding) VALUES (?, ?, ?, ?, ?, ?, ?) """, (namespace, content, content_hash, "note", json.dumps(tags), source_client, embedding_blob), )normalize_embeddings=True这步很重要。向量归一化之后,余弦相似度可以直接用点积计算,省去一步归一化运算,也提高检索精度。去重用哈希而不是全文比对,是因为 SQLite 里直接做文本相似度比较在数据量大了之后非常慢,哈希索引走 B-tree 会快很多。当然,哈希去重无法处理“同一意思不同表述”的近似重复,所以我在写入前还会做一个轻量的“近重复检查”,比如用上一节召回的向量相似度做预判,相似度超过 0.95 就自动忽略。
3.3 记忆召回:BM25 和向量混合检索
如果只用向量检索,遇到专有名词、拼写错误、生僻词的时候准确率会掉得厉害。所以我做了混合检索:一路走向量相似度,另一路走 BM25 关键词匹配(中文先做 jieba 分词),最后用 RRF(Reciprocal Rank Fusion,倒数排名融合)合并结果。
def rrf_fusion(scores_list, k=60): fused = {} for scores in scores_list: for rank, doc_id in enumerate(scores): fused[doc_id] = fused.get(doc_id, 0) + 1.0 / (k + rank + 1) return sorted(fused.items(), key=lambda x: x[1], reverse=True)RRF 的核心思路是只看“排名”不看“绝对分数”。向量相似度 0.8 和 BM25 得分 12.5 之间根本没有可比性,硬把它们相乘相加都是自找麻烦。RRF 把两个排序结果映射到同一套排名权重上,稳定且不敏感。
检索完成之后,我还会做一次基于时间的“活性加权”:
active_score = round(1.0 / (1.0 + (now - last_access_at).seconds / 3600), 4) rrf_score += active_score * 0.1这个细节很关键。同样相关的两条记忆,一条是昨天记的,一条是三个月前记的,默认应该优先给出昨天的。因为对话场景下“时效性”本身就是重要上下文。这个 0.1 的系数我试过很多值,太大会让最近的低相关记录霸榜,太小又起不到时间偏置效果,目前 0.1 是我用下来比较平衡的点。
3.4 MCP 工具定义与参数设计
MCP 工具定义本质上是给模型看的说明书,所以描述写得越清楚,模型调用就越准确。这是recall_memory的工具描述:
{ "name": "recall_memory", "description": "从 MemTether 记忆中检索与给定查询相关的历史内容。当用户的问题涉及之前讨论过的项目、决策或个人偏好时,应该调用此工具。", "inputSchema": { "type": "object", "properties": { "query": { "type": "string", "description": "查询的关键词或问题描述,越具体越好。" }, "namespace": { "type": "string", "description": "命名空间,默认是 default。项目建议使用 project:xxx。" }, "top_k": { "type": "integer", "description": "返回的条数,默认 5,最大 10。" } }, "required": ["query"] } }注意 description 里的措辞,我明确写了“当用户的问题涉及之前讨论过的项目、决策或个人偏好时,应该调用此工具”。这是我从实践中摸索出来的——如果描述太宽泛,模型会没事就调一次,浪费 token;描述太窄,又容易漏调。精准描述工具触发条件,是做好 MCP Server 的关键细节。
4. 一分钟跑起来:部署与接入实操
4.1 安装和初始化
我把项目发布成了 Python 包,安装非常简单:
pip install memtether memtether init --db ~/.memtether/memory.db memtether serve --host 127.0.0.1 --port 8765这里强烈建议host用127.0.0.1而不是0.0.0.0。默认只监听本机回环地址,意味着只有本机进程能访问,网络上的其他设备连不上。如果你只想自用,这是最安全的配置。后面我会单独讲需要远程访问时的加固方案。
启动后服务会输出一条日志:
MemTether listening on http://127.0.0.1:8765 MCP endpoint: /mcp顺手验证一下健康检查接口:
curl http://127.0.0.1:8765/health4.2 接入 Claude Desktop 这类 MCP 客户端
如果你的客户端支持 MCP,比如目前主流的桌面 ChatGPT 类应用和 Claude 类客户端,接入方式是在客户端配置文件里加一个 MCP Server 条目。
以 Claude Desktop 为例,就是把claude_desktop_config.json里的mcpServers字段加上去:
{ "mcpServers": { "memtether": { "command": "uvx", "args": ["--from", "memtether", "memtether-mcp"], "env": { "MEMTETHER_API": "http://127.0.0.1:8765", "MEMTETHER_TOKEN": "你的访问令牌" } } } }这里的command建议先用uvx,它会自动管理 Python 环境和依赖,省去手动建虚拟环境的麻烦。如果你的机器上没有uvx,也可以改成npx或python -m memtether.mcp,但需要确保运行客户端的用户能访问到对应的命令路径。配置完重启客户端,再问一句“你现在能用记忆工具吗”,如果模型回答能,就说明 MCP 已经接上了。
4.3 接入 Open WebUI 及其他自建前端
如果是 Open WebUI 这类可以自定义 Function/Pipeline 的自建前端,不需要依赖 MCP,直接写一个轻量函数调用 HTTP API 就行。我这里贴一个示意图代码,具体版本函数形参会随着版本更新有些变化,但思路是通用的:
def pipe(self, body): query = body["messages"][-1]["content"] memories = recall_memory(query, namespace="default") if memories: memory_section = "\n".join(f"- {m}" for m in memories) body["messages"].insert(0, { "role": "system", "content": f"以下是与用户问题相关的历史记忆,请优先参考:\n{memory_section}" }) return body核心思路是在每个请求前先拉取一次相关记忆,把记忆作为 system prompt 的一部分注入。注意这里插入的位置要在最前面,因为后面的用户消息和工具消息都是在这个基础上组织的,位置错了可能覆盖其他系统指令。
4.4 移动端和自定义项目的接入方式
我平时用手机连回家里电脑上的 Open WebUI,也想让手机端共享同一份记忆。最简单的方式不是装 App,而是走同一个 HTTP API——手机端本质上也是“另一个客户端”。
MemTether 的 REST API 设计得非常直观:
# 写入记忆 curl -X POST http://127.0.0.1:8765/v1/memories \ -H "Authorization: Bearer 你的令牌" \ -H "Content-Type: application/json" \ -d '{"namespace": "project:demo", "content": "这个项目的默认数据库用 SQLite,不引入外部服务"}' # 检索记忆 curl -X GET "http://127.0.0.1:8765/v1/recall?q=项目数据库&namespace=project:demo" \ -H "Authorization: Bearer 你的令牌"如果你想从公网访问,千万别直接把 8765 端口映射到公网。这个时候正确的做法是走内网穿透或者虚拟组网方案,并且仍然要在上面套一层鉴权。没有加密保护和 token 校验的记忆数据直接暴露在公网上,等于把你的个人资料挂到门口,这件事我劝你千万不要做。
4.5 验证记忆共享是否生效
接完客户端之后,我建议做一个完整的闭环测试,别光看“能连上”就以为通了。
第一步,在客户端 A 里说:“记住这件事:以后凡是涉及数据迁移的方案,优先考虑 SQLite 的备份导出工具,不要引入新的数据库。” 模型调用save_memory写入。
第二步,在客户端 B 里新开一个会话,问:“关于数据迁移,我之前有没有什么偏好?” 模型应该能通过recall_memory找到刚才那条记忆。这一步你不需要提前说任何背景,只要 B 能答出“你更倾向用 SQLite 备份导出工具”,就说明共享记忆真正生效了。
第三步,检查数据文件里确实有记录:
sqlite3 ~/.memtether/memory.db "select namespace, content, source_client from memories;"这套测试流程我每次换一台新机器部署都会跑一遍,既验证服务本身,也验证接入层配置有没有写对。
5. 上线两个月遇到的问题与排查实录
5.1 常见故障速查表
我实际用下来,遇到的高频问题就那几类,整理成一张表方便对照:
| 问题现象 | 可能原因 | 解决方式 |
|---|---|---|
| MCP 客户端显示连接失败 | MemTether 服务没启动,或 MCP 路径配置错误 | 先确认服务在跑,再检查客户端配置文件里的 command 是否能被正常执行 |
| 召回结果为空 | 没有写入过记忆,或嵌入模型失败导致向量为空 | 查看服务日志,确认模型加载成功;手动调一次 API 验证 |
| “database is locked” 报错 | 并发写入,且没有开 WAL 模式 | 确认 PRAGMA journal_mode=WAL,并设置 busy_timeout |
| 召回结果混乱,串命名空间 | 检索时没有带 namespace 参数 | 检查客户端的工具调用,确认 namespace 传递正确 |
| 模型频繁调用记忆工具浪费 token | 工具描述写得太宽泛 | 把 description 写精确,限定触发条件 |
5.2 记忆脏数据问题:会写进去也要会忘掉
如果只做“无限写入”,记忆系统迟早会变成垃圾场。这个问题我是在上线第一周就撞上的:我让所有客户端的对话历史都自动写入记忆,结果三天之后,召回结果里全是各种碎片化的中间讨论。比如用户问“今天要改哪个文件?”模型返回了三条记忆,加在一起都没有一句完整的结论,检索效果反而比没有记忆更差。
所以我后来加了两个机制。第一个是expire_at字段,支持为一条记忆设置 TTL,比如某个临时任务的讨论记录,48 小时后自动过期。第二个是“沉淀规则”,只有被标记为“decision”“preference”“project_state”的记忆才会长期保留,普通的闲聊记录默认只保留 7 天。这相当于给记忆系统加了“遗忘”的能力,而且遗忘是有策略的,不是乱删。
5.3 嵌入模型选型与离线部署经验
嵌入模型的选择直接影响召回效果。我一开始图省事直接用了系统里的通用 embedding,结果中文长文本的召回效果很差。后来换成了 BGE 系列,效果明显改善。
| 模型 | 维度 | 中文效果 | 速度 | 模型大小 |
|---|---|---|---|---|
| BAAI/bge-small-zh-v1.5 | 512 | 良好 | 快 | 约 95 MB |
| BAAI/bge-m3 | 1024 | 优秀 | 较慢 | 约 1.2 GB |
| text-embedding-ada-002 | 1536 | 一般 | 中 | API 在线调用 |
我自己在离线机器上的选择是bge-small-zh-v1.5。它模型体积小,CPU 也能跑得动,中文语义理解对于记忆检索这个场景足够用。bge-m3效果确实更好,但 1.2GB 的模型文件如果只是本地个人用,加载和推理都会拖慢响应。
另外,离线环境里第一次加载模型会尝试从 Hugging Face 下载。我建议提前把模型文件下载好,放到~/.cache/huggingface/hub目录下,然后在加载时指定本地路径。这个细节能避免你在没网或者网络受限的机器上卡半天。
5.4 暴露到局域网的安全加固
我在 4.4 提过,不要直接暴露端口到公网。即使只在局域网用,我也做了三层加固。第一层,服务启动时只监听127.0.0.1;如果确实需要局域网访问,我会改监听0.0.0.0,但立刻在配置里开 token 校验。第二层,token 放在配置文件里,用环境变量引用,不硬编码到代码或命令行历史里。第三层,可选开启字段级加密,对content字段做加密存储。
字段级加密这里稍微复杂一点:加密后就没法做 BM25 关键词检索了,只能走向量检索。所以我把加密做成可选项,默认关闭,只有那些你确实不想明文落盘的内容才需要开。对大多数个人使用场景,本地磁盘上的 SQLite 文件做好系统级磁盘加密已经足够,字段级加密不是必须的。
从直接分享的角度说几句
从 v0.1 到现在,MemTether 给我最大的启发是:记忆工具真正难的其实不是存,而是怎么忘。我最初一股脑把所有聊天记录都灌进去,结果召回时全是过期信息,比没有记忆还糟糕。后来加上命名空间、TTL、点击衰减,才慢慢有了“像人一样记事情”的感觉——重要的记住,过期的不死守,不同场景的记忆不互相干扰。
这个项目我目前不会停。下一步想做的是支持多用户权限和跨设备的记忆同步——同一套服务跑在一台机器上,多台设备一起用,每个用户只能读写自己的命名空间。如果你也在做多客户端共享记忆的方案,我的建议是先别想着一上来就做通用平台,把一个项目场景跑通,再逐渐扩展到全局命名空间,这样迭代起来会踏实很多。现在源码和文档都放在开源仓库里,你在使用中遇到的问题也欢迎反馈,我会持续更新这篇踩坑记录。