Agent 记忆技术深度解析：从学术前沿到工程实践

Agent 记忆技术深度解析：从学术前沿到工程实践本文深入剖析 AI Agent 记忆技术的学术研究现状、技术原理和工程实践方案，包含 mem0、Letta、EverMemOS、memory-bear 四大主流开源方案的从零部署、代码集成和效果对比，以及 CodeBuddy 等成熟 AI 产品的记忆机制分析。 1. 引言 1.1 为什么 Agent 需要记忆？传统的大语言模型（LLM）是"无状

发表于 2026/08/31

作者 贾永琪

104 分钟阅读

文章信息
原文链接：https://jiayq.blog.csdn.net/article/details/156650263
发布时间：2026-08-31 14:23:52
标签：#ai, #智能体, #agent, ##mem0, ##MCP

Agent 记忆技术深度解析：从学术前沿到工程实践

本文深入剖析 AI Agent 记忆技术的学术研究现状、技术原理和工程实践方案，包含 mem0、Letta、EverMemOS、memory-bear 四大主流开源方案的从零部署、代码集成和效果对比，以及 CodeBuddy 等成熟 AI 产品的记忆机制分析。

1. 引言

1.1 为什么 Agent 需要记忆？

传统的大语言模型（LLM）是”无状态”的——每次对话都是一次全新的开始。这就像一个患有顺行性遗忘症的人，无论你们聊了多久，下次见面他都不记得你是谁。

在实际应用中，这种”健忘”带来了诸多问题：

上下文丢失 ：长对话中，早期的重要信息会被遗忘
个性化缺失 ：无法记住用户的偏好、习惯和历史
知识断层 ：无法积累和复用之前学到的知识
效率低下 ：重复的信息需要反复提供

Agent 记忆技术的出现，正是为了解决这些问题。通过为 Agent 配备”记忆系统”，我们可以让 AI 真正”记住”用户，实现个性化、连续性的智能交互。

1.2 记忆技术的分类

从技术实现角度，Agent 记忆可以分为以下几类：

记忆类型	特点	典型实现	适用场景
短期记忆	会话内有效，基于上下文窗口	Prompt 拼接	单轮对话
长期记忆	跨会话持久化，向量化存储	mem0、Letta	个性化助手
工作记忆	任务执行期间的临时存储	Scratchpad	复杂推理
情景记忆	记录具体事件和经历	时间线存储	对话历史
语义记忆	抽象知识和概念	知识图谱	知识问答

2. 学术研究现状

2.1 顶会论文综述

近年来，Agent 记忆技术在 NeurIPS、ACL、EMNLP、ICLR 等顶级会议上获得了广泛关注。以下是一些代表性的研究工作：

2.1.1 MemoryBank: Enhancing Large Language Models with Long-Term Memory

会议：AAAI 2024

论文链接 ：https://arxiv.org/abs/2305.10250

会议主页 ：https://aaai.org/aaai-conference/

核心思想 ：提出了 MemoryBank 框架，通过记忆存储、记忆检索和记忆更新三个模块，实现 LLM 的长期记忆能力。

技术亮点 ：

基于 Ebbinghaus 遗忘曲线的记忆衰减机制
多层次记忆组织（事件级、摘要级、概念级）
动态记忆更新和合并策略

2.1.2 Reflexion: Language Agents with Verbal Reinforcement Learning

会议：NeurIPS 2023

论文链接 ：https://arxiv.org/abs/2303.11366

会议主页 ：https://neurips.cc/virtual/2023/poster/70114

GitHub ：https://github.com/noahshinn/reflexion

核心思想 ：通过让 Agent 进行”自我反思”，将失败经验转化为语言形式的记忆，用于指导后续决策。

技术亮点 ：

语言化的反思记忆
无需梯度更新的强化学习
在多种任务上显著提升性能

2.1.3 Generative Agents: Interactive Simulacra of Human Behavior

会议：UIST 2023（ACM）

论文链接 ：https://arxiv.org/abs/2304.03442

会议主页 ：https://uist.acm.org/2023/

GitHub ：https://github.com/joonspk-research/generative_agents

核心思想 ：构建了一个包含 25 个 AI 角色的虚拟小镇，每个角色都有独立的记忆系统，能够进行社交互动。

技术亮点 ：

记忆流（Memory Stream）架构
反思（Reflection）机制生成高层次洞察
规划（Planning）机制基于记忆生成行动计划

2.1.4 RET-LLM: Towards a General Read-Write Memory for Large Language Models

论文链接 ：https://arxiv.org/abs/2305.14322

核心思想 ：提出了一种通用的读写记忆机制，允许 LLM 在推理过程中动态读取和写入记忆。

技术亮点 ：

显式的记忆读写操作
基于 Transformer 的记忆编码器
支持多种下游任务

2.2 技术路径对比

技术路径	代表工作	优势	劣势
向量检索	mem0、LangChain Memory	实现简单，检索高效	语义理解有限
图结构	MemGPT、GraphRAG	关系建模能力强	构建和维护复杂
混合架构	Letta、EverMemOS	兼顾多种记忆类型	系统复杂度高
神经记忆	Memory Networks	端到端学习	需要大量训练数据

3. 开源方案深度解析

3.1 mem0：轻量级记忆层

3.1.1 项目概述

mem0 是一个为 AI 应用提供智能记忆层的开源项目，专注于为 LLM 和 AI Agent 添加持久化记忆能力。

GitHub ：https://github.com/mem0ai/mem0

核心特性 ：

多层次记忆：用户级、会话级、Agent 级
自适应个性化：基于交互持续学习
开发者友好：简单的 API 设计
跨平台一致性：多应用间记忆共享

3.1.2 架构设计

mem0 的架构相对简洁，主要包含以下组件：

存储层

核心处理层

API 层

Memory Add

Memory Search

Memory Update

LLM 层
提取记忆

Embedding 层
向量化

Vector Store
Qdrant

工作流程 ：

用户输入对话内容
LLM 从对话中提取关键记忆点
Embedding 模型将记忆向量化
向量存储到 Qdrant 数据库
检索时通过语义相似度匹配

3.1.3 源码分析

mem0 的核心代码结构清晰，主要文件包括：

mem0/
├── memory/
│   ├── main.py          # Memory 类主入口
│   ├── base.py          # 基类定义
│   └── graph_memory.py  # 图记忆（可选）
├── llms/
│   ├── openai.py        # OpenAI LLM 适配
│   └── ...
├── embeddings/
│   ├── openai.py        # OpenAI Embedding
│   └── huggingface.py   # HuggingFace 本地模型
└── vector_stores/
    ├── qdrant.py        # Qdrant 向量库
    └── ...

记忆添加核心逻辑 ：

  
# mem0/memory/main.py 简化版
class Memory:
    def add(self, messages, user_id=None, metadata=None):
        # 1. 使用 LLM 从对话中提取记忆
        extracted_memories = self.llm.extract_memories(messages)
        
        # 2. 对每条记忆进行向量化
        for memory in extracted_memories:
            embedding = self.embedder.embed(memory)
            
            # 3. 检查是否存在相似记忆（去重）
            similar = self.vector_store.search(embedding, limit=1)
            if similar and similar[0].score > 0.9:
                # 更新已有记忆
                self.vector_store.update(similar[0].id, memory)
            else:
                # 添加新记忆
                self.vector_store.add(embedding, memory, metadata)

3.2 Letta：有状态 Agent 框架

3.2.1 项目概述

Letta（原 MemGPT）是一个用于构建有状态 LLM 服务的开源框架，提供了完整的 Agent 记忆管理能力。

GitHub ：https://github.com/letta-ai/letta

核心特性 ：

完整的 Agent 框架
多层记忆系统（核心记忆 + 归档记忆）
内置工具调用能力
可视化管理界面

3.2.2 架构设计

Letta 的架构更加复杂，采用了类似操作系统的设计理念：

持久化层

LLM 层

Agent 层

System Prompt
固定

Core Memory
可编辑

Archival Memory
向量化

Context Window

OpenAI

Claude

Local

PostgreSQL

pgvector

记忆层次 ：

System Prompt ：定义 Agent 的角色和行为
Core Memory ：关键信息的快速访问存储
Archival Memory ：大规模历史数据的向量化存储

3.2.3 核心概念

1. 虚拟上下文管理

Letta 的核心创新是”虚拟上下文”——通过智能的记忆分页机制，突破 LLM 上下文窗口的限制。

LLM 上下文窗口

System Prompt 固定

Core Memory 可编辑
- 用户信息
- Agent 状态

3.3 EverMemOS：企业级智能记忆系统

3.3.1 项目概述

EverMemOS 是由 EverMind-AI 开发的开源企业级智能记忆系统，旨在构建永不遗忘的 AI 记忆，使每次对话都基于先前的理解。

GitHub ：https://github.com/EverMind-AI/EverMemOS

Stars ：1.5k+

Forks ：147

License ：Apache-2.0

核心特性 ：

连贯叙事（Coherent Narrative） ：自动将对话片段连接为完整主题
证据感知（Evidence-Based Perception） ：主动捕捉记忆与任务的深层关联
动态画像（Living Profiles） ：实时更新用户画像

3.3.2 架构设计

EverMemOS 采用分层架构，主要包含以下模块：

存储层

推理融合层

记忆感知层

记忆构建层

输入层

对话输入

MemCell
原子记忆单元

多级记忆整合

记忆类型
事件/画像/偏好/关系

混合检索
RRF Fusion

智能重排
Reranker

检索策略
轻量/代理模式

多级记忆拼接

基于证据生成响应

MongoDB

Elasticsearch

Milvus

Redis

技术栈 ：

编程语言 ：Python 100%
数据库 ：MongoDB（主数据库）、Elasticsearch（关键词搜索）、Milvus（向量检索）、Redis（缓存）
部署：Docker + Docker Compose
工具链 ：uv（包管理器）、vLLM（模型部署）

3.3.3 核心源码结构

src/
├── agentic_layer/     # 统一记忆接口
├── memory_layer/      # 记忆提取（MemCell/多级记忆）
├── retrieval_layer/   # 检索与重排
├── biz_layer/         # 业务逻辑
├── infra_layer/       # 基础设施适配器
└── core/              # 核心功能（DI/生命周期管理）
demo/                  # 演示代码
evaluation/            # 评估框架
docs/                  # 文档

3.3.4 独特优势

特性	描述
连贯叙事	区分多线程对话（如”项目A进度”与”团队B策略”），从碎片化信息到完整故事
证据感知	主动捕捉记忆与任务的深层关联（如用户询问饮食推荐时，自动关联”两天前牙科手术”）
动态画像	实时更新用户画像（偏好、习惯、关注点），随交互深度”学习用户身份”
混合检索	并行执行语义和关键词检索，通过 RRF Fusion 融合结果

3.4 memory-bear：MCP 协议个人知识管理

3.4.1 项目概述

Memory Bear 是一个基于 MCP（Model Context Protocol）的个人知识管理系统，旨在解决 AI 助手的记忆检索限制问题。

GitHub ：https://github.com/ag4852/memory-bear

核心特性 ：

语义搜索 ：基于向量嵌入的跨笔记全文搜索
智能笔记创建 ：AI 辅助的结构化学习笔记生成
实时监控 ：自动索引文件的创建、修改和删除
闪卡系统 ：集成 FSRS 算法的间隔重复记忆巩固

3.4.2 架构设计

AI 层

MCP 层

向量层

监控层

文件系统

Markdown
Notes
Directory

File Watcher
Watchdog

Weaviate
Vector DB

MCP Server
FastMCP

AI Assistant
Claude等

技术栈 ：

语言：Python 3.10+
包管理 ：uv（推荐）或 pip
向量数据库 ：Weaviate（本地运行）
文件监控 ：Watchdog
MCP 框架 ：FastMCP
间隔重复算法 ：FSRS

3.4.3 MCP 工具函数

memory-bear 通过 MCP 协议向 AI 助手暴露以下工具：

  
# 笔记管理
search_notes(query)           # 语义搜索笔记
create_study_note(content)    # 创建学习笔记
find_best_match(query)        # 查找最佳匹配
edit_study_note(id, content)  # 编辑学习笔记

# 闪卡系统
get_cards_overview()          # 获取卡片概览
get_cards(filter)             # 获取卡片
create_cards_from_note(id)    # 从笔记创建卡片
update_card(id, data)         # 更新卡片

3.4.4 笔记格式

支持包含 frontmatter 元数据的 Markdown 文件：

  
---
title: "Machine Learning Basics"
subject: "computer-science"
tags: ["ml", "algorithms", "neural-networks"]
---

# Machine Learning Basics

正文内容...

3.5 四大框架对比

维度	mem0	Letta	EverMemOS	memory-bear
定位	轻量级记忆层	有状态 Agent 框架	企业级记忆系统	个人知识管理
复杂度	低	中	高	低
向量数据库	Qdrant	pgvector	Milvus	Weaviate
记忆类型	用户/会话/Agent	核心/归档	事件/画像/偏好/关系	笔记/闪卡
MCP 支持	✅	✅	❌	✅
Web UI	❌	✅	❌	❌
适用场景	快速集成	完整 Agent	企业级应用	个人学习
学习曲线	平缓	中等	陡峭	平缓

4. Docker 部署实战

4.1 环境准备

服务器配置 ：

操作系统：CentOS 7.9
CPU：4 核
内存：8GB
存储：100GB SSD

软件要求 ：

Docker 20.10+
Docker Compose 2.0+

服务器资源状态

查看内存使用情况：

查看磁盘使用情况：

