app.py

import os
import json
from dotenv import load_dotenv
import uuid
from datetime import datetime
from groq import Groq

from constants import SYSTEM_PROMPT, INTERVIEW_INSTRUCTIONS, MOCK_INTERVIEW_PROMPT,EXPECTED_OUTPUT
from constants import JOB_MATCHING_INSTRUCTIONS,JOB_MATCHING_PROMPT,JOB_ROLES

from phi.agent import Agent
from phi.model.google import Gemini
from phi.tools.duckduckgo import DuckDuckGo

from elevenlabs.client import ElevenLabs
from elevenlabs import VoiceSettings

from gradio import (
    Blocks, Chatbot, Row, Column, Radio, Dropdown,
    Button, Audio, Textbox, State, HTML
)

load_dotenv()

# Initialize clients
eleven_client = ElevenLabs(api_key=os.getenv("ELEVEN_API_KEY"))
groq_client = Groq(api_key=os.getenv("GROQ_API_KEY"))

def get_current_datetime() -> str:
    return json.dumps({
        "current_datetime": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    })

def create_agent(mode):
    base_config = {
        "model": Gemini(id="gemini-2.0-flash-exp", api_key=os.getenv("GOOGLE_API_KEY")),
        "show_tool_calls": True,
        "tools": [DuckDuckGo(fixed_max_results=10), get_current_datetime],
        "add_history_to_messages": True,
    }
    
    if mode == "Interview Guide":
        return Agent(
            **base_config,
            system_prompt=SYSTEM_PROMPT,
            instructions=INTERVIEW_INSTRUCTIONS,
            num_history_responses=6,
            expected_output=EXPECTED_OUTPUT
        )
    elif mode == "Job Matching":
        return Agent(
            **base_config,
            system_prompt=JOB_MATCHING_PROMPT,
            instructions=JOB_MATCHING_INSTRUCTIONS,
        )
    else:
        return Agent(
            **base_config,
            system_prompt=MOCK_INTERVIEW_PROMPT,
            instructions=["Conduct a technical mock interview.", "Ask one question at a time.", "Evaluate the candidate's responses in simple concise words"],
            num_history_responses=3,
        )

def text_to_speech_file(text: str, client: ElevenLabs) -> str:
    """
    Converts text to speech using ElevenLabs API and saves as MP3.
    
    Args:
        text (str): Text to convert to speech
        client (ElevenLabs): Initialized ElevenLabs client
    
    Returns:
        str: Path to saved audio file
    """
    try:
        response = client.text_to_speech.convert(
            voice_id="pNInz6obpgDQGcFmaJgB",  # Adam voice
            optimize_streaming_latency="0",
            output_format="mp3_22050_32",
            text=text,
            model_id="eleven_turbo_v2",
            voice_settings=VoiceSettings(
                stability=0.0,
                similarity_boost=1.0,
                style=0.0,
                use_speaker_boost=True,
            ),
        )
        
        save_file_path = f"{uuid.uuid4()}.mp3"
        with open(save_file_path, "wb") as f:
            for chunk in response:
                if chunk:
                    f.write(chunk)
        
        return save_file_path
    except Exception as e:
        raise Exception(f"Text-to-speech conversion failed: {str(e)}")

def handle_mock_interview(audio_path, history, agent_state, voice_choice):
    """Handle audio for Mock Interview mode"""
    if not audio_path:
        return history, agent_state, None, ""
    
    try:
        # Transcribe audio
        with open(audio_path, "rb") as audio_file:
            transcription = groq_client.audio.transcriptions.create(
                file=("recording.wav", audio_file.read(), "audio/wav"),
                model="whisper-large-v3-turbo",
                response_format="text"
            )
        
        # Initialize agent if needed
        if "agent" not in agent_state or agent_state.get("mode") != "Mock Interview":
            agent_state["agent"] = create_agent("Mock Interview")
            agent_state["mode"] = "Mock Interview"
        
        # Get agent response
        agent = agent_state["agent"]
        response = agent.run(transcription).content
        
        try:
            # Generate audio using the improved text-to-speech function
            audio_output = text_to_speech_file(response, eleven_client)
            
            # Update history
            history.append({"role": "user", "content": f"[Audio]: {transcription}"})
            history.append({"role": "assistant", "content": response})
            
            return history, agent_state, audio_output, ""
            
        except Exception as audio_error:
            return history, agent_state, None, f"Audio generation error: {str(audio_error)}"
            
    except Exception as e:
        return history, agent_state, None, f"Error: {str(e)}"

