Hologres基于Mem0框架和向量检索实现大模型长记忆，持久化存储对话历史，实现跨会话长期记忆。 - 实时数仓Hologres

随着大语言模型在智能客服、虚拟助手等场景中的广泛应用，其上下文长度限制带来的"短期记忆"问题日益凸显。Hologres 基于 Mem0 框架，结合高性能向量检索能力，推出大模型长记忆方案。本文介绍如何基于 Mem0 + Hologres 构建大模型长记忆系统。

方案简介

Hologres Mem0 方案充分利用 Hologres 作为实时数仓的高并发写入与更新、毫秒级向量检索等能力，构建一个专用于存储和管理用户个性化记忆的向量引擎。

在该方案中，用户的对话历史、偏好设置、关键事实等信息被自动提取、向量化并持久化存储到 Hologres 表中。当用户发起新请求时，系统会根据当前查询语义，从 Hologres 中实时检索最相关的记忆片段，并将其动态拼接到大模型的输入上下文中，从而在不增加模型负担的前提下，显著提升模型对历史信息的利用能力，实现跨会话、跨时间的"长记忆"效果，构建更智能、更可靠的企业级 AI 应用。

使用 Hologres 作为 Mem0 长记忆方案的优势如下：

高性能向量检索：依托 Hologres 内置的向量引擎，支持亿级向量毫秒级检索，满足高并发在线推理场景需求。
实时写入与更新：支持每秒数万条记忆记录的实时写入，确保用户最新行为能即时生效，避免记忆滞后。
低成本高可用：复用 Hologres 作为统一数据平台，一份数据实现向量检索、全文检索、OLAP 分析、在线服务，降低运维复杂度与总体拥有成本。
开箱即用&开源开放：基于开源 Mem0 框架开发，代码已公开，快速上手、灵活定制。

方案架构

Hologres Mem0 方案主要包含以下核心组件：

记忆提取器（Memory Extractor）
集成在应用服务中，负责监听用户与大模型的交互过程，识别并提取具有长期价值的信息（如用户偏好、事实陈述、任务目标等），生成结构化的记忆条目。
向量化引擎（Embedding Engine）
使用文本嵌入模型将记忆条目转换为高维向量，确保语义相似的记忆在向量空间中距离相近。
Hologres 向量存储层
Hologres 作为核心记忆库，提供：
- 支持向量字段、文本字段、标量字段混合存储的表结构。
- 高性能 HGraph 向量索引。
- 毫秒级向量相似度检索。
- 多租户隔离与权限控制。
- 向量 + 全文 + 标量字段混合检索。
记忆检索器（Memory Retriever）
在每次大模型推理前，将用户当前查询向量化，并在 Hologres 中执行 Top-K 相似向量检索，返回最相关的记忆片段。
上下文融合器（Context Integrator）
将检索到的记忆按相关性排序后，以自然语言形式拼接到 Prompt 中，形成增强后的上下文输入给大模型。

整个流程闭环运行，无需人工干预，可无缝集成到现有 LLM 应用架构中。

实践步骤

Mem0 支持 OpenAI、Gemini、DeepSeek 等多种大语言模型，本文实践基于 Mem0 框架 + 阿里云大模型服务平台百炼 + Hologres 构建长记忆系统。实践步骤如下。

获取示例代码：mem0_hologres。
连接 Hologres 实例，创建数据库 mem0。
```
CREATE DATABASE mem0;
```

打开示例代码所在目录，创建、激活虚拟环境，并安装所需的依赖库。

cd /home/mem0_hologres
python3.11 -m venv myenv
source myenv/bin/activate
pip install -e .
pip install "psycopg[pool]" / pip install psycopg2-binary

阿里云大模型服务平台百炼为开发者提供了兼容 OpenAI 的 API 及全链路模型服务，本文提供以下两个示例用于测试。百炼 API Key 获取方式。

示例一：基础记忆操作

from mem0 import Memory

config = {
    "llm" : {
        "provider": "openai",
        "config": {"openai_base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
                   "api_key": "<YOUR_DASHSCOPE_API_KEY>",
                   "model": "qwen-plus"},
    },
    "embedder": {
        "provider": "openai",
        "config": {"openai_base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
                   "api_key": "<YOUR_DASHSCOPE_API_KEY>",
                   "model": "text-embedding-v4", "embedding_dims": 1536},
    },
    "vector_store": {
        "provider": "hologres",
        "config": {
            "host": "<hologres-host>",
            "port": <hologres-port>,
            "dbname": "mem0",
            "user": "<hologres-username>",
            "password": "<hologres-password>",
            "collection_name": "memories",
        },
    }
}

m = Memory.from_config(config_dict=config)

debug = True
# Using only user_id
result = m.add("I like pizza", user_id="alice")
if debug:
    print(result)

