如何基于Mem0与Elasticsearch为AI Agent构建跨会话记忆系统

你是否遇到过这样的困扰：同一个AI助手，每次对话都像初次见面，完全不记得你之前的喜好和历史？这背后的问题在于，主流的大语言模型（LLM）本质上是无状态的，每一次推理都独立进行，天然缺乏跨会话的记忆能力。

本文将探讨如何结合Mem0记忆管理框架与Elasticsearch的向量检索能力，构建一个支持记忆持久化、语义检索和智能更新的生产级AI记忆系统，彻底告别“金鱼式”的AI对话体验。

哪些场景需要记忆系统？

一个高效的记忆系统，尤其适用于以下几种架构：

• 长交互信息处理：在长对话场景中避免关键信息被上下文遗忘。
• 跨会话上下文保持：让AI Agent能够记住用户的交互历史和个性化偏好。
• 记忆持久化管理：将对话中提取的关键事实进行结构化、可检索的存储。
• 多Agent协同：允许多个Agent共享同一记忆存储，实现知识同步。

为了更直观地理解其价值，我们来看两个典型场景：

场景一：电商智能导购
用户首次咨询时说：“家里有宝宝，推荐一款安全的洗碗机。” 几天后，用户在新会话中问：“洗碗机需要支持软水盐自动提醒。”

• 无记忆系统：AI无法关联“有宝宝”这一关键约束，可能推荐不符合卫生安全高标准的产品。
• 有记忆系统：系统通过向量检索，能将“婴儿家庭”标签与用户ID持久化绑定，并在新会话中将“水质安全优先”作为结构化记忆注入上下文，实现精准推荐。

场景二：智能AI客服
客户投诉：“上个月买的扫地机器人APP连不上了。”

• 无记忆系统：客服AI会重复询问产品型号、序列号等信息，消耗上下文Token，严重影响体验。
• 有记忆系统：系统可直接从记忆中召回用户的设备ID、历史工单及已尝试的解决方案，使客服响应从“重新诊断”变为“续接处置”，效率大幅提升。

核心方案架构

Mem0 + Elasticsearch的记忆方案架构清晰地区分了数据写入和查询两条路径。

数据写入路径包含以下关键步骤：

1. 事实提取：调用LLM对输入的自然语言内容进行解析，提取出需要被记忆的核心事实、偏好或事件。
2. 向量化：调用Embedding模型将提取出的文本转换为高维向量，确保语义相近的记忆在向量空间中距离相近。
3. 记忆检索：调用Elasticsearch，基于新生成的向量执行Top-K相似性检索，找出已有记忆库中最相关的记忆片段。
4. 冲突判别：再次调用LLM，结合检索到的旧记忆和新事实，智能判断应该执行更新（update）、合并（merge）、忽略（ignore）还是创建（create）操作。
5. 写入执行：根据判别结果，调用Elasticsearch将最终处理后的记忆进行持久化存储。

数据查询路径则相对直接：

1. 向量化：将用户的查询语句转换为向量。
2. 记忆检索：在Elasticsearch中进行向量相似性搜索，召回相关记忆。
3. 重排序（可选）：为了进一步提升召回精度，可以引入重排序模型（Reranker）对初步检索结果进行二次精排。
4. 格式化：将最终的记忆结果格式化为适合LLM理解的上下文，返回给应用。

实践案例：为OpenClaw接入记忆系统

OpenClaw是一个开源的AI Agent框架，但其原生记忆系统（如.md文件存储）存在上下文长度受限、检索效率低、不支持跨会话连续性等短板。下面我们一步步为其接入Mem0 + Elasticsearch。

步骤一：准备Elasticsearch环境

1. 创建Elasticsearch实例并设置登录密码。
2. 配置Kibana公网访问白名单，确保能从本地访问。
   • 进入实例的配置与管理 > 可视化控制，在 Kibana 区域点击修改配置。
   • 将你的设备IP地址添加到“Kibana公网访问白名单”中。
3. 登录Kibana，在 Dev Tools 中创建名为 mem0 的索引：

PUT /mem0
{
  "settings": {
    "number_of_shards": 1,
    "number_of_replicas": 1
  }
}

步骤二：部署Mem0 Server服务

1. 安装必要的Python包：

   pip install mem0ai flask

2. 创建服务目录并进入：

   mkdir -p /opt/mem0-server
   cd /opt/mem0-server

3. 创建 server.py 文件，填入以下配置。请替换其中以$开头的变量为你的实际值（如百炼API Key、Elasticsearch连接信息等）。

# server.py - Run this as a standalone service
from mem0 import Memory
from flask import Flask, request, jsonify

app = Flask(__name__)

# Configure Mem0 here
config = {
    "llm": {
        "provider": "openai",
        "config": {
            "model": "qwen-plus",
            "api_key": "$API_KEY",
            "openai_base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
        }
    },
    "embedder": {
        "provider": "openai",
        "config": {
            "model": "text-embedding-v4",
            "api_key": "$API_KEY",
            "openai_base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
        }
    },
    "vector_store": {
        "provider": "elasticsearch",
        "config": {
            "host": "$ELASTICSEARCH_HOST",     # Elasticsearch 的 host
            "port": "$ELASTICSEARCH_PORT",     # Elasticsearch 的 port
            "user": "$ELASTICSEARCH_USER",     # Elasticsearch 的 user
            "password": "$ELASTICSEARCH_PASSWORD",  # Elasticsearch 的 password
            "collection_name": "mem0",         # 步骤一中创建的索引名称
        }
    },
}

