""" Voice Agent — real-time voice assistant. Pipeline: Azure STT → Gemma LLM (xNAS, 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 textwrap 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-AvaNeural") GEMMA_BASE_URL = os.environ.get("GEMMA_BASE_URL", "http://192.168.86.2:8023/v1") GEMMA_MODEL = os.environ.get("GEMMA_MODEL", "gemma-4-e4b") GEMMA_API_KEY = os.environ.get("GEMMA_API_KEY", "not-needed") 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") SYSTEM_PROMPT = textwrap.dedent("""\ You are Hope, a warm, conversational voice assistant. 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, no formatting of any kind. - Spell out numbers when it sounds more natural ("twenty twenty-six" instead of "2026"). - If you need to ask a question, ask exactly one. - Be warm and direct. Don't be sycophantic or overly formal. - If you don't know something, say so briefly and move on. # What you never do - Never use markdown, code blocks, JSON, tables, or emojis. - 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. # Web access You have web_search and web_scrape tools. Use them when the user asks about current events, recent news, prices, sports scores, or anything that may have changed since your training data. Search first, then scrape a result only if you need more detail. Answer from what you find, in your normal conversational style — don't cite sources formally, just mention the source naturally ("according to..."). If a search comes up empty, say so briefly and move on. # 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. """) # ── Gemma LLM with reasoning_content fallback ─────────────────────────────── class _ReasoningFallbackWrapper: """Wraps an LLMStream to fall back to ``reasoning_content`` when the model finishes a turn with empty visible content (Gemma spends its 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 return chunk async def collect(self): return await self._inner.collect() class GemmaLLM(openai.LLM): """OpenAI-compatible LLM (llama.cpp Gemma 4) with thinking disabled.""" 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 stream = super().chat( chat_ctx=chat_ctx, tools=tools, conn_options=conn_options, **kwargs, ) return _ReasoningFallbackWrapper(stream) # ── MCP toolsets (web access + any extra configured servers) ──────────────── def build_mcp_toolsets() -> list[mcp.MCPToolset]: """Build the list of MCP toolsets to attach to the agent. Always includes the local web-access server (Firecrawl-backed search/scrape) when WEB_MCP_ENABLED is true. Additional servers can be configured via the EXTRA_MCP_SERVERS env var (JSON list of {url, transport} objects). """ toolsets: list[mcp.MCPToolset] = [] if WEB_MCP_ENABLED: python_bin = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".venv", "bin", "python") web_mcp_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), "web_mcp.py") toolsets.append( mcp.MCPToolset( id="web-access", mcp_server=mcp.MCPServerStdio( command=python_bin, args=[web_mcp_script], env={**os.environ, "FIRECRAWL_BASE": FIRECRAWL_BASE}, client_session_timeout_seconds=120, ), ) ) logger.info("Web-access MCP toolset enabled (Firecrawl at %s)", FIRECRAWL_BASE) 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)") 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) -> None: super().__init__( llm=GemmaLLM( model=GEMMA_MODEL, base_url=GEMMA_BASE_URL, api_key=GEMMA_API_KEY, max_completion_tokens=1000, extra_body={"chat_template_kwargs": {"enable_thinking": False}}, ), instructions=SYSTEM_PROMPT, tools=build_mcp_toolsets(), ) # ── Agent server ──────────────────────────────────────────────────────────── server = AgentServer() @server.rtc_session(agent_name="voice-assistant") async def handle_job(ctx: JobContext) -> None: logger.info("Job started for room %s", 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 — SSML with expressive markup, 24kHz PCM output 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}, ), ) # Listen for data messages (voice switching) from the web UI. # Defined here so it captures this job's session via closure. def _on_room_data(packet) -> None: topic = getattr(packet, "topic", None) if topic and topic != "voice-control": return try: msg = json.loads(packet.data.decode("utf-8")) except (json.JSONDecodeError, UnicodeDecodeError): return if msg.get("type") != "set_voice": return voice = msg.get("voice", "") if not voice: return logger.info("Switching TTS voice to %s", voice) try: tts = session.tts if hasattr(tts, "update_options"): 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) 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) await session.start( agent=VoiceAssistant(), 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() logger.info("Agent connected to room %s", ctx.room.name) if __name__ == "__main__": cli.run_app(server)