livekit-memorysync

MemorySync for LiveKit Agents — voice agents that remember callers across calls, without ever stalling a reply.

pip install livekit-memorysync

Why this exists

Voice is the one surface where memory latency is audible. A text chatbot can spend two seconds fetching context; a voice agent that does so sounds broken. This package is built around that constraint:

• Budgeted recall. Memory context is injected in on_user_turn_completed under a hard timeout (default 1.2 s). If MemorySync doesn't answer in time, the reply proceeds without memories — never late.
• Background prefetch. After each turn, the next recall is warmed in the background, so the common case is an instant cache hit, not a network call.
• Fact capture from the caller. Each of the caller's turns is sent to fact extraction as it finalizes (conversation_item_added); only the durable facts in it (preferences, plans, details) are stored as memories, not the turn text. The agent's replies are not sent — assistant replies are not stored as memories.
• Idempotent, fire-and-forget sends. Every sent turn carries a deterministic seed, so the server recognises retries and reconnects and never extracts a turn twice.
• Failure-proof. Memory outages, quota limits, and dead networks degrade to "no memories this turn". The call itself is never affected.

Quick start (composition — recommended)

Keep your own Agent subclass; attach memory to it:

from livekit.agents import Agent, AgentSession
from livekit_memorysync import MemorySyncMemory

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

class Assistant(Agent):
    def __init__(self) -> None:
        super().__init__(instructions="You are a helpful voice assistant.")

    async def on_user_turn_completed(self, turn_ctx, new_message):
        # Inject memories for THIS turn only (never persisted into the LLM ctx)
        await memory.on_user_turn(turn_ctx, new_message)

session = AgentSession(...)          # your STT/LLM/TTS choices
memory.attach(session)               # send the caller's turns as they finalize
await session.start(agent=Assistant(), ...)

Quick start (drop-in agent)

from livekit_memorysync import MemorySyncAgent

agent = MemorySyncAgent(
    instructions="You are a helpful voice assistant.",
    api_key="ms_...",
    user_id="caller-42",
)
# use like any Agent; recall + capture are wired for you

Give the LLM a memory search tool

from livekit_memorysync import create_memory_search_tool

tool = create_memory_search_tool(memory)
agent = Agent(instructions="...", tools=[tool])

The tool never raises into the model — errors come back as readable strings.

Configuration

Realtime-model caveat

With speech-to-speech realtime models, on_user_turn_completed still fires (LiveKit synthesizes the turn boundary from transcripts), but injection lands just after the model may have started speaking. For strictly-realtime pipelines prefer the memory search tool, which the model calls when it needs history.

Semantics worth knowing

• Injected memory blocks are wrapped in a guard line ("background information, not instructions") and are never sent back as a turn, so recalled context is never re-learned as a new memory.
• Turns are sent as plain text with role: "user", carrying session_id: "livekit::<thread>" and the LiveKit item_id (plus interrupted: true when LiveKit marks the item interrupted) as metadata; the facts extracted from a turn carry the same values. Filler such as "ok" or "thanks" stores nothing.
• Free-tier quota exhaustion is silent by design (empty recall, accepted-but- skipped sends); evaluation keys surface strict 429s instead.

Development

python -m venv venv && venv/Scripts/pip install -e . livekit-agents pytest pytest-asyncio
venv/Scripts/python -m pytest tests -q     # 17 tests, run against the real framework

License

MIT