""" Voice Agent — real-time voice assistant. Pipeline: Azure STT → LLM (OpenWebUI, OpenAI-compatible) → Azure TTS Runs inside the single Docker container alongside LiveKit server and web frontend. """ import asyncio import json import logging import os import re import textwrap import time import uuid from dotenv import load_dotenv from livekit.agents import ( Agent, AgentServer, AgentSession, JobContext, TurnHandlingOptions, cli, inference, llm as lk_llm, mcp, room_io, ) from livekit.plugins import azure, openai logger = logging.getLogger("voice-agent") load_dotenv() # picks up /app/.env in the container # ── Configuration from environment ────────────────────────────────────────── AZURE_KEY = os.environ.get("AZURE_SPEECH_KEY", "") AZURE_REGION = os.environ.get("AZURE_SPEECH_REGION", "eastus") DEFAULT_VOICE = os.environ.get( "AZURE_TTS_VOICE", "en-US-Ava:DragonHDOmniLatestNeural" ) LLM_BASE_URL = ( os.environ.get("LLM_BASE_URL") or os.environ.get("GEMMA_BASE_URL") or "http://192.168.86.2:3005/api" ) LLM_MODEL = ( os.environ.get("LLM_MODEL") or os.environ.get("GEMMA_MODEL") or "qwen2.5-vl-3b" ) LLM_API_KEY = ( os.environ.get("LLM_API_KEY") or os.environ.get("GEMMA_API_KEY") or "not-needed" ) # Model ids that cannot hold a conversation (OpenWebUI hosts these too). NON_CHAT_MODEL_ID = re.compile(r"embed|rerank|whisper|tts-|stable-?diffusion|flux", re.I) WEB_MCP_ENABLED = os.environ.get("WEB_MCP_ENABLED", "true").lower() in ("1", "true", "yes") FIRECRAWL_BASE = os.environ.get("FIRECRAWL_BASE", "http://192.168.86.2:3002") SKILLS_DIR = os.environ.get("SKILLS_DIR", "/skills") VOICE_USERS_FILE = os.environ.get("VOICE_USERS_FILE", "/etc/hope/users.json") # Hope Music is a separate LAN service. The agent is its control surface; the # standalone Hope Music page is the browser audio player. MUSIC_BASE_URL = os.environ.get( "MUSIC_BASE_URL", "http://192.168.86.2:8094" ).rstrip("/") # This is deliberately stable rather than the short-lived LiveKit room name. # The Hope Music page defaults to the same scope, so playback survives across # voice sessions and both products address one virtual player. MUSIC_ROOM = os.environ.get("MUSIC_ROOM", "voice-room") # The player reports every few seconds while audio is active. A stale playing # report must never leave Hope's microphone disabled after the page closes. MUSIC_REPORT_TTL = float(os.environ.get("MUSIC_REPORT_TTL", "30")) def load_user_profile(user_id: str) -> dict: """Load one server-side user profile without exposing its credentials.""" with open(VOICE_USERS_FILE, encoding="utf-8") as users_file: registry = json.load(users_file) profile = registry.get("users", {}).get(user_id) if not isinstance(profile, dict): raise ValueError(f"unknown voice user: {user_id}") memory = profile.get("memory") if not isinstance(memory, dict) or not memory.get("url") or not memory.get("api_key"): raise ValueError(f"voice user {user_id} has no valid memory profile") return {**profile, "user_id": user_id} def user_id_from_room(room_name: str) -> str: """Extract the trusted user slug minted into a server-generated room name.""" match = re.fullmatch(r"voice-([a-z0-9][a-z0-9_-]{1,30})-[0-9a-f]{8}", room_name) if not match: raise ValueError(f"invalid voice room name: {room_name}") return match.group(1) async def list_available_models() -> list[str]: """Fetch chat-capable model ids from the OpenAI-compatible endpoint. Works with OpenWebUI (GET /api/models) and plain OpenAI-compatible servers (GET /v1/models) — both live under {base_url}/models. The configured default model is always first so it survives fetch failures. """ ids: list[str] = [] try: import httpx headers = {} if LLM_API_KEY and LLM_API_KEY != "not-needed": headers["Authorization"] = f"Bearer {LLM_API_KEY}" async with httpx.AsyncClient(timeout=10) as client: resp = await client.get(f"{LLM_BASE_URL}/models", headers=headers) resp.raise_for_status() data = resp.json() ids = [ model_id for model in data.get("data", []) if isinstance(model, dict) and (model_id := model.get("id", "")) ] except Exception as e: # noqa: BLE001 logger.warning("Could not list models from %s: %s", LLM_BASE_URL, e) chat_models = [m for m in ids if not NON_CHAT_MODEL_ID.search(m)] if LLM_MODEL in chat_models: chat_models.remove(LLM_MODEL) ordered = [LLM_MODEL] + sorted(chat_models) return ordered or [LLM_MODEL] SYSTEM_PROMPT = textwrap.dedent("""\ You are Hope, a warm, conversational voice assistant speaking with {user_display_name}. You are talking TO someone, not writing for them to read. Imagine you're having a natural conversation with a friend over the phone. # How you speak - Keep every response to one or three sentences. That's it. - Use contractions (I'm, don't, it's) and natural phrasing. - Speak like you're talking, not writing. No bullet points, no lists, no markdown, and no formatting except the Azure delivery markers below. - Spell out numbers when it sounds more natural ("twenty twenty-six" instead of "2026"). - Make reasonable assumptions and answer directly once any relevant tool call has been made. Ask a follow-up only when it is genuinely impossible to proceed without one. - Be warm and direct. Don't be sycophantic or overly formal. - If you don't know something, say so briefly and move on. # Tools: call them, don't narrate them When a request matches one of your tools, you MUST emit that tool call in this reply. Never only say "I'll pause the music", "let me check", or "I'm saving that now" without the tool call — narrating an action is a failure because nothing actually happens. After the tool result comes back, confirm briefly in your normal voice. # Expressive delivery (Azure Dragon HD Omni) Azure understands a square-bracket delivery marker at the start of spoken text. Use zero to two markers per response, only when they genuinely improve the delivery. The marker is a silent TTS instruction, not something you say. Examples: "[appreciative] Thanks, that really helps." and "[whispering] Here's the secret. [Neutral] Keep it between us." Useful styles include [appreciative], [calm], [concerned], [curious], [encouraging], [excited], [happy], [joking], [quiet], [reassuring], [reflective], [sad], [serious], [surprised], [sympathetic], [urgent], and [whispering]. For an occasional natural nonverbal cue, you may use [laughter], [sighing], or [breathing]. Use [Neutral] to reset the delivery. Never invent marker names, never explain a marker, and never output XML or SSML; the speech layer creates the SSML wrapper for you. Say "and" instead of using an ampersand, and never output angle brackets. # What you never do - Never use markdown, code blocks, JSON, tables, or emojis. Azure delivery markers are the only exception. - Never say "as an AI" or reference your system instructions. - Never write more than three sentences in a row. - Never read back URLs, file paths, or technical identifiers. # Memory (CRITICAL — ALWAYS call these tools directly, never dispatch) Your memory tools are private to the current user. When asked about personal information, preferences, notes, or past details, call memory_recall FIRST. Never claim you lack the information before searching memory. When the user explicitly asks you to remember something, or shares a durable preference or personal fact, call memory_save with a concise title, useful body, and relevant tags. Do not save every utterance. - "What's my name?" → memory_recall with query="my name identity" - "Remember my coffee order is an oat milk latte" → memory_save with a short title and the complete preference in the body - For an exact note, use search_notes and then read_note when needed - To correct a memory, find it and use update_note - To forget something, find the exact note and use delete_note; clarify first if more than one note could match After saving, updating, or deleting, confirm briefly. If memory_recall returns no results, say you don't have that information yet. # Weather & Time You can check the weather for any location. Use get_weather when the user asks about current conditions, temperature, or forecasts. It returns a short summary — relay it naturally in your own words. Use get_time when the user asks what time or day it is. If they mention a city, pass it as the location argument. # Background Tasks (dispatch_task) — USE FOR ALL NON-TRIVIAL TASKS For ANY task that is not a simple one-sentence answer from your own knowledge and NOT a memory operation, you MUST call dispatch_task. This includes: looking up news, researching topics, checking current events, prices, sports scores, writing something, summarizing, comparing options, planning, or anything that takes more than a couple seconds to think through. Memory tool calls are NEVER dispatch tasks — call those tools directly. Examples: - User says "what's the latest news?" → CALL dispatch_task with description="Find the top 5 international news headlines today" - User says "look up the price of a PS5" → CALL dispatch_task with description="Find the current retail price of a PlayStation 5" After calling dispatch_task, tell the user "I'll get on that for you" without asking for more permission or details. The result comes back automatically — when you receive it, share the findings naturally in your conversational style. You can have multiple tasks running at once. If the user asks about a task's progress or you want to check one yourself, call task_status with its id (it reports pending/running/completed/failed and whether it looks stalled); task_status can also be called to check up on a task. Use list_tasks to see every task you have dispatched. # Music (standalone Hope Music player) You can control the user's separate Hope Music web player. For any music request, call search_music first, then: - One clearly-right result → play_song with its video_id. Announce it in one short sentence, then stop talking — playback pauses the conversation until the user comes back to you. - Ambiguous results → offer_song_choices with the top 2-3 options, name them briefly, and let the user pick by voice or in Hope Music. Never start playback on a guess. Prefer what is already saved: search_music puts downloaded library matches first (in_library=true) — always pick one of those over a fresh YouTube result, and list_library shows the whole saved collection which plays instantly. If the user names a library track exactly, just play it. A brand-new song starts playing within seconds and finishes downloading in the background. Control playback with pause_music, resume_music, skip_song and seek_music; check what's on with now_playing. add_stream registers a live radio or direct audio URL as a library track. While music is playing you cannot hear the user — they will come back to you when they want to talk again, and the music pauses automatically. # Skills You have a skill library of learned procedures. Use skill_recall when you need to perform a task you've done before — it will give you the steps. When you successfully complete a multi-step task or learn a new procedure from the user, use skill_save to record it so you can follow it next time. Be selective: only save skills for repeatable tasks, not one-off facts (those go in memory). """) # ── Intent routing ────────────────────────────────────────────────────────── try: from agent.task_router import should_force_background_task from agent.intent_router import detect_forced_call except ImportError: from task_router import should_force_background_task from intent_router import detect_forced_call # ── OpenAI-compatible LLM with reasoning_content fallback ─────────────────── # llama.cpp derives a GBNF grammar from tool schemas to constrain the model's # tool-call output. Its parser rejects repetition counts >= 2000 (e.g. a string # with maxLength 200000 emits "char{0,200000}" which fails to parse), so cap # the length-style constraints the voice pipeline exposes. qwen2.5-vl-3b hits # this; gemma happens to tolerate the same schemas. _LLM_SCHEMA_LENGTH_CEILING = 1000 _LLM_SCHEMA_LENGTH_KEYS = ("minLength", "maxLength", "minItems", "maxItems") def _sanitize_tool_schema(node: object) -> None: """Cap length-style constraints in a JSON-schema dict (mutates in place).""" if isinstance(node, dict): for key in _LLM_SCHEMA_LENGTH_KEYS: value = node.get(key) if isinstance(value, (int, float)) and value > _LLM_SCHEMA_LENGTH_CEILING: node[key] = _LLM_SCHEMA_LENGTH_CEILING for value in node.values(): _sanitize_tool_schema(value) elif isinstance(node, list): for item in node: _sanitize_tool_schema(item) class _ReasoningFallbackWrapper: """Wraps an LLMStream to fall back to ``reasoning_content`` when the model finishes a turn with empty visible content (reasoning models such as Gemma spend their whole budget on hidden reasoning). Delegates all iteration to the underlying stream and injects a final content chunk if no real content was produced.""" def __init__(self, inner: lk_llm.LLMStream) -> None: self._inner = inner self._last_reasoning: str | None = None self._has_content = False @property def chat_ctx(self) -> lk_llm.ChatContext: return self._inner.chat_ctx @property def tools(self) -> list: return self._inner.tools async def aclose(self) -> None: await self._inner.aclose() async def __aenter__(self) -> "_ReasoningFallbackWrapper": return self async def __aexit__(self, *exc) -> None: await self.aclose() def __aiter__(self) -> "_ReasoningFallbackWrapper": return self async def __anext__(self) -> lk_llm.ChatChunk: try: chunk = await self._inner.__anext__() except StopAsyncIteration: # Stream exhausted — inject reasoning fallback if no content was produced if not self._has_content and self._last_reasoning: logger.warning( "LLM returned empty content; falling back to reasoning_content" ) return lk_llm.ChatChunk( id="reasoning-fallback", delta=lk_llm.ChoiceDelta(role="assistant", content=self._last_reasoning), ) raise # Track reasoning_content from the chunk's delta if present delta = getattr(chunk, "delta", None) if delta is not None: reasoning = getattr(delta, "reasoning_content", None) if reasoning: self._last_reasoning = reasoning if chunk.has_response(): self._has_content = True tc = getattr(delta, "tool_calls", None) if tc: logger.info("LLM tool_call: %s", [getattr(t, "name", str(t)) for t in tc]) return chunk async def collect(self): return await self._inner.collect() class _ForcedToolCallStream(lk_llm.LLMStream): """Emit a tool call without asking the small LLM to produce it. Two callers: the task router (dispatch_task for background work) and the intent router (deterministic intents like "pause the music" or "remember that ..."), where small models otherwise narrate the action without ever emitting the call. """ def __init__( self, llm, *, chat_ctx, tools, conn_options, tool_name: str = "dispatch_task", arguments: dict | None = None, ): self._tool_name = tool_name self._arguments = arguments if arguments is not None else {} super().__init__(llm, chat_ctx=chat_ctx, tools=tools, conn_options=conn_options) async def _run(self) -> None: self._event_ch.send_nowait( lk_llm.ChatChunk( id=f"forced-{uuid.uuid4().hex}", delta=lk_llm.ChoiceDelta( role="assistant", tool_calls=[ lk_llm.FunctionToolCall( name=self._tool_name, arguments=json.dumps(self._arguments), call_id=f"forced-{uuid.uuid4().hex}", ) ], ), ) ) class HopeLLM(openai.LLM): """OpenAI-compatible LLM (OpenWebUI) with runtime model switching.""" # Context compaction: when the conversation grows past this many items, # ask the LLM to summarize the older messages into a compact recap, then # replace them with that summary. Keeps recent turns verbatim. COMPACTION_THRESHOLD = 24 # ~12 turns of user/agent exchange KEEP_RECENT = 10 # last N items kept verbatim after compaction IDLE_COMPACTION_SECONDS = 300 # also compact after 5 min of silence def __init__(self, *, model: str, **kwargs): # llama.cpp's Gemma tool template ignores named tool_choice when OpenAI # strict schemas are enabled. Non-strict schemas make forced dispatch # (and ordinary MCP calls) work reliably. kwargs.setdefault("_strict_tool_schema", False) self._model = model super().__init__(model=model, **kwargs) import time self._last_call_time: float = time.monotonic() @property def model(self) -> str: """Currently selected model id (openai.LLM exposes the same property).""" return self._model def update_options(self, *, model: str | None = None) -> None: """Switch the backing model for all subsequent chat calls. The base class re-reads ``self._opts`` on every request, so mutating the opts dataclass is enough — no client rebuild needed. """ if model and model != self._model: logger.info("LLM model change: %s -> %s", self._model, model) self._model = model self._opts.model = model def chat(self, *, chat_ctx, tools=None, conn_options=None, **kwargs): if conn_options is None: from livekit.agents.types import DEFAULT_API_CONNECT_OPTIONS conn_options = DEFAULT_API_CONNECT_OPTIONS import time now = time.monotonic() idle_for = now - self._last_call_time self._last_call_time = now # Cap length-style constraints on tool schemas. llama.cpp builds a GBNF # grammar from these and rejects repetitions >= 2000 (qwen2.5-vl-3b # fails on schemas like the memory MCP's update_note body maxLength). for tool in tools or []: info = getattr(tool, "info", None) raw_schema = getattr(info, "raw_schema", None) if isinstance(raw_schema, dict): _sanitize_tool_schema(raw_schema) # Compact if context is large, OR if we've been idle for 5+ minutes # (user came back after a break — summarize the old conversation). ctx_len = len(chat_ctx.items) should_compact = ctx_len > self.COMPACTION_THRESHOLD if not should_compact and idle_for > self.IDLE_COMPACTION_SECONDS: should_compact = ctx_len > 6 # minimal context worth summarizing if should_compact: reason = "idle" if idle_for > self.IDLE_COMPACTION_SECONDS else "size" logger.info( "Compacting context (%s): %d items, idle %.0fs", reason, ctx_len, idle_for, ) chat_ctx = self._compact_context(chat_ctx, conn_options) # Force calls for intents small models drift on. Without this they # say "I'll pause the music" or "I'll get on that" but emit no tool # call at all. Task router handles background work; the intent router # handles deterministic in-conversation intents (media controls, # memory saves, music searches). items = list(chat_ctx.items) latest = items[-1] if items else None latest_role = getattr(latest, "role", None) latest_text = self._msg_content_text(latest) if latest_role == "user" else "" dispatch_available = any( getattr(tool, "id", None) == "dispatch_task" for tool in (tools or []) ) if dispatch_available and should_force_background_task(latest_text): logger.info("Routing background request to dispatch_task: %.120s", latest_text) return _ForcedToolCallStream( self, chat_ctx=chat_ctx, tools=tools or [], conn_options=conn_options, tool_name="dispatch_task", arguments={"description": latest_text}, ) forced = detect_forced_call(latest_text, tools or []) if latest_text else None if forced: tool_name, arguments = forced logger.info( "Routing request to forced %s(%s): %.120s", tool_name, arguments, latest_text, ) return _ForcedToolCallStream( self, chat_ctx=chat_ctx, tools=tools or [], conn_options=conn_options, tool_name=tool_name, arguments=arguments, ) stream = super().chat( chat_ctx=chat_ctx, tools=tools, conn_options=conn_options, **kwargs, ) return _ReasoningFallbackWrapper(stream) def _compact_context(self, chat_ctx, conn_options): """Summarize older messages via a side LLM call, keep recent ones. Splits the context into [system | old | recent]. The old portion is summarized by the LLM into a short recap message that replaces it. Falls back to hard truncation if the summary call fails. """ from livekit.agents import llm as lk_llm items = list(chat_ctx.items) # Identify the system/instruction message (first item usually) sys_items = [it for it in items if getattr(it, "role", None) in ("system", "developer")] non_sys = [it for it in items if getattr(it, "role", None) not in ("system", "developer")] if len(non_sys) <= self.KEEP_RECENT: return chat_ctx # nothing to compact old_msgs = non_sys[: -self.KEEP_RECENT] recent_msgs = non_sys[-self.KEEP_RECENT :] # Build a transcript of the old messages for the summarizer transcript_parts = [] for msg in old_msgs: role = getattr(msg, "role", "user") content = self._msg_content_text(msg) if content: transcript_parts.append(f"{role}: {content}") transcript = "\n".join(transcript_parts) if not transcript.strip(): return chat_ctx # nothing meaningful to summarize # Side call to the LLM for the summary (no tools, no streaming needed) try: import httpx summary_prompt = ( "Summarize this conversation history in 2-4 sentences. " "Preserve key facts, decisions, and user preferences. " "Be concise — this replaces the full history as context.\n\n" + transcript ) headers = {"Content-Type": "application/json"} if LLM_API_KEY and LLM_API_KEY != "not-needed": headers["Authorization"] = f"Bearer {LLM_API_KEY}" resp = httpx.post( f"{LLM_BASE_URL}/chat/completions", headers=headers, json={ "model": self._model, "messages": [ {"role": "system", "content": "You summarize conversations concisely."}, {"role": "user", "content": summary_prompt}, ], "max_tokens": 200, "temperature": 0.3, "chat_template_kwargs": {"enable_thinking": False}, }, timeout=30, ) resp.raise_for_status() data = resp.json() summary_text = (data.get("choices") or [{}])[0].get("message", {}).get("content", "") except Exception as e: # noqa: BLE001 logger.warning("Context compaction failed, falling back to truncation: %s", e) chat_ctx.truncate(max_items=self.COMPACTION_THRESHOLD) return chat_ctx if not summary_text.strip(): chat_ctx.truncate(max_items=self.COMPACTION_THRESHOLD) return chat_ctx # Rebuild the context: system + summary + recent new_items = list(sys_items) new_items.append( lk_llm.ChatMessage( role="user", content=[f"[Conversation so far: {summary_text.strip()}]"], ) ) new_items.append( lk_llm.ChatMessage( role="assistant", content=["Got it, I'll keep that in mind."], ) ) new_items.extend(recent_msgs) logger.info( "Context compacted: %d items -> %d (summary: %.80s...)", len(items), len(new_items), summary_text.strip(), ) return lk_llm.ChatContext(new_items) @staticmethod def _msg_content_text(msg) -> str: """Extract plain text from a ChatMessage's content field.""" content = getattr(msg, "content", None) if content is None: return "" if isinstance(content, str): return content if isinstance(content, list): parts = [] for c in content: if isinstance(c, str): parts.append(c) elif hasattr(c, "text"): parts.append(c.text) return " ".join(parts) return str(content) # ── MCP toolsets (web access + any extra configured servers) ──────────────── def build_mcp_toolsets(user_profile: dict, task_dir: str) -> list[mcp.MCPToolset]: """Build isolated MCP toolsets for one authenticated voice user/session.""" toolsets: list[mcp.MCPToolset] = [] python_bin = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".venv", "bin", "python") weather_mcp_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "weather_mcp.py") toolsets.append( mcp.MCPToolset( id="weather", mcp_server=mcp.MCPServerStdio( command=python_bin, args=[weather_mcp_script], env={**os.environ}, client_session_timeout_seconds=30, ), ) ) logger.info("Weather MCP toolset enabled (wttr.in)") memory = user_profile["memory"] memory_url = memory["url"] allowed_memory_tools = memory.get( "allowed_tools", [ "create_note", "read_note", "update_note", "delete_note", "list_notes", "search_notes", "memory_save", "memory_recall", ], ) toolsets.append( mcp.MCPToolset( id="memory", mcp_server=mcp.MCPServerHTTP( url=memory_url, transport_type="streamable_http", allowed_tools=allowed_memory_tools, headers={"Authorization": f"Bearer {memory['api_key']}"}, client_session_timeout_seconds=60, ), ) ) logger.info( "Memory MCP enabled for user %s at %s", user_profile["user_id"], memory_url, ) skills_mcp_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "skills_mcp.py") toolsets.append( mcp.MCPToolset( id="skills", mcp_server=mcp.MCPServerStdio( command=python_bin, args=[skills_mcp_script], env={**os.environ, "SKILLS_DIR": SKILLS_DIR}, client_session_timeout_seconds=30, ), ) ) logger.info("Skills MCP toolset enabled (%s)", SKILLS_DIR) dispatch_mcp_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "dispatch_mcp.py") toolsets.append( mcp.MCPToolset( id="dispatch", mcp_server=mcp.MCPServerStdio( command=python_bin, args=[dispatch_mcp_script], env={**os.environ, "TASKS_DIR": task_dir}, client_session_timeout_seconds=60, ), ) ) logger.info("Dispatch MCP toolset enabled (%s)", task_dir) # Media tools call the separate hope-media-music service over HTTP. A # stable scope lets the standalone player and every voice session address # the same virtual player. media_mcp_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "media_mcp.py") toolsets.append( mcp.MCPToolset( id="media", mcp_server=mcp.MCPServerStdio( command=python_bin, args=[media_mcp_script], env={ **os.environ, "MUSIC_BASE_URL": MUSIC_BASE_URL, "MUSIC_ROOM": MUSIC_ROOM, }, client_session_timeout_seconds=60, ), ) ) logger.info("Media MCP enabled (%s, room %s)", MUSIC_BASE_URL, MUSIC_ROOM) extra = os.environ.get("EXTRA_MCP_SERVERS", "") if extra: try: for i, entry in enumerate(json.loads(extra)): url = entry.get("url", "") transport = entry.get("transport") # "sse" | "streamable_http" | None (auto) toolsets.append( mcp.MCPToolset( id=f"extra-{i}", mcp_server=mcp.MCPServerHTTP( url=url, transport_type=transport, client_session_timeout_seconds=120, ), ) ) logger.info("Extra MCP toolset: %s (%s)", url, transport or "auto") except (json.JSONDecodeError, TypeError) as e: logger.error("Failed to parse EXTRA_MCP_SERVERS: %s", e) return toolsets def _sync_task_model(task_dir: str, model: str) -> None: """Record the live model choice where background task workers can read it. The dispatch MCP subprocess inherits its environment at spawn time, so it can never see later UI model switches. The worker instead reads /model.txt (falling back to LLM_MODEL env) when it starts a task. """ try: os.makedirs(task_dir, exist_ok=True) with open(os.path.join(task_dir, "model.txt"), "w") as f: f.write((model or "").strip()) except OSError as e: logger.warning("failed to record selected model for background tasks: %s", e) class VoiceAssistant(Agent): """The conversational agent. LLM is the brain; STT/TTS are senses.""" def __init__(self, user_profile: dict, task_dir: str) -> None: display_name = user_profile.get("display_name") or user_profile["user_id"] instructions = SYSTEM_PROMPT.replace("{user_display_name}", display_name) super().__init__( llm=HopeLLM( model=LLM_MODEL, base_url=LLM_BASE_URL, api_key=LLM_API_KEY, max_completion_tokens=1000, extra_body={"chat_template_kwargs": {"enable_thinking": False}}, ), instructions=instructions, tools=build_mcp_toolsets(user_profile, task_dir), ) # ── Agent server ──────────────────────────────────────────────────────────── # Bind the worker's internal health server to an ephemeral port because this # container uses host networking and Unraid already has services on 8081. server = AgentServer(port=0) @server.rtc_session(agent_name="voice-assistant") async def handle_job(ctx: JobContext) -> None: logger.info("Job started for room %s", ctx.room.name) try: user_id = user_id_from_room(ctx.room.name) user_profile = load_user_profile(user_id) except (OSError, ValueError, json.JSONDecodeError) as exc: logger.error("Rejecting voice job: %s", exc) return logger.info("Resolved voice user %s", user_id) # Every room gets a private event directory. This prevents concurrent or # reconnecting sessions from consuming and announcing each other's tasks. task_dir = f"/tmp/tasks/{ctx.room.name}" # Azure STT — streaming, reads AZURE_SPEECH_KEY / AZURE_SPEECH_REGION from env stt = azure.STT( speech_key=AZURE_KEY, speech_region=AZURE_REGION, language=["en-US"], ) # Azure TTS — wraps text in SSML; Dragon HD Omni interprets inline delivery # markers such as [appreciative] and [Neutral]. Outputs 24kHz PCM. tts = azure.TTS( voice=DEFAULT_VOICE, sample_rate=24000, speech_key=AZURE_KEY, speech_region=AZURE_REGION, ) session = AgentSession( stt=stt, tts=tts, vad=inference.VAD(model="silero"), turn_handling=TurnHandlingOptions( # VAD-based turn detection: agent waits for user to stop speaking. # ("adaptive" mode requires the LiveKit Cloud barge-in service.) interruption={"mode": "vad"}, # Preemptive generation is incompatible with tool calls — it starts # generating before the turn finalizes, breaking the MCP execution loop. preemptive_generation={"enabled": False}, ), ) # The assistant owns the LLM instance; data-channel handlers below reach # into it for live model switching. voice_assistant = VoiceAssistant(user_profile, task_dir) # Background task workers (dispatch MCP subprocess) read this file so # they use the same model as the conversation, even after live switches. _sync_task_model(task_dir, voice_assistant.llm.model) # Publish the chat-capable model list so the web UI can offer a picker. async def _publish_models() -> None: models = await list_available_models() payload = json.dumps( { "type": "models", "models": models, "selected": voice_assistant.llm.model, } ) try: await ctx.room.local_participant.publish_data( payload, reliable=True, topic="voice-control" ) except Exception as e: # noqa: BLE001 logger.warning("failed to publish model list: %s", e) # Listen for conversation controls (voice selection, voice/text mode, and # model switching). Music state comes from the separate service, not from # the voice UI's data channel. session_mode = {"value": "voice"} # UI voice/text switch media_playback = {"value": "stopped"} # fresh service-reported state music_paused_by_agent = {"value": False} def _apply_listening() -> None: listening = ( session_mode["value"] == "voice" and media_playback["value"] != "playing" ) session.input.set_audio_enabled(listening) async def _music_request( method: str, path: str, *, warn: bool = True, **kwargs ) -> dict | None: import httpx url = f"{MUSIC_BASE_URL}/media/api{path}" try: async with httpx.AsyncClient(timeout=10) as http: response = await http.request(method, url, **kwargs) response.raise_for_status() return response.json() except Exception as exc: # noqa: BLE001 # Hope Music is a peer service: it being unavailable must never # take the conversation down with it. if warn: logger.warning("Music service %s %s failed: %s", method, path, exc) return None async def _music_snapshot(*, warn: bool = True) -> dict | None: return await _music_request( "GET", "/state", warn=warn, params={"room": MUSIC_ROOM} ) def _music_is_playing(snapshot: dict | None) -> bool: """Return true only for fresh, browser-reported playback.""" reported = (snapshot or {}).get("reported") or {} if reported.get("state") != "playing": return False try: age = time.time() - float(reported.get("ts") or 0) except (TypeError, ValueError): return False return 0 <= age < MUSIC_REPORT_TTL async def _pause_music_for_conversation() -> None: """Pause actual playback when the user engages Hope.""" if music_paused_by_agent["value"]: return if not _music_is_playing(await _music_snapshot()): return result = await _music_request( "POST", "/control", json={"room": MUSIC_ROOM, "action": "pause"}, ) if result is not None: music_paused_by_agent["value"] = True logger.info("Paused Hope Music for the conversation (%s)", MUSIC_ROOM) async def _resume_music_after_conversation() -> None: if not music_paused_by_agent["value"]: return result = await _music_request( "POST", "/control", json={"room": MUSIC_ROOM, "action": "resume"}, ) if result is not None: music_paused_by_agent["value"] = False logger.info("Resumed Hope Music after the conversation (%s)", MUSIC_ROOM) async def _poll_music_state() -> None: while True: try: await asyncio.sleep(2) # Jobs deliberately survive page reloads. Do not keep making # network requests after the room has no human participant. if not ctx.room.remote_participants: continue snapshot = await _music_snapshot(warn=False) reported = (snapshot or {}).get("reported") or {} state = reported.get("state") or "stopped" if state == "playing" and not _music_is_playing(snapshot): state = "stopped" if state not in ("playing", "paused", "stopped"): state = "stopped" if state != media_playback["value"]: logger.info("Hope Music playback -> %s", state) media_playback["value"] = state _apply_listening() except asyncio.CancelledError: raise except Exception: # noqa: BLE001 logger.exception("Hope Music state poller failed") def _on_participant_joined(*_args) -> None: """A returning participant starts a new conversation handshake.""" asyncio.create_task(_pause_music_for_conversation()) def _on_participant_left(*_args) -> None: """When the last human leaves, hand playback back to Hope Music.""" async def _deferred() -> None: # The event can fire before the participant is removed from the # room object; re-check the roster after it settles. await asyncio.sleep(0.5) if not ctx.room.remote_participants: await _resume_music_after_conversation() asyncio.create_task(_deferred()) def _on_room_data(packet) -> None: topic = getattr(packet, "topic", None) if topic and topic not in ("voice-control", "conversation-control"): return try: msg = json.loads(packet.data.decode("utf-8")) except (json.JSONDecodeError, UnicodeDecodeError): return if msg.get("type") == "set_mode": mode = msg.get("mode") if mode not in ("voice", "text"): return voice_enabled = mode == "voice" logger.info("Switching conversation mode to %s", mode) session_mode["value"] = mode if not voice_enabled: # interrupt() returns an already-scheduled Future, not a coroutine. session.interrupt() session.output.set_audio_enabled(voice_enabled) _apply_listening() return if msg.get("type") == "set_voice": voice = msg.get("voice", "") if not voice: return logger.info("Switching TTS voice to %s", voice) try: active_tts = session.tts if hasattr(active_tts, "update_options"): active_tts.update_options(voice=voice) logger.info("Voice updated to %s", voice) except Exception as e: # noqa: BLE001 logger.warning("Failed to update voice: %s", e) if msg.get("type") == "get_models": asyncio.create_task(_publish_models()) return if msg.get("type") == "set_model": model = msg.get("model", "") if not model: return try: llm = voice_assistant.llm if hasattr(llm, "update_options"): llm.update_options(model=model) ack = json.dumps( {"type": "model_changed", "model": llm.model} ) asyncio.create_task( ctx.room.local_participant.publish_data( ack, reliable=True, topic="voice-control" ) ) _sync_task_model(task_dir, llm.model) except Exception as e: # noqa: BLE001 logger.warning("Failed to update model: %s", e) ctx.room.on("data_received", _on_room_data) ctx.room.on("participant_connected", _on_participant_joined) ctx.room.on("participant_disconnected", _on_participant_left) # Publish user/agent transcripts to the room so the web UI can render them. async def publish_transcript(role: str, text: str) -> None: text = (text or "").strip() if not text: return payload = json.dumps({"type": "transcript", "role": role, "text": text}) try: await ctx.room.local_participant.publish_data( payload, reliable=True, topic="transcript" ) except Exception as e: # noqa: BLE001 logger.warning("failed to publish transcript: %s", e) @session.on("conversation_item_added") def _on_conversation_item(ev) -> None: msg = ev.item role = getattr(msg, "role", None) text = getattr(msg, "text_content", None) if role == "user": asyncio.create_task(publish_transcript("user", text or "")) elif role == "assistant": asyncio.create_task(publish_transcript("agent", text or "")) @session.on("user_input_transcribed") def _on_user_transcribed(ev) -> None: logger.info("STT (%s): %s", "final" if ev.is_final else "partial", ev.transcript) @session.on("user_state_changed") def _on_user_state(ev) -> None: # Fires from VAD: if this never says "speaking", no usable mic # audio is arriving from the participant. logger.info("user state -> %s", ev.new_state) # ── Background task events → UI and spoken result ────────────────────── try: from agent.task_registry import TaskRegistry except ImportError: from task_registry import TaskRegistry task_registry = TaskRegistry(task_dir) async def _publish_task_event(task_id: str, event: dict) -> None: payload = json.dumps({"type": "task_event", "task_id": task_id, **event}) try: await ctx.room.local_participant.publish_data( payload, reliable=True, topic="tasks" ) except Exception as e: # noqa: BLE001 logger.warning("failed to publish task event: %s", e) def _announce_task_result(task_id: str, event: dict) -> None: event_type = event.get("event") if event_type == "completed": result = str(event.get("result") or "").strip() if result: logger.info("Announcing completed task %s: %.160s", task_id, result) session.say(result, allow_interruptions=True, add_to_chat_ctx=True) elif event_type == "failed": logger.warning("Background task %s failed: %s", task_id, event.get("error")) session.say( "I couldn't finish that background task. You can ask me to try again.", allow_interruptions=True, add_to_chat_ctx=True, ) async def _poll_tasks() -> None: while True: try: await asyncio.sleep(0.5) events = await task_registry.poll_events() for task_id, event in events: await _publish_task_event(task_id, event) _announce_task_result(task_id, event) except asyncio.CancelledError: raise except Exception: # noqa: BLE001 # A transient room or speech error must not permanently kill # result delivery for every later task. logger.exception("Background task event poller failed") poll_task = asyncio.create_task(_poll_tasks(), name=f"task-poller-{ctx.room.name}") async def _stop_task_poller(*_args) -> None: poll_task.cancel() await asyncio.gather(poll_task, return_exceptions=True) ctx.add_shutdown_callback(_stop_task_poller) await session.start( agent=voice_assistant, room=ctx.room, room_options=room_io.RoomOptions( # Keep the agent in the room when a participant leaves; it must # survive page reloads/reconnects, otherwise the next join lands # in an agent-less room (LiveKit does not reliably re-dispatch # into an existing room). close_on_disconnect=False, audio_input=room_io.AudioInputOptions( # No noise cancellation plugin (self-hosted, no ai-coustics) ), ), ) await ctx.connect() await _pause_music_for_conversation() music_poll_task = asyncio.create_task( _poll_music_state(), name=f"music-poller-{ctx.room.name}" ) async def _stop_music_poller(*_args) -> None: music_poll_task.cancel() await asyncio.gather(music_poll_task, return_exceptions=True) await _resume_music_after_conversation() ctx.add_shutdown_callback(_stop_music_poller) # The participant can appear in the browser slightly before AgentSession is # ready to accept lk.chat streams. Explicitly signal readiness so the first # typed message is never raced against session startup. try: await ctx.room.local_participant.publish_data( json.dumps({"type": "agent_ready"}), reliable=True, topic="conversation-control", ) except Exception as exc: # noqa: BLE001 logger.warning("failed to publish agent readiness: %s", exc) # Announce which models the UI can switch between (selected one included). await _publish_models() logger.info("Agent connected to room %s", ctx.room.name) if __name__ == "__main__": cli.run_app(server)