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

跟 ChatGPT 对话，比跟真人社交还累！真人好歹能记住你名字吧？

想象一下——你昨天刚把沙发位置、爆米花口味、爱看的电影都告诉了 ChatGPT，而它永远是那个热情又健忘的助理，下次再对话还是会问：“哦？是吗？那太好了！请问您对什么类型的电影感兴趣呢？”

受够了这种单方面的“社牛”表演？Mem0 来了，专治 AI 失忆症，给你的“金鱼脑”助理装个大容量硬盘，让你们下次见面，能直接跳过多余的问答，从“好久不见”开始。

Mem0 是为 AI 智能体开发打造的记忆层。它就像一个持久的“大脑”，能帮助 AI 智能体完成以下内容：

• 随时调取历史对话，追溯关键信息

• 精准记住用户的个人偏好与重要事实

• 在实践中总结经验，不断自我完善

git地址：https://github.com/mem0ai/mem0

记忆层的作用

如下图所示，无记忆层的情况下，即使 LLM 有超大的上下文窗口的情况下，再开一个新会话后上下文都会被重置。有记忆层 Mem0 的情况下，将保留上下文，召回需要的内容，并持续优化自身存储。

记忆层在 AI 智能体开发中的作用

如下图所示，Mem0 会与检索器(RAG)、LLM 、上下文并肩工作。与传统的基于检索的系统（如 RAG）不同，Mem0 会记录过往交互、保存长期知识，并让智能体的行为随时间而进化。仅会将记忆中相关的知识合并到 prompt 之中，输入给 LLM。

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

下面是 Mem0 记忆层和使用 LLM 上下文窗口的主要区别：

记忆层和 RAG 对比

以 Mem0 为代表的记忆层与传统 RAG 对比有以下区别：

• 实体关联：理解并跨会话关联人物、主题，而非仅检索静态文档。

• 记忆策略：优先近期、高相关记忆，旧信息自动衰减。

• 会话连续：长期保留上下文，使得虚拟伴侣、学习助手等场景更连贯。

• 持续学习：根据用户反馈实时微调，个性化随时间更精准。

• 动态更新：新交互即时写入记忆，无需重新索引文档。

Mem0 核心流程

Mem0 的核心工作流程包括以下步骤：

1. 语义捕获：利用 LLM 对会话流进行智能解析，自动捕获并抽象出具备长期价值的核心语义信息。

2. 内容向量化：通过嵌入模型将这些语义信息编码为高维度的向量，为后续的相似度计算和高效检索奠定基础。

3. 向量存储：将上一步生成的向量存储至向量数据库中，该数据库需要支持大规模、低延迟的语义搜索，在后面的例子中我们将使用阿里云 Milvus。

4. 检索：系统接收到新的用户输入后，会立即在向量空间中进行语义相似度匹配，精准地调用出与当前情境最关联的历史记忆。

5. 上下文增强：将调用出的历史记忆注入到当前的推理链路中，与现有上下文相结合，从而生成逻辑更连贯、内容更具个性化的响应。

基本原理与架构概述

Milvus 是专为向量相似性搜索设计的分布式数据库，其核心基于以下关键技术：

• 近似最近邻搜索（ANN）：通过 HNSW、IVF、PQ 等算法实现高效向量检索，平衡精度与速度。

• 向量索引与查询分离：支持动态构建多种索引类型（如FLAT、IVF_FLAT、IVF_PQ、HNSW），适配不同场景需求。

• 向量数据分片与分布式计算：数据水平切分（Sharding）并行处理，实现高吞吐与低延迟。

采用云原生和存算分离的微服务架构。该架构分为接入、协调、执行和存储四层。各组件可独立扩展，确保了系统的高性能、高可用性和弹性。它依赖成熟的第三方组件（如 etcd、对象存储）进行数据和元数据管理，稳定可靠。

阿里云 Milvus 系统架构图

使用场景

阿里云 Milvus 适用于任何需要进行“相似性”匹配的场景。其核心应用包括：