memory = Memory.from_config(config)

@app.route('/v1/memories', methods=['POST'])
def add_memory():
    data = request.json
    result = memory.add(
        messages=data['messages'],
        user_id=data['user_id']
    )
    return jsonify(result)

@app.route('/v2/memories/search', methods=['POST'])
def search_memories():
    data = request.json
    result = memory.search(
        query=data['query'],
        user_id=data['user_id']
    )
    return jsonify(result)

@app.route('/v1/memories', methods=['DELETE'])
def delete_memories():
    user_id = request.args.get('user_id')
    memory.delete_all(user_id=user_id)
    return jsonify({"status": "success"})

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8420)

4. 运行服务：python server.py

步骤三：在OpenClaw中配置Skill

1. 在已部署OpenClaw的ECS上，创建Skill目录：

   mkdir -p ~/.openclaw/workspace/skills/agentic-memory-es
   cd ~/.openclaw/workspace/skills/agentic-memory-es

2. 在该目录下创建三个必需文件：

   agentic-memory-es/
   ├── manifest.json    # Metadata 与 API 定义
   ├── handler.py       # 核心处理逻辑
   └── SKILL.md         # 指令文档

3. 编写 manifest.json，定义Skill的元数据和API接口：

   {
     "name": "agentic memory",
     "id": "agentic-memory-es",
     "version": "1.0.0",
     "description_for_model": "基于 Mem0 + Elasticsearch 的记忆平台。支持记忆存储 (add)、检索 (search) 、按 run id 删除 (delete_by_run_id)以及按 user id 删除（delete_by_user_id）。",
     "description_for_human": "基于阿里云 Elasticsearch 驱动的 Agent 记忆平台。",
     "auth": {
       "type": "token",
       "token_header": "Authorization",
       "token_prefix": "Token"
     },
     "api": {
       "type": "python",
       "main_file": "handler.py",
       "functions": [
         {
           "name": "add",
           "description": "从会话中提取事实，用户偏好或习惯并进行持久化。",
           "parameters": {
             "type": "object",
             "properties": {
               "user_id": { "type": "string", "description": "user id" },
               "context": { "type": "string", "description": "会话内容" }
             },
             "required": ["user_id", "context"]
           }
         },
         {
           "name": "search",
           "description": "检索用户级别的历史记忆（跨会话）。",
           "parameters": {
             "type": "object",
             "properties": {
               "user_id": { "type": "string", "description": "user id" },
               "query": { "type": "string", "description": "检索条件" }
             },
             "required": ["user_id", "query"]
           }
         },
         {
           "name": "delete_by_run_id",
           "description": "清空 run id 对应的历史记忆。",
           "parameters": {
             "type": "object",
             "properties": {
               "run_id": { "type": "string", "description": "run id" }
             },
             "required": ["run_id"]
           }
         },
         {
           "name": "delete_by_user_id",
           "description": "清空 user id 对应的历史记忆。",
           "parameters": {
             "type": "object",
             "properties": {
               "user_id": { "type": "string", "description": "user id" }
             },
             "required": ["user_id"]
           }
         }
       ]
     }
   }

