import logging from dotenv import load_dotenv from livekit import agents from livekit.agents import Agent, AgentServer, AgentSession, JobContext, room_io # from livekit.plugins import noise_cancellation, silero from livekit.plugins import noise_cancellation, silero from livekit.agents import llm, stt, tts, inference from livekit.plugins.turn_detector.multilingual import MultilingualModel from livekit.agents import AgentStateChangedEvent, MetricsCollectedEvent, metrics import time import httpx from livekit.agents import function_tool, RunContext, ToolError from livekit.agents import mcp from livekit.plugins import cartesia load_dotenv() logger = logging.getLogger(__name__) # Define your agent's behavior by extending the Agent class class Assistant(Agent): def __init__(self) -> None: super().__init__( instructions=( "Your default and primary language is Indonesian. Always respond in Indonesian unless prompted otherwise. " "If the user addresses you in English, you must immediately switch to English and respond in English." "Maintain the language chosen by the user throughout the conversation until they switch back to the other language." "Be professional, friendly, and accurate in both languages." "If the user asks a question in Indonesian, provide a comprehensive, clear, and natural-sounding Indonesian response." "If the user asks a question in English, provide a comprehensive, clear, and natural-sounding English response." "Always maintain the persona regardless of the language being used." "Help the caller fix issues without rambling, and keep replies under 3 sentences. " "You can also look up the weather if asked." "LiveKit by searching the documentation. When users ask about LiveKit " "features, APIs, or how to build something, use the docs search tools " "to find accurate information." ), ) # The @function_tool decorator registers this method as a tool the LLM can call @function_tool() async def lookup_weather( self, context: RunContext, # Gives access to the session, speech handle, and user data location: str, # Type hints help the LLM understand what arguments to pass ) -> dict: """Look up current weather for a location. Args: location: City name or location to get weather for. """ # The docstring above becomes the tool description the LLM sees # when deciding which tool to call async with httpx.AsyncClient() as client: # First, geocode the location to get coordinates geo_response = await client.get( "https://geocoding-api.open-meteo.com/v1/search", params={"name": location, "count": 1} ) geo_data = geo_response.json() if not geo_data.get("results"): raise ToolError(f"Could not find location: {location}") lat = geo_data["results"][0]["latitude"] lon = geo_data["results"][0]["longitude"] place_name = geo_data["results"][0]["name"] # Get current weather for those coordinates weather_response = await client.get( "https://api.open-meteo.com/v1/forecast", params={ "latitude": lat, "longitude": lon, "current": "temperature_2m,weather_code", "temperature_unit": "fahrenheit" } ) weather = weather_response.json() # Return a dict with the weather data # The LLM will use this to form a natural response return { "location": place_name, "temperature_f": weather["current"]["temperature_2m"], "conditions": weather["current"]["weather_code"] } server = AgentServer() # The entrypoint function runs when a participant joins the room @server.rtc_session() async def entrypoint(ctx: JobContext): # Configure the voice pipeline with STT, LLM, TTS, and VAD providers session = AgentSession( # LLM with fallback: OpenAI primary, Gemini backup llm=llm.FallbackAdapter( [ inference.LLM(model="openai/gpt-4.1-mini"), inference.LLM(model="google/gemini-2.5-flash"), ] ), # STT with fallback: AssemblyAI primary, Deepgram backup stt=stt.FallbackAdapter( [ inference.STT.from_model_string("assemblyai/universal-streaming:en"), inference.STT.from_model_string("deepgram/nova-3"), ] ), # TTS with fallback: Cartesia primary, Inworld backup tts=tts.FallbackAdapter( [ cartesia.TTS( model="sonic-3", voice="f786b574-daa5-4673-aa0c-cbe3e8534c02", ), #inference.TTS.from_model_string("cartesia/sonic-3:72a1dea2-302f-4f6b-8936-e28f8d1a51ba"), inference.TTS.from_model_string("inworld/inworld-tts-1"), ] ), vad=silero.VAD.load(), turn_detection=MultilingualModel(), preemptive_generation=True, mcp_servers=[ mcp.MCPServerHTTP(url="https://docs.livekit.io/mcp") ] ) # Aggregate data across all conversation turns usage_collector = metrics.UsageCollector() # Track End of Utterance timing (when turn detector decides user finished speaking) last_eou_metrics: metrics.EOUMetrics | None = None @session.on("metrics_collected") def _on_metrics_collected(ev: MetricsCollectedEvent): nonlocal last_eou_metrics # Capture EOU metrics for TTFA calculation if ev.metrics.type == "eou_metrics": last_eou_metrics = ev.metrics # Log each metric as it arrives and add to usage collector metrics.log_metrics(ev.metrics) usage_collector.collect(ev.metrics) async def log_usage(): # Print per-session summary (tokens, audio duration, costs) summary = usage_collector.get_summary() logger.info("Usage summary: %s", summary) ctx.add_shutdown_callback(log_usage) @session.on("agent_state_changed") def _on_agent_state_changed(ev: AgentStateChangedEvent): if ev.new_state == "speaking": if last_eou_metrics: # Calculate time since user finished speaking elapsed = time.time() - last_eou_metrics.timestamp logger.info(f"Time to first audio: {elapsed:.3f}s") # Start the session with noise cancellation enabled await session.start( agent=Assistant(), room=ctx.room, room_options=room_io.RoomOptions( audio_input=room_io.AudioInputOptions( noise_cancellation=noise_cancellation.BVC(), # Background voice cancellation ), ), record=False, # Enable recording for metrics and debugging ) if __name__ == "__main__": logging.basicConfig(level=logging.INFO) agents.cli.run_app(server)