Haystack Mem0MemoryStore 深度解析基于 Mem0 云端的长期记忆存储 API 实战指南【免费下载链接】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 为后端的记忆存储实现负责把ChatMessage记忆写入 Mem0、按语义检索记忆并通过user_id/run_id/agent_id等实体 ID 隔离不同用户与会话的记忆空间。本文以版本 2.24 的实验性 API 参考文档为主体逐方法拆解Mem0MemoryStore的初始化、写入、检索、删除、过滤与序列化语义并结合作者仓库中的用户指南与集成源码补充可运行的管线级实战示例帮助读者为 Agent 和对话系统搭建持久化长期记忆层。Mem0MemoryStore 在 Haystack 记忆体系中的定位Mem0 是一个面向 AI Agent 的长期记忆平台提供云端记忆的存储与语义检索能力。Mem0MemoryStore把该能力封装成 Haystack 统一的内存存储接口所有记忆都以ChatMessage为载体调用方无需关心 Mem0 REST API 的细节。在本仓库的文档体系中Mem0MemoryStore是记忆功能共享的数据层被三类上层组件共同使用Mem0MemoryRetriever从 store 检索长期记忆并输出为系统角色ChatMessage常放在 Agent 或 Chat Generator 之前Mem0MemoryWriter把ChatMessage列表持久化为长期记忆常放在管线末端Mem0 Memory ToolsMem0MemoryRetrieverTool与Mem0MemoryWriterTool让 Agent 在推理过程中自主读取与写入记忆。三者共享同一个Mem0MemoryStore实例因此只要在管线中维护一个 store就能实现读取—推理—写入的完整记忆闭环。模块路径说明版本差异版本 2.24 的 API 参考文档中该类位于实验性模块haystack_experimental.memory_stores.mem0.memory_store属于experimental标记组件API 可能随版本调整参见 experimental.py 中关于实验性组件可能变更或被移除的警告逻辑。在当前仓库的正式文档如 mem0memorystore.mdx中导入路径已迁移为haystack_integrations.memory_stores.mem0。实际使用时请以你安装的mem0-haystack版本对应的导入路径为准。安装与 API Key 配置pip install mem0-haystack然后设置 Mem0 云 API 密钥export MEM0_API_KEYyour-mem0-api-key密钥也可以通过构造参数显式传入见下文__init__两种方式等价。类与方法总览Mem0MemoryStore的核心方法可归纳为四类能力能力方法作用初始化__init__配置 Mem0 API 密钥写入add_memories将ChatMessage列表写入 Mem0检索search_memories按查询返回匹配的ChatMessage列表检索search_memories_as_single_message把检索结果合并为单个ChatMessage删除delete_all_memories/delete_memory按作用域删除 / 按 ID 删除过滤normalize_filters把 Haystack 过滤器转换为 Mem0 过滤器静态方法序列化to_dict/from_dict配置的字典序列化与反序列化初始化__init__与密钥管理def __init__(*, api_key: Secret Secret.from_env_var(MEM0_API_KEY))api_keyMem0 API 密钥类型为Secret。默认从环境变量MEM0_API_KEY读取也可以显式传入。从集成 API 参考mem0.md可以确认底层实现细节Mem0 客户端采用懒初始化策略构造 store 时并不会真正创建客户端而是在首次调用warm_up()或首次调用需要客户端的业务方法时才创建。如果你希望在首次管线运行前提前校验密钥或建立连接可以显式调用warm_up()。典型用法from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore() # 读取 MEM0_API_KEY 环境变量写入记忆add_memoriesdef 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记忆归属的用户 IDrun_idstr \| NoneNone记忆归属的运行会话IDagent_idstr \| NoneNone记忆归属的 Agent ID若希望 Mem0 存储来自 assistant 的聊天消息必须设置agent_idasync_modeboolFalse是否异步写入为True时方法立即返回记忆在后台异步添加kwargsAny—透传给 Mem0client.add的额外关键字参数两个关键语义infer控制记忆粒度。inferTrue适合写入 Agent 完整一轮的对话用户消息、工具上下文、最终回复由 Mem0 自动抽取其中值得长期保存的事实inferFalse适合上游已经选好精确记忆文本的场景直接按原文存储。消息级元数据会被忽略。Mem0 不允许为列表中的每条消息单独传元数据因此列表中ChatMessage.meta的内容会被忽略若需要为整批记忆附加元数据请通过metadata关键字参数传给 Mem0。返回值包含memory_id与对应记忆文本的对象列表可用于后续精确删除。完整示例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, # 原文入存不做事实抽取 )检索记忆search_memoriesdef 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]作用域规则务必注意若不提供filters则user_id、run_id、agent_id中至少设置一个若提供filters则检索范围以 filters 为准其他 ID 参数将被忽略。补充在当前仓库较新版本的集成 API 参考中该规则已演进为filters 与实体 ID 同时提供时按 AND 条件组合并新增了app_id实体 ID——详见 mem0.md使用时请以所装版本文档为准。参数说明参数类型默认值说明querystr \| NoneNone检索查询文本不传则返回作用域内全部记忆filtersdict[str, Any] \| NoneNone应用于检索的 Haystack 过滤器语法见 Metadata Filteringtop_kint5最大返回条数user_id/run_id/agent_idstr \| NoneNone记忆作用域实体 IDinclude_memory_metadataboolFalse是否在ChatMessage中附带 Mem0 相关元数据kwargsAny—透传给 Mem0 客户端的额外参数传了query则透传给client.search未传query则透传给client.get_all元数据处理当include_memory_metadataTrue时检索出的ChatMessage.meta会额外包含mem0_memory_metadata键其值为 Mem0 相关的检索元数据如memory_id、score等为False时ChatMessage.meta仅包含用户自定义元数据。返回值匹配条件的ChatMessage记忆列表。完整示例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])返回作用域内全部记忆传queryNone或不传query即可跳过相关性检索、直接返回作用域内所有记忆all_memories store.search_memories(queryNone, user_idalice) print([msg.text for msg in all_memories])合并检索search_memories_as_single_messagedef 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的变体检索规则、作用域规则无filters时至少设置一个 ID有filters时以 filters 为准与kwargs透传规则完全一致区别在于返回值是单个ChatMessage对象把所有匹配记忆合并成一条消息。当后续组件只接受单条消息输入例如直接拼入 prompt时使用该方法更便捷。删除记忆delete_all_memories与delete_memorydef delete_all_memories(*, user_id: str | None None, run_id: str | None None, agent_id: str | None None, **kwargs: Any) - None def delete_memory(memory_id: str, **kwargs: Any) - Nonedelete_all_memories按作用域批量删除。user_id、run_id、agent_id至少设置一个kwargs透传给 Mem0client.delete_all。适合用户注销、会话重置、Agent 下线等需要清空某作用域记忆的场景。delete_memory按记忆 ID 精确删除单条记忆。memory_id可以从add_memories的返回值中获得kwargs透传给 Mem0client.delete。适合修正错误记忆、去重后的清理等精细操作。过滤器转换normalize_filtersstaticmethod def normalize_filters(filters: dict[str, Any]) - dict[str, Any]静态方法把Haystack 风格过滤器转换为 Mem0 过滤器。Haystack 与 Mem0 的过滤器语法并不相同该方法作为内部适配层让上层组件能够以 Haystack 统一的过滤写法参考 Metadata Filtering检索记忆而无需关心 Mem0 原生过滤结构。结合集成 API 参考可知Mem0 侧支持一组固定的原生字段与操作符如user_id等实体字段非原生字段会被当作 Mem0 元数据字段处理。序列化to_dict与from_dictdef to_dict() - dict[str, Any] classmethod def from_dict(cls, data: dict[str, Any]) - Mem0MemoryStoreto_dict()把 store 的配置API 密钥引用等序列化为字典用于管线/组件的 YAML 或 JSON 持久化from_dict(data)类方法从字典反序列化重建Mem0MemoryStore实例配合 Haystack 组件的to_dict/from_dict约定使 store 可以无缝嵌入可序列化的管线配置中。实战组合把记忆层接入 Pipeline 与 Agent下面是在 Mem0MemoryRetriever 文档中给出的完整示例检索记忆后与当前用户消息拼接再喂给 Agent 作为长期记忆上下文。from 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)]}, }, )对称地Mem0MemoryWriter 文档展示了如何把 Agent 输出的完整messages接到 writer 上配合inferTrue让 Mem0 从整轮对话中抽取并持久化事实pipeline.add_component(writer, Mem0MemoryWriter(memory_storestore, inferTrue)) pipeline.connect(agent.messages, writer.messages)若希望 Agent 主动管理记忆还可使用 Mem0 Memory Tools 中的Mem0MemoryRetrieverTool默认名为retrieve_memories与Mem0MemoryWriterTool默认名为store_memory把读取/写入记忆的能力以工具形式暴露给 LLM。这两个工具默认通过inputs_from_state从 Agent State 注入user_id同一工具实例即可服务多个用户如需更细粒度作用域可在state_schema中增加字段并通过inputs_from_state映射到run_id、agent_id等参数。最佳实践与注意事项作用域是记忆安全的第一道防线。写入、检索、删除都遵循至少一个实体 ID的约束user_id/run_id/agent_id较新版本另有app_id。多租户场景务必为每次调用显式传入正确的 ID避免跨用户串记忆。记忆 assistant 消息必须设置agent_id。这是add_memories文档明确标注的行为若写入对象包含 assistant 消息却未提供agent_id这部分消息不会按预期存储。infer的选择取决于上游。inferTrue交给 Mem0 抽取事实适合整轮对话落库inferFalse保留原文适合已精确选定的记忆文本能避免重复抽取与噪声。元数据限制。add_memories忽略消息级ChatMessage.meta整批元数据需通过metadatakwarg 传入检索时如需 Mem0 返回的memory_id、score等字段记得打开include_memory_metadataTrue。实验性 API 的变更风险。2.24 参考文档对应haystack_experimental模块受experimental机制约束见 experimental.py签名与行为可能演进——例如作用域 ID 的数量、filters 与 ID 的组合方式在不同版本间已有差异升级依赖前应核对目标版本的 集成 API 参考。小结Mem0MemoryStore以统一的内存存储接口把 Mem0 云记忆能力接入 Haystack通过add_memories落库、search_memories/search_memories_as_single_message检索、delete_all_memories/delete_memory清理配合normalize_filters的过滤器适配与to_dict/from_dict的序列化约定再叠加user_id/run_id/agent_id的多维作用域即可在单个 store 实例之上支撑多用户、多会话、多 Agent 的长期记忆服务。结合 Mem0MemoryRetriever、Mem0MemoryWriter 与 Mem0 Memory Tools你可以在 Pipeline 或 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/haystack创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考