基于MCP协议与向量数据库构建AI记忆系统：从原理到实践-洪萨配资

1. 项目概述：AI记忆系统的核心价值

最近在折腾AI应用开发，特别是想让AI助手能记住我们之前的对话，实现更连贯、个性化的服务。这让我想起了GitHub上一个挺有意思的项目：ermermermermidk/mcp-ai-memory。简单来说，这是一个基于MCP（Model Context Protocol）协议实现的AI记忆系统。它的核心目标，是解决当前大语言模型（LLM）普遍存在的“健忘症”问题——每次对话都像初次见面，无法利用历史交互信息来提供更精准、更懂你的服务。

想象一下，你告诉AI助手“我喜欢喝美式咖啡”，下次你问“推荐一杯咖啡”时，它如果能直接说“还是来杯美式吗？”，这种体验就完全不一样了。mcp-ai-memory项目正是为了实现这种“记忆”能力而生的。它通过标准化的MCP协议，为各种AI应用（比如Claude Desktop、Cursor等）提供一个外挂的、可持久化的记忆存储与检索服务。开发者无需在每个应用里重复造轮子，只需让应用通过MCP与这个记忆服务器通信，就能轻松为AI赋予记忆能力。

这个项目适合所有正在构建或使用AI助手的开发者、产品经理甚至高级用户。如果你厌倦了每次和AI聊天都要从头交代背景，或者你想打造一个真正“认识”用户的智能体，那么这个项目及其背后的思路，绝对值得你深入研究。接下来，我将从设计思路、核心实现、实操部署到问题排查，完整拆解如何构建和用好这样一个AI记忆系统。

2. 核心架构与设计思路拆解

2.1 为什么选择MCP协议？

在深入代码之前，首先要理解项目为什么基于MCP构建。MCP（Model Context Protocol）是由Anthropic提出的一种开放协议，旨在标准化AI应用与外部工具、数据源之间的通信方式。你可以把它想象成AI世界的“USB标准”——只要设备（AI应用）和配件（工具、数据库）都支持USB（MCP），它们就能即插即用。

对于记忆系统而言，采用MCP协议带来了几个关键优势：

应用无关性：任何支持MCP的AI客户端（如Claude Desktop、自行开发的AI前端）都可以无缝接入这个记忆服务器，无需为每个客户端单独开发适配层。
功能标准化：MCP定义了标准的资源（Resources）和工具（Tools）模型。记忆系统可以暴露“保存记忆”、“搜索相关记忆”等标准工具，客户端以统一的方式调用。
协议层安全与传输：MCP基于SSE（Server-Sent Events）和JSON-RPC，处理了连接管理、请求路由等底层细节，开发者可以专注于记忆业务逻辑本身。

项目的核心设计思路是：将记忆系统抽象为一个独立的、通过MCP提供服务的“记忆微服务”。AI应用（客户端）通过MCP协议向这个服务发送指令（如“保存这段对话”），服务负责将记忆向量化后存入数据库，并在需要时执行相似度搜索，返回最相关的历史记忆片段给客户端，由客户端将其作为上下文喂给LLM。

2.2 记忆系统的核心组件设计

一个实用的AI记忆系统不能只是简单存储文本，它需要解决几个核心问题：记什么、怎么存、怎么找。mcp-ai-memory项目的架构围绕这几个问题展开：

记忆内容（记什么）：通常包括用户陈述的事实（“我在北京工作”）、用户的偏好（“不喜欢吃香菜”）、对话的摘要或关键结论。项目需要定义记忆的数据结构，通常包含：记忆ID、内容文本、关联的用户/会话ID、元数据（如时间戳、重要性标签）、以及最重要的——内容的向量嵌入（Embedding）。
存储与索引（怎么存）：纯文本存储无法实现基于语义的搜索。因此，项目核心是使用文本嵌入模型（如OpenAI的text-embedding-3-small，或开源的BAAI/bge-small-en-v1.5）将记忆内容转换为高维向量，然后存入支持向量相似度搜索的数据库中，例如ChromaDB、Qdrant或PGVector（PostgreSQL扩展）。这种“向量数据库”能快速找到与当前问题语义最相近的历史记忆。
检索逻辑（怎么找）：当用户提出新问题时，系统需要从海量记忆中快速找到相关的。这不仅仅是简单的关键词匹配，而是语义搜索。流程是：1) 将当前问题也转化为向量；2) 在向量数据库中执行相似度计算（通常用余弦相似度）；3) 返回相似度最高的前k条记忆。更高级的策略还会结合时间衰减（越近的记忆权重越高）、重要性加权等。