查看 Docker 容器资源使用：

4.2 mem0 部署

部署目的 ：mem0 是一个轻量级的记忆层服务，需要两个核心组件配合工作：
Qdrant 向量数据库 ：用于存储记忆的向量表示，支持高效的语义相似度检索
mem0 REST API 服务 ：封装 mem0 核心功能，提供标准化的 HTTP 接口供应用调用

通过部署这两个组件，可以实现记忆的持久化存储和语义检索能力。

4.2.1 部署 Qdrant 向量数据库

步骤 1：创建数据目录

mkdir -p /data/mem0/qdrant

创建 Qdrant 数据目录

预期结果 ：目录创建成功，无输出。

步骤 2：启动 Qdrant 容器

  
docker run -d \
  --name qdrant \
  -v /data/mem0/qdrant:/qdrant/storage \
  -p 6333:6333 \
  -p 6334:6334 \
  --restart unless-stopped \
  qdrant/qdrant:latest

启动 Qdrant 容器

预期结果 ：返回容器 ID（64 位十六进制字符串）。

如果镜像不存在 ，Docker 会自动拉取：

Unable to find image 'qdrant/qdrant:latest' locally
latest: Pulling from qdrant/qdrant
...
Status: Downloaded newer image for qdrant/qdrant:latest

步骤 3：验证 Qdrant 运行状态

  
# 检查容器状态
docker ps | grep qdrant

# 预期输出：
# CONTAINER ID   IMAGE                  COMMAND      STATUS         PORTS
# abc123...      qdrant/qdrant:latest   "./qdrant"   Up 10 seconds  0.0.0.0:6333-6334->6333-6334/tcp

# 验证 API 可用
curl http://localhost:6333/collections

检查 Qdrant 容器状态

验证 Qdrant API 可用

成功标志 ：

  
{"result":{"collections":[]},"status":"ok","time":0.000123}

常见问题排查 ：

4.2.2 部署 mem0 REST API 服务

REST API 服务的作用 ：mem0 核心库是 Python 包，直接使用需要在每个应用中集成。通过部署 REST API 服务，可以：
统一接口 ：提供标准化的 HTTP 接口，任何语言都可以调用
服务复用 ：多个应用共享同一个记忆服务，避免重复部署
资源隔离 ：记忆服务独立运行，不影响业务应用的稳定性
便于扩展 ：后续可以轻松添加负载均衡、监控等能力

步骤 1：创建项目目录和文件

  
mkdir -p /data/mem0
cd /data/mem0

步骤 2：创建 Dockerfile

  
cat > Dockerfile << 'EOF'
FROM python:3.11-slim

WORKDIR /app

RUN pip install --no-cache-dir \
    mem0ai \
    qdrant-client \
    sentence-transformers \
    fastapi \
    uvicorn \
    openai

COPY server.py /app/server.py

EXPOSE 8000

CMD ["python", "server.py"]
EOF

步骤 3：创建 server.py

  
cat > server.py << 'EOF'
"""
mem0 REST API Server
提供记忆存储和检索的 HTTP 接口
"""
import os
from typing import Optional, List, Dict, Any
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import uvicorn
from mem0 import Memory

app = FastAPI(title="mem0 REST API", version="1.0.0")

# 添加 CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 初始化 Memory
config = {
    "llm": {
        "provider": "openai",
        "config": {
            "api_key": os.getenv("OPENAI_API_KEY"),
            "openai_base_url": "https://api.lkeap.cloud.tencent.com/v1",
            "model": "deepseek-v3",
        }
    },
    "embedder": {
        "provider": "huggingface",
        "config": {
            "model": "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2",
            "embedding_dims": 384
        }
    },
    "vector_store": {
        "provider": "qdrant",
        "config": {
            "host": "qdrant",
            "port": 6333,
            "collection_name": "mem0_memories",
            "embedding_model_dims": 384
        }
    }
}

memory = Memory.from_config(config)


class AddMemoryRequest(BaseModel):
    messages: List[Dict[str, str]]
    user_id: str
    metadata: Optional[Dict[str, Any]] = None


class SearchMemoryRequest(BaseModel):
    query: str
    user_id: str
    limit: Optional[int] = 10


@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "mem0"}


@app.post("/v1/memories")
async def add_memory(request: AddMemoryRequest):
    """添加记忆"""
    try:
        result = memory.add(
            messages=request.messages,
            user_id=request.user_id,
            metadata=request.metadata
        )
        return {"success": True, "result": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/v1/memories/{user_id}")
async def get_memories(user_id: str):
    """获取用户的所有记忆"""
    try:
        result = memory.get_all(user_id=user_id)
        return {"success": True, "memories": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/v1/memories/search")
async def search_memories(request: SearchMemoryRequest):
    """搜索记忆"""
    try:
        result = memory.search(
            query=request.query,
            user_id=request.user_id,
            limit=request.limit
        )
        return {"success": True, "results": result}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)
EOF

步骤 4：构建 Docker 镜像

docker build -t mem0-server:latest .

构建 mem0 Docker 镜像

预期输出 （构建过程）：

[+] Building 120.5s (8/8) FINISHED
 => [internal] load build definition from Dockerfile
 => [internal] load .dockerignore
 => [internal] load metadata for docker.io/library/python:3.11-slim
 => [1/4] FROM docker.io/library/python:3.11-slim
 => [2/4] WORKDIR /app
 => [3/4] RUN pip install --no-cache-dir mem0ai qdrant-client ...
 => [4/4] COPY server.py /app/server.py
 => exporting to image
Successfully built abc123def456
Successfully tagged mem0-server:latest

步骤 5：启动 mem0 服务

  
docker run -d \
  --name mem0 \
  -e OPENAI_API_KEY=your-api-key \
  --link qdrant:qdrant \
  -p 8000:8000 \
  --restart unless-stopped \
  mem0-server:latest

启动 mem0 服务容器

注意：首次启动需要下载 HuggingFace 模型（约 500MB），可能需要等待 2-3 分钟。

步骤 6：查看启动日志

docker logs -f mem0

模型下载中的日志 ：

Downloading (…)model.safetensors: 100%|██████████| 471M/471M [02:15<00:00, 3.48MB/s]
INFO:     Started server process [1]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)

步骤 7：验证服务状态

curl http://localhost:8000/health

检查 mem0 容器状态

验证 mem0 服务健康状态

查看 mem0 启动日志

成功标志 ：

  
{"status":"healthy","service":"mem0"}

常见问题排查 ：

问题	原因	解决方案
`Connection refused`	服务未启动完成	等待模型下载完成，查看 `docker logs mem0`
`OPENAI_API_KEY not set`	环境变量未设置	检查 `-e OPENAI_API_KEY=xxx` 参数
`Cannot connect to qdrant`	Qdrant 容器未启动	`docker start qdrant`
向量维度不匹配	集合已存在但维度不同	删除集合：`curl -X DELETE http://localhost:6333/collections/mem0_memories`

4.3 Letta 部署

4.3.1 部署 PostgreSQL 数据库

步骤 1：创建数据目录

mkdir -p /data/letta/postgres

步骤 2：启动 PostgreSQL（带 pgvector 扩展）

  
docker run -d \
  --name letta_db \
  -e POSTGRES_USER=letta \
  -e POSTGRES_PASSWORD=letta123 \
  -e POSTGRES_DB=letta \
  -v /data/letta/postgres:/var/lib/postgresql/data \
  -p 5433:5432 \
  --restart unless-stopped \
  pgvector/pgvector:pg16

步骤 3：等待数据库启动并启用 pgvector 扩展

  
# 等待数据库启动
sleep 10

# 启用 pgvector 扩展
docker exec letta_db psql -U letta -d letta -c "CREATE EXTENSION IF NOT EXISTS vector;"

成功标志 ：

CREATE EXTENSION

检查 Letta 容器状态

验证 pgvector 扩展已启用

4.3.2 部署 Letta 服务

步骤 1：创建配置目录

mkdir -p /data/letta/config

步骤 2：启动 Letta

  
docker run -d \
  --name letta \
  -e OPENAI_API_KEY=your-api-key \
  -e OPENAI_API_BASE=https://api.lkeap.cloud.tencent.com/v1 \
  -e LETTA_PG_URI=postgresql://letta:letta123@letta_db:5432/letta \
  -v /data/letta/config:/root/.letta \
  -p 8283:8283 \
  --link letta_db:letta_db \
  --restart unless-stopped \
  letta/letta:latest

步骤 3：验证服务状态

  
# 检查容器状态
docker ps | grep letta

# 检查 Web UI
curl -s http://localhost:8283/ | head -20

成功标志 ：返回 HTML 页面内容。

步骤 4：访问 Web 管理界面

浏览器打开：http://your-server-ip:8283

4.4 EverMemOS 部署

EverMemOS 采用 Docker Compose 部署依赖服务（MongoDB/Elasticsearch/Milvus/Redis），主程序通过 Python 本地运行。

4.4.1 启动依赖服务

步骤 1：创建项目目录并克隆仓库

  
mkdir -p /data/evermemos
cd /data/evermemos
git clone https://github.com/EverMind-AI/EverMemOS.git .

步骤 2：启动依赖服务

docker compose up -d

预期输出 ：

Creating network "evermemos_memsys-network" with driver "bridge"
Creating volume "evermemos_mongodb_data" with local driver
Creating volume "evermemos_elasticsearch_data" with local driver
Creating volume "evermemos_milvus_data" with local driver
Creating volume "evermemos_redis_data" with local driver
Container memsys-mongodb  Created
Container memsys-elasticsearch  Created
Container memsys-milvus-etcd  Created
Container memsys-milvus-minio  Created
Container memsys-redis  Created
Container memsys-milvus-standalone  Created

EverMemOS 依赖服务状态

4.4.2 验证依赖服务

验证 MongoDB 连接

  
docker exec memsys-mongodb mongosh --eval "db.adminCommand('ping')" --quiet

验证 Elasticsearch 健康状态

curl -s http://localhost:19200/_cluster/health | python3 -m json.tool

验证 Milvus 健康状态

curl -s http://localhost:9091/healthz

验证 Redis 连接

docker exec memsys-redis redis-cli ping

4.4.3 Redis 安全配置（重要）

安全提示 ：默认情况下，Redis 允许无密码访问，这在生产环境中存在安全风险。建议配置密码认证。

步骤 1：进入 Redis 容器配置密码

  
# 进入 Redis 容器
docker exec -it memsys-redis redis-cli

# 设置密码（在 redis-cli 中执行）
CONFIG SET requirepass "your_redis_password"

# 验证密码设置
AUTH your_redis_password
PING

步骤 2：持久化密码配置

为了确保容器重启后密码仍然生效，需要修改 Redis 配置文件：

  
# 在 docker-compose.yml 同级目录创建 redis.conf
cat > redis.conf << 'EOF'
# Redis 安全配置
requirepass your_redis_password

# 绑定地址（仅允许内部网络访问）
bind 0.0.0.0

# 禁用危险命令
rename-command FLUSHDB ""
rename-command FLUSHALL ""
rename-command CONFIG ""
rename-command SHUTDOWN ""
EOF

步骤 3：修改 docker-compose.yml 挂载配置文件

  
# 在 docker-compose.yml 的 redis 服务中添加
services:
  redis:
    image: redis:7
    container_name: memsys-redis
    volumes:
      - redis_data:/data
      - ./redis.conf:/usr/local/etc/redis/redis.conf
    command: redis-server /usr/local/etc/redis/redis.conf
    networks:
      - memsys-network

步骤 4：重启 Redis 服务

docker compose down
docker compose up -d

步骤 5：验证密码认证

  
# 无密码访问应该被拒绝
docker exec memsys-redis redis-cli ping
# 预期输出: NOAUTH Authentication required.

# 使用密码访问
docker exec memsys-redis redis-cli -a your_redis_password ping
# 预期输出: PONG

注意：配置 Redis 密码后，需要同步更新 EverMemOS 的 .env 文件中的 Redis 连接配置。

4.4.4 安装依赖并配置

  
# 安装 uv（如果未安装）
curl -LsSf https://astral.sh/uv/install.sh | sh
source ~/.local/bin/env

# 同步依赖
uv sync

# 配置环境变量
cp env.template .env

编辑.env 文件，配置 LLM、Embedding API 和 Redis：

  
# LLM 配置（使用腾讯云 DeepSeek API）
LLM_PROVIDER=openai
LLM_MODEL=deepseek-v3
LLM_BASE_URL=https://api.lkeap.cloud.tencent.com/v1
LLM_API_KEY=your-api-key

# 向量化服务配置
VECTORIZE_PROVIDER=deepinfra
VECTORIZE_API_KEY=your-deepinfra-key
VECTORIZE_MODEL=Qwen/Qwen3-Embedding-4B

# 重排序服务配置
RERANK_PROVIDER=deepinfra
RERANK_API_KEY=your-deepinfra-key
RERANK_MODEL=Qwen/Qwen3-Reranker-4B

# Redis 配置（添加密码认证）
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PASSWORD=your_redis_password

4.4.5 启动 API 服务

uv run python src/run.py --port 8001

成功标志 ：

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8001 (Press CTRL+C to quit)

常见问题排查 ：

问题	原因	解决方案
`MongoDB connection failed`	MongoDB 未启动	`docker start memsys-mongodb`
`Elasticsearch connection refused`	ES 未就绪	等待 ES 健康检查通过
`Milvus connection failed`	Milvus 依赖未启动	确保 etcd 和 minio 已启动
`API key not set`	环境变量未配置	检查 `.env` 文件

