Haystack 中的 Cognee 知识图谱记忆集成CogneeMemoryStore、CogneeRetriever 与 CogneeWriter 深度指南【免费下载链接】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 框架中的 Cognee 记忆集成cognee-haystack包以 Cognee 知识图谱 API 为后端为 Agent 与对话系统提供跨会话的长期记忆能力。你将掌握CogneeMemoryStore、CogneeRetriever、CogneeWriter三个核心组件的完整 API、两级记忆架构永久图谱与会话缓存、improve图谱增强机制以及如何将它们接入 Haystack Pipeline 构建记忆增强的 RAG/Agent 应用。背景为什么需要知识图谱记忆在构建生产级 LLM 应用时会话内上下文短期记忆只是记忆的一个维度。用户偏好、历史事实、跨会话的持久信息长期记忆通常需要独立于对话状态进行保存与检索。Haystack 将记忆抽象为memory store retriever/writer的组合Memory store负责记忆的持久化与查询是共享数据层Retriever在模型生成前将记忆注入 Agent 或 Chat GeneratorWriter在 Agent 或生成器完成一轮交互后将对话事实写入记忆。Cognee 是其中的一种记忆后端。它基于知识图谱运作写入时对文本做 LLM 抽取、构建实体与关系的图结构查询时支持图谱补全graph completion等更丰富的检索策略。这一集成的定位可以参考迁移指南中的说明——在与 LangGraph/LangChain 对比时Haystack 提供 Mem0MemoryStore 与 CogneeMemoryStore 用于跨会话持久化对话历史。本文所讲解内容的权威 API 参考见 Cognee 集成 API 参考version-2.19对应的使用指南见 CogneeMemoryStore 使用指南、CogneeRetriever 使用指南 与 CogneeWriter 使用指南。安装与前置配置Cognee 集成通过独立包分发安装命令pip install cognee-haystackCognee 自身的配置LLM 提供方、数据库、向量存储从环境变量读取。图抽取与查询依赖 LLM因此必须设置 LLM API 密钥export LLM_API_KEYyour-llm-api-key可选地设置独立的 embedding API 密钥不设置时默认回退到LLM_API_KEYexport EMBEDDING_API_KEYyour-embedding-api-key架构总览两级记忆与四个核心操作CogneeMemoryStore封装了 Cognee V2 记忆 API将四个方法直接映射到底层 Cognee 操作Haystack 方法Cognee 底层调用作用add_memoriescognee.remember将ChatMessage持久化为记忆search_memoriescognee.recall按自然语言查询召回记忆improvecognee.improve将会话缓存内容提升promote到永久图谱delete_all_memoriescognee.forget删除数据集dataset下的全部记忆永久图谱与会话缓存session_id决定记忆层级Cognee 支持两个记忆层级由session_id参数选择永久知识图谱permanent graphsession_id为None时使用。写入过程中 Cognee 会执行 LLM 抽取构建实体与关系的图节点支持图补全GRAPH_COMPLETION等丰富查询会话缓存session cache设置session_id后使用。写入快、无 LLM 抽取召回是会话感知session-aware的适合临时对话上下文。值得强调的是session_id的作用域是per-call每次调用级别的覆盖写入器Writer的session_id会覆盖 store 自身的session_id因此一个 store 可以同时支撑多个指向不同记忆层级的 writerstore 本身也可以在写入时按调用传入不同的session_id分别操作两个层级。这正是该集成在灵活性上的关键设计记忆的层级路由发生在调用点而不是绑定在 store 实例上。CogneeMemoryStore记忆后端核心初始化参数__init__( *, search_type: CogneeSearchType GRAPH_COMPLETION, top_k: int 5, dataset_name: str haystack_memory, session_id: str | None None, self_improvement: bool True, timeout: float 300 ) - None各参数说明如下参数默认值含义与建议search_typeGRAPH_COMPLETIONCogneesearch_memories使用的检索策略。GRAPH_COMPLETION为图补全其他常用值包括CHUNKS原始块检索与SUMMARIES返回摘要后的图节点top_k5search_memories的默认最大返回条数可被调用级参数覆盖dataset_namehaystack_memory该 store 背后对应的 Cognee 数据集名称session_idNone设置后读写目标为会话缓存层级None时使用永久知识图谱self_improvementTrue转发给cognee.remember默认True与 cognee 一致。置为False时improve()成为唯一的图谱增强触发点timeout300任何单个 Cognee 调用的超时上限秒超时抛concurrent.futures.TimeoutError。单消息的 Agent 记忆写入默认值 300s 绰绰有余长文档的批量摄入可能需要调大add_memories写入记忆add_memories( *, messages: list[ChatMessage], user_id: str | None None, session_id: str | None None ) - None通过cognee.remember持久化消息。实现细节上两个层级采取不同的批处理策略永久层级将所有文本合并为一次调用批量写入会话层级每条消息单独写入一条记录与 cognee 官方会话示例一致。空消息会被跳过。user_id用于将记忆归属到特定 Cognee 用户传None时使用 cognee 默认用户session_id为本次调用的层级覆盖。search_memories检索记忆search_memories( *, query: str | None None, top_k: int | None None, user_id: str | None None ) - list[ChatMessage]通过cognee.recall检索每个命中结果被包装成一条 system 角色的ChatMessage返回。注意边界行为query为空或None时直接返回空列表[]。top_k为调用级覆盖未传时回落到 store 默认值。improve将会话缓存提升到永久图谱improve(*, session_id: str | None None, user_id: str | None None) - None通过cognee.improve将会话缓存内容提升到永久图谱session_id缺省时使用 store 自身的session_id。不带任何session_id的调用是一次纯粹的图谱增强graph-enrichment过程。关于self_improvement有一个易踩的坑值得单独说明当self_improvementTrue默认时永久层级会在写入后内联等待improve完成会话层级则将improve作为 fire-and-forget 后台任务调度。若你希望improve()是唯一的图谱增强触发点务必把self_improvement设为False——否则显式调用一次improve()会导致 improve 被执行两次产生近乎重复的图节点。delete_all_memories删除记忆delete_all_memories(*, user_id: str | None None) - None通过cognee.forget(dataset...)删除该 store 对应的数据集。注意会话缓存不受影响会话不按数据集划分作用域。若需彻底清空包括会话缓存需要直接调用 cogneeimport asyncio import cognee asyncio.run(cognee.forget(everythingTrue))序列化支持store 实现了to_dict()与from_dict(data)用于 Pipeline 的持久化YAML 序列化/反序列化场景保证组件可被完整地保存与恢复。CogneeRetriever记忆注入适配器__init__(*, memory_store: CogneeMemoryStore, top_k: int | None None) - NoneCogneeRetriever是一个薄的 Pipeline 适配器搜索行为search_type、dataset_name、session_id等全部配置在 store 上retriever 只负责把query转发给search_memories并返回ChatMessage列表。memory_store要查询的后端 store必填top_k默认最大结果数传None时回落到 store 的top_k。run( query: str, top_k: int | None None, user_id: str | None None ) - dict[str, list[ChatMessage]]run 参数类型说明querystr自然语言查询top_kint \| None单次调用覆盖值回退链为调用级top_k→ init 时top_k→ store 默认值user_idstr \| NoneCognee 用户 UUID将检索作用域限定到该用户输出为{messages: [system ChatMessage, ...]}。在记忆增强的 Pipeline 中它通常位于 Agent 或 Chat Generator之前把长期记忆作为 system 消息注入对话。独立使用示例from haystack.dataclasses import ChatMessage from haystack_integrations.components.retrievers.cognee import CogneeRetriever from haystack_integrations.memory_stores.cognee import CogneeMemoryStore store CogneeMemoryStore(search_typeGRAPH_COMPLETION, top_k5) # 先写入一些记忆 store.add_memories( messages[ChatMessage.from_user(Alice prefers concise Python examples.)], user_ida1b2c3d4-e5f6-7890-abcd-ef1234567890, ) retriever CogneeRetriever(memory_storestore, top_k3) result retriever.run( queryWhat does Alice prefer?, user_ida1b2c3d4-e5f6-7890-abcd-ef1234567890, ) memories result[messages] print([message.text for message in memories])接入 Agent Pipeline记忆前置注入下面的完整示例展示了标准用法先用 retriever 召回记忆再用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.dataclasses import ChatMessage from haystack_integrations.components.retrievers.cognee import CogneeRetriever from haystack_integrations.memory_stores.cognee import CogneeMemoryStore store CogneeMemoryStore(dataset_namemy_agent_memory, session_idalice_session_1) pipeline Pipeline() pipeline.add_component(retriever, CogneeRetriever(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. ), ), ) pipeline.connect(retriever.messages, memory_context.memories) pipeline.connect(memory_context.output, agent.messages) query Give me a short implementation tip. pipeline.run( { retriever: { query: query, user_id: a1b2c3d4-e5f6-7890-abcd-ef1234567890, }, memory_context: { user_messages: [ChatMessage.from_user(query)], }, } )注意这里的 store 配置了session_idalice_session_1因此检索发生在会话缓存层级。若希望从永久图谱检索保持session_idNone即可——这也是同一 store 通过不同组件/调用覆盖层级设计思想的体现。CogneeWriter对话事实持久化__init__(*, memory_store: CogneeMemoryStore, session_id: str | None None) - NoneCogneeWriter将ChatMessage列表持久化到CogneeMemoryStore用于在 Agent 或生成器完成一轮交互后保存对话事实与用户偏好。memory_store要写入的后端 store必填session_id覆盖 store 的session_id仅作用于该 writer 的写入。省略或传None写入永久图谱设置后写入会话缓存。run( messages: list[ChatMessage], user_id: str | None None ) - dict[str, list[ChatMessage]]run 参数类型说明messageslist[ChatMessage]要持久化的消息user_idstr \| NoneCognee 用户 UUID将写入作用域限定到该用户输出为{messages_written: list[ChatMessage]}——消息原样透传pass-through因此 writer 可以安全地链在 Agent 或生成器之后而不破坏 Pipeline 的消息流。独立使用示例from haystack.dataclasses import ChatMessage from haystack_integrations.components.writers.cognee import CogneeWriter from haystack_integrations.memory_stores.cognee import CogneeMemoryStore store CogneeMemoryStore() writer CogneeWriter(memory_storestore) result writer.run( messages[ChatMessage.from_user(Alice prefers concise Python examples.)], user_ida1b2c3d4-e5f6-7890-abcd-ef1234567890, ) print(result[messages_written])若想写入会话缓存而不是永久图谱给 writer 传session_idsession_writer CogneeWriter(memory_storestore, session_idalice_session_1) session_writer.run( messages[ ChatMessage.from_user(Alice is currently debugging a vector store issue.) ], user_ida1b2c3d4-e5f6-7890-abcd-ef1234567890, )接入 Agent Pipeline写入记忆将 Agent 的完整messages输出连接到CogneeWriter让 Cognee 把整轮对话存储到永久图谱from haystack import Pipeline from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack_integrations.components.writers.cognee import CogneeWriter from haystack_integrations.memory_stores.cognee import CogneeMemoryStore store CogneeMemoryStore(dataset_namemy_agent_memory) 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. ), ), ) pipeline.add_component(writer, CogneeWriter(memory_storestore)) 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: a1b2c3d4-e5f6-7890-abcd-ef1234567890, }, }, ) print(result[writer][messages_written])两级记忆协作完整示例会话级写入 提升到永久图谱结合CogneeMemoryStore的方法级session_id覆盖与improve()可以构建临时上下文入会话缓存、重要事实沉淀为长期图谱的完整记忆策略。以下示例取自 CogneeMemoryStore 使用指南from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.cognee import CogneeMemoryStore store CogneeMemoryStore(dataset_namemy_agent_memory, self_improvementFalse) # 长期事实写入永久图谱不带 session_id。 store.add_memories( messages[ChatMessage.from_user(Alice is a senior data scientist at Acme Corp.)], ) # 临时会话上下文写入会话缓存。 store.add_memories( messages[ ChatMessage.from_user(Alice is currently debugging a vector store issue.) ], session_idalice_session_1, ) # 将会话缓存提升到永久图谱。 store.improve(session_idalice_session_1)这里self_improvementFalse是刻意为之确保improve()是唯一的图谱增强触发点避免重复提升产生近似重复的图节点。组合成记忆增强闭环将上述组件拼装即可得到一个完整的记忆增强 Agent闭环CogneeRetriever在模型生成前把与该用户相关的历史记忆systemChatMessage注入 Agent 上下文Agent 结合记忆回答问题CogneeWriter在每轮交互后将对话中的新事实写入 store定期调用store.improve()或依赖self_improvementTrue把值得沉淀的会话内容提升到永久图谱。由于 writer 的session_id覆盖是 per-call 的同一个CogneeMemoryStore实例可以被多个 writer 同时用于写永久事实和写会话上下文两种职责无需创建多个 store 或数据集。这是该集成在生产中组织记忆分级的关键手段。边界行为与注意事项汇总以下行为均来自 API 参考文档version-2.19 参考在实际使用时务必留意空查询search_memories的query为空或None时返回[]不会触发底层召回空消息跳过add_memories会跳过空消息批处理差异永久层级批量合并文本为单次调用会话层级逐条写入双 improve 陷阱self_improvementTrue时显式调用improve()会导致 improve 被执行两次并产生近似重复图节点需要显式控制提升时机时务必设置self_improvementFalse删除作用域delete_all_memories只删除数据集会话缓存不受影响彻底清空需直接调用cognee.forget(everythingTrue)超时任何单次 cognee 调用超过timeout默认 300s会抛concurrent.futures.TimeoutError长文档批量摄入建议调大层级覆盖规则writer 的session_id覆盖 store 的session_idretriever 的top_k覆盖链为调用级 → init 级 → store 默认值检索结果角色search_memories将每个命中包装为 system 角色ChatMessage可直接作为上下文注入。序列化与持久化三个组件CogneeMemoryStore、CogneeRetriever、CogneeWriter均实现了标准的to_dict()/from_dict(data)接口用于 Pipeline 的序列化与反序列化。这意味着包含 Cognee 记忆组件的 Pipeline 可以完整地导出为字典并进一步由 Haystack 的 marshaller 转为 YAML 等格式保存与恢复。底层上CogneeMemoryStore需要保存search_type、top_k、dataset_name、session_id、self_improvement、timeout等配置以在恢复时重建等价实例retriever 与 writer 则保存其关联 store 的序列化形式及各自的覆盖参数。延伸阅读CogneeMemoryStore 使用指南两级记忆、参数表与删除/清空示例CogneeRetriever 使用指南retriever 的独立使用与 Pipeline 注入示例CogneeWriter 使用指南writer 的独立使用与 Pipeline 持久化示例Cognee 集成 API 参考当前版本本文所述 API 的现行权威参考本文基于 version-2.19 的版本化参考撰写接口与该参考保持一致记忆存储总览含 Mem0 对比Haystack 提供的另一类长期记忆后端可用于选型对比从 LangGraph/LangChain 迁移视角看记忆设计迁移指南其中将 Agent State 视为短期记忆、将 Cognee/Mem0 memory store 视为长期记忆。【免费下载链接】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 中的 Cognee 知识图谱记忆集成：CogneeMemoryStore、CogneeRetriever 与 CogneeWriter 深度指南
Haystack 中的 Cognee 知识图谱记忆集成CogneeMemoryStore、CogneeRetriever 与 CogneeWriter 深度指南【免费下载链接】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 框架中的 Cognee 记忆集成cognee-haystack包以 Cognee 知识图谱 API 为后端为 Agent 与对话系统提供跨会话的长期记忆能力。你将掌握CogneeMemoryStore、CogneeRetriever、CogneeWriter三个核心组件的完整 API、两级记忆架构永久图谱与会话缓存、improve图谱增强机制以及如何将它们接入 Haystack Pipeline 构建记忆增强的 RAG/Agent 应用。背景为什么需要知识图谱记忆在构建生产级 LLM 应用时会话内上下文短期记忆只是记忆的一个维度。用户偏好、历史事实、跨会话的持久信息长期记忆通常需要独立于对话状态进行保存与检索。Haystack 将记忆抽象为memory store retriever/writer的组合Memory store负责记忆的持久化与查询是共享数据层Retriever在模型生成前将记忆注入 Agent 或 Chat GeneratorWriter在 Agent 或生成器完成一轮交互后将对话事实写入记忆。Cognee 是其中的一种记忆后端。它基于知识图谱运作写入时对文本做 LLM 抽取、构建实体与关系的图结构查询时支持图谱补全graph completion等更丰富的检索策略。这一集成的定位可以参考迁移指南中的说明——在与 LangGraph/LangChain 对比时Haystack 提供 Mem0MemoryStore 与 CogneeMemoryStore 用于跨会话持久化对话历史。本文所讲解内容的权威 API 参考见 Cognee 集成 API 参考version-2.19对应的使用指南见 CogneeMemoryStore 使用指南、CogneeRetriever 使用指南 与 CogneeWriter 使用指南。安装与前置配置Cognee 集成通过独立包分发安装命令pip install cognee-haystackCognee 自身的配置LLM 提供方、数据库、向量存储从环境变量读取。图抽取与查询依赖 LLM因此必须设置 LLM API 密钥export LLM_API_KEYyour-llm-api-key可选地设置独立的 embedding API 密钥不设置时默认回退到LLM_API_KEYexport EMBEDDING_API_KEYyour-embedding-api-key架构总览两级记忆与四个核心操作CogneeMemoryStore封装了 Cognee V2 记忆 API将四个方法直接映射到底层 Cognee 操作Haystack 方法Cognee 底层调用作用add_memoriescognee.remember将ChatMessage持久化为记忆search_memoriescognee.recall按自然语言查询召回记忆improvecognee.improve将会话缓存内容提升promote到永久图谱delete_all_memoriescognee.forget删除数据集dataset下的全部记忆永久图谱与会话缓存session_id决定记忆层级Cognee 支持两个记忆层级由session_id参数选择永久知识图谱permanent graphsession_id为None时使用。写入过程中 Cognee 会执行 LLM 抽取构建实体与关系的图节点支持图补全GRAPH_COMPLETION等丰富查询会话缓存session cache设置session_id后使用。写入快、无 LLM 抽取召回是会话感知session-aware的适合临时对话上下文。值得强调的是session_id的作用域是per-call每次调用级别的覆盖写入器Writer的session_id会覆盖 store 自身的session_id因此一个 store 可以同时支撑多个指向不同记忆层级的 writerstore 本身也可以在写入时按调用传入不同的session_id分别操作两个层级。这正是该集成在灵活性上的关键设计记忆的层级路由发生在调用点而不是绑定在 store 实例上。CogneeMemoryStore记忆后端核心初始化参数__init__( *, search_type: CogneeSearchType GRAPH_COMPLETION, top_k: int 5, dataset_name: str haystack_memory, session_id: str | None None, self_improvement: bool True, timeout: float 300 ) - None各参数说明如下参数默认值含义与建议search_typeGRAPH_COMPLETIONCogneesearch_memories使用的检索策略。GRAPH_COMPLETION为图补全其他常用值包括CHUNKS原始块检索与SUMMARIES返回摘要后的图节点top_k5search_memories的默认最大返回条数可被调用级参数覆盖dataset_namehaystack_memory该 store 背后对应的 Cognee 数据集名称session_idNone设置后读写目标为会话缓存层级None时使用永久知识图谱self_improvementTrue转发给cognee.remember默认True与 cognee 一致。置为False时improve()成为唯一的图谱增强触发点timeout300任何单个 Cognee 调用的超时上限秒超时抛concurrent.futures.TimeoutError。单消息的 Agent 记忆写入默认值 300s 绰绰有余长文档的批量摄入可能需要调大add_memories写入记忆add_memories( *, messages: list[ChatMessage], user_id: str | None None, session_id: str | None None ) - None通过cognee.remember持久化消息。实现细节上两个层级采取不同的批处理策略永久层级将所有文本合并为一次调用批量写入会话层级每条消息单独写入一条记录与 cognee 官方会话示例一致。空消息会被跳过。user_id用于将记忆归属到特定 Cognee 用户传None时使用 cognee 默认用户session_id为本次调用的层级覆盖。search_memories检索记忆search_memories( *, query: str | None None, top_k: int | None None, user_id: str | None None ) - list[ChatMessage]通过cognee.recall检索每个命中结果被包装成一条 system 角色的ChatMessage返回。注意边界行为query为空或None时直接返回空列表[]。top_k为调用级覆盖未传时回落到 store 默认值。improve将会话缓存提升到永久图谱improve(*, session_id: str | None None, user_id: str | None None) - None通过cognee.improve将会话缓存内容提升到永久图谱session_id缺省时使用 store 自身的session_id。不带任何session_id的调用是一次纯粹的图谱增强graph-enrichment过程。关于self_improvement有一个易踩的坑值得单独说明当self_improvementTrue默认时永久层级会在写入后内联等待improve完成会话层级则将improve作为 fire-and-forget 后台任务调度。若你希望improve()是唯一的图谱增强触发点务必把self_improvement设为False——否则显式调用一次improve()会导致 improve 被执行两次产生近乎重复的图节点。delete_all_memories删除记忆delete_all_memories(*, user_id: str | None None) - None通过cognee.forget(dataset...)删除该 store 对应的数据集。注意会话缓存不受影响会话不按数据集划分作用域。若需彻底清空包括会话缓存需要直接调用 cogneeimport asyncio import cognee asyncio.run(cognee.forget(everythingTrue))序列化支持store 实现了to_dict()与from_dict(data)用于 Pipeline 的持久化YAML 序列化/反序列化场景保证组件可被完整地保存与恢复。CogneeRetriever记忆注入适配器__init__(*, memory_store: CogneeMemoryStore, top_k: int | None None) - NoneCogneeRetriever是一个薄的 Pipeline 适配器搜索行为search_type、dataset_name、session_id等全部配置在 store 上retriever 只负责把query转发给search_memories并返回ChatMessage列表。memory_store要查询的后端 store必填top_k默认最大结果数传None时回落到 store 的top_k。run( query: str, top_k: int | None None, user_id: str | None None ) - dict[str, list[ChatMessage]]run 参数类型说明querystr自然语言查询top_kint \| None单次调用覆盖值回退链为调用级top_k→ init 时top_k→ store 默认值user_idstr \| NoneCognee 用户 UUID将检索作用域限定到该用户输出为{messages: [system ChatMessage, ...]}。在记忆增强的 Pipeline 中它通常位于 Agent 或 Chat Generator之前把长期记忆作为 system 消息注入对话。独立使用示例from haystack.dataclasses import ChatMessage from haystack_integrations.components.retrievers.cognee import CogneeRetriever from haystack_integrations.memory_stores.cognee import CogneeMemoryStore store CogneeMemoryStore(search_typeGRAPH_COMPLETION, top_k5) # 先写入一些记忆 store.add_memories( messages[ChatMessage.from_user(Alice prefers concise Python examples.)], user_ida1b2c3d4-e5f6-7890-abcd-ef1234567890, ) retriever CogneeRetriever(memory_storestore, top_k3) result retriever.run( queryWhat does Alice prefer?, user_ida1b2c3d4-e5f6-7890-abcd-ef1234567890, ) memories result[messages] print([message.text for message in memories])接入 Agent Pipeline记忆前置注入下面的完整示例展示了标准用法先用 retriever 召回记忆再用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.dataclasses import ChatMessage from haystack_integrations.components.retrievers.cognee import CogneeRetriever from haystack_integrations.memory_stores.cognee import CogneeMemoryStore store CogneeMemoryStore(dataset_namemy_agent_memory, session_idalice_session_1) pipeline Pipeline() pipeline.add_component(retriever, CogneeRetriever(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. ), ), ) pipeline.connect(retriever.messages, memory_context.memories) pipeline.connect(memory_context.output, agent.messages) query Give me a short implementation tip. pipeline.run( { retriever: { query: query, user_id: a1b2c3d4-e5f6-7890-abcd-ef1234567890, }, memory_context: { user_messages: [ChatMessage.from_user(query)], }, } )注意这里的 store 配置了session_idalice_session_1因此检索发生在会话缓存层级。若希望从永久图谱检索保持session_idNone即可——这也是同一 store 通过不同组件/调用覆盖层级设计思想的体现。CogneeWriter对话事实持久化__init__(*, memory_store: CogneeMemoryStore, session_id: str | None None) - NoneCogneeWriter将ChatMessage列表持久化到CogneeMemoryStore用于在 Agent 或生成器完成一轮交互后保存对话事实与用户偏好。memory_store要写入的后端 store必填session_id覆盖 store 的session_id仅作用于该 writer 的写入。省略或传None写入永久图谱设置后写入会话缓存。run( messages: list[ChatMessage], user_id: str | None None ) - dict[str, list[ChatMessage]]run 参数类型说明messageslist[ChatMessage]要持久化的消息user_idstr \| NoneCognee 用户 UUID将写入作用域限定到该用户输出为{messages_written: list[ChatMessage]}——消息原样透传pass-through因此 writer 可以安全地链在 Agent 或生成器之后而不破坏 Pipeline 的消息流。独立使用示例from haystack.dataclasses import ChatMessage from haystack_integrations.components.writers.cognee import CogneeWriter from haystack_integrations.memory_stores.cognee import CogneeMemoryStore store CogneeMemoryStore() writer CogneeWriter(memory_storestore) result writer.run( messages[ChatMessage.from_user(Alice prefers concise Python examples.)], user_ida1b2c3d4-e5f6-7890-abcd-ef1234567890, ) print(result[messages_written])若想写入会话缓存而不是永久图谱给 writer 传session_idsession_writer CogneeWriter(memory_storestore, session_idalice_session_1) session_writer.run( messages[ ChatMessage.from_user(Alice is currently debugging a vector store issue.) ], user_ida1b2c3d4-e5f6-7890-abcd-ef1234567890, )接入 Agent Pipeline写入记忆将 Agent 的完整messages输出连接到CogneeWriter让 Cognee 把整轮对话存储到永久图谱from haystack import Pipeline from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack_integrations.components.writers.cognee import CogneeWriter from haystack_integrations.memory_stores.cognee import CogneeMemoryStore store CogneeMemoryStore(dataset_namemy_agent_memory) 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. ), ), ) pipeline.add_component(writer, CogneeWriter(memory_storestore)) 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: a1b2c3d4-e5f6-7890-abcd-ef1234567890, }, }, ) print(result[writer][messages_written])两级记忆协作完整示例会话级写入 提升到永久图谱结合CogneeMemoryStore的方法级session_id覆盖与improve()可以构建临时上下文入会话缓存、重要事实沉淀为长期图谱的完整记忆策略。以下示例取自 CogneeMemoryStore 使用指南from haystack.dataclasses import ChatMessage from haystack_integrations.memory_stores.cognee import CogneeMemoryStore store CogneeMemoryStore(dataset_namemy_agent_memory, self_improvementFalse) # 长期事实写入永久图谱不带 session_id。 store.add_memories( messages[ChatMessage.from_user(Alice is a senior data scientist at Acme Corp.)], ) # 临时会话上下文写入会话缓存。 store.add_memories( messages[ ChatMessage.from_user(Alice is currently debugging a vector store issue.) ], session_idalice_session_1, ) # 将会话缓存提升到永久图谱。 store.improve(session_idalice_session_1)这里self_improvementFalse是刻意为之确保improve()是唯一的图谱增强触发点避免重复提升产生近似重复的图节点。组合成记忆增强闭环将上述组件拼装即可得到一个完整的记忆增强 Agent闭环CogneeRetriever在模型生成前把与该用户相关的历史记忆systemChatMessage注入 Agent 上下文Agent 结合记忆回答问题CogneeWriter在每轮交互后将对话中的新事实写入 store定期调用store.improve()或依赖self_improvementTrue把值得沉淀的会话内容提升到永久图谱。由于 writer 的session_id覆盖是 per-call 的同一个CogneeMemoryStore实例可以被多个 writer 同时用于写永久事实和写会话上下文两种职责无需创建多个 store 或数据集。这是该集成在生产中组织记忆分级的关键手段。边界行为与注意事项汇总以下行为均来自 API 参考文档version-2.19 参考在实际使用时务必留意空查询search_memories的query为空或None时返回[]不会触发底层召回空消息跳过add_memories会跳过空消息批处理差异永久层级批量合并文本为单次调用会话层级逐条写入双 improve 陷阱self_improvementTrue时显式调用improve()会导致 improve 被执行两次并产生近似重复图节点需要显式控制提升时机时务必设置self_improvementFalse删除作用域delete_all_memories只删除数据集会话缓存不受影响彻底清空需直接调用cognee.forget(everythingTrue)超时任何单次 cognee 调用超过timeout默认 300s会抛concurrent.futures.TimeoutError长文档批量摄入建议调大层级覆盖规则writer 的session_id覆盖 store 的session_idretriever 的top_k覆盖链为调用级 → init 级 → store 默认值检索结果角色search_memories将每个命中包装为 system 角色ChatMessage可直接作为上下文注入。序列化与持久化三个组件CogneeMemoryStore、CogneeRetriever、CogneeWriter均实现了标准的to_dict()/from_dict(data)接口用于 Pipeline 的序列化与反序列化。这意味着包含 Cognee 记忆组件的 Pipeline 可以完整地导出为字典并进一步由 Haystack 的 marshaller 转为 YAML 等格式保存与恢复。底层上CogneeMemoryStore需要保存search_type、top_k、dataset_name、session_id、self_improvement、timeout等配置以在恢复时重建等价实例retriever 与 writer 则保存其关联 store 的序列化形式及各自的覆盖参数。延伸阅读CogneeMemoryStore 使用指南两级记忆、参数表与删除/清空示例CogneeRetriever 使用指南retriever 的独立使用与 Pipeline 注入示例CogneeWriter 使用指南writer 的独立使用与 Pipeline 持久化示例Cognee 集成 API 参考当前版本本文所述 API 的现行权威参考本文基于 version-2.19 的版本化参考撰写接口与该参考保持一致记忆存储总览含 Mem0 对比Haystack 提供的另一类长期记忆后端可用于选型对比从 LangGraph/LangChain 迁移视角看记忆设计迁移指南其中将 Agent State 视为短期记忆、将 Cognee/Mem0 memory store 视为长期记忆。【免费下载链接】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），仅供参考