Mem0 + Milvus：为人工智能构建持久化长时记忆

与 ChatGPT 对话常让人感到疲惫，因其无法记住用户信息，每次交互都需重复说明。例如，即便你已告知偏好电影类型和沙发位置，下次对话时它仍会重新提问。

这种“健忘”限制了 AI 助手的实用性。Mem0 的出现正是为解决这一问题——它为 AI 智能体提供持久化记忆能力，让对话无需反复铺垫，真正实现个性化连续交互。

Mem0 介绍

Mem0 是专为 AI 智能体设计的记忆层，赋予其长期记忆与持续学习能力，使其能够：

• 追溯历史对话中的关键信息
• 精准记忆用户的个人偏好与事实
• 基于交互经验不断优化响应行为

项目地址：https://github.com/mem0ai/mem0

记忆层的作用

传统 LLM 虽具备大上下文窗口，但新会话开启后上下文即被重置。Mem0 提供持久化记忆机制，可保留并召回历史内容，持续优化存储结构。

Mem0 在 AI 智能体架构中协同 RAG、LLM 和上下文模块工作。不同于仅检索静态文档的传统系统，Mem0 记录交互历史、保存长期知识，并推动智能体行为随时间进化，仅将相关记忆注入 prompt 输入给 LLM。

Mem0在AI智能体中的处理方式

Mem0 与 LLM 上下文窗口对比

能力	LLM上下文窗口	Mem0 记忆存储
记忆力	临时	持久化
token 消耗	随输入增加而增长	优化（仅提取所需内容）
内容召回	依赖 LLM 对长文本的理解能力	压缩上下文，更贴近用户意图
个性化	无	记录历史会话

记忆层与 RAG 对比

Mem0 类记忆层相较于传统 RAG 具备以下优势：

• 实体关联：跨会话识别并关联人物、主题，而非简单检索文档。
• 记忆策略：优先调用近期及高相关性记忆，旧信息自动衰减。
• 会话连续：支持长期上下文保留，适用于虚拟伴侣、学习助手等场景。
• 持续学习：根据反馈实时调整，个性化程度随时间提升。
• 动态更新：新交互即时写入记忆，无需重新索引。

Mem0 核心流程

Mem0 工作流程包括五个关键步骤：

1. 语义捕获：利用 LLM 解析会话流，提取具有长期价值的核心语义信息。
2. 内容向量化：通过嵌入模型将语义编码为高维向量，支撑高效检索。
3. 向量存储：将向量存入支持大规模低延迟搜索的向量数据库（如阿里云 Milvus）。
4. 检索：基于用户新输入，在向量空间中进行语义匹配，召回最相关记忆。
5. 上下文增强：将历史记忆融合至当前推理链路，生成更具连贯性与个性化的响应。

阿里云Milvus基本原理介绍

基本原理与架构概述

Milvus 是专为向量相似性搜索构建的分布式数据库，核心技术包括：

• 近似最近邻搜索（ANN）：采用 HNSW、IVF、PQ 等算法平衡检索精度与速度。
• 索引与查询分离：支持多种索引类型（FLAT、IVF_FLAT、IVF_PQ、HNSW），适配不同场景。
• 数据分片与分布式计算：通过水平切分实现高吞吐、低延迟。

其采用云原生存算分离架构，分为接入、协调、执行和存储四层，各组件可独立扩展，保障高性能、高可用与弹性伸缩。系统依赖 etcd、对象存储等成熟组件管理元数据与数据，确保稳定性。

阿里云 Milvus 系统架构图

使用场景

阿里云 Milvus 广泛应用于需“相似性”匹配的领域：

• 图像视频搜索：电商平台以图搜图、人脸识别、视频轨迹追踪。
• 文本语义搜索：智能客服、企业知识库、代码搜索引擎，理解意图而非关键词匹配。
• 个性化推荐：基于用户行为向量实时推荐商品、音乐、新闻等内容。
• 前沿科研与安全：药物分子筛选、异常流量检测、欺诈识别。
• 智能驾驶：对点云、音视频等多模态数据进行向量化实时查询。

更多详情请访问：https://www.aliyun.com/product/milvus

实践一：有记忆的AI Agent开发流程

前提条件

• 已创建阿里云 Milvus 实例。
• 已开通服务并获取 API-KEY。

