基于 mem9 与 TiDB 构建本地私有 AI 记忆中枢：从向量检索到数据自主

1. 项目缘起：为什么我们需要一个“不上云”的记忆中枢？

最近几年，AI驱动的个人助理和知识管理工具层出不穷。从Notion AI到各种笔记应用的智能插件，它们都在尝试做一件事：理解你，并基于你的历史信息提供更精准的服务。这背后依赖的核心能力，就是“记忆”——一个关于你个人偏好、工作习惯、项目上下文乃至生活琐事的庞大数据库。

然而，一个现实问题摆在我们面前：这些记忆数据存放在哪里？绝大多数服务商的选择是“上云”。你的个人数据被加密后存储在服务商的云端服务器上，模型在调用时通过网络API获取。这带来了几个无法回避的痛点：

1. 隐私与数据主权的焦虑：即便服务商承诺加密和安全，但“数据在别人服务器上”这个事实本身，就让许多对隐私敏感的用户（尤其是处理敏感信息的从业者）感到不安。你无法完全控制数据的生命周期，也无法审计其被访问的完整日志。
2. 网络依赖与延迟：每一次AI回忆你的偏好或调用历史文档，都可能意味着一次网络请求。在弱网环境或需要高频、低延迟交互的场景下（比如本地IDE插件、实时对话助理），这种延迟是不可接受的。
3. 成本与锁定风险：云服务的API调用通常按次数或token量计费，长期使用是一笔不小的开销。更关键的是，你的记忆数据格式往往与特定服务深度绑定，一旦想迁移，历史记忆可能无法完整带走，形成了事实上的“数据锁死”。
4. 自定义与深度集成的限制：云端服务通常是“黑盒”，你很难根据自己独特的业务逻辑去定制记忆的存储结构、检索策略或更新机制。比如，你想把记忆系统和本地的代码仓库、邮件客户端甚至智能家居日志深度关联，云端API往往力不从心。

正是这些痛点，催生了“记忆不上云”的需求。我们需要的不是一个功能阉割的本地替代品，而是一个能力对标甚至超越云端服务，但完全运行在私有环境下的“记忆中枢”。它应该具备强大的向量化检索能力（这是AI理解非结构化记忆的核心），支持海量数据的低成本存储与毫秒级查询，并且能轻松与各类本地应用集成。这就是我启动“OpenClaw 私有记忆中枢”项目的初衷：用mem9和TiDB这两把利器，在本地打造一个完全属于自己、性能强悍、可无限扩展的AI记忆引擎。

2. 技术选型解析：为什么是 mem9 + TiDB？

构建一个私有记忆中枢，核心是解决两个问题：“记什么、怎么记”和“存哪里、怎么查”。前者对应记忆的嵌入（Embedding）与向量化，后者对应向量的存储与检索。我的选择是mem9负责前者，TiDB负责后者。

2.1 mem9：轻量高效的记忆嵌入引擎

mem9不是一个广为人知的流行框架，它是我基于实际需求，整合多个轻量级库封装的一个工具集。它的核心目标就一个：以最低的资源开销，将各类非结构化数据（文本、图片摘要、代码片段）转化为高质量的向量（Embedding）。

为什么不用更流行的sentence-transformers或直接调用 OpenAI 的 Embedding API？

• 完全离线：mem9内置了像all-MiniLM-L6-v2这类优秀的开源小模型，仅百兆级别，在普通CPU上就能流畅运行，彻底杜绝网络请求。
• 定制化预处理：云端API通常有输入长度限制，且预处理逻辑固定。mem9允许我针对代码、邮件、聊天记录等不同来源的数据，设计不同的清洗、分段（chunking）和标准化流程。例如，对于代码文件，我会按函数或类进行分割，并保留关键的上下文信息（如导入语句、父类名），这对于后续的代码记忆检索至关重要。
• 多模态支持（简易版）：虽然核心是文本，但mem9可以集成CLIP的ViT-B/32图像编码器，将图片的简短描述或OCR提取的文字进行向量化，实现跨模态的“图文关联记忆”。

注意：mem9选择的模型在绝对精度上可能不如参数量巨大的商用模型，但对于个人或小团队的知识库、记忆库场景，其效果已经足够出色。关键在于，你拥有了对“记忆表示”的完全控制权。