项目的设计巧妙之处在于，它将这套复杂的流程封装成几个简单的MCP工具。对客户端开发者来说，他只需要调用save_memory(content)和search_memories(query, limit=5)，背后的向量化、存储、检索全部由记忆服务透明完成。

3. 关键技术细节与实现解析

3.1 向量化模型的选择与权衡

记忆系统的效果，很大程度上取决于将文本转换为向量的模型（嵌入模型）的质量。mcp-ai-memory项目通常需要在这里做出选择。

1. 闭源API模型（如OpenAI Embeddings）

优点：效果通常最好，尤其是对英文和复杂语义的理解；无需本地部署，省去计算资源；简单易用。
缺点：产生持续API费用；有网络延迟；数据需要发送到第三方，可能涉及隐私合规考量。
适用场景：快速原型验证、对效果要求高的生产环境（且能接受成本与隐私策略）。

2. 开源本地模型（如Sentence Transformers系列）

优点：数据完全私有，安全性高；一次部署，无后续调用费用；网络延迟极低。
缺点：需要本地GPU或CPU资源进行推理，对服务器有一定要求；模型效果可能略逊于顶级闭源模型；需要自己管理模型加载和推理服务。
适用场景：对数据隐私要求极高的场景；希望控制长期成本的项目；具备一定的运维能力。

实操心得：在项目初期，我建议直接使用OpenAI的嵌入API（如text-embedding-3-small）进行开发测试，它的效果稳定，能让你快速验证记忆功能的核心价值。当产品方向确定，并开始处理真实用户数据时，再评估是否迁移到像BAAI/bge-small-zh-v1.5（中文优化）这类开源模型本地部署。迁移时，注意对比新旧模型在相似度分数上的分布差异，可能需要调整检索的相似度阈值。

3.2 向量数据库的集成与实践

选择了嵌入模型后，下一步就是存储和检索这些向量。mcp-ai-memory项目常选用ChromaDB，因为它设计简单，与Python生态集成好，可以纯内存运行也可持久化。

# 示例：使用ChromaDB创建集合（Collection）并插入记忆向量 import chromadb from chromadb.config import Settings # 初始化客户端，持久化到磁盘 client = chromadb.PersistentClient(path="./memory_db") # 创建或获取一个以用户ID命名的集合 collection = client.get_or_create_collection(name="user_12345") # 假设我们已经有了记忆文本的嵌入向量 `embeddings` memory_embeddings = [...] # 来自嵌入模型的向量列表 memory_texts = ["用户喜欢喝美式咖啡", "用户是软件工程师"] memory_metadatas = [{"timestamp": "2023-10-01"}, {"timestamp": "2023-10-02"}] memory_ids = ["mem_001", "mem_002"] # 将记忆添加到集合中 collection.add( embeddings=memory_embeddings, documents=memory_texts, metadatas=memory_metadatas, ids=memory_ids ) # 检索：根据查询向量查找相似记忆 query_embedding = [...] # 当前用户问题的向量 results = collection.query( query_embeddings=[query_embedding], n_results=3 # 返回最相关的3条记忆 ) print(results['documents']) # 打印找到的记忆文本

关键配置解析：

persistent_client：确保记忆在服务重启后不丢失。
collection：通常按用户或会话隔离，避免不同用户记忆混淆。这是实现多租户记忆的关键。
metadata：强烈建议存储timestamp。在检索时，可以结合相似度和时间进行加权排序，让近期记忆更优先。

注意事项：

维度一致性：嵌入模型输出的向量维度必须与创建集合时定义的维度（或数据库字段）一致。OpenAItext-embedding-3-small是1536维，而bge-small-en是384维。
ID唯一性：ids需要自己维护唯一性，通常可以用uuid或用户ID_时间戳的组合。
元数据过滤：ChromaDB支持基于元数据的过滤查询（如where={"user_id": "123"}），这在实现复杂检索逻辑时非常有用。

