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/haystackCognee 是面向 LLM 应用的知识图谱记忆后端而 Haystack 通过cognee-haystack集成包将其封装为三个可组合的组件持久化记忆层CogneeMemoryStore、检索组件CogneeRetriever与写入组件CogneeWriter。本文以仓库内 Cognee 集成 API 参考 为核心骨架结合 CogneeMemoryStore 使用指南、CogneeRetriever 使用指南 与 CogneeWriter 使用指南完整讲解三个组件的参数语义、双记忆层级会话缓存 / 永久知识图谱机制并给出可在 Pipeline 与 Agent 工作流中直接运行的代码示例。读完本文你将能够在 Haystack 中搭建写入对话记忆 → 检索长期记忆 → 注入 Agent 上下文的完整记忆增强链路。集成全景三个组件如何协作Cognee 集成遵循 Haystack 记忆类组件的标准分工模式CogneeMemoryStore是共享数据层封装了 Cognee V2 记忆 API 的全部底层操作被另外两个组件共同依赖CogneeWriter负责把对话消息ChatMessage写入记忆通常放置在 Agent 或 Chat Generator 之后CogneeRetriever负责读取记忆并转换为ChatMessage列表通常放置在 Agent 或 Chat Generator 之前。仓库内三份组件文档的关键属性对比如下组件必填 init 参数可选 init 参数运行输入运行输出典型位置CogneeMemoryStore无search_type、top_k、dataset_name、session_id、self_improvement、timeout各方法自有参数见方法说明数据层被二者共享CogneeRetrievermemory_storetop_kquery、user_idmessagesAgent / Chat Generator 之前CogneeWritermemory_storesession_idmessages、user_idmessages_writtenAgent / Chat Generator 之后安装与环境变量集成以独立包cognee-haystack发布安装命令为pip install cognee-haystackCognee 在写入LLM 抽取知识图谱与查询阶段都需要 LLM 服务其配置LLM 提供方、数据库、向量存储全部通过环境变量读取。最低限度的配置是设置 LLM API Keyexport LLM_API_KEYyour-llm-api-key可选地可以单独指定 Embedding API Key未设置时默认回退到LLM_API_KEYexport EMBEDDING_API_KEYyour-embedding-api-key注意仓库内 CogneeMemoryStore 文档 明确说明 LLM 提供方、数据库与向量存储的完整配置均来自环境变量具体项请参考 Cognee 官方文档完成初始化。CogneeMemoryStore知识图谱记忆的数据层CogneeMemoryStore是记忆后端的核心封装。它把 Cognee V2 记忆 API 做了四对一的薄封装见 API 参考CogneeMemoryStore 方法底层 Cognee API职责add_memoriescognee.remember持久化消息search_memoriescognee.recall检索记忆improvecognee.improve图谱增强 / 会话提升delete_all_memoriescognee.forget删除数据集初始化参数__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各参数语义均来自 API 参考文档search_type可选默认GRAPH_COMPLETIONsearch_memories使用的 Cognee 召回策略。除图补全外常用取值还有CHUNKS原始片段召回与SUMMARIES图节点的摘要召回。top_k可选默认5search_memories的默认最大返回条数可被每次调用覆盖。dataset_name可选默认haystack_memory该 store 对应的 Cognee 数据集名称。session_id可选默认None设置后读写指向会话缓存层级为None时使用永久知识图谱。self_improvement可选默认True透传给cognee.remember与 cognee 默认值一致。为True时在永久层级上每次写入后内联等待improve执行完毕在会话层级上则以 fire-and-forget 后台任务方式调度improve。timeout可选默认300任何单个 cognee 调用的超时上限秒超时抛concurrent.futures.TimeoutError。默认 300 秒可从容覆盖单条消息的 agent 记忆写入对长文档的批量摄取可能需要调大。方法详解add_memories通过cognee.remember持久化消息。add_memories(*, messages: list[ChatMessage], user_id: str | None None, session_id: str | None None) - None永久层级会把所有文本合并为一次调用批量写入会话层级则为每条消息单独写一条与 cognee 官方会话示例保持一致。空消息会被跳过。user_id为 None 时使用 cognee 的默认用户session_id可对 store 级配置做单次覆盖。search_memories通过cognee.recall检索并把每条命中包装为一条system 角色的ChatMessage。search_memories(*, query: str | None None, top_k: int | None None, user_id: str | None None) - list[ChatMessage]query为空或None时直接返回[]。top_k为单次覆盖缺省时用 store 默认值。improve通过cognee.improve把会话缓存内容提升到永久知识图谱。improve(*, session_id: str | None None, user_id: str | None None) - None不带session_id时是一次纯粹的图谱增强enrichment操作session_id缺省时取 store 自身的session_id。delete_all_memories通过cognee.forget(dataset...)删除该数据集。delete_all_memories(*, user_id: str | None None) - None会话缓存不受影响会话并非按数据集作用域隔离如需完整清空须直接调用cognee.forget(everythingTrue)。双记忆层级会话缓存与永久知识图谱这是该集成最核心的设计。通过session_id选择目标层级永久知识图谱session_idNone写入时 Cognee 执行 LLM 抽取生成富含实体关系的图谱节点支持更丰富的图补全GRAPH_COMPLETION查询读取走cognee.recall。会话缓存设置session_id写入速度快、不做 LLM 抽取召回是会话感知的内容后续可通过improve()提升为永久记忆。使用指南 给出了一段同时使用两个层级的示例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)CogneeRetriever把记忆注入 Agent 上下文CogneeRetriever是search_memories之上的薄管道适配器API 参考搜索行为search_type、数据集、会话层级全部配置在 store 上__init__(*, memory_store: CogneeMemoryStore, top_k: int | None None) - None run(query: str, top_k: int | None None, user_id: str | None None) - dict[str, list[ChatMessage]]top_k优先取每次调用传入值其次取 init 时的值最后回退到 store 默认值三级回退链。user_id把检索限定到特定 Cognee 用户None时使用 cognee 默认用户。返回的messages均为 system 角色ChatMessage可直接拼接到 Agent 的消息列表开头充当长期记忆。独立使用示例retriever 文档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])在 Pipeline 中为 Agent 注入记忆retriever 文档 给出了完整的记忆增强管道检索记忆 → 用OutputAdapter把记忆 当前用户消息合并 → 送入 Agent。其中OutputAdapter使用 Jinja 模板{{ memories user_messages }}将两组ChatMessage拼接output_typelist[ChatMessage]且需unsafeTruefrom 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)], }, } )这里的Agent与OpenAIChatGenerator均来自仓库主库见 Agent 组件源码 与 ChatMessage 数据类记忆检索结果以 system 消息形式排在对话开头Agent 的系统提示词明确指示模型将开头的 system 消息视作长期记忆使用。CogneeWriter把对话回合持久化为记忆CogneeWriter把ChatMessage列表写入CogneeMemoryStoreAPI 参考通常放在 Agent 回合之后__init__(*, memory_store: CogneeMemoryStore, session_id: str | None None) - None run(messages: list[ChatMessage], user_id: str | None None) - dict[str, list[ChatMessage]]透传语义写入的消息会原样透传到输出键messages_written因此可以安全地串联在 Agent 或 Generator 之后而不破坏管道数据流。层级覆盖writer 的session_id会在每次调用时覆盖 store 自身的session_id因此一个 store 可以支撑多个 writer 写入不同层级——这是该组件最重要的设计点。不传session_id或为None写入永久图谱LLM 抽取、图补全就绪传入session_id则写入会话缓存快、无抽取之后可用store.improve()提升。独立使用示例writer 文档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])写入会话缓存只需在 init 时传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, )在 Pipeline 中记录 Agent 回合writer 文档 演示了把 Agent 的完整messages输出接到 writer 上使 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])self_improvement 与 improve() 的正确用法这是最容易踩坑的地方API 参考给出了明确的语义默认self_improvementTrue与 cognee 一致。此时每次remember写入后永久层级内联等待improve完成会话层级则以 fire-and-forget 后台任务调度improve。如果业务上把improve()作为唯一的增强触发点必须显式设置self_improvementFalse。否则显式调用improve()会与写入时自动触发的 improve叠加执行两次产生近重复的图谱节点near-duplicate graph nodes。典型推荐组合永久记忆写入 手动批量提升时store 配置self_improvementFalse写完后由你主动调用improve()。删除记忆与清理delete_all_memories()只删除该 store 的dataset_name数据集内部走cognee.forget(dataset...)。由于会话缓存不属于数据集作用域它会在本次删除中存活下来。需要完整清空含会话缓存时直接调用 cognee 原生 API# 只删除当前数据集会话缓存不受影响。 store.delete_all_memories() # 完整清空含会话缓存。 import asyncio import cognee asyncio.run(cognee.forget(everythingTrue))实战要点与边界行为汇总三级 top_k 回退链retriever 单次调用top_k→ retriever inittop_k→ storetop_k默认5。空查询短路search_memories对空/None的query直接返回[]避免无谓的 LLM 调用。空消息跳过add_memories跳过空消息永久层级批量合并为一次remember调用会话层级逐条写入。超时保护timeout默认 300s对每个 cognee 调用生效批量摄取长文档时应调大该值。system 角色输出search_memories把每条命中包装为 system 角色ChatMessage与 Agent 的 system-prompt 语义天然兼容。用户隔离user_id贯穿写入、检索与删除全流程None时落到 cognee 默认用户多用户场景请务必显式传入 Cognee 用户 UUID。相关资源Cognee 集成 API 参考三个组件的完整签名与参数文档CogneeMemoryStore 使用指南数据层参数与双层级示例CogneeRetriever 使用指南检索器独立与 Pipeline 用法CogneeWriter 使用指南写入器独立与 Pipeline 用法Agent 组件源码 与 ChatMessage 数据类记忆注入的目标组件【免费下载链接】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），仅供参考 
长河君 · 2026/9/12 2:57:47 · 阅读 1.2k 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/haystackCognee 是面向 LLM 应用的知识图谱记忆后端而 Haystack 通过cognee-haystack集成包将其封装为三个可组合的组件持久化记忆层CogneeMemoryStore、检索组件CogneeRetriever与写入组件CogneeWriter。本文以仓库内 Cognee 集成 API 参考 为核心骨架结合 CogneeMemoryStore 使用指南、CogneeRetriever 使用指南 与 CogneeWriter 使用指南完整讲解三个组件的参数语义、双记忆层级会话缓存 / 永久知识图谱机制并给出可在 Pipeline 与 Agent 工作流中直接运行的代码示例。读完本文你将能够在 Haystack 中搭建写入对话记忆 → 检索长期记忆 → 注入 Agent 上下文的完整记忆增强链路。集成全景三个组件如何协作Cognee 集成遵循 Haystack 记忆类组件的标准分工模式CogneeMemoryStore是共享数据层封装了 Cognee V2 记忆 API 的全部底层操作被另外两个组件共同依赖CogneeWriter负责把对话消息ChatMessage写入记忆通常放置在 Agent 或 Chat Generator 之后CogneeRetriever负责读取记忆并转换为ChatMessage列表通常放置在 Agent 或 Chat Generator 之前。仓库内三份组件文档的关键属性对比如下组件必填 init 参数可选 init 参数运行输入运行输出典型位置CogneeMemoryStore无search_type、top_k、dataset_name、session_id、self_improvement、timeout各方法自有参数见方法说明数据层被二者共享CogneeRetrievermemory_storetop_kquery、user_idmessagesAgent / Chat Generator 之前CogneeWritermemory_storesession_idmessages、user_idmessages_writtenAgent / Chat Generator 之后安装与环境变量集成以独立包cognee-haystack发布安装命令为pip install cognee-haystackCognee 在写入LLM 抽取知识图谱与查询阶段都需要 LLM 服务其配置LLM 提供方、数据库、向量存储全部通过环境变量读取。最低限度的配置是设置 LLM API Keyexport LLM_API_KEYyour-llm-api-key可选地可以单独指定 Embedding API Key未设置时默认回退到LLM_API_KEYexport EMBEDDING_API_KEYyour-embedding-api-key注意仓库内 CogneeMemoryStore 文档 明确说明 LLM 提供方、数据库与向量存储的完整配置均来自环境变量具体项请参考 Cognee 官方文档完成初始化。CogneeMemoryStore知识图谱记忆的数据层CogneeMemoryStore是记忆后端的核心封装。它把 Cognee V2 记忆 API 做了四对一的薄封装见 API 参考CogneeMemoryStore 方法底层 Cognee API职责add_memoriescognee.remember持久化消息search_memoriescognee.recall检索记忆improvecognee.improve图谱增强 / 会话提升delete_all_memoriescognee.forget删除数据集初始化参数__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各参数语义均来自 API 参考文档search_type可选默认GRAPH_COMPLETIONsearch_memories使用的 Cognee 召回策略。除图补全外常用取值还有CHUNKS原始片段召回与SUMMARIES图节点的摘要召回。top_k可选默认5search_memories的默认最大返回条数可被每次调用覆盖。dataset_name可选默认haystack_memory该 store 对应的 Cognee 数据集名称。session_id可选默认None设置后读写指向会话缓存层级为None时使用永久知识图谱。self_improvement可选默认True透传给cognee.remember与 cognee 默认值一致。为True时在永久层级上每次写入后内联等待improve执行完毕在会话层级上则以 fire-and-forget 后台任务方式调度improve。timeout可选默认300任何单个 cognee 调用的超时上限秒超时抛concurrent.futures.TimeoutError。默认 300 秒可从容覆盖单条消息的 agent 记忆写入对长文档的批量摄取可能需要调大。方法详解add_memories通过cognee.remember持久化消息。add_memories(*, messages: list[ChatMessage], user_id: str | None None, session_id: str | None None) - None永久层级会把所有文本合并为一次调用批量写入会话层级则为每条消息单独写一条与 cognee 官方会话示例保持一致。空消息会被跳过。user_id为 None 时使用 cognee 的默认用户session_id可对 store 级配置做单次覆盖。search_memories通过cognee.recall检索并把每条命中包装为一条system 角色的ChatMessage。search_memories(*, query: str | None None, top_k: int | None None, user_id: str | None None) - list[ChatMessage]query为空或None时直接返回[]。top_k为单次覆盖缺省时用 store 默认值。improve通过cognee.improve把会话缓存内容提升到永久知识图谱。improve(*, session_id: str | None None, user_id: str | None None) - None不带session_id时是一次纯粹的图谱增强enrichment操作session_id缺省时取 store 自身的session_id。delete_all_memories通过cognee.forget(dataset...)删除该数据集。delete_all_memories(*, user_id: str | None None) - None会话缓存不受影响会话并非按数据集作用域隔离如需完整清空须直接调用cognee.forget(everythingTrue)。双记忆层级会话缓存与永久知识图谱这是该集成最核心的设计。通过session_id选择目标层级永久知识图谱session_idNone写入时 Cognee 执行 LLM 抽取生成富含实体关系的图谱节点支持更丰富的图补全GRAPH_COMPLETION查询读取走cognee.recall。会话缓存设置session_id写入速度快、不做 LLM 抽取召回是会话感知的内容后续可通过improve()提升为永久记忆。使用指南 给出了一段同时使用两个层级的示例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)CogneeRetriever把记忆注入 Agent 上下文CogneeRetriever是search_memories之上的薄管道适配器API 参考搜索行为search_type、数据集、会话层级全部配置在 store 上__init__(*, memory_store: CogneeMemoryStore, top_k: int | None None) - None run(query: str, top_k: int | None None, user_id: str | None None) - dict[str, list[ChatMessage]]top_k优先取每次调用传入值其次取 init 时的值最后回退到 store 默认值三级回退链。user_id把检索限定到特定 Cognee 用户None时使用 cognee 默认用户。返回的messages均为 system 角色ChatMessage可直接拼接到 Agent 的消息列表开头充当长期记忆。独立使用示例retriever 文档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])在 Pipeline 中为 Agent 注入记忆retriever 文档 给出了完整的记忆增强管道检索记忆 → 用OutputAdapter把记忆 当前用户消息合并 → 送入 Agent。其中OutputAdapter使用 Jinja 模板{{ memories user_messages }}将两组ChatMessage拼接output_typelist[ChatMessage]且需unsafeTruefrom 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)], }, } )这里的Agent与OpenAIChatGenerator均来自仓库主库见 Agent 组件源码 与 ChatMessage 数据类记忆检索结果以 system 消息形式排在对话开头Agent 的系统提示词明确指示模型将开头的 system 消息视作长期记忆使用。CogneeWriter把对话回合持久化为记忆CogneeWriter把ChatMessage列表写入CogneeMemoryStoreAPI 参考通常放在 Agent 回合之后__init__(*, memory_store: CogneeMemoryStore, session_id: str | None None) - None run(messages: list[ChatMessage], user_id: str | None None) - dict[str, list[ChatMessage]]透传语义写入的消息会原样透传到输出键messages_written因此可以安全地串联在 Agent 或 Generator 之后而不破坏管道数据流。层级覆盖writer 的session_id会在每次调用时覆盖 store 自身的session_id因此一个 store 可以支撑多个 writer 写入不同层级——这是该组件最重要的设计点。不传session_id或为None写入永久图谱LLM 抽取、图补全就绪传入session_id则写入会话缓存快、无抽取之后可用store.improve()提升。独立使用示例writer 文档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])写入会话缓存只需在 init 时传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, )在 Pipeline 中记录 Agent 回合writer 文档 演示了把 Agent 的完整messages输出接到 writer 上使 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])self_improvement 与 improve() 的正确用法这是最容易踩坑的地方API 参考给出了明确的语义默认self_improvementTrue与 cognee 一致。此时每次remember写入后永久层级内联等待improve完成会话层级则以 fire-and-forget 后台任务调度improve。如果业务上把improve()作为唯一的增强触发点必须显式设置self_improvementFalse。否则显式调用improve()会与写入时自动触发的 improve叠加执行两次产生近重复的图谱节点near-duplicate graph nodes。典型推荐组合永久记忆写入 手动批量提升时store 配置self_improvementFalse写完后由你主动调用improve()。删除记忆与清理delete_all_memories()只删除该 store 的dataset_name数据集内部走cognee.forget(dataset...)。由于会话缓存不属于数据集作用域它会在本次删除中存活下来。需要完整清空含会话缓存时直接调用 cognee 原生 API# 只删除当前数据集会话缓存不受影响。 store.delete_all_memories() # 完整清空含会话缓存。 import asyncio import cognee asyncio.run(cognee.forget(everythingTrue))实战要点与边界行为汇总三级 top_k 回退链retriever 单次调用top_k→ retriever inittop_k→ storetop_k默认5。空查询短路search_memories对空/None的query直接返回[]避免无谓的 LLM 调用。空消息跳过add_memories跳过空消息永久层级批量合并为一次remember调用会话层级逐条写入。超时保护timeout默认 300s对每个 cognee 调用生效批量摄取长文档时应调大该值。system 角色输出search_memories把每条命中包装为 system 角色ChatMessage与 Agent 的 system-prompt 语义天然兼容。用户隔离user_id贯穿写入、检索与删除全流程None时落到 cognee 默认用户多用户场景请务必显式传入 Cognee 用户 UUID。相关资源Cognee 集成 API 参考三个组件的完整签名与参数文档CogneeMemoryStore 使用指南数据层参数与双层级示例CogneeRetriever 使用指南检索器独立与 Pipeline 用法CogneeWriter 使用指南写入器独立与 Pipeline 用法Agent 组件源码 与 ChatMessage 数据类记忆注入的目标组件【免费下载链接】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），仅供参考