livekit-plugins-engram 0.1.0

Engram memory plugin for LiveKit Agents — durable, explainable memory for voice and realtime AI.

livekit-plugins-engram

Durable, explainable memory for LiveKit Agents, powered by Engram (Lumetra).

Voice and realtime agents are stateless by default — every call starts from zero. This plugin gives your Agent a one-line memory backend so it can remember what the caller told you last Tuesday, recall it mid-sentence, and explain why it surfaced any given fact (BM25 + vector + knowledge-graph retrieval).

Install

pip install livekit-plugins-engram

Set your Engram API key:

export ENGRAM_API_KEY=eng_live_...

Get a key at https://lumetra.io.

Usage

The plugin exposes a single Engram class. Use it from your Agent's lifecycle hooks (record what the user just said, recall relevant context before the LLM responds), or hand the model a pair of function tools so it can manage memory itself.

Pattern 1 — record + recall in on_user_turn_completed

from livekit.agents import Agent, AgentSession, JobContext, WorkerOptions, cli
from livekit.plugins import openai, deepgram, silero
from livekit.plugins.engram import Engram

memory = Engram(bucket="caller-default")  # picks up ENGRAM_API_KEY

class Receptionist(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions=(
                "You are a friendly receptionist. Use the memory provided "
                "in the system messages to personalize replies."
            ),
        )

    async def on_user_turn_completed(self, chat_ctx, new_message):
        text = new_message.text_content
        if not text:
            return
        # Save what the user just said
        await memory.astore_memory(text)
        # Pull anything relevant from prior sessions
        recall = await memory.aquery_memory(text)
        if recall.get("answer"):
            chat_ctx.add_message(
                role="system",
                content=f"Relevant memory: {recall['answer']}",
            )

async def entrypoint(ctx: JobContext):
    session = AgentSession(
        stt=deepgram.STT(),
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=openai.TTS(),
        vad=silero.VAD.load(),
    )
    await session.start(agent=Receptionist(), room=ctx.room)

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))

Pattern 2 — give the model memory tools

from livekit.agents import Agent
from livekit.plugins.engram import Engram, engram_tools

memory = Engram(bucket="caller-default")

agent = Agent(
    instructions="When the user shares a fact, call store_memory. "
                 "When you need history, call query_memory.",
    tools=engram_tools(memory),
)

Per-caller buckets

In a voice deployment you typically want one bucket per caller / room / customer:

memory = Engram(bucket=f"caller-{ctx.room.name}")

Buckets are created lazily on first write — no admin step required.

API

Engram(*, api_key=None, bucket=None, base_url="https://api.lumetra.io", timeout=30.0)

Method (sync / async)	Returns
store_memory(content, bucket=None) / astore_memory(...)	{} (201)
query_memory(query, bucket=None) / aquery_memory(...)	{"success": True, "answer": "..."}
list_memories(bucket=None, *, limit=50) / alist_memories(...)	{"memories": [...], "total": N}
list_buckets(*, limit=50, offset=0) / alist_buckets(...)	{"buckets": [...]}
delete_memory(memory_id, bucket=None) / adelete_memory(...)	{}
clear_memories(bucket=None) / aclear_memories(...)	{}

Inside a LiveKit Agent (which is asyncio), prefer the a* coroutines.

Configuration

Env var	Purpose
ENGRAM_API_KEY	API key (eng_live_...)

Links

• Engram docs — https://docs.lumetra.io
• LiveKit Agents — https://docs.livekit.io/agents/
• Source — https://github.com/lumetra-ai/engram-livekit

License

MIT © Lumetra