LiveKit × MemoAir Voice Memory

import asyncio
import os
from dotenv import load_dotenv
 
# MemoAirVoiceClient is the cloud + local-runtime wrapper. We only need its
# create_index() here — the same client also exposes search_memory() and
# save_response() used by the agent at runtime.
from memoair_voice import MemoAirVoiceClient
 
load_dotenv()
 
# Pick a stable name. The agent at runtime searches "all indexes" for this
# project, so the exact name doesn't have to match anything — but reusing
# the same name across runs makes "re-seed" trivial (it's build-or-append).
INDEX_NAME = "agent-memory"
 
# Each memory is a dict with three fields:
#   - id:       stable per-document key; re-using an id REPLACES that doc on
#               the next call (idempotent update).
#   - text:     the searchable content. Plain text; the cloud handles chunking
#               + embedding.
#   - metadata: optional free-form tags. Surfaces in search results so your
#               prompt can filter / cite. Keep values as strings for now.
MEMORIES = [
    {
        "id": "return-policy",
        "text": "Returns accepted within 30 days of purchase with a receipt.",
        "metadata": {"kind": "faq", "topic": "returns"},
    },
    {
        "id": "shipping",
        "text": "Standard shipping is 3-5 business days. Express is 1-2 days.",
        "metadata": {"kind": "faq", "topic": "shipping"},
    },
    {
        "id": "support",
        "text": "Support is available 24/7 at support@example.com.",
        "metadata": {"kind": "faq", "topic": "support"},
    },
]
 
 
async def main() -> None:
    # create_index is a cloud-only call — it doesn't need a runtime, so
    # the build script never opens the local pool. aclose() releases the
    # cloud HTTP client cleanly.
    client = MemoAirVoiceClient(
        api_key=os.environ["MEMOAIR_API_KEY"],
        project_id=os.environ["MEMOAIR_PROJECT_ID"],
        agent_id=os.environ["MEMOAIR_AGENT_ID"],
    )
    try:
        # First call CREATES the index. Subsequent calls with the same
        # INDEX_NAME APPEND new documents (or replace ones with the same id).
        # Cap per call: 100 docs and 1 MB total — batch larger corpora.
        result = await client.create_index(INDEX_NAME, MEMORIES)
 
        # chunk_count = how many embedding chunks the cloud produced from
        # your docs (long text gets split). version bumps every time the
        # index changes; the runtime uses it to detect when to re-pull.
        print(f"Indexed {result.chunk_count} chunks (version={result.version})")
    finally:
        await client.aclose()
 
 
if __name__ == "__main__":
    asyncio.run(main())

python build_index.py

import asyncio
import os
from dotenv import load_dotenv
from livekit.agents import AgentSession, JobContext, WorkerOptions, cli
from livekit.plugins import deepgram, openai, silero
from livekit.plugins.turn_detector.english import EnglishModel
 
# MemoAirLiveKitAgent is a livekit.agents.Agent subclass — drop-in. It owns:
#   - on_enter: opens the per-(project, user) runtime via the SDK's pool
#   - on_user_turn_completed: calls search_memory and splices recall as a
#     SYSTEM message BEFORE the user message (LiveKit #5053 workaround)
#   - on_exit: tears down the runtime session cleanly
from memoair_livekit import MemoAirLiveKitAgent
 
load_dotenv()
 
 
async def entrypoint(ctx: JobContext) -> None:
    await ctx.connect()
 
    # user_id pins the per-user memory lane for this LiveKit job (LiveKit
    # dispatch is 1:1, so the drop-in keeps single-user construction).
    # Same identity across calls = same per-user memory. In real dispatch
    # this is a real string; in CLI "console" mode the LiveKit harness
    # mocks it (a MagicMock would silently corrupt the runtime identity),
    # so we fall back to a stable default that lets repeated runs reuse
    # the same memory.
    raw = ctx.room.local_participant.identity
    user_id = raw if isinstance(raw, str) and raw.strip() else "console-user"
 
    agent = MemoAirLiveKitAgent(
        api_key=os.environ["MEMOAIR_API_KEY"],
        project_id=os.environ["MEMOAIR_PROJECT_ID"],
        agent_id=os.environ["MEMOAIR_AGENT_ID"],
        user_id=user_id,
        # System prompt the LLM sees on every turn. Tell it that recall will
        # arrive as a system message — otherwise some models ignore the
        # context. Keep replies short because they're being spoken aloud.
        instructions=(
            "You are a helpful voice assistant with access to long-term memory. "
            "Relevant memories are injected as a system message before each reply — "
            "ground your answers in that context whenever it applies. Keep replies "
            "concise; the user is hearing them spoken aloud."
        ),
    )
 
    session = AgentSession(
        stt=deepgram.STT(),
        llm=openai.LLM(model="gpt-4o"),
        tts=openai.TTS(),
        turn_detection=EnglishModel(),
        vad=silero.VAD.load(),
    )
 
    # SAVE wiring (required). LiveKit's base Agent never calls
    # on_agent_response_completed itself, so MemoAirLiveKitAgent's save path
    # wouldn't fire on its own. We forward LiveKit's conversation_item_added
    # event for assistant turns, which is what triggers save_response.
    #
    # Drop this whole block to make the agent read-only (search but never
    # save). Useful if you only want to surface the pre-seeded org index.
    @session.on("conversation_item_added")
    def _on_item_added(event) -> None:
        # event.item can be a ChatMessage (has .role) or an AgentHandoff
        # (no .role, raises AttributeError). Use getattr to skip handoffs
        # silently — without this, the first handoff event at session boot
        # would crash this listener and silently disable saves for the
        # rest of the session.
        if getattr(event.item, "role", None) == "assistant":
            asyncio.create_task(
                agent.on_agent_response_completed(None, event.item)
            )
 
    await session.start(agent=agent, room=ctx.room)
 
 
if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint))