recall-livekit

Long-term memory for LiveKit Agents voice agents, backed by Recall by Polign.

The agent loads everything it knows about the caller before the first reply, and the model saves new facts through a remember tool. Facts are typed: each one is a predicate from a closed registry with a value, so "call me Sam" replaces the old name instead of piling up a second one, and the history of the change is kept. No second model runs on the read path, and no embedding service is needed.

pip install recall-livekit "livekit-agents[openai,deepgram,cartesia,silero]"

That is the whole install. pip also brings the polign_db package with the polign CLI and polign-server binaries for Linux, macOS and Windows, so there is no separate database to download, and a worker image needs nothing beyond pip install.

Usage

from livekit.agents import AgentServer, AgentSession, JobContext, JobProcess, cli
from livekit.plugins import cartesia, deepgram, openai, silero
from recall_livekit import VOICE_REGISTRY, RecallAgent, RecallMemory

server = AgentServer()


def setup(proc: JobProcess) -> None:
    # one Recall subprocess per worker process
    proc.userdata["recall"] = RecallMemory.open(
        local_dir="./recall-data",            # the database lives here
        predicates=VOICE_REGISTRY,            # or your own registry file
    )


server.setup_fnc = setup


@server.rtc_session()
async def entrypoint(ctx: JobContext) -> None:
    participant = await ctx.wait_for_participant()
    memory = ctx.proc.userdata["recall"].for_subject(participant.identity)

    session = AgentSession(
        stt=deepgram.STT(), llm=openai.LLM(model="gpt-4.1-mini"),
        tts=cartesia.TTS(), vad=silero.VAD.load(),
    )
    await session.start(
        agent=RecallAgent(
            memory=memory,
            instructions="You are the support line for Acme. Use what you remember about the caller.",
        ),
        room=ctx.room,
    )


if __name__ == "__main__":
    cli.run_app(server)

What happens on a call:

1. on_enter loads the caller's current beliefs and appends them to the instructions inside a <recall_memory> block, before the greeting.
2. The caller says "actually call me Sam, and I moved to Denver". The model calls remember twice. Recall supersedes the old name and timezone, and the agent rewrites its instructions so the next sentence already uses Sam.
3. A week later the same identity calls back and step 1 finds the facts.

Nothing is searched per turn. A caller's typed facts are a short list, so the whole set fits in the prompt. When a caller has more beliefs than limit (default 20), each user turn also runs a search and adds matching facts to that turn only.

Where the memory is stored

local_dir="./recall-data" keeps the database on the worker's machine. The first worker process to open the directory starts a polign-server for it in the background, listening on localhost only, and every other process shares that server. It keeps running between worker restarts; its process id is in recall-data/runtime.json and its log in recall-data/server.log. This is the quickest way to try the package and is fine for a single machine. It works on Linux and macOS.

Workers on several machines need one shared server. Run polign-server somewhere they can all reach (Get started shows how, including storing into S3, GCS or Azure), then connect to it instead:

RecallMemory.open(
    url="http://memory.internal:23000",   # or POLIGN_URL in the environment
    api_key=os.environ["POLIGN_API_KEY"],
    predicates=VOICE_REGISTRY,
)

For S3, run the server with -store s3://your-bucket/recall-livekit and give that server an AWS identity with access to the bucket. The worker connects to its HTTP(S) URL; do not pass an S3 URI as url or local_dir. The LiveKit example includes API-key setup and a check for memory recovery after a server restart.

GCS uses the same worker connection: start the server with -store gcs://your-bucket/recall-livekit and Google Application Default Credentials. See the GCS setup.

Your own Agent subclass

from recall_livekit import attach

agent = FrontDesk()                      # any Agent with string instructions
await attach(agent, memory, who="the guest")
await session.start(agent, room=ctx.room)

attach adds the memory block and the remember tool. It does not do the per-turn search for overflowing callers; use RecallAgent for that.

Resuming a call after a worker dies

If the worker running a call crashes, LiveKit hands the room to another worker, and by default the new agent starts the conversation over. With resume, it continues instead:

