import gradio as gr import tempfile import os import json from io import BytesIO from collections import deque from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain.schema import HumanMessage, SystemMessage from langchain.chains import LLMChain from langchain.prompts import PromptTemplate from openai import OpenAI import time # Imports - Keep only what's actually used. I've organized them. from generatorgr import ( generate_and_save_questions as generate_questions_manager, update_max_questions, ) from generator import ( PROFESSIONS_FILE, TYPES_FILE, OUTPUT_FILE, load_json_data, generate_questions, # Keep if needed, but ensure it exists ) from splitgpt import ( generate_and_save_questions_from_pdf3, generate_questions_from_job_description, ) # ai_config.py is no longer directly imported, functions are redefined here to handle missing API key. # from ai_config import convert_text_to_speech # Redundant import, redefined below. from knowledge_retrieval import get_next_response, get_initial_question from prompt_instructions import get_interview_initial_message_hr from settings import language from utils import save_interview_history from tools import store_interview_report, read_questions_from_json load_dotenv() # Load .env variables class InterviewState: """Manages the state of the interview.""" def __init__(self): self.reset() def reset(self, voice="alloy"): self.question_count = 0 # Corrected history format: List of [user_msg, bot_msg] pairs. self.interview_history = [] self.selected_interviewer = voice self.interview_finished = False self.audio_enabled = True self.temp_audio_files = [] self.initial_audio_path = None self.interview_chain = None self.report_chain = None self.current_questions = [] self.history_limit = 5 # Limit the history (good for performance) def get_voice_setting(self): return self.selected_interviewer interview_state = InterviewState() def initialize_chains(): """Initializes the LangChain LLM chains.""" openai_api_key = os.getenv("OPENAI_API_KEY") if not openai_api_key: print("OpenAI API key not found. Chains will not be initialized.") interview_state.interview_chain = None # Set to None to indicate not initialized interview_state.report_chain = None return False # Indicate chains were not initialized try: llm = ChatOpenAI( openai_api_key=openai_api_key, model="gpt-4o", temperature=0.7, max_tokens=750 ) interview_prompt_template = """ You are Sarah, an empathetic HR interviewer conducting a technical interview in {language}. Current Question: {current_question} Previous conversation history: {history} User's response to current question: {user_input} Your response: """ interview_prompt = PromptTemplate( input_variables=["language", "current_question", "history", "user_input"], template=interview_prompt_template, ) interview_state.interview_chain = LLMChain(prompt=interview_prompt, llm=llm) report_prompt_template = """ You are an HR assistant tasked with generating a concise report based on the following interview transcript in {language}: {interview_transcript} Summarize the candidate's performance, highlighting strengths and areas for improvement. Keep it to 3-5 sentences. Report: """ report_prompt = PromptTemplate( input_variables=["language", "interview_transcript"], template=report_prompt_template ) interview_state.report_chain = LLMChain(prompt=report_prompt, llm=llm) return True # Indicate chains were initialized except Exception as e: print(f"Error initializing chains: {e}") interview_state.interview_chain = None interview_state.report_chain = None return False # Indicate chains were not initialized def generate_report(report_chain, history, language): """Generates a concise interview report.""" if report_chain is None: return "Report generation is unavailable because the API key is not set." # Handle uninitialized chain # Convert the Gradio-style history to a plain text transcript. transcript = "" for user_msg, bot_msg in history: transcript += f"User: {user_msg}\nAssistant: {bot_msg}\n" report = report_chain.invoke({"language": language, "interview_transcript": transcript}) return report["text"] def reset_interview_action(voice): """Resets the interview state and prepares the initial message.""" interview_state.reset(voice) if not initialize_chains(): # Initialize chains and check if successful initial_message_text = "OpenAI API key is not configured. Please set it in the Admin Panel to start the interview with full functionality." initial_audio_path = convert_text_to_speech_updated(initial_message_text) # Still try TTS for error message return ( [[None, initial_message_text]], # [user_msg, bot_msg]. User starts with None. gr.Audio(value=initial_audio_path, autoplay=True) if initial_audio_path else None, # Audio output might be None gr.Textbox(interactive=False), # Disable textbox if API key is missing, or keep interactive? Let's keep disabled for now. ) print(f"[DEBUG] Interview reset. Voice: {voice}") initial_message_text = get_interview_initial_message_hr(5) # Get initial message # Convert to speech and save to a temporary file. initial_audio_path = convert_text_to_speech_updated(initial_message_text, voice) # Return values in the correct format for Gradio. return ( [[None, initial_message_text]], # [user_msg, bot_msg]. User starts with None. gr.Audio(value=initial_audio_path, autoplay=True) if initial_audio_path else None, # Audio output might be None gr.Textbox(interactive=True), # Enable the textbox ) def start_interview(): """Starts the interview (used by the Gradio button).""" return reset_interview_action(interview_state.selected_interviewer) def construct_history_string(history): """Constructs a history string for the LangChain prompt.""" history_str = "" for user_msg, bot_msg in history: history_str += f"User: {user_msg}\nAssistant: {bot_msg}\n" return history_str def bot_response(chatbot, user_message_text): """Handles the bot's response logic.""" voice = interview_state.get_voice_setting() history_str = construct_history_string(chatbot) if interview_state.interview_chain is None: # Check if chain is initialized chatbot.append([user_message_text, "Please set up the OpenAI API key in the Admin Panel to continue the interview."]) return chatbot, None, gr.File(visible=False) # No audio or report if chain is not initialized if interview_state.question_count < len(interview_state.current_questions): current_question = interview_state.current_questions[interview_state.question_count] response_obj = interview_state.interview_chain.invoke( { "language": language, "current_question": current_question, "history": history_str, "user_input": user_message_text, } ) response = response_obj["text"] interview_state.question_count += 1 # Text-to-speech temp_audio_path = convert_text_to_speech_updated(response, voice) # Update chatbot history in the correct format. chatbot.append([user_message_text, response]) # Add user and bot messages return chatbot, gr.Audio(value=temp_audio_path, autoplay=True) if temp_audio_path else None, gr.File(visible=False) else: # Interview finished interview_state.interview_finished = True conclusion_message = "Thank you for your time. The interview is complete. Please review your report." # Text-to-speech for conclusion temp_conclusion_audio_path = convert_text_to_speech_updated(conclusion_message, voice) # Update chatbot history. chatbot.append([user_message_text, conclusion_message]) # Generate and save report. report_content = generate_report( interview_state.report_chain, chatbot, language ) # Pass Gradio history txt_path = save_interview_history( [f"User: {user}\nAssistant: {bot}" for user, bot in chatbot], language ) # Create plain text history report_file_path = store_interview_report(report_content) print(f"[DEBUG] Interview report saved at: {report_file_path}") return ( chatbot, gr.Audio(value=temp_conclusion_audio_path, autoplay=True) if temp_conclusion_audio_path else None, gr.File(visible=True, value=txt_path), ) def convert_text_to_speech_updated(text, voice="alloy"): """Converts text to speech and returns the file path, handles missing API key.""" api_key = os.getenv("OPENAI_API_KEY") if not api_key: print("API key is missing, text-to-speech disabled.") return None # Return None when API key is missing try: client = OpenAI(api_key=api_key) response = client.audio.speech.create(model="tts-1", voice=voice, input=text) with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as tmp_file: for chunk in response.iter_bytes(): tmp_file.write(chunk) temp_audio_path = tmp_file.name return temp_audio_path except Exception as e: print(f"Error in text-to-speech: {e}") return None def transcribe_audio(audio_file_path): """Transcribes audio to text, handles missing API key.""" api_key = os.getenv("OPENAI_API_KEY") if not api_key: print("API key is missing, audio transcription disabled.") return "" # Return empty string, transcription is unavailable try: client = OpenAI(api_key=api_key) with open(audio_file_path, "rb") as audio_file: transcription = client.audio.transcriptions.create( model="whisper-1", file=audio_file ) return transcription.text except Exception as e: print(f"Error in transcription: {e}") return "" def conduct_interview_updated(questions, language="English", history_limit=5): """Conducts the interview (LangChain/OpenAI), handles missing API key.""" openai_api_key = os.getenv("OPENAI_API_KEY") if not openai_api_key: # Return a placeholder interview step if API key is missing initial_message = "⚠️ OpenAI API Key not configured. Please enter your API key in the Admin Panel to start the interview with full functionality. Text responses will be displayed, but advanced features are disabled." placeholder_audio_path = convert_text_to_speech_updated(initial_message) def placeholder_interview_step(user_input, audio_input, history): history.append([None, initial_message]) # bot message in history return history, "", placeholder_audio_path, gr.Textbox(interactive=False) # Textbox disabled return placeholder_interview_step, initial_message, "API key missing" # Return placeholder and flag chat = ChatOpenAI( openai_api_key=openai_api_key, model="gpt-4o", temperature=0.7, max_tokens=750 ) conversation_history = deque(maxlen=history_limit) # For LangChain, not Gradio system_prompt = ( f"You are Sarah, an empathetic HR interviewer conducting a technical interview in {language}. " "Respond to user follow-up questions politely and concisely. Keep responses brief." ) interview_data = [] # Store Q&A for potential later use current_question_index = [0] is_interview_finished = [False] initial_message = ( "👋 Hi there, I'm Sarah, your friendly AI HR assistant! " "I'll guide you through a series of interview questions. " "Take your time." ) final_message = "That wraps up our interview. Thank you for your responses!" def interview_step(user_input, audio_input, history): nonlocal current_question_index, is_interview_finished if is_interview_finished[0]: return history, "", None, gr.Textbox(interactive=False) # No further interaction, textbox disabled if audio_input: user_input = transcribe_audio(audio_input) if not user_input: history.append(["", "I couldn't understand your audio. Could you please repeat or type?"]) #Empty string "" so the user input is not None audio_path = convert_text_to_speech_updated(history[-1][1]) #Access the content return history, "", audio_path, gr.Textbox(interactive=True) # Keep textbox interactive if user_input.lower() in ["exit", "quit"]: history.append(["", "The interview has ended. Thank you."])#Empty string "" so the user input is not None is_interview_finished[0] = True return history, "", None, gr.Textbox(interactive=False) # Disable textbox after exit # Crucial: Add USER INPUT to history *before* getting bot response. history.append([user_input, ""]) # Add user input, bot response pending question_text = questions[current_question_index[0]] # Prepare history for LangChain (not Gradio chatbot format) history_content = "\n".join( [ f"Q: {entry['question']}\nA: {entry['answer']}" for entry in conversation_history ] ) combined_prompt = ( f"{system_prompt}\n\nPrevious conversation history:\n{history_content}\n\n" f"Current question: {question_text}\nUser's input: {user_input}\n\n" "Respond warmly." ) messages = [ SystemMessage(content=system_prompt), HumanMessage(content=combined_prompt), ] response = chat.invoke(messages) response_content = response.content.strip() audio_path = convert_text_to_speech_updated(response_content) conversation_history.append({"question": question_text, "answer": user_input}) interview_data.append({"question": question_text, "answer": user_input}) # Update Gradio-compatible history. Crucial for display. history[-1][1] = response_content # Update the last entry with the bot's response interactive_textbox = gr.Textbox(interactive=True) # Keep textbox interactive in most steps if current_question_index[0] + 1 < len(questions): current_question_index[0] += 1 next_question = f"Next question: {questions[current_question_index[0]]}" next_question_audio_path = convert_text_to_speech_updated(next_question) # No need to add the "Next Question:" prompt to the displayed history. # The bot will say it. Adding it here would cause a double entry. return history, "", next_question_audio_path, interactive_textbox else: final_message_audio = convert_text_to_speech_updated(final_message) history.append([None, final_message]) # Final message, no user input. is_interview_finished[0] = True interactive_textbox = gr.Textbox(interactive=False) # Disable textbox at the end return history, "", final_message_audio, interactive_textbox return interview_step, initial_message, final_message def launch_candidate_app_updated(): """Launches the Gradio app for candidates.""" QUESTIONS_FILE_PATH = "questions.json" try: questions = read_questions_from_json(QUESTIONS_FILE_PATH) if not questions: raise ValueError("No questions found.") except (FileNotFoundError, json.JSONDecodeError, ValueError) as e: print(f"Error loading questions: {e}") with gr.Blocks() as error_app: gr.Markdown(f"# Error: {e}") return error_app interview_func, initial_message, api_status = conduct_interview_updated(questions) # Get API status def start_interview_ui(): """Starts the interview.""" history = [] if api_status == "API key missing": # Check API status from conduct_interview_updated initial_combined = initial_message # Initial message already indicates API key missing textbox_interactive = gr.Textbox(interactive=False) # Disable textbox if API key missing else: initial_combined = ( initial_message + " Let's begin! Here's the first question: " + questions[0] ) textbox_interactive = gr.Textbox(interactive=True) # Enable textbox if API key OK initial_audio_path = convert_text_to_speech_updated(initial_combined) history.append(["", initial_combined]) # Correct format: [user, bot] Empty string for user. return history, "", initial_audio_path, textbox_interactive # Return interactive textbox status def clear_interview_ui(): """Clears the interview and resets.""" # Recreate the object in order to clear the history of the interview nonlocal interview_func, initial_message, api_status # Include api_status to reset properly interview_func, initial_message, api_status = conduct_interview_updated(questions) # Re-init, get API status textbox_interactive = gr.Textbox(interactive= (api_status != "API key missing")) # Enable if API key is OK after clear, disable if missing. return [], "", None, textbox_interactive # Return textbox interactive state def interview_step_wrapper(user_response, audio_response, history): """Wrapper for the interview step function.""" history, user_text, audio_path, new_textbox_interactive = interview_func(user_response, audio_response, history) return history, "", audio_path, new_textbox_interactive def on_enter_submit(history, user_response): """Handles submission when Enter is pressed.""" if not user_response.strip(): return history, "", None, gr.Textbox(interactive=True) # Prevent empty submissions history, _, audio_path, new_textbox_interactive = interview_step_wrapper( user_response, None, history ) # No audio on Enter return history, "", audio_path, new_textbox_interactive with gr.Blocks(title="AI HR Interview Assistant") as candidate_app: gr.Markdown( "

