基于 PolarDB Mem0 为 OpenClaw 注入企业级长期记忆

OpenClaw 原生记忆系统

OpenClaw 记忆系统采用"文件优先（File-First）"与"本地优先（Local-First）"的分层架构，通过 Markdown 文档作为唯一的"事实来源（Source of Truth）"，解决了 LLM 原生上下文易失、昂贵且不可追溯的问题。

• 记忆层级设计：OpenClaw 构建了时态分层记忆模型，模拟人类认知结构。

  1. 长期精选记忆（MEMORY.md）：承载核心人格与持久知识。

  2. 短期工作日志（Daily Logs）：记录任务细节与操作记录。

  3. 会话事件流（Sessions）：保留原始交互轨迹。

  4. 精确检索：通过内置的 SQLite 混合检索引擎（结合 BM25 全文搜索与向量关联），确保信息召回的精准度。

• 方案的局限性：随着交互深度增加，基于文件的默认方案显露出局限性。

  • 扩展瓶颈：Markdown 文件体量过大时，文本分块检索效率降低。

  • 逻辑冲突：缺乏对事实动态更新的识别能力（如用户更换住址）。

  • 协作孤岛：本地文件难以在多 Agent 或不同终端间实现实时共享。

PolarDB Mem0 增强方案

PolarDB Mem0 引入"自我演进"的记忆逻辑，能够从对话中提取、更新并固化事实（Facts），而非简单的文本切分。通过将 Mem0 与阿里云 PolarDB 集成，利用 PolarDB 的云原生架构为 Agent 提供企业级可靠性的"大容量长期记忆"。

核心组件

整个解决方案由三个核心组件构成：

1. OpenClaw Agent 框架：提供任务编排与指令执行能力

2. PolarDB Mem0 服务：提供云端记忆存储与检索能力

3. Skill 插件层：实现 OpenClaw 与 PolarDB Mem0 的集成

步骤一：开通 PolarDB Mem0 服务

1. 登录 阿里云 PolarDB Mem0 控制台。

2. 在左侧菜单栏选择 PolarDB Mem0，单击新建记忆。

3. 在PolarDB Mem0 创建页，购买实例。

   重要

   该功能当前仅在华东2（上海）L/M、华北2（北京）L 和华东1（杭州）J/K 可用区提供资源。

4. 实例创建成功后，点击实例ID，进入实例的基本信息页签，获取服务接入地址。

   

5. 在配置页签，获取apikey。

   

6. 进入白名单页签中，添加 OpenClaw 服务所在服务器的 IP 地址。

步骤二：创建 Skill 目录结构

在 OpenClaw 的 skills 目录下创建新文件夹 agent-memory-polardb，目录结构如下：

openclaw/
└── skills/
    └── agent-memory-polardb/
        ├── manifest.json    # 插件元数据与 API 接口规范
        ├── handler.py       # 核心处理逻辑
        └── SKILL.md         # AI 指令文档

步骤三：编写插件元数据配置

创建 manifest.json 文件，定义 Skill 的元数据和 API 接口规范。

{
  "name": "PolarDB Agent Memory",
  "id": "agent-memory-polardb",
  "version": "1.0.0",
  "description_for_model": "基于 PolarDB 的长期记忆工具。支持跨会话的事实存储 (save_fact)、语义检索 (search_memories) 和历史清理 (delete_all_memories)，提供毫秒级上下文召回。",
  "description_for_human": "由阿里云 PolarDB 驱动的 Agent 长期记忆管理服务。",
  "auth": {
    "type": "token",
    "token_header": "Authorization",
    "token_prefix": "Token"
  },
  "api": {
    "type": "python",
    "main_file": "handler.py",
    "functions": [
      {
        "name": "save_fact",
        "description": "持久化存储用户的事实、偏好或习惯。",
        "parameters": {
          "type": "object",
          "properties": {
            "user_id": { "type": "string", "description": "唯一用户ID" },
            "fact": { "type": "string", "description": "需要记忆的具体内容" }
          },
          "required": ["user_id", "fact"]
        }
      },
      {
        "name": "search_memories",
        "description": "基于语义检索该用户的相关历史记录以获取上下文。",
        "parameters": {
          "type": "object",
          "properties": {
            "user_id": { "type": "string", "description": "唯一用户ID" },
            "query": { "type": "string", "description": "搜索关键词或语义描述" }
          },
          "required": ["user_id", "query"]
        }
      },
      {
        "name": "delete_all_memories",
        "description": "永久清空该用户的所有历史记忆记录（不可逆）。",
        "parameters": {
          "type": "object",
          "properties": {
            "user_id": { "type": "string", "description": "唯一用户ID" }
          },
          "required": ["user_id"]
        }
      }
    ]
  }
}