def setup(proc: JobProcess) -> None:
    proc.userdata["recall"] = RecallMemory.open(
        local_dir="./recall-data", predicates=VOICE_REGISTRY, agent=True,
    )

# in the entrypoint
agent = RecallAgent(
    memory=memory,
    instructions="You are the support line for Acme.",
    resume=ctx.room.name,     # names the call, not the caller
)

What changes:

1. on_enter resumes the call's records before the first reply. When the call is new, nothing else happens. When it was cut off, a <recall_resume> block goes after the memory block with what the agent had noted and the last turns, and tells the model not to greet the caller again.
2. Every user and agent message, and every tool call with its result, is recorded as a turn in the background, in order.
3. The model gets an update_working_state tool to note what the caller needs, what is done, and what it is doing now. That note, with the recent turns, is what the next worker continues from.
4. on_exit (which also runs when the session closes) writes any queued turns and releases the call.

Only one worker may write for a call, and a worker that died still holds it until its lease runs out. The new agent does not wait for that: it reads the call's records without the lease, so it has the briefing and speaks at once. Its turns wait in a buffer while it takes the lease in the background, then are written in order, after anything the dead worker managed to write. The default lease_ttl of 6 seconds keeps that gap short (the server accepts 5 seconds to 1 hour). If another worker still holds the call after wait seconds, this one records nothing and logs a warning. Like the memory reads, resuming and recording fail open: if Recall is down, the call goes on without a briefing and a warning is logged.

The id must be 1-128 letters, digits, ., _ or -; map room names with other characters to one. For more options, pass an AgentResume:

from recall_livekit import AgentResume

resume = AgentResume(proc.userdata["recall"], ctx.room.name, lease_ttl=10, token_budget=2000)
agent = RecallAgent(memory=memory, instructions="...", resume=resume)

Subclasses that override on_exit must await super().on_exit(). attach does not resume; use RecallAgent for that. Needs a polign CLI newer than 0.7.4, the first with polign mcp -agent.

Options

Custom predicates

Predicates are a JSON file. The remember tool's schema is built from it, so the model only ever sees the names you allow.

{
  "name":       { "cardinality": "single", "value_type": "string",  "description": "The name the caller asks to be called" },
  "open_issue": { "cardinality": "multi",  "value_type": "string",  "description": "A problem the caller reported that is not resolved yet" }
}

single means a newer value replaces the old one; multi means each value is an additional fact. value_type is string, number, or boolean. The file replaces Recall's built-in registry entirely. VOICE_REGISTRY is a starter set for callers: name, preferred language, callback number, email, timezone, account tier, recording consent, response style, open issues, owned products, and technologies used.

Subjects and retention

Use a stable, auth-derived identifier as the subject, such as the participant identity your token server issued. Never the room name. Recall keeps the record of every change while the current view updates; enable forget_tool if callers should be able to withdraw a fact by asking, and see the Recall docs for retention and export.

Development

cd python/recall-livekit
python -m pip install -e . pytest pytest-asyncio
pytest tests/unit_tests -q                # fake Recall subprocess, scripted model
pytest tests/integration_tests -v         # real polign-server and polign CLI

To include S3 persistence and authentication coverage, install pip install 'moto[server]>=5.1,<6' before running the integration tests. For GCS, install go install github.com/fsouza/fake-gcs-server@v1.55.0 and put the binary on PATH, or set FAKE_GCS_SERVER to its path. The same test runs against both emulators, writes through the real binaries, and checks corrections and withdrawals after server restarts with the disk cache disabled. It does not verify cloud IAM, real-bucket connectivity, or GCS gRPC/zonal modes. CI runs both.

The integration tests locate the binaries like the SDK's tests do (POLIGN_SERVER, POLIGN_SOURCE, POLIGN_SERVER_VERSION, or the latest release download) and skip when none is found. To test against the binaries pip installed, set POLIGN_SERVER="$(python -c 'import polign_db; print(polign_db.find_bin("polign-server"))')".