""" 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 ( extract_transcript_entries, generate_livekit_report, notify_backend_completed, post_transcripts_to_backend, upload_session_report_to_s3, ) 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.getLogger().setLevel(logging.INFO) logger = logging.getLogger(__name__) # Suppress noisy LiveKit logs livekit_logger = logging.getLogger("livekit") livekit_logger.propagate = False livekit_logger.setLevel(logging.WARNING) # Timeout constants (in seconds) START_SIGNAL_TIMEOUT = 60 # seconds to wait for candidate's START signal 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" S3_RECORDINGS_FOLDER = "interview_recordings_dev" 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"{S3_RECORDINGS_FOLDER}/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 = candidate_info.get("skills", []) candidate_seniority = candidate_info.get("seniority_level") or "not specified" role_title = role_info.get("title", "Position") designation = role_info.get("designation") or role_title role_description = role_info.get("description", "") role_seniority = role_info.get("seniority_level") or "not specified" question_topics = role_info.get("question_topics", []) duration_minutes = interview_metadata.get("duration_minutes", 20) main_minutes = max(1, int(duration_minutes * WRAP_UP_RATIO) - 1) greeting_time = "30 seconds" if duration_minutes >= 20 else "15 seconds" num_topics = max(1, len(question_topics)) exchanges_per_topic = max(2, main_minutes // (num_topics * 2)) minutes_per_topic = max(2, main_minutes // num_topics) candidate_skills_lower = {s.lower() for s in candidate_skills} gap_topics = [t for t in question_topics if t.lower() not in candidate_skills_lower] overlap_topics = [t for t in question_topics if t.lower() in candidate_skills_lower] if gap_topics: overlap_str = ", ".join(overlap_topics) if overlap_topics else "none" gap_str = ", ".join(gap_topics) skill_gap_context = ( f"The candidate has listed {overlap_str} of the required topics. " f"They have NOT listed {gap_str} — probe these areas more carefully." ) else: skill_gap_context = ( "The candidate's listed skills cover all required topics. " "Go deeper — focus on edge cases, tradeoffs, and real-world application." ) if question_topics: question_topics_list = "\n".join(f" {i + 1}. {topic}" for i, topic in enumerate(question_topics)) else: question_topics_list = ( " (No specific topics provided — ask broad questions " "grounded in the role description and required skills.)" ) instructions = INTERVIEW_AGENT_INSTRUCTIONS.format( role_title=role_title, designation=designation, role_description=role_description, role_seniority=role_seniority, candidate_name=candidate_name, candidate_seniority=candidate_seniority, candidate_skills=", ".join(candidate_skills) if candidate_skills else "not specified", duration_minutes=duration_minutes, greeting_time=greeting_time, main_minutes=main_minutes, num_topics=num_topics, question_topics_list=question_topics_list, exchanges_per_topic=exchanges_per_topic, minutes_per_topic=minutes_per_topic, skill_gap_context=skill_gap_context, ) 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(f"START signal not received within {START_SIGNAL_TIMEOUT}s — 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" ) # Clear any VAD/STT turn buffered while waiting for the start signal # (ambient noise, mic sounds) so the greeting fires immediately. session.interrupt() await session.generate_reply(instructions="Begin the interview now with the greeting.") # 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=( f"The interview time is almost up. Tell {candidate_name} you have time for one final question, " "then ask it. After they answer, thank them warmly and let them know the team will be in touch " "within three to five business days. Do not ask any more questions after that." ) ) # 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( livekit_report: dict | None, transcript_entries: list[dict], metadata: dict, interview_id, egress_id: str | None, ): """Run post-interview cleanup in a thread to survive process shutdown.""" logger.info(f"[CLEANUP] Starting cleanup for interview_id={interview_id}") def _cleanup(): s3_key = upload_session_report_to_s3(livekit_report, metadata) notify_backend_completed(interview_id, s3_key) if transcript_entries: post_transcripts_to_backend(interview_id, transcript_entries) else: logger.warning("[CLEANUP] No transcript entries extracted — skipping transcript POST") 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) # Notify the frontend that the agent is ready to receive the START signal. # If the frontend sent START during agent setup (~7s window), it missed the # data_received listener. The frontend should resend START on AGENT_READY. try: await ctx.room.local_participant.publish_data( json.dumps({"action": "AGENT_READY"}).encode("utf-8"), reliable=True, ) logger.info("Sent AGENT_READY signal to room") except Exception as e: logger.warning(f"Failed to send AGENT_READY signal: {e}") 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) livekit_report = generate_livekit_report(ctx) transcript_entries = extract_transcript_entries(livekit_report or {}) logger.info(f"[CLEANUP] Extracted {len(transcript_entries)} transcript entries") # Disconnect room if still connected (timeout path) try: await ctx.room.disconnect() except Exception: pass run_cleanup(livekit_report, transcript_entries, 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, ) )