LiveKit

Give a LiveKit voice agent a memory of each caller. Before every reply, what is known about the caller that bears on what they just said goes in front of the LLM; once the agent has answered, the exchange is saved. The next call starts knowing what was said in this one.

Install

Terminal

pip install "livekit-agents[openai]" geniffy

Terminal

uv add "livekit-agents[openai]" geniffy

Add the speech plugins your agent already uses. Set GENIFFY_API_KEY from API keys in the Geniffy app.

The agent

agent.py

import asyncio

from geniffy import AsyncGeniffy
from livekit.agents import Agent, ConversationItemAddedEvent

geniffy = AsyncGeniffy()                          # reads GENIFFY_API_KEY


class Assistant(Agent):
    """A voice agent that knows each caller: what they said before bears on this call."""

    def __init__(self, user_id: str) -> None:
        super().__init__(instructions="You are a helpful voice assistant. Keep answers short.")
        self.mem = geniffy.space(f"user_{user_id}")
        self.held = ("", "")                      # the last question looked up, and what was found
        self.saving: set[asyncio.Task] = set()

    async def on_enter(self) -> None:
        self.session.on("conversation_item_added", self.remember)

    async def llm_node(self, chat_ctx, tools, model_settings):
        # before each reply, what is known about this caller that bears on what they just said
        asked = next((m.text_content for m in reversed(chat_ctx.items)
                      if getattr(m, "role", None) == "user" and m.text_content), "")
        if asked:
            if self.held[0] != asked:             # a reply with tool calls asks once, not once per step
                self.held = (asked, await self.mem.context(asked))
            chat_ctx = chat_ctx.copy()
            chat_ctx.add_message(role="system", content=f"<memory>\n{self.held[1]}\n</memory>")
        async for chunk in Agent.default.llm_node(self, chat_ctx, tools, model_settings):
            yield chunk

    def remember(self, ev: ConversationItemAddedEvent) -> None:
        # once the agent has replied, the exchange is saved to this caller's memory
        if getattr(ev.item, "role", None) != "assistant" or not ev.item.text_content or not self.held[0]:
            return
        task = asyncio.create_task(self.mem.memories.add(messages=[
            {"role": "user", "content": self.held[0]},
            {"role": "assistant", "content": ev.item.text_content},
        ]))
        self.saving.add(task)
        task.add_done_callback(self.saving.discard)

The memory goes into a copy of the chat context for that reply only, so a long call never fills up with memory blocks. It works the same whether the caller spoke or typed. When nothing is known, the block says so in one sentence, so the agent says it doesn't know instead of guessing.

In your worker

Start the agent with the caller from your own sign-in, such as the participant's identity or metadata:

async def entrypoint(ctx: JobContext):
    await ctx.connect()
    caller = await ctx.wait_for_participant()
    session = AgentSession(stt=stt, llm=llm, tts=tts, vad=vad)
    await session.start(agent=Assistant(user_id=caller.identity), room=ctx.room)

Each exchange is added as a conversation. Geniffy keeps who said what, so what the caller said becomes a fact about them, and what your agent said stays the agent's. A caller who asks to be forgotten is one call: forget_space(f"user_{caller.identity}").

Last updated October 6, 2026