尧图网页设计 
Haystack × Mem0 集成指南为 LLM Agent 与 Pipeline 构建云上长期记忆系统【免费下载链接】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/haystackMem0 集成是 Haystack 生态中面向长期记忆场景的官方组件它将 Mem0 云 API 的能力封装为四个可直接嵌入 Haystack 体系的构件底层存储Mem0MemoryStore、管线组件Mem0MemoryRetriever与Mem0MemoryWriter以及供 Agent 调用的工具Mem0MemoryRetrieverTool与Mem0MemoryWriterTool。本文以 integrations-api/mem0.md 为核心骨架结合仓库内的 Mem0MemoryStore 用户指南 与 ChatMessage 数据类实现完整讲解如何为多用户、多会话的 LLM 应用接入跨对话持久记忆并深入每个类的方法签名、参数语义与序列化机制读完即可在 Pipeline 与 Agent 中落地一套可复用的记忆层。集成概览记忆从会话内走向跨会话LLM 应用尤其是 Agent天然存在一个短板每次对话上下文只存在于当次会话用户偏好、历史事实、长期目标在下一次交互中全部丢失。Mem0 集成解决的就是这个持久化记忆问题——把值得记住的信息以记忆memory的形式存入 Mem0 云服务并在需要时按语义相关性检索回来。从 API 参考文档可以看到该集成由三个层级构成存储层Mem0MemoryStore—— 封装 Mem0 云 API 的MemoryClient提供add_memories写入与search_memories检索两个核心方法组件层Mem0MemoryRetriever检索与Mem0MemoryWriter写入它们是标准的 Haystack 组件可直接接入 Pipeline工具层Mem0MemoryRetrieverTool与Mem0MemoryWriterTool基于 Haystack 的Tool基类把记忆检索/存储暴露给 Agent 调用。一个贯穿全部层级的设计准则是作用域scoping记忆通过user_id、run_id、agent_id、app_id四个实体 ID 进行隔离。这些 ID 全部是运行时参数runtime parameters因此同一个 store、同一个 pipeline、同一个 tool 实例可以同时服务成百上千个用户而互不串扰——这是该集成面向生产环境的关键设计。这一点在仓库的 Mem0MemoryStore 用户指南 中被明确强调These are runtime parameters, so a single store instance can serve multiple users or sessions.此外记忆的载体统一使用 Haystack 的ChatMessage对象检索结果以 system 角色的消息返回可以直接拼接进模型上下文。从 ChatMessage 实现 可以看到它通过from_user、from_system等类方法构造并支持meta携带附加信息这与 Mem0 组件的元数据处理逻辑完全对应。安装与环境准备安装mem0-haystack集成包该包名在 Mem0MemoryStore 用户指南 的 key-value 表中明确列出pip install mem0-haystack然后配置 Mem0 API 密钥有两种方式二者等价# 方式一环境变量推荐也是默认行为 export MEM0_API_KEYyour-mem0-api-key # 方式二构造时显式传入 from haystack.utils import Secret from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore(api_keySecret.from_token(your-mem0-api-key))关键实现细节Mem0MemoryStore.__init__的签名是__init__(*, api_key: Secret Secret.from_env_var(MEM0_API_KEY))即默认从MEM0_API_KEY环境变量读取密钥。同时Mem0 客户端是惰性创建的——__init__只保存配置直到首次warm_up()或第一次需要客户端的方法被调用时才真正初始化。这意味着你可以在未设置密钥时就完成对象构造只要在真正运行前配置好环境变量即可。如果想在第一次 Pipeline 运行前主动校验密钥是否有效可以显式调用warm_up()重复调用是幂等的后续调用为空操作。底层存储Mem0MemoryStoreMem0MemoryStore是整个集成的数据层其余三个构件两个组件、两个工具全部依赖它。它直接持有 Mem0 的MemoryClient通过client属性暴露所有 Mem0 API 调用失败时都会抛出Mem0MemoryStoreError继承自RuntimeError。写入记忆add_memoriesadd_memories( *, messages: list[ChatMessage], user_id: str | None None, run_id: str | None None, agent_id: str | None None, app_id: str | None None, infer: bool True, **kwargs: Any ) - list[dict[str, Any]]参数语义与使用要点messages要存储的ChatMessage列表user_id/run_id/agent_id/app_id四个作用域 ID用于标记这批记忆属于谁。特别地文档明确指出如果要让 Mem0 存储 assistant助手角色的消息必须提供agent_idinfer控制 Mem0 对消息的处理方式。inferTrue默认Mem0 自动从消息中抽取值得记忆的事实memory extraction适合存储完整的 Agent 回合一整轮对话后让 Mem0 提炼要点inferFalse把消息文本原样存为记忆适合上游已经精确挑选好记忆文本的场景**kwargs透传给 Mem0 client 的add方法。注意一个容易踩坑的细节ChatMessage.meta会被忽略因为 Mem0 不支持逐条消息的元数据如果需要为整批记忆附加元数据请通过metadata关键字参数传给kwargs。返回值是list[dict[str, Any]]每个字典包含memory_id与memory文本可用于确认写入结果或后续按 ID 删除。检索记忆search_memoriessearch_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, app_id: str | None None, **kwargs: Any ) - list[ChatMessage]参数语义与使用要点作用域规则要么提供filters要么至少提供一个实体 IDuser_id、run_id、agent_id、app_id中之一否则调用会失败当filters与 ID 同时提供时二者以AND 逻辑组合query用于相关性检索的文本。传None则不做相关性搜索返回作用域内的全部记忆——这一全量召回模式在需要初始化 Agent 上下文时非常有用filtersHaystack 风格的过滤器字典会转换为 Mem0 原生过滤器。Mem0 支持一组固定的原生字段与操作符对应 Mem0 的 Search Memories API 与 Memory Filters 规范不属于 Mem0 原生过滤字段的键会被当作 Mem0 metadata 字段处理top_k返回结果上限默认 5**kwargs透传给 Mem0 client。返回值是list[ChatMessage]全部为system 角色的消息由ChatMessage.from_system构造。每条消息的meta中用户自定义的 Mem0 metadata 直接放在消息的meta里Mem0 的检索字段——memory_id、user_id、score、时间戳等——统一放在meta[mem0]键下。也就是说你可以从message.meta[mem0][memory_id]拿到记忆 ID 用于后续删除从message.meta[mem0][score]评估检索相关性。独立使用示例不接 Pipelinestore 可以单独使用from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore() # 写入inferFalse 表示文本原样存储 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 返回作用域内全部记忆 all_memories store.search_memories(queryNone, user_idalice) print([msg.text for msg in all_memories])多实体 ID 组合作用域四个 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])组件层在 Pipeline 中读写记忆组件层是管线化的入口让记忆读写成为 Pipeline 的一个节点可以与其他组件检索器、生成器、Agent自由编排。Mem0MemoryRetriever检索记忆组件在 Pipeline 中Mem0MemoryRetriever负责在把上下文交给语言模型或 Agent 之前先取回相关记忆。其构造函数为__init__(*, memory_store: Mem0MemoryStore, top_k: int 5) - Nonetop_k是初始化时的默认返回条数默认 5运行时可被覆盖。其run方法签名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时返回作用域内全部记忆作用域 ID、filters与top_k均为运行时参数top_k可覆盖初始化默认值filters与 ID 同时提供时按 AND 组合与 store 行为一致返回{memories: [ChatMessage, ...]}memories键下的消息为 system 角色可直接拼入后续模型调用。from haystack_integrations.components.retrievers.mem0 import Mem0MemoryRetriever from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore() retriever Mem0MemoryRetriever(memory_storestore, top_k3) result retriever.run(queryWhat does Alice like?, user_idalice) memories result[memories] print([message.text for message in memories]) # queryNone 全量召回作用域内记忆 all_memories retriever.run(queryNone, user_idalice)[memories]Mem0MemoryWriter写入记忆组件Mem0MemoryWriter把对话消息持久化为记忆其构造函数__init__(*, memory_store: Mem0MemoryStore, infer: bool True) - Noneinfer默认TrueMem0 自动抽取记忆运行时可覆盖。run方法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]由于作用域 ID 全部是运行时参数同一个 writer 实例可以为多个用户或 Agent 服务——这正是文档强调的same pipeline instance can serve multiple users or agents。返回{memories_written: int}即本次写入的记忆条数。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])组合进 Pipeline 的典型模式一个典型的写入-检索闭环 Pipeline 可以这样组织对话结束后用Mem0MemoryWriter沉淀记忆下一次对话开始时用Mem0MemoryRetriever召回相关记忆并注入 Prompt。由于两个组件都接受运行时参数可以用同一个 Pipeline 通过不同的user_id服务不同用户from haystack import Pipeline memory_pipeline Pipeline() memory_pipeline.add_component(writer, Mem0MemoryWriter(memory_storestore, inferTrue)) memory_pipeline.add_component(retriever, Mem0MemoryRetriever(memory_storestore, top_k5)) # 会话结束后写入 memory_pipeline.run( {writer: {messages: [ChatMessage.from_user(Alice likes Rust.)], user_id: alice}} ) # 下次会话开始前召回 result memory_pipeline.run( {retriever: {query: What does Alice like?, user_id: alice}} )工具层让 Agent 自主读写记忆工具层把记忆能力包装成 Agent 可调用的工具。Haystack 的 Agent 会依据工具描述让 LLM 决定何时调用因此工具的参数暴露策略经过精心设计LLM 只能看到最少的参数而作用域 ID 由 Agent 的 State 在运行时自动注入——这正是一个工具实例服务多用户的实现机制。Mem0MemoryRetrieverToolAgent 侧的记忆检索构造函数__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 ) - Nonename暴露给 LLM 的工具名默认retrieve_memoriesdescription工具描述LLM 据此判断何时调用parameters暴露给 LLM 的 JSON Schema默认只包含可选的query与top_kinputs_from_state从 Agent State 到工具参数的映射默认{user_id: user_id}即把state[user_id]注入工具的user_id参数。键是 Agent State 的键值是工具的参数名。核心方法retrieveretrieve( query: str | None None, *, top_k: int | None None, user_id: str | None None, run_id: str | None None, agent_id: str | None None, app_id: str | None None ) - strquery省略或为None时返回注入作用域内的全部记忆top_k可覆盖工具默认值返回格式化后的记忆字符串供 Agent 阅读无匹配记忆时返回相应提示信息。Agent 集成示例来自 API 参考文档from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore from haystack_integrations.tools.mem0 import Mem0MemoryRetrieverTool store Mem0MemoryStore() retrieve_memories Mem0MemoryRetrieverTool(memory_storestore, top_k5) agent Agent( chat_generatorOpenAIChatGenerator(modelgpt-4o-mini), tools[retrieve_memories], state_schema{user_id: {type: str}, session_id: {type: str}}, ) # Agent 可以带 query 定向召回也可以不带 query 全量获取作用域内记忆 result agent.run( messages[ChatMessage.from_user(What do you remember about me?)], user_idalice, session_idchat-42, ) print(result[last_message].text)注入更多实体 ID若要同时注入run_id、agent_id、app_id需要两步先在 Agent 的state_schema中声明对应状态字段再传给工具的inputs_from_state。例如把会话 ID 映射到run_idretrieve_memories Mem0MemoryRetrieverTool( memory_storestore, inputs_from_state{user_id: user_id, session_id: run_id}, ) # 运行时 state[session_id] 会注入工具的 run_id 参数完整的多 ID 映射示例为{user_id: user_id, session_id: run_id, agent_name: agent_id, app_name: app_id}。Mem0MemoryWriterToolAgent 侧的记忆存储构造函数__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_memory暴露给 LLM 的参数只有text和inferinputs_from_state默认同样为{user_id: user_id}。核心方法storestore( text: str, *, infer: bool False, user_id: str | None None, run_id: str | None None, agent_id: str | None None, app_id: str | None None ) - strtext要存为记忆的信息inferTrue时 Mem0 从文本中抽取记忆False默认时文本原样存储——注意与 writer 组件默认True不同工具层默认False因为 LLM 已经生成了明确的记忆文本返回存储条数的说明字符串。Agent 集成示例from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore from haystack_integrations.tools.mem0 import Mem0MemoryWriterTool store Mem0MemoryStore() store_memory Mem0MemoryWriterTool(memory_storestore) agent Agent( chat_generatorOpenAIChatGenerator(modelgpt-4o-mini), tools[store_memory], state_schema{user_id: {type: str}, session_id: {type: str}}, ) result agent.run( messages[ChatMessage.from_user(Remember that I prefer concise Python examples.)], user_idalice, session_idchat-42, ) print(result[last_message].text)两个工具都实现了warm_up()首次调用初始化 Mem0 client之后为空操作并实现了完整的to_dict()/from_dict()序列化接口。序列化支持to_dict 与 from_dict集成的所有构件都实现了完整的序列化接口这是 Haystack 组件生态的标准要求保证 Pipeline 可以导出为 YAML/JSON 再恢复类to_dict / from_dictMem0MemoryStore序列化 store 配置含 API key 的安全处理Mem0MemoryRetriever序列化组件配置Mem0MemoryWriter序列化组件配置Mem0MemoryRetrieverTool序列化工具配置Mem0MemoryWriterTool序列化工具配置这意味着包含记忆读写节点的 Pipeline 可以被保存、版本化、跨环境迁移而api_key通过 Haystack 的Secret机制管理不会以明文落入序列化文件。错误处理与实战要点异常类型所有 Mem0 API 操作失败都会抛出Mem0MemoryStoreErrorRuntimeError子类捕获它即可统一处理网络/鉴权/限流等错误作用域校验search_memories要求有 filters 或至少一个实体 ID编写调用时务必满足否则会直接报错filters与 ID 同时给出时按 AND 组合assistant 消息存储要让 Mem0 保存助手角色的消息必须设置agent_idmeta 处理的两条规则写入时ChatMessage.meta被忽略如需批量元数据用metadatakwarg 传给 Mem0读取时用户元数据进入message.metaMem0 检索字段memory_id、score、时间戳等进入message.meta[mem0]惰性初始化store 与工具的 Mem0 client 都在首次使用时才创建warm_up()可用于提前校验密钥或预连接过滤器兼容性filters走 Haystack 风格但 Mem0 只支持固定的一组原生字段与操作符非原生字段会被当作 Mem0 metadata 字段处理——构造过滤器时需对照 Mem0 的 Search Memories API 与 Memory Filters 规范。小结Mem0 集成为 Haystack 应用补上了跨会话长期记忆这一关键能力且从存储到组件再到 Agent 工具每一层都贯彻了同一套设计以ChatMessage为记忆载体、以四个实体 ID 做运行时作用域隔离、以惰性客户端和完整序列化接口保证生产可用。开发者既可以在 Pipeline 中用Mem0MemoryRetriever/Mem0MemoryWriter显式编排记忆读写也可以借助两个Tool让 Agent 自主决定何时记住、何时回忆从而构建出真正具备持久记忆能力的多用户 LLM 应用。更完整的组件级用法可继续参考仓库中的 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），仅供参考 Haystack × Mem0 集成指南：为 LLM Agent 与 Pipeline 构建云上长期记忆系统
Haystack × Mem0 集成指南为 LLM Agent 与 Pipeline 构建云上长期记忆系统【免费下载链接】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/haystackMem0 集成是 Haystack 生态中面向长期记忆场景的官方组件它将 Mem0 云 API 的能力封装为四个可直接嵌入 Haystack 体系的构件底层存储Mem0MemoryStore、管线组件Mem0MemoryRetriever与Mem0MemoryWriter以及供 Agent 调用的工具Mem0MemoryRetrieverTool与Mem0MemoryWriterTool。本文以 integrations-api/mem0.md 为核心骨架结合仓库内的 Mem0MemoryStore 用户指南 与 ChatMessage 数据类实现完整讲解如何为多用户、多会话的 LLM 应用接入跨对话持久记忆并深入每个类的方法签名、参数语义与序列化机制读完即可在 Pipeline 与 Agent 中落地一套可复用的记忆层。集成概览记忆从会话内走向跨会话LLM 应用尤其是 Agent天然存在一个短板每次对话上下文只存在于当次会话用户偏好、历史事实、长期目标在下一次交互中全部丢失。Mem0 集成解决的就是这个持久化记忆问题——把值得记住的信息以记忆memory的形式存入 Mem0 云服务并在需要时按语义相关性检索回来。从 API 参考文档可以看到该集成由三个层级构成存储层Mem0MemoryStore—— 封装 Mem0 云 API 的MemoryClient提供add_memories写入与search_memories检索两个核心方法组件层Mem0MemoryRetriever检索与Mem0MemoryWriter写入它们是标准的 Haystack 组件可直接接入 Pipeline工具层Mem0MemoryRetrieverTool与Mem0MemoryWriterTool基于 Haystack 的Tool基类把记忆检索/存储暴露给 Agent 调用。一个贯穿全部层级的设计准则是作用域scoping记忆通过user_id、run_id、agent_id、app_id四个实体 ID 进行隔离。这些 ID 全部是运行时参数runtime parameters因此同一个 store、同一个 pipeline、同一个 tool 实例可以同时服务成百上千个用户而互不串扰——这是该集成面向生产环境的关键设计。这一点在仓库的 Mem0MemoryStore 用户指南 中被明确强调These are runtime parameters, so a single store instance can serve multiple users or sessions.此外记忆的载体统一使用 Haystack 的ChatMessage对象检索结果以 system 角色的消息返回可以直接拼接进模型上下文。从 ChatMessage 实现 可以看到它通过from_user、from_system等类方法构造并支持meta携带附加信息这与 Mem0 组件的元数据处理逻辑完全对应。安装与环境准备安装mem0-haystack集成包该包名在 Mem0MemoryStore 用户指南 的 key-value 表中明确列出pip install mem0-haystack然后配置 Mem0 API 密钥有两种方式二者等价# 方式一环境变量推荐也是默认行为 export MEM0_API_KEYyour-mem0-api-key # 方式二构造时显式传入 from haystack.utils import Secret from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore(api_keySecret.from_token(your-mem0-api-key))关键实现细节Mem0MemoryStore.__init__的签名是__init__(*, api_key: Secret Secret.from_env_var(MEM0_API_KEY))即默认从MEM0_API_KEY环境变量读取密钥。同时Mem0 客户端是惰性创建的——__init__只保存配置直到首次warm_up()或第一次需要客户端的方法被调用时才真正初始化。这意味着你可以在未设置密钥时就完成对象构造只要在真正运行前配置好环境变量即可。如果想在第一次 Pipeline 运行前主动校验密钥是否有效可以显式调用warm_up()重复调用是幂等的后续调用为空操作。底层存储Mem0MemoryStoreMem0MemoryStore是整个集成的数据层其余三个构件两个组件、两个工具全部依赖它。它直接持有 Mem0 的MemoryClient通过client属性暴露所有 Mem0 API 调用失败时都会抛出Mem0MemoryStoreError继承自RuntimeError。写入记忆add_memoriesadd_memories( *, messages: list[ChatMessage], user_id: str | None None, run_id: str | None None, agent_id: str | None None, app_id: str | None None, infer: bool True, **kwargs: Any ) - list[dict[str, Any]]参数语义与使用要点messages要存储的ChatMessage列表user_id/run_id/agent_id/app_id四个作用域 ID用于标记这批记忆属于谁。特别地文档明确指出如果要让 Mem0 存储 assistant助手角色的消息必须提供agent_idinfer控制 Mem0 对消息的处理方式。inferTrue默认Mem0 自动从消息中抽取值得记忆的事实memory extraction适合存储完整的 Agent 回合一整轮对话后让 Mem0 提炼要点inferFalse把消息文本原样存为记忆适合上游已经精确挑选好记忆文本的场景**kwargs透传给 Mem0 client 的add方法。注意一个容易踩坑的细节ChatMessage.meta会被忽略因为 Mem0 不支持逐条消息的元数据如果需要为整批记忆附加元数据请通过metadata关键字参数传给kwargs。返回值是list[dict[str, Any]]每个字典包含memory_id与memory文本可用于确认写入结果或后续按 ID 删除。检索记忆search_memoriessearch_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, app_id: str | None None, **kwargs: Any ) - list[ChatMessage]参数语义与使用要点作用域规则要么提供filters要么至少提供一个实体 IDuser_id、run_id、agent_id、app_id中之一否则调用会失败当filters与 ID 同时提供时二者以AND 逻辑组合query用于相关性检索的文本。传None则不做相关性搜索返回作用域内的全部记忆——这一全量召回模式在需要初始化 Agent 上下文时非常有用filtersHaystack 风格的过滤器字典会转换为 Mem0 原生过滤器。Mem0 支持一组固定的原生字段与操作符对应 Mem0 的 Search Memories API 与 Memory Filters 规范不属于 Mem0 原生过滤字段的键会被当作 Mem0 metadata 字段处理top_k返回结果上限默认 5**kwargs透传给 Mem0 client。返回值是list[ChatMessage]全部为system 角色的消息由ChatMessage.from_system构造。每条消息的meta中用户自定义的 Mem0 metadata 直接放在消息的meta里Mem0 的检索字段——memory_id、user_id、score、时间戳等——统一放在meta[mem0]键下。也就是说你可以从message.meta[mem0][memory_id]拿到记忆 ID 用于后续删除从message.meta[mem0][score]评估检索相关性。独立使用示例不接 Pipelinestore 可以单独使用from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore() # 写入inferFalse 表示文本原样存储 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 返回作用域内全部记忆 all_memories store.search_memories(queryNone, user_idalice) print([msg.text for msg in all_memories])多实体 ID 组合作用域四个 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])组件层在 Pipeline 中读写记忆组件层是管线化的入口让记忆读写成为 Pipeline 的一个节点可以与其他组件检索器、生成器、Agent自由编排。Mem0MemoryRetriever检索记忆组件在 Pipeline 中Mem0MemoryRetriever负责在把上下文交给语言模型或 Agent 之前先取回相关记忆。其构造函数为__init__(*, memory_store: Mem0MemoryStore, top_k: int 5) - Nonetop_k是初始化时的默认返回条数默认 5运行时可被覆盖。其run方法签名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时返回作用域内全部记忆作用域 ID、filters与top_k均为运行时参数top_k可覆盖初始化默认值filters与 ID 同时提供时按 AND 组合与 store 行为一致返回{memories: [ChatMessage, ...]}memories键下的消息为 system 角色可直接拼入后续模型调用。from haystack_integrations.components.retrievers.mem0 import Mem0MemoryRetriever from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store Mem0MemoryStore() retriever Mem0MemoryRetriever(memory_storestore, top_k3) result retriever.run(queryWhat does Alice like?, user_idalice) memories result[memories] print([message.text for message in memories]) # queryNone 全量召回作用域内记忆 all_memories retriever.run(queryNone, user_idalice)[memories]Mem0MemoryWriter写入记忆组件Mem0MemoryWriter把对话消息持久化为记忆其构造函数__init__(*, memory_store: Mem0MemoryStore, infer: bool True) - Noneinfer默认TrueMem0 自动抽取记忆运行时可覆盖。run方法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]由于作用域 ID 全部是运行时参数同一个 writer 实例可以为多个用户或 Agent 服务——这正是文档强调的same pipeline instance can serve multiple users or agents。返回{memories_written: int}即本次写入的记忆条数。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])组合进 Pipeline 的典型模式一个典型的写入-检索闭环 Pipeline 可以这样组织对话结束后用Mem0MemoryWriter沉淀记忆下一次对话开始时用Mem0MemoryRetriever召回相关记忆并注入 Prompt。由于两个组件都接受运行时参数可以用同一个 Pipeline 通过不同的user_id服务不同用户from haystack import Pipeline memory_pipeline Pipeline() memory_pipeline.add_component(writer, Mem0MemoryWriter(memory_storestore, inferTrue)) memory_pipeline.add_component(retriever, Mem0MemoryRetriever(memory_storestore, top_k5)) # 会话结束后写入 memory_pipeline.run( {writer: {messages: [ChatMessage.from_user(Alice likes Rust.)], user_id: alice}} ) # 下次会话开始前召回 result memory_pipeline.run( {retriever: {query: What does Alice like?, user_id: alice}} )工具层让 Agent 自主读写记忆工具层把记忆能力包装成 Agent 可调用的工具。Haystack 的 Agent 会依据工具描述让 LLM 决定何时调用因此工具的参数暴露策略经过精心设计LLM 只能看到最少的参数而作用域 ID 由 Agent 的 State 在运行时自动注入——这正是一个工具实例服务多用户的实现机制。Mem0MemoryRetrieverToolAgent 侧的记忆检索构造函数__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 ) - Nonename暴露给 LLM 的工具名默认retrieve_memoriesdescription工具描述LLM 据此判断何时调用parameters暴露给 LLM 的 JSON Schema默认只包含可选的query与top_kinputs_from_state从 Agent State 到工具参数的映射默认{user_id: user_id}即把state[user_id]注入工具的user_id参数。键是 Agent State 的键值是工具的参数名。核心方法retrieveretrieve( query: str | None None, *, top_k: int | None None, user_id: str | None None, run_id: str | None None, agent_id: str | None None, app_id: str | None None ) - strquery省略或为None时返回注入作用域内的全部记忆top_k可覆盖工具默认值返回格式化后的记忆字符串供 Agent 阅读无匹配记忆时返回相应提示信息。Agent 集成示例来自 API 参考文档from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore from haystack_integrations.tools.mem0 import Mem0MemoryRetrieverTool store Mem0MemoryStore() retrieve_memories Mem0MemoryRetrieverTool(memory_storestore, top_k5) agent Agent( chat_generatorOpenAIChatGenerator(modelgpt-4o-mini), tools[retrieve_memories], state_schema{user_id: {type: str}, session_id: {type: str}}, ) # Agent 可以带 query 定向召回也可以不带 query 全量获取作用域内记忆 result agent.run( messages[ChatMessage.from_user(What do you remember about me?)], user_idalice, session_idchat-42, ) print(result[last_message].text)注入更多实体 ID若要同时注入run_id、agent_id、app_id需要两步先在 Agent 的state_schema中声明对应状态字段再传给工具的inputs_from_state。例如把会话 ID 映射到run_idretrieve_memories Mem0MemoryRetrieverTool( memory_storestore, inputs_from_state{user_id: user_id, session_id: run_id}, ) # 运行时 state[session_id] 会注入工具的 run_id 参数完整的多 ID 映射示例为{user_id: user_id, session_id: run_id, agent_name: agent_id, app_name: app_id}。Mem0MemoryWriterToolAgent 侧的记忆存储构造函数__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_memory暴露给 LLM 的参数只有text和inferinputs_from_state默认同样为{user_id: user_id}。核心方法storestore( text: str, *, infer: bool False, user_id: str | None None, run_id: str | None None, agent_id: str | None None, app_id: str | None None ) - strtext要存为记忆的信息inferTrue时 Mem0 从文本中抽取记忆False默认时文本原样存储——注意与 writer 组件默认True不同工具层默认False因为 LLM 已经生成了明确的记忆文本返回存储条数的说明字符串。Agent 集成示例from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore from haystack_integrations.tools.mem0 import Mem0MemoryWriterTool store Mem0MemoryStore() store_memory Mem0MemoryWriterTool(memory_storestore) agent Agent( chat_generatorOpenAIChatGenerator(modelgpt-4o-mini), tools[store_memory], state_schema{user_id: {type: str}, session_id: {type: str}}, ) result agent.run( messages[ChatMessage.from_user(Remember that I prefer concise Python examples.)], user_idalice, session_idchat-42, ) print(result[last_message].text)两个工具都实现了warm_up()首次调用初始化 Mem0 client之后为空操作并实现了完整的to_dict()/from_dict()序列化接口。序列化支持to_dict 与 from_dict集成的所有构件都实现了完整的序列化接口这是 Haystack 组件生态的标准要求保证 Pipeline 可以导出为 YAML/JSON 再恢复类to_dict / from_dictMem0MemoryStore序列化 store 配置含 API key 的安全处理Mem0MemoryRetriever序列化组件配置Mem0MemoryWriter序列化组件配置Mem0MemoryRetrieverTool序列化工具配置Mem0MemoryWriterTool序列化工具配置这意味着包含记忆读写节点的 Pipeline 可以被保存、版本化、跨环境迁移而api_key通过 Haystack 的Secret机制管理不会以明文落入序列化文件。错误处理与实战要点异常类型所有 Mem0 API 操作失败都会抛出Mem0MemoryStoreErrorRuntimeError子类捕获它即可统一处理网络/鉴权/限流等错误作用域校验search_memories要求有 filters 或至少一个实体 ID编写调用时务必满足否则会直接报错filters与 ID 同时给出时按 AND 组合assistant 消息存储要让 Mem0 保存助手角色的消息必须设置agent_idmeta 处理的两条规则写入时ChatMessage.meta被忽略如需批量元数据用metadatakwarg 传给 Mem0读取时用户元数据进入message.metaMem0 检索字段memory_id、score、时间戳等进入message.meta[mem0]惰性初始化store 与工具的 Mem0 client 都在首次使用时才创建warm_up()可用于提前校验密钥或预连接过滤器兼容性filters走 Haystack 风格但 Mem0 只支持固定的一组原生字段与操作符非原生字段会被当作 Mem0 metadata 字段处理——构造过滤器时需对照 Mem0 的 Search Memories API 与 Memory Filters 规范。小结Mem0 集成为 Haystack 应用补上了跨会话长期记忆这一关键能力且从存储到组件再到 Agent 工具每一层都贯彻了同一套设计以ChatMessage为记忆载体、以四个实体 ID 做运行时作用域隔离、以惰性客户端和完整序列化接口保证生产可用。开发者既可以在 Pipeline 中用Mem0MemoryRetriever/Mem0MemoryWriter显式编排记忆读写也可以借助两个Tool让 Agent 自主决定何时记住、何时回忆从而构建出真正具备持久记忆能力的多用户 LLM 应用。更完整的组件级用法可继续参考仓库中的 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），仅供参考