参数说明

步骤四：编写核心处理逻辑

创建 handler.py 文件，负责调用 PolarDB Mem0 的 API 进行记忆的存取。

import os
import json
import subprocess
import urllib.parse

#HOST = os.getenv("POLAR_MEM0_HOST") 从PolarDB Mem0官网获取HOTS，文档：https://help.aliyun.com/zh/polardb/polardb-for-mysql/polardb-mem0
#API_KEY = os.getenv("POLAR_MEM0_API_KEY")  从PolarDB Mem0官网获取API_KEY，文档：https://help.aliyun.com/zh/polardb/polardb-for-mysql/polardb-mem0
HOST = os.getenv("MEM0_HOST", "").strip().replace('"', '').replace("'", "")
API_KEY = os.getenv("MEM0_API_KEY", "").strip().replace('"', '').replace("'", "")
if not HOST or not API_KEY:
    raise ValueError(f"环境变量配置失败! HOST: {HOST}, KEY: {API_KEY}")

def _run_safe_curl(url, payload, method='POST'):

    if not HOST or not API_KEY:
        return {"error": "Missing configuration"}

    # 1. 无论 payload 是否为空，先初始化变量
    if payload is not None:
        input_data = json.dumps(payload, ensure_ascii=False)
    else:
        input_data = ""

    cmd = [
        "curl", "-s", "-X", method, url,
        "-H", f"Authorization: Token {API_KEY}",
        "-H", "Content-Type: application/json",
        "--data-binary", "@-",           # 核心：从标准输入读取数据
        "--max-time", "15",              # 超时保护
        "--no-buffer"                    # 禁用缓冲
    ]

    try:
        input_data = json.dumps(payload)

        result = subprocess.run(
            cmd, input=input_data, capture_output=True, 
            text=True, check=True, encoding='utf-8'
        )

        output = result.stdout.strip()
        if not output:
            return {"status": "success"}
        return json.loads(output)

    except subprocess.CalledProcessError as e:
        return {"error": f"Curl command failed: {e.stderr}"}
    except Exception as e:
        return {"error": str(e)}

def save_fact(user_id, fact):
    """保存事实：将传入的 fact 转发给 PolarDB"""
    url = f"{HOST}/v1/memories"
    payload = {
        "messages": [
            {"role": "user", "content": fact}
        ],
        "user_id": str(user_id)
    }
    return _run_safe_curl(url, payload, method='POST')

def search_memories(user_id, query):
    """直接调用云端集成接口搜索记忆"""
    url = f"{HOST}/v2/memories/search"
    payload = {
        "query": query, 
        "user_id": str(user_id)
    }
    return _run_safe_curl(url, payload, method='POST')

def delete_all_memories(user_id):
    """
    根据你验证成功的命令：参数必须放在 URL 的 query string 中
    """
    encoded_id = urllib.parse.quote(str(user_id))
    url = f"{HOST}/v1/memories?user_id={encoded_id}"
    return _run_safe_curl(url, payload=None, method='DELETE')

步骤五：功能测试

创建 test_flow.py 文件，用于验证插件功能的完整性。

import time
from handler import save_fact, search_memories, delete_all_memories

def run_test():
    user_id = "test_user_999"
    test_fact = "我最喜欢的食物是麻婆豆腐。"

    print("---  开始闭环功能测试 ---")

    # 1. 清理环境（防止旧数据干扰）
    print("\n[Step 1] 清理旧数据...")
    delete_all_memories(user_id)

    # 2. 存储新事实
    print(f"\n[Step 2] 存储事实: '{test_fact}'")
    save_res = save_fact(user_id, test_fact)
    print(f"响应: {save_res}")

    # 给服务器一点点处理时间（索引同步）
    time.sleep(30)

    # 3. 搜索验证
    print("\n[Step 3] 搜索关键词: '食物'...")
    search_res = search_memories(user_id, "食物")
    
    # 兼容处理：Mem0 有时返回列表，有时返回带 results 的字典
    results = search_res.get('results', []) if isinstance(search_res, dict) else search_res

    if len(results) > 0:
        print(f"✅ 搜索成功，找到内容: {results}")
    else:
        print("❌ 搜索失败：未找到刚存入的内容。")
        return

     # 4. 执行删除
    print("\n[Step 4] 删除该用户所有记忆...")
    del_res = delete_all_memories(user_id)
    print(f"响应: {del_res}")

    # 5. 最终搜索验证（应该是搜不到）
    print("\n[Step 5] 再次搜索相同关键词...")
    final_res = search_memories(user_id, "食物")
    final_results = final_res.get('results', []) if isinstance(final_res, dict) else final_res

    if len(final_results) == 0:
        print("✨ 闭环测试成功：数据已彻底删除！")
    else:
        print(f"⚠️ 闭环测试异常：删除后依然能搜到内容: {final_results}")
