RoomIO

AgentSession

TranscriptSynchronizer

DataStreamIO

AgentSession

TranscriptSynchronizer

RoomIO

RoomIO

AgentSession

<http://api.simli.ai|api.simli.ai>

host <http://api.simli.ai|api.simli.ai>

curl -v <https://api.simli.ai>

8.8.8.8

navigence-85r0jkqk.livekit.cloud

*.livekit.cloud

*.livekit.cloud

*.turn.livekit.cloud

*.turn.livekit.cloud

*.turn.livekit.cloud

host stun:ip-161-118-222-18.host.livekit.cloud

host turn:ip-161-118-222-18.host.livekit.cloud

iceTransportPolicy: 'relay'

RoomIO

AgentSession

TranscriptSynchronizer

DataStreamIO

AgentSession

TranscriptSynchronizer

RoomIO

RoomIO

AgentSession

from dotenv import load_dotenv
import os,aiohttp,json
from livekit import agents
from livekit.agents import WorkerOptions, WorkerType,cli,AutoSubscribe,JobProcess
from livekit.agents import AgentSession, Agent, RoomInputOptions,metrics
from livekit.plugins import (
    openai,
    noise_cancellation,
    simli,
    elevenlabs,
    silero
)
from jinja2 import Template
from livekit.agents.voice import MetricsCollectedEvent
from livekit.agents.metrics import LLMMetrics,STTMetrics,TTSMetrics
from app.core.config import settings


prompt_template=f"""You are an AI interviewer conducting a professional interview. Your interface with users will be voice, so you must avoid usage of unpronounceable punctuation and should use short and concise responses. Deliver the response in plain text without any Markdown or formatting.
 
 You are conducting an interview for the position of {{job_title}} at {{company_name}}. This is a {{interview_type}} interview.
 
 Your role is to:
 1. Conduct a professional, engaging interview
 2. Ask relevant questions based on the position and company
 3. Listen actively to responses and ask follow-up questions
 4. Maintain a friendly but professional tone
 5. Guide the conversation naturally
 6. Provide feedback when appropriate
 7. Keep the interview flowing smoothly
 
 Interview guidelines:
 - Start with a warm welcome and brief introduction
 - Ask 5-7 main questions plus follow-ups
 - Allow the candidate to ask questions at the end
 - Provide constructive feedback
 - Keep responses concise and natural for voice interaction
 - Be encouraging and supportive
 
 Remember: You are speaking, not writing. Use natural speech patterns and avoid complex punctuation or formatting."""

def formate_template(data):
    data=json.loads(data)
    template = Template(prompt_template)
    rendered_prompt = template.render(**data)
    return rendered_prompt

def validate_configuration():
    """Validate that all required configuration is present"""
    missing_configs = []
    
    if not settings.OPENAI_API_KEY:
        missing_configs.append("OPENAI_API_KEY")
    
    if not settings.ELEVENLABS_API_KEY:
        missing_configs.append("ELEVENLABS_API_KEY")
        print("WARNING: ELEVENLABS_API_KEY not found. Will use OpenAI TTS as fallback.")
    
    if not settings.SIMLI_API_KEY:
        missing_configs.append("SIMLI_API_KEY")
        print("WARNING: SIMLI_API_KEY not found. Avatar functionality will be disabled.")
    
    if not settings.LIVEKIT_URL or not settings.LIVEKIT_API_KEY or not settings.LIVEKIT_API_SECRET:
        missing_configs.extend(["LIVEKIT_URL", "LIVEKIT_API_KEY", "LIVEKIT_API_SECRET"])
    
    if missing_configs:
        print(f"Missing required configuration: {', '.join(missing_configs)}")
        print("Please check your .env file and ensure all required API keys are set.")
        return False
    
    print("Configuration validation passed.")
    return True

class Assistant(Agent):
    def __init__(self,ctx) -> None:
        super().__init__(instructions= formate_template(ctx.room.metadata))

def prewarm_process(proc: JobProcess):
    proc.userdata["vad"] = silero.VAD.load(min_silence_duration=0.3)
    