2.2 TiDB：作为向量数据库的降维打击

向量存储和检索是另一个核心。市面上有专门的向量数据库如Pinecone、Weaviate（也有开源版），以及PostgreSQL的pgvector扩展。我最终选择了TiDB，一个分布式 NewSQL 数据库，原因如下：

1. 原生支持向量索引（从 v7.4.0 起实验性支持）：TiDB 的向量索引基于IVF_FLAT或HNSW算法，通过CREATE VECTOR INDEX语句即可创建，查询时使用VEC_COSINE_DISTANCE()等函数，语法直观，与SQL生态无缝集成。
2. 强大的结构化数据协同能力：记忆不仅仅是向量。一条记忆条目（Memory Item）通常包含丰富的元数据：来源（哪个文件、哪个网页URL）、创建时间、关联标签、访问频率、原始文本片段等。TiDB 作为关系型数据库，可以轻松地用一张表来管理这些结构化元数据，并与向量列存放在同一行。一次查询，既能通过向量相似度找到相关内容，又能通过元数据（如WHERE source='work_notion' AND created_at > '2024-01-01'）进行高效过滤。这是纯向量数据库需要额外架构设计才能实现的。
3. 水平扩展性与高可用：TiDB 天生的分布式架构意味着，当你的记忆库从几万条增长到几亿条时，你无需重构整个系统。通过增加 TiKV 节点（存储节点）即可线性扩展存储和计算能力。对于企业级或重度用户，这一点是单机向量数据库或pgvector难以比拟的。
4. 完整的生态与运维工具：TiDB 拥有成熟的监控（TiDB Dashboard）、备份恢复（BR）、数据迁移（TiDB Data Migration）工具链。管理一个TiDB集群，比维护一堆自建的开源向量数据库服务要规范和省心得多。

简单来说，TiDB 提供了一个“数据库级”的全栈解决方案，而不仅仅是一个“向量检索组件”。它把向量检索变成了一个如同“WHERE id = 1”一样普通的数据库操作，同时赋予了整个记忆系统处理海量结构化关联数据、随时弹性扩容的“超能力”。对于追求长期稳定、可控和集成的私有化部署场景，这个选择显得尤为合适。

3. OpenClaw 记忆中枢的架构设计与核心流程

OpenClaw 不是一个单一的软件，而是一个设计模式或参考架构。它定义了记忆从产生、处理、存储到被消费的完整生命周期。下图勾勒了其核心组件与数据流：

（注：此处用文字描述架构，因禁止使用Mermaid图表） 整个系统分为三层：

• 接入层：由一系列“采集器”（Collector）组成，它们是轻量级的守护进程或脚本，监控不同数据源。例如：fs-watcher监控指定目录的文件变动；mail-fetcher定期拉取邮件摘要；browser-extension捕获划词和网页保存。
• 处理与存储层：核心层。采集器将原始数据发送到“记忆处理中心”（一个常驻服务）。该中心调用mem9对内容进行清洗、分段、向量化，然后将向量和结构化元数据作为一个完整的事务，写入 TiDB 集群的memory_items表中。TiDB 表同时包含id,content_text,embedding_vector,source,tags,timestamp等字段，并在embedding_vector列上创建向量索引。
• 应用层：提供统一的查询接口（如gRPC或REST API）。任何需要“记忆”的应用（如本地AI助手、IDE插件、笔记软件）都可以向该接口发起查询。查询可以是纯文本（会被mem9实时向量化），也可以是带有元数据过滤条件的混合查询。接口将请求转化为SQL，利用 TiDB 的向量索引和条件索引进行毫秒级检索，返回最相关的几条记忆及其完整上下文。

核心工作流程示例：保存一篇技术博客

1. 你用浏览器插件将一篇关于“Rust并发模型”的博客保存到OpenClaw。
2. 采集器将博客URL、标题、正文内容、保存时间戳打包发送。
3. 记忆处理中心启动mem9管道：先清理HTML标签，然后按章节将正文分割成多个语义段落（每个段落约200-300词）。接着，用all-MiniLM-L6-v2模型将每个段落转化为一个384维的向量。
4. 对于每一个段落，在 TiDB 中插入一条记录：content_text存储段落原文，embedding_vector存储向量，source存储博客URL，tags自动打上["rust", "concurrency", "blog"]等标签。
5. 整个过程在数秒内异步完成，你对这篇博客的“记忆”已经牢固地存储在你的私有集群中。

4. 从零到一：搭建你的私有记忆中枢

理论说再多，不如动手搭一个。下面我将以一台配置尚可的Linux开发机（至少4核CPU，8GB内存）为例，演示如何搭建一个最小可用的OpenClaw系统。

4.1 基础环境与TiDB集群部署

我们使用 TiDB 的离线包和tiup工具在单机上部署一个迷你测试集群（1个PD， 1个TiKV， 1个TiDB）。生产环境请参考官方文档部署多节点集群。

# 1. 安装 tiup curl --proto '=https' --tlsv1.2 -sSf https://tiup-mirrors.pingcap.com/install.sh | sh source ~/.bashrc # 2. 创建拓扑配置文件 tidb-test.yaml cat > tidb-test.yaml <<EOF global: user: "tidb" ssh_port: 22 deploy_dir: "/home/tidb/deploy" data_dir: "/home/tidb/data" pd_servers: - host: 127.0.0.1 tidb_servers: - host: 127.0.0.1 tikv_servers: - host: 127.0.0.1 monitoring_servers: - host: 127.0.0.1 grafana_servers: - host: 127.0.0.1 EOF # 3. 部署集群（指定版本，确保支持向量索引） tiup cluster deploy test-cluster v7.5.0 ./tidb-test.yaml --user root -p tiup cluster start test-cluster # 4. 安装 MySQL 客户端并连接 sudo apt-get install mysql-client mysql -h 127.0.0.1 -P 4000 -u root

连接成功后，创建我们的记忆库数据库和表：