1. 图像视频搜索：如电商平台的以图搜图、安防领域的人脸识别和视频轨迹追踪。

2. 文本语义搜索：构建智能客服、企业内部文档知识库和代码搜索引擎，能精准理解用户意图，而非简单的关键词匹配。

3. 个性化推荐系统：根据用户的行为和偏好向量，实时推荐最相似的商品、音乐、新闻或视频。

4. 前沿科学与安全：在生物信息学中加速药物分子筛选，或在网络安全领域进行异常流量和欺诈行为检测。

5. 智能驾驶数据准备与挖掘：对点云图像、车载传感器收集的音视频等多模态数据进行向量数据的实时查询。

更多介绍：https://www.aliyun.com/product/milvus

接下来，本教程将通过两个示例，带你实践如何结合 Mem0 与 Milvus 实现：

• 构建具备长期记忆的 AI Agent

• 利用图谱引擎与向量引擎协同分析信息间的复杂关联。

• 已创建阿里云 Milvus 实例。具体操作，请参见 快速创建 Milvus 实例 [1]。

• 已开通服务并获得 API-KEY。具体操作，请参见 开通 DashScope 并创建 API-KEY [2]。

LangGraph 是一个业界成熟的用于构建有状态和多角色的Agents 应用的框架。限于篇幅将不对 LangGraph 过多介绍，可以参考官方文档 [3]。

• 依赖库安装

pip install langgraph langchain-openai mem0ai

• 核心代码

包含以下核心步骤：

1. 环境变量设置 OpenAI 方式访问百炼 qwen 大模型；LLM 设置 qwen-plus 作为语言大模型；Mem0 配置 qwen-plus 作为语义识别和处理大模型、使用 text-embedding-v3 作为 embedding 模型、使用 Milvus 作为向量存储数据库。

2. 设置 LangGraph 会话状态，用于获取对话上下文。

3. 对话 Agent 开发，使用 Mem0 的 search 接口获取相关的记忆、使用 Mem0 的 add 接口存储相关记忆到向量库 Milvus 中。

4. 编排 LangGraph，设置节点和边。

5. 设置 LangGraph 流式输出。

6. 入口 main 函数进行人机交互。

from typing import Annotated, TypedDict, Listfrom langgraph.graph import StateGraph, STARTfrom langgraph.graph.message import add_messagesfrom langchain_openai import ChatOpenAIfrom mem0 import Memoryimport osfrom langchain_core.messages import SystemMessage, HumanMessage, AIMessage# 设置环境变量，百炼平台qwen模型的key和baseurlos.environ["OPENAI_API_KEY"] = "sk-xx"os.environ["OPENAI_BASE_URL"] = "https://dashscope.aliyuncs.com/compatible-mode/v1"# LLM设置llm = ChatOpenAI(model="qwen-plus", temperature=0.2, max_tokens=2000)# Mem0设置，LLM、embedding和向量库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)# 设置LangGraph对话状态class State(TypedDict):    messages: Annotated[List[HumanMessage | AIMessage], add_messages]    mem0_user_id: strgraph = StateGraph(State)# 对话Agent开发，包含Mem0记忆读取和记忆存储def chatbot(state: State):    messages = state["messages"]    user_id = state["mem0_user_id"]    try:        # Retrieve relevant memories        memories = mem0.search(messages[-1].content, user_id=user_id,)        # Handle dict response format        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)        # Store the interaction in Mem0        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}")        # Fallback response without memory context        response = llm.invoke(messages)        return {"messages": [response]}# 设置LangGraph调度节点和边graph.add_node("chatbot", chatbot)graph.add_edge(START, "chatbot")graph.add_edge("chatbot", "chatbot")compiled_graph = graph.compile()# 设置LangGraph流式输出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"  # You can generate or retrieve this based on your user management system    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 说了不喜欢惊悚片，LLM 最终根据我们的要求推荐了一些合适的影片。

第一次提问