async def entrypoint(ctx: agents.JobContext):
    # Validate configuration first
    validate_configuration()
    
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
    print(f"CTX.Room.Name: {ctx.room.name}")

    metadata=json.loads(ctx.room.metadata)
    # Prefer explicit voice_id from metadata; fallback to gender defaults
    voice_id = metadata.get("voice_id")
    if not voice_id:
        if metadata.get('avatar_type', '').lower()=="female":
            voice_id="21m00Tcm4TlvDq8ikWAM"
        elif metadata.get('avatar_type', '').lower()=="male":
            voice_id="GBv7mTt0atIp3Br8iCZE"
        else:
            voice_id = "21m00Tcm4TlvDq8ikWAM"

    print("Agent Session")
    total_llm_tokens = 0
    total_stt_duration = 0.0
    total_tts_duration = 0.0

    def _on_metrics_collected(event: MetricsCollectedEvent):
        # Use 'nonlocal' to modify variables from the outer scope
        nonlocal total_llm_tokens, total_stt_duration, total_tts_duration
        
        metrics = event.metrics

        if isinstance(metrics, LLMMetrics):
            # This is the missing block from your previous code
            print("LLM METRICS RECEIVED:", metrics.model_dump())
            tokens_used = metrics.completion_tokens
            total_llm_tokens += tokens_used
            print(f"Tokens Used in this turn: {tokens_used}")

        elif isinstance(metrics, STTMetrics):
            total_stt_duration += metrics.audio_duration

        elif isinstance(metrics, TTSMetrics):
            total_tts_duration += metrics.audio_duration
    async def print_final_summary():
        print("\n--- FINAL USAGE SUMMARY ---")
        print(f"Total LLM Tokens Used: {total_llm_tokens}")
        print(f"Total STT Audio Duration: {total_stt_duration:.2f} seconds")
        print(f"Total TTS Audio Duration: {total_tts_duration:.2f} seconds")
        print("---------------------------\n")

    # Use only VAD to avoid turn-detector model requirement
    # Initialize TTS with better error handling
    tts_instance = None
    try:
        if settings.ELEVENLABS_API_KEY:
            print(f"Initializing ElevenLabs TTS with voice_id: {voice_id}")
            tts_instance = elevenlabs.TTS(
                api_key=settings.ELEVENLABS_API_KEY,
                voice_id=voice_id,
                # Add connection timeout and retry settings
                timeout=30.0,
                retry_attempts=3
            )
        else:
            print("ELEVENLABS_API_KEY not found, using OpenAI TTS as fallback")
            tts_instance = openai.TTS(api_key=settings.OPENAI_API_KEY)
    except Exception as e:
        print(f"Failed to initialize ElevenLabs TTS: {e}")
        print("Falling back to OpenAI TTS")
        tts_instance = openai.TTS(api_key=settings.OPENAI_API_KEY)
    
    session = AgentSession(
        turn_detection=None,
        vad=ctx.proc.userdata["vad"],
        stt=openai.STT(api_key=settings.OPENAI_API_KEY),
        tts=tts_instance,
        allow_interruptions=True,
        llm=openai.LLM(
            model="gpt-4o-mini",  # or gpt-4 / gpt-3.5-turbo depending on your needs
            api_key=settings.OPENAI_API_KEY
        ),
        max_endpointing_delay=0.3
    )
    async with aiohttp.ClientSession() as http_session:
        enable_avatar = os.environ.get("SIMLI_ENABLE_AVATAR", "0") in ["1", "true", "True"]
        avatar = None
        if enable_avatar:
            avatar_face_id = metadata.get("avatar_face_id") or metadata.get("face_id") or "cace3ef7-a4c4-425d-a8cf-a5358eb0c427"
            avatar = simli.AvatarSession(
                simli_config=simli.SimliConfig(
                    api_key=settings.SIMLI_API_KEY,
                    face_id=avatar_face_id,
                ),
            )


    try:
        if avatar is not None:
            print("Avatar Starting...")
            await avatar.start(session, room=ctx.room)
        else:
            print("SIMLI_ENABLE_AVATAR disabled. Running audio-only agent.")
    except Exception as e:
        print(f"Failed to start avatar: {e}")
        if avatar is not None:
            print("Trying again with hardcoded conversation_id (NOT recommended for production)")
            try:
                avatar.conversation_id = metadata.get("session_id")
                await avatar.start(session, room=ctx.room)
            except Exception as e2:
                print(f"Avatar failed again: {e2}")
                print("Continuing with audio-only mode")
                avatar = None

    print("Agent Start")
    print(f"Room:{ctx.room}")

    print("ctx connect")
    await ctx.connect()
    session.on("metrics_collected", _on_metrics_collected)
    ctx.add_shutdown_callback(print_final_summary)

    await session.start(
        room=ctx.room,
        agent=Assistant(ctx),  
        room_input_options=RoomInputOptions(
            noise_cancellation=noise_cancellation.BVC(),
            video_enabled=True,
            audio_enabled=True   
            
        ),
    )
    print("generate Reply...")
    greeting = f"""Hello! Welcome to your interview for the {metadata['job_title']} position at {metadata['company_name']}. I'm your AI interviewer, and I'm excited to learn more about you today. Let's start with a brief introduction. Can you tell me about yourself and what interests you about this role?"""
    try:
        await session.say(greeting)
    except Exception as e:
        print(f"Failed to say greeting: {e}")
        # Continue without greeting if TTS fails

if __name__ == "__main__":
    # agent_name ="Test"
    agent_name =os.environ["AGENT_NAME"]
    print(f"Agent_Name in creation: {agent_name}, Type: {type(agent_name)}")
    opts = WorkerOptions(
        entrypoint_fnc=entrypoint,
        prewarm_fnc=prewarm_process,
        worker_type=WorkerType.ROOM,
        agent_name=agent_name,
        initialize_process_timeout=90
    )    
    
    cli.run_app(opts)

preemptive_generation

AgentSession

prewarm