3.3 MCP服务器工具的实现

这是项目最核心的部分，即如何将记忆的保存和检索能力，通过MCP协议暴露出去。我们使用Python的mcp库来实现一个标准的MCP服务器。

# 示例：核心MCP工具实现骨架 import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client # 假设我们已经有了一个MemoryManager类，封装了上述向量数据库操作 from memory_manager import MemoryManager class MemoryMCPServer: def __init__(self, user_id: str): self.memory_mgr = MemoryManager(user_id) self.session = None async def save_memory_tool(self, content: str, metadata: dict = None) -> str: """MCP工具：保存一条记忆""" try: memory_id = self.memory_mgr.save(content, metadata) return f"Memory saved successfully with ID: {memory_id}" except Exception as e: return f"Failed to save memory: {str(e)}" async def search_memories_tool(self, query: str, limit: int = 5) -> list: """MCP工具：搜索相关记忆""" try: memories = self.memory_mgr.search(query, limit) # 格式化返回结果，便于AI客户端阅读 formatted = [] for mem in memories: formatted.append(f"- [{mem['timestamp']}] {mem['content']} (score: {mem['score']:.3f})") return "\n".join(formatted) if formatted else "No relevant memories found." except Exception as e: return f"Search failed: {str(e)}" async def start(self): """启动MCP服务器，注册工具""" # 创建与客户端（如Claude Desktop）的stdio通信 server_params = StdioServerParameters( command="python", # 这里实际上是自己调用自己，真实场景可能是独立进程 args=["-m", "mcp_server_main"] # 指向实际的服务入口脚本 ) async with stdio_client(server_params) as (read, write): self.session = ClientSession(read, write) await self.session.initialize() # 向客户端声明本服务器提供的工具 tools = [ { "name": "save_memory", "description": "Save a piece of information to the user's long-term memory.", "inputSchema": { "type": "object", "properties": { "content": {"type": "string", "description": "The content to remember."}, "metadata": {"type": "object", "description": "Optional metadata like tags."} }, "required": ["content"] } }, { "name": "search_memories", "description": "Search the user's past memories for relevant information.", "inputSchema": { "type": "object", "properties": { "query": {"type": "string", "description": "The search query."}, "limit": {"type": "integer", "description": "Max number of results.", "default": 5} }, "required": ["query"] } } ] await self.session.list_tools() # 实际协议中，需要通过特定调用注册工具 # ... 后续进入主循环，监听客户端请求并调用对应的工具函数

实现要点：

工具定义标准化：name、description和inputSchema必须清晰明确，这决定了AI客户端（如Claude）如何理解和使用这个工具。
错误处理：工具函数内部必须有完善的try...except，返回友好的错误信息，避免服务器崩溃。
会话管理：上述示例将user_id在初始化时传入。在实际多用户服务中，user_id应该作为工具调用的一个参数，由客户端传入，服务器根据它来切换不同的记忆集合（Collection）。

4. 完整部署与集成实操指南

4.1 本地开发环境搭建

假设我们从零开始，基于ermermermermidk/mcp-ai-memory的思路构建自己的记忆服务。

步骤1：项目初始化与依赖安装

# 创建项目目录 mkdir ai-memory-server && cd ai-memory-server python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate # 创建核心文件 touch memory_manager.py mcp_server.py config.py requirements.txt # 编辑requirements.txt，添加依赖 echo "chromadb>=0.4.0 openai>=1.0.0 # 如果使用OpenAI嵌入 sentence-transformers # 如果使用开源嵌入模型 mcp>=0.1.0 pydantic>=2.0.0 numpy " > requirements.txt # 安装依赖 pip install -r requirements.txt

步骤2：配置管理（config.py）使用配置文件或环境变量管理敏感信息和变量，是生产级应用的好习惯。

