构建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数据库。我选择它，主要看中两点：

1. 强大的水平扩展能力：虽然初期数据量不大，但考虑到Agent记忆可能随着时间快速增长，TiDB的分布式架构让我无需担心未来的扩容问题。它可以通过增加节点来线性提升存储和计算能力。
2. 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的记忆管理层

记忆管理层是整套系统的“大脑”，它需要完成以下几项核心工作：

1. 记忆格式化：将Agent的原始输出（文本、JSON等）转换成结构化的记忆对象。
2. 向量化：调用嵌入模型（Embedding Model）将记忆文本转换为向量。
3. 双写存储：将记忆的元数据写入TiDB，将向量和文本写入mem9。
4. 记忆检索：根据查询条件（关键词、语义、时间、类型等）从TiDB和mem9中联合检索出相关记忆。
5. 记忆注入：将检索到的记忆，以合适的格式（如系统提示词、上下文）重新注入到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维），却发现已经存入的向量维度不匹配，无法检索。

解决：

1. 前期确定模型：在项目开始时就选定嵌入模型，并固定下来。如果必须更换，需要编写数据迁移脚本，将mem9中所有已有向量用新模型重新生成一遍。
2. 维度参数化：将嵌入模型的维度作为配置项，在初始化MemoryManager和创建mem9集合时动态传入，避免硬编码。
3. 性能权衡：高维度向量精度可能更好，但存储和计算成本也高。对于Agent记忆这种对精度要求并非极致的场景，384维或768维的模型通常是性价比之选。

5.2 TiDB连接池与长连接管理

问题：在Agent长时间运行，频繁调用记忆功能后，出现了“MySQL server has gone away”的错误。

原因：TiDB（兼容MySQL协议）默认会关闭长时间空闲的连接。我们的MemoryManager在__init__中创建了一个连接，如果Agent闲置一段时间，这个连接就被服务器断开了。

解决：

1. 使用连接池：不要用单一的pymysql.connect，改用DBUtils或SQLAlchemy提供的连接池。这样每次操作从池中获取连接，用完后归还，池会自动处理失效连接。
2. 增加重试逻辑：在数据库操作外围包裹一个重试装饰器，当捕获到连接异常时，尝试重新建立连接后再执行操作。

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方法中，我强化了过滤条件。

1. 会话隔离：默认优先搜索当前会话（session_id）内的记忆，这符合大多数对话场景。
2. 时间衰减：在计算最终排序分数时，可以引入时间衰减因子，让更近的记忆有更高的权重。这需要在从TiDB获取元数据后，在应用层进行分数融合计算。
3. 类型过滤：根据当前Agent在进行的任务类型（如正在调用工具、进行总结），可以指定只检索tool_call或knowledge类型的记忆，提高相关性。

5.4 记忆的“修剪”与重要性评分更新

问题：如果无限制地存储所有记忆，数据量会越来越大，检索效率会下降，而且无关的旧记忆可能会干扰Agent的当前决策。

解决：实现一个简单的记忆管理策略。

1. 重要性评分动态更新：每次一段记忆被成功召回并帮助Agent做出了正确决策，就提高它的importance_score。反之，长期不被使用的记忆，其分数可以缓慢衰减。
2. 定期修剪：可以设置一个后台任务，定期（如每天）清理那些importance_score低于某个阈值且创建时间过久（如30天前）的记忆。删除时，需要同时从TiDB和mem9中删除对应记录。
3. 摘要化：对于非常长的记忆内容（如一篇文档），可以在存储时同时存储一个由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，希望这篇详尽的实践记录能给你带来一些切实可行的思路。