import json
import logging
import os
import time
from concurrent.futures import ThreadPoolExecutor
from livekit import rtc
from livekit.agents import (
Agent,
AgentServer,
AgentSession,
JobContext,
JobProcess,
cli,
inference,
room_io,
)
from livekit.plugins import cartesia, noise_cancellation, silero
from livekit.plugins import elevenlabs
from livekit.plugins.turn_detector.multilingual import MultilingualModel
from conversation_logger import logger as conversation_logger
from mcube_integration.env_load import load_agent_runtime_dotenv
logger = logging.getLogger("agent")
# Single-thread executor for conversation logging so file I/O never blocks the voice pipeline.
_log_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="conv_log")
load_agent_runtime_dotenv()
DEFAULT_INSTRUCTIONS = """You are a helpful voice AI for real-time voice. Reply in short phrases. Start with a very short phrase (1-5 words) when possible, e.g. "Sure," "One moment," "Let me think," then continue briefly. Keep each turn under 15 words when possible. No formatting, emojis, or symbols. Be friendly. Speed matters: quick short replies beat long ones."""
class Assistant(Agent):
def __init__(self, instructions: str | None = None, first_message: str | None = None) -> None:
super().__init__(
instructions=instructions or DEFAULT_INSTRUCTIONS,
)
self._first_message = (first_message or "").strip()
# Phase 1.6: TTS cache for fixed phrases to reduce latency on repeated phrases
self._tts_cache: dict[str, bytes] = {}
# Phase 2.6: Pronunciation dictionary for Indian names, places, and technical terms
self._pronunciation_dict = {
# Indian names and places
"Mumbai": "Mumbai",
"Chennai": "Chennai",
"Bangalore": "Bangalore",
"Kolkata": "Kolkata",
"Delhi": "Delhi",
"Hyderabad": "Hyderabad",
"Pune": "Pune",
"Ahmedabad": "Ahmedabad",
}
def _apply_pronunciation(self, text: str) -> str:
"""Apply pronunciation customization using SSML phonemes."""
result = text
for word, phoneme in self._pronunciation_dict.items():
# Case-insensitive replacement
import re
result = re.sub(f"\\b{word}\\b", phoneme, result, flags=re.IGNORECASE)
return result
async def on_enter(self) -> None:
"""If a first message is set, the bot says it when the session starts; otherwise waits for user."""
if self._first_message:
# Apply pronunciation customization to first message
enhanced_message = self._apply_pronunciation(self._first_message)
self.session.say(enhanced_message)
logger.info("First message spoken: %s", self._first_message[:80])
# To add tools, use the @function_tool decorator.
# Here's an example that adds a simple weather tool.
# You also have to add `from livekit.agents import function_tool, RunContext` to the top of this file
# @function_tool
# async def lookup_weather(self, context: RunContext, location: str):
# """Use this tool to look up current weather information in the given location.
#
# If the location is not supported by the weather service, the tool will indicate this. You must tell the user the location's weather is unavailable.
#
# Args:
# location: The location to look up weather information for (e.g. city name)
# """
#
# logger.info(f"Looking up weather for {location}")
#
# return "sunny with a temperature of 70 degrees."
server = AgentServer()
logger.info("=" * 60)
logger.info("AGENT SERVER INITIALIZED")
logger.info("Agent is ready to accept connections")
logger.info("=" * 60)
def _warm_elevenlabs_tts():
"""Warm ElevenLabs TTS so first 'Hello' response is faster (reduces cold start)."""
key = os.getenv("ELEVENLABS_API_KEY") or os.getenv("ELEVEN_API_KEY")
if not key:
return
try:
# Use environment variables for dynamic model/voice configuration (Phase 1.3 fix)
voice_id = os.getenv("MCUBE_TTS_VOICE_ID") or os.getenv("ELEVENLABS_TTS_VOICE_ID") or "EXAVITQu4vr4xnSDxMaL"
model = os.getenv("MCUBE_TTS_MODEL") or os.getenv("ELEVENLABS_TTS_MODEL") or "eleven_turbo_v2_5"
tts = elevenlabs.TTS(
voice_id=voice_id,
model=model,
api_key=key,
streaming_latency=0,
)
if hasattr(tts, "prewarm"):
tts.prewarm()
logger.info("ElevenLabs TTS prewarm done (model=%s voice=%s)", model, voice_id)
except Exception as e:
logger.debug("ElevenLabs TTS prewarm failed (non-fatal): %s", e)
def _warm_cartesia_tts():
"""Warm Cartesia TTS so first 'Hello' response is faster (reduces cold start)."""
key = os.getenv("CARTESIA_API_KEY")
if not key:
return
try:
# Use environment variables for dynamic model/voice configuration (Phase 1.3 fix)
model = os.getenv("MCUBE_TTS_MODEL") or os.getenv("CARTESIA_TTS_MODEL") or "sonic-3"
voice = os.getenv("MCUBE_TTS_VOICE_ID") or os.getenv("CARTESIA_TTS_VOICE_ID") or "95d51f79-c397-46f9-a48a-23763d3eaa2d"
tts = cartesia.TTS(
model=model,
voice=voice,
api_key=key,
encoding="pcm_s16le",
sample_rate=24000,
)
if hasattr(tts, "prewarm"):
tts.prewarm()
logger.info("Cartesia TTS prewarm done (model=%s voice=%s)", model, voice)
except Exception as e:
logger.debug("Cartesia TTS prewarm failed (non-fatal): %s", e)
def prewarm(proc: JobProcess):
logger.info("Prewarming agent process...")
proc.userdata["vad"] = silero.VAD.load()
# Phase 1.3: Enable TTS prewarming with timeout to avoid blocking join window
# Prewarm TTS in background with 5s timeout to balance cold start reduction vs join window
import asyncio
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
def _prewarm_tts_with_timeout():
try:
with ThreadPoolExecutor(max_workers=2) as executor:
# Submit both prewarm tasks
elevenlabs_future = executor.submit(_warm_elevenlabs_tts)
cartesia_future = executor.submit(_warm_cartesia_tts)
# Wait for both with 5s timeout total
try:
elevenlabs_future.result(timeout=5)
except FuturesTimeoutError:
logger.warning("ElevenLabs TTS prewarm timed out (non-fatal)")
try:
cartesia_future.result(timeout=5)
except FuturesTimeoutError:
logger.warning("Cartesia TTS prewarm timed out (non-fatal)")
except Exception as e:
logger.warning("TTS prewarm encountered error (non-fatal): %s", e)
# Run prewarm in thread to not block the prewarm function
import threading
prewarm_thread = threading.Thread(target=_prewarm_tts_with_timeout, daemon=True)
prewarm_thread.start()
logger.info("Prewarm complete - VAD loaded, TTS prewarming in background")
server.setup_fnc = prewarm
# agent_name must match the name requested by the frontend (room_config.agents[0].agent_name)
@server.rtc_session(agent_name="default")
async def my_agent(ctx: JobContext):
# === STARTUP LATENCY: track where time goes from job received to "listening" ===
t_job_received = time.time()
logger.info("Job received: room=%s, job_id=%s", ctx.room.name, getattr(ctx.job, "id", "?"))
# Logging setup
# Add any other context you want in all log entries here
ctx.log_context_fields = {
"room": ctx.room.name,
}
# Extract custom configuration from multiple sources
# Priority: 1) Agent dispatch metadata, 2) Room metadata, 3) Participant metadata
metadata_dict = {}
# Try to get metadata from agent dispatch first (minimal logging to stay within 10s join window)
if hasattr(ctx.job, 'agent_metadata') and ctx.job.agent_metadata:
try:
metadata_dict = json.loads(ctx.job.agent_metadata)
logger.info(f"Using agent dispatch metadata: {metadata_dict}")
except json.JSONDecodeError:
logger.warning(f"Failed to parse agent metadata as JSON: {ctx.job.agent_metadata}")
# Fall back to room metadata if agent metadata is empty
if not metadata_dict and ctx.room.metadata:
try:
metadata_dict = json.loads(ctx.room.metadata)
logger.info(f"Using room metadata: {metadata_dict}")
except json.JSONDecodeError:
logger.warning(f"Failed to parse room metadata as JSON: {ctx.room.metadata}")
# Connect to the room as early as possible so LiveKit sees the agent within the 10s join window.
# Then we build the pipeline and start the voice session (RoomIO).
livekit_url = os.getenv("LIVEKIT_URL")
livekit_key = os.getenv("LIVEKIT_API_KEY")
livekit_secret = os.getenv("LIVEKIT_API_SECRET")
if not livekit_url or not livekit_key or not livekit_secret:
logger.error(
"Missing LiveKit env: LIVEKIT_URL, LIVEKIT_API_KEY, LIVEKIT_API_SECRET "
"(set in backend/agent_runtime/.env.local and run from backend/agent_runtime/)"
)
raise RuntimeError(
"Agent cannot join: missing LIVEKIT_URL, LIVEKIT_API_KEY, or LIVEKIT_API_SECRET. "
"Set them in backend/agent_runtime/.env.local and run from backend/agent_runtime/."
)
await ctx.connect()
t_after_connect = time.time()
logger.info(
"[LATENCY] Startup: ctx.connect(): %.0f ms (total since job: %.0f ms)",
(t_after_connect - t_job_received) * 1000,
(t_after_connect - t_job_received) * 1000,
)
# Re-read room metadata after connect; it may only be available once we're in the room.
# This fixes the case where Cartesia (or other config) is sent in the token but the agent
# previously saw empty metadata and defaulted to ElevenLabs.
if ctx.room.metadata:
try:
room_meta = json.loads(ctx.room.metadata)
if not room_meta:
pass
elif not metadata_dict:
metadata_dict = room_meta
logger.info("Using room metadata (after connect): %s", list(metadata_dict.keys()))
else:
# Merge in any missing keys so we don't lose e.g. agent_tts_provider from room
for k, v in room_meta.items():
if k not in metadata_dict or metadata_dict[k] in (None, ""):
metadata_dict[k] = v
logger.info(
"Merged room metadata after connect; agent_tts_provider=%s, agent_stt_provider=%s",
metadata_dict.get("agent_tts_provider"),
metadata_dict.get("agent_stt_provider"),
)
except json.JSONDecodeError:
logger.warning("Failed to parse room metadata (after connect): %s", ctx.room.metadata[:100])
custom_instructions = metadata_dict.get("agent_instructions") if metadata_dict else None
custom_first_message = (metadata_dict.get("agent_first_message") or "").strip() if metadata_dict else ""
# STT and TTS providers can be chosen independently: "elevenlabs" or "cartesia"
custom_stt_provider = (metadata_dict.get("agent_stt_provider") or metadata_dict.get("agent_tts_provider") or "elevenlabs") if metadata_dict else "elevenlabs"
custom_tts_provider = (metadata_dict.get("agent_tts_provider") or "elevenlabs") if metadata_dict else "elevenlabs"
# Default voice must exist on your ElevenLabs/Cartesia account (per TTS provider)
custom_voice = metadata_dict.get("agent_voice", "EXAVITQu4vr4xnSDxMaL") if metadata_dict else "EXAVITQu4vr4xnSDxMaL"
custom_language = metadata_dict.get("agent_language", "en") if metadata_dict else "en"
custom_llm_model = metadata_dict.get("agent_llm_model", "openai/gpt-4.1-mini") if metadata_dict else "openai/gpt-4.1-mini"
custom_stt_model = metadata_dict.get("agent_stt_model", "default") if metadata_dict else "default"
custom_stt_language_code = metadata_dict.get("agent_stt_language_code", "en") if metadata_dict else "en"
custom_tts_model = metadata_dict.get("agent_tts_model", "eleven_turbo_v2_5") if metadata_dict else "eleven_turbo_v2_5"
custom_tts_encoding = metadata_dict.get("agent_tts_encoding") if metadata_dict else None
# Track message timing for logging
user_message_start = None
bot_message_start = None
# Latency tracking (during conversation): user_commit -> LLM (conversation_item_added) -> TTS (speech_created).
# Logs show: "[LATENCY] LLM (user_commit -> agent_response): X ms" and "[LATENCY] TTS (agent_response -> first audio): Y ms"
# so you can see whether delay is from LLM or TTS. Startup latency is logged separately (job -> connect -> session.start).
t_user_commit = None
t_agent_response = None
t_agent_speak = None
# Set up voice AI pipeline: STT and TTS can use different providers (ElevenLabs or Cartesia)
elevenlabs_key = os.getenv("ELEVENLABS_API_KEY") or os.getenv("ELEVEN_API_KEY")
cartesia_key = os.getenv("CARTESIA_API_KEY")
logger.info(
"Providers: STT=%s, TTS=%s (from agent_stt_provider / agent_tts_provider)",
"CARTESIA" if custom_stt_provider == "cartesia" else "ELEVENLABS",
"CARTESIA" if custom_tts_provider == "cartesia" else "ELEVENLABS",
)
# --- STT: build from STT provider only ---
if custom_stt_provider == "cartesia":
if not cartesia_key:
logger.error("CARTESIA_API_KEY not set – required for Cartesia STT")
stt = cartesia.STT(
model="ink-whisper",
language=custom_stt_language_code,
api_key=cartesia_key,
)
logger.info("STT: Cartesia model=ink-whisper language=%s", custom_stt_language_code)
else:
if not elevenlabs_key:
logger.error("ELEVENLABS_API_KEY (or ELEVEN_API_KEY) not set – required for ElevenLabs STT")
# Phase 4.1: Use scribe_v2_realtime for multilingual support including Indian languages
stt_kwargs = {
"model_id": "scribe_v2_realtime",
"language_code": custom_stt_language_code,
}
if elevenlabs_key:
stt_kwargs["api_key"] = elevenlabs_key
stt = elevenlabs.STT(**stt_kwargs)
logger.info("STT: ElevenLabs model=scribe_v2_realtime language=%s", custom_stt_language_code)
# --- TTS: build from TTS provider only ---
if custom_tts_provider == "cartesia":
if not cartesia_key:
logger.error("CARTESIA_API_KEY not set – required for Cartesia TTS")
# Phase 1.5: Add streaming_latency=0 for Cartesia to match ElevenLabs low-latency configuration
tts = cartesia.TTS(
model=custom_tts_model or "sonic-2",
voice=custom_voice,
language=custom_language,
api_key=cartesia_key,
streaming_latency=0,
)
logger.info("TTS: Cartesia model=%s voice=%s language=%s", custom_tts_model or "sonic-3", custom_voice, custom_language)
else:
if not elevenlabs_key:
logger.error("ELEVENLABS_API_KEY (or ELEVEN_API_KEY) not set – required for ElevenLabs TTS")
# Phase 4.3: Add language parameter for Indian language support
tts_kwargs = {
"voice_id": custom_voice,
"model": custom_tts_model,
"language": custom_language,
}
if elevenlabs_key:
tts_kwargs["api_key"] = elevenlabs_key
if custom_tts_encoding:
tts_kwargs["encoding"] = custom_tts_encoding
tts_kwargs.setdefault("streaming_latency", 0)
tts = elevenlabs.TTS(**tts_kwargs)
logger.info("TTS: ElevenLabs model=%s voice=%s language=%s", custom_tts_model, custom_voice, custom_language)
session = AgentSession(
stt=stt,
llm=inference.LLM(model=custom_llm_model),
tts=tts,
turn_detection=MultilingualModel(),
vad=ctx.proc.userdata["vad"],
preemptive_generation=True,
# Phase 1.4: Reduced endpointing delays for lower latency
min_endpointing_delay=0.05,
max_endpointing_delay=0.2,
# Phase 3.2: Enable adaptive interruption handling to distinguish true interruptions from backchanneling
turn_handling={
"turn_detection": MultilingualModel(),
"interruption": {"mode": "adaptive"},
},
)
t_after_pipeline_setup = time.time()
logger.info(
"[LATENCY] Startup: pipeline setup (metadata + STT + TTS + LLM + AgentSession): %.0f ms",
(t_after_pipeline_setup - t_job_received) * 1000,
)
# Add event handlers to log conversations and latency (Python SDK event names)
def _ev(ev, key, default=None):
if ev is None:
return default
if hasattr(ev, key):
return getattr(ev, key)
if isinstance(ev, dict):
return ev.get(key, default)
return default
# user_input_transcribed: STT output (is_final=True = user turn committed)
@session.on("user_input_transcribed")
def on_user_input_transcribed(ev):
nonlocal user_message_start, t_user_commit
transcript = _ev(ev, "transcript") or ""
is_final = _ev(ev, "is_final") or False
logger.info(f"[EVENT] user_input_transcribed is_final={is_final} transcript={transcript!r}")
if is_final:
_log_executor.submit(conversation_logger.log_user_message, transcript, None)
t_user_commit = time.time()
if user_message_start is not None:
user_turn_ms = (t_user_commit - user_message_start) * 1000
logger.info(f"[LATENCY] User turn (speaking + STT final): {user_turn_ms:.0f} ms")
user_message_start = None
logger.info(f"[LATENCY] user_speech_committed at T+0 (reference for LLM/TTS timing)")
# conversation_item_added: when a message is committed (user or assistant)
@session.on("conversation_item_added")
def on_conversation_item_added(ev):
nonlocal t_user_commit, t_agent_response, bot_message_start
item = _ev(ev, "item")
if item is None:
return
role = _ev(item, "role") if hasattr(item, "role") else (item.get("role") if isinstance(item, dict) else None)
content = _ev(item, "content") if hasattr(item, "content") else (item.get("content", []) if isinstance(item, dict) else [])
if content is None:
content = []
text = ""
if content:
parts = content if isinstance(content, list) else [content]
for part in parts:
t = getattr(part, "text", None)
if t:
text += t
if not text:
text = str(content)[:200]
if role == "assistant":
t_agent_response = time.time()
llm_ms = (t_agent_response - t_user_commit) * 1000 if t_user_commit is not None else None
logger.info(f"[EVENT] conversation_item_added role=assistant text={text[:80]!r}...")
if llm_ms is not None:
logger.info(f"[LATENCY] LLM (user_commit -> agent_response): {llm_ms:.0f} ms")
# Run synchronously so the bot message exists before speech_created can run
# update_last_bot_latency; otherwise e2e/tts get applied to the wrong message.
conversation_logger.log_bot_message(text, None, llm_ms=llm_ms)
bot_message_start = t_agent_response
elif role == "user":
logger.info(f"[EVENT] conversation_item_added role=user text={text[:80]!r}...")
# speech_created: agent started producing TTS (first audio)
@session.on("speech_created")
def on_speech_created(ev):
nonlocal bot_message_start, t_user_commit, t_agent_response, t_agent_speak
t_agent_speak = time.time()
bot_message_start = t_agent_speak
source = _ev(ev, "source") or "?"
logger.info(f"[EVENT] speech_created source={source}")
e2e_ms = (t_agent_speak - t_user_commit) * 1000 if t_user_commit is not None else None
tts_ms = (t_agent_speak - t_agent_response) * 1000 if t_agent_response is not None else None
if e2e_ms is not None:
logger.info(f"[LATENCY] End-to-end (user_commit -> first audio): {e2e_ms:.0f} ms")
if tts_ms is not None:
logger.info(f"[LATENCY] TTS (agent_response -> first audio): {tts_ms:.0f} ms")
_log_executor.submit(
lambda: conversation_logger.update_last_bot_latency(tts_ms=tts_ms, e2e_ms=e2e_ms)
)
# agent_state_changed: e.g. idle -> speaking (backup for "agent started")
@session.on("agent_state_changed")
def on_agent_state_changed(ev):
old_s = _ev(ev, "old_state")
new_s = _ev(ev, "new_state")
logger.info(f"[EVENT] agent_state_changed {old_s} -> {new_s}")
if new_s == "speaking" and old_s != "speaking":
nonlocal bot_message_start
bot_message_start = time.time()
# user_state_changed: e.g. when user starts speaking (for user_message_start)
@session.on("user_state_changed")
def on_user_state_changed(ev):
nonlocal user_message_start
state = _ev(ev, "new_state") or _ev(ev, "state")
logger.info(f"[EVENT] user_state_changed state={state}")
if state == "speaking":
user_message_start = time.time()
# To use a realtime model instead of a voice pipeline, use the following session setup instead.
# (Note: This is for the OpenAI Realtime API. For other providers, see https://docs.livekit.io/agents/models/realtime/))
# 1. Install livekit-agents[openai]
# 2. Set OPENAI_API_KEY in .env.local
# 3. Add `from livekit.plugins import openai` to the top of this file
# 4. Use the following session setup instead of the version above
# session = AgentSession(
# llm=openai.realtime.RealtimeModel(voice="marin")
# )
# # Add a virtual avatar to the session, if desired
# # For other providers, see https://docs.livekit.io/agents/models/avatar/
# avatar = hedra.AvatarSession(
# avatar_id="...", # See https://docs.livekit.io/agents/models/avatar/plugins/hedra
# )
# # Start the avatar and wait for it to join
# await avatar.start(session, room=ctx.room)
elapsed_before_start = (time.time() - t_job_received) * 1000
if elapsed_before_start > 8000:
logger.warning(
"Slow startup: %.0f ms since job before session.start(); LiveKit join window is 10s",
elapsed_before_start,
)
try:
await session.start(
agent=Assistant(instructions=custom_instructions, first_message=custom_first_message or None),
room=ctx.room,
room_options=room_io.RoomOptions(
audio_input=room_io.AudioInputOptions(
noise_cancellation=lambda params: noise_cancellation.BVCTelephony()
if params.participant.kind == rtc.ParticipantKind.PARTICIPANT_KIND_SIP
else noise_cancellation.BVC(),
),
),
)
except Exception as e:
logger.exception(
"Failed to start voice session (check API keys: ELEVENLABS/CARTESIA per provider, OPENAI_API_KEY / LiveKit Inference): %s",
e,
)
raise
t_after_session_start = time.time()
logger.info(
"[LATENCY] Startup: pipeline build + session.start() (since connect): %.0f ms (total since job: %.0f ms)",
(t_after_session_start - t_after_connect) * 1000,
(t_after_session_start - t_job_received) * 1000,
)
logger.info(
"[LATENCY] Startup TOTAL (job received -> agent listening): %.0f ms",
(t_after_session_start - t_job_received) * 1000,
)
# Now that we're in the room, start conversation logging and log config (no longer on critical path)
try:
conversation_logger.start_session(
room_name=ctx.room.name,
instructions=custom_instructions or DEFAULT_INSTRUCTIONS,
voice_id=custom_voice,
language=custom_language,
llm_model=custom_llm_model,
first_message=custom_first_message or None,
)
except Exception as e:
logger.error("Conversation logging start failed: %s", e)
logger.info("Agent is in the room and listening. User can say hello.")
# Set up cleanup handler to save logs when session ends
async def cleanup():
logger.info("Session ending, saving conversation log...")
log_file = conversation_logger.end_session()
logger.info(f"Conversation log saved to: {log_file}")
# Register cleanup handler
ctx.add_shutdown_callback(cleanup)
if __name__ == "__main__":
cli.run_app(server)