# config.py import os from pydantic_settings import BaseSettings class Settings(BaseSettings): # 嵌入模型配置 EMBEDDING_MODEL: str = os.getenv("EMBEDDING_MODEL", "text-embedding-3-small") OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "") LOCAL_EMBEDDING_MODEL: str = os.getenv("LOCAL_EMBEDDING_MODEL", "BAAI/bge-small-en-v1.5") # 向量数据库配置 CHROMA_PERSIST_DIR: str = os.getenv("CHROMA_PERSIST_DIR", "./chroma_db") # 服务器配置 MCP_SERVER_HOST: str = "0.0.0.0" MCP_SERVER_PORT: int = 8000 settings = Settings()

4.2 记忆管理核心类实现

这是业务逻辑的核心，负责与向量数据库交互。

# memory_manager.py import chromadb from chromadb.config import Settings as ChromaSettings import numpy as np from typing import List, Dict, Any, Optional import uuid from datetime import datetime # 根据配置选择嵌入客户端 from config import settings import openai # 或 from sentence_transformers import SentenceTransformer class EmbeddingClient: """嵌入客户端工厂，根据配置选择不同的模型""" def __init__(self): self.client = None self.model_name = settings.EMBEDDING_MODEL self._init_client() def _init_client(self): if self.model_name.startswith("text-embedding"): # 使用OpenAI API if not settings.OPENAI_API_KEY: raise ValueError("OPENAI_API_KEY is required for OpenAI embeddings.") self.client = openai.OpenAI(api_key=settings.OPENAI_API_KEY) self.type = "openai" else: # 使用本地Sentence Transformer模型 from sentence_transformers import SentenceTransformer self.client = SentenceTransformer(settings.LOCAL_EMBEDDING_MODEL) self.type = "local" def get_embedding(self, text: str) -> List[float]: if self.type == "openai": response = self.client.embeddings.create( model=self.model_name, input=text ) return response.data[0].embedding else: # 本地模型推理 embedding = self.client.encode(text, normalize_embeddings=True) return embedding.tolist() class MemoryManager: def __init__(self, collection_name: str = "default"): self.embedder = EmbeddingClient() self.chroma_client = chromadb.PersistentClient(path=settings.CHROMA_PERSIST_DIR) self.collection = self.chroma_client.get_or_create_collection( name=collection_name, metadata={"hnsw:space": "cosine"} # 使用余弦相似度进行搜索 ) def save(self, content: str, metadata: Optional[Dict] = None) -> str: """保存一条记忆，返回记忆ID""" if not metadata: metadata = {} metadata.setdefault("timestamp", datetime.utcnow().isoformat()) memory_id = str(uuid.uuid4()) embedding = self.embedder.get_embedding(content) self.collection.add( embeddings=[embedding], documents=[content], metadatas=[metadata], ids=[memory_id] ) return memory_id def search(self, query: str, limit: int = 5, filter_dict: Optional[Dict] = None) -> List[Dict[str, Any]]: """搜索相关记忆""" query_embedding = self.embedder.get_embedding(query) results = self.collection.query( query_embeddings=[query_embedding], n_results=limit, where=filter_dict, # 可选的元数据过滤，如 {"user_id": "123"} include=["documents", "metadatas", "distances"] ) memories = [] for i in range(len(results['ids'][0])): memories.append({ "id": results['ids'][0][i], "content": results['documents'][0][i], "metadata": results['metadatas'][0][i], "score": 1 - results['distances'][0][i] # Chroma返回的是距离，转换为相似度分数 }) return memories

4.3 MCP服务器主程序

现在，我们将记忆管理器与MCP协议结合起来。

