""" 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 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") 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. 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. # 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. # Music (media player in the web UI) You can play music on the user's 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 the player panel. Never start playback on a guess. The library is permanent: list_library shows what is already downloaded, and those tracks play instantly without searching. If the user names a library track exactly, just play it. 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). """) # ── Background-task routing ───────────────────────────────────────────────── try: from agent.task_router import should_force_background_task except ImportError: from task_router import should_force_background_task # ── 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 dispatch_task call without asking the small LLM to route it.""" def __init__(self, llm, *, chat_ctx, tools, conn_options, description: str): self._description = description 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-dispatch-{uuid.uuid4().hex}", delta=lk_llm.ChoiceDelta( role="assistant", tool_calls=[ lk_llm.FunctionToolCall( name="dispatch_task", arguments=json.dumps({"description": self._description}), call_id=f"dispatch-{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 dispatch for clear background-work requests. Without this, # small local models often say "I'll get on that" but emit no tool # call at all. 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, description=latest_text, ) 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 talk to the local media server; the room name scopes player # state so concurrent sessions don't fight over one virtual player. media_server_url = os.environ.get("MEDIA_SERVER_URL", "http://127.0.0.1:8092") 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, "MEDIA_SERVER_URL": media_server_url, "MEDIA_ROOM": os.path.basename(task_dir.rstrip("/")) or "default", }, client_session_timeout_seconds=60, ), ) ) logger.info("Media MCP toolset enabled (%s)", media_server_url) 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 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) # 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 and voice/text mode). # Defined here so it captures this job's session via closure. # Music and conversation are mutually exclusive by design: while a track # plays the agent's ears are off; re-engaging the conversation pauses the # music, and ending the conversation lets the user resume it. session_mode = {"value": "voice"} # UI voice/text switch media_playback = {"value": "stopped"} # UI-reported player state def _apply_listening() -> None: listening = ( session_mode["value"] == "voice" and media_playback["value"] != "playing" ) session.input.set_audio_enabled(listening) def _on_room_data(packet) -> None: topic = getattr(packet, "topic", None) if topic and topic not in ("voice-control", "conversation-control", "media-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") == "media_state": state = msg.get("state") if state not in ("playing", "paused", "stopped"): return logger.info("Media playback -> %s (agent listening: %s)", state, session_mode["value"] == "voice" and state != "playing") media_playback["value"] = state _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" ) ) except Exception as e: # noqa: BLE001 logger.warning("Failed to update model: %s", e) ctx.room.on("data_received", _on_room_data) # 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)) elif role == "assistant": asyncio.create_task(publish_transcript("agent", text)) @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() # 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)