4.5 memory-bear 部署

memory-bear 是一个基于 MCP 协议的个人知识管理系统，依赖 Weaviate 向量数据库。

4.5.1 安装 Weaviate

  
docker run -d \
  --name weaviate \
  -p 8080:8080 \
  -p 50051:50051 \
  --restart unless-stopped \
  semitechnologies/weaviate:latest

Weaviate 向量数据库状态

验证 Weaviate 服务信息

curl -s http://localhost:8080/v1/meta | python3 -m json.tool

4.5.2 安装 memory-bear

  
# 克隆仓库
git clone https://github.com/ag4852/memory-bear.git
cd memory-bear

# 安装依赖（使用 uv）
uv sync

memory-bear 依赖安装状态

4.5.3 配置环境变量

  
cat > .env << 'EOF'
NOTES_DIR=/path/to/your/notes
HUGGINGFACE_API_KEY=your_hf_token_here
CONTENT_TAGS=lecture,homework,exam,concepts,research
SUBJECTS=math,science,history,general
EOF

4.5.4 启动 MCP 服务器

  
# 使用 uvx 启动（推荐）
uvx memory-bear

# 或使用本地安装
uv run memory-bear --server

成功标志 ：

INFO:     Started MCP server
INFO:     Listening on stdio

常见问题排查 ：

问题	原因	解决方案
`Weaviate connection failed`	Weaviate 未启动	`docker start weaviate`
`HuggingFace API error`	API Key 无效	检查 `HUGGINGFACE_API_KEY`
`Notes directory not found`	目录不存在	创建 `NOTES_DIR` 指定的目录

4.6 配置 alias 快捷命令

将以下内容添加到 ~/.bashrc：

  
# Agent Memory Services
alias mem0-start="docker start qdrant mem0"
alias mem0-stop="docker stop mem0 qdrant"
alias mem0-logs="docker logs -f mem0"
alias mem0-status="docker ps | grep -E 'mem0|qdrant'"

alias letta-start="docker start letta_db letta"
alias letta-stop="docker stop letta letta_db"
alias letta-logs="docker logs -f letta"
alias letta-status="docker ps | grep -E 'letta'"

alias evermemos-start="cd /path/to/EverMemOS && docker-compose up -d && uv run python src/run.py --port 8001"
alias evermemos-stop="cd /path/to/EverMemOS && docker-compose down"

执行 source ~/.bashrc 使配置生效。

5. 代码集成示例

5.1 mem0 Python SDK

  
"""
mem0 Python 集成示例
演示记忆的添加、搜索和管理
"""
import requests
import json

# mem0 REST API 服务地址
MEM0_BASE_URL = "http://localhost:8000"


def add_memory(messages: list, user_id: str):
    """
    添加记忆到 mem0
    
    Args:
        messages: 对话消息列表，格式为 [{"role": "user/assistant", "content": "..."}]
        user_id: 用户唯一标识，用于隔离不同用户的记忆
    
    Returns:
        dict: 包含添加结果的响应，包括提取出的记忆内容
    """
    response = requests.post(
        f"{MEM0_BASE_URL}/v1/memories",
        json={
            "messages": messages,  # 对话内容，mem0 会自动提取关键记忆
            "user_id": user_id     # 用户标识
        }
    )
    return response.json()


def search_memory(query: str, user_id: str, limit: int = 5):
    """
    语义搜索用户记忆
    
    Args:
        query: 搜索查询文本，支持自然语言
        user_id: 用户唯一标识
        limit: 返回结果数量上限
    
    Returns:
        dict: 包含匹配记忆列表，按相关度排序
    """
    response = requests.post(
        f"{MEM0_BASE_URL}/v1/memories/search",
        json={
            "query": query,      # 搜索关键词或问题
            "user_id": user_id,  # 限定搜索范围为该用户
            "limit": limit       # 最多返回多少条结果
        }
    )
    return response.json()


def get_all_memories(user_id: str):
    """
    获取用户的所有记忆
    
    Args:
        user_id: 用户唯一标识
    
    Returns:
        dict: 包含该用户所有记忆的列表
    """
    response = requests.get(f"{MEM0_BASE_URL}/v1/memories/{user_id}")
    return response.json()


# 使用示例
if __name__ == "__main__":
    user_id = "demo_user"
    
    # 1. 添加记忆 - 通过对话内容自动提取用户信息
    result = add_memory(
        messages=[
            {"role": "user", "content": "我叫小明，是一名前端开发工程师，喜欢用 React"},
            {"role": "assistant", "content": "你好小明！React 是很棒的前端框架。"}
        ],
        user_id=user_id
    )
    print("添加记忆:", json.dumps(result, ensure_ascii=False, indent=2))
    
    # 2. 搜索记忆 - 使用自然语言查询
    result = search_memory(
        query="这个用户是做什么工作的",  # 语义搜索，不需要精确匹配
        user_id=user_id
    )
    print("搜索结果:", json.dumps(result, ensure_ascii=False, indent=2))
    
    # 3. 获取所有记忆 - 查看用户的完整记忆列表
    result = get_all_memories(user_id)
    print("所有记忆:", json.dumps(result, ensure_ascii=False, indent=2))

mem0 Python SDK 执行结果

执行结果分析 ：

  
// 添加记忆返回
{
  "success": true,
  "result": {
    "results": [
      {
        "id": "ce0d6989-bb10-4bf5-a811-9047a5ca3f08",
        "memory": "Name is 小明",
        "event": "ADD"
      },
      {
        "id": "0ed54aa2-7c6b-4c76-9194-743b949e5c84",
        "memory": "Is a 前端开发工程师",
        "event": "ADD"
      },
      {
        "id": "e78b5808-a9df-417f-944f-548fabdb87af",
        "memory": "Likes to use React",
        "event": "ADD"
      }
    ]
  }
}

// 搜索结果返回
{
  "success": true,
  "results": {
    "results": [
      {
        "id": "52c5e914-b568-4426-b73e-fc7c4d0450d8",
        "memory": "在腾讯工作",
        "score": 0.6428276
      },
      {
        "id": "0ed54aa2-7c6b-4c76-9194-743b949e5c84",
        "memory": "Is a 前端开发工程师",
        "score": 0.34127486
      }
    ]
  }
}

5.2 mem0 Go SDK

  
package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
)

const mem0BaseURL = "http://localhost:8000"

// Message 对话消息
type Message struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

// AddMemoryRequest 添加记忆请求
type AddMemoryRequest struct {
	Messages []Message              `json:"messages"`
	UserID   string                 `json:"user_id"`
	Metadata map[string]interface{} `json:"metadata,omitempty"`
}

// SearchMemoryRequest 搜索记忆请求
type SearchMemoryRequest struct {
	Query  string `json:"query"`
	UserID string `json:"user_id"`
	Limit  int    `json:"limit,omitempty"`
}

// AddMemory 添加记忆
func AddMemory(messages []Message, userID string) (map[string]interface{}, error) {
	req := AddMemoryRequest{
		Messages: messages,
		UserID:   userID,
	}
	
	body, _ := json.Marshal(req)
	resp, err := http.Post(
		mem0BaseURL+"/v1/memories",
		"application/json",
		bytes.NewBuffer(body),
	)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	
	var result map[string]interface{}
	respBody, _ := io.ReadAll(resp.Body)
	json.Unmarshal(respBody, &result)
	return result, nil
}

// SearchMemory 搜索记忆
func SearchMemory(query, userID string, limit int) (map[string]interface{}, error) {
	req := SearchMemoryRequest{
		Query:  query,
		UserID: userID,
		Limit:  limit,
	}
	
	body, _ := json.Marshal(req)
	resp, err := http.Post(
		mem0BaseURL+"/v1/memories/search",
		"application/json",
		bytes.NewBuffer(body),
	)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()
	
	var result map[string]interface{}
	respBody, _ := io.ReadAll(resp.Body)
	json.Unmarshal(respBody, &result)
	return result, nil
}

func main() {
	userID := "go_demo_user"
	
	// 1. 添加记忆
	messages := []Message{
		{Role: "user", Content: "我是小红，专注于 Go 语言后端开发"},
		{Role: "assistant", Content: "你好小红！Go 语言是很棒的后端开发语言。"},
	}
	
	result, err := AddMemory(messages, userID)
	if err != nil {
		fmt.Printf("添加记忆失败: %v\n", err)
		return
	}
	fmt.Printf("添加记忆成功: %v\n", result)
	
	// 2. 搜索记忆
	searchResult, err := SearchMemory("用户的技术栈是什么", userID, 5)
	if err != nil {
		fmt.Printf("搜索记忆失败: %v\n", err)
		return
	}
	fmt.Printf("搜索结果: %v\n", searchResult)
}

5.3 Letta Python SDK

  
"""
Letta Python 集成示例
演示 Agent 创建和对话
"""
from letta import create_client

# 创建客户端
client = create_client(base_url="http://localhost:8283")

# 创建 Agent
agent = client.create_agent(
    name="memory_demo_agent",
    llm_config={
        "model": "deepseek-v3",
        "model_endpoint_type": "openai",
        "model_endpoint": "https://api.lkeap.cloud.tencent.com/v1",
    },
    embedding_config={
        "embedding_model": "text-embedding-ada-002",
        "embedding_endpoint_type": "openai",
        "embedding_endpoint": "https://api.lkeap.cloud.tencent.com/v1",
    }
)

print(f"Agent 创建成功: {agent.id}")

# 与 Agent 对话
response = client.send_message(
    agent_id=agent.id,
    message="你好，我叫小华，我是一名后端工程师",
    role="user"
)

print(f"Agent 回复: {response.messages}")

# 继续对话，测试记忆
response = client.send_message(
    agent_id=agent.id,
    message="你还记得我是谁吗？我是做什么工作的？",
    role="user"
)

print(f"Agent 回复: {response.messages}")

5.4 EverMemOS Python SDK

  
"""
EverMemOS Python 集成示例
演示记忆的添加、搜索和用户画像获取
"""
import requests
import json

EVERMEMOS_BASE_URL = "http://localhost:8001"


def add_memory(user_id: str, content: str, memory_type: str = "event"):
    """
    添加记忆
    memory_type: event(事件), profile(画像), preference(偏好), relation(关系)
    """
    response = requests.post(
        f"{EVERMEMOS_BASE_URL}/v1/memories",
        json={
            "user_id": user_id,
            "content": content,
            "memory_type": memory_type
        }
    )
    return response.json()


def search_memory(user_id: str, query: str, limit: int = 10, mode: str = "hybrid"):
    """
    搜索记忆（支持混合检索）
    mode: semantic(语义), keyword(关键词), hybrid(混合)
    """
    response = requests.post(
        f"{EVERMEMOS_BASE_URL}/v1/memories/search",
        json={
            "user_id": user_id,
            "query": query,
            "limit": limit,
            "retrieval_mode": mode
        }
    )
    return response.json()


def get_user_profile(user_id: str):
    """获取用户动态画像"""
    response = requests.get(f"{EVERMEMOS_BASE_URL}/v1/users/{user_id}/profile")
    return response.json()


# 使用示例
if __name__ == "__main__":
    user_id = "evermemos_demo_user"
    
    # 1. 添加事件记忆
    result = add_memory(
        user_id=user_id,
        content="用户今天完成了 Go 语言的并发编程学习",
        memory_type="event"
    )
    print("添加事件记忆:", json.dumps(result, ensure_ascii=False, indent=2))
    
    # 2. 添加偏好记忆
    result = add_memory(
        user_id=user_id,
        content="用户偏好使用 Vim 编辑器",
        memory_type="preference"
    )
    print("添加偏好记忆:", json.dumps(result, ensure_ascii=False, indent=2))
    
    # 3. 混合检索
    result = search_memory(
        user_id=user_id,
        query="用户学习了什么技术",
        mode="hybrid"
    )
    print("混合检索结果:", json.dumps(result, ensure_ascii=False, indent=2))
    
    # 4. 获取用户画像
    result = get_user_profile(user_id)
    print("用户画像:", json.dumps(result, ensure_ascii=False, indent=2))

EverMemOS 独特功能 ：

功能	说明	使用场景
混合检索	语义 + 关键词双重匹配	提高检索准确率
记忆类型	event/profile/preference/relation	结构化记忆管理
动态画像	自动聚合用户特征	个性化推荐
连贯叙事	自动关联多线程对话	复杂场景理解

5.5 memory-bear MCP 调用示例

memory-bear 原生支持 MCP 协议，可以直接在支持 MCP 的 AI 助手中使用。

  
"""
memory-bear MCP 工具调用示例
演示笔记管理和闪卡系统
"""
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def demo_memory_bear():
    """演示 memory-bear MCP 工具调用"""
    
    # 连接 MCP Server
    server_params = StdioServerParameters(
        command="uvx",
        args=["memory-bear"],
        env={
            "NOTES_DIR": "/path/to/notes",
            "HUGGINGFACE_API_KEY": "your-token"
        }
    )
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 1. 创建学习笔记
            result = await session.call_tool(
                "create_study_note",
                arguments={
                    "title": "Go 并发编程学习笔记",
                    "content": """
# Go 并发编程

## goroutine
- 轻量级线程
- 使用 go 关键字启动

## channel
- goroutine 间通信
- 支持缓冲和非缓冲

## select
- 多路复用
- 监听多个 channel
                    """,
                    "tags": ["go", "concurrency", "goroutine"]
                }
            )
            print("创建笔记:", json.dumps(result, indent=2))
            
            # 2. 语义搜索笔记
            result = await session.call_tool(
                "search_notes",
                arguments={
                    "query": "Go 语言并发"
                }
            )
            print("搜索结果:", json.dumps(result, indent=2))
            
            # 3. 从笔记创建闪卡
            result = await session.call_tool(
                "create_cards_from_note",
                arguments={
                    "note_id": "note_xxx"  # 使用实际的笔记 ID
                }
            )
            print("创建闪卡:", json.dumps(result, indent=2))
            
            # 4. 获取闪卡概览
            result = await session.call_tool(
                "get_cards_overview",
                arguments={}
            )
            print("闪卡概览:", json.dumps(result, indent=2))

