""" LiveKit AI Interview Agent Worker Runs as a separate process: python -m agent.interview_agent Registers with LiveKit as "interview-agent" and gets dispatched when a candidate joins an interview room. Ported from the standalone POC at ai-interview-system/agent/. """ import asyncio import json import logging import os import threading from config import ( LLM_MODEL, STT_LANGUAGE, STT_MODEL, TTS_MODEL, TTS_VOICE, VAD_MIN_ENDPOINTING_DELAY, ) from helpers import notify_backend_completed, save_session_report from livekit import agents, api, rtc from livekit.agents import ( Agent, AgentSession, JobContext, RoomInputOptions, WorkerOptions, cli, ) from livekit.plugins import openai, silero from prompts import INTERVIEW_AGENT_INSTRUCTIONS from s3_client import get_s3_bucket, get_s3_client logging.basicConfig(level=logging.INFO, format="%(levelname)s:%(name)s:%(message)s") logger = logging.getLogger(__name__) # Suppress noisy LiveKit logs livekit_logger = logging.getLogger("livekit") livekit_logger.propagate = False livekit_logger.addHandler(logging.StreamHandler()) livekit_logger.setLevel(logging.WARNING) # Timeout constants (in seconds) START_SIGNAL_TIMEOUT = 300 # 5 minutes to wait for candidate to click start CLEANUP_THREAD_TIMEOUT = 60 # Max time to wait for cleanup thread # Feature toggle for egress recording ENABLE_EGRESS_RECORDING = os.getenv("ENABLE_EGRESS_RECORDING", "true").lower() == "true" async def start_egress_recording(room_name: str, interview_id, livekit_url: str): """Start room composite egress recording to S3. Returns egress_id or None.""" if not ENABLE_EGRESS_RECORDING: logger.info("Egress recording disabled via ENABLE_EGRESS_RECORDING=false") return None s3_client = get_s3_client() s3_bucket = get_s3_bucket() if not s3_client or not s3_bucket: logger.warning("S3 not configured — skipping egress recording") return None aws_access_key = os.getenv("AWS_ACCESS_KEY_ID") aws_secret_key = os.getenv("AWS_SECRET_ACCESS_KEY") aws_s3_region = os.getenv("AWS_S3_REGION", "us-east-1") try: lkapi = api.LiveKitAPI(livekit_url) s3_upload = api.S3Upload( access_key=aws_access_key, secret=aws_secret_key, bucket=s3_bucket, region=aws_s3_region, ) egress_request = api.RoomCompositeEgressRequest( room_name=room_name, file_outputs=[ api.EncodedFileOutput( file_type=api.EncodedFileType.MP4, filepath=f"interview_{interview_id}.mp4", s3=s3_upload, ) ], ) egress_info = await lkapi.egress.start_room_composite_egress(egress_request) egress_id = egress_info.egress_id logger.info(f"Egress recording started: {egress_id}") await lkapi.aclose() return egress_id except api.TwirpError as e: logger.error(f"LiveKit API error starting egress: {e}") return None except Exception as e: logger.error(f"Failed to start egress recording: {e}", exc_info=True) return None def create_interview_agent(interview_metadata: dict) -> Agent: """Create an AI Interview Agent with personalized instructions.""" candidate_info = interview_metadata.get("candidate", {}) role_info = interview_metadata.get("role", {}) candidate_name = candidate_info.get("name", "Candidate") candidate_skills = ", ".join(candidate_info.get("skills", [])) candidate_experience = candidate_info.get("experience_years", 0) role_title = role_info.get("title", "Position") question_topics = role_info.get("question_topics", []) duration_minutes = interview_metadata.get("duration_minutes", 20) greeting_time = "30 seconds" if duration_minutes >= 20 else "15 seconds" main_time = f"{duration_minutes - 2} minutes" if duration_minutes > 5 else f"{duration_minutes - 1} minutes" closing_time = "2 minutes" if duration_minutes >= 20 else "1 minute" # Format numbered list of question topics question_topics_list = "\n".join(f" {i + 1}. {topic}" for i, topic in enumerate(question_topics)) instructions = INTERVIEW_AGENT_INSTRUCTIONS.format( role_title=role_title, candidate_name=candidate_name, candidate_experience=candidate_experience, candidate_skills=candidate_skills, duration_minutes=duration_minutes, greeting_time=greeting_time, main_time=main_time, closing_time=closing_time, question_topics_list=question_topics_list, ) return Agent(instructions=instructions) def parse_job_metadata(ctx: JobContext) -> dict: """Parse and validate job metadata. Returns metadata dict or empty dict on failure.""" try: metadata = json.loads(ctx.job.metadata) if ctx.job.metadata else {} candidate = metadata.get("candidate", {}).get("name") role = metadata.get("role", {}).get("title") logger.info(f"Interview metadata: candidate={candidate}, role={role}") return metadata except (json.JSONDecodeError, TypeError) as e: logger.error(f"Failed to parse job metadata: {e}") return {} def setup_room_events( ctx: JobContext, start_event: asyncio.Event, disconnect_event: asyncio.Event, candidate_name: str ): """Register room event listeners for data_received and disconnected.""" @ctx.room.on("data_received") def on_data(data: rtc.DataPacket): try: message = json.loads(data.data.decode("utf-8")) if message.get("action") == "START": logger.info(f"{candidate_name} is ready — starting interview") start_event.set() except (json.JSONDecodeError, UnicodeDecodeError) as e: logger.error(f"Error decoding data packet: {e}") except Exception as e: logger.error(f"Error processing data packet: {e}") def on_disconnect(reason): logger.info(f"Room disconnected: {reason}") disconnect_event.set() start_event.set() # Unblock START wait if still waiting ctx.room.on("disconnected", on_disconnect) async def wait_for_start_signal(start_event: asyncio.Event, candidate_name: str): """Wait for the candidate to send a START signal, with a 5-minute timeout.""" logger.info(f"Waiting for {candidate_name} to start the interview...") try: await asyncio.wait_for(start_event.wait(), timeout=START_SIGNAL_TIMEOUT) except asyncio.TimeoutError: logger.warning("Candidate didn't click start within 5 minutes — starting anyway") WRAP_UP_RATIO = 0.85 # Nudge agent to close at 85% of total time async def run_interview( session: AgentSession, disconnect_event: asyncio.Event, candidate_name: str, duration_seconds: int ): """Run the interview: generate greeting, nudge to wrap up near end, hard stop at timeout.""" if disconnect_event.is_set(): logger.info("Room already disconnected before interview started — skipping to cleanup") return wrap_up_at = int(duration_seconds * WRAP_UP_RATIO) remaining_after_wrap_up = duration_seconds - wrap_up_at interview_start = asyncio.get_event_loop().time() logger.info( f"Interview timing: total={duration_seconds}s, " f"wrap_up_at={wrap_up_at}s, closing_buffer={remaining_after_wrap_up}s" ) await session.generate_reply( instructions=( f"Start the interview now. Greet {candidate_name} warmly " "and ask the first question from topic 1." ) ) # Subtract time already spent on greeting from the wrap-up timeout elapsed = asyncio.get_event_loop().time() - interview_start phase1_timeout = max(0, wrap_up_at - elapsed) logger.info(f"Greeting took {elapsed:.1f}s, phase 1 timeout adjusted to {phase1_timeout:.1f}s") # Phase 1: Main interview — wait for disconnect or wrap-up time try: await asyncio.wait_for(disconnect_event.wait(), timeout=phase1_timeout) logger.info("Interview ended by user") return except asyncio.TimeoutError: logger.info("Wrap-up time reached — asking agent to close the interview") # Phase 2: Nudge agent to wrap up naturally await session.generate_reply( instructions=( "TIME IS UP. You must wrap up the interview NOW. " f"Thank {candidate_name} for their time, tell them the team will be in touch " "within 3-5 business days, and say goodbye. Do NOT ask any more questions." ) ) # Phase 3: Hard cutoff — allow remaining time for the goodbye try: await asyncio.wait_for(disconnect_event.wait(), timeout=remaining_after_wrap_up) logger.info("Interview ended by user during closing") except asyncio.TimeoutError: logger.info(f"Interview hard timeout reached ({duration_seconds // 60} minutes)") def run_cleanup(ctx: JobContext, metadata: dict, interview_id, egress_id: str | None): """Run post-interview cleanup synchronously in a thread to survive process shutdown.""" logger.info(f"[CLEANUP] Starting cleanup for interview_id={interview_id}") def _cleanup(): s3_key = save_session_report(ctx, metadata, egress_id=egress_id) logger.info(f"[CLEANUP] save_session_report returned s3_key={s3_key}") notify_backend_completed(interview_id, s3_key) logger.info(f"[CLEANUP] Cleanup complete for interview_id={interview_id}") cleanup_thread = threading.Thread(target=_cleanup, daemon=False) cleanup_thread.start() cleanup_thread.join(timeout=CLEANUP_THREAD_TIMEOUT) async def entrypoint(ctx: JobContext): """Main entrypoint for the interview agent worker.""" logger.info(f"Starting interview agent job: {ctx.job.id}") openai_key = os.getenv("OPENAI_API_KEY") livekit_url = os.getenv("LIVEKIT_URL") if not openai_key: logger.error("Missing OPENAI_API_KEY — agent cannot initialize") return await ctx.connect() metadata = parse_job_metadata(ctx) if not metadata.get("candidate") or not metadata.get("role"): logger.error("Missing required metadata: candidate and role information") return interview_id = metadata.get("interview_id", "unknown") candidate_name = metadata.get("candidate", {}).get("name", "Candidate") duration_minutes = metadata.get("duration_minutes", 20) duration_seconds = duration_minutes * 60 # Start egress recording egress_id = await start_egress_recording(ctx.room.name, interview_id, livekit_url) # Create agent and session interview_agent = create_interview_agent(interview_metadata=metadata) logger.info(f"[AGENT] Starting session: llm={LLM_MODEL} stt={STT_MODEL} tts={TTS_MODEL}") session = AgentSession( vad=silero.VAD.load(), stt=openai.STT(model=STT_MODEL, language=STT_LANGUAGE), llm=openai.LLM(model=LLM_MODEL), tts=openai.TTS(model=TTS_MODEL, voice=TTS_VOICE), min_endpointing_delay=VAD_MIN_ENDPOINTING_DELAY, ) @session.on("user_speech_committed") def on_user_speech(msg: agents.llm.ChatMessage): if msg.content: logger.info(f"[USER] {msg.content}") @session.on("agent_speech_committed") def on_agent_speech(msg: agents.llm.ChatMessage): if msg.content: logger.info(f"[AGENT] {msg.content}") # Register room event listeners BEFORE session.start() to avoid race conditions start_event = asyncio.Event() disconnect_event = asyncio.Event() setup_room_events(ctx, start_event, disconnect_event, candidate_name) await session.start( agent=interview_agent, room=ctx.room, room_input_options=RoomInputOptions(close_on_disconnect=False), ) await wait_for_start_signal(start_event, candidate_name) await run_interview(session, disconnect_event, candidate_name, duration_seconds) # Disconnect room if still connected (timeout path) try: await ctx.room.disconnect() except Exception: pass run_cleanup(ctx, metadata, interview_id, egress_id) if __name__ == "__main__": cli.run_app( WorkerOptions( entrypoint_fnc=entrypoint, agent_name=os.getenv("LIVEKIT_AGENT_NAME", "interview-agent"), num_idle_processes=1, ) )