def handle_text_input(message, history, mode, agent_state):
    """Handle text input for Interview Guide mode"""
    if not message.strip():
        return history, agent_state, "", ""
    
    if "agent" not in agent_state or agent_state.get("mode") != mode:
        agent_state["agent"] = create_agent(mode)
        agent_state["mode"] = mode
    
    agent = agent_state["agent"]
    history = history + [{"role": "user", "content": message}]
    
    try:
        # Using direct response instead of streaming
        response = agent.run(message).content
        history.append({"role": "assistant", "content": response})
        return history, agent_state, "", ""
    except Exception as e:
        return history, agent_state, "", f"Error: {str(e)}"

def handle_job_matching(role, experience, location, history, agent_state):
    """Handle job matching mode inputs"""
    # Initialize history if None
    if history is None:
        history = []
    
    # Initialize agent_state if None
    if agent_state is None:
        agent_state = {}
        
    if not all([role, experience, location]):
        return history, agent_state, "Please fill in all fields"
    
    # Initialize agent if needed
    if "agent" not in agent_state or agent_state.get("mode") != "Job Matching":
        agent_state["agent"] = create_agent("Job Matching")
        agent_state["mode"] = "Job Matching"
    
    query = f"""Find relevant jobs for:
    Role: {role}
    Experience: {experience}
    Location: {location}
    
    Please search for current job listings and provide details including:
    1. Company name
    2. Job title
    3. Location
    4. Key requirements
    5. Application link or process
    """
    
    try:
        agent = agent_state["agent"]
        history = history + [{"role": "user", "content": query}]
        response = agent.run(query).content
        history.append({"role": "assistant", "content": response})
        return history, agent_state, ""
    except Exception as e:
        return history, agent_state, f"Error: {str(e)}"
    
def clear_chat():
    return [], {}, None, ""

with Blocks(title="AI Interview Assistant") as demo:
    # State
    chat_history = State([])
    agent_state = State({})
    
    mode = Radio(
        choices=["Interview Guide", "Mock Interview", "Job Matching"],
        label="Mode",
        value="Interview Guide"
    )

    chatbot = Chatbot(label="Conversation", height=500, type="messages")
    error_msg = HTML()
    
    with Row():
        with Column(visible=True) as text_col:
            text_input = Textbox(
                label="Type your message",
                placeholder="Ask about interview preparation...",
                lines=3
            )
            submit_btn = Button("Send Message")
        
        # Audio input for Mock Interview
        with Column(visible=False) as audio_col:
            voice_select = Dropdown(
                choices=["Brian", "Rachel", "Sam"],
                value="Brian",
                label="Assistant Voice"
            )
            audio_input = Audio(
                sources=["microphone"],
                type="filepath",
                label="Record your answer"
            )
            audio_output = Audio(
                label="Assistant's Response",
                visible=True
            )
        
        with Column(visible=False) as job_col:
            role_select = Dropdown(
                choices=JOB_ROLES,
                label="Select Job Role",
                value="Software Engineer"
            )
            experience = Dropdown(
                choices=["0-2 years", "2-5 years", "5-8 years", "8+ years"],
                label="Experience Level",
                value="0-2 years"
            )
            location = Textbox(
                label="Preferred Location",
                placeholder="Enter city, state, or 'Remote'",
                lines=1
            )
            search_btn = Button("Search Jobs")
    
    # Clear button
    clear_btn = Button("Clear Chat")
    
    # Event handlers
    def update_mode(mode_value):
        return (
            Column(visible=mode_value == "Interview Guide"),
            Column(visible=mode_value == "Mock Interview"),
            Column(visible=mode_value == "Job Matching")
        )
    
    mode.change(
        update_mode,
        inputs=[mode],
        outputs=[text_col,audio_col,job_col]
    )
    
    submit_btn.click(
        handle_text_input,
        inputs=[text_input, chat_history, mode, agent_state],
        outputs=[chatbot, agent_state, text_input, error_msg]
    )
    
    audio_input.change(
        handle_mock_interview,
        inputs=[audio_input, chat_history, agent_state, voice_select],
        outputs=[chatbot, agent_state, audio_output, error_msg]
    )

    search_btn.click(
        handle_job_matching,
        inputs=[role_select, experience, location, chat_history, agent_state],
        outputs=[chatbot, agent_state, error_msg]
    )
    
    clear_btn.click(
        clear_chat,
        outputs=[chat_history, agent_state, audio_output, error_msg]
    )

if __name__ == "__main__":
    demo.launch()