hope-voice-api/agent/agent.py

"""Hope's speech/transport layer. Each authenticated room talks directly to Hermes."""
import asyncio
import json
import logging
import os
import re

import httpx
from dotenv import load_dotenv
from livekit.agents import Agent, AgentServer, AgentSession, JobContext, TurnHandlingOptions, cli, inference, room_io
from livekit.plugins import azure
from livekit import rtc
from pathlib import Path

try:
    from agent.conversation_llm import ConversationLLM
except ImportError:
    from conversation_llm import ConversationLLM

try:
    from web.voice_catalog import normalize_default_voice
except ImportError:
    from voice_catalog import normalize_default_voice

load_dotenv()
logger = logging.getLogger("voice-agent")
CONTROL_URL = os.getenv("CONTROL_URL", "http://127.0.0.1:8092").rstrip("/")
CONTROL_SECRET = os.getenv("CONTROL_SERVICE_SECRET", "")
DEFAULT_VOICE = normalize_default_voice(os.getenv("AZURE_TTS_VOICE", ""))
SYSTEM_PROMPT = """You are Hope, {display_name}'s personal voice assistant, powered by their private Hermes agent.
Speak conversationally and keep ordinary answers to one to three sentences. Do not use markdown,
read URLs aloud, or expose hidden reasoning. Use your Hermes tools whenever needed; never claim an
operation succeeded before its result. For longer research provide a concise spoken summary of
what you actually found. Remember durable personal preferences using your private memory tools.
The voice client provides wait/progress feedback automatically; focus on the eventual answer.
"""

async def resolve_room(room_name: str) -> dict:
    if not CONTROL_SECRET:
        raise RuntimeError("Control-plane authentication is not configured")
    async with httpx.AsyncClient(timeout=10) as client:
        response = await client.get(f"{CONTROL_URL}/internal/rooms/{room_name}", headers={"X-Control-Secret": CONTROL_SECRET})
        response.raise_for_status()
        return response.json()

# Host networking shares ports with other NAS services; allocate a private
# ephemeral worker health port rather than claiming the SDK default 8081.
server = AgentServer(host="127.0.0.1", port=0, num_idle_processes=2)

@server.on("worker_registered")
def worker_registered(_worker_id, _server_info):
    # SDK exposes the allocated HTTP port only after start. Store PID too so a
    # stale file cannot mark an exited worker healthy.
    port = server._http_server.port
    Path("/tmp/hope-worker-health").write_text(f"{os.getpid()} {port}")

