Pipecat - Maximem Synap

Add persistent memory to a Pipecat voice pipeline as two frame processors: one that injects the user’s relevant memories before the LLM sees a frame, and one that records each completed turn after the response.

Requires Python 3.11+.

Overview

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

Inject the most relevant memories as a system message before the LLM responds
Record each completed user + assistant turn back to Synap
Compose with any other Pipecat frame processor without changing the pipeline shape

The Synap Pipecat integration ships two frame processors. Both follow Pipecat’s processor contract, so they slot into any pipeline alongside STT, LLM, and TTS.

Class	Pipeline position	Purpose
`SynapMemoryProcessor`	Before LLM	Prepends relevant memories to the system message
`SynapRecorder`	After response	Records the completed turn back to Synap

Setup

Install the package alongside Pipecat:

pip install maximem-synap-pipecat pipecat-ai

uv add maximem-synap-pipecat pipecat-ai
# pip-compatible (existing venv): uv pip install maximem-synap-pipecat pipecat-ai

The pip package is maximem-synap-pipecat, but the import drops the maximem- prefix and uses underscores: from synap_pipecat import ....

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 the worker’s 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 adds both processors to a standard voice pipeline: memory injection before the LLM, recording after the response. No other pipeline changes are needed:

# pip install maximem-synap-pipecat pipecat-ai
import uuid
from maximem_synap import MaximemSynapSDK
from pipecat.pipeline.pipeline import Pipeline
from synap_pipecat import SynapMemoryProcessor, SynapRecorder

sdk = MaximemSynapSDK()
await sdk.initialize()

memory = SynapMemoryProcessor(
    sdk=sdk,
    user_id="alice",
    customer_id="acme",   # optional, required for B2B instances
    max_results=6,
)

# If you pass conversation_id explicitly it must be a valid UUID;
# omit it to let Synap auto-generate one.
recorder = SynapRecorder(
    sdk=sdk,
    user_id="alice",
    customer_id="acme",
    conversation_id=str(uuid.uuid4()),   # optional; auto-generated if omitted
)

pipeline = Pipeline([
    transport.input(),
    stt,
    memory,           # inject memory before LLM
    user_aggregator,
    llm,
    tts,
    transport.output(),
    assistant_aggregator,
    recorder,         # record turn after response
])

Memory injection failures degrade gracefully: the frame passes through unmodified if context retrieval fails. Recording failures surface explicitly as SynapIntegrationError, which Pipecat’s frame-error handling catches and logs.

Core concepts

Memory processor

SynapMemoryProcessor intercepts LLMMessagesFrame events and prepends a system message containing the user’s relevant memories before the frame reaches the LLM service:

from synap_pipecat import SynapMemoryProcessor

memory = SynapMemoryProcessor(
    sdk=sdk,
    user_id="alice",
    customer_id="acme",
    max_results=6,
    mode="fast",          # "fast" or "accurate"
)

Voice latency is tight, so mode="fast" is the default. The two retrieval modes trade latency against comprehensiveness:

	`fast`	`accurate`
Search	Vector + graph (no LLM subquery decomposition)	Vector + graph + LLM subquery decomposition + reranking
Best for	Real-time chat	Multi-entity queries

fast is lower-latency and suited to the hot path; accurate adds LLM-driven query decomposition and reranking for relationship-aware queries at a higher latency cost. Failures degrade gracefully: if context retrieval fails, the LLMMessagesFrame passes through unmodified rather than blocking the call.

Recorder

SynapRecorder intercepts TranscriptionFrame (user side) and LLMFullResponseEndFrame (assistant side) and ingests the completed turn into Synap asynchronously:

from synap_pipecat import SynapRecorder

recorder = SynapRecorder(
    sdk=sdk,
    user_id="alice",
    customer_id="acme",
    conversation_id=str(uuid.uuid4()),
)

Recording happens out-of-band: it never blocks the audio path. Write failures surface as SynapIntegrationError, which propagates through Pipecat’s frame-error handling so the failure is visible rather than silent.

Positioning in the pipeline

The two processors expect specific positions:

transport.input()
    │
    ▼
   STT
    │
    ▼
