企业官网建设流程全解析

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

Cognee 是面向 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 之前。

仓库内三份组件文档的关键属性对比如下：

安装与环境变量

集成以独立包cognee-haystack发布，安装命令为：

pip install cognee-haystack

Cognee 在写入（LLM 抽取知识图谱）与查询阶段都需要 LLM 服务，其配置（LLM 提供方、数据库、向量存储）全部通过环境变量读取。最低限度的配置是设置 LLM API Key：

export LLM_API_KEY="your-llm-api-key"

可选地，可以单独指定 Embedding API Key；未设置时默认回退到LLM_API_KEY：

export EMBEDDING_API_KEY="your-embedding-api-key"

注意：仓库内 CogneeMemoryStore 文档 明确说明 LLM 提供方、数据库与向量存储的完整配置均来自环境变量，具体项请参考 Cognee 官方文档完成初始化。

CogneeMemoryStore：知识图谱记忆的数据层

CogneeMemoryStore是记忆后端的核心封装。它把 Cognee V2 记忆 API 做了四对一的薄封装（见 API 参考）：

初始化参数

__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_COMPLETION"）：search_memories使用的 Cognee 召回策略。除图补全外，常用取值还有"CHUNKS"（原始片段召回）与"SUMMARIES"（图节点的摘要召回）。
• top_k（可选，默认5）：search_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(everything=True)。

双记忆层级：会话缓存与永久知识图谱

这是该集成最核心的设计。通过session_id选择目标层级：

• 永久知识图谱（session_id=None）：写入时 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_name="my_agent_memory", self_improvement=False) # 写入长期事实到永久图谱（不传 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_id="alice_session_1", ) # 将会话缓存提升到永久图谱。 store.improve(session_id="alice_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_type="GRAPH_COMPLETION", top_k=5) # 先写入一些记忆 store.add_memories( messages=[ChatMessage.from_user("Alice prefers concise Python examples.")], user_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890", ) retriever = CogneeRetriever(memory_store=store, top_k=3) result = retriever.run( query="What does Alice prefer?", user_id="a1b2c3d4-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_type=list[ChatMessage]且需unsafe=True：

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.dataclasses import ChatMessage from haystack_integrations.components.retrievers.cognee import CogneeRetriever from haystack_integrations.memory_stores.cognee import CogneeMemoryStore store = CogneeMemoryStore(dataset_name="my_agent_memory", session_id="alice_session_1") pipeline = Pipeline() pipeline.add_component("retriever", CogneeRetriever(memory_store=store, top_k=5)) pipeline.add_component( "memory_context", OutputAdapter( template="{{ memories + user_messages }}", output_type=list[ChatMessage], unsafe=True, ), ) pipeline.add_component( "agent", Agent( chat_generator=OpenAIChatGenerator(model="gpt-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列表写入CogneeMemoryStore（API 参考），通常放在 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_store=store) result = writer.run( messages=[ChatMessage.from_user("Alice prefers concise Python examples.")], user_id="a1b2c3d4-e5f6-7890-abcd-ef1234567890", ) print(result["messages_written"])

写入会话缓存只需在 init 时传session_id：

session_writer = CogneeWriter(memory_store=store, session_id="alice_session_1") session_writer.run( messages=[ ChatMessage.from_user("Alice is currently debugging a vector store issue.") ], user_id="a1b2c3d4-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_name="my_agent_memory") pipeline = Pipeline() pipeline.add_component( "agent", Agent( chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"), system_prompt=( "Answer the user and preserve durable user facts or preferences for future conversations." ), ), ) pipeline.add_component("writer", CogneeWriter(memory_store=store)) 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_improvement=True（与 cognee 一致）。此时每次remember写入后：永久层级内联等待improve完成；会话层级则以 fire-and-forget 后台任务调度improve。
• 如果业务上把improve()作为唯一的增强触发点，必须显式设置self_improvement=False。否则显式调用improve()会与写入时自动触发的 improve叠加执行两次，产生近重复的图谱节点（near-duplicate graph nodes）。

典型推荐组合：永久记忆写入 + 手动批量提升时，store 配置self_improvement=False，写完后由你主动调用improve()。

删除记忆与清理

delete_all_memories()只删除该 store 的dataset_name数据集（内部走cognee.forget(dataset=...)）。由于会话缓存不属于数据集作用域，它会在本次删除中存活下来。需要完整清空（含会话缓存）时，直接调用 cognee 原生 API：

# 只删除当前数据集（会话缓存不受影响）。 store.delete_all_memories() # 完整清空（含会话缓存）。 import asyncio import cognee asyncio.run(cognee.forget(everything=True))

实战要点与边界行为汇总

• 三级 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