result = m.add("I hate pizza", user_id="alice")
if debug:
    print(result)

result = m.add("I like football", user_id="alice", metadata={"category": "hobbies"})
if debug:
    print(result)

# Using both user_id and agent_id
result = m.add("I like pizza", user_id="alice", agent_id="food-assistant")
if debug:
    print(result)

# Get all memories for a user
result = m.get_all(user_id="alice")
if debug:
    print(result)

# Get all memories for a specific agent belonging to a user
result = m.get_all(user_id="alice", agent_id="food-assistant")
if debug:
    print(result)

# Search memories for a user
result = m.search("tell me my name.", user_id="alice")
if debug:
    print(result)

# Search memories for a specific agent belonging to a user
result = m.search("tell me my name.", user_id="alice", agent_id="food-assistant")
if debug:
    print(result)

# Search memories for a user
result = m.search("what food do you like.", user_id="alice")
if debug:
    print(result)

# Search memories for a user
result = m.search("what sport do you like.", user_id="alice")
if debug:
    print(result)

# Delete all memories for a user
result = m.delete_all(user_id="alice")
if debug:
    print(result)

# Delete all memories for a specific agent belonging to a user
result = m.delete_all(user_id="alice", agent_id="food-assistant")
if debug:
    print(result)

输出：

{'results': [{'id': 'c62cba4d-2261-4399-9ef2-9f2af8537615', 'memory': 'Likes pizza', 'event': 'ADD'}]}
{'results': [{'id': 'c62cba4d-2261-4399-9ef2-9f2af8537615', 'memory': 'Likes pizza', 'event': 'DELETE'}]}
{'results': [{'id': 'd5099c77-0458-4012-a2be-98b2960a6159', 'memory': 'Likes football', 'event': 'ADD'}]}
{'results': [{'id': 'f4b3a456-b9d4-435b-ae0f-08d0b32630ac', 'memory': 'Likes pizza', 'event': 'ADD'}]}
...
Resetting index memories...
{'message': 'Memories deleted successfully!'}
Resetting index memories...
{'message': 'Memories deleted successfully!'}

示例二：对话记忆与语义检索

import json
from mem0 import Memory

config = {
    "llm" : {
        "provider": "openai",
        "config": {"openai_base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
                   "api_key": "<YOUR_DASHSCOPE_API_KEY>",
                   "model": "qwen-plus",  "temperature": 0.2, "max_tokens": 2000, "top_p": 1.0},
    },
    "embedder": {
        "provider": "openai",
        "config": {"openai_base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
                   "api_key": "<YOUR_DASHSCOPE_API_KEY>",
                   "model": "text-embedding-v4", "embedding_dims": 1536},
    },
    "vector_store": {
        "provider": "hologres",
        "config": {
            "host": "<hologres-host>",
            "port": <hologres-port>,
            "dbname": "mem0",
            "user": "<hologres-username>",
            "password": "<hologres-password>",
            "collection_name": "memories",
        },
    }
}

m = Memory.from_config(config_dict=config)

# 添加对话记忆
messages = [
    {"role": "user", "content": "我喜欢科幻电影，尤其是星际穿越。"},
    {"role": "assistant", "content": "星际穿越确实是一部经典的科幻电影！我会记住你喜欢这类电影。"},
    {"role": "user", "content": "我也喜欢克里斯托弗·诺兰的其他电影。"},
    {"role": "assistant", "content": "诺兰的作品确实很出色！除了《星际穿越》，他还导演了《盗梦空间》、《信条》、《记忆��片》等经典作品。您对这些电影有什么特别的看法吗？"},
]

# 添加记忆，并附加元数据
m.add(messages, user_id="alice", metadata={"category": "movies", "tags": ["sci-fi", "preferences"], "importance": "high"})

# 检索记忆
memories = m.search(query="帮我推荐喜欢的电影", user_id="alice", filters={"category": "movies"})
print(json.dumps(memories, indent=2, ensure_ascii=False))

# 获取所有记忆
all_memories = m.get_all(user_id="alice")
print(json.dumps(all_memories, indent=2, ensure_ascii=False))

# Delete all memories for a user
result = m.delete_all(user_id="alice")
print(result)

输出：

{
  "results": [
    {
      "id": "3b84616f-f705-4b92-be2d-4374ce3644f9",
      "memory": "尤其喜欢星际穿越",
      "score": 0.634332,
      "user_id": "alice"
    },
    {
      "id": "210f505f-4d6f-4b0c-ae59-6b55bf2b7cc1",
      "memory": "喜欢科幻电影",
      "score": 0.604246,
      "user_id": "alice"
    },
    {
      "id": "59265409-1c75-4b49-b2b4-efa8a9af4869",
      "memory": "喜欢克里斯托弗·诺兰的其他电影",
      "score": 0.490508,
      "user_id": "alice"
    }
  ]
}