ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

本地AI记忆系统实战:架构、MCP协议与本地模型接入

本地AI记忆系统实战:架构、MCP协议与本地模型接入 1. 为什么本地 AI 记忆系统值得认真做一次先把话说直白现在大部分人在用的 AI 助手本质上是个金鱼脑。你昨天跟它聊了两个小时的项目架构今天开个新会话它对你一无所知。你反复解释自己的技术栈、代码规范、命名习惯、甚至我不喜欢用某个库它每次都像第一次见你。这不是模型不够聪明而是记忆这件事根本没被当成一等公民来设计。上下文窗口再大也是临时的、会话级的、关掉就清零的。RAG 能补一部分但 RAG 检索的是文档不是你这个人。所以当我们决定做一套本地 AI 记忆系统的时候核心命题其实就一句话让 AI 记住关于你的一切并且这些记忆完全跑在你自己的机器上。这件事为什么现在做、为什么值得做我拆成三个层面讲。1.1 云端记忆的三个硬伤第一是隐私边界。你的聊天记录里有什么可能有公司内部架构、未公开的产品思路、个人日程、甚至一些随口说的情绪化判断。这些东西一旦上传到别人的服务器你就失去了控制权。哪怕对方承诺加密、承诺不训练你也没法验证。第二是记忆的归属权。云端记忆是平台资产不是你的资产。换个工具记忆带不走平台改政策你的数据可能被清平台倒闭记忆直接蒸发。这跟当年网盘大战一个道理——你以为存的是你的文件其实存的是别人的商业模式。第三是可定制性。云端记忆系统是黑盒你没法告诉它这条记忆权重高一点那条三天后过期这类信息永远不要记。而本地系统你可以直接改数据库、改检索策略、改衰减曲线。1.2 本地化不等于断网这里要澄清一个常见误解。很多人一听本地就以为是完全离线、不联网。不是的。本地 AI 记忆系统的本地指的是记忆的存储、检索、管理全部在本地完成。至于推理用的模型你可以用本地部署的开源模型比如通过 LM Studio、Ollama 加载也可以调用云端 API——但即使调用云端 API你发给它的也只是检索出来的相关记忆片段而不是全量历史。这个区别很关键。它意味着你可以在享受大模型能力的同时把最敏感的那层数据留在自己手里。1.3 这套系统到底解决什么问题具体到使用场景它能解决这些事你换了个新会话AI 依然知道你在做 Rust 项目、知道你偏好函数式风格、知道你上周卡在某个并发问题上。你问上次那个方案后来怎么定的它能从记忆里翻出来而不是让你重新讲一遍。你明确说这件事别记它就真的不记而且你能在数据库里验证它没记。你想让记忆跨工具共享——今天用这个客户端明天用那个 IDE 插件记忆是同一份。说白了这是把 AI 从一次性工具变成长期协作者的基础设施。而找技术合伙人这件事恰恰说明这不是一个人能轻松搞定的活——它涉及存储、检索、模型接入、协议设计、客户端适配是个典型的系统工程。2. 记忆系统的核心架构不是加个数据库那么简单很多人第一反应是不就是把聊天记录存进 SQLite然后检索的时候塞回 prompt 吗如果你真这么干跑一周就会崩。原因很简单记忆不是日志日志是append-only的流水记忆是需要被组织、被衰减、被关联、被召回的结构化知识。我按数据流把架构拆成四层每层都有坑。2.1 写入层什么该记什么不该记写入层要解决的核心问题是记忆的准入。你不能把所有对话都无脑存进去否则检索质量会被噪声淹没。我们的做法是分三档档位触发条件处理方式显式记忆用户说记住这个以后都这样直接写入高权重隐式记忆模型判断信息有长期价值写入候选池需二次确认或达阈值才转正临时上下文普通对话只存短期缓存按时间衰减这里有个经验隐式记忆的判定不要完全交给模型。我们试过让模型自己决定这条要不要记结果它过于积极把今天天气不错也记了。后来改成模型提议 规则过滤 用户可撤销三段式噪声率降了一个数量级。规则过滤包括长度阈值太短的丢弃、去重语义相似度超过阈值的合并、敏感词过滤这个必须本地做不能送出去。2.2 存储层向量库 关系库的双写纯向量库的问题是它能做语义检索但做不了精确的结构化查询。比如你想问我上周提到的所有关于数据库的记忆向量检索给不了你准确的时间范围过滤。所以我们的方案是双写向量库本地用 Chroma、Qdrant 或 LanceDB存 embedding负责语义召回。关系库SQLite 足够存元数据时间戳、来源会话、权重、标签、访问次数、衰减状态。两边用同一个 UUID 关联。检索时先在关系库做条件过滤时间、标签、权重拿到候选 ID 集合再去向量库做相似度排序。这个先过滤后检索的顺序很重要反过来会慢很多。提示SQLite 在单机场景下完全够用别一上来就上 PostgreSQL。我们早期为了看起来专业上了 PG结果部署复杂度陡增后来退回 SQLite性能没差运维省心一半。2.3 检索层召回策略决定体验上限检索层是整个系统最考验功力的地方。简单说它要回答当前这个问题应该唤起哪些记忆我们的召回是多路融合语义召回query embedding 和记忆 embedding 算余弦相似度取 top-K。时间召回最近 N 天的记忆无论语义是否相关都给一定权重因为近期上下文通常更重要。关联召回如果当前 query 命中了某个实体比如项目名、人名把和这个实体关联的所有记忆拉出来。频次召回被反复访问的记忆说明它重要加权。四路结果做加权融合再送去重和截断。权重怎么定我们没有拍脑袋而是做了个小规模的 A/B让几个真实用户标注这条记忆该不该被召回然后调参。最后发现语义权重 0.5、时间权重 0.2、关联权重 0.2、频次权重 0.1是个不错的起点。2.4 注入层怎么塞进 prompt 才不浪费 token检索出来的记忆最终要拼进 prompt。这里有个反直觉的点不是塞得越多越好。我们试过把 top-20 记忆全塞进去结果模型开始分心回答质量反而下降。后来改成只注入 top-5 到 top-8且每条记忆压缩成一句话摘要。记忆按重要性排序重要的放前面模型对 prompt 前部更敏感。加一个明确的边界标记比如[相关记忆开始]...[相关记忆结束]让模型知道这是背景而非指令。这个边界标记很关键。没有它模型有时会把记忆内容当成用户当前的要求产生误答。3. MCP 协议让记忆系统变成可插拔的基础设施聊到技术合伙人就绕不开一个现实问题你做的记忆系统怎么让别人的工具用上如果每个客户端都要单独适配那这个项目永远做不大。这时候 MCPModel Context Protocol就成了关键。3.1 MCP 到底解决了什么MCP 本质是一个标准化的工具/资源暴露协议。它让一个能力提供方比如我们的记忆系统能以统一的方式被各种能力消费方IDE 插件、聊天客户端、Agent 框架调用。打个比方以前每个电器都要配自己的插座MCP 就是那个统一的标准插座。你的记忆系统做成一个 MCP Server任何支持 MCP 的客户端都能接进来读写记忆。这就把记忆系统从一个 App变成了一层基础设施。价值完全不同。3.2 把记忆系统封装成 MCP Server 的实操核心是暴露几个工具tool{ tools: [ { name: memory_write, description: 写入一条记忆, inputSchema: { type: object, properties: { content: {type: string}, tags: {type: array, items: {type: string}}, weight: {type: number, default: 1.0} }, required: [content] } }, { name: memory_search, description: 检索相关记忆, inputSchema: { type: object, properties: { query: {type: string}, top_k: {type: integer, default: 5} }, required: [query] } }, { name: memory_forget, description: 删除指定记忆, inputSchema: { type: object, properties: { memory_id: {type: string} }, required: [memory_id] } } ] }三个工具写、查、删。看起来简单但每个都有讲究。memory_write的weight参数是给显式记忆用的用户说这个很重要时传高权重。memory_search的top_k默认给 5是因为前面说的注入层经验。memory_forget必须存在——一个不能删除的记忆系统是危险的用户得有被遗忘权。3.3 MCP 接入时的几个坑第一个坑是授权。MCP 客户端接入时通常需要确认授权范围。我们早期没做细粒度权限结果某个客户端能读全部记忆这不行。后来加了 scope读、写、删分开授权。第二个坑是并发写。多个客户端同时写记忆时SQLite 的写锁会冲突。解决办法是加一个写入队列串行化处理或者用 WAL 模式提升并发。第三个坑是协议版本兼容。MCP 还在演进不同客户端支持的版本不一样。我们的做法是在 Server 端做版本协商老版本走兼容路径。注意如果你打算把记忆系统做成 MCP Server一定要先把删除和权限做扎实。这两块出问题用户信任直接归零。4. 本地模型接入LM Studio、Ollama 与推理性能的取舍记忆系统本身不产生智能它依赖模型来做 embedding 和生成。所以模型接入是绕不过去的一环。4.1 embedding 模型的选择embedding 是记忆系统的索引引擎它的质量直接决定检索准不准。本地可选的方案模型维度中文效果资源占用适用场景bge-small-zh512好低轻量部署、边缘设备bge-base-zh768很好中主流选择bge-large-zh1024最好高对精度要求极高m3e-base768好中备选我们的默认是bge-base-zh。理由small 在长文本上召回率明显下降large 的收益相对 base 提升有限但资源翻倍。base 是性价比拐点。这里有个实操细节embedding 模型和生成模型最好分开部署。因为 embedding 是高频调用每次写入和检索都要用生成是低频调用。混在一起会互相抢资源。4.2 通过 LM Studio / Ollama 加载本地模型这两个工具都能把本地模型暴露成 OpenAI 兼容的 API接入成本很低。Ollama 的方式ollama pull bge-m3 ollama serve然后在记忆系统里配置import openai client openai.OpenAI( base_urlhttp://localhost:11434/v1, api_keyollama # 本地不需要真实 key ) def get_embedding(text): resp client.embeddings.create( modelbge-m3, inputtext ) return resp.data[0].embeddingLM Studio 类似只是端口和模型名不同。它的优势是有 GUI方便切换模型和调参。4.3 在 Jetson Orin 这类边缘设备上的部署如果目标是完全本地、低功耗Jetson Orin 是个常见选择。但要注意内存是瓶颈。Orin 的显存和内存共享跑 7B 模型 embedding 模型会比较紧张。建议生成模型用 4-bit 量化embedding 用 small 版本。推理速度。Orin 上跑 7B 模型生成速度大概在 10-20 token/s够用但不快。记忆检索本身很快毫秒级瓶颈在生成。散热。长时间推理会降频要做好散热设计。我们实测下来Orin 上跑small embedding 4bit 7B 生成是可行的但别指望它做高并发。5. 技术合伙人视角这个项目真正难在哪既然标题是找技术合伙人我就从一个潜在合伙人会关心什么的角度把项目的真实难点摊开讲。5.1 难点一记忆的质量而非数量新手容易陷入我要存下所有东西的执念。但记忆系统的价值不在存了多少而在召回时准不准。我们做过一个测试存 1000 条记忆检索 top-5人工评估相关性。早期版本准确率只有 40% 左右大量噪声。经过写入过滤、权重调整、召回融合三轮优化提升到 75% 以上。这个提升不是靠堆数据是靠减法。所以合伙人要理解这是个做减法的工程不是做加法。5.2 难点二跨工具的协议一致性记忆系统要真正有用必须跨工具。但不同工具对记忆的理解不一样IDE 插件关心代码上下文聊天客户端关心对话历史Agent 框架关心任务状态。MCP 提供了协议层的一致性但语义层的一致性还得自己定义。我们设计了一套记忆的 schema规定每条记忆必须有内容、类型、来源、时间、权重、标签。所有工具写入时都遵循这个 schema检索时才能统一处理。5.3 难点三隐私与可验证性本地是卖点但用户凭什么信你我们的做法是开源 可审计。数据库是明文的 SQLite用户随时能用任何工具打开看。所有网络请求都有日志用户能确认哪些数据出去了、哪些没出去。这一点对合伙人很重要如果你打算做闭源那本地的信任基础就弱了一半。5.4 难点四长期维护的衰减策略记忆会越来越多如果不做衰减检索会越来越慢、越来越不准。我们设计了一套衰减机制每条记忆有个decay_score随时间下降。被访问时decay_score回升类似 LRU。低于阈值的记忆进入冷存储不参与常规检索但可被显式查询。极低阈值的记忆自动归档或删除可配置。这套机制让系统能长期运行而不膨胀。但阈值怎么定需要根据实际使用调没有万能值。6. 从零搭一个最小可用版本可复现的步骤前面讲了架构和难点这一节给一套能跑起来的最小实现。目标本地存储 语义检索 MCP 暴露不追求完美但求跑通。6.1 环境与依赖python -m venv venv source venv/bin/activate pip install chromadb sqlite-utils openai mcp sentence-transformersChroma 做向量库SQLite 做元数据sentence-transformers 做本地 embeddingmcp 做协议暴露。6.2 数据模型import sqlite3 def init_db(pathmemory.db): conn sqlite3.connect(path) conn.execute( CREATE TABLE IF NOT EXISTS memories ( id TEXT PRIMARY KEY, content TEXT NOT NULL, tags TEXT, weight REAL DEFAULT 1.0, created_at REAL, last_access REAL, decay_score REAL DEFAULT 1.0, archived INTEGER DEFAULT 0 ) ) conn.commit() return conn字段设计对应前面讲的权重、时间、衰减、归档状态。6.3 写入与检索import uuid, time from sentence_transformers import SentenceTransformer import chromadb model SentenceTransformer(BAAI/bge-base-zh-v1.5) client chromadb.PersistentClient(path./chroma) collection client.get_or_create_collection(memories) def write_memory(conn, content, tagsNone, weight1.0): mid str(uuid.uuid4()) now time.time() conn.execute( INSERT INTO memories (id, content, tags, weight, created_at, last_access, decay_score) VALUES (?, ?, ?, ?, ?, ?, ?), (mid, content, ,.join(tags or []), weight, now, now, 1.0) ) conn.commit() emb model.encode(content).tolist() collection.add(ids[mid], embeddings[emb], documents[content]) return mid def search_memory(conn, query, top_k5): emb model.encode(query).tolist() results collection.query(query_embeddings[emb], n_resultstop_k * 2) ids results[ids][0] # 关系库过滤排除归档、按衰减排序 placeholders ,.join(? * len(ids)) rows conn.execute( fSELECT id, content, decay_score FROM memories fWHERE id IN ({placeholders}) AND archived 0 fORDER BY decay_score DESC LIMIT ?, (*ids, top_k) ).fetchall() return rows这段代码就是最小闭环。注意检索时先拿 2 倍候选再用关系库过滤这是前面说的先过滤后检索的变体。6.4 衰减任务def decay_task(conn, half_life_days30): now time.time() half_life_sec half_life_days * 86400 rows conn.execute(SELECT id, last_access, decay_score FROM memories).fetchall() for mid, last_access, score in rows: elapsed now - last_access new_score score * (0.5 ** (elapsed / half_life_sec)) archived 1 if new_score 0.05 else 0 conn.execute( UPDATE memories SET decay_score ?, archived ? WHERE id ?, (new_score, archived, mid) ) conn.commit()半衰期 30 天是个起点实际用下来可以调到 14 或 60看你的使用频率。6.5 封装成 MCP Serverfrom mcp.server import Server from mcp.types import Tool, TextContent app Server(local-memory) app.list_tools() async def list_tools(): return [ Tool(namememory_write, description写入记忆, inputSchema{type: object, properties: {content: {type: string}}, required: [content]}), Tool(namememory_search, description检索记忆, inputSchema{type: object, properties: {query: {type: string}}, required: [query]}), ] app.call_tool() async def call_tool(name, arguments): if name memory_write: mid write_memory(conn, arguments[content]) return [TextContent(typetext, textf已写入 {mid})] if name memory_search: rows search_memory(conn, arguments[query]) return [TextContent(typetext, text\n.join(r[1] for r in rows))]跑起来之后任何支持 MCP 的客户端都能接进来。这就是可插拔的起点。7. 几个我踩过的坑和对应的解法最后这部分是纯经验文档里不会写但实际做的时候一定会遇到。坑一embedding 模型换了历史记忆全废。不同模型的向量空间不兼容换模型意味着所有历史记忆要重新 embedding。我们的解法是把原始文本永远保留embedding 只是索引可重建。重建任务做成后台批处理不阻塞使用。坑二SQLite 的 WAL 模式没开并发写直接锁死。多客户端接入后写冲突频繁。开启 WAL 后好很多PRAGMA journal_modeWAL; PRAGMA busy_timeout5000;坑三记忆注入导致模型幻觉引用。模型有时会把记忆里的内容当成事实即使记忆本身是错的。解法是在注入时加一句以下为历史记忆可能过时请以当前对话为准。坑四中文分词的坑。如果做关键词召回中文分词质量直接影响效果。我们最后放弃了关键词召回纯靠语义 元数据过滤反而更稳。坑五别过早优化。我们早期花了两周做分布式存储结果单机 SQLite 完全够用。先跑通再优化这是本地项目最容易违反也最该遵守的原则。这套系统到现在还在迭代MCP 的生态也在快速变化。但核心逻辑是稳的本地存储、语义检索、协议暴露、可审计可删除。如果你对这个方向有兴趣或者正在做类似的事欢迎交流——尤其是如果你对 MCP 协议、本地推理优化、或者记忆衰减策略有想法这些正是最需要合伙人的地方。
返回列表