Haystack

Retriever and memory-writer pipeline components for Haystack RAG pipelines.

Add persistent, per-user memory to a Haystack pipeline as two regular components: a retriever for the read side and a writer for the write side. Both drop into existing pipelines without restructuring.

Overview

This guide shows how to add Synap to a Haystack application to build pipelines that:

• Retrieve user-scoped memories as standard Document objects in a RAG flow
• Persist each conversation turn back to Synap so future runs benefit from it
• Compose freely with any other Haystack component (rerankers, prompt builders, generators)

The Synap Haystack integration ships two drop-in pipeline components — both follow Haystack’s component contract so you can wire them into pipelines exactly like any built-in component.

Component	Role	Purpose
SynapRetriever	Read	Fetches Synap memories as Document objects
SynapMemoryWriter	Write	Records conversation turns back to Synap

Setup

Install the package alongside Haystack:

pip install maximem-synap-haystack haystack-ai

Configure your API key. Generate one from the Synap Dashboard.

.env

SYNAP_API_KEY=synap_your_key_here
OPENAI_API_KEY=your-openai-api-key

Initialize the SDK once at application startup:

from maximem_synap import MaximemSynapSDK

sdk = MaximemSynapSDK()
await sdk.initialize()

See SDK Initialization for the full lifecycle and configuration options.

Basic integration

The smallest useful integration plugs SynapRetriever into a pipeline and routes its documents output to a prompt builder:

from haystack import Pipeline
from haystack.components.builders import PromptBuilder
from synap_haystack import SynapRetriever

retriever = SynapRetriever(
    sdk=sdk,
    user_id="alice",
    customer_id="acme",   # optional — required for B2B instances
    max_results=6,
    mode="fast",          # "fast" or "accurate"
)

pipeline = Pipeline()
pipeline.add_component("retriever", retriever)
pipeline.add_component("prompt_builder", PromptBuilder(template=your_template))
pipeline.connect("retriever.documents", "prompt_builder.documents")

result = pipeline.run({"retriever": {"query": "project deadlines"}})

Retrieval failures degrade gracefully — SynapRetriever emits an empty documents list and logs an error, so the rest of the pipeline keeps running.

To close the loop and persist new turns for future retrievals, add SynapMemoryWriter after the generator.

---

Core concepts

Retriever

SynapRetriever is a Haystack component that takes a query input and emits a documents output. Each returned Document has:

• content — the memory text
• meta["type"] — memory type (e.g. "fact", "preference")
• meta["confidence"] — relevance score

from synap_haystack import SynapRetriever

retriever = SynapRetriever(
    sdk=sdk,
    user_id="alice",
    customer_id="acme",
    max_results=6,
    mode="fast",
)

The two retrieval modes trade latency against comprehensiveness:

	fast	accurate
Latency	50-100ms	200-500ms
Search	Vector similarity	Vector + graph + re-ranking
Best for	Real-time chat	Multi-entity queries

Because the output shape matches Haystack’s standard Document, you can route it through any reranker, prompt builder, or filter that accepts documents.

Memory writer

SynapMemoryWriter is the write side. Place it at the end of a pipeline so each LLM reply is captured as a memory for future retrievals:

from synap_haystack import SynapMemoryWriter

writer = SynapMemoryWriter(
    sdk=sdk,
    conversation_id="conv-001",
    user_id="alice",
    customer_id="acme",
)

It accepts a replies input (matching the output of standard generators like OpenAIGenerator) and emits a result summary. Write failures surface explicitly — SynapMemoryWriter raises SynapIntegrationError so the pipeline knows if persistence failed.

---

Complete example: full RAG pipeline with memory loop

The following pipeline retrieves user-scoped memories, builds a prompt, generates a response, and writes the response back to Synap — in one pass:

from haystack import Pipeline
from haystack.components.builders import PromptBuilder
from haystack.components.generators import OpenAIGenerator
from synap_haystack import SynapRetriever, SynapMemoryWriter


def build_pipeline(sdk, user_id: str, conversation_id: str, customer_id: str | None = None) -> Pipeline:
    retriever = SynapRetriever(
        sdk=sdk,
        user_id=user_id,
        customer_id=customer_id,
        max_results=6,
        mode="fast",
    )
    writer = SynapMemoryWriter(
        sdk=sdk,
        conversation_id=conversation_id,
        user_id=user_id,
        customer_id=customer_id,
    )

    template = """
Given this context about the user:
{% for doc in documents %}
- {{ doc.content }}
{% endfor %}

Answer the question: {{ query }}
"""

    pipeline = Pipeline()
    pipeline.add_component("retriever", retriever)
    pipeline.add_component("prompt", PromptBuilder(template=template))
    pipeline.add_component("llm", OpenAIGenerator(model="gpt-4o"))
    pipeline.add_component("writer", writer)

    pipeline.connect("retriever.documents", "prompt.documents")
    pipeline.connect("prompt.prompt", "llm.prompt")
    pipeline.connect("llm.replies", "writer.replies")

    return pipeline


# Usage
pipeline = build_pipeline(sdk, user_id="alice", conversation_id="conv-001", customer_id="acme")

result = pipeline.run({
    "retriever": {"query": "What are my priorities?"},
    "prompt": {"query": "What are my priorities?"},
})

print(result["llm"]["replies"][0])

Three things to notice in this pattern:

1. Memory is just another retriever. SynapRetriever emits standard Document objects, so you can mix it with any document store retriever via a DocumentJoiner if you want corpus context too.
2. The write loop closes itself. Each pipeline run ends by persisting the reply — every subsequent run benefits from accumulating context.
3. Scope is bound at construction. The retriever and writer carry the user/customer scope; the pipeline graph never needs to know about user identity.

---

Advanced patterns

Multi-tenant scoping

Both components accept the standard scoping triple — user_id (required), optional customer_id, optional conversation_id. customer_id is required on B2B Synap instances and ignored on single-tenant ones. See Memory Scopes.

# User-scoped only
retriever = SynapRetriever(sdk=sdk, user_id="alice")

# Organization-scoped
retriever = SynapRetriever(sdk=sdk, user_id="alice", customer_id="acme")

For multi-tenant services, build the pipeline (or at least the retriever/writer components) per request so each invocation has the correct scope baked in.

Combining with document retrieval

SynapRetriever’s output is a standard Document list, so it slots into a DocumentJoiner next to your existing document store retriever:

pipeline.add_component("doc_retriever", your_existing_retriever)
pipeline.add_component("synap", SynapRetriever(sdk=sdk, user_id="alice"))
pipeline.add_component("joiner", DocumentJoiner())
pipeline.connect("doc_retriever.documents", "joiner.documents")
pipeline.connect("synap.documents", "joiner.documents")

User-specific facts and corpus chunks come back as a single ranked list to the prompt builder.

Failure semantics

The integration follows the Synap-wide contract:

• SynapRetriever degrades gracefully — emits an empty documents list and logs an error if Synap is unreachable.
• SynapMemoryWriter surfaces failures — raises SynapIntegrationError so the pipeline (and caller) know persistence failed.

This is by design: read failures shouldn’t break a user-facing turn, but silent write failures would let the memory drift away from reality.

---

Next steps

LangChain

Memory and retriever for LangChain.

LlamaIndex

BaseMemory and retriever for LlamaIndex.

Context Fetch

The retrieval API that powers SynapRetriever — modes, scopes, and response shapes.

Memory Scopes

How user_id, customer_id, and conversation_id interact across reads.