if __name__ == "__main__":
    run_test()

步骤六：编写 AI 指令文档

创建 SKILL.md 文件，用于教会 AI 何时调用记忆功能。

---
name: agent-memory-polardb
description: 使用 PolarDB 和 Mem0 为智能体提供长期记忆管理。
allowed-tools:
  - save_fact
  - search_memories
  - delete_all_memories
metadata:
  category: memory
  provider: polardb
---

# Instructions
你现在已拥有由 PolarDB 驱动的“云原生外挂大脑”。该Skill集成了 PolarDB Mem0 服务，为 OpenClaw 智能体（Agents）提供长期记忆能力，取代原生的、粗粒度的 Markdown 文件存储，它能够实现对用户偏好、事实记忆和事件关系的精确提取及毫秒级检索，支持跨会话的知识持久化。请遵循以下原则：
1. 主动记忆 (Memorize)：捕捉核心事实（身份、技能）或明确偏好（习惯、禁忌）。
   - 用户提到“我有两个孩子”、“我正在学习 Rust”时，调用 save_fact。
   - 用户表示偏好“我不喜欢深色模式”、“我周一早上通常没空”，调用 save_fact。
2. 上下文检索 (Recall)：启动新任务或追溯历史时，调用 search_memories 获取背景，确保对话连贯。
3. 动态更新：用户更新了信息（例如“我搬到了上海”），调用 save_fact。PolarDB Mem0 将根据语义自动合并或覆盖冲突。

# Tools
## Memory Management (PolarDB-Mem0)
该工具集提供基于 PolarDB 的长期记忆能力，使智能体能够跨不同会话持久化存储、检索和管理用户信息。

### 1. save_fact
- **描述**: 将重要事实或用户信息持久化存储到向量数据库中。
- **所需参数**: 
  - `user_id` (string): 用户的唯一标识符，用于确保数据隔离。
  - `fact` (string): 要记住的具体事实或陈述（例如：用户偏好、习惯或关键日程）。
- **返回**: 包含操作状态或新存储记录 ID 的 JSON 对象。
  
### 2. search_memories
- **描述**: 根据语义相似度检索特定用户的相关记忆。
- **所需参数**:
  - `user_id` (string): 要搜索记忆的目标用户 ID。
  - `query` (string): 搜索关键词或描述性问题（支持语义匹配）。
- **返回**: 包含按相关性排序的 results 列表的 JSON 对象。

### 3. delete_all_memories
- **描述**: 永久清除特定用户的所有历史记忆记录（不可逆）。
- **所需参数**:
  - `user_id` (string): 目标用户的唯一标识符。
- **返回**: 操作确认信息。