代码开发

LangGraph 是用于构建有状态 Agents 应用的成熟框架，本文基于该框架集成 Mem0 与 Milvus。

• 依赖库安装

pip install langgraph langchain-openai mem0ai

• 核心实现步骤

1. 配置环境变量，通过 OpenAI 接口调用百炼 Qwen 大模型；设置 qwen-plus 为语言模型，text-embedding-v3 为 embedding 模型，Milvus 为向量数据库。

2. 定义 LangGraph 会话状态，用于维护对话上下文。

3. 开发对话 Agent，调用 Mem0 的 search 接口获取记忆，add 接口将交互存入 Milvus。

4. 编排 LangGraph 节点与边。

5. 启用流式输出功能。

6. 实现 main 函数作为人机交互入口。

from typing import Annotated, TypedDict, List
from langgraph.graph import StateGraph, START
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from mem0 import Memory
import os
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

os.environ["OPENAI_API_KEY"] = "sk-xx"
os.environ["OPENAI_BASE_URL"] = "https://dashscope.aliyuncs.com/compatible-mode/v1"

llm = ChatOpenAI(model="qwen-plus", temperature=0.2, max_tokens=2000)

config = {
  "llm": {
    "provider": "openai",
    "config": {
      "model": "qwen-plus",
      "temperature": 0.2,
      "max_tokens": 2000,
    }
  },
  "embedder": {
    "provider": "openai",
    "config": {
      "model": "text-embedding-v3",
      "embedding_dims": 128,
    }
  },
  "vector_store": {
    "provider": "milvus",
    "config": {
      "collection_name": "mem0_test1",
      "embedding_model_dims": "128",
      "url": "http://c-xxx.milvus.aliyuncs.com:19530",
      "token": "root:xxx",
      "db_name": "default",
    }
  },
  "version": "v1.1",
}
mem0 = Memory.from_config(config)

class State(TypedDict):
  messages: Annotated[List[HumanMessage | AIMessage], add_messages]
  mem0_user_id: str

graph = StateGraph(State)

def chatbot(state: State):
  messages = state["messages"]
  user_id = state["mem0_user_id"]
  try:
    memories = mem0.search(messages[-1].content, user_id=user_id)
    memory_list = memories['results']
    context = "Relevant information from previous conversations:\n"
    for memory in memory_list:
      context += f"- {memory['memory']}\n"
    system_message = SystemMessage(content=f"""You are a helpful customer support assistant. Use the provided context to personalize your responses and remember user preferences and past interactions.
{context}""")
    full_messages = [system_message] + messages
    print(full_messages)
    response = llm.invoke(full_messages)
    try:
      interaction = [
        {"role": "user", "content": messages[-1].content},
        {"role": "assistant", "content": response.content}
      ]
      result = mem0.add(interaction, user_id=user_id)
      print(f"Memory saved: {len(result.get('results', []))} memories added")
    except Exception as e:
      print(f"Error saving memory: {e}")
    return {"messages": [response]}
  except Exception as e:
    print(f"Error in chatbot: {e}")
    response = llm.invoke(messages)
    return {"messages": [response]}

graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", "chatbot")
compiled_graph = graph.compile()

def run_conversation(user_input: str, mem0_user_id: str):
  config = {"configurable": {"thread_id": mem0_user_id}}
  state = {"messages": [HumanMessage(content=user_input)], "mem0_user_id": mem0_user_id}
  for event in compiled_graph.stream(state, config):
    for value in event.values():
      if value.get("messages"):
        print("Customer Support:", value["messages"][-1].content)
        return

if __name__ == "__main__":
  print("Welcome to Customer Support! How can I assist you today?")
  mem0_user_id = "alice"
  while True:
    user_input = input("You: ")
    if user_input.lower() in ['quit', 'exit', 'bye']:
      print("Customer Support: Thank you for contacting us. Have a great day!")
      break
    run_conversation(user_input, mem0_user_id)

验证效果

首次运行时无上下文记忆，当用户表达对电影的偏好（如不喜欢惊悚片）后，LLM 可据此推荐合适影片。

首次提问

控制台显示“Memory saved”，查看 Milvus 实例可见对应 collection 已生成多个 Entity。

进入数据页面可见，Mem0 已将上下文经 LLM 抽象后存入 metadata 字段，用户标识为 alice，同时向量化内容存入 vectors 字段以支持检索。

