> ## Documentation Index
> Fetch the complete documentation index at: https://docs.maximem.ai/llms.txt
> Use this file to discover all available pages before exploring further.
# Pipecat
> Frame processors that add memory injection and turn recording to Pipecat voice pipelines.
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:
```bash pip theme={null}
pip install maximem-synap-pipecat pipecat-ai
```
```bash uv theme={null}
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](https://synap.maximem.ai).
```bash .env theme={null}
SYNAP_API_KEY=synap_your_key_here
OPENAI_API_KEY=your-openai-api-key
```
Initialize the SDK once at the worker's startup:
```python theme={null}
from maximem_synap import MaximemSynapSDK
sdk = MaximemSynapSDK()
await sdk.initialize()
```
See [SDK Initialization](/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:
```python theme={null}
# 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:
```python theme={null}
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:
```python theme={null}
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:
```text theme={null}
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:
```python theme={null}
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:
1. **Memory injection happens once per LLM turn**, not once per audio frame; the processor only acts on `LLMMessagesFrame` events.
2. **Recording is async and non-blocking.** The audio path never waits on a Synap write.
3. **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](/concepts/memory-scopes).
```python theme={null}
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](/patterns/voice-agent-livekit): the reference architecture for memory-backed voice agents.
* [Voice concierge cookbook](/cookbook/voice-concierge): an end-to-end worked voice example.
***
## Next steps
Context preloading and recording for LiveKit voice agents.
Hooks and MCP server for the Claude Agent SDK.
The retrieval API behind `SynapMemoryProcessor`: modes, scopes, and response shapes.
How `user_id`, `customer_id`, and `conversation_id` interact across reads.