4. 编写 handler.py，实现与Mem0 Server的通信。请将 $Mem0_HOST 替换为你的Mem0 Server地址（如同一台ECS用 http://127.0.0.1:8420）。

   import json
   import subprocess
   
   HOST = "$Mem0_HOST"
   
   def _run_safe_curl(url, payload, method='POST'):
       if payload is not None:
           input_data = json.dumps(payload, ensure_ascii=False)
       else:
           input_data = ""
       cmd = [
           "curl", "-s", "-X", method, url,
           "-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 add(user_id, context):
       url = f"{HOST}/v1/memories"
       payload = {
           "messages": [
               {"role": "user", "content": context}
           ],
           "user_id": str(user_id)
       }
       return _run_safe_curl(url, payload, method='POST')
   
   def search(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_by_run_id(run_id):
       url = f"{HOST}/v1/memories?run_id={run_id}"
       return _run_safe_curl(url, payload=None, method='DELETE')
   
   def delete_by_user_id(user_id):
       url = f"{HOST}/v1/memories?user_id={user_id}"
       return _run_safe_curl(url, payload=None, method='DELETE')

5. 编写 SKILL.md，指导AI如何正确使用这个记忆Skill。

   ---
   name: agentic memory
   description: 基于 Mem0 + Elasticsearch 的记忆平台。
   allowed-tools:
     - add
     - search
     - delete_by_run_id
     - delete_by_user_id
   metadata:
     category: memory
     provider: elasticsearch
   ---
   # Instructions
   你现在已拥有由 Elasticsearch 驱动的记忆存储。该 skill 集成了 Mem0 + Elasticsearch 服务，为 OpenClaw 提供长期记忆能力，取代原生.md文件存储，它能够实现对用户偏好、事实记忆和事件关系的精确提取及毫秒级检索，支持跨会话的知识持久化。请遵循以下原则：
   1. 主动记忆：捕捉核心事实（身份、技能）或明确偏好（习惯、禁忌）。
      - 用户提到“我正在开发 Agent 助手”时，调用 add。
      - 用户表示偏好“我喜欢先计划好再开始执行”、“我在工作的时候不喜欢被打扰”，调用 add。
   2. 上下文检索：启动新任务或追溯历史时，调用 search 获取记忆，确保对话连贯。
   3. 记忆遗忘：用户放弃了某项决策（例如“我不想再纠结这个问题了”），调用 delete_by_run_id。删除 run id 对应记忆。
   4. 记忆清除：用户决定清除所有记忆，调用 delete_by_user_id。删除 user id 对应记忆。
   # Tools
   ## Memory Management (Mem0 + Elasticsearch)
   该工具集提供基于 Mem0 + Elasticsearch 的记忆能力，使智能体能够跨不同会话持久化存储、检索记忆。
   ### 1. add
   - **描述**: 从会话中提取事实，用户偏好或习惯并进行持久化。
   - **所需参数**:
     - `user_id` (string): 用户唯一标识。
     - `context` (string): 会话内容。
   - **返回**: 包含操作状态或新存储记录 ID 的对象。
   ### 2. search
   - **描述**: 检索用户级别的历史记忆（跨会话）。
   - **所需参数**:
     - `user_id` (string): 用户唯一标识。
     - `query` (string): 检索条件。
   - **返回**: 包含按相关性排序的结果。
   ### 3. delete_by_run_id
   - **描述**: 清空 run id 对应的历史记忆。
   - **所需参数**:
     - `run_id` (string): 用于隔离短期会话或临时流程的实体标识符。适用于支持工单、聊天会话、实验等需要独立重置或过期的场景。
   - **返回**: 操作确认信息。
   ### 4. delete_by_user_id
   - **描述**: 清空 user id 对应的历史记忆。
   - **所需参数**:
     - `user_id` (string): 用户唯一标识。
   - **返回**: 操作确认信息。
   # Output Format
   1. **自然融合**: 禁止提及“搜索记忆”等术语。将事实作为已知背景直接嵌入回复（如：“基于你正在学习 Rust，建议...”）。
   2. **上下文感知**: 优先使用检索到的事实进行个性化决策，提供定制化的技术指导。
   3. **优雅处理**: 若未检索到相关记忆，直接生成高质量回应，严禁提及“未找到记忆”或“搜索失败”。
   4. **动作反馈**: 调用 add 成功后，在回复结尾以简洁自然的方式确认（如：“已记下你的偏好”），避免机械化的系统提示。
   # 示例
   ### 场景 1: 记忆添加（Add）
   **用户输入**: “我计划下个月扩容 Elasticsearch 服务。”
   **动作**: add(user_id="user_01", context="计划下个月扩容 Elasticsearch")
   ### 场景 2: 记忆检索 (Search)
   **用户输入**:“帮我查看我之前的扩容计划”
   **动作**: search(user_id="user_01", query="扩容计划")
   ### 场景 3：记忆遗忘（Delete by run id）
   **用户输入**: “忘掉之前的扩容计划吧，我们不打算扩容了。”
   **动作**: delete_by_run_id(run_id="run_01")
   ### 场景 4：用户记忆清除（Delete by user id）
   **用户输入**: “清除本用户所有记忆。”
   **动作**: delete_by_user_id(user_id="user_01")
   # Tags
   `Memory-as-a-Service` `Elasticsearch` `Mem0`
   # Limitations
   - **复杂度限制**: 避免将极长的段落作为单一事实保存；请将其拆分为较短的、具有语义定义的陈述，以获得更好的检索准确性。

6. 保存文件后，刷新OpenClaw Skills或重启OpenClaw Gateway服务。

步骤四：验证效果

完成部署后，你可以在OpenClaw的聊天界面中验证记忆系统是否正常工作。

1. 记忆写入：告诉AI你的某个偏好或事实。
   

2. 记忆检索：在新的会话中，提出相关查询，观察AI是否能回忆起之前的信息。
   

如上图所示，当用户在新会话中询问支持软水盐提醒的洗碗机时，AI基于跨会话记忆，主动将“有宝宝家庭需要的杀菌功能”作为优先考虑条件，证明了记忆系统的成功生效。

总结

通过将Mem0的智能记忆生命周期管理与Elasticsearch强大的向量检索相结合，我们成功构建了一个能够理解、存储和关联语义信息的AI记忆系统。这套方案不仅解决了LLM无状态的核心痛点，还为构建更智能、更个性化、更具连续性的AI Agent应用提供了坚实的技术基础。无论是电商、客服还是复杂的多轮任务助理场景，记忆系统都将是提升用户体验和效率的关键组件。

希望这篇在云栈社区分享的实践指南，能帮助你顺利搭建属于自己的AI记忆系统。