Metadata-Version: 2.5
Name: recall-livekit
Version: 0.3.1
Summary: Typed, correctable long-term memory for LiveKit voice agents, backed by Recall by Polign
Project-URL: Homepage, https://polign.com
Project-URL: Documentation, https://polign.com/integrations.html
Project-URL: Repository, https://github.com/Polign/polign
Project-URL: Issues, https://github.com/Polign/polign/issues
Author: Polign
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: agent memory,livekit,polign,recall,voice agent
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Requires-Dist: livekit-agents<2,>=1.8
Requires-Dist: polign-recall>=0.18.0
Description-Content-Type: text/markdown

# recall-livekit

Long-term memory for [LiveKit Agents](https://github.com/livekit/agents) voice
agents, backed by [Recall by Polign](https://polign.com/recall.html).

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.

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

That is the whole install. pip also brings the
[`polign_db`](https://pypi.org/project/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

```python
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](https://polign.com/developers.html)
shows how, including storing into S3, GCS or Azure), then connect to it
instead:

```python
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](../../examples/livekit/README.md#use-an-s3-backed-store)
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](../../examples/livekit/README.md#use-a-gcs-backed-store).

## Your own Agent subclass

```python
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:

```python
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`:

```python
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

| Where | Option | Default | What it does |
|---|---|---|---|
| `RecallMemory.open` | `local_dir` | none | Keep the database in this directory and run its server; exclusive with `url` and `api_key` |
| `RecallMemory.open` | `url`, `api_key`, `collection`, `predicates` | worker environment | Connection for the subprocess (`POLIGN_URL`, `POLIGN_API_KEY`, `POLIGN_COLLECTION`, `POLIGN_PREDICATES`) |
| `RecallMemory.open` | `command` | `polign mcp -memory-only -write` | The subprocess argv, to run a `polign` binary other than the one pip installed |
| `RecallMemory.open` | `agent` | off | Turn on the agent resume tools that `resume` needs (adds `-agent` to the default argv) |
| `for_subject` | `limit` | 20 | Beliefs loaded into the prompt; above it, per-turn search kicks in |
| `for_subject` | `read_timeout`, `write_timeout` | 0.5 s, 5 s | Reads fail open (last known beliefs); writes tell the model the fact was not saved |
| `RecallAgent` | `context_template` | `<recall_memory>\n{context}\n</recall_memory>` | Wrapper around the block; must contain `{context}` |
| `RecallAgent` | `who` | `"the caller"` | How the block refers to the person |
| `RecallAgent` | `remember_tool`, `forget_tool` | on, off | Which tools the model gets |
| `RecallAgent` | `search_when_overflowed` | on | Per-turn search when the caller has more beliefs than `limit` |
| `RecallAgent` | `resume` | off | The call's agent id (such as the room name) or an `AgentResume`; continue the call from its records after a worker dies |
| `RecallAgent` | `resume_template` | `<recall_resume>\n{context}\n</recall_resume>` | Wrapper around the resume block; must contain `{context}` |
| `AgentResume` | `lease_ttl`, `wait` | 6 s, `lease_ttl` + 5 s | How long a dead worker keeps the call, and how long a new one keeps trying to take it before it stops recording |
| `AgentResume` | `token_budget`, `timeout` | server default (8000), 5 s | Bound on the briefing; bound on each record, update and release |

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

```json
{
  "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

```bash
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"))')"`.