👋 Welcome to Your AI HR Interview Assistant

" ) start_btn = gr.Button("Start Interview", variant="primary") chatbot = gr.Chatbot(label="Interview Chat", height=650) audio_input = gr.Audio( sources=["microphone"], type="filepath", label="Record Your Answer" ) user_input = gr.Textbox( label="Your Response", placeholder="Type your answer here or use the microphone...", lines=1, interactive=True, # Textbox interactive status is controlled by functions ) audio_output = gr.Audio(label="Response Audio", autoplay=True) with gr.Row(): submit_btn = gr.Button("Submit", variant="primary") clear_btn = gr.Button("Clear Chat") start_btn.click( start_interview_ui, inputs=[], outputs=[chatbot, user_input, audio_output, user_input] # user_input for textbox interactive status ) audio_input.stop_recording( interview_step_wrapper, inputs=[user_input, audio_input, chatbot], outputs=[chatbot, user_input, audio_output, user_input], # user_input for textbox interactive status ) submit_btn.click( interview_step_wrapper, inputs=[user_input, audio_input, chatbot], outputs=[chatbot, user_input, audio_output, user_input], # user_input for textbox interactive status ) user_input.submit( on_enter_submit, inputs=[chatbot, user_input], outputs=[chatbot, user_input, audio_output, user_input], # user_input for textbox interactive status ) clear_btn.click( clear_interview_ui, inputs=[], outputs=[chatbot, user_input, audio_output, user_input] # user_input for textbox interactive status ) return candidate_app # --- (End of Candidate Interview Implementation) --- def cleanup(): """Cleans up temporary audio files.""" for audio_file in interview_state.temp_audio_files: try: if os.path.exists(audio_file): os.unlink(audio_file) except Exception as e: print(f"Error deleting file {audio_file}: {e}")