再次运行程序并提问“我喜欢什么电影”，Mem0 成功从 Milvus 中召回记忆，合并至 prompt 发送给 LLM，无需重复说明即可获得个性化推荐。

实践二：通过图谱引擎+向量引擎解析信息之间复杂关系

方案概述

Mem0 支持图谱记忆（Graph Memory），结合向量检索与图谱技术，挖掘信息间的深层关联，生成更精准、上下文感知更强的响应。

添加记忆时，Mem0 利用 LLM 提取内容，同时将语义向量化存入向量库，并提取实体与关系存入图谱数据库。

记忆层添加记录

检索时，Mem0 并行执行向量检索与图谱检索，整合结果后输出，提升信息完整性与准确性。

记忆层检索记录

前提条件

• 已创建阿里云 Milvus 实例。
• 已开通服务并获取 API-KEY。

代码开发

• 依赖库安装

pip install kuzu rank-bm25 mem0ai

• 核心实现步骤

1. 配置环境变量，调用百炼 Qwen 模型；设置 qwen-plus 为语言模型，text-embedding-v3 为 embedding 模型，Milvus 为向量数据库，kuzu 为图谱数据库。

2. 初始化 Mem0，添加数据时同步写入向量库与图谱库。

3. 执行查询测试。

from langchain_openai import ChatOpenAI
from mem0 import Memory
import os
import json

os.environ["OPENAI_API_KEY"] = "sk-xx"
os.environ["OPENAI_BASE_URL"] = "https://dashscope.aliyuncs.com/compatible-mode/v1"

llm = ChatOpenAI(model="qwen-plus", temperature=0.2, max_tokens=2000)

config = {
  "llm": {
    "provider": "openai",
    "config": {
      "model": "qwen-plus",
      "temperature": 0.2,
      "max_tokens": 2000,
    }
  },
  "embedder": {
    "provider": "openai",
    "config": {
      "model": "text-embedding-v3",
      "embedding_dims": 128,
    }
  },
  "vector_store": {
    "provider": "milvus",
    "config": {
      "collection_name": "mem0_test3",
      "embedding_model_dims": "128",
      "url": "http://c-xxx.milvus.aliyuncs.com:19530",
      "token": "root:xxx",
      "db_name": "default",
    }
  },
  "graph_store": {
    "provider": "kuzu",
    "config": {
      "db": "./mem0-example.kuzu"
    }
  },
  "version": "v1.1",
}

m = Memory.from_config(config)
m.add("我喜欢去徒步旅行", user_id="alice123")
m.add("我喜欢打羽毛球", user_id="alice123")
m.add("我讨厌打羽毛球", user_id="alice123")
m.add("我的朋友叫约翰，约翰有一只叫汤米的狗", user_id="alice123")
m.add("我的名字是爱丽丝", user_id="alice123")
m.add("约翰喜欢徒步旅行，哈利也喜欢徒步旅行", user_id="alice123")
m.add("我的朋友彼得是蜘蛛侠", user_id="alice123")

def get_res(res):
  sorted_results = sorted(res['results'], key=lambda x: x['score'], reverse=True)
  res['results'] = sorted_results
  print(json.dumps(res, ensure_ascii=False, indent=2))

get_res(m.search("我的名字是什么？", user_id="alice123"))
get_res(m.search("谁是蜘蛛侠？", user_id="alice123"))

验证效果

查询“我的名字是什么？”返回结果显示：向量检索部分，“名字是爱丽丝”得分较低；图谱检索部分则准确解析出“爱丽丝”为姓名，关系为 has_name。

查询“谁是蜘蛛侠？”返回中，向量检索结果“朋友彼得是蜘蛛侠”得分最低，而图谱检索明确识别出“彼得”是蜘蛛侠，关系为“是”。

上述案例表明，图谱能力有效弥补了纯向量检索在深层次关系分析上的不足。

随着 AI 深入日常生活，系统对用户上下文与历史信息的理解愈发重要。Mem0 联合 Milvus 提供了一套高效、可扩展的长时记忆解决方案。通过向量数据库持久化语义记忆，AI 不仅能记住过往交互，还能持续调用与更新信息，显著提升智能助手、客服机器人等应用的连贯性、个性化与实用性。