synap-pipecat

Synap integration for Pipecat — preload long-term memory into voice pipelines and record every turn back to Synap.

Install

pip install synap-pipecat

Requires pipecat-ai>=0.0.80, maximem-synap>=0.7.1.

Quickstart

from pipecat.pipeline.pipeline import Pipeline
from pipecat.processors.aggregators.llm_context import LLMContext
from pipecat.processors.aggregators.llm_response_universal import (
    LLMContextAggregatorPair,
    LLMUserAggregator,
    LLMAssistantAggregator,
)
from maximem_synap import MaximemSynapSDK
from synap_pipecat import SynapMemoryProcessor, SynapRecorder

sdk = MaximemSynapSDK(api_key="sk-...")
context = LLMContext(messages=[{"role": "system", "content": "You are a helpful assistant."}])

user_agg = LLMUserAggregator(context=context)
asst_agg = LLMAssistantAggregator(context=context)

pipeline = Pipeline([
    transport.input(),
    stt,
    SynapMemoryProcessor(sdk, user_id="alice", customer_id="acme", context=context),
    user_agg,
    llm,
    tts,
    transport.output(),
    asst_agg,
    SynapRecorder(sdk, user_id="alice", customer_id="acme", context=context),
])

Pass context= to both processors. SynapMemoryProcessor without a context is a read-only tap; it fetches nothing into the prompt. On Pipecat ≥1.x the universal aggregators consume TranscriptionFrame, so a SynapRecorder at the end of the pipeline recovers the user turn from the shared context, and it is also the context Synap's compaction is spliced into. Without context= it records only the assistant half.

Scope

• SynapMemoryProcessor — sits BEFORE the user context aggregator. On each TranscriptionFrame, fetches relevant context from Synap via sdk.fetch() and appends it as a system message to the shared LLMContext. The transcription frame itself flows through unchanged so the user aggregator can record it as the user turn.
• SynapRecorder: sits AFTER the assistant context aggregator. Records both turns of each exchange. The user turn comes from the latest TranscriptionFrame to reach the processor, or (on universal-aggregator pipelines where that frame never arrives) from the shared context. The assistant turn comes from streamed LLMTextFrames or the TTSTextFrame stream, or, behind the universal LLMAssistantAggregator, from LLMContextAssistantTurnFrame (see below).
• SynapShortTermProcessor: sits BEFORE the user aggregator, with the conversation_id of the recorder. Keeps one tagged system block at the head of the context, and never repeats a turn the context already holds.

What the recorder sees depends on where it sits

Pipecat 1.x's LLMAssistantAggregator consumes LLMFullResponseStartFrame/EndFrame, every TextFrame (so TTSTextFrame too), LLMThoughtTextFrame and the function-call frames, and pushes none of them on (llm_response_universal.py, LLMAssistantAggregator.process_frame). Behind it, as in the quickstart, the recorder gets one end-of-turn frame, LLMContextAssistantTurnFrame (pipecat-ai 1.7+), and records the turn from that. Tool calls and results are read back out of the shared context after the turn, since no frame carries them; reasoning is not recoverable there.

To get tool calls at the moment they are dispatched, and reasoning, put the recorder BEFORE the assistant aggregator (for example llm, SynapRecorder(...), tts, ..., asst_agg). It then ends the turn on LLMFullResponseEndFrame, and the aggregator's LLMContextAssistantTurnFrame is ignored so a turn is never recorded twice.

The gRPC stream is the default; REST only as a fallback, never both

The first turn, tool call or tool result opens the stream (sdk.instance.listen()) if you did not, and everything after rides it. The server persists streamed turns itself, so sdk.conversation.record_message is called only when the stream cannot carry the turn. Set SYNAP_AUTO_LISTEN=false (or set_auto_listen(sdk, False) from synap_integrations_common) to stay on REST.

Synap's compaction replaces the earlier turns

With context= given, after each turn it records the recorder checks whether Synap has compacted the conversation. If so, the shared LLMContext is rewritten in place (set_messages, the same call pipecat's own LLMContextSummarizer ends in): the earlier turns become one <synap_conversation_history> system message plus the turns after the compaction, and the current exchange is untouched. Before the first compaction, or if Synap does not answer within 1.5s, the context is left alone and enable_auto_context_summarization (if you turned it on) stays the fallback. SynapShortTermProcessor does the same before each user turn.

Error policy

• Reads (SynapMemoryProcessor) degrade gracefully — SDK failures log at ERROR and the frame continues downstream with no Synap context injected. A Synap blip must never break a live voice call.
• Writes (SynapRecorder) that fail on both the stream and REST are logged and pushed upstream as an ErrorFrame carrying a SynapIntegrationError; the pipeline itself stays alive (errors propagate as frames, not exceptions).
• Non-matching frames pass through untouched in both processors — standard Pipecat push_frame contract.

Deeper scope ladders

If your account's scope ladder has levels beyond customer and user, pass the level to read and write at as scope_path (key order does not matter; the server orders by ladder position). It is forwarded to sdk.fetch and to the user and assistant turns, only when set; nothing changes for callers who do not pass it, and no extra network call is made per turn.

path = {"customer": "acme", "team": "payments", "user": "dana"}  # e.g. from ScopeAddress.scope_path
memory = SynapMemoryProcessor(sdk, user_id="dana", customer_id="acme", context=ctx, scope_path=path)
recorder = SynapRecorder(sdk, user_id="dana", customer_id="acme", context=ctx, scope_path=path)

The server lets a path's user and customer levels win over flat ids, so user_id / customer_id that disagree with the path raise SynapIntegrationError at construction, before anything is sent. If you reassign processor.user_id or .customer_id later, the processor skips that read or turn and logs an ERROR (it never raises inside the pipeline).

SynapShortTermProcessor reads through the conversation, which remembers its own level, so it takes no scope_path. Needs a maximem-synap release whose sdk.fetch accepts scope_path.

Constructor args (both processors)

• sdk: a configured MaximemSynapSDK
• user_id: required — Synap memory is user-scoped
• customer_id: B2B instances only, where it is required. Not accepted on a B2C instance, which rejects it with HTTP 400.
• scope_path (memory processor, recorder): optional {level_key: value} for a deeper scope ladder; see above
• conversation_id (recorder, short-term processor): optional explicit id for the recorder (auto-generated when absent), required for the short-term processor: pass the same value to both