☰
MemTether:用 MCP 为多 AI 客户端构建统一长期记忆服务
2026/10/6 6:41:52 网站建设 项目流程

说个我最近的折腾成果:我自己写了一个开源小工具,叫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/health

4.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.5512良好快约 95 MB
BAAI/bge-m31024优秀较慢约 1.2 GB
text-embedding-ada-0021536一般中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、点击衰减,才慢慢有了“像人一样记事情”的感觉——重要的记住,过期的不死守,不同场景的记忆不互相干扰。

这个项目我目前不会停。下一步想做的是支持多用户权限和跨设备的记忆同步——同一套服务跑在一台机器上,多台设备一起用,每个用户只能读写自己的命名空间。如果你也在做多客户端共享记忆的方案,我的建议是先别想着一上来就做通用平台,把一个项目场景跑通,再逐渐扩展到全局命名空间,这样迭代起来会踏实很多。现在源码和文档都放在开源仓库里,你在使用中遇到的问题也欢迎反馈,我会持续更新这篇踩坑记录。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询