# mcp_server.py import asyncio import json from typing import Any from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from memory_manager import MemoryManager class AIMemoryServer: def __init__(self): # 注意：这里为了简化，使用全局记忆管理器。实际应按会话或用户隔离。 self.memory_mgr = MemoryManager(collection_name="global_memories") self.session: ClientSession | None = None async def handle_save_memory(self, arguments: dict) -> str: content = arguments.get("content") if not content: return "Error: 'content' field is required." metadata = arguments.get("metadata", {}) try: memory_id = self.memory_mgr.save(content, metadata) return json.dumps({"status": "success", "memory_id": memory_id}) except Exception as e: return json.dumps({"status": "error", "message": str(e)}) async def handle_search_memories(self, arguments: dict) -> str: query = arguments.get("query") if not query: return "Error: 'query' field is required." limit = arguments.get("limit", 5) try: memories = self.memory_mgr.search(query, limit) # 格式化输出，便于AI阅读 if not memories: return "No relevant memories found." result_lines = ["Here are relevant past memories:"] for mem in memories: result_lines.append(f"- {mem['content']} (Relevance: {mem['score']:.2f})") return "\n".join(result_lines) except Exception as e: return f"Search error: {str(e)}" async def run(self): """启动MCP服务器（通过stdio与客户端通信）""" # 注意：这是一个简化的示例。实际MCP服务器启动方式可能更复杂。 # 这里模拟通过标准输入输出与父进程（如Claude Desktop）通信。 print("Initializing AI Memory MCP Server...", flush=True) # 在实际实现中，你需要遵循MCP协议，通过stdio读取JSON-RPC请求并响应。 # 以下是一个极其简化的模拟循环，用于说明概念。 try: while True: # 从标准输入读取一行（模拟请求） line = await asyncio.get_event_loop().run_in_executor(None, input, "") if not line: continue try: request = json.loads(line) method = request.get("method") params = request.get("params", {}) if method == "save_memory": result = await self.handle_save_memory(params) elif method == "search_memories": result = await self.handle_search_memories(params) else: result = json.dumps({"error": f"Unknown method: {method}"}) # 将结果写回标准输出（模拟响应） print(json.dumps({"result": result}), flush=True) except json.JSONDecodeError: print(json.dumps({"error": "Invalid JSON"}), flush=True) except (EOFError, KeyboardInterrupt): print("Server shutting down.", flush=True) if __name__ == "__main__": server = AIMemoryServer() asyncio.run(server.run())

4.4 与Claude Desktop集成

这是让记忆服务生效的关键一步。Claude Desktop支持通过本地配置文件添加MCP服务器。

首先，确保你的记忆服务器脚本（mcp_server.py）可以运行。
在Claude Desktop的配置目录下创建或编辑MCP配置文件。
- macOS:~/Library/Application Support/Claude/claude_desktop_config.json
- Windows:%APPDATA%\Claude\claude_desktop_config.json
编辑配置文件，添加你的记忆服务器：

{ "mcpServers": { "ai-memory": { "command": "python", "args": ["/绝对路径/到/你的/项目/mcp_server.py"], "env": { "OPENAI_API_KEY": "你的密钥", "EMBEDDING_MODEL": "text-embedding-3-small" } } } }

重启Claude Desktop。重启后，Claude应该能连接到你的记忆服务器。你可以在对话中测试，例如说：“请记住，我最喜欢的编程语言是Python。”然后稍后问：“我之前说过我喜欢什么编程语言？”。Claude应该会调用search_memories工具来找到答案。

重要提示：上述mcp_server.py是一个高度简化的概念验证版本。实际生产级的MCP服务器实现需要严格遵循MCP协议规范，处理初始化握手、工具列表声明、并发请求等。建议直接参考ermermermermidk/mcp-ai-memory项目的源码，或使用更成熟的MCP服务器框架。

5. 高级功能与优化策略

一个基础的记忆系统搭建完成后，可以考虑以下优化来提升其实用性和智能度。

5.1 记忆的自动摘要与压缩

LLM的上下文长度有限，不能无限制地塞入所有历史记忆。因此，需要对记忆进行智能管理。

策略：定期（例如每10轮对话后）或当记忆条数超过阈值时，触发一个摘要任务。使用LLM（如GPT-4或Claude Haiku）对近期的一组相关记忆进行总结，生成一条浓缩的“摘要记忆”，并替换或归档原有的多条细节记忆。
实现：新增一个MCP工具summarize_memories，它调用记忆管理器的搜索功能找到近期记忆，然后调用LLM API生成摘要，最后保存摘要并删除原始记忆（或标记为已归档）。

5.2 基于时间与重要性的检索加权

不是所有记忆都同等重要。昨天的午餐内容可能不如“我对花生严重过敏”这条信息重要。

