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.2.0.

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. And on Pipecat ≥1.x the universal aggregators consume TranscriptionFrame while TTS services consume LLMTextFrame, so a SynapRecorder at the end of the pipeline only sees the turns via the shared context (user) and the TTSTextFrame stream (assistant). Without context= it records nothing.

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. On LLMFullResponseEndFrame, calls sdk.conversation.record_message() for both turns. 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 from the TTSTextFrame stream (the words actually spoken, which is also the correct record under interruption).

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) surface SynapIntegrationError via wrap_sdk_errors_async. The processor logs the failure and pushes an ErrorFrame upstream; 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.

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.
• conversation_id: optional explicit conversation id (auto-generated per process lifetime when absent, shared across both processors if you pass the same instance)