Haystack 集成 Mem0 长期记忆Mem0MemoryStore、Retriever/Writer 组件与 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本文基于 Haystack 2.23 版本文档编写。Mem0 是面向 LLM 应用的长期记忆平台本文档介绍的是mem0-haystack集成包它以Mem0MemoryStore为共享数据层向外提供Mem0MemoryRetriever检索组件、Mem0MemoryWriter写入组件以及供 Agent 直接调用的Mem0MemoryRetrieverTool/Mem0MemoryWriterTool两个现成工具。读完本文你将掌握如何在 Haystack Pipeline 中接入 Mem0 云端记忆为 Agent 或对话系统注入跨会话的用户长期记忆并学会用实体 ID 与过滤器精确圈定记忆的作用域。一、集成概览四件套与共享数据层在 Haystack 2.23 中Mem0 集成由haystack_integrations即mem0-haystack包提供共包含四个核心模块模块类角色haystack_integrations.components.retrievers.mem0Mem0MemoryRetriever从Mem0MemoryStore检索记忆输出ChatMessage列表system 角色haystack_integrations.components.writers.mem0Mem0MemoryWriter将ChatMessage列表写入Mem0MemoryStore输出写入数量haystack_integrations.memory_stores.mem0Mem0MemoryStore基于 Mem0 云 API 的记忆存储层读写两个核心方法haystack_integrations.tools.mem0Mem0MemoryRetrieverTool/Mem0MemoryWriterTool供 Agent 直接调用的现成工具Mem0MemoryStore是这四个模块共享的数据层检索器、写入器和两个 Agent 工具都围绕它工作。它与haystack_experimental中的实验性Mem0MemoryStore见 experimental_mem0_memory_store_api.md不同属于正式集成 API。二、安装与 API Key 配置安装集成包pip install mem0-haystack设置 Mem0 API KeyMem0MemoryStore默认从MEM0_API_KEY环境变量读取密钥export MEM0_API_KEYyour-mem0-api-key也可以显式传入Mem0MemoryStore(api_keySecret.from_token(your-mem0-api-key))。密钥通过 Haystack 的Secret机制管理默认构造器签名即__init__(*, api_key: Secret Secret.from_env_var(MEM0_API_KEY))。三、Mem0MemoryStore核心数据层Mem0MemoryStore是 A memory store backed by the Mem0 cloud API即基于 Mem0 云 API 的记忆存储。它提供两个核心方法add_memories—— 将一组ChatMessage对象作为记忆写入 Mem0search_memories—— 从 Mem0 检索与查询相关的记忆。它支持用至少一个 Mem0 实体 ID 圈定作用域user_id、run_id、agent_id或app_id。这些 ID 全部是运行时参数因此一个 store 实例可以同时服务多个用户或会话。3.1 客户端懒加载与 warm_upMem0 客户端采用懒创建策略__init__时并不会真正连接只有首次调用需要客户端的方法或显式调用warm_up()时才会创建。warm_up()适合在首次 Pipeline 运行前显式调用用于提前校验 API Key 或建立预连接。client属性会自动触发warm_up()。3.2 add_memories 写入记忆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, )参数说明messages要写入的ChatMessage列表infer默认True若为True由 Mem0 自动从消息中抽取记忆若为False则按消息原文原样存储user_id/run_id/agent_id/app_id作用域 ID。特别地agent_id是 Mem0 存储 assistant 消息时的必填项kwargs透传给 Mem0 客户端add方法的额外关键字参数。注意ChatMessage.meta会被忽略因为 Mem0 不支持逐条消息的元数据如需元数据请通过metadata关键字参数挂到整批记忆上。返回值list[dict[str, Any]]每个元素包含memory_id与memory文本。若 Mem0 API 调用失败抛出Mem0MemoryStoreError继承自RuntimeError。3.3 search_memories 检索记忆memories store.search_memories( queryWhat does Alice prefer?, user_idalice, top_k3, ) print([msg.text for msg in memories])参数说明query文本查询。省略或传None时返回作用域内全部记忆不做相关性搜索filtersHaystack 风格过滤器。Mem0 要求过滤器内必须包含实体 ID且只支持一组固定的原生字段与操作符对应 Mem0 的 Search Memories API 与 Memory Filters不属于 Mem0 原生过滤字段的键会被当作 Mem0 元数据字段处理。过滤器与 ID 同时提供时以AND条件组合search_memories要求filters与四个 ID 至少提供其一top_k最大返回条数默认5kwargs透传给 Mem0 客户端的额外参数。注意当传了query时透传给client.search未传时透传给client.get_all此行为在实验版 API 文档中有明确说明。返回值list[ChatMessage]system 角色。用户自定义的 Mem0 元数据进入每条消息的meta而memory_id、user_id、score、时间戳等 Mem0 检索字段则统一放在meta[mem0]下。若 API 调用失败抛出Mem0MemoryStoreError。3.4 多实体 ID 组合作用域Mem0 允许用四个 ID 任意组合来收窄读写范围store.add_memories( messages[ ChatMessage.from_user(Alice is working on a documentation search system.) ], user_idalice, run_iddocs-assistant-session-1, inferTrue, ) memories store.search_memories( queryWhat project is Alice working on?, user_idalice, run_iddocs-assistant-session-1, ) print([msg.text for msg in memories])3.5 检索作用域内全部记忆传queryNone即可按作用域返回全部记忆跳过相关性搜索all_memories store.search_memories(queryNone, user_idalice) print([msg.text for msg in all_memories])3.6 序列化to_dict()将 store 配置序列化为字典from_dict(data)从字典反序列化便于 YAML/JSON 方式持久化 Pipeline 配置。四、Mem0MemoryRetriever把记忆接进 PipelineMem0MemoryRetriever从Mem0MemoryStore检索记忆并输出为ChatMessage对象system 角色最常见的摆放位置是在 Agent 或 Chat Generator 之前用于在模型生成回复前注入长期记忆。4.1 初始化与运行签名__init__(*, memory_store: Mem0MemoryStore, top_k: int 5) - Nonememory_store必填Mem0MemoryStore实例top_k每次查询默认返回的最大记忆数默认5。运行签名run( query: str | None, *, user_id: str | None None, run_id: str | None None, agent_id: str | None None, app_id: str | None None, filters: dict[str, Any] | None None, top_k: int | None None ) - dict[str, list[ChatMessage]]query用于搜索相关记忆的文本传None则返回作用域内全部记忆user_id/run_id/agent_id/app_id圈定搜索作用域的实体 IDfiltersHaystack 风格过滤器与 ID 参数同时提供时按 AND 组合Mem0 要求过滤器内包含实体 ID非原生字段视为 Mem0 元数据字段top_k运行时覆盖初始化默认值。输出字典键为memories值为ChatMessage列表用户自定义 Mem0 元数据进入每条消息的metamemory_id、user_id、score、时间戳等检索字段放在meta[mem0]。4.2 单独使用from haystack.dataclasses import ChatMessage from haystack_integrations.components.retrievers.mem0 import Mem0MemoryRetriever 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, ) retriever Mem0MemoryRetriever(memory_storestore, top_k3) result retriever.run(queryanswer style, user_idalice) memories result[memories] for memory in memories: print(memory.text)检索作用域内全部记忆all_memories retriever.run(queryNone, user_idalice)[memories] print([memory.text for memory in all_memories])4.3 在 Pipeline 中记忆增强把检索到的记忆拼到当前用户消息之前再一起交给 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), ], }, }, )这里OutputAdapter用 Jinja 模板{{ memories user_messages }}把检索到的记忆与用户消息合并为完整消息列表并作为agent.messages输入。user_id作为运行时输入传给 retriever使同一个 Pipeline 实例可以为不同用户服务。五、Mem0MemoryWriter把对话沉淀为记忆Mem0MemoryWriter将ChatMessage列表写入Mem0MemoryStore最常见的摆放位置是在 Agent 或 Chat Generator 之后用于把会话事实、用户偏好、项目上下文持久化到后续运行。5.1 初始化与运行签名__init__(*, memory_store: Mem0MemoryStore, infer: bool True) - Nonememory_store必填Mem0MemoryStore实例infer若为True由 Mem0 从消息中抽取记忆若为False按消息原文存储。运行签名run( messages: list[ChatMessage], *, user_id: str | None None, run_id: str | None None, agent_id: str | None None, app_id: str | None None ) - dict[str, int]输出字典键memories_written值为实际写入的记忆条目数int。5.2 单独使用from haystack.dataclasses import ChatMessage from haystack_integrations.components.writers.mem0 import Mem0MemoryWriter from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore() writer Mem0MemoryWriter(memory_storestore, inferFalse) result writer.run( messages[ChatMessage.from_user(Alice prefers concise Python examples.)], user_idalice, ) print(result[memories_written])5.3 在 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])六、Agent 记忆工具让 LLM 自己读写记忆除了组件化接入 PipelineMem0 集成还提供两个现成 Tool让 Agent 在运行过程中自主决定何时读取、何时写入记忆retrieve_memoriesMem0MemoryRetrieverTool——搜索长期记忆不传 query 时返回作用域内全部记忆。store_memoryMem0MemoryWriterTool——把持久事实、偏好与上下文存为长期记忆。两者的核心设计是user_id通过inputs_from_state从 Agent State 在运行时注入因此一个 Tool 实例可以服务多个用户且用户 ID 不会作为工具调用参数暴露给 LLM。6.1 Mem0MemoryRetrieverTool初始化签名__init__( *, memory_store: Mem0MemoryStore, top_k: int 5, name: str retrieve_memories, description: str _DEFAULT_DESCRIPTION, parameters: dict[str, Any] _PARAMETERS, inputs_from_state: dict[str, str] _DEFAULT_INPUTS_FROM_STATE ) - Nonememory_store必填top_k默认5LLM 可覆盖name暴露给 LLM 的工具名默认retrieve_memoriesdescription暴露给 LLM 的工具描述parameters暴露给 LLM 的参数 JSON Schema默认只有可选的query与top_kinputs_from_stateAgent State 键到本工具参数名的映射默认{user_id: user_id}即把state[user_id]注入user_id参数。运行方法retrieve(queryNone, *, top_kNone, user_idNone, run_idNone, agent_idNone, app_idNone) - str默认只向 LLM 暴露query与top_kLLM 省略query或传None时返回注入作用域内的全部记忆返回值为格式化后的记忆字符串无匹配时返回提示消息。warm_up()初始化 Mem0 客户端后续调用为空操作。6.2 Mem0MemoryWriterTool初始化签名__init__( *, memory_store: Mem0MemoryStore, name: str store_memory, description: str _DEFAULT_DESCRIPTION, parameters: dict[str, Any] _PARAMETERS, inputs_from_state: dict[str, str] _DEFAULT_INPUTS_FROM_STATE ) - None运行方法store(text: str, *, infer: bool False, user_idNone, run_idNone, agent_idNone, app_idNone) - str默认只向 LLM 暴露text与infer写入工具默认inferFalse让 Agent 精确存储自己选定的记忆文本如需从较长文本如对话记录中抽取记忆可用inferTrue。返回值为表示已存记忆条目数的字符串。两个工具的to_dict()/from_dict()均支持序列化。6.3 完整 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-4o-mini), 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, )注意state_schema{user_id: {type: str}}是必须的inputs_from_state默认从 Agent State 读取user_idState 中必须先声明该键。6.4 通过 Agent State 注入更多 ID默认只暴露user_id但你可以在不扩充 LLM 可见参数的前提下通过inputs_from_state把更多 Mem0 实体 ID 注入工具from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.generators.utils import print_streaming_chunk from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore from haystack_integrations.tools.mem0 import ( Mem0MemoryRetrieverTool, Mem0MemoryWriterTool, ) store Mem0MemoryStore() inputs_from_state { user_id: user_id, # Map the Agent State key conversation_id to the tools run_id parameter. conversation_id: run_id, } retrieve_memories Mem0MemoryRetrieverTool( memory_storestore, inputs_from_stateinputs_from_state, ) store_memory Mem0MemoryWriterTool( memory_storestore, inputs_from_stateinputs_from_state, ) agent Agent( chat_generatorOpenAIChatGenerator(modelgpt-4o-mini), tools[retrieve_memories, store_memory], state_schema{ user_id: {type: str}, conversation_id: {type: str}, }, streaming_callbackprint_streaming_chunk, ) result agent.run( messages[ ChatMessage.from_user( Remember that this conversation is about the docs assistant prototype., ), ], user_idalice, conversation_iddocs-assistant-prototype, )映射规则是键是 Agent State 的键值是本工具的参数名。例如{user_id: user_id, session_id: run_id, agent_name: agent_id, app_name: app_id}会把state[session_id]传给工具的run_id参数、state[agent_name]传给agent_id、state[app_name]传给app_id。同时必须在 Agent 的state_schema中声明相应字段。七、记忆作用域与过滤器使用要点7.1 实体 ID 作用域规则检索或写入记忆时至少需要提供一个 Mem0 实体 IDuser_id、run_id、agent_id、app_id或filters四个 ID 可以任意组合同时传filters与 ID 时按AND条件组合常见做法单用户场景用user_id需要更窄作用域时叠加run_id会话、agent_id智能体、app_id应用agent_id对存储 assistant 消息是必需的见实验版 API 文档中 add_memories 的参数说明由于 ID 都是运行时参数单个 store / Pipeline / Tool 实例可以服务多个用户、会话、Agent 或应用。7.2 filters 使用规则filters是 Haystack 风格过滤器通用语法可参考 metadata-filtering.mdx当前 2.23 版本目录下对应的概念文档或仓库 docs-website/docs/concepts/metadata-filtering.mdxMem0 要求过滤器内必须包含实体 ID且只支持固定的原生字段与操作符集合对应 Mem0 Search Memories API 与 Memory Filters不属于 Mem0 原生过滤字段的键会被当作 Mem0 元数据字段参与过滤。八、错误处理与序列化Mem0MemoryStoreError位于haystack_integrations.memory_stores.mem0.errors继承自RuntimeError当 Mem0 API 操作失败时抛出。add_memories与search_memories均声明可能抛出该异常生产环境中应捕获处理序列化四个组件类均实现to_dict()/from_dict()可将配置持久化为 YAML/JSON配合 Haystack Pipeline 的序列化机制使用懒加载Mem0MemoryStore的客户端在首次使用时才创建可在 Pipeline 运行前显式warm_up()校验 API Key。九、典型落地模式总结Pipeline 注入式Mem0MemoryRetriever放在 Agent/Chat Generator 之前检索记忆拼接为 system 上下文见 4.3Mem0MemoryWriter放在 Agent 之后持久化整轮对话见 5.3。Agent 工具式将Mem0MemoryRetrieverTool与Mem0MemoryWriterTool一并挂给 Agent让 LLM 自主决定开篇检索已知记忆、对话中沉淀新事实并配合state_schemainputs_from_state完成多用户隔离。直接数据层使用不经过 Pipeline直接用Mem0MemoryStore.add_memories/search_memories完成程序化读写见 3.2、3.3。参考文档本集成 API 参考integrations-api/mem0.md实验版 Mem0MemoryStoreexperimental_mem0_memory_store_api.mdMem0MemoryStore 使用指南mem0memorystore.mdxMem0MemoryRetriever 使用指南mem0memoryretriever.mdxMem0MemoryWriter 使用指南mem0memorywriter.mdxMem0 Memory Tools 使用指南mem0memorytools.mdx【免费下载链接】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），仅供参考