可以看到了一些 Memory saved 的打印，查看 Milvus 向量库，可以看到对应的 collection 已经有了几个 Entity。

点开数据页面，可以看到 Mem0 已将上下文经过 LLM 处理概括地保存到 metadata 字段中，并且对应的用户是 alice，数据为了可以被检索也已经被向量化存储到 vectors 字段中。

重新执行代码，因为我们已经有了记忆的存在，再问一下“我喜欢什么电影”，可以看到 Mem0 从 Milvus 中召回了相关的内容，并将内容合并到了 prompt 中发送给 LLM，我们得到了相关的电影推荐而不需要再和 LLM 重复介绍我们的喜好。

方案概述

Mem0 支持图谱记忆（Graph Memory）。借助图谱记忆，用户可以创建并利用信息之间的复杂关系，从而生成更细致、更具上下文感知能力的响应。这一融合使用户能够同时发挥向量检索与图谱技术的优势，实现更准确、更全面的信息检索与内容生成。

记忆层添加记录的方式如下图所示，Mem0 通过 LLM 提取内容后，通过添加或者更新的方式，同时将内容 embedding 到向量库和提取实体&关系到图谱数据库中。

记忆层添加记录

记忆层检索记录的方式如下图所示，Mem0 通过 LLM 提取内容后，同时将内容 embedding 到向量库检索和提取实体&关系到图谱数据库中检索，双路检索后将结果合并输出。

记忆层检索记录

前提条件

• 已创建阿里云 Milvus 实例。具体操作，请参见 快速创建 Milvus 实例 [1]。

• 已开通服务并获得 API-KEY。具体操作，请参见 开通 DashScope 并创建 API-KEY [2]。

代码开发

• 依赖库安装

pip install kuzu rank-bm25 mem0ai

• 核心代码

包含以下核心步骤：

1. 环境变量设置 OpenAI 方式访问百炼 qwen 大模型；LLM 设置 qwen-plus 作为语言大模型；Mem0 配置 qwen-plus 作为语义识别和处理大模型、使用 text-embedding-v3 作为 embedding 模型、使用 Milvus 作为向量存储数据库、使用 kuzu 作为图谱数据库。

2. 初始化 Mem0，添加数据，将同时添加内容到向量库和图谱库中。

3. 提问测试。

from langchain_openai import ChatOpenAIfrom mem0 import Memory# 设置环境变量，百炼平台qwen模型的key和baseurlos.environ["OPENAI_API_KEY"] = "sk-xx"os.environ["OPENAI_BASE_URL"] = "https://dashscope.aliyuncs.com/compatible-mode/v1"# LLM设置llm = ChatOpenAI(model="qwen-plus", temperature=0.2, max_tokens=2000)# Mem0设置，LLM、embedding和向量库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",}# 初始化Mem0，添加数据，将同时添加内容到向量库和图谱库中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")# 按照score分数倒序排列，输出结果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"))

验证效果

如下图所示，是“我的名字是什么？”的返回，可以看到 results 中是向量返回，返回的“名字是爱丽丝”得分并不高，relations 中是图谱返回，解析出了我的名字是“爱丽丝”，关系为 has_name。

如下图所示，是“谁是蜘蛛侠？”的返回，可以看到 results 中是向量返回，返回的“朋友彼得是蜘蛛侠”得分最低，relations 中是图谱返回，解析出了蜘蛛侠的名字是“彼得”，关系为是。

通过以上两个例子，可以发现，有图谱能力的加持，可以补齐向量库缺失信息之间深层分析的短板。

随着 AI 应用越来越深入日常生活，系统对用户上下文和历史信息的理解变得尤为重要。Mem0 与 Milvus 的结合，为人工智能提供了一套高效、可扩展的长时记忆解决方案。通过向量数据库持久化存储语义记忆，AI 不仅能记住过去的交互，还能在后续对话中持续调用和更新这些信息。这一能力让智能助手、客服机器人等应用更加连贯、个性化和实用