CREATE DATABASE IF NOT EXISTS openclaw_memory; USE openclaw_memory; CREATE TABLE memory_items ( id BIGINT AUTO_INCREMENT PRIMARY KEY, content_text TEXT NOT NULL COMMENT '原始文本内容', embedding_vector VECTOR(384) NOT NULL COMMENT 'mem9生成的384维向量', source VARCHAR(500) COMMENT '来源，如文件路径、URL', content_hash CHAR(64) COMMENT '内容SHA256，用于去重', tags JSON COMMENT '标签数组', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_source (source(100)), INDEX idx_created_at (created_at), INDEX idx_content_hash (content_hash), VECTOR INDEX idx_vector (embedding_vector) USING IVF_FLAT ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;

关键点：VECTOR(384)指定了向量维度，需与mem9使用的模型维度一致。VECTOR INDEX是查询性能的关键。content_hash用于去重，避免同一内容被重复存储。

4.2 构建与配置 mem9 处理服务

mem9的核心是一个Python服务。我们创建一个项目目录。

mkdir openclaw-processor && cd openclaw-processor python -m venv venv source venv/bin/activate pip install sentence-transformers pillow torch numpy pymysql

接下来是mem9的核心代码文件mem9_encoder.py：

# mem9_encoder.py import numpy as np from sentence_transformers import SentenceTransformer from typing import List, Optional import logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class Mem9Encoder: def __init__(self, model_name: str = 'all-MiniLM-L6-v2', device: str = 'cpu'): """ 初始化记忆编码器。 :param model_name: 本地或HuggingFace模型名 :param device: 'cpu' 或 'cuda' """ logger.info(f"Loading model: {model_name} on {device}") self.model = SentenceTransformer(model_name, device=device) self.dimension = self.model.get_sentence_embedding_dimension() logger.info(f"Model loaded. Vector dimension: {self.dimension}") def chunk_text(self, text: str, chunk_size: int = 300, overlap: int = 50) -> List[str]: """ 简单的按词滑动窗口分块。生产环境建议使用更智能的分段（如按句子、段落）。 """ words = text.split() chunks = [] for i in range(0, len(words), chunk_size - overlap): chunk = ' '.join(words[i:i + chunk_size]) chunks.append(chunk) if i + chunk_size >= len(words): break return chunks def encode_text(self, text: str, do_chunk: bool = True) -> List[np.ndarray]: """ 将文本编码为向量列表。 :param text: 输入文本 :param do_chunk: 是否先分块 :return: 向量列表，每个numpy数组 shape=(dimension,) """ if do_chunk: text_chunks = self.chunk_text(text) else: text_chunks = [text] if not text_chunks: return [] # 批量编码提升效率 embeddings = self.model.encode(text_chunks, convert_to_numpy=True, show_progress_bar=False, normalize_embeddings=True) # 归一化便于余弦相似度计算 return [emb for emb in embeddings] # 初始化一个全局编码器实例 encoder = Mem9Encoder()

然后，我们编写主处理服务processor_service.py，它负责连接数据库、处理数据入库。

# processor_service.py import pymysql import hashlib import json from mem9_encoder import encoder import threading from queue import Queue import time class MemoryProcessor: def __init__(self, db_config): self.db_config = db_config self.task_queue = Queue() self.worker_thread = threading.Thread(target=self._worker, daemon=True) self.worker_thread.start() def _get_db_connection(self): return pymysql.connect(**self.db_config) def add_memory_task(self, content: str, source: str, tags: list = None): """外部调用，添加一个记忆处理任务到队列""" task = { 'content': content, 'source': source, 'tags': tags or [] } self.task_queue.put(task) def _worker(self): """后台工作线程，从队列取任务处理""" while True: task = self.task_queue.get() try: self._process_single_task(task) except Exception as e: print(f"Error processing task from {task.get('source')}: {e}") finally: self.task_queue.task_done() def _process_single_task(self, task): content = task['content'] source = task['source'] tags = task['tags'] # 1. 计算内容哈希用于去重 content_hash = hashlib.sha256(content.encode('utf-8')).hexdigest() # 2. 检查是否已存在 conn = self._get_db_connection() try: with conn.cursor() as cursor: cursor.execute("SELECT id FROM memory_items WHERE content_hash = %s", (content_hash,)) if cursor.fetchone(): print(f"Content already exists, skipped. Hash: {content_hash[:16]}...") return finally: conn.close() # 3. 编码文本为向量 vectors = encoder.encode_text(content) if not vectors: print("No vectors generated, skipped.") return # 4. 将每个向量块存入数据库 conn = self._get_db_connection() try: with conn.cursor() as cursor: for idx, vec in enumerate(vectors): # 将numpy数组转换为逗号分隔的字符串，TiDB的VECTOR类型需要这种格式 vector_str = ','.join([str(x) for x in vec]) # 提取该块对应的文本（这里简化处理，实际应存储对应分块原文） chunk_text = content[:500] + "..." if len(content) > 500 else content # 示例，实际应存储分块后的文本 sql = """ INSERT INTO memory_items (content_text, embedding_vector, source, content_hash, tags) VALUES (%s, %s, %s, %s, %s) """ cursor.execute(sql, (chunk_text, vector_str, source, content_hash, json.dumps(tags, ensure_ascii=False))) conn.commit() print(f"Successfully saved memory from {source}, {len(vectors)} chunks inserted.") except Exception as e: conn.rollback() raise e finally: conn.close() # 配置和启动 if __name__ == "__main__": db_config = { 'host': '127.0.0.1', 'port': 4000, 'user': 'root', 'password': '', # 你的密码 'database': 'openclaw_memory', 'charset': 'utf8mb4' } processor = MemoryProcessor(db_config) # 模拟添加一个任务 test_content = """ TiDB is an open-source, distributed SQL database that supports Hybrid Transactional and Analytical Processing (HTAP) workloads. It is MySQL compatible and features horizontal scalability, strong consistency, and high availability. """ processor.add_memory_task(test_content, source="manual_test", tags=["database", "tidb", "opensource"]) # 保持主线程运行，等待队列处理 time.sleep(5)

运行这个服务，它将在后台监听任务队列。你可以通过add_memory_task方法从任何地方（如文件监控脚本、Web API）添加需要记忆的内容。

4.3 实现记忆查询接口

记忆存进去了，怎么用？我们需要一个查询接口。创建一个简单的 Flask 应用query_api.py：

# query_api.py from flask import Flask, request, jsonify import pymysql import json from mem9_encoder import encoder import numpy as np app = Flask(__name__) # 数据库配置 DB_CONFIG = { 'host': '127.0.0.1', 'port': 4000, 'user': 'root', 'password': '', 'database': 'openclaw_memory', 'charset': 'utf8mb4' } def search_memories(query_text: str, top_k: int = 5, filter_source: str = None): """ 核心检索函数：将查询文本向量化，并在TiDB中执行向量相似度搜索。 """ # 1. 将查询文本编码为向量 query_vector_list = encoder.encode_text(query_text, do_chunk=False) if not query_vector_list: return [] query_vector = query_vector_list[0] # 查询通常不分块 query_vector_str = ','.join([str(x) for x in query_vector]) # 2. 构建SQL查询 conn = pymysql.connect(**DB_CONFIG) try: with conn.cursor(pymysql.cursors.DictCursor) as cursor: sql = """ SELECT id, content_text, source, tags, VEC_COSINE_DISTANCE(embedding_vector, %s) as distance FROM memory_items WHERE 1=1 """ params = [query_vector_str] if filter_source: sql += " AND source = %s" params.append(filter_source) sql += " ORDER BY distance ASC LIMIT %s" params.append(top_k) cursor.execute(sql, params) results = cursor.fetchall() # 余弦距离越小越相似，转换为相似度分数 (1 - distance) for r in results: r['similarity'] = 1.0 - float(r['distance']) del r['distance'] return results finally: conn.close() @app.route('/search', methods=['POST']) def search(): data = request.json query = data.get('query', '') top_k = data.get('top_k', 5) source = data.get('source', None) # 可选过滤条件 if not query: return jsonify({'error': 'Query text is required'}), 400 try: memories = search_memories(query, top_k, source) return jsonify({'results': memories}) except Exception as e: return jsonify({'error': str(e)}), 500 if __name__ == '__main__': app.run(host='0.0.0.0', port=5000, debug=False)

启动这个API服务 (python query_api.py)，你就可以通过发送一个POST请求来检索记忆了：

curl -X POST http://localhost:5000/search \ -H "Content-Type: application/json" \ -d '{"query": "What is a distributed SQL database?", "top_k": 3}'

返回的结果会包含与你问题最相关的记忆片段、来源、标签和相似度分数。

5. 踩坑实录：向量索引的优化与查询调优

在初步搭建系统并灌入数万条测试数据后，我遇到了第一个性能瓶颈：查询速度不稳定，有时很快（<100ms），有时慢得惊人（>2s）。

5.1 问题定位：全表扫描与向量索引失效

首先，我检查了TiDB的慢查询日志。发现一些VEC_COSINE_DISTANCE查询没有使用我们创建的idx_vector索引，而是进行了全表扫描。原因在于我的查询SQL中包含了WHERE source = 'some_source'这样的过滤条件。TiDB的优化器在评估使用向量索引后还需要回表过滤source字段的成本时，可能错误地选择了全表扫描。

解决方案：使用强制索引提示或优化查询结构。

-- 修改前的查询（可能不走向量索引） SELECT id, content_text, VEC_COSINE_DISTANCE(embedding_vector, %s) as dist FROM memory_items WHERE source = 'notion' ORDER BY dist ASC LIMIT 10; -- 方案1：使用FORCE INDEX SELECT id, content_text, VEC_COSINE_DISTANCE(embedding_vector, %s) as dist FROM memory_items FORCE INDEX(idx_vector) WHERE source = 'notion' ORDER BY dist ASC LIMIT 10; -- 方案2：子查询优化（先向量检索，后过滤） SELECT t.id, t.content_text, t.dist FROM ( SELECT id, content_text, source, VEC_COSINE_DISTANCE(embedding_vector, %s) as dist FROM memory_items ORDER BY dist ASC LIMIT 100 -- 先取更多候选 ) AS t WHERE t.source = 'notion' ORDER BY t.dist ASC LIMIT 10;

经过测试，在数据量较大（>10万条）且过滤条件选择性较强时，方案2（子查询）往往更优。因为它先利用向量索引快速缩小范围（取前100个最相似的），再在这小范围结果里进行精确过滤，避免了强制索引可能带来的代价估算错误。我在query_api.py的search_memories函数中最终采用了这种模式。

5.2 参数调优：IVF_FLAT 索引的 nlist 参数

创建向量索引时，我最初使用了默认参数。但随着数据量增长，召回精度和查询速度的平衡出现了问题。TiDB的IVF_FLAT索引有一个关键参数nlist，它决定了聚类中心的数量。

-- 创建索引时指定 nlist CREATE VECTOR INDEX idx_vector_optimized ON memory_items (embedding_vector) USING IVF_FLAT WITH (nlist = 2048);

• nlist太小（如128）：每个聚类包含的向量很多，查询时需要计算查询向量与大量候选向量的距离，查询速度慢，但召回率（找到真正最相似向量的概率）高。
• nlist太大（如10000）：每个聚类包含的向量少，查询速度快，但可能因为“搜索粒度”太粗，而漏掉真正最相似的向量（它可能在相邻的聚类里）。

如何确定最佳nlist？一个经验法则是nlist = sqrt(N)，其中 N 是向量总数。对于百万级数据，nlist=2048或4096是个不错的起点。我通过一个简单的测试脚本，在验证集上对比了不同nlist值下的查询耗时和召回率（与暴力扫描结果对比），最终为我的数据规模（约50万条）选择了nlist=1024。

实操心得：向量索引的调优是一个实验性过程。建议在数据量达到一定规模后（比如超过10万），重新评估并重建索引。可以准备一个小型但具有代表性的查询测试集，用来自动化评估不同索引参数下的性能与精度。

5.3 内存与批量处理：避免“记忆洪水”

另一个坑出现在“批量导入”历史数据时。我写了一个脚本，一次性读取上万篇旧博客文章往系统里灌。很快，处理服务的内存占用飙升，然后被系统OOM（内存溢出）杀死。

问题出在mem9的编码环节。sentence-transformers的encode函数默认会一次性处理所有传入的文本，如果一次性传入几万个文本块，它会试图在内存中同时为它们所有计算向量，这自然会导致内存爆炸。

解决方案：实现分批次编码与数据库提交。

我修改了MemoryProcessor._process_single_task方法中处理批量任务的部分（如果是单个大文本分块很多，也适用）：

def _process_single_task_batch(self, task_list): """处理批量任务，避免内存溢出""" conn = self._get_db_connection() cursor = conn.cursor() batch_size = 32 # 根据GPU/CPU内存调整 try: for i in range(0, len(task_list), batch_size): batch = task_list[i:i+batch_size] all_vectors_for_batch = [] all_db_params_for_batch = [] for task in batch: # ... 计算hash，去重检查（可批量优化）... vectors = encoder.encode_text(task['content']) for vec in vectors: vector_str = ','.join([str(x) for x in vec]) # 收集参数 all_db_params_for_batch.append(( task['chunk_text'], vector_str, task['source'], task['content_hash'], json.dumps(task['tags']) )) # 批量插入 if all_db_params_for_batch: placeholders = ','.join(['(%s, %s, %s, %s, %s)'] * len(all_db_params_for_batch)) sql = f"INSERT INTO memory_items (content_text, embedding_vector, source, content_hash, tags) VALUES {placeholders}" # 扁平化参数列表 flat_params = [item for sublist in all_db_params_for_batch for item in sublist] cursor.execute(sql, flat_params) conn.commit() print(f"Committed batch {i//batch_size + 1}") except Exception as e: conn.rollback() raise e finally: cursor.close() conn.close()

通过控制编码和插入的批次大小，系统内存使用变得平稳，吞吐量也保持在高位。

6. 进阶玩法：让记忆“活”起来

基础的系统搭建好后，就可以在此基础上玩出很多花样，让记忆真正成为你数字生活的“第二大脑”。

6.1 记忆的主动推送与上下文关联

一个被动的记忆库，需要你主动去查询。一个主动的记忆中枢，应该能在你需要的时候“跳出来”提醒你。我实现了一个简单的“上下文关联推送”机制。

在我的IDE（VSCode）中，一个后台插件会分析我当前正在编辑的文件（比如一个Python文件），提取关键实体（函数名、类名、导入的库），然后实时向OpenClaw查询接口发送这些关键词。系统返回相关的记忆片段，比如我过去写的关于这个函数的注释、解决过的类似bug的日志、相关的API文档摘要。这些信息以非侵入式的形式显示在IDE侧边栏，当我卡住时，瞥一眼就能获得灵感。

实现的关键在于丰富记忆条目的元数据标签。在记忆入库时，不仅使用通用NLP模型提取关键词，还针对特定类型内容使用专门解析器。例如，对于代码，使用tree-sitter解析语法树，提取出函数签名、类名、使用的库作为标签。这样，当查询“pandas.DataFrame.merge”时，系统能精准找到我过去学习merge用法的笔记和实战代码片段，而不是泛泛的关于Python或数据分析的文章。

6.2 记忆的衰减与强化：模拟遗忘曲线

人的记忆会遗忘，AI的记忆是否需要“遗忘”？我认为，对于私有记忆中枢，有选择的遗忘（归档）和强化是提升质量的关键。

我在memory_items表中增加了两个字段：access_count（访问次数）和last_accessed（最后访问时间）。每次记忆被查询并最终被用户点击查看详情时，这两个字段就会更新。

我设置了一个定时任务（Cron Job），每周运行一次“记忆整理”：

1. 强化高频记忆：对于access_count高且last_accessed较近的记忆，将其向量表示进行“微调”。这不是重新训练模型，而是将其与近期相关的其他记忆向量进行加权平均，产生一个更“泛化”、更“核心”的新向量，并新增一条记录，同时给旧记录打上archived标签。这模拟了大脑中重要记忆被反复强化、变得愈发稳固和抽象的过程。
2. 归档低频记忆：对于超过半年未访问且access_count极低的记忆，将其移入单独的memory_archive表，并从主表中删除。主表只保留“活跃记忆”，保证查询速度。归档的记忆仍然可以被专门的“深度挖掘”查询检索到，只是不在高频路径上。这解决了数据无限膨胀带来的性能和管理问题。

6.3 与本地大模型（LLM）的集成：从记忆到思考

最终的形态，是让OpenClaw成为本地大模型（如通过ollama运行的Llama 3、Qwen）的“长期记忆体”。当我在终端与LLM对话时，对话历史首先被摘要并存入OpenClaw。当我提出一个新问题时：

1. 问题文本被向量化，在OpenClaw中检索出最相关的若干条历史记忆和知识片段。
2. 这些记忆片段作为“上下文”，和我的新问题一起，构造成一个Prompt，发送给本地LLM。
3. LLM基于我个人的历史对话风格、已知信息和偏好来生成回答，实现了真正的“个性化AI”。

这个集成的效果是颠覆性的。AI不再每次对话都是“金鱼脑”，它能记住我上周让它分析的某个数据模式，记得我更喜欢代码解释用Python而不是伪代码，甚至能在我问“我们上次讨论的那个方案”时，准确找到上下文。这一切，完全运行在本地，没有数据离开我的机器。

7. 总结与展望

构建“OpenClaw 私有记忆中枢”的过程，是一个将前沿的向量检索技术与成熟的分布式数据库相结合，来解决实际隐私和可控性需求的实践。mem9提供了灵活、本地的记忆编码能力，而TiDB则赋予了这套系统工业级的存储、查询和管理能力。这套方案的优势在于：

• 完全自主可控：从数据到算力，都在自己手里。
• 性能与规模兼顾：得益于TiDB的分布式架构，记忆库可以从小型个人笔记库平滑扩展为企业级知识库。
• 强大的关联查询：向量相似度搜索与结构化元数据过滤的有机结合，让检索更加精准。
• 生态友好：标准的MySQL协议和SQL语法，使得任何能连接MySQL的工具或应用都能与之交互，集成成本极低。

当然，这套系统目前还有很多可以深化的方向。例如，探索更高效的多模态向量融合方式，实现对图片、音频的直接记忆；实现记忆之间的图关系存储，不仅能根据相似度检索，还能根据逻辑关系（如“是A的原因”、“是B的步骤”）进行推理式检索；优化向量索引的自动调参策略等。

对我个人而言，这个项目最大的收获不仅仅是技术上的，更是一种思维上的转变：在AI时代，个人数据的主权和控制权至关重要。通过开源工具和自建系统，我们完全有能力打造一个不逊于云端服务、且完全属于自己的智能基础设施。这不仅仅是“记忆不上云”，更是“智能不下线”，是我们在数字世界中构建自主人格与能力的一次扎实尝试。如果你也对数据隐私和个性化AI感兴趣，不妨从搭建一个微型的OpenClaw开始，感受一下完全属于自己的“记忆宫殿”所带来的安全感和强大能力。