@server.rtc_session(agent_name=os.getenv("VOICE_AGENT_NAME", "voice-assistant"))
async def entrypoint(ctx: JobContext):
    try:
        account = await resolve_room(ctx.room.name)
    except Exception:
        logger.warning("Rejected unauthenticated or unavailable room")
        ctx.shutdown("Account unavailable")
        return
    user = account["user"]
    user_id = user["id"]
    expected_identity = f"user:{user_id}"
    mode = {"value": "voice"}
    tasks: set[asyncio.Task] = set()

    def spawn(coro):
        task = asyncio.create_task(coro)
        tasks.add(task)
        task.add_done_callback(tasks.discard)
        return task

    async def publish(payload: dict, topic: str):
        try:
            await ctx.room.local_participant.publish_data(json.dumps(payload), reliable=True, topic=topic)
        except Exception:
            logger.warning("Unable to deliver room event (%s)", topic)

    async def progress(payload: dict):
        await publish(payload, "assistant-progress")

    # AgentSession.say queues behind the response being generated. A separate
    # audio track delivers the notice while Hermes is working, without interrupting
    # Hermes or inserting synthetic speech into its conversation history.
    notice_source = rtc.AudioSource(24000, 1)
    notice_task = None

    async def stop_notice():
        nonlocal notice_task
        if notice_task:
            notice_task.cancel()
            await asyncio.gather(notice_task, return_exceptions=True)
            notice_task = None
        notice_source.clear_queue()

    async def play_notice(text: str):
        try:
            async with tts.synthesize(text) as stream:
                async for audio in stream:
                    if mode["value"] != "voice":
                        return
                    await notice_source.capture_frame(audio.frame)
            await notice_source.wait_for_playout()
        except asyncio.CancelledError:
            raise
        except Exception:
            logger.warning("Wait notice audio unavailable")

    async def wait_notice(text: str):
        nonlocal notice_task
        if mode["value"] == "voice":
            await stop_notice()
            notice_task = spawn(play_notice(text))

    async def model_settings():
        # Fetch at each turn, not just on join/poll: saved account changes apply
        # immediately to the next request even in already-connected rooms.
        current = await resolve_room(ctx.room.name)
        if current["user"]["id"] != user_id:
            raise RuntimeError("Account mismatch")
        return current["model_settings"]

    bridge = ConversationLLM(control_url=CONTROL_URL, secret=CONTROL_SECRET,
                             room_id=ctx.room.name, on_progress=progress,
                             on_wait_notice=wait_notice, on_answer_start=stop_notice)
    tts = azure.TTS(voice=DEFAULT_VOICE, sample_rate=24000,
                    speech_key=os.getenv("AZURE_SPEECH_KEY", ""), speech_region=os.getenv("AZURE_SPEECH_REGION", "eastus"))
    session = AgentSession(stt=azure.STT(speech_key=os.getenv("AZURE_SPEECH_KEY", ""),
                           speech_region=os.getenv("AZURE_SPEECH_REGION", "eastus"), language=["en-US"]),
                           tts=tts, vad=inference.VAD(model="silero"),
                           turn_handling=TurnHandlingOptions(interruption={"mode": "vad"},
                           preemptive_generation={"enabled": False}))
    assistant = Agent(instructions=SYSTEM_PROMPT.format(display_name=user.get("display_name") or user["username"]), llm=bridge)

    def on_data(packet):
        sender = getattr(packet, "participant", None)
        if not sender or sender.identity != expected_identity:
            return
        topic = getattr(packet, "topic", "")
        if topic not in ("voice-control", "conversation-control") or len(packet.data) > 4096:
            return
        try:
            message = json.loads(packet.data.decode())
        except (ValueError, UnicodeDecodeError):
            return
        if not isinstance(message, dict):
            return
        if topic == "conversation-control" and message.get("type") == "set_mode":
            value = message.get("mode")
            if value not in ("voice", "text"):
                return
            mode["value"] = value
            if value == "text":
                spawn(stop_notice())
                session.interrupt()
            session.output.set_audio_enabled(value == "voice")
            session.input.set_audio_enabled(value == "voice")
        elif topic == "voice-control" and message.get("type") == "set_voice":
            voice = message.get("voice")
            if isinstance(voice, str) and re.fullmatch(r"[A-Za-z0-9:_-]{1,120}", voice):
                tts.update_options(voice=voice)
        # Model/tool routing is server-managed. No model-change or local MCP path.

    ctx.room.on("data_received", on_data)

    @session.on("conversation_item_added")
    def on_item(event):
        item = event.item
        role = getattr(item, "role", None)
        text = getattr(item, "text_content", None)
        if role in ("user", "assistant") and text:
            spawn(publish({"type": "transcript", "role": "agent" if role == "assistant" else "user", "text": text}, "transcript"))

    @session.on("error")
    def on_error(event):
        if not getattr(event, "recoverable", False):
            spawn(publish({"type": "assistant_progress", "request_id": "session", "status": "failed",
                           "message": "Hope couldn't complete that response. Please try again."}, "assistant-progress"))

    async def check_access():
        """Account deletion/logout revokes existing rooms, not only future joins."""
        while True:
            await asyncio.sleep(2)
            try:
                await resolve_room(ctx.room.name)
            except httpx.HTTPStatusError as exc:
                if exc.response.status_code in (401, 403, 404, 410):
                    session.interrupt()
                    ctx.shutdown("Account access revoked")
                    return
            except (httpx.HTTPError, RuntimeError):
                # Fail closed after a control-plane outage rather than let an
                # unvalidated private session continue indefinitely.
                session.interrupt()
                ctx.shutdown("Account validation unavailable")
                return

    @session.on("user_state_changed")
    def on_user_state(event):
        if event.new_state == "speaking":
            spawn(stop_notice())

    async def cleanup(*_args):
        await stop_notice()
        for task in tuple(tasks):
            task.cancel()
        await asyncio.gather(*tuple(tasks), return_exceptions=True)
        await bridge.aclose()
        await notice_source.aclose()

    ctx.add_shutdown_callback(cleanup)
    await session.start(agent=assistant, room=ctx.room,
                        room_options=room_io.RoomOptions(close_on_disconnect=True))
    await ctx.connect()
    notice_track = rtc.LocalAudioTrack.create_audio_track("hope-wait-notice", notice_source)
    await ctx.room.local_participant.publish_track(notice_track, rtc.TrackPublishOptions(source=rtc.TrackSource.SOURCE_MICROPHONE))
    # Admit only the account identity granted by the authenticated bootstrap.
    for participant in ctx.room.remote_participants.values():
        if participant.identity != expected_identity:
            ctx.shutdown("Unexpected participant")
            return
    spawn(check_access())
    await publish({"type": "agent_ready"}, "conversation-control")

if __name__ == "__main__":
    cli.run_app(server)