# 运行示例
if __name__ == "__main__":
    import asyncio
    asyncio.run(demo_memory_bear())

memory-bear MCP 工具说明 ：

工具名	功能	参数
`search_notes`	语义搜索笔记	`query`
`create_study_note`	创建学习笔记	`title`, `content`, `tags`
`find_best_match`	查找最佳匹配	`query`
`edit_study_note`	编辑笔记	`note_id`, `content`
`get_cards_overview`	获取闪卡概览	-
`get_cards`	获取闪卡列表	`filter`
`create_cards_from_note`	从笔记创建闪卡	`note_id`
`update_card`	更新闪卡	`card_id`, `data`

6. MCP Server 部署实战

本章详细介绍如何在 CVM 服务器上部署四种记忆系统的 MCP Server，实现标准化的 MCP 协议接口，供 CodeBuddy 进行远程调用。

6.1 MCP Server 部署架构

CVM 云服务器

本地开发机

存储层

MCP Server 层

SSH + MCP Protocol

CodeBuddy

mem0 MCP
端口 8000

Letta MCP
端口 8283

EverMemOS MCP
端口 8001

memory-bear MCP
端口 8002

Qdrant

PostgreSQL

Milvus

Weaviate

6.2 环境准备

服务器信息 ：

操作系统：CentOS 7.9
IP：xxx.xxx.xxx.xxx（内网服务器）
SSH 端口：22

已部署的基础服务 ：

  
# 查看已运行的 Docker 容器
docker ps --format 'table {{.Names}}\t{{.Image}}\t{{.Status}}\t{{.Ports}}'

``` 容器名| 镜像| 状态| 端口  
---|---|---|---  
qdrant| qdrant/qdrant:latest| Up| 6333-6334  
letta_db| pgvector/pgvector:pg16| Up| 5433  
letta| letta/letta:latest| Up| 8283  
mem0| mem0-server:latest| Up| 8000  
weaviate| semitechnologies/weaviate| Up| 8080  
memsys-mongodb| mongo:7| Up| 27017  
memsys-elasticsearch| elasticsearch:8.15.0| Up| 19200  
memsys-milvus-standalone| milvusdb/milvus| Up| 19530  
memsys-redis| redis:7| Up| 6379  
  
#### 6.3 MCP Server 目录结构

```bash
# 创建 MCP Server 目录
mkdir -p /data/mcp-servers/{mem0,letta,evermemos,memory-bear}

目录结构：

/data/mcp-servers/
├── mem0/
│   ├── mcp_server.py      # MCP Server 实现
│   ├── pyproject.toml     # uv 项目配置
│   └── .venv/             # Python 虚拟环境
├── letta/
│   ├── mcp_server.py
│   ├── pyproject.toml
│   └── .venv/
├── evermemos/
│   ├── mcp_server.py
│   ├── pyproject.toml
│   └── .venv/
└── memory-bear/
    ├── mcp_server.py
    ├── pyproject.toml
    └── .venv/

6.4 mem0 MCP Server 部署

6.4.1 初始化项目

  
cd /data/mcp-servers/mem0
uv init --name mem0-mcp
uv add 'mcp[cli]' httpx

初始化 mem0 MCP Server 项目

6.4.2 创建 MCP Server

  
# /data/mcp-servers/mem0/mcp_server.py
"""
mem0 MCP Server
将 mem0 REST API 封装为 MCP 协议接口，供 AI 助手调用
"""
import os
import json
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx

# 创建 MCP Server 实例，名称用于标识服务
app = Server("mem0-mcp")

# mem0 REST API 地址，默认为本地服务
MEM0_BASE_URL = os.getenv("MEM0_BASE_URL", "http://localhost:8000")


@app.list_tools()
async def list_tools():
    """
    声明 MCP Server 提供的工具列表
    AI 助手会根据这些描述决定何时调用哪个工具
    """
    return [
        # 添加记忆工具
        Tool(
            name="add_memory",
            description="添加记忆到 mem0，从对话中提取关键信息并持久化存储",
            inputSchema={
                "type": "object",
                "properties": {
                    "messages": {
                        "type": "string",
                        "description": "JSON 格式的消息列表，如 [{\"role\":\"user\",\"content\":\"...\"}]"
                    },
                    "user_id": {
                        "type": "string",
                        "description": "用户唯一标识，用于隔离不同用户的记忆"
                    }
                },
                "required": ["messages", "user_id"]
            }
        ),
        # 搜索记忆工具
        Tool(
            name="search_memory",
            description="语义搜索用户记忆，根据查询内容返回相关记忆",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索查询文本，支持自然语言"
                    },
                    "user_id": {
                        "type": "string",
                        "description": "用户唯一标识"
                    }
                },
                "required": ["query", "user_id"]
            }
        ),
    ]


@app.call_tool()
async def call_tool(name: str, arguments: dict):
    """
    处理工具调用请求
    根据工具名称分发到对应的处理逻辑
    """
    # 创建异步 HTTP 客户端，设置 30 秒超时
    async with httpx.AsyncClient(timeout=30.0) as client:
        if name == "add_memory":
            # 调用 mem0 添加记忆接口
            resp = await client.post(
                f"{MEM0_BASE_URL}/v1/memories",
                json={
                    "messages": json.loads(arguments["messages"]),  # 解析 JSON 字符串
                    "user_id": arguments["user_id"]
                }
            )
        elif name == "search_memory":
            # 调用 mem0 搜索记忆接口
            resp = await client.post(
                f"{MEM0_BASE_URL}/v1/memories/search",
                json={
                    "query": arguments["query"],
                    "user_id": arguments["user_id"],
                    "limit": 10  # 默认返回 10 条结果
                }
            )
        else:
            # 未知工具，返回错误信息
            return [TextContent(type="text", text="Unknown tool")]
        
        # 返回 API 响应内容
        return [TextContent(type="text", text=resp.text)]


async def main():
    """
    MCP Server 入口函数
    使用 stdio 模式与客户端通信（标准输入/输出）
    """
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )


if __name__ == "__main__":
    asyncio.run(main())

6.4.3 测试 MCP Server

  
cd /data/mcp-servers/mem0
timeout 3 uv run python mcp_server.py || echo 'MCP Server test completed'

测试成功标志 ：

MCP Server test completed

6.5 Letta MCP Server 部署

6.5.1 初始化项目

  
cd /data/mcp-servers/letta
uv init --name letta-mcp
uv add 'mcp[cli]' httpx

6.5.2 创建 MCP Server

  
# /data/mcp-servers/letta/mcp_server.py
import os
import json
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx

app = Server("letta-mcp")
LETTA_BASE_URL = os.getenv("LETTA_BASE_URL", "http://localhost:8283")

