> ## Documentation Index
> Fetch the complete documentation index at: https://docs.cognee.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Haystack
> Add persistent memory to Haystack agents and pipelines with Cognee.
Give your [Haystack](https://haystack.deepset.ai/) agents and pipelines persistent memory powered by cognee. The `cognee-haystack` package ships a memory store plus a writer and a retriever component, so you can store `ChatMessage`s in cognee's knowledge graph and feed relevant memories back into an `Agent` on every turn.
This integration is maintained by deepset in [haystack-core-integrations](https://github.com/deepset-ai/haystack-core-integrations/tree/main/integrations/cognee) and listed on the [Haystack integrations page](https://haystack.deepset.ai/integrations/cognee).
## Why Use This Integration
* **Native Haystack components**: `CogneeWriter` and `CogneeRetriever` plug into any `Pipeline` and serialize with it
* **Two memory tiers**: Write straight to the permanent knowledge graph, or to a cheap per-session cache that you promote later
* **Natural language recall**: Retrieve memories with any cognee search type (default `GRAPH_COMPLETION`) as system `ChatMessage`s
* **Per-user scoping**: Pass a cognee user UUID as `user_id` to scope writes and searches
## Installation
```bash theme={null}
pip install cognee-haystack
```
Requires Python 3.10+. The package depends on `haystack-ai>=2.24.0` and `cognee>=1.0.9`.
## Quick Start
Set your keys (cognee extracts knowledge with an LLM; Haystack's `OpenAIChatGenerator` reads `OPENAI_API_KEY`):
```bash theme={null}
export LLM_API_KEY="your-openai-api-key-here" # used by cognee
export OPENAI_API_KEY="your-openai-api-key-here" # used by Haystack
```
Store some facts, then build a pipeline that injects retrieved memories ahead of the user's message:
```python theme={null}
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.components.writers.cognee import CogneeWriter
from haystack_integrations.memory_stores.cognee import CogneeMemoryStore
store = CogneeMemoryStore(dataset_name="agent_memory")
# Store long-lived facts in the permanent graph
CogneeWriter(memory_store=store).run(messages=[
ChatMessage.from_user("My name is Alice. I'm a data scientist at Acme Corp."),
ChatMessage.from_user("I'm building a documentation search system with Haystack and Cognee."),
])
# Retriever -> injector -> agent
pipeline = Pipeline()
pipeline.add_component("retriever", CogneeRetriever(memory_store=store))
pipeline.add_component(
"injector",
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="System messages at the start of the conversation contain relevant memories.",
),
)
pipeline.connect("retriever.messages", "injector.memories")
pipeline.connect("injector.output", "agent.messages")
question = "What project am I working on?"
result = pipeline.run({
"retriever": {"query": question},
"injector": {"user_messages": [ChatMessage.from_user(question)]},
})
print(result["agent"]["last_message"].text)
```
The components are synchronous. When called from inside a running event loop, cognee calls run on a shared background loop, so they also work in async applications and notebooks.
## Components
| Component | Import | Description |
| ------------------- | ---------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------ |
| `CogneeMemoryStore` | `haystack_integrations.memory_stores.cognee` | Memory backend wrapping `cognee.remember`, `cognee.recall`, `cognee.improve`, and `cognee.forget` |
| `CogneeWriter` | `haystack_integrations.components.writers.cognee` | Stores `messages` in the store and passes them through as `messages_written`; optional `session_id` overrides the store's tier |
| `CogneeRetriever` | `haystack_integrations.components.retrievers.cognee` | Searches the store for `query` and outputs matching memories as system `ChatMessage`s under `messages`; optional `top_k` |
Both components accept an optional `user_id` (a cognee user UUID) at run time; omit it to use cognee's default user.
### Memory Store Options
`CogneeMemoryStore(*, search_type="GRAPH_COMPLETION", top_k=5, dataset_name="haystack_memory", session_id=None, self_improvement=True, timeout=300)`
| Parameter | Description |
| ------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------- |
| `search_type` | cognee search type used for retrieval, such as `GRAPH_COMPLETION`, `CHUNKS`, `RAG_COMPLETION`, or `SUMMARIES` |
| `top_k` | Default maximum number of results per search |
| `dataset_name` | cognee dataset backing the store |
| `session_id` | When set, writes go to that session's cache and searches combine that session's cache with the permanent graph; `None` uses the permanent graph |
| `self_improvement` | Forwarded to `cognee.remember`. Set to `False` if you call `improve()` yourself, otherwise improve runs twice and can create near-duplicate nodes |
| `timeout` | Per-call timeout in seconds, applied when the components run inside an async event loop; raise it for bulk ingestion of long documents |
The store also exposes `improve(session_id=None, user_id=None)` and `delete_all_memories(user_id=None)`.
## Session Memory
By default, writes go straight to the **permanent knowledge graph**. Give a writer a `session_id` to write to that session's lightweight cache instead, then promote the session into the graph with `improve()`:
```python theme={null}
SESSION_ID = "alice_chat_42"
# One store, two writers: session_id on the writer picks the tier
seed_store = CogneeMemoryStore(dataset_name="agent_memory", self_improvement=False)
CogneeWriter(memory_store=seed_store).run(messages=[...]) # permanent graph
CogneeWriter(memory_store=seed_store, session_id=SESSION_ID).run(messages=[...]) # session cache
# A session-scoped store makes the retriever's searches session-aware
chat_store = CogneeMemoryStore(dataset_name="agent_memory", session_id=SESSION_ID)
retriever = CogneeRetriever(memory_store=chat_store)
# ... run your agent pipeline ...
# Promote the session cache into the permanent graph
chat_store.improve()
```
With a session-scoped store and a completion search type (such as the default `GRAPH_COMPLETION`), cognee records each question and answer in the session cache as you search, so you don't need a writer in the chat loop. `delete_all_memories()` forgets only the store's dataset; the session cache survives. Use `cognee.forget(everything=True)` for a full wipe.
## How It Works
1. **Write**: `CogneeWriter` stores message text with `cognee.remember`, batched into one call on the permanent tier, or one entry per message on the session tier. Empty messages are skipped.
2. **Retrieve**: `CogneeRetriever` runs `cognee.recall` against the store's dataset and wraps each result in a system `ChatMessage`
3. **Inject**: An `OutputAdapter` (or any component you like) prepends the memories to the user's messages before the `Agent` runs
4. **Improve**: `improve()` runs `cognee.improve` to enrich the graph and promote session content into it
***
View source code and examples
Runnable memory agent demo