LiveKit

import asyncio
import os

import httpx
from livekit.agents import Agent, ChatContext, ChatMessage, ConversationItemAddedEvent

TWIN = "https://api.twinmemory.dev/v1/memory"


async def twin(user_id: str, route: str, body: dict) -> dict | None:
    async with httpx.AsyncClient(timeout=30) as http:
        r = await http.post(
            f"{TWIN}/{route}",
            headers={"Authorization": f"Bearer {os.environ['TWIN_MEMORY_API_KEY']}", "X-Twin-Account": user_id},
            json=body,
        )
    if r.status_code == 404:  # no memory for this end user yet
        return None
    r.raise_for_status()
    return r.json()


class Assistant(Agent):
    def __init__(self, user_id: str) -> None:
        super().__init__(instructions="You are a helpful voice assistant. Use the memory you are given.")
        self.user_id = user_id

    async def on_user_turn_completed(self, turn_ctx: ChatContext, new_message: ChatMessage) -> None:
        brief = await twin(self.user_id, "search", {"query": new_message.text_content}) or {}
        lines = [*brief.get("known", []), *(f"- {f['text']}" for f in brief.get("facts", []))]
        if lines:
            turn_ctx.add_message(role="assistant", content="What I remember:\n" + "\n".join(lines))


def remember_turns(session, user_id: str, call_id: str) -> None:
    turns = 0

    @session.on("conversation_item_added")
    def on_item(event: ConversationItemAddedEvent) -> None:
        nonlocal turns
        text = event.item.text_content
        if event.item.role not in ("user", "assistant") or not text:
            return
        turns += 1
        item = {"id": f"{call_id}-{turns:04d}", "session": call_id, "role": event.item.role, "text": text}
        asyncio.create_task(twin(user_id, "add", {"items": [item]}))