SynapMemoryProcessor  ← fetches context, prepends to system prompt
    │
    ▼
 UserAggregator
    │
    ▼
   LLM
    │
    ▼
   TTS
    │
    ▼
transport.output()
    │
    ▼
AssistantAggregator
    │
    ▼
 SynapRecorder        ← records completed user + assistant turn

SynapMemoryProcessor must be between STT and the user aggregator; SynapRecorder after the assistant aggregator. Any other placement will not see the right frame types.

Complete example: full voice pipeline with memory

The pattern below sets up an end-to-end voice pipeline with Synap-backed memory. Scope is pulled per-call from the transport’s connection metadata, so the same worker can serve multiple users:

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineTask
from pipecat.services.openai import OpenAILLMService
from pipecat.transports.network.daily import DailyTransport
from synap_pipecat import SynapMemoryProcessor, SynapRecorder


async def build_pipeline(sdk, transport, user_id: str, customer_id: str | None = None) -> Pipeline:
    stt = ...   # your STT service
    llm = OpenAILLMService(model="gpt-4o")
    tts = ...   # your TTS service
    user_aggregator = ...
    assistant_aggregator = ...

    memory = SynapMemoryProcessor(
        sdk=sdk,
        user_id=user_id,
        customer_id=customer_id,
        max_results=6,
        mode="fast",
    )

    recorder = SynapRecorder(
        sdk=sdk,
        user_id=user_id,
        customer_id=customer_id,
    )

    return Pipeline([
        transport.input(),
        stt,
        memory,
        user_aggregator,
        llm,
        tts,
        transport.output(),
        assistant_aggregator,
        recorder,
    ])


# Usage: invoked per call
async def run_call(sdk, room_url: str, user_id: str, customer_id: str | None = None):
    transport = DailyTransport(room_url, ...)
    pipeline = await build_pipeline(sdk, transport, user_id=user_id, customer_id=customer_id)
    task = PipelineTask(pipeline)
    await task.run()

Three things to notice in this pattern:

Memory injection happens once per LLM turn, not once per audio frame; the processor only acts on LLMMessagesFrame events.
Recording is async and non-blocking. The audio path never waits on a Synap write.
Scope is per-call. Each run_call invocation gets its own processor instances with the right user_id / customer_id.

Advanced patterns

Multi-tenant scoping

Both processors 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.

memory = SynapMemoryProcessor(sdk=sdk, user_id="alice", customer_id="acme")

For multi-tenant deployments, build processors per call rather than caching them globally; each call should have its scope baked in.

Tuning retrieval mode

mode="fast" is the default and the right choice for most voice flows. Switch to "accurate" only for use cases where missing relevant memory is worse than the additional pre-LLM latency that LLM-driven query decomposition and reranking add.

Failure semantics

The integration follows the Synap-wide contract, adapted for voice latency:

SynapMemoryProcessor degrades gracefully: frame passes through unmodified if context retrieval fails.
SynapRecorder surfaces failures: raises SynapIntegrationError which Pipecat’s frame-error path catches and logs.

This is by design: a voice call should never break because of a transient memory glitch, but write failures must be visible to monitoring.

Going further

Voice agent pattern: the reference architecture for memory-backed voice agents.
Voice concierge cookbook: an end-to-end worked voice example.

Next steps

LiveKit Agents

Context preloading and recording for LiveKit voice agents.

Claude Agent SDK

Hooks and MCP server for the Claude Agent SDK.

Context Fetch

The retrieval API behind SynapMemoryProcessor: modes, scopes, and response shapes.

Memory Scopes

How user_id, customer_id, and conversation_id interact across reads.

​Overview

​Setup

​Basic integration

​Core concepts

​Memory processor

​Recorder

​Positioning in the pipeline

​Complete example: full voice pipeline with memory

​Advanced patterns

​Multi-tenant scoping

​Tuning retrieval mode

​Failure semantics

​Going further

​Next steps

LiveKit Agents

Claude Agent SDK

Context Fetch

Memory Scopes

Overview

Setup

Basic integration

Core concepts

Memory processor

Recorder

Positioning in the pipeline

Complete example: full voice pipeline with memory

Advanced patterns

Multi-tenant scoping

Tuning retrieval mode

Failure semantics

Going further

Next steps