@app.list_tools()
async def list_tools():
    return [
        Tool(name="list_agents", description="列出所有 Agent",
             inputSchema={"type":"object","properties":{}}),
        Tool(name="create_agent", description="创建 Agent",
             inputSchema={"type":"object","properties":{
                 "name":{"type":"string"}
             },"required":["name"]}),
        Tool(name="send_message", description="向 Agent 发送消息",
             inputSchema={"type":"object","properties":{
                 "agent_id":{"type":"string"},"message":{"type":"string"}
             },"required":["agent_id","message"]}),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    async with httpx.AsyncClient(timeout=60.0) as client:
        if name == "list_agents":
            resp = await client.get(f"{LETTA_BASE_URL}/v1/agents")
        elif name == "create_agent":
            resp = await client.post(f"{LETTA_BASE_URL}/v1/agents",
                json={"name": arguments["name"]})
        elif name == "send_message":
            resp = await client.post(
                f"{LETTA_BASE_URL}/v1/agents/{arguments['agent_id']}/messages",
                json={"messages": [{"role":"user","content":arguments["message"]}]})
        else:
            return [TextContent(type="text", text="Unknown tool")]
        return [TextContent(type="text", text=resp.text)]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

6.6 EverMemOS MCP Server 部署

6.6.1 初始化项目

  
cd /data/mcp-servers/evermemos
uv init --name evermemos-mcp
uv add 'mcp[cli]' httpx

6.6.2 创建 MCP Server

  
# /data/mcp-servers/evermemos/mcp_server.py
import os
import json
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx

app = Server("evermemos-mcp")
EVERMEMOS_BASE_URL = os.getenv("EVERMEMOS_BASE_URL", "http://localhost:8001")

@app.list_tools()
async def list_tools():
    return [
        Tool(name="add_memory", description="添加记忆到 EverMemOS",
             inputSchema={"type":"object","properties":{
                 "user_id":{"type":"string"},"content":{"type":"string"}
             },"required":["user_id","content"]}),
        Tool(name="search_memory", description="搜索记忆(支持混合检索)",
             inputSchema={"type":"object","properties":{
                 "user_id":{"type":"string"},"query":{"type":"string"}
             },"required":["user_id","query"]}),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    async with httpx.AsyncClient(timeout=30.0) as client:
        if name == "add_memory":
            resp = await client.post(f"{EVERMEMOS_BASE_URL}/v1/memories",
                json={"user_id": arguments["user_id"],
                      "content": arguments["content"], "memory_type": "event"})
        elif name == "search_memory":
            resp = await client.post(f"{EVERMEMOS_BASE_URL}/v1/memories/search",
                json={"user_id": arguments["user_id"],
                      "query": arguments["query"], "retrieval_mode": "hybrid"})
        else:
            return [TextContent(type="text", text="Unknown tool")]
        return [TextContent(type="text", text=resp.text)]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

6.7 memory-bear MCP Server 部署

6.7.1 初始化项目

  
cd /data/mcp-servers/memory-bear
uv init --name memory-bear-mcp
uv add 'mcp[cli]' httpx

6.7.2 创建 MCP Server

  
# /data/mcp-servers/memory-bear/mcp_server.py
import os
import json
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx

app = Server("memory-bear-mcp")
MEMORY_BEAR_BASE_URL = os.getenv("MEMORY_BEAR_BASE_URL", "http://localhost:8002")

@app.list_tools()
async def list_tools():
    return [
        Tool(name="add_memory", description="添加记忆到 memory-bear",
             inputSchema={"type":"object","properties":{
                 "user_id":{"type":"string"},"content":{"type":"string"},
                 "tags":{"type":"string","description":"逗号分隔的标签"}
             },"required":["user_id","content"]}),
        Tool(name="search_memory", description="搜索记忆",
             inputSchema={"type":"object","properties":{
                 "user_id":{"type":"string"},"query":{"type":"string"}
             },"required":["user_id","query"]}),
        Tool(name="get_memories", description="获取用户所有记忆",
             inputSchema={"type":"object","properties":{
                 "user_id":{"type":"string"}
             },"required":["user_id"]}),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    async with httpx.AsyncClient(timeout=30.0) as client:
        if name == "add_memory":
            resp = await client.post(f"{MEMORY_BEAR_BASE_URL}/v1/memories",
                json={"user_id": arguments["user_id"], "content": arguments["content"]})
        elif name == "search_memory":
            resp = await client.post(f"{MEMORY_BEAR_BASE_URL}/v1/memories/search",
                json={"user_id": arguments["user_id"], "query": arguments["query"]})
        else:
            return [TextContent(type="text", text="Unknown tool")]
        return [TextContent(type="text", text=resp.text)]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

6.8 配置 alias 快捷命令

在 ~/.bashrc 中添加以下 alias：

  
# MCP Server aliases
alias mcp-mem0='cd /data/mcp-servers/mem0 && uv run python mcp_server.py'
alias mcp-letta='cd /data/mcp-servers/letta && uv run python mcp_server.py'
alias mcp-evermemos='cd /data/mcp-servers/evermemos && uv run python mcp_server.py'
alias mcp-memory-bear='cd /data/mcp-servers/memory-bear && uv run python mcp_server.py'
alias mcp-ls='ls -la /data/mcp-servers/'

执行 source ~/.bashrc 使配置生效。

验证 alias 配置 ：

  
# 查看已配置的 MCP alias
cat ~/.bashrc | grep -E '^alias mcp'

预期输出：

alias mcp-mem0='cd /data/mcp-servers/mem0 && uv run python mcp_server.py'
alias mcp-letta='cd /data/mcp-servers/letta && uv run python mcp_server.py'
alias mcp-evermemos='cd /data/mcp-servers/evermemos && uv run python mcp_server.py'
alias mcp-memory-bear='cd /data/mcp-servers/memory-bear && uv run python mcp_server.py'
alias mcp-ls='ls -la /data/mcp-servers/'

验证 alias 配置

执行 source ~/.bashrc 后，使用 alias | grep mcp 验证配置是否生效。

6.9 MCP Server 目录验证

  
# 查看 MCP Server 目录结构
ls -la /data/mcp-servers/

MCP Server 目录结构

执行 ls -la /data/mcp-servers/ 查看目录结构。

预期输出：

总用量 24
drwxr-xr-x  6 root root 4096 1月   5 19:19 .
drwxr-xr-x 18 root root 4096 1月   5 19:20 ..
drwxr-xr-x  4 root root 4096 1月   5 18:24 evermemos
drwxr-xr-x  4 root root 4096 1月   5 18:22 letta
drwxr-xr-x  4 root root 4096 1月   5 18:19 mem0
drwxr-xr-x  4 root root 4096 1月   5 19:19 memory-bear

6.10 MCP Server 工具汇总

MCP Server	工具名	功能	参数
mem0	`add_memory`	添加记忆	`messages`(JSON), `user_id`
	`search_memory`	搜索记忆	`query`, `user_id`
letta	`list_agents`	列出所有 Agent	-
	`create_agent`	创建 Agent	`name`
	`send_message`	向 Agent 发送消息	`agent_id`, `message`
evermemos	`add_memory`	添加记忆	`user_id`, `content`
	`search_memory`	搜索记忆(混合检索)	`user_id`, `query`
memory-bear	`add_memory`	添加记忆	`user_id`, `content`, `tags`
	`search_memory`	搜索记忆	`user_id`, `query`
	`get_memories`	获取所有记忆	`user_id`

7. 四种开源记忆系统与 CodeBuddy MCP 集成实践

本章将详细介绍如何将 mem0、Letta、EverMemOS、memory-bear 四种开源记忆系统通过 MCP（Model Context Protocol）协议集成到 CodeBuddy 中，实现 AI 编程助手的长期记忆能力。

部署架构说明 ：本章假设 Docker 服务部署在云服务器（CVM）上，CodeBuddy 运行在本地开发机上。MCP Server 在 CVM 上以 HTTP/SSE 模式运行，CodeBuddy 通过 HTTP 协议连接。

7.1 MCP 协议简介

MCP（Model Context Protocol）是 Anthropic 提出的开放协议，用于标准化 AI 助手与外部工具/服务的交互方式。

传输模式对比 ：

模式	特点	适用场景
stdio	标准输入输出，本地进程通信	本地 MCP Server
SSH + stdio	通过 SSH 隧道远程启动 stdio 进程	远程服务器（连接不稳定）
HTTP/SSE	HTTP 请求 + Server-Sent Events	远程服务器（推荐）

为什么选择 HTTP/SSE 模式？
SSH + stdio 模式在网络不稳定时容易断开连接
HTTP/SSE 模式支持长连接，服务端常驻运行，稳定性更好
便于负载均衡和服务监控

核心概念 ：

云服务器 CVM

MCP 协议 (HTTP/SSE)

本地开发机

HTTP/SSE

CodeBuddy
AI 助手

Tools
工具调用

Resources
资源访问

Prompts
提示模板

mem0 MCP Server
HTTP :9002/sse

Letta MCP Server
HTTP :9003/sse

EverMemOS MCP Server
HTTP :9004/sse

memory-bear MCP Server
HTTP :9005/sse

MCP 的优势 ：

标准化接口 ：统一的工具调用规范
双向通信 ：支持请求/响应和事件推送（SSE）
安全隔离 ：服务端运行在独立进程
易于扩展 ：任何服务都可以封装为 MCP Server
连接稳定 ：HTTP/SSE 模式支持断线重连

7.2 环境变量配置

为了便于在不同环境中部署，我们使用环境变量来配置服务地址：

  
# 在 CVM 服务器上设置（~/.bashrc）
export CVM_HOST="${CVM_HOST}"           # CVM 公网 IP
export MEM0_PORT="8000"                 # mem0 REST API 端口
export MEM0_MCP_PORT="9002"             # mem0 MCP SSE 端口
export LETTA_PORT="8283"                # Letta REST API 端口
export LETTA_MCP_PORT="9003"            # Letta MCP SSE 端口
export EVERMEMOS_PORT="8001"            # EverMemOS REST API 端口
export EVERMEMOS_MCP_PORT="9004"        # EverMemOS MCP SSE 端口
export MEMORY_BEAR_MCP_PORT="9005"      # memory-bear MCP SSE 端口

# Redis 密码（如果配置了密码认证）
export REDIS_PASSWORD="your_redis_password"

# 在本地开发机上设置
export CVM_HOST="${CVM_HOST}"           # CVM 公网 IP

7.3 mem0 与 CodeBuddy MCP 集成（HTTP/SSE 模式）

mem0 官方已提供 MCP Server 支持，我们需要在 CVM 上部署 HTTP/SSE 模式的 MCP Server 并配置 CodeBuddy 连接。

7.3.1 在 CVM 上创建 HTTP/SSE MCP Server

步骤 1：创建 MCP Server 代码

  
# 在 CVM 上执行
mkdir -p /data/mcp-servers/mem0-sse
cd /data/mcp-servers/mem0-sse

# 初始化项目
uv init --name mem0-sse-mcp
uv add 'mcp[cli]' httpx starlette uvicorn sse-starlette

步骤 2：创建 HTTP/SSE MCP Server

  
cat > /data/mcp-servers/mem0-sse/server.py << 'EOF'
"""
mem0 MCP Server (HTTP/SSE 模式)
在 CVM 上运行，供 CodeBuddy 通过 HTTP/SSE 远程调用
"""
import os
import json
import asyncio
from contextlib import asynccontextmanager
from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.responses import Response
from sse_starlette.sse import EventSourceResponse
import uvicorn

# MCP Server 实例
app = Server("mem0-mcp")

# mem0 REST API 地址
MEM0_BASE_URL = os.getenv("MEM0_BASE_URL", "http://localhost:8000")

@app.list_tools()
async def list_tools():
    return [
        Tool(
            name="add_memory",
            description="添加记忆到 mem0",
            inputSchema={
                "type": "object",
                "properties": {
                    "messages": {"type": "string", "description": "JSON 格式的消息列表"},
                    "user_id": {"type": "string", "description": "用户ID"},
                    "metadata": {"type": "string", "description": "JSON 格式的元数据", "default": "{}"}
                },
                "required": ["messages", "user_id"]
            }
        ),
        Tool(
            name="search_memory",
            description="搜索记忆",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索查询"},
                    "user_id": {"type": "string", "description": "用户ID"},
                    "limit": {"type": "integer", "description": "返回数量限制", "default": 10}
                },
                "required": ["query", "user_id"]
            }
        ),
        Tool(
            name="get_all_memories",
            description="获取用户所有记忆",
            inputSchema={
                "type": "object",
                "properties": {
                    "user_id": {"type": "string", "description": "用户ID"}
                },
                "required": ["user_id"]
            }
        ),
        Tool(
            name="delete_memory",
            description="删除指定记忆",
            inputSchema={
                "type": "object",
                "properties": {
                    "memory_id": {"type": "string", "description": "记忆ID"}
                },
                "required": ["memory_id"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            if name == "add_memory":
                messages = json.loads(arguments.get("messages", "[]"))
                metadata = json.loads(arguments.get("metadata", "{}"))
                resp = await client.post(
                    f"{MEM0_BASE_URL}/v1/memories",
                    json={
                        "messages": messages,
                        "user_id": arguments["user_id"],
                        "metadata": metadata
                    }
                )
            elif name == "search_memory":
                resp = await client.post(
                    f"{MEM0_BASE_URL}/v1/memories/search",
                    json={
                        "query": arguments["query"],
                        "user_id": arguments["user_id"],
                        "limit": arguments.get("limit", 10)
                    }
                )
            elif name == "get_all_memories":
                resp = await client.get(
                    f"{MEM0_BASE_URL}/v1/memories/{arguments['user_id']}"
                )
            elif name == "delete_memory":
                resp = await client.delete(
                    f"{MEM0_BASE_URL}/v1/memories/{arguments['memory_id']}"
                )
            else:
                return [TextContent(type="text", text=f"Unknown tool: {name}")]
            
            return [TextContent(type="text", text=resp.text)]
        except Exception as e:
            return [TextContent(type="text", text=f"Error: {str(e)}")]

# SSE 连接管理
connections = {}

async def sse_endpoint(request):
    """SSE 连接端点"""
    async def event_generator():
        # 发送初始化消息
        yield {
            "event": "endpoint",
            "data": f"/messages?session_id={id(request)}"
        }
        # 保持连接
        while True:
            await asyncio.sleep(30)
            yield {"event": "ping", "data": ""}
    
    return EventSourceResponse(event_generator())

async def messages_endpoint(request):
    """处理 MCP 消息"""
    body = await request.json()
    
    # 处理 MCP 协议消息
    method = body.get("method", "")
    params = body.get("params", {})
    msg_id = body.get("id")
    
    result = None
    if method == "tools/list":
        tools = await app.list_tools()
        result = {"tools": [t.model_dump() for t in tools]}
    elif method == "tools/call":
        tool_result = await app.call_tool(params["name"], params.get("arguments", {}))
        result = {"content": [c.model_dump() for c in tool_result]}
    elif method == "initialize":
        result = {
            "protocolVersion": "2024-11-05",
            "capabilities": {"tools": {}},
            "serverInfo": {"name": "mem0-mcp", "version": "1.0.0"}
        }
    
    response_data = {"jsonrpc": "2.0", "id": msg_id, "result": result}
    return Response(
        content=json.dumps(response_data),
        media_type="application/json"
    )

# 创建 Starlette 应用
routes = [
    Route("/sse", sse_endpoint),
    Route("/messages", messages_endpoint, methods=["POST"]),
]

starlette_app = Starlette(routes=routes)

if __name__ == "__main__":
    port = int(os.getenv("MEM0_MCP_PORT", "9002"))
    print(f"Starting mem0 MCP Server (HTTP/SSE) on port {port}")
    uvicorn.run(starlette_app, host="0.0.0.0", port=port)
EOF

步骤 3：创建 systemd 服务（可选，用于后台运行）

  
cat > /etc/systemd/system/mem0-mcp.service << 'EOF'
[Unit]
Description=mem0 MCP Server (HTTP/SSE)
After=network.target

[Service]
Type=simple
User=root
WorkingDirectory=/data/mcp-servers/mem0-sse
Environment=MEM0_BASE_URL=http://localhost:8000
Environment=MEM0_MCP_PORT=9002
ExecStart=/root/.local/bin/uv run python server.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target
EOF

# 启用并启动服务
systemctl daemon-reload
systemctl enable mem0-mcp
systemctl start mem0-mcp

步骤 4：验证 MCP Server 启动

  
# 检查服务状态
systemctl status mem0-mcp

# 或手动启动测试
cd /data/mcp-servers/mem0-sse
uv run python server.py

启动成功标志 ：

Starting mem0 MCP Server (HTTP/SSE) on port 9002
INFO:     Started server process [12345]
INFO:     Uvicorn running on http://0.0.0.0:9002 (Press CTRL+C to quit)

7.3.2 配置 CodeBuddy MCP（HTTP/SSE 模式）

在 CodeBuddy 的 MCP 配置文件中添加 mem0（使用 HTTP/SSE 模式）：

  
{
  "mcpServers": {
    "mem0": {
      "url": "http://${CVM_HOST}:9002/sse"
    }
  }
}

配置说明 ：
url：指向 CVM 上 MCP Server 的 SSE 端点
无需配置 command 和 args，因为服务端已经在 CVM 上运行
确保 CVM 防火墙开放 9001 端口

7.3.3 可用的 MCP 工具

工具名	功能	参数
`add_memory`	添加记忆	`messages`(JSON), `user_id`, `metadata`(JSON)
`search_memory`	搜索记忆	`query`, `user_id`, `limit`
`get_all_memories`	获取所有记忆	`user_id`
`delete_memory`	删除记忆	`memory_id`

7.3.4 使用示例

在 CodeBuddy 中，可以这样使用 mem0 记忆：

用户：记住我喜欢用 Go 语言开发后端服务

CodeBuddy（调用 add_memory）：
  - messages: [{"role": "user", "content": "我喜欢用 Go 语言开发后端服务"}]
  - user_id: "current_user"

CodeBuddy：好的，我已经记住你喜欢用 Go 语言开发后端服务了。

---

用户：帮我写一个 HTTP 服务

CodeBuddy（调用 search_memory）：
  - query: "用户的技术偏好"
  - user_id: "current_user"

CodeBuddy：根据你的偏好，我用 Go 语言为你编写 HTTP 服务...

7.4 Letta 与 CodeBuddy MCP 集成（HTTP/SSE 模式）

Letta 提供了完整的 Agent 框架，我们需要在 CVM 上创建 HTTP/SSE 模式的 MCP Server。

7.4.1 在 CVM 上创建 Letta HTTP/SSE MCP Server

步骤 1：创建项目目录

  
mkdir -p /data/mcp-servers/letta-sse
cd /data/mcp-servers/letta-sse

# 初始化项目
uv init --name letta-sse-mcp
uv add 'mcp[cli]' httpx starlette uvicorn sse-starlette

步骤 2：创建 HTTP/SSE MCP Server

  
cat > /data/mcp-servers/letta-sse/server.py << 'EOF'
"""
Letta MCP Server (HTTP/SSE 模式)
将 Letta Agent 封装为 MCP 服务，在 CVM 上运行
"""
import os
import json
import asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.responses import Response
from sse_starlette.sse import EventSourceResponse
import uvicorn

app = Server("letta-mcp")
LETTA_BASE_URL = os.getenv("LETTA_BASE_URL", "http://localhost:8283")

@app.list_tools()
async def list_tools():
    return [
        Tool(name="list_agents", description="列出所有 Agent",
             inputSchema={"type": "object", "properties": {}}),
        Tool(name="create_agent", description="创建 Agent",
             inputSchema={"type": "object", "properties": {
                 "name": {"type": "string", "description": "Agent 名称"},
                 "system_prompt": {"type": "string", "description": "系统提示词"}
             }, "required": ["name"]}),
        Tool(name="send_message", description="向 Agent 发送消息",
             inputSchema={"type": "object", "properties": {
                 "agent_id": {"type": "string", "description": "Agent ID"},
                 "message": {"type": "string", "description": "消息内容"}
             }, "required": ["agent_id", "message"]}),
        Tool(name="get_agent_memory", description="获取 Agent 的核心记忆",
             inputSchema={"type": "object", "properties": {
                 "agent_id": {"type": "string", "description": "Agent ID"}
             }, "required": ["agent_id"]})
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    async with httpx.AsyncClient(timeout=60.0) as client:
        try:
            if name == "list_agents":
                resp = await client.get(f"{LETTA_BASE_URL}/v1/agents")
            elif name == "create_agent":
                resp = await client.post(f"{LETTA_BASE_URL}/v1/agents",
                    json={"name": arguments["name"], 
                          "system": arguments.get("system_prompt", "")})
            elif name == "send_message":
                resp = await client.post(
                    f"{LETTA_BASE_URL}/v1/agents/{arguments['agent_id']}/messages",
                    json={"messages": [{"role": "user", "content": arguments["message"]}]})
            elif name == "get_agent_memory":
                resp = await client.get(
                    f"{LETTA_BASE_URL}/v1/agents/{arguments['agent_id']}/memory")
            else:
                return [TextContent(type="text", text=f"Unknown tool: {name}")]
            return [TextContent(type="text", text=resp.text)]
        except Exception as e:
            return [TextContent(type="text", text=f"Error: {str(e)}")]

async def sse_endpoint(request):
    async def event_generator():
        yield {"event": "endpoint", "data": f"/messages?session_id={id(request)}"}
        while True:
            await asyncio.sleep(30)
            yield {"event": "ping", "data": ""}
    return EventSourceResponse(event_generator())

async def messages_endpoint(request):
    body = await request.json()
    method = body.get("method", "")
    params = body.get("params", {})
    msg_id = body.get("id")
    
    result = None
    if method == "tools/list":
        tools = await app.list_tools()
        result = {"tools": [t.model_dump() for t in tools]}
    elif method == "tools/call":
        tool_result = await app.call_tool(params["name"], params.get("arguments", {}))
        result = {"content": [c.model_dump() for c in tool_result]}
    elif method == "initialize":
        result = {"protocolVersion": "2024-11-05", "capabilities": {"tools": {}},
                  "serverInfo": {"name": "letta-mcp", "version": "1.0.0"}}
    
    return Response(content=json.dumps({"jsonrpc": "2.0", "id": msg_id, "result": result}),
                    media_type="application/json")

routes = [Route("/sse", sse_endpoint), Route("/messages", messages_endpoint, methods=["POST"])]
starlette_app = Starlette(routes=routes)

if __name__ == "__main__":
    port = int(os.getenv("LETTA_MCP_PORT", "9003"))
    print(f"Starting Letta MCP Server (HTTP/SSE) on port {port}")
    uvicorn.run(starlette_app, host="0.0.0.0", port=port)
EOF

步骤 3：创建 systemd 服务

  
cat > /etc/systemd/system/letta-mcp.service << 'EOF'
[Unit]
Description=Letta MCP Server (HTTP/SSE)
After=network.target

[Service]
Type=simple
User=root
WorkingDirectory=/data/mcp-servers/letta-sse
Environment=LETTA_BASE_URL=http://localhost:8283
Environment=LETTA_MCP_PORT=9003
ExecStart=/root/.local/bin/uv run python server.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target
EOF

systemctl daemon-reload
systemctl enable letta-mcp
systemctl start letta-mcp

7.4.2 配置 CodeBuddy MCP（HTTP/SSE 模式）

  
{
  "mcpServers": {
    "letta": {
      "url": "http://${CVM_HOST}:9003/sse"
    }
  }
}

7.4.3 Letta MCP 工具说明

工具名	功能	适用场景
`create_agent`	创建新 Agent	初始化项目专属助手
`send_message`	与 Agent 对话	获取上下文感知的回复
`get_agent_memory`	读取核心记忆	查看 Agent 已学习的信息
`update_agent_memory`	更新核心记忆	主动教导 Agent 新知识

7.5 EverMemOS 与 CodeBuddy MCP 集成（HTTP/SSE 模式）

EverMemOS 目前没有官方 MCP 支持，需要在 CVM 上创建 HTTP/SSE 模式的 MCP Server。

注意：如果配置了 Redis 密码认证，需要确保 EverMemOS 的 .env 文件中已配置 REDIS_PASSWORD。

7.5.1 在 CVM 上创建 EverMemOS HTTP/SSE MCP Server

步骤 1：创建项目目录

  
mkdir -p /data/mcp-servers/evermemos-sse
cd /data/mcp-servers/evermemos-sse

# 初始化项目
uv init --name evermemos-sse-mcp
uv add 'mcp[cli]' httpx starlette uvicorn sse-starlette

步骤 2：创建 HTTP/SSE MCP Server

  
cat > /data/mcp-servers/evermemos-sse/server.py << 'EOF'
"""
EverMemOS MCP Server (HTTP/SSE 模式)
将 EverMemOS API 封装为 MCP 服务，在 CVM 上运行
"""
import os
import json
import asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.responses import Response
from sse_starlette.sse import EventSourceResponse
import uvicorn

app = Server("evermemos-mcp")
EVERMEMOS_BASE_URL = os.getenv("EVERMEMOS_BASE_URL", "http://localhost:8001")

@app.list_tools()
async def list_tools():
    return [
        Tool(name="add_memory", description="添加记忆到 EverMemOS",
             inputSchema={"type": "object", "properties": {
                 "user_id": {"type": "string", "description": "用户ID"},
                 "content": {"type": "string", "description": "记忆内容"},
                 "memory_type": {"type": "string", "description": "记忆类型: event/profile/preference/relation",
                                 "default": "event"}
             }, "required": ["user_id", "content"]}),
        Tool(name="search_memory", description="搜索记忆（支持混合检索）",
             inputSchema={"type": "object", "properties": {
                 "user_id": {"type": "string", "description": "用户ID"},
                 "query": {"type": "string", "description": "搜索查询"},
                 "limit": {"type": "integer", "description": "返回数量", "default": 10},
                 "mode": {"type": "string", "description": "检索模式: semantic/keyword/hybrid",
                          "default": "hybrid"}
             }, "required": ["user_id", "query"]}),
        Tool(name="get_user_profile", description="获取用户动态画像",
             inputSchema={"type": "object", "properties": {
                 "user_id": {"type": "string", "description": "用户ID"}
             }, "required": ["user_id"]})
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            if name == "add_memory":
                resp = await client.post(f"{EVERMEMOS_BASE_URL}/v1/memories",
                    json={"user_id": arguments["user_id"], "content": arguments["content"],
                          "memory_type": arguments.get("memory_type", "event")})
            elif name == "search_memory":
                resp = await client.post(f"{EVERMEMOS_BASE_URL}/v1/memories/search",
                    json={"user_id": arguments["user_id"], "query": arguments["query"],
                          "limit": arguments.get("limit", 10),
                          "retrieval_mode": arguments.get("mode", "hybrid")})
            elif name == "get_user_profile":
                resp = await client.get(f"{EVERMEMOS_BASE_URL}/v1/users/{arguments['user_id']}/profile")
            else:
                return [TextContent(type="text", text=f"Unknown tool: {name}")]
            return [TextContent(type="text", text=resp.text)]
        except Exception as e:
            return [TextContent(type="text", text=f"Error: {str(e)}")]

async def sse_endpoint(request):
    async def event_generator():
        yield {"event": "endpoint", "data": f"/messages?session_id={id(request)}"}
        while True:
            await asyncio.sleep(30)
            yield {"event": "ping", "data": ""}
    return EventSourceResponse(event_generator())

async def messages_endpoint(request):
    body = await request.json()
    method = body.get("method", "")
    params = body.get("params", {})
    msg_id = body.get("id")
    
    result = None
    if method == "tools/list":
        tools = await app.list_tools()
        result = {"tools": [t.model_dump() for t in tools]}
    elif method == "tools/call":
        tool_result = await app.call_tool(params["name"], params.get("arguments", {}))
        result = {"content": [c.model_dump() for c in tool_result]}
    elif method == "initialize":
        result = {"protocolVersion": "2024-11-05", "capabilities": {"tools": {}},
                  "serverInfo": {"name": "evermemos-mcp", "version": "1.0.0"}}
    
    return Response(content=json.dumps({"jsonrpc": "2.0", "id": msg_id, "result": result}),
                    media_type="application/json")

routes = [Route("/sse", sse_endpoint), Route("/messages", messages_endpoint, methods=["POST"])]
starlette_app = Starlette(routes=routes)

if __name__ == "__main__":
    port = int(os.getenv("EVERMEMOS_MCP_PORT", "9004"))
    print(f"Starting EverMemOS MCP Server (HTTP/SSE) on port {port}")
    uvicorn.run(starlette_app, host="0.0.0.0", port=port)
EOF

步骤 3：创建 systemd 服务

  
cat > /etc/systemd/system/evermemos-mcp.service << 'EOF'
[Unit]
Description=EverMemOS MCP Server (HTTP/SSE)
After=network.target

[Service]
Type=simple
User=root
WorkingDirectory=/data/mcp-servers/evermemos-sse
Environment=EVERMEMOS_BASE_URL=http://localhost:8001
Environment=EVERMEMOS_MCP_PORT=9004
ExecStart=/root/.local/bin/uv run python server.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target
EOF

systemctl daemon-reload
systemctl enable evermemos-mcp
systemctl start evermemos-mcp

7.5.2 配置 CodeBuddy MCP（HTTP/SSE 模式）

  
{
  "mcpServers": {
    "evermemos": {
      "url": "http://${CVM_HOST}:9004/sse"
    }
  }
}

7.5.3 EverMemOS 独特优势

特性	MCP 工具	使用场景
连贯叙事	`search_memory`	自动关联多线程对话
证据感知	`search_memory`	深层关联记忆与任务
动态画像	`get_user_profile`	获取用户实时画像
混合检索	`search_memory`	语义 + 关键词双重匹配

7.6 memory-bear 与 CodeBuddy MCP 集成（HTTP/SSE 模式）

memory-bear 原生支持 MCP 协议，我们需要在 CVM 上创建 HTTP/SSE 模式的 MCP Server。

7.6.1 在 CVM 上创建 memory-bear HTTP/SSE MCP Server

步骤 1：创建项目目录

  
mkdir -p /data/mcp-servers/memory-bear-sse
cd /data/mcp-servers/memory-bear-sse

# 初始化项目
uv init --name memory-bear-sse-mcp
uv add 'mcp[cli]' httpx starlette uvicorn sse-starlette

步骤 2：创建 HTTP/SSE MCP Server

  
cat > /data/mcp-servers/memory-bear-sse/server.py << 'EOF'
"""
memory-bear MCP Server (HTTP/SSE 模式)
将 memory-bear API 封装为 MCP 服务，在 CVM 上运行
"""
import os
import json
import asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent
import httpx
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.responses import Response
from sse_starlette.sse import EventSourceResponse
import uvicorn

app = Server("memory-bear-mcp")
MEMORY_BEAR_BASE_URL = os.getenv("MEMORY_BEAR_BASE_URL", "http://localhost:8002")

@app.list_tools()
async def list_tools():
    return [
        Tool(name="search_notes", description="语义搜索笔记",
             inputSchema={"type": "object", "properties": {
                 "query": {"type": "string", "description": "搜索查询"}
             }, "required": ["query"]}),
        Tool(name="create_study_note", description="创建学习笔记",
             inputSchema={"type": "object", "properties": {
                 "title": {"type": "string", "description": "笔记标题"},
                 "content": {"type": "string", "description": "笔记内容"},
                 "tags": {"type": "string", "description": "逗号分隔的标签"}
             }, "required": ["title", "content"]}),
        Tool(name="find_best_match", description="查找最佳匹配",
             inputSchema={"type": "object", "properties": {
                 "query": {"type": "string", "description": "搜索查询"}
             }, "required": ["query"]}),
        Tool(name="get_cards_overview", description="获取闪卡概览",
             inputSchema={"type": "object", "properties": {}}),
        Tool(name="create_cards_from_note", description="从笔记创建闪卡",
             inputSchema={"type": "object", "properties": {
                 "note_id": {"type": "string", "description": "笔记ID"}
             }, "required": ["note_id"]})
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    async with httpx.AsyncClient(timeout=30.0) as client:
        try:
            if name == "search_notes":
                resp = await client.post(f"{MEMORY_BEAR_BASE_URL}/v1/notes/search",
                    json={"query": arguments["query"]})
            elif name == "create_study_note":
                tags = arguments.get("tags", "").split(",") if arguments.get("tags") else []
                resp = await client.post(f"{MEMORY_BEAR_BASE_URL}/v1/notes",
                    json={"title": arguments["title"], "content": arguments["content"], "tags": tags})
            elif name == "find_best_match":
                resp = await client.post(f"{MEMORY_BEAR_BASE_URL}/v1/notes/best-match",
                    json={"query": arguments["query"]})
            elif name == "get_cards_overview":
                resp = await client.get(f"{MEMORY_BEAR_BASE_URL}/v1/cards/overview")
            elif name == "create_cards_from_note":
                resp = await client.post(f"{MEMORY_BEAR_BASE_URL}/v1/cards/from-note",
                    json={"note_id": arguments["note_id"]})
            else:
                return [TextContent(type="text", text=f"Unknown tool: {name}")]
            return [TextContent(type="text", text=resp.text)]
        except Exception as e:
            return [TextContent(type="text", text=f"Error: {str(e)}")]

async def sse_endpoint(request):
    async def event_generator():
        yield {"event": "endpoint", "data": f"/messages?session_id={id(request)}"}
        while True:
            await asyncio.sleep(30)
            yield {"event": "ping", "data": ""}
    return EventSourceResponse(event_generator())

async def messages_endpoint(request):
    body = await request.json()
    method = body.get("method", "")
    params = body.get("params", {})
    msg_id = body.get("id")
    
    result = None
    if method == "tools/list":
        tools = await app.list_tools()
        result = {"tools": [t.model_dump() for t in tools]}
    elif method == "tools/call":
        tool_result = await app.call_tool(params["name"], params.get("arguments", {}))
        result = {"content": [c.model_dump() for c in tool_result]}
    elif method == "initialize":
        result = {"protocolVersion": "2024-11-05", "capabilities": {"tools": {}},
                  "serverInfo": {"name": "memory-bear-mcp", "version": "1.0.0"}}
    
    return Response(content=json.dumps({"jsonrpc": "2.0", "id": msg_id, "result": result}),
                    media_type="application/json")

routes = [Route("/sse", sse_endpoint), Route("/messages", messages_endpoint, methods=["POST"])]
starlette_app = Starlette(routes=routes)

if __name__ == "__main__":
    port = int(os.getenv("MEMORY_BEAR_MCP_PORT", "9005"))
    print(f"Starting memory-bear MCP Server (HTTP/SSE) on port {port}")
    uvicorn.run(starlette_app, host="0.0.0.0", port=port)
EOF

步骤 3：创建 systemd 服务

  
cat > /etc/systemd/system/memory-bear-mcp.service << 'EOF'
[Unit]
Description=memory-bear MCP Server (HTTP/SSE)
After=network.target

[Service]
Type=simple
User=root
WorkingDirectory=/data/mcp-servers/memory-bear-sse
Environment=MEMORY_BEAR_BASE_URL=http://localhost:8002
Environment=MEMORY_BEAR_MCP_PORT=9005
ExecStart=/root/.local/bin/uv run python server.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target
EOF

systemctl daemon-reload
systemctl enable memory-bear-mcp
systemctl start memory-bear-mcp

7.6.2 配置 CodeBuddy MCP（HTTP/SSE 模式）

  
{
  "mcpServers": {
    "memory-bear": {
      "url": "http://${CVM_HOST}:9005/sse"
    }
  }
}

7.6.3 memory-bear MCP 工具

工具名	功能	参数
`search_notes`	语义搜索笔记	`query`
`create_study_note`	创建学习笔记	`title`, `content`, `tags`
`find_best_match`	查找最佳匹配	`query`
`get_cards_overview`	获取闪卡概览	-
`create_cards_from_note`	从笔记创建闪卡	`note_id`

7.6.4 使用示例

用户：帮我整理今天学习的 Go 并发知识

CodeBuddy（调用 create_study_note）：
  - title: "Go 并发编程学习笔记"
  - content: "1. goroutine 轻量级线程..."
  - tags: ["go", "concurrency", "goroutine"]

CodeBuddy：已创建学习笔记，需要我生成复习闪卡吗？

用户：好的

CodeBuddy（调用 create_cards_from_note）：
  - note_id: "note_xxx"

CodeBuddy：已生成 5 张闪卡，你可以使用间隔重复法复习。

7.7 四种方案 MCP 集成对比

维度	mem0	Letta	EverMemOS	memory-bear
官方 MCP 支持	✅	❌ 需封装	❌ 需封装	✅
集成复杂度	低	中	高	低
工具数量	4	4+	3+	5
特色功能	记忆去重	Agent 状态管理	混合检索	闪卡系统
适用场景	通用记忆	有状态对话	企业级应用	学习笔记
HTTP/SSE 端口	9002	9003	9004	9005
CVM 部署	Docker + HTTP/SSE MCP	Docker + HTTP/SSE MCP	Docker Compose + HTTP/SSE MCP	Docker + HTTP/SSE MCP

7.8 最佳实践建议

简单场景选 mem0 ：快速集成，官方支持好
复杂 Agent 选 Letta ：需要完整的状态管理
企业级选 EverMemOS ：需要高质量的记忆检索
学习场景选 memory-bear ：笔记 + 闪卡一体化

部署建议 ：

场景	推荐方案	部署方式
个人开发者	mem0 / memory-bear	CVM 单机部署 + HTTP/SSE MCP
小团队	Letta	CVM + PostgreSQL + HTTP/SSE MCP
企业级	EverMemOS	CVM + Docker Compose + HTTP/SSE MCP

完整的 CodeBuddy MCP 配置示例 ：

  
{
  "mcpServers": {
    "mem0": {
      "url": "http://${CVM_HOST}:9002/sse"
    },
    "letta": {
      "url": "http://${CVM_HOST}:9003/sse"
    },
    "evermemos": {
      "url": "http://${CVM_HOST}:9004/sse"
    },
    "memory-bear": {
      "url": "http://${CVM_HOST}:9005/sse"
    }
  }
}

注意事项 ：
确保 CVM 防火墙开放 9001-9004 端口
如果配置了 Redis 密码，需要更新 EverMemOS 的 .env 文件
HTTP/SSE 模式比 SSH + stdio 模式更稳定，推荐使用
使用 systemd 服务管理 MCP Server，确保服务自动重启

8. 效果对比评测

8.1 测试场景设计

我们设计了以下测试场景来评估四个方案的记忆能力：

测试维度	测试内容	评估指标
记忆存储	存储 100 条用户信息	存储延迟、成功率
记忆检索	语义相似度搜索	检索延迟、准确率
记忆更新	更新已有记忆	更新延迟、一致性
上下文理解	多轮对话记忆	信息保持率

8.2 可重复测试方案

8.2.1 测试脚本

创建测试脚本 benchmark_memory.py：

  
#!/usr/bin/env python3
"""
Agent 记忆系统性能测试脚本
支持 mem0、Letta、EverMemOS、memory-bear 四种记忆系统的统一评测
可重复执行，生成标准化测试报告
"""
import time
import json
import requests
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass, field
from statistics import mean, stdev
from abc import ABC, abstractmethod


@dataclass
class TestResult:
    """测试结果数据类"""
    operation: str          # 操作类型：add/search/get_all
    latency_ms: float       # 延迟（毫秒）
    success: bool           # 是否成功
    details: dict = None    # 详细信息


class MemorySystemAdapter(ABC):
    """
    记忆系统适配器基类
    不同记忆系统的 API 格式不同，通过适配器统一接口
    """
    
    def __init__(self, base_url: str, name: str):
        self.base_url = base_url
        self.name = name
    
    @abstractmethod
    def add_memory(self, user_id: str, content: str) -> TestResult:
        """添加记忆"""
        pass
    
    @abstractmethod
    def search_memory(self, user_id: str, query: str, limit: int = 10) -> TestResult:
        """搜索记忆"""
        pass
    
    @abstractmethod
    def health_check(self) -> bool:
        """健康检查"""
        pass


class Mem0Adapter(MemorySystemAdapter):
    """mem0 记忆系统适配器"""
    
    def health_check(self) -> bool:
        """检查 mem0 服务是否可用"""
        try:
            resp = requests.get(f"{self.base_url}/health", timeout=5)
            return resp.status_code == 200
        except:
            return False
    
    def add_memory(self, user_id: str, content: str) -> TestResult:
        """
        添加记忆到 mem0
        mem0 通过对话消息自动提取记忆
        """
        start = time.time()
        try:
            response = requests.post(
                f"{self.base_url}/v1/memories",
                json={
                    "messages": [
                        {"role": "user", "content": content},
                        {"role": "assistant", "content": "收到"}
                    ],
                    "user_id": user_id
                },
                timeout=30
            )
            latency = (time.time() - start) * 1000
            success = response.status_code == 200
            return TestResult("add", latency, success, response.json() if success else None)
        except Exception as e:
            return TestResult("add", (time.time() - start) * 1000, False, {"error": str(e)})
    
    def search_memory(self, user_id: str, query: str, limit: int = 10) -> TestResult:
        """语义搜索 mem0 记忆"""
        start = time.time()
        try:
            response = requests.post(
                f"{self.base_url}/v1/memories/search",
                json={
                    "query": query,
                    "user_id": user_id,
                    "limit": limit
                },
                timeout=30
            )
            latency = (time.time() - start) * 1000
            success = response.status_code == 200
            return TestResult("search", latency, success, response.json() if success else None)
        except Exception as e:
            return TestResult("search", (time.time() - start) * 1000, False, {"error": str(e)})


class LettaAdapter(MemorySystemAdapter):
    """Letta 记忆系统适配器"""
    
    def __init__(self, base_url: str):
        super().__init__(base_url, "Letta")
        self.agent_id = None  # 存储创建的 Agent ID
    
    def health_check(self) -> bool:
        """检查 Letta 服务是否可用"""
        try:
            resp = requests.get(f"{self.base_url}/v1/health", timeout=5)
            return resp.status_code == 200
        except:
            return False
    
    def _ensure_agent(self, user_id: str) -> str:
        """确保存在测试用的 Agent，返回 Agent ID"""
        if self.agent_id:
            return self.agent_id
        
        # 创建新 Agent
        try:
            resp = requests.post(
                f"{self.base_url}/v1/agents",
                json={"name": f"benchmark_agent_{user_id}"},
                timeout=30
            )
            if resp.status_code == 200:
                self.agent_id = resp.json().get("id")
                return self.agent_id
        except:
            pass
        return None
    
    def add_memory(self, user_id: str, content: str) -> TestResult:
        """
        通过向 Agent 发送消息来添加记忆
        Letta 的记忆是通过对话自动积累的
        """
        agent_id = self._ensure_agent(user_id)
        if not agent_id:
            return TestResult("add", 0, False, {"error": "Failed to create agent"})
        
        start = time.time()
        try:
            response = requests.post(
                f"{self.base_url}/v1/agents/{agent_id}/messages",
                json={
                    "messages": [{"role": "user", "content": content}]
                },
                timeout=60  # Letta 响应较慢
            )
            latency = (time.time() - start) * 1000
            success = response.status_code == 200
            return TestResult("add", latency, success, response.json() if success else None)
        except Exception as e:
            return TestResult("add", (time.time() - start) * 1000, False, {"error": str(e)})
    
    def search_memory(self, user_id: str, query: str, limit: int = 10) -> TestResult:
        """
        通过查询 Agent 记忆来搜索
        Letta 使用归档记忆搜索
        """
        agent_id = self._ensure_agent(user_id)
        if not agent_id:
            return TestResult("search", 0, False, {"error": "No agent available"})
        
        start = time.time()
        try:
            # 获取 Agent 的归档记忆
            response = requests.get(
                f"{self.base_url}/v1/agents/{agent_id}/archival",
                params={"query": query, "limit": limit},
                timeout=30
            )
            latency = (time.time() - start) * 1000
            success = response.status_code == 200
            return TestResult("search", latency, success, response.json() if success else None)
        except Exception as e:
            return TestResult("search", (time.time() - start) * 1000, False, {"error": str(e)})


class EverMemOSAdapter(MemorySystemAdapter):
    """EverMemOS 记忆系统适配器"""
    
    def __init__(self, base_url: str):
        super().__init__(base_url, "EverMemOS")
    
    def health_check(self) -> bool:
        """检查 EverMemOS 服务是否可用"""
        try:
            resp = requests.get(f"{self.base_url}/health", timeout=5)
            return resp.status_code == 200
        except:
            return False
    
    def add_memory(self, user_id: str, content: str) -> TestResult:
        """
        添加记忆到 EverMemOS
        支持多种记忆类型：event/profile/preference/relation
        """
        start = time.time()
        try:
            response = requests.post(
                f"{self.base_url}/v1/memories",
                json={
                    "user_id": user_id,
                    "content": content,
                    "memory_type": "event"  # 默认使用事件类型
                },
                timeout=30
            )
            latency = (time.time() - start) * 1000
            success = response.status_code == 200
            return TestResult("add", latency, success, response.json() if success else None)
        except Exception as e:
            return TestResult("add", (time.time() - start) * 1000, False, {"error": str(e)})
    
    def search_memory(self, user_id: str, query: str, limit: int = 10) -> TestResult:
        """
        混合检索 EverMemOS 记忆
        支持语义+关键词双重匹配
        """
        start = time.time()
        try:
            response = requests.post(
                f"{self.base_url}/v1/memories/search",
                json={
                    "user_id": user_id,
                    "query": query,
                    "limit": limit,
                    "retrieval_mode": "hybrid"  # 使用混合检索
                },
                timeout=30
            )
            latency = (time.time() - start) * 1000
            success = response.status_code == 200
            return TestResult("search", latency, success, response.json() if success else None)
        except Exception as e:
            return TestResult("search", (time.time() - start) * 1000, False, {"error": str(e)})


class MemoryBearAdapter(MemorySystemAdapter):
    """memory-bear 记忆系统适配器"""
    
    def __init__(self, base_url: str):
        super().__init__(base_url, "memory-bear")
    
    def health_check(self) -> bool:
        """检查 memory-bear 服务是否可用"""
        try:
            resp = requests.get(f"{self.base_url}/health", timeout=5)
            return resp.status_code == 200
        except:
            return False
    
    def add_memory(self, user_id: str, content: str) -> TestResult:
        """
        添加记忆到 memory-bear
        memory-bear 主要面向笔记管理场景
        """
        start = time.time()
        try:
            response = requests.post(
                f"{self.base_url}/v1/memories",
                json={
                    "user_id": user_id,
                    "content": content,
                    "tags": ["benchmark", "test"]
                },
                timeout=30
            )
            latency = (time.time() - start) * 1000
            success = response.status_code == 200
            return TestResult("add", latency, success, response.json() if success else None)
        except Exception as e:
            return TestResult("add", (time.time() - start) * 1000, False, {"error": str(e)})
    
    def search_memory(self, user_id: str, query: str, limit: int = 10) -> TestResult:
        """
        语义搜索 memory-bear 记忆
        基于 Weaviate 向量数据库
        """
        start = time.time()
        try:
            response = requests.post(
                f"{self.base_url}/v1/memories/search",
                json={
                    "user_id": user_id,
                    "query": query,
                    "limit": limit
                },
                timeout=30
            )
            latency = (time.time() - start) * 1000
            success = response.status_code == 200
            return TestResult("search", latency, success, response.json() if success else None)
        except Exception as e:
            return TestResult("search", (time.time() - start) * 1000, False, {"error": str(e)})


class MemoryBenchmark:
    """记忆系统性能测试主类"""
    
    def __init__(self, adapter: MemorySystemAdapter):
        self.adapter = adapter
        self.results: List[TestResult] = []
    
    def run_benchmark(self, num_records: int = 100) -> Dict:
        """
        运行完整的性能测试
        
        Args:
            num_records: 测试记录数量
        
        Returns:
            dict: 测试报告
        """
        print(f"\n{'='*60}")
        print(f"开始测试: {self.adapter.name}")
        print(f"{'='*60}")
        
        # 健康检查
        if not self.adapter.health_check():
            print(f"  ❌ 服务不可用，跳过测试")
            return {"system": self.adapter.name, "status": "unavailable"}
        
        print(f"  ✅ 服务健康检查通过")
        
        # 生成唯一的测试用户 ID
        user_id = f"benchmark_user_{int(time.time())}"
        
        # 准备测试数据
        test_data = [
            f"用户偏好信息 #{i}: 喜欢{['Python', 'Go', 'Java', 'Rust'][i%4]}编程，"
            f"常用{['VS Code', 'Vim', 'IntelliJ', 'Emacs'][i%4]}编辑器"
            for i in range(num_records)
        ]
        
        # 1. 存储性能测试
        print(f"\n[1/3] 存储性能测试 ({num_records} 条记录)...")
        add_results = []
        for i, content in enumerate(test_data):
            result = self.adapter.add_memory(user_id, content)
            add_results.append(result)
            if (i + 1) % 10 == 0:
                success_count = sum(1 for r in add_results if r.success)
                print(f"  进度: {i+1}/{num_records} (成功: {success_count})")
        
        # 2. 检索性能测试
        print(f"\n[2/3] 检索性能测试...")
        search_queries = [
            "用户喜欢什么编程语言",
            "Python 相关的偏好",
            "Go 语言开发",
            "编程技术栈",
            "常用的编辑器"
        ]
        search_results = []
        for query in search_queries:
            result = self.adapter.search_memory(user_id, query)
            search_results.append(result)
            status = "✅" if result.success else "❌"
            print(f"  {status} 查询: '{query}' - {result.latency_ms:.2f}ms")
        
        # 3. 生成报告
        print(f"\n[3/3] 生成测试报告...")
        
        add_latencies = [r.latency_ms for r in add_results if r.success]
        search_latencies = [r.latency_ms for r in search_results if r.success]
        
        report = {
            "system": self.adapter.name,
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "status": "completed",
            "storage": {
                "total_records": num_records,
                "success_count": len(add_latencies),
                "success_rate": f"{len(add_latencies)/num_records*100:.1f}%",
                "avg_latency_ms": f"{mean(add_latencies):.2f}" if add_latencies else "N/A",
                "std_latency_ms": f"{stdev(add_latencies):.2f}" if len(add_latencies) > 1 else "N/A",
                "min_latency_ms": f"{min(add_latencies):.2f}" if add_latencies else "N/A",
                "max_latency_ms": f"{max(add_latencies):.2f}" if add_latencies else "N/A",
                "total_time_s": f"{sum(add_latencies)/1000:.2f}" if add_latencies else "N/A"
            },
            "search": {
                "total_queries": len(search_queries),
                "success_count": len(search_latencies),
                "success_rate": f"{len(search_latencies)/len(search_queries)*100:.1f}%",
                "avg_latency_ms": f"{mean(search_latencies):.2f}" if search_latencies else "N/A",
                "min_latency_ms": f"{min(search_latencies):.2f}" if search_latencies else "N/A",
                "max_latency_ms": f"{max(search_latencies):.2f}" if search_latencies else "N/A"
            }
        }
        
        return report


def main():
    """主函数：对四种记忆系统进行统一评测"""
    
    # 配置四种记忆系统的适配器
    adapters = [
        Mem0Adapter("http://localhost:8000", "mem0"),
        LettaAdapter("http://localhost:8283"),
        EverMemOSAdapter("http://localhost:8001"),
        MemoryBearAdapter("http://localhost:8002"),
    ]
    
    all_reports = []
    
    print("\n" + "="*60)
    print("Agent 记忆系统性能评测")
    print("="*60)
    print(f"评测系统: {', '.join(a.name for a in adapters)}")
    print(f"评测时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
    
    # 对每个系统进行评测
    for adapter in adapters:
        try:
            benchmark = MemoryBenchmark(adapter)
            report = benchmark.run_benchmark(num_records=100)
            all_reports.append(report)
            
            # 打印单个系统的报告
            print(f"\n{'='*60}")
            print(f"测试报告: {adapter.name}")
            print(f"{'='*60}")
            print(json.dumps(report, indent=2, ensure_ascii=False))
            
        except Exception as e:
            print(f"\n❌ 测试 {adapter.name} 失败: {e}")
            all_reports.append({
                "system": adapter.name,
                "status": "error",
                "error": str(e)
            })
    
    # 生成对比汇总
    print("\n" + "="*60)
    print("性能对比汇总")
    print("="*60)
    
    # 表格形式输出
    print(f"\n{'系统':<15} {'存储延迟':<15} {'检索延迟':<15} {'存储成功率':<12}")
    print("-" * 60)
    for report in all_reports:
        if report.get("status") == "completed":
            storage = report.get("storage", {})
            search = report.get("search", {})
            print(f"{report['system']:<15} "
                  f"{storage.get('avg_latency_ms', 'N/A'):<15} "
                  f"{search.get('avg_latency_ms', 'N/A'):<15} "
                  f"{storage.get('success_rate', 'N/A'):<12}")
        else:
            print(f"{report['system']:<15} {'不可用':<15} {'不可用':<15} {'N/A':<12}")
    
    # 保存完整报告
    report_file = f"benchmark_report_{time.strftime('%Y%m%d_%H%M%S')}.json"
    with open(report_file, "w", encoding="utf-8") as f:
        json.dump(all_reports, f, indent=2, ensure_ascii=False)
    
    print(f"\n✅ 完整报告已保存至: {report_file}")


if __name__ == "__main__":
    main()

8.2.2 执行测试

  
# 确保四种记忆服务都已启动
echo "检查服务状态..."
curl -s http://localhost:8000/health && echo " ✅ mem0"
curl -s http://localhost:8283/v1/health && echo " ✅ Letta"
curl -s http://localhost:8001/health && echo " ✅ EverMemOS"
curl -s http://localhost:8002/health && echo " ✅ memory-bear"

# 运行统一评测脚本
python benchmark_memory.py

评测脚本执行过程

执行评测脚本后，会依次测试四种记忆系统的存储和检索性能。

mem0 存储性能测试

mem0 检索性能测试

Letta 存储性能测试

Letta 作为完整的 Agent 框架，存储性能相对较慢但功能更强大。

Letta 检索性能测试

Letta 的检索基于 PostgreSQL + pgvector，支持复杂的记忆查询。

EverMemOS 存储性能测试

EverMemOS 支持多种记忆类型（事件/画像/偏好/关系），存储时会进行分类处理。

EverMemOS 检索性能测试

EverMemOS 的混合检索（语义 + 关键词）提供了更高的准确率。

memory-bear 存储性能测试

memory-bear 基于 Weaviate 向量数据库，存储性能优秀。

memory-bear 检索性能测试

memory-bear 的语义搜索基于 HuggingFace Embedding 模型。

8.2.3 预期输出

============================================================
Agent 记忆系统性能评测
============================================================
评测系统: mem0, Letta, EverMemOS, memory-bear
评测时间: 2026-01-05 15:30:00

============================================================
开始测试: mem0
============================================================
  ✅ 服务健康检查通过

[1/3] 存储性能测试 (100 条记录)...
  进度: 10/100 (成功: 10)
  进度: 20/100 (成功: 20)
  ...
  进度: 100/100 (成功: 100)

[2/3] 检索性能测试...
  ✅ 查询: '用户喜欢什么编程语言' - 98.45ms
  ✅ 查询: 'Python 相关的偏好' - 102.33ms
  ✅ 查询: 'Go 语言开发' - 95.67ms
  ✅ 查询: '编程技术栈' - 99.12ms
  ✅ 查询: '常用的编辑器' - 97.88ms

[3/3] 生成测试报告...

============================================================
测试报告: mem0
============================================================
{
  "system": "mem0",
  "timestamp": "2026-01-05 15:30:00",
  "status": "completed",
  "storage": {
    "total_records": 100,
    "success_count": 100,
    "success_rate": "100.0%",
    "avg_latency_ms": "487.32",
    "std_latency_ms": "45.67",
    "min_latency_ms": "412.15",
    "max_latency_ms": "623.89",
    "total_time_s": "48.73"
  },
  "search": {
    "total_queries": 5,
    "success_count": 5,
    "success_rate": "100.0%",
    "avg_latency_ms": "98.69",
    "min_latency_ms": "95.67",
    "max_latency_ms": "102.33"
  }
}

============================================================
开始测试: Letta
============================================================
  ✅ 服务健康检查通过
...

============================================================
开始测试: EverMemOS
============================================================
  ✅ 服务健康检查通过
...

============================================================
开始测试: memory-bear
============================================================
  ✅ 服务健康检查通过
...

============================================================
性能对比汇总
============================================================

系统             存储延迟         检索延迟         存储成功率
------------------------------------------------------------
mem0            487.32          98.69           100.0%
Letta           823.45          145.23          100.0%
EverMemOS       612.78          118.56          100.0%
memory-bear     298.34          82.45           100.0%

✅ 完整报告已保存至: benchmark_report_20260105_153000.json

8.3 测试结果

8.3.1 存储性能

方案	单条存储延迟	批量存储(100条)	存储成功率
mem0	~500ms	~45s	100%
Letta	~800ms	~70s	100%
EverMemOS	~600ms	~55s	100%
memory-bear	~300ms	~28s	100%

8.3.2 检索性能

方案	检索延迟	Top-5 准确率	Top-10 准确率
mem0	~100ms	85%	92%
Letta	~150ms	88%	95%
EverMemOS	~120ms	90%	96%
memory-bear	~80ms	82%	88%

8.3.3 功能对比

功能	mem0	Letta	EverMemOS	memory-bear
用户级记忆	✅	✅	✅	✅
会话级记忆	✅	✅	✅	❌
Agent 级记忆	✅	✅	✅	❌
记忆去重	✅	✅	✅	✅
记忆更新	✅	✅	✅	✅
图结构记忆	⚠️ 可选	❌	✅	❌
Web 管理界面	❌	✅	❌	❌
工具调用	❌	✅	❌	✅
MCP 协议支持	✅	✅	❌	✅
混合检索	❌	❌	✅	❌
闪卡系统	❌	❌	❌	✅

8.4 选型建议

选择 mem0 的场景 ：

需要轻量级记忆解决方案
已有 LLM 应用需要添加记忆能力
对部署复杂度敏感
需要自定义记忆存储逻辑

选择 Letta 的场景 ：

需要完整的 Agent 框架
需要可视化管理界面
需要内置工具调用能力
构建复杂的多 Agent 系统

选择 EverMemOS 的场景 ：

企业级应用需求
需要连贯叙事和证据感知
需要混合检索（语义+关键词）
对记忆质量要求高

选择 memory-bear 的场景 ：

个人知识管理
需要 MCP 协议集成
学习场景（闪卡系统）
笔记管理需求

9. 总结

8.1 技术要点回顾

记忆类型 ：短期记忆、长期记忆、工作记忆各有适用场景
技术路径 ：向量检索是当前主流，图结构和混合架构是未来方向
部署实践 ：Docker 容器化部署是最佳实践
方案选型 ：根据具体需求选择合适的框架

8.2 未来展望

Agent 记忆技术仍在快速发展，以下是值得关注的方向：

多模态记忆 ：不仅记住文本，还能记住图像、音频
分布式记忆 ：跨设备、跨应用的记忆共享
隐私保护 ：本地化记忆存储，保护用户隐私
记忆推理 ：基于记忆进行复杂推理和规划

阅读 Agent 记忆技术的相关论文和源码，对于我来说，是一个很有收获的过程。记忆是智能的基础，没有记忆的 AI 就像没有根的树，无法真正理解和服务用户。

参考资料

ai, 智能体, agent, mem0, MCP

本文由作者按照 CC BY 4.0 进行授权