时间衰减：在检索评分中引入时间因子。公式可以简化为：最终分数 = 语义相似度分数 * exp(-衰减系数 * 时间差)。这样，越旧的记忆，即使语义相关，排名也会靠后。
重要性标签：允许在保存记忆时添加重要性元数据（如importance: “high”）。在检索时，对高重要性的记忆进行加分。重要性可以由用户显式标注，也可以由AI在保存时根据内容自动判断（例如，涉及健康、安全、强烈偏好的内容判为高重要性）。

5.3 多租户与记忆隔离

在服务多个用户或不同AI代理时，必须严格隔离记忆。

实现：将user_id和agent_id（或session_id）作为记忆的必需元数据。在MemoryManager初始化或每个工具调用时，必须传入这些ID。向量数据库的集合（Collection）命名可以直接使用f”user_{user_id}_agent_{agent_id}”，或者在同一集合中使用元数据字段进行过滤查询（where={“user_id”: “xxx”}）。
安全：确保从MCP客户端传来的用户身份是经过验证的，防止用户A查询到用户B的记忆。

6. 常见问题与故障排查实录

在实际部署和使用过程中，你几乎一定会遇到下面这些问题。

6.1 连接与通信问题

问题1：Claude Desktop无法连接MCP服务器，提示“Connection refused”或超时。

排查：
1. 检查配置文件路径和格式是否正确。JSON格式必须严格正确。
2. 检查command和args指向的Python解释器和脚本路径是否有效。建议使用绝对路径。
3. 在终端手动运行配置中的命令（如python /path/to/mcp_server.py），看脚本是否能正常启动，有无报错（如缺少依赖）。
4. 确保你的MCP服务器脚本正确地通过stdio进行通信。一个简单的测试方法是：运行你的服务器脚本，然后手动在终端输入一行模拟的JSON-RPC请求，看是否有正确的JSON响应输出。

问题2：Claude能连接，但看不到save_memory等工具。

排查：这通常是MCP服务器没有在初始化时正确声明工具列表。根据MCP协议，服务器在初始化阶段必须响应tools/list调用，返回工具清单。请仔细检查你的服务器代码，是否实现了list_tools方法或等价的功能。

6.2 记忆检索效果不佳

问题3：AI总是找不到明明已经保存过的记忆。

可能原因1：相似度阈值过高。向量搜索返回的结果默认会按相似度排序，但如果没有设置最低分数阈值，一些低分结果（不相关）也会被返回，而高分结果可能因为分数不够高被忽略。尝试在search函数中设置一个阈值（如score > 0.7），并调试观察不同查询的分数分布。
可能原因2：嵌入模型不匹配或质量差。如果你使用的是小型开源模型，对复杂或专业语句的语义捕捉能力可能有限。尝试用更高质量的模型（如text-embedding-3-large或BAAI/bge-large-en），或者确保查询语句和记忆内容在表述上更接近。
可能原因3：记忆内容过于冗长或噪声大。保存记忆前，可以考虑用LLM对内容进行简要提炼或提取关键词，只保存核心事实。例如，将一整段对话总结为“用户计划下周去上海出差”。

问题4：检索到了记忆，但AI在回答中不会合理利用。

排查：这可能是提示工程（Prompt Engineering）的问题。当AI客户端收到检索到的记忆列表后，需要将其以合适的格式插入到给LLM的提示词中。通常的格式是：
```
以下是用户的相关历史信息，供你参考： - [记忆1内容] - [记忆2内容] 当前问题：[用户的新问题]
```
确保你的客户端在调用搜索工具后，将返回的记忆文本清晰地组织到上下文中。有时需要明确指示AI：“请根据以上历史信息来回答。”

6.3 性能与扩展性问题

问题5：记忆数量很大后，检索速度变慢。

优化：
1. 索引优化：确保向量数据库使用了合适的索引（如Chroma默认的HNSW）。对于生产环境大量数据，考虑使用专业的向量数据库如Qdrant、Weaviate或PGVector（搭配IVFFlat或HNSW索引）。
2. 分页与过滤：在搜索时，尽量使用元数据过滤（如时间范围、标签）来缩小搜索集合，而不是在全库中搜索。
3. 记忆归档：将很久以前且不重要的记忆转移到冷存储（如普通对象存储），只在向量数据库中保留近期和重要的热点记忆。

问题6：如何备份和迁移记忆数据？