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/haystack

Mem0 集成是 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_KEY="your-mem0-api-key" # 方式二：构造时显式传入 from haystack.utils import Secret from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store = Mem0MemoryStore(api_key=Secret.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()（重复调用是幂等的，后续调用为空操作）。

底层存储：Mem0MemoryStore

Mem0MemoryStore是整个集成的数据层，其余三个构件（两个组件、两个工具）全部依赖它。它直接持有 Mem0 的MemoryClient（通过client属性暴露），所有 Mem0 API 调用失败时都会抛出Mem0MemoryStoreError（继承自RuntimeError）。

写入记忆：add_memories

add_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_id；
• infer：控制 Mem0 对消息的处理方式。
  • infer=True（默认）：Mem0 自动从消息中抽取值得记忆的事实（memory extraction），适合存储完整的 Agent 回合（一整轮对话）后让 Mem0 提炼要点；
  • infer=False：把消息文本原样存为记忆，适合上游已经精确挑选好记忆文本的场景；
• **kwargs：透传给 Mem0 client 的add方法。注意一个容易踩坑的细节：ChatMessage.meta会被忽略，因为 Mem0 不支持逐条消息的元数据；如果需要为整批记忆附加元数据，请通过metadata关键字参数传给kwargs。

返回值是list[dict[str, Any]]，每个字典包含memory_id与memory文本，可用于确认写入结果或后续按 ID 删除。

检索记忆：search_memories

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, app_id: str | None = None, **kwargs: Any ) -> list[ChatMessage]

参数语义与使用要点：

• 作用域规则：要么提供filters，要么至少提供一个实体 ID（user_id、run_id、agent_id、app_id中之一），否则调用会失败；当filters与 ID 同时提供时，二者以AND 逻辑组合；
• query：用于相关性检索的文本。传None则不做相关性搜索，返回作用域内的全部记忆——这一"全量召回"模式在需要初始化 Agent 上下文时非常有用；
• filters：Haystack 风格的过滤器字典，会转换为 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"]评估检索相关性。

独立使用示例

不接 Pipeline，store 可以单独使用：

from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.mem0 import Mem0MemoryStore store = Mem0MemoryStore() # 写入：infer=False 表示文本原样存储 store.add_memories( messages=[ChatMessage.from_user("Alice prefers concise Python examples.")], user_id="alice", infer=False, ) # 检索：按相关性搜索 memories = store.search_memories( query="What does Alice prefer?", user_id="alice", top_k=3, ) print([msg.text for msg in memories]) # 全量召回：query=None 返回作用域内全部记忆 all_memories = store.search_memories(query=None, user_id="alice") 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_id="alice", run_id="docs-assistant-session-1", infer=True, # 让 Mem0 自动提炼记忆 ) memories = store.search_memories( query="What project is Alice working on?", user_id="alice", run_id="docs-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) -> None

top_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_store=store, top_k=3) result = retriever.run(query="What does Alice like?", user_id="alice") memories = result["memories"] print([message.text for message in memories]) # query=None 全量召回作用域内记忆 all_memories = retriever.run(query=None, user_id="alice")["memories"]

Mem0MemoryWriter：写入记忆组件

Mem0MemoryWriter把对话消息持久化为记忆，其构造函数：

__init__(*, memory_store: Mem0MemoryStore, infer: bool = True) -> None

infer默认True（Mem0 自动抽取记忆），运行时可覆盖。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_store=store, infer=False) result = writer.run( messages=[ChatMessage.from_user("Alice prefers concise Python examples.")], user_id="alice", ) 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_store=store, infer=True)) memory_pipeline.add_component("retriever", Mem0MemoryRetriever(memory_store=store, top_k=5)) # 会话结束后写入 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 在运行时自动注入——这正是"一个工具实例服务多用户"的实现机制。

Mem0MemoryRetrieverTool：Agent 侧的记忆检索

构造函数：

__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 ) -> None

• name：暴露给 LLM 的工具名，默认retrieve_memories；
• description：工具描述，LLM 据此判断何时调用；
• parameters：暴露给 LLM 的 JSON Schema，默认只包含可选的query与top_k；
• inputs_from_state：从 Agent State 到工具参数的映射，默认{"user_id": "user_id"}，即把state["user_id"]注入工具的user_id参数。键是 Agent State 的键，值是工具的参数名。

核心方法retrieve：

retrieve( 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 ) -> str

• query省略或为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_store=store, top_k=5) agent = Agent( chat_generator=OpenAIChatGenerator(model="gpt-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_id="alice", session_id="chat-42", ) print(result["last_message"].text)

注入更多实体 ID：若要同时注入run_id、agent_id、app_id，需要两步：先在 Agent 的state_schema中声明对应状态字段，再传给工具的inputs_from_state。例如把会话 ID 映射到run_id：

retrieve_memories = Mem0MemoryRetrieverTool( memory_store=store, 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"}。

Mem0MemoryWriterTool：Agent 侧的记忆存储

构造函数：

__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和infer，inputs_from_state默认同样为{"user_id": "user_id"}。

核心方法store：

store( 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 ) -> str

• text：要存为记忆的信息；
• infer：True时 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_store=store) agent = Agent( chat_generator=OpenAIChatGenerator(model="gpt-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_id="alice", session_id="chat-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_dict
Mem0MemoryStore	序列化 store 配置（含 API key 的安全处理）
Mem0MemoryRetriever	序列化组件配置
Mem0MemoryWriter	序列化组件配置
Mem0MemoryRetrieverTool	序列化工具配置
Mem0MemoryWriterTool	序列化工具配置

这意味着包含记忆读写节点的 Pipeline 可以被保存、版本化、跨环境迁移，而api_key通过 Haystack 的Secret机制管理，不会以明文落入序列化文件。

错误处理与实战要点

• 异常类型：所有 Mem0 API 操作失败都会抛出Mem0MemoryStoreError（RuntimeError子类），捕获它即可统一处理网络/鉴权/限流等错误；
• 作用域校验：search_memories要求"有 filters 或至少一个实体 ID"，编写调用时务必满足，否则会直接报错；filters与 ID 同时给出时按 AND 组合；
• assistant 消息存储：要让 Mem0 保存助手角色的消息，必须设置agent_id；
• meta 处理的两条规则：写入时ChatMessage.meta被忽略（如需批量元数据，用metadatakwarg 传给 Mem0）；读取时用户元数据进入message.meta，Mem0 检索字段（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