pipecat-memorysync

MemorySync for Pipecat — long-term memory for voice pipelines that never stalls a reply and never re-stores what it already knows.

pip install pipecat-memorysync

Where it sits

MemorySyncMemoryService is a FrameProcessor. Place it between your context aggregator and your LLM service:

transport.input() → stt → context_aggregator.user()
    → MemorySyncMemoryService        ← enriches + captures here
    → llm → tts → transport.output() → context_aggregator.assistant()

from pipecat_memorysync import MemorySyncMemoryService

memory = MemorySyncMemoryService(
    api_key="ms_...",                 # or MEMORYSYNC_API_KEY env var
    user_id="caller-42",              # stable end-user id
    session_id="call-123",            # optional: scope to this call
)

pipeline = Pipeline([
    transport.input(),
    stt,
    context_aggregator.user(),
    memory,
    llm,
    tts,
    transport.output(),
    context_aggregator.assistant(),
])

Every LLMContextFrame that flows through is enriched with relevant memories (as a system message) and mined for new turns to persist — then pushed on, enriched or not, on time.

Design guarantees

• Budgeted recall. Enrichment runs under a hard timeout (default 1.2 s). A slow or dead memory backend means an unenriched frame, never a stalled voice reply.
• Delta-only capture. Only messages not seen before are stored, tracked by deterministic idempotency seeds. Growing a 50-message context does not re-store 50 messages per turn (a real flaw in some in-tree memory services, which re-send the entire context every frame — O(n²) writes per call).
• Injection exclusion. The memory block this service adds is never captured back as a new memory.
• Graceful end, salvaged abort. On EndFrame, queued writes get a bounded window (3 s) to land before the pipeline stops — the call's final exchange is not lost. On CancelFrame, the frame is pushed first and writes get a brief salvage window.
• Failure-proof. HTTP errors, quota limits, and timeouts all degrade to "no memories this turn". Nothing propagates into the pipeline.

Configuration (InputParams)

from pipecat_memorysync import MemorySyncMemoryService

memory = MemorySyncMemoryService(
    api_key="ms_...",
    user_id="caller-42",
    params=MemorySyncMemoryService.InputParams(
        top_k=5,                  # memories injected per turn
        recall_timeout=1.2,       # hard budget, seconds
        add_as_system_message=True,
        position="end",           # where the memory block lands in the context
        min_prompt_chars=8,       # skip enrichment for shorter user prompts
    ),
)

Semantics worth knowing

• Both user and assistant turns are persisted, with role fidelity.
• Idempotency seeds make retries/reconnects duplicate-free server-side.
• Free-tier quota exhaustion is silent by design (empty recall, accepted-but- dropped writes); evaluation keys surface strict 429s instead.
• The service is reusable across pipeline runs; call await memory.aclose() from application shutdown if you want an explicit flush + client close.

Development

python -m venv venv && venv/Scripts/pip install -e . pipecat-ai pytest pytest-asyncio
venv/Scripts/python -m pytest tests -q    # 14 tests, run via pipecat's official test harness

License

MIT