Haystack Mem0MemoryStore 深度解析基于 Mem0 云端 API 构建 Agent 长期记忆存储【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystackMem0MemoryStore是 Haystack 中一个以 Mem0 云端 API 为后端的记忆存储memory store组件负责把对话中的事实、偏好与项目上下文持久化为可检索的长期记忆。本文以 v2.19 实验版 API 参考 为骨架结合当前仓库中 Mem0 集成mem0memorystore.mdx、Mem0MemoryRetriever、Mem0MemoryWriter、Mem0 Memory Tools的完整文档逐方法讲解其初始化、写入、检索、删除与过滤器转换机制并给出可运行的实战示例。读完本文你将掌握如何在 Haystack 中用它为多用户、多会话的 Agent 系统注入可跨对话持久化的长期记忆。一、组件定位记忆存储家族中的云端实现在 Haystack 的记忆存储memory store体系中Mem0MemoryStore与CogneeMemoryStore是两个基于外部服务的实现前者以Mem0 云端 API为后端后者以 Cognee 知识图谱 API 为后端。Mem0MemoryStore是共享的数据层被三类上层组件共同使用Mem0MemoryRetriever在 Agent 或 Chat Generator 之前检索长期记忆返回ChatMessage列表Mem0MemoryWriter在 Agent 或 Chat Generator 之后把本轮对话写入长期记忆Mem0MemoryRetrieverTool/Mem0MemoryWriterTool作为 Agent 可直接调用的工具让模型自己决定何时读取、何时存储。从 v2.19 的 API 参考看该组件最初以实验形态出现在haystack_experimental包中模块路径为haystack_experimental.memory_stores.mem0.memory_store在当前版本的文档中它已随mem0-haystack集成包发布导入路径为haystack_integrations.memory_stores.mem0.Mem0MemoryStore。两种形态的类名与核心方法签名保持一致。二、安装与 API Key 配置在使用前需要安装 Mem0 集成并配置云端 API Keypip install mem0-haystackexport MEM0_API_KEYyour-mem0-api-keyMem0MemoryStore的构造函数只接受一个关键字参数api_key类型为Secret默认值是从环境变量MEM0_API_KEY读取def __init__(*, api_key: Secret Secret.from_env_var(MEM0_API_KEY))这意味着你可以直接Mem0MemoryStore()无参实例化此时依赖已导出的环境变量也可以显式传入Secret.from_token(...)覆盖默认值。将敏感信息封装为Secret对象是 Haystack 的标准做法——序列化时密钥不会以明文写入to_dict()结果。序列化与反序列化组件实现了两个生命周期方法便于将配置保存到 YAML/JSON 管线定义或从其中恢复def to_dict() - dict[str, Any] classmethod def from_dict(cls, data: dict[str, Any]) - Mem0MemoryStoreto_dict()把存储配置序列化为字典from_dict(data)从字典反序列化出新的Mem0MemoryStore实例。这两个方法是 Haystack 组件可序列化协议的一部分保证Mem0MemoryStore可以安全地嵌入 Pipeline 的 YAML 序列化机制 与Pipeline.loads()/dumps()流程中。三、写入记忆add_memoriesadd_memories是记忆的入口把一组ChatMessage作为记忆写入 Mem0def add_memories(*, messages: list[ChatMessage], infer: bool True, user_id: str | None None, run_id: str | None None, agent_id: str | None None, async_mode: bool False, **kwargs: Any) - list[dict[str, Any]]各参数语义如下参数类型默认值说明messageslist[ChatMessage]必填携带记忆元数据的 ChatMessage 对象列表inferboolTrue是否让 Mem0 从消息中自动推断事实。为False时整条消息文本原样作为记忆存储user_idstr \| NoneNone用户 ID用于限定记忆的存储与检索范围run_idstr \| NoneNone运行会话ID用于限定记忆范围agent_idstr \| NoneNoneAgent ID。若希望 Mem0 存储来自 assistant 的聊天消息必须设置该值async_modeboolFalse是否异步写入。为True时方法立即返回记忆在后台添加**kwargsAny—透传给Mem0 client.add的附加参数关于infer的取舍官方文档给出了明确的使用场景建议inferTrue默认让 Mem0 自动从消息中抽取记忆适合存储完整的 Agent 回合用户消息 工具上下文 最终回答inferFalse把消息文本原样入库适合上游组件已经挑选好精确记忆文本的场景例如直接存储ChatMessage.from_user(Alice prefers concise Python examples.)。注意两个易错点逐条消息元数据会被忽略Mem0 不允许在消息列表中为每条消息单独传 metadata因此messages中每条ChatMessage.meta的内容不会生效。若要为整条记忆附加元数据必须通过**kwargs里的metadata关键字传给 Mem0 client。assistant 消息需要agent_id只有设置了agent_idMem0 才会存储来自 assistant 的聊天消息否则可能只处理用户侧内容。返回值是“带有memory_id与记忆内容的对象”列表便于调用方跟踪每次写入生成的记忆 ID。异步写入模式当async_modeTrue时add_memories会立刻返回实际写入在后台完成。这在长对话场景下可以避免主链路阻塞——例如 Agent 的回合结束后用后台任务把整轮对话异步沉淀为记忆而不延长用户的等待时间。注意此时返回值不再代表“已写入完成”的记忆列表调用方不应依赖其同步结果做后续校验。四、检索记忆search_memories 与 search_memories_as_single_messagesearch_memoriessearch_memories是记忆的读取入口返回匹配条件的ChatMessage列表def search_memories(*, query: str | None None, filters: dict[str, Any] | None None, top_k: int 5, user_id: str | None None, run_id: str | None None, agent_id: str | None None, include_memory_metadata: bool False, **kwargs: Any) - list[ChatMessage]核心语义query文本查询。不传时返回所有记忆此时走get_all语义不做相关性检索filtersHaystack 风格的过滤器用于限定检索范围。作用域规则如果提供了filters检索只以过滤器为界user_id/run_id/agent_id会被忽略如果未提供过滤器则user_id、run_id、agent_id三者至少设置一个否则无法确定检索范围top_k返回的最大结果数默认5include_memory_metadata是否把 Mem0 侧的检索元数据memory_id、score等并入返回的ChatMessage。为True时这些字段会写入ChatMessage.meta的mem0_memory_metadata键下为False时ChatMessage.meta只包含用户自定义的元数据**kwargs透传给 Mem0 client。分流规则传了query则传给client.search方法未传query则传给client.get_all方法。search_memories_as_single_message该方法是search_memories的便捷变体把检索结果合并成单个ChatMessage返回def search_memories_as_single_message(*, query: str | None None, filters: dict[str, Any] | None None, top_k: int 5, user_id: str | None None, run_id: str | None None, agent_id: str | None None, **kwargs: Any) - ChatMessage参数语义与search_memories基本一致作用域规则相同无过滤器时至少设置一个 IDkwargs按query是否存在分流给search或get_all。区别只在返回值形态它把多条记忆折叠进一条ChatMessage适合直接作为 system 消息拼进模型的输入消息列表——例如在 Agent 的messages开头插入一条包含全部已检索记忆的消息模型无需处理多条结构。检索时的元数据约定结合 Mem0MemoryRetriever 的文档约定用户自定义的 Mem0 元数据会进入每条返回消息的meta而 Mem0 检索字段如memory_id、user_id、score、时间戳会放在meta[mem0]键下。这与 v2.19 API 参考中include_memory_metadata的语义一致——需要注意 API 参考中该键名为mem0_memory_metadata集成层文档中展示为mem0不同版本间键名可能演进读取时以所装版本为准。五、删除记忆delete_all_memories 与 delete_memory记忆的生命周期管理提供两个删除方法def delete_all_memories(*, user_id: str | None None, run_id: str | None None, agent_id: str | None None, **kwargs: Any) - None按作用域批量删除记忆记录user_id、run_id、agent_id至少设置一个kwargs透传给Mem0 client.delete_all方法。def delete_memory(memory_id: str, **kwargs: Any) - None按memory_id精确删除单条记忆该 ID 来自add_memories的返回值或检索结果中的memory_idkwargs透传给Mem0 client.delete方法。这两个方法让开发者可以实现“记忆遗忘”例如用户主动清除历史数据、或 Agent 判断某条记忆已过期时可按 ID 定点删除或按用户/会话整体清空避免长期记忆库无限膨胀或残留过期事实。六、过滤器转换normalize_filtersstaticmethod def normalize_filters(filters: dict[str, Any]) - dict[str, Any]这是一个静态方法负责把Haystack 风格的过滤器转换为 Mem0 风格的过滤器。Haystack 有自己的过滤器 DSL逻辑操作符如AND/OR/NOT、比较操作符如、、in等而 Mem0 client 需要其原生过滤器格式该方法是两者之间的适配桥梁在search_memories内部对filters参数做归一化处理。这意味着上层代码可以沿用 Haystack 统一的过滤语法编写查询条件而无需关心 Mem0 的具体过滤结构同时search_memories_as_single_message中提到的“更多 Mem0 过滤器细节”则直接透传两种过滤能力可以互补使用。七、作用域机制user_id / run_id / agent_idMem0 的记忆是多租户作用域化的。user_id、run_id、agent_id三个实体 ID 用于窄化读写范围它们都是运行时参数而非构造参数——因此单个Mem0MemoryStore实例可以服务多个用户、多个会话或多个 Agent无需为每个租户创建独立存储对象user_id最常见的粒度按用户隔离记忆run_id按会话/运行隔离适合同一用户的不同会话互不干扰agent_id按 Agent 隔离同时是存储 assistant 消息的前提条件。当前版本集成文档中还提到app_id作为第四个作用域实体可参考 mem0memorystore.mdxv2.19 实验版 API 参考的方法签名中仅显式列出user_id、run_id、agent_id三个参数其余实体通过**kwargs透传。八、实战示例独立使用 Mem0MemoryStore基础写入与检索from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore() store.add_memories( messages[ChatMessage.from_user(Alice prefers concise Python examples.)], user_idalice, inferFalse, # 精确文本直接入库不做事实抽取 ) memories store.search_memories( queryWhat does Alice prefer?, user_idalice, top_k3, ) print([msg.text for msg in memories])多实体 ID 复合作用域store.add_memories( messages[ ChatMessage.from_user(Alice is working on a documentation search system.) ], user_idalice, run_iddocs-assistant-session-1, inferTrue, # 让 Mem0 从消息中抽取事实 ) memories store.search_memories( queryWhat project is Alice working on?, user_idalice, run_iddocs-assistant-session-1, ) print([msg.text for msg in memories])复合作用域让同一用户的记忆可以进一步按会话细分避免跨会话的上下文污染。检索作用域内全部记忆传queryNone即可不做相关性检索返回作用域内的全部记忆all_memories store.search_memories(queryNone, user_idalice) print([msg.text for msg in all_memories])这在 Agent 需要“先盘点已知记忆、再决定是否精确检索”的场景中非常有用详见下文 Tools 部分。九、组合成 Agent 记忆管线Mem0MemoryStore的价值在组合使用时体现得最充分。以下模式来自 Mem0MemoryRetriever 与 Mem0MemoryWriter 文档展示了三种典型用法。模式一Pipeline 中“先检索、再回答”通过OutputAdapter把检索到的记忆与当前用户消息拼接注入 Agentfrom haystack import Pipeline from haystack.components.agents import Agent from haystack.components.converters import OutputAdapter from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.generators.utils import print_streaming_chunk from haystack.dataclasses import ChatMessage from haystack_integrations.components.retrievers.mem0 import Mem0MemoryRetriever from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore() pipeline Pipeline() pipeline.add_component(retriever, Mem0MemoryRetriever(memory_storestore, top_k5)) pipeline.add_component( memory_context, OutputAdapter( template{{ memories user_messages }}, output_typelist[ChatMessage], unsafeTrue, ), ) pipeline.add_component( agent, Agent( chat_generatorOpenAIChatGenerator(modelgpt-4o-mini), system_prompt( Use any system messages at the start of the conversation as long-term memory. Answer concisely. ), streaming_callbackprint_streaming_chunk, ), ) pipeline.connect(retriever.memories, memory_context.memories) pipeline.connect(memory_context.output, agent.messages) query Give me a short implementation tip. pipeline.run( { retriever: {query: query, user_id: alice}, memory_context: {user_messages: [ChatMessage.from_user(query)]}, }, )Mem0MemoryRetriever的关键运行语义与 store 一致query为字符串时做相关性检索并应用top_kqueryNone时返回作用域内全部记忆检索结果以systemChatMessage形态输出到memories输出端。模式二Pipeline 中“回合结束、写入记忆”把 Agent 完整输出的messages接到Mem0MemoryWriter用inferTrue让 Mem0 从整轮对话中抽取事实from haystack import Pipeline from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.generators.utils import print_streaming_chunk from haystack.dataclasses import ChatMessage from haystack_integrations.components.writers.mem0 import Mem0MemoryWriter from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore() pipeline Pipeline() pipeline.add_component( agent, Agent( chat_generatorOpenAIChatGenerator(modelgpt-4o-mini), system_prompt( Answer the user and preserve durable user facts or preferences for future conversations. ), streaming_callbackprint_streaming_chunk, ), ) pipeline.add_component(writer, Mem0MemoryWriter(memory_storestore, inferTrue)) pipeline.connect(agent.messages, writer.messages) result pipeline.run( { agent: { messages: [ ChatMessage.from_user(My name is Alice and I prefer concise Python examples.), ], }, writer: {user_id: alice}, }, ) print(result[writer][memories_written])Mem0MemoryWriter输出端为memories_written写入的记忆条数。写入同样需要至少一个作用域 IDuser_id/run_id/agent_id且infer的语义与add_memories完全一致——写整轮 Agent 回合用inferTrue上游已选定精确文本时用inferFalse。模式三让 Agent 自己决定读写Mem0 Memory ToolsMem0 Memory Tools 提供Mem0MemoryRetrieverTool默认工具名retrieve_memories与Mem0MemoryWriterTool默认工具名store_memory二者共享同一个Mem0MemoryStore。默认情况下工具通过inputs_from_state{user_id: user_id}从 Agent State 注入user_id这样用户 ID 不会作为工具参数暴露给 LLM一个 Agent 实例即可服务多用户from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.generators.utils import print_streaming_chunk from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore from haystack_integrations.tools.mem0 import ( Mem0MemoryRetrieverTool, Mem0MemoryWriterTool, ) store Mem0MemoryStore() retrieve_memories Mem0MemoryRetrieverTool(memory_storestore, top_k10) store_memory Mem0MemoryWriterTool(memory_storestore) agent Agent( chat_generatorOpenAIChatGenerator(modelgpt-5.4), tools[retrieve_memories, store_memory], system_promptYou are a helpful assistant with long-term memory. At the beginning of each turn, call retrieve_memories without a query to inspect known memories. Use store_memory only for new durable user-specific facts, preferences, or project context. Before storing, compare the proposed memory with retrieved memories and avoid duplicates. Do not store transient requests that are only useful in the current conversation. , streaming_callbackprint_streaming_chunk, state_schema{user_id: {type: str}}, ) result agent.run( messages[ ChatMessage.from_user( My name is Alice. Please remember that I prefer concise Python examples., ), ], user_idalice, )工具的参数约定Mem0MemoryRetrieverToolmemory_store必填top_k可选默认5name默认retrieve_memoriesdescription、parameters可选用于覆盖暴露给 LLM 的 JSON Schemainputs_from_state默认{user_id: user_id}。工具向 LLM 暴露query与top_k当 Agent 省略query或传null时返回注入作用域内的全部记忆Mem0MemoryWriterToolmemory_store必填name默认store_memory其余可选参数同上。工具向 LLM 暴露text与infer且默认inferFalse保证 Agent 存储的是它自己选定的精确文本。若需要注入更多作用域 ID如把 State 中的session_id映射到工具的run_id参数可扩展state_schema并用inputs_from_state{user_id: user_id, session_id: run_id}完成映射无需让 LLM 感知这些内部字段。十、从实验包到集成包的演进v2.19 的 experimental_mem0_memory_store_api.md 展示了该组件的实验形态位于haystack_experimental包中接口以haystack_experimental.memory_stores.mem0.memory_store.Mem0MemoryStore暴露。在当前仓库的主文档mem0memorystore.mdx中组件已迁移为正式集成mem0-haystack导入路径为haystack_integrations.memory_stores.mem0.Mem0MemoryStore并配套了 Retriever、Writer 与 Tools 的完整生态。核心方法add_memories、search_memories及作用域、infer等语义在两个版本间保持一致功能上当前集成还补充了app_id作用域、include_memory_metadata/mem0_memory_metadata元数据回传等增强。如果你是 v2.19 用户升级时只需调整导入路径与包依赖业务调用代码基本可平移。十一、小结与最佳实践综合 v2.19 API 参考与当前集成文档使用Mem0MemoryStore时建议遵循以下实践作用域先行任何读写操作都至少提供一个作用域 IDuser_id/run_id/agent_id或用filters限定范围无过滤器时三者缺一不可否则调用会失败按需选择infer写入完整 Agent 回合用inferTrue让 Mem0 抽取事实上游已确定精确文本时用inferFalse避免语义漂移善用queryNone需要盘点上下文时检索全部记忆需要精准命中时再提供query并配合top_k理解元数据流向写入时逐消息meta会被忽略改用kwargs中的metadata读取时用include_memory_metadata控制是否带回memory_id、score等 Mem0 检索元数据生命周期管理用delete_memory(memory_id)定点清理过期记忆用delete_all_memories(...)按用户/会话整体遗忘避免长期记忆库噪声累积组合而非孤立Mem0MemoryStore本身是数据层与Mem0MemoryRetriever、Mem0MemoryWriter或两个 Memory Tools 搭配才能构成“读取记忆 → 生成回答 → 沉淀新记忆”的完整长期记忆闭环。如需深入底层实现可继续阅读 ChatMessage 数据结构、Mem0MemoryRetriever、Mem0MemoryWriter 以及 Mem0 Memory Tools 的完整文档。【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/13 8:16:48 · 阅读 1.2k Haystack Mem0MemoryStore 深度解析：基于 Mem0 云端 API 构建 Agent 长期记忆存储
Haystack Mem0MemoryStore 深度解析基于 Mem0 云端 API 构建 Agent 长期记忆存储【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystackMem0MemoryStore是 Haystack 中一个以 Mem0 云端 API 为后端的记忆存储memory store组件负责把对话中的事实、偏好与项目上下文持久化为可检索的长期记忆。本文以 v2.19 实验版 API 参考 为骨架结合当前仓库中 Mem0 集成mem0memorystore.mdx、Mem0MemoryRetriever、Mem0MemoryWriter、Mem0 Memory Tools的完整文档逐方法讲解其初始化、写入、检索、删除与过滤器转换机制并给出可运行的实战示例。读完本文你将掌握如何在 Haystack 中用它为多用户、多会话的 Agent 系统注入可跨对话持久化的长期记忆。一、组件定位记忆存储家族中的云端实现在 Haystack 的记忆存储memory store体系中Mem0MemoryStore与CogneeMemoryStore是两个基于外部服务的实现前者以Mem0 云端 API为后端后者以 Cognee 知识图谱 API 为后端。Mem0MemoryStore是共享的数据层被三类上层组件共同使用Mem0MemoryRetriever在 Agent 或 Chat Generator 之前检索长期记忆返回ChatMessage列表Mem0MemoryWriter在 Agent 或 Chat Generator 之后把本轮对话写入长期记忆Mem0MemoryRetrieverTool/Mem0MemoryWriterTool作为 Agent 可直接调用的工具让模型自己决定何时读取、何时存储。从 v2.19 的 API 参考看该组件最初以实验形态出现在haystack_experimental包中模块路径为haystack_experimental.memory_stores.mem0.memory_store在当前版本的文档中它已随mem0-haystack集成包发布导入路径为haystack_integrations.memory_stores.mem0.Mem0MemoryStore。两种形态的类名与核心方法签名保持一致。二、安装与 API Key 配置在使用前需要安装 Mem0 集成并配置云端 API Keypip install mem0-haystackexport MEM0_API_KEYyour-mem0-api-keyMem0MemoryStore的构造函数只接受一个关键字参数api_key类型为Secret默认值是从环境变量MEM0_API_KEY读取def __init__(*, api_key: Secret Secret.from_env_var(MEM0_API_KEY))这意味着你可以直接Mem0MemoryStore()无参实例化此时依赖已导出的环境变量也可以显式传入Secret.from_token(...)覆盖默认值。将敏感信息封装为Secret对象是 Haystack 的标准做法——序列化时密钥不会以明文写入to_dict()结果。序列化与反序列化组件实现了两个生命周期方法便于将配置保存到 YAML/JSON 管线定义或从其中恢复def to_dict() - dict[str, Any] classmethod def from_dict(cls, data: dict[str, Any]) - Mem0MemoryStoreto_dict()把存储配置序列化为字典from_dict(data)从字典反序列化出新的Mem0MemoryStore实例。这两个方法是 Haystack 组件可序列化协议的一部分保证Mem0MemoryStore可以安全地嵌入 Pipeline 的 YAML 序列化机制 与Pipeline.loads()/dumps()流程中。三、写入记忆add_memoriesadd_memories是记忆的入口把一组ChatMessage作为记忆写入 Mem0def add_memories(*, messages: list[ChatMessage], infer: bool True, user_id: str | None None, run_id: str | None None, agent_id: str | None None, async_mode: bool False, **kwargs: Any) - list[dict[str, Any]]各参数语义如下参数类型默认值说明messageslist[ChatMessage]必填携带记忆元数据的 ChatMessage 对象列表inferboolTrue是否让 Mem0 从消息中自动推断事实。为False时整条消息文本原样作为记忆存储user_idstr \| NoneNone用户 ID用于限定记忆的存储与检索范围run_idstr \| NoneNone运行会话ID用于限定记忆范围agent_idstr \| NoneNoneAgent ID。若希望 Mem0 存储来自 assistant 的聊天消息必须设置该值async_modeboolFalse是否异步写入。为True时方法立即返回记忆在后台添加**kwargsAny—透传给Mem0 client.add的附加参数关于infer的取舍官方文档给出了明确的使用场景建议inferTrue默认让 Mem0 自动从消息中抽取记忆适合存储完整的 Agent 回合用户消息 工具上下文 最终回答inferFalse把消息文本原样入库适合上游组件已经挑选好精确记忆文本的场景例如直接存储ChatMessage.from_user(Alice prefers concise Python examples.)。注意两个易错点逐条消息元数据会被忽略Mem0 不允许在消息列表中为每条消息单独传 metadata因此messages中每条ChatMessage.meta的内容不会生效。若要为整条记忆附加元数据必须通过**kwargs里的metadata关键字传给 Mem0 client。assistant 消息需要agent_id只有设置了agent_idMem0 才会存储来自 assistant 的聊天消息否则可能只处理用户侧内容。返回值是“带有memory_id与记忆内容的对象”列表便于调用方跟踪每次写入生成的记忆 ID。异步写入模式当async_modeTrue时add_memories会立刻返回实际写入在后台完成。这在长对话场景下可以避免主链路阻塞——例如 Agent 的回合结束后用后台任务把整轮对话异步沉淀为记忆而不延长用户的等待时间。注意此时返回值不再代表“已写入完成”的记忆列表调用方不应依赖其同步结果做后续校验。四、检索记忆search_memories 与 search_memories_as_single_messagesearch_memoriessearch_memories是记忆的读取入口返回匹配条件的ChatMessage列表def search_memories(*, query: str | None None, filters: dict[str, Any] | None None, top_k: int 5, user_id: str | None None, run_id: str | None None, agent_id: str | None None, include_memory_metadata: bool False, **kwargs: Any) - list[ChatMessage]核心语义query文本查询。不传时返回所有记忆此时走get_all语义不做相关性检索filtersHaystack 风格的过滤器用于限定检索范围。作用域规则如果提供了filters检索只以过滤器为界user_id/run_id/agent_id会被忽略如果未提供过滤器则user_id、run_id、agent_id三者至少设置一个否则无法确定检索范围top_k返回的最大结果数默认5include_memory_metadata是否把 Mem0 侧的检索元数据memory_id、score等并入返回的ChatMessage。为True时这些字段会写入ChatMessage.meta的mem0_memory_metadata键下为False时ChatMessage.meta只包含用户自定义的元数据**kwargs透传给 Mem0 client。分流规则传了query则传给client.search方法未传query则传给client.get_all方法。search_memories_as_single_message该方法是search_memories的便捷变体把检索结果合并成单个ChatMessage返回def search_memories_as_single_message(*, query: str | None None, filters: dict[str, Any] | None None, top_k: int 5, user_id: str | None None, run_id: str | None None, agent_id: str | None None, **kwargs: Any) - ChatMessage参数语义与search_memories基本一致作用域规则相同无过滤器时至少设置一个 IDkwargs按query是否存在分流给search或get_all。区别只在返回值形态它把多条记忆折叠进一条ChatMessage适合直接作为 system 消息拼进模型的输入消息列表——例如在 Agent 的messages开头插入一条包含全部已检索记忆的消息模型无需处理多条结构。检索时的元数据约定结合 Mem0MemoryRetriever 的文档约定用户自定义的 Mem0 元数据会进入每条返回消息的meta而 Mem0 检索字段如memory_id、user_id、score、时间戳会放在meta[mem0]键下。这与 v2.19 API 参考中include_memory_metadata的语义一致——需要注意 API 参考中该键名为mem0_memory_metadata集成层文档中展示为mem0不同版本间键名可能演进读取时以所装版本为准。五、删除记忆delete_all_memories 与 delete_memory记忆的生命周期管理提供两个删除方法def delete_all_memories(*, user_id: str | None None, run_id: str | None None, agent_id: str | None None, **kwargs: Any) - None按作用域批量删除记忆记录user_id、run_id、agent_id至少设置一个kwargs透传给Mem0 client.delete_all方法。def delete_memory(memory_id: str, **kwargs: Any) - None按memory_id精确删除单条记忆该 ID 来自add_memories的返回值或检索结果中的memory_idkwargs透传给Mem0 client.delete方法。这两个方法让开发者可以实现“记忆遗忘”例如用户主动清除历史数据、或 Agent 判断某条记忆已过期时可按 ID 定点删除或按用户/会话整体清空避免长期记忆库无限膨胀或残留过期事实。六、过滤器转换normalize_filtersstaticmethod def normalize_filters(filters: dict[str, Any]) - dict[str, Any]这是一个静态方法负责把Haystack 风格的过滤器转换为 Mem0 风格的过滤器。Haystack 有自己的过滤器 DSL逻辑操作符如AND/OR/NOT、比较操作符如、、in等而 Mem0 client 需要其原生过滤器格式该方法是两者之间的适配桥梁在search_memories内部对filters参数做归一化处理。这意味着上层代码可以沿用 Haystack 统一的过滤语法编写查询条件而无需关心 Mem0 的具体过滤结构同时search_memories_as_single_message中提到的“更多 Mem0 过滤器细节”则直接透传两种过滤能力可以互补使用。七、作用域机制user_id / run_id / agent_idMem0 的记忆是多租户作用域化的。user_id、run_id、agent_id三个实体 ID 用于窄化读写范围它们都是运行时参数而非构造参数——因此单个Mem0MemoryStore实例可以服务多个用户、多个会话或多个 Agent无需为每个租户创建独立存储对象user_id最常见的粒度按用户隔离记忆run_id按会话/运行隔离适合同一用户的不同会话互不干扰agent_id按 Agent 隔离同时是存储 assistant 消息的前提条件。当前版本集成文档中还提到app_id作为第四个作用域实体可参考 mem0memorystore.mdxv2.19 实验版 API 参考的方法签名中仅显式列出user_id、run_id、agent_id三个参数其余实体通过**kwargs透传。八、实战示例独立使用 Mem0MemoryStore基础写入与检索from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore() store.add_memories( messages[ChatMessage.from_user(Alice prefers concise Python examples.)], user_idalice, inferFalse, # 精确文本直接入库不做事实抽取 ) memories store.search_memories( queryWhat does Alice prefer?, user_idalice, top_k3, ) print([msg.text for msg in memories])多实体 ID 复合作用域store.add_memories( messages[ ChatMessage.from_user(Alice is working on a documentation search system.) ], user_idalice, run_iddocs-assistant-session-1, inferTrue, # 让 Mem0 从消息中抽取事实 ) memories store.search_memories( queryWhat project is Alice working on?, user_idalice, run_iddocs-assistant-session-1, ) print([msg.text for msg in memories])复合作用域让同一用户的记忆可以进一步按会话细分避免跨会话的上下文污染。检索作用域内全部记忆传queryNone即可不做相关性检索返回作用域内的全部记忆all_memories store.search_memories(queryNone, user_idalice) print([msg.text for msg in all_memories])这在 Agent 需要“先盘点已知记忆、再决定是否精确检索”的场景中非常有用详见下文 Tools 部分。九、组合成 Agent 记忆管线Mem0MemoryStore的价值在组合使用时体现得最充分。以下模式来自 Mem0MemoryRetriever 与 Mem0MemoryWriter 文档展示了三种典型用法。模式一Pipeline 中“先检索、再回答”通过OutputAdapter把检索到的记忆与当前用户消息拼接注入 Agentfrom haystack import Pipeline from haystack.components.agents import Agent from haystack.components.converters import OutputAdapter from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.generators.utils import print_streaming_chunk from haystack.dataclasses import ChatMessage from haystack_integrations.components.retrievers.mem0 import Mem0MemoryRetriever from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore() pipeline Pipeline() pipeline.add_component(retriever, Mem0MemoryRetriever(memory_storestore, top_k5)) pipeline.add_component( memory_context, OutputAdapter( template{{ memories user_messages }}, output_typelist[ChatMessage], unsafeTrue, ), ) pipeline.add_component( agent, Agent( chat_generatorOpenAIChatGenerator(modelgpt-4o-mini), system_prompt( Use any system messages at the start of the conversation as long-term memory. Answer concisely. ), streaming_callbackprint_streaming_chunk, ), ) pipeline.connect(retriever.memories, memory_context.memories) pipeline.connect(memory_context.output, agent.messages) query Give me a short implementation tip. pipeline.run( { retriever: {query: query, user_id: alice}, memory_context: {user_messages: [ChatMessage.from_user(query)]}, }, )Mem0MemoryRetriever的关键运行语义与 store 一致query为字符串时做相关性检索并应用top_kqueryNone时返回作用域内全部记忆检索结果以systemChatMessage形态输出到memories输出端。模式二Pipeline 中“回合结束、写入记忆”把 Agent 完整输出的messages接到Mem0MemoryWriter用inferTrue让 Mem0 从整轮对话中抽取事实from haystack import Pipeline from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.generators.utils import print_streaming_chunk from haystack.dataclasses import ChatMessage from haystack_integrations.components.writers.mem0 import Mem0MemoryWriter from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore() pipeline Pipeline() pipeline.add_component( agent, Agent( chat_generatorOpenAIChatGenerator(modelgpt-4o-mini), system_prompt( Answer the user and preserve durable user facts or preferences for future conversations. ), streaming_callbackprint_streaming_chunk, ), ) pipeline.add_component(writer, Mem0MemoryWriter(memory_storestore, inferTrue)) pipeline.connect(agent.messages, writer.messages) result pipeline.run( { agent: { messages: [ ChatMessage.from_user(My name is Alice and I prefer concise Python examples.), ], }, writer: {user_id: alice}, }, ) print(result[writer][memories_written])Mem0MemoryWriter输出端为memories_written写入的记忆条数。写入同样需要至少一个作用域 IDuser_id/run_id/agent_id且infer的语义与add_memories完全一致——写整轮 Agent 回合用inferTrue上游已选定精确文本时用inferFalse。模式三让 Agent 自己决定读写Mem0 Memory ToolsMem0 Memory Tools 提供Mem0MemoryRetrieverTool默认工具名retrieve_memories与Mem0MemoryWriterTool默认工具名store_memory二者共享同一个Mem0MemoryStore。默认情况下工具通过inputs_from_state{user_id: user_id}从 Agent State 注入user_id这样用户 ID 不会作为工具参数暴露给 LLM一个 Agent 实例即可服务多用户from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.generators.utils import print_streaming_chunk from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore from haystack_integrations.tools.mem0 import ( Mem0MemoryRetrieverTool, Mem0MemoryWriterTool, ) store Mem0MemoryStore() retrieve_memories Mem0MemoryRetrieverTool(memory_storestore, top_k10) store_memory Mem0MemoryWriterTool(memory_storestore) agent Agent( chat_generatorOpenAIChatGenerator(modelgpt-5.4), tools[retrieve_memories, store_memory], system_promptYou are a helpful assistant with long-term memory. At the beginning of each turn, call retrieve_memories without a query to inspect known memories. Use store_memory only for new durable user-specific facts, preferences, or project context. Before storing, compare the proposed memory with retrieved memories and avoid duplicates. Do not store transient requests that are only useful in the current conversation. , streaming_callbackprint_streaming_chunk, state_schema{user_id: {type: str}}, ) result agent.run( messages[ ChatMessage.from_user( My name is Alice. Please remember that I prefer concise Python examples., ), ], user_idalice, )工具的参数约定Mem0MemoryRetrieverToolmemory_store必填top_k可选默认5name默认retrieve_memoriesdescription、parameters可选用于覆盖暴露给 LLM 的 JSON Schemainputs_from_state默认{user_id: user_id}。工具向 LLM 暴露query与top_k当 Agent 省略query或传null时返回注入作用域内的全部记忆Mem0MemoryWriterToolmemory_store必填name默认store_memory其余可选参数同上。工具向 LLM 暴露text与infer且默认inferFalse保证 Agent 存储的是它自己选定的精确文本。若需要注入更多作用域 ID如把 State 中的session_id映射到工具的run_id参数可扩展state_schema并用inputs_from_state{user_id: user_id, session_id: run_id}完成映射无需让 LLM 感知这些内部字段。十、从实验包到集成包的演进v2.19 的 experimental_mem0_memory_store_api.md 展示了该组件的实验形态位于haystack_experimental包中接口以haystack_experimental.memory_stores.mem0.memory_store.Mem0MemoryStore暴露。在当前仓库的主文档mem0memorystore.mdx中组件已迁移为正式集成mem0-haystack导入路径为haystack_integrations.memory_stores.mem0.Mem0MemoryStore并配套了 Retriever、Writer 与 Tools 的完整生态。核心方法add_memories、search_memories及作用域、infer等语义在两个版本间保持一致功能上当前集成还补充了app_id作用域、include_memory_metadata/mem0_memory_metadata元数据回传等增强。如果你是 v2.19 用户升级时只需调整导入路径与包依赖业务调用代码基本可平移。十一、小结与最佳实践综合 v2.19 API 参考与当前集成文档使用Mem0MemoryStore时建议遵循以下实践作用域先行任何读写操作都至少提供一个作用域 IDuser_id/run_id/agent_id或用filters限定范围无过滤器时三者缺一不可否则调用会失败按需选择infer写入完整 Agent 回合用inferTrue让 Mem0 抽取事实上游已确定精确文本时用inferFalse避免语义漂移善用queryNone需要盘点上下文时检索全部记忆需要精准命中时再提供query并配合top_k理解元数据流向写入时逐消息meta会被忽略改用kwargs中的metadata读取时用include_memory_metadata控制是否带回memory_id、score等 Mem0 检索元数据生命周期管理用delete_memory(memory_id)定点清理过期记忆用delete_all_memories(...)按用户/会话整体遗忘避免长期记忆库噪声累积组合而非孤立Mem0MemoryStore本身是数据层与Mem0MemoryRetriever、Mem0MemoryWriter或两个 Memory Tools 搭配才能构成“读取记忆 → 生成回答 → 沉淀新记忆”的完整长期记忆闭环。如需深入底层实现可继续阅读 ChatMessage 数据结构、Mem0MemoryRetriever、Mem0MemoryWriter 以及 Mem0 Memory Tools 的完整文档。【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考