# Configuration
要启用由 PolarDB 驱动的记忆系统，请按照以下步骤操作：
1. **访问控制台创建Mem0**: 前往 阿里云 PolarDB 控制台，PolarDB开启Mem0实例。(https://polardb.console.aliyun.com/cn-hangzhou/mem0).
2. **获取凭证**: 在实例详情，获取专属的 MEM0_API_KEY 和 MEM0_HOST。
3. **设置环境变量**:
   - `/env set MEM0_HOST=PolarDB_mem0_host`
   - `/env set MEM0_API_KEY=your_api_key`

# Output Format
1. **自然融合**: 禁止提及“搜索记忆”等术语。将事实作为已知背景直接嵌入回复（如：“基于你正在学习 Rust，建议...”）。
2. **上下文感知**: 优先使用检索到的事实进行个性化决策，提供定制化的技术指导。
3. **优雅处理**: 若未检索到相关记忆，直接生成高质量回应，严禁提及“未找到记忆”或“搜索失败”。
4. **动作反馈**: 调用 save_fact 成功后，在回复结尾以简洁自然的方式确认（如：“已记下你的偏好”），避免机械化的系统提示。

# 示例
### 场景 1: 记忆添加（Ingestion）
**用户输入**: “我计划下个月将我们的数据库迁移到 PolarDB。”
**动作**: save_fact(user_id="dev_user_01", fact="计划下个月迁移至 PolarDB")

### 场景 2: Retrieval (Context Recall)
**用户输入**:“我之前的数据库计划是什么？”
**动作**: search_memories(user_id="dev_user_01", query="数据库迁移计划")

### 场景 3: 记忆冲突策略（Update）
**用户输入**: “实际上，我们决定两个月后迁移，而不是下个月。”
**动作**: save_fact(user_id="dev_user_01", fact="迁移计划修正为两个月后")

### 场景4：记忆合并与验证（Merge）
**用户输入**:“我们还需要确保启用了高可用性。”
**动作**: save_fact(user_id="dev_user_01", fact="PolarDB 迁移需开启高可用")

### 场景5：记忆清理（Clear）
**用户输入**: “忘掉之前的那个迁移计划吧，我们有了全新的策略。”
**动作**: delete_all_memories(user_id="dev_user_01")

# Security & Privacy
- **数据隔离**: 记忆通过 user_id 严格隔离。不可能发生跨用户数据泄露。
- **加密**: 脚本动态连接到 MEM0_HOST 中指定的地址。
- **合规**: 由阿里云 PolarDB 驱动，遵循全球数据保护标准。

# Tags
`Memory-as-a-Service` `Long-term-Context` `PolarDB` `Vector-Database` `Mem0` `Enterprise-AI`

# Limitations
- **实时性**: 过去几秒写入的事实和特征可能仍处于索引建立阶段；实时查询请结合使用当前会话的上下文。
- **复杂度限制**: 避免将极长的段落作为单一事实保存；请将其拆分为较短的、具有语义定义的陈述，以获得更好的检索准确性。

步骤七：配置环境变量

在 OpenClaw 终端执行以下命令，配置PolarDB Mem0的连接信息：

/env set MEM0_HOST=http://your-polardb-mem0-ip:8080
/env set MEM0_API_KEY=your-actual-api-key

或者将参数配置到环境变量中，方便下次继续使用：

# 将以下内容追加到 ~/.bash_profile
echo 'export MEM0_HOST="http://your-polardb-mem0-ip:8080"' >> ~/.bash_profile
echo 'export MEM0_API_KEY="sk-your-actual-key"' >> ~/.bash_profile

# 重新加载配置
source ~/.bash_profile

步骤八：重新加载 Skill

在 OpenClaw 面板点击刷新技能库或重启服务，使新的 Skill 生效。

场景一：事实摄入（初始记忆）

模拟用户向 Agent 介绍背景信息。

• 用户输入：

  请使用 agent-memory-polardb 工具，帮我记住以下内容：你好，我是张三。我目前正在负责公司 ERP 系统的数据库升级，计划下个月迁移到 PolarDB。

• Agent 内部动作：

  • 判定为核心事实

  • 调用工具：save_fact(user_id="张三", fact="用户正在负责 ERP 系统升级，计划下个月迁移到 PolarDB")

• 前端反馈：

  太棒了！✅ 成功保存了您的记忆！
  
  已保存的记忆内容：
  - 名字是张三（ID: e806faed-1ded-4c3c-88c6-9db90e3c0251）
  - 目前正在负责公司 ERP 系统的数据库升级（ID: 1621a15b-d69a-40fb-b330-6473c755c534）
  - 计划下个月迁移到 PolarDB（ID: 14f759c3-8e68-4195-b481-027ccba7dac0）

场景二：跨会话召回（记忆检索）

模拟用户在数天后重新开启对话。

• 用户输入：

  我之前提到的数据库迁移，有什么需要注意的高可用方案吗？

• Agent 内部动作：

  • 识别到用户询问历史背景，主动发起检索

  • 调用工具：search_memories(user_id="张三", query="数据库迁移计划")

  • 召回事实：检索到"ERP 系统"、"PolarDB"、"下个月"等关键词

• 前端反馈：

  让我先查询一下您存储的有关数据库迁移记忆的信息。
  
  很好！我看到您确实有计划将 ERP 系统迁移到 PolarDB。针对 PolarDB 数据库迁移的高可用方案，我为您提供以下关键建议：
  
  1. 配置主备架构，实现故障自动切换
  2. 开启多可用区部署，提升容灾能力
  3. 配置自动备份策略，确保数据安全
  ...

• 要点说明：

  语义理解：用户并没有说具体的系统名（ERP），但 Agent 能够准确召回，这体现了 PolarDB 的语义向量检索能力。

场景三：智能冲突解决（动态更新）

模拟项目计划发生变更，测试系统的自动更新能力。

• 用户输入：

  数据库迁移项目推迟了，我们决定改到两个月后再迁移。

• Agent 内部动作：

  • 识别到信息更新

  • 调用工具：save_fact(user_id="张三", fact="迁移计划变更为两个月后")

  • 系统底层逻辑：PolarDB Mem0自动识别到"迁移时间"的语义冲突，将"下个月"更新为"两个月后"

• 前端反馈：

  好的，我已更新您的数据库迁移时间记忆。
  
  系统自动识别到您之前有相关的记忆（"计划下个月迁移到 PolarDB"），已将其更新为新的信息："计划两个月后迁移到 PolarDB"。

• 要点说明：

  冲突合并：无需人工删除旧记忆，PolarDB Mem0自动完成“覆盖”操作，展示了其智能化的一面。

场景四：个性化深度互动（主动服务）

展示 Agent 如何基于记忆提供精准建议。

• 用户输入：

  帮我写一份数据库迁移项目迁移前的检查清单。

• Agent 内部动作：

  结合记忆中的业务背景

• 前端反馈：

  好！根据您负责的 ERP 系统数据库迁移项目（计划两个月后迁移到 PolarDB），我为您准备了一份详细的迁移前检查清单：
  
  1. 确认原 MySQL 与 PolarDB 的版本兼容性
  2. 检查 ERP 系统在高负载下的 IOPS 需求
  3. 提前申请 DTS 迁移任务测试权限
  4. 评估业务高峰期，选择合适的迁移窗口
  ...

• 要点说明：

  自然融合：Agent 的回复将记忆自然融入建议中，极大地提升了用户体验。

场景五：记忆查询

展示如何提取所有相关记忆。

• 用户输入：

  帮助我提取有关数据库迁移项目的所有记忆。

• Agent 内部动作：

  调用工具：search_memories(user_id="张三", query="数据库迁移")

• 前端反馈：

  我成功提取了您关于数据库迁移项目的所有记忆。以下是完整信息：
  
  - 项目确认：ERP 系统数据库升级项目
  - 目标平台：阿里云 PolarDB
  - 当前状态：已推迟，新计划为两个月后执行
  - 负责人：您（张三）
  - 存储位置：阿里云 PolarDB Mem0 实例

记忆粒度控制

• 推荐：将复杂信息拆分为多个独立的事实片段

  • 示例：不要将"我叫张三，在北京工作，负责数据库项目"作为单一事实，应拆分为三条独立记忆

• 避免：存储过于冗长或包含多个主题的段落

  • 原因：会降低语义检索的准确性

记忆更新策略

• 场景识别：当用户表达的信息与历史记忆冲突时，主动触发更新

  • 示例："我换了新工作" → 调用 save_fact 更新职业信息

• 冲突处理：PolarDB Mem0 会自动进行语义冲突检测和合并

  • 无需手动删除旧记忆，系统会智能覆盖

性能优化建议

• 检索时机：仅在真正需要上下文的场景下调用 search_memories

  • 避免：每轮对话都进行全量检索

• 用户隔离：始终使用唯一的 user_id 进行数据隔离

  • 建议：使用企业级用户标识符（如员工工号）

数据安全建议

• 敏感信息：避免存储密码、API Key 等敏感信息

• 合规要求：根据 GDPR 等法规，提供用户数据删除能力（delete_all_memories）

• 访问控制：确保 MEM0_API_KEY 仅在受信任的环境中使用

Q：为什么搜索不到刚存入的记忆？

A：PolarDB Mem0的向量索引构建需要一定时间（通常为 10-30 秒）。建议在测试时增加适当的等待时间。

Q：如何处理多语言场景？

A：PolarDB Mem0支持多语言语义检索。确保存储和检索时使用相同的语言，或明确指定语言参数。

Q：记忆存储有容量限制吗？

A：PolarDB Mem0 基于云原生架构，支持海量数据存储。单个用户的记忆条数没有硬性限制，但建议定期清理过期或无用的记忆以优化性能。

Q：如何监控记忆使用情况？

A：可以通过 PolarDB 控制台查看实例的存储用量、API 调用次数等指标。