构建AI Agent私有记忆中枢:基于TiDB与向量数据库的本地化实践
1. 项目缘起:当AI Agent需要记住“我是谁”
最近在折腾一个叫OpenClaw的AI Agent项目,它本质上是一个可以自主执行任务、调用工具、处理信息的智能体框架。玩过一阵子后,我发现了一个挺要命的问题: 这Agent记性太差了 。每次对话重启,它就像得了健忘症,完全不记得之前聊过什么、做过什么决策、执行过哪些步骤。这导致它无法进行连续、复杂的多轮任务,更别提形成个性化的“记忆”和“经验”了。
这让我意识到,一个真正有用的AI Agent,除了强大的推理和工具调用能力,还必须有一个可靠的“记忆中枢”。这个中枢要能持久化存储它的对话历史、任务上下文、学到的知识、用户偏好,甚至是它自己总结出的“经验教训”。更重要的是,出于对数据隐私、合规性以及定制化需求的考虑,这个记忆中枢 绝对不能上云 。我不想把我的Agent思考过程、用户交互数据、乃至可能涉及的业务逻辑,都托付给某个远方的服务器。
于是,我开始寻找解决方案。目标很明确:一个能本地或私有化部署、高性能、可扩展的存储后端。最终,我锁定了两个组件: mem9 和 TiDB 。前者是一个为AI应用设计的向量数据库,后者则是一个成熟的分布式关系型数据库。将它们与OpenClaw结合,我构建了一个名为“私有记忆中枢”的架构。今天,我就来详细拆解这个项目的设计思路、技术选型、部署踩坑和实战心得。
2. 技术选型:为什么是 mem9 + TiDB?
在决定自己动手之前,我调研了市面上几种常见的方案。最简单的是直接用OpenClaw默认的、基于内存或本地文件的存储,但这显然无法满足持久化和多实例共享的需求。也考虑过直接用PostgreSQL的 pgvector 插件,或者专门的向量数据库如Milvus、Qdrant。但经过一番权衡,我选择了mem9和TiDB的组合,原因如下。
2.1 mem9:专为AI记忆场景优化的向量数据库
mem9并不是最知名的向量数据库,但它有几个特性非常契合“AI记忆”这个场景。
首先,它的设计哲学是“轻量”和“易嵌入”。它不像Milvus那样是一个庞大的分布式系统,更像是一个可以轻松集成到应用中的库或服务。这对于OpenClaw这种需要灵活部署的Agent框架来说,减少了运维复杂度。
其次,mem9在存储和检索“会话”和“记忆片段”这类数据上做了优化。它原生支持将一段文本(比如一次对话轮次或一个任务步骤)与其向量嵌入(embedding)关联存储,并可以方便地基于语义相似度进行检索。这正是我们构建Agent记忆的核心操作:把Agent的每一次“思考”和“行动”作为一段记忆存起来,后续需要时能快速、准确地回想起来。
最后,mem9的API相对简洁,与Python生态集成良好,这对于主要用Python开发的OpenClaw来说,接入成本较低。
注意:选择mem9的一个潜在风险是,它的社区生态和成熟度可能不如Milvus或Qdrant。但在我的测试中,对于中小规模的记忆存储和检索需求,它的性能和稳定性是完全足够的。如果你的项目对向量检索的规模(比如十亿级)和性能(超低延迟)有极致要求,可能需要重新评估。
2.2 TiDB:作为“元数据”和“关系型记忆”的基石
光有向量数据库还不够。Agent的记忆不仅仅是模糊的语义片段,还有很多是结构化的“元数据”。例如:
- 记忆的归属 :这段记忆属于哪个会话(Session)?哪个用户(User)?
- 记忆的类型 :这是对话历史、工具调用记录、还是学到的知识(Knowledge)?
- 记忆的时间戳 :什么时候创建或访问的?
- 记忆的标签(Tag)和重要性评分 :方便后续的筛选和优先级排序。
这些数据是典型的关系型数据,适合用SQL来高效查询和管理。此外,Agent的某些“记忆”本身就是结构化的,比如从网页中提取的表格数据、用户提供的配置文件片段等。
这就是TiDB出场的原因。TiDB是一个兼容MySQL协议的分布式NewSQL数据库。我选择它,主要看中两点:
- 强大的水平扩展能力 :虽然初期数据量不大,但考虑到Agent记忆可能随着时间快速增长,TiDB的分布式架构让我无需担心未来的扩容问题。它可以通过增加节点来线性提升存储和计算能力。
- HTAP混合负载能力 :TiDB同时支持在线事务处理(OLTP)和在线分析处理(OLAP)。这意味着我既可以用它来快速插入和查询单条记忆的元数据(OLTP),也可以在后期对海量记忆数据进行聚合分析,比如“分析过去一周Agent最常调用的工具”(OLAP),而无需引入另一个复杂的分析数据库。
将mem9和TiDB结合,就形成了一个分工明确的记忆存储层: TiDB负责存储结构化的元数据和关系型记忆,mem9负责存储非结构化的文本及其向量嵌入,并通过一个唯一ID(如 memory_id )与TiDB中的记录关联 。
2.3 整体架构视图
基于以上选型,我设计的“私有记忆中枢”架构如下:
+-----------------------+
| OpenClaw Agent |
+-----------------------+
|
| 读写记忆
v
+---------------------------------------------+
| 记忆管理层 (Memory Manager) |
| (负责记忆的创建、编码、存储、检索和召回) |
+---------------------------------------------+
| |
| 存储元数据/关系记忆 | 存储向量/语义记忆
v v
+-------------+ +-------------+
| TiDB | | mem9 |
| (关系数据库) | | (向量数据库) |
+-------------+ +-------------+
这个架构的核心是“记忆管理层”,它是OpenClaw与底层存储之间的桥梁。接下来,我们就重点看看如何实现这一层。
3. 核心实现:构建OpenClaw的记忆管理层
记忆管理层是整套系统的“大脑”,它需要完成以下几项核心工作:
- 记忆格式化 :将Agent的原始输出(文本、JSON等)转换成结构化的记忆对象。
- 向量化 :调用嵌入模型(Embedding Model)将记忆文本转换为向量。
- 双写存储 :将记忆的元数据写入TiDB,将向量和文本写入mem9。
- 记忆检索 :根据查询条件(关键词、语义、时间、类型等)从TiDB和mem9中联合检索出相关记忆。
- 记忆注入 :将检索到的记忆,以合适的格式(如系统提示词、上下文)重新注入到OpenClaw Agent的推理循环中。
下面,我以Python代码为例,分步骤说明关键实现。
3.1 环境准备与依赖安装
首先,确保你的部署环境(我用的Ubuntu)已经准备好。
# 1. 安装Docker和Docker Compose (用于部署TiDB和mem9)
sudo apt-get update
sudo apt-get install docker.io docker-compose
# 2. 在OpenClaw的Python环境中安装必要的库
# 假设你的OpenClaw项目在一个虚拟环境中
pip install pymysql # 用于连接TiDB (MySQL协议)
pip install mem9-client # mem9的Python客户端,具体包名请查阅mem9官方文档
pip install sentence-transformers # 用于本地生成文本嵌入向量,可选,也可调用API
3.2 部署TiDB和mem9服务
为了私有化,我们使用Docker Compose在本地或内网服务器上启动服务。
docker-compose.yml
version: '3'
services:
tidb:
image: pingcap/tidb:latest
container_name: openclaw_tidb
ports:
- "4000:4000" # TiDB 服务端口
- "10080:10080" # 状态端口
command:
- --store=tikv
- --path=/data/tidb
volumes:
- ./tidb_data:/data/tidb
environment:
- TZ=Asia/Shanghai
mem9:
image: mem9db/mem9:latest # 请替换为mem9的实际官方镜像
container_name: openclaw_mem9
ports:
- "8080:8080" # mem9的API端口,假设为8080
volumes:
- ./mem9_data:/data/mem9
environment:
- MEM9_DATA_PATH=/data/mem9
运行 docker-compose up -d 启动服务。之后,你需要初始化TiDB数据库和mem9的集合(Collection)。
3.3 设计记忆数据模型(TiDB表结构)
在TiDB中,我们需要创建表来存储记忆的元数据。以下是一个核心表的设计:
-- 连接到TiDB: mysql -h 127.0.0.1 -P 4000 -u root
CREATE DATABASE IF NOT EXISTS openclaw_memory;
USE openclaw_memory;
CREATE TABLE memories (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
memory_uid VARCHAR(255) NOT NULL UNIQUE COMMENT '全局唯一记忆ID,用于关联mem9',
session_id VARCHAR(255) NOT NULL COMMENT '所属会话ID',
user_id VARCHAR(255) COMMENT '用户ID',
memory_type ENUM('conversation', 'tool_call', 'knowledge', 'reflection') NOT NULL COMMENT '记忆类型',
content TEXT COMMENT '记忆的原始文本内容(可能为摘要或完整内容)',
metadata JSON COMMENT '扩展元数据,如工具名称、参数、结果状态等',
importance_score FLOAT DEFAULT 0.0 COMMENT '重要性评分,可动态调整',
tags JSON COMMENT '标签数组,如 ["urgent", "project_x"]',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
INDEX idx_session (session_id),
INDEX idx_user (user_id),
INDEX idx_type (memory_type),
INDEX idx_created (created_at),
INDEX idx_importance (importance_score)
) COMMENT='记忆元数据主表';
这张表记录了记忆的核心元信息。 memory_uid 是关键,它将作为桥梁,关联到mem9中存储的对应向量。
3.4 实现记忆管理类(MemoryManager)
这是最核心的代码部分。我们创建一个 MemoryManager 类。
import json
import uuid
from typing import List, Dict, Any, Optional
import pymysql
from pymysql.cursors import DictCursor
# 假设mem9客户端可以这样导入
from mem9_client import Mem9Client
from sentence_transformers import SentenceTransformer
class MemoryManager:
def __init__(self, tidb_config, mem9_config, embed_model_name='all-MiniLM-L6-v2'):
"""
初始化记忆管理器。
:param tidb_config: TiDB连接配置字典
:param mem9_config: mem9连接配置字典
:param embed_model_name: 本地嵌入模型名称,如果为None则需调用API
"""
# 连接TiDB
self.tidb_conn = pymysql.connect(**tidb_config, cursorclass=DictCursor)
# 连接mem9
self.mem9_client = Mem9Client(**mem9_config)
# 在mem9中创建一个集合(Collection),类似于表
self.collection_name = "agent_memories"
try:
self.mem9_client.create_collection(self.collection_name, dimension=384) # dimension根据嵌入模型定
except Exception as e:
# 集合可能已存在
print(f"Collection might already exist: {e}")
# 初始化嵌入模型(本地模式,节省成本且隐私)
self.embed_model = SentenceTransformer(embed_model_name) if embed_model_name else None
self.embed_dim = 384 # all-MiniLM-L6-v2的维度
def _generate_embedding(self, text: str) -> List[float]:
"""生成文本的向量嵌入。"""
if self.embed_model:
# 本地模型生成
embedding = self.embed_model.encode(text).tolist()
else:
# 或者调用OpenAI/Cohere等API (需网络和API Key)
# embedding = openai.Embedding.create(...)['data'][0]['embedding']
raise NotImplementedError("请配置嵌入模型或API")
return embedding
def create_memory(self,
session_id: str,
content: str,
memory_type: str,
user_id: Optional[str] = None,
metadata: Optional[Dict] = None,
tags: Optional[List[str]] = None) -> str:
"""
创建一段新记忆。
1. 生成唯一ID和向量。
2. 将元数据写入TiDB。
3. 将向量和文本写入mem9。
"""
memory_uid = str(uuid.uuid4())
# 1. 生成向量
embedding = self._generate_embedding(content)
# 2. 写入TiDB (元数据)
with self.tidb_conn.cursor() as cursor:
sql = """
INSERT INTO memories
(memory_uid, session_id, user_id, memory_type, content, metadata, tags)
VALUES (%s, %s, %s, %s, %s, %s, %s)
"""
cursor.execute(sql, (
memory_uid, session_id, user_id, memory_type,
content[:500], # 只存摘要或前500字符到TiDB,完整内容在mem9
json.dumps(metadata) if metadata else None,
json.dumps(tags) if tags else None
))
self.tidb_conn.commit()
# 3. 写入mem9 (向量和完整内容)
mem9_record = {
"id": memory_uid, # 使用相同的UID作为mem9中的ID
"embedding": embedding,
"content": content, # 存储完整内容
"session_id": session_id,
"type": memory_type
}
# 假设mem9 client的插入接口如此
self.mem9_client.insert(self.collection_name, [mem9_record])
print(f"Memory created with UID: {memory_uid}")
return memory_uid
def search_memories(self,
query_text: str,
session_id: Optional[str] = None,
memory_type: Optional[str] = None,
limit: int = 10) -> List[Dict]:
"""
检索相关记忆。
1. 将查询文本向量化。
2. 在mem9中进行向量相似度搜索。
3. 根据mem9返回的ID,从TiDB中获取完整的元数据。
"""
# 1. 向量化查询
query_embedding = self._generate_embedding(query_text)
# 2. 在mem9中搜索
# 可以添加基于session_id或type的过滤条件,如果mem9支持元数据过滤
search_params = {"vector": query_embedding, "top_k": limit}
if session_id:
search_params["filter"] = {"session_id": session_id} # 假设mem9支持过滤
mem9_results = self.mem9_client.search(self.collection_name, **search_params)
# 3. 从TiDB获取详细信息
memory_uids = [str(res['id']) for res in mem9_results] # 假设返回结果中有'id'
if not memory_uids:
return []
with self.tidb_conn.cursor() as cursor:
# 使用IN查询,注意SQL注入风险,这里uid是生成的UUID,相对安全
placeholders = ', '.join(['%s'] * len(memory_uids))
sql = f"""
SELECT * FROM memories
WHERE memory_uid IN ({placeholders})
ORDER BY FIELD(memory_uid, {placeholders}) -- 保持mem9返回的顺序
"""
# 参数需要重复一次用于FIELD函数
params = memory_uids + memory_uids
cursor.execute(sql, params)
db_results = cursor.fetchall()
# 将TiDB的元数据与mem9的相似度分数合并
# 这里需要根据mem9返回的数据结构做匹配
result_map = {row['memory_uid']: row for row in db_results}
final_results = []
for mem9_res in mem9_results:
uid = str(mem9_res['id'])
if uid in result_map:
final_result = dict(result_map[uid])
final_result['similarity_score'] = mem9_res.get('score', 0) # 相似度分数
# 可以从mem9_res中取出完整content,如果TiDB中只存了摘要
final_result['full_content'] = mem9_res.get('content', final_result.get('content'))
final_results.append(final_result)
return final_results
def __del__(self):
"""清理连接。"""
if hasattr(self, 'tidb_conn'):
self.tidb_conn.close()
# mem9 client的关闭逻辑取决于其实现
这个 MemoryManager 类提供了最核心的创建和检索功能。在实际的OpenClaw Agent中,你需要在关键节点(如每轮对话结束、工具调用完成后)调用 create_memory ,并在Agent需要“回忆”时调用 search_memories 。
4. 与OpenClaw的集成实战
OpenClaw本身是一个框架,其核心是 Agent 和 Skill 。我们需要将记忆功能注入到它的运行生命周期中。这里没有标准答案,我分享我的两种集成思路。
4.1 方案一:作为“记忆Skill”集成
OpenClaw的Skill机制允许扩展Agent的能力。我们可以创建一个 MemorySkill 。
# memory_skill.py
from openclaw.skill import Skill, skill
from openclaw.models import Message
class MemorySkill(Skill):
def __init__(self, memory_manager: MemoryManager):
super().__init__()
self.memory_manager = memory_manager
self.current_session_id = None
@skill
async def remember(self, content: str, memory_type: str = "conversation", **kwargs):
"""记录一段记忆。"""
if not self.current_session_id:
# 可以从传入的Message或上下文中获取session_id
self.current_session_id = kwargs.get('session_id', 'default_session')
memory_uid = self.memory_manager.create_memory(
session_id=self.current_session_id,
content=content,
memory_type=memory_type,
metadata=kwargs.get('metadata'),
tags=kwargs.get('tags')
)
return {"status": "success", "memory_id": memory_uid}
@skill
async def recall(self, query: str, limit: int = 5, **kwargs):
"""回忆相关的记忆。"""
session_id = kwargs.get('session_id', self.current_session_id)
memories = self.memory_manager.search_memories(
query_text=query,
session_id=session_id,
limit=limit
)
# 将记忆格式化成Agent容易理解的文本
formatted = []
for mem in memories:
formatted.append(f"[{mem['memory_type']}] {mem['full_content'][:200]}... (Score: {mem['similarity_score']:.3f})")
return {"memories": memories, "formatted": "\n".join(formatted)}
然后,在你的主Agent初始化时,加载这个Skill。这样,Agent在运行过程中,就可以通过调用 @skill 装饰的方法来主动记录或回忆了。
4.2 方案二:通过“Harness”层拦截与自动记录
OpenClaw的文档提到了“Harness”的概念,它像是一个包裹在Agent核心逻辑之外的基础设施层。我们可以实现一个自定义的Harness,在Agent的输入输出管道中自动完成记忆操作。
# memory_harness.py
from typing import Callable, Any
from openclaw.harness import Harness
class MemoryHarness(Harness):
def __init__(self, memory_manager: MemoryManager):
self.memory_manager = memory_manager
async def around_process(self, agent, message: Message, next_fn: Callable) -> Any:
"""在Agent处理消息前后介入。"""
# 1. 在处理前,先回忆相关记忆,并注入到message的上下文中
relevant_memories = self.memory_manager.search_memories(
query_text=message.content,
session_id=message.session_id,
limit=3
)
if relevant_memories:
memory_context = "Here are some relevant past memories for reference:\n"
for mem in relevant_memories:
memory_context += f"- {mem['full_content'][:150]}\n"
# 修改或附加message,将记忆上下文加进去。具体方式取决于OpenClaw版本。
# 例如,可以放在 message.metadata 或一个单独的字段。
message.context['past_memories'] = memory_context
# 2. 调用真正的Agent处理逻辑
response = await next_fn(agent, message)
# 3. 处理完成后,将本次交互记录为记忆
memory_content = f"User: {message.content}\nAgent: {response.content}"
self.memory_manager.create_memory(
session_id=message.session_id,
content=memory_content,
memory_type="conversation",
user_id=message.user_id,
metadata={"input": message.content, "output": response.content}
)
return response
在创建Agent时,将这个Harness添加进去。这样,记忆的“记录”和“回忆”就变成了全自动、无感的过程,对Agent的核心逻辑侵入最小。这是我认为更优雅的一种方式。
5. 踩坑实录与性能调优
在实际部署和测试中,我遇到了不少问题,这里把关键的坑和解决方案记录下来。
5.1 向量模型的选择与维度对齐
问题 :最初我选用了一个维度很高的模型(如 text-embedding-ada-002 的1536维),但mem9在创建集合时需要指定维度。后来想换一个更轻量的本地模型(如 all-MiniLM-L6-v2 ,384维),却发现已经存入的向量维度不匹配,无法检索。
解决 :
- 前期确定模型 :在项目开始时就选定嵌入模型,并固定下来。如果必须更换,需要编写数据迁移脚本,将mem9中所有已有向量用新模型重新生成一遍。
- 维度参数化 :将嵌入模型的维度作为配置项,在初始化
MemoryManager和创建mem9集合时动态传入,避免硬编码。 - 性能权衡 :高维度向量精度可能更好,但存储和计算成本也高。对于Agent记忆这种对精度要求并非极致的场景,384维或768维的模型通常是性价比之选。
5.2 TiDB连接池与长连接管理
问题 :在Agent长时间运行,频繁调用记忆功能后,出现了“MySQL server has gone away”的错误。
原因 :TiDB(兼容MySQL协议)默认会关闭长时间空闲的连接。我们的 MemoryManager 在 __init__ 中创建了一个连接,如果Agent闲置一段时间,这个连接就被服务器断开了。
解决 :
- 使用连接池 :不要用单一的
pymysql.connect,改用DBUtils或SQLAlchemy提供的连接池。这样每次操作从池中获取连接,用完后归还,池会自动处理失效连接。 - 增加重试逻辑 :在数据库操作外围包裹一个重试装饰器,当捕获到连接异常时,尝试重新建立连接后再执行操作。
from dbutils.pooled_db import PooledDB
class MemoryManager:
def __init__(self, tidb_config, ...):
# 创建连接池
self.tidb_pool = PooledDB(
creator=pymysql,
maxconnections=5,
mincached=2,
**tidb_config
)
...
def _get_tidb_connection(self):
"""从连接池获取连接。"""
return self.tidb_pool.connection()
def create_memory(self, ...):
conn = self._get_tidb_connection()
try:
with conn.cursor() as cursor:
# ... 执行SQL
conn.commit()
finally:
conn.close() # 实际上是归还给连接池
5.3 记忆检索的混合查询策略
问题 :单纯依靠向量相似度搜索,有时会召回一些语义相关但上下文无关的记忆(比如来自不同会话的相似对话)。
优化 : 在 search_memories 方法中,我强化了过滤条件。
- 会话隔离 :默认优先搜索当前会话(
session_id)内的记忆,这符合大多数对话场景。 - 时间衰减 :在计算最终排序分数时,可以引入时间衰减因子,让更近的记忆有更高的权重。这需要在从TiDB获取元数据后,在应用层进行分数融合计算。
- 类型过滤 :根据当前Agent在进行的任务类型(如正在调用工具、进行总结),可以指定只检索
tool_call或knowledge类型的记忆,提高相关性。
5.4 记忆的“修剪”与重要性评分更新
问题 :如果无限制地存储所有记忆,数据量会越来越大,检索效率会下降,而且无关的旧记忆可能会干扰Agent的当前决策。
解决 :实现一个简单的记忆管理策略。
- 重要性评分动态更新 :每次一段记忆被成功召回并帮助Agent做出了正确决策,就提高它的
importance_score。反之,长期不被使用的记忆,其分数可以缓慢衰减。 - 定期修剪 :可以设置一个后台任务,定期(如每天)清理那些
importance_score低于某个阈值且创建时间过久(如30天前)的记忆。删除时,需要同时从TiDB和mem9中删除对应记录。 - 摘要化 :对于非常长的记忆内容(如一篇文档),可以在存储时同时存储一个由LLM生成的简短摘要。向量化时,既可以用全文,也可以用摘要,以平衡精度和存储开销。
6. 效果评估与未来展望
部署了这套私有记忆中枢后,我的OpenClaw Agent表现出了明显的“成长性”。最直接的感受是,在跨会话的复杂任务中,它不再是从零开始。例如,我让它“帮我整理上周提到的关于项目A的文档”,它能够回忆起上周我们对话中提到的文档关键词和存放位置,大大减少了我的重复解释。
从技术指标上看:
- 记忆召回准确率 :在测试集上,基于语义的召回Top-5相关记忆的准确率(人工判断是否相关)达到了85%以上,足够实用。
- 响应延迟 :由于嵌入模型在本地,且TiDB和mem9都在内网,单次“记忆-回忆”循环的延迟增加在100-200毫秒内,对于非实时性要求极高的对话场景可以接受。
- 系统资源占用 :TiDB和mem9的Docker容器内存占用各约500MB,加上本地嵌入模型,对服务器有一定要求,但在可接受范围内。
当然,这套方案还有很大的优化空间:
- 更智能的记忆融合 :目前只是简单地将相关记忆文本拼接到上下文中。未来可以尝试用一个小型LLM来对多段相关记忆进行总结、去重和推理,生成更精炼的“记忆提示”再给主Agent。
- 记忆图谱 :目前的记忆是扁平的片段。可以尝试建立记忆之间的关系(如“导致”、“发生于”、“关于”),形成知识图谱,让Agent能进行更复杂的逻辑推理。
- 与更多Skill联动 :例如,当
WebSearchSkill搜索到新信息时,自动将其作为knowledge类型的记忆存储起来,丰富Agent的知识库。
这个“记忆不上云”的项目,让我深刻体会到,赋予AI Agent持久化记忆,不仅仅是加一个数据库那么简单。它涉及到数据模型设计、多模态存储、检索策略、与Agent框架的深度集成以及长期的数据治理。通过mem9和TiDB的组合,我实现了一个在性能、隐私和扩展性上都能满足当前需求的解决方案。如果你也在构建需要长期记忆的AI Agent,希望这篇详尽的实践记录能给你带来一些切实可行的思路。
更多推荐



所有评论(0)