Job-Interview / gptgr-manager.py
ruslanmv's picture
First commit
5798cfc
raw
history blame
7.64 kB
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
# Load environment variables
load_dotenv()
class InterviewState:
def __init__(self):
self.reset()
def reset(self, voice="alloy"):
self.question_count = 0
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.admin_authenticated = False
self.document_loaded = False
self.knowledge_retrieval_setup = False
self.interview_chain = None
self.report_chain = None
def get_voice_setting(self):
return self.selected_interviewer
interview_state = InterviewState()
# Function to read questions from JSON
def read_questions_from_json(file_path):
if not os.path.exists(file_path):
raise FileNotFoundError(f"The file '{file_path}' does not exist.")
with open(file_path, 'r') as f:
questions_list = json.load(f)
if not questions_list:
raise ValueError("The JSON file is empty or has invalid content.")
return questions_list
# Conduct interview and handle user input
def conduct_interview(questions, language="English", history_limit=5):
openai_api_key = os.getenv("OPENAI_API_KEY")
if not openai_api_key:
raise RuntimeError("OpenAI API key not found. Please add it to your .env file as OPENAI_API_KEY.")
chat = ChatOpenAI(
openai_api_key=openai_api_key, model="gpt-4", temperature=0.7, max_tokens=750
)
conversation_history = deque(maxlen=history_limit)
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. If the user is confused, provide clear clarification.")
interview_data = []
current_question_index = [0]
initial_message = ("👋 Hi there, I'm Sarah, your friendly AI HR assistant! "
"I'll guide you through a series of interview questions to learn more about you. "
"Take your time and answer each question thoughtfully.")
def interview_step(user_input, history):
if user_input.lower() in ["exit", "quit"]:
history.append({"role": "assistant", "content": "The interview has ended at your request. Thank you for your time!"})
return history, ""
question_text = questions[current_question_index[0]]
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 in a warm and conversational way, offering natural follow-ups if needed.")
messages = [
SystemMessage(content=system_prompt),
HumanMessage(content=combined_prompt)
]
response = chat.invoke(messages)
response_content = response.content.strip()
conversation_history.append({"question": question_text, "answer": user_input})
interview_data.append({"question": question_text, "answer": user_input})
history.append({"role": "user", "content": user_input})
history.append({"role": "assistant", "content": response_content})
if current_question_index[0] + 1 < len(questions):
current_question_index[0] += 1
next_question = f"Alright, let's move on. {questions[current_question_index[0]]}"
history.append({"role": "assistant", "content": next_question})
return history, ""
else:
history.append({"role": "assistant", "content": "That wraps up our interview. Thank you so much for your responses—it's been great learning more about you!"})
return history, ""
return interview_step, initial_message
def launch_candidate_app():
QUESTIONS_FILE_PATH = "questions.json"
try:
questions = read_questions_from_json(QUESTIONS_FILE_PATH)
interview_func, initial_message = conduct_interview(questions)
def start_interview_ui():
history = [{"role": "assistant", "content": initial_message}]
history.append({"role": "assistant", "content": "Let's begin! Here's your first question: " + questions[0]})
return history, ""
def clear_interview_ui():
return [], ""
def on_enter_submit_ui(history, user_response):
if not user_response.strip():
return history, ""
history, _ = interview_func(user_response, history)
return history, ""
with gr.Blocks(title="AI HR Interview Assistant") as candidate_app:
gr.Markdown("<h1 style='text-align: center;'>👋 Welcome to Your AI HR Interview Assistant</h1>")
start_btn = gr.Button("Start Interview", variant="primary")
chatbot = gr.Chatbot(label="Interview Chat", height=650, type="messages")
user_input = gr.Textbox(label="Your Response", placeholder="Type your answer here...", lines=1)
with gr.Row():
submit_btn = gr.Button("Submit")
clear_btn = gr.Button("Clear Chat")
start_btn.click(start_interview_ui, inputs=[], outputs=[chatbot, user_input])
submit_btn.click(on_enter_submit_ui, inputs=[chatbot, user_input], outputs=[chatbot, user_input])
user_input.submit(on_enter_submit_ui, inputs=[chatbot, user_input], outputs=[chatbot, user_input])
clear_btn.click(clear_interview_ui, inputs=[], outputs=[chatbot, user_input])
return candidate_app
except Exception as e:
print(f"Error: {e}")
return None
def create_manager_app():
with gr.Blocks(title="AI HR Interviewer Manager") as manager_app:
gr.HTML("<h1 style='text-align: center;'>AI HR Interviewer Manager</h1>")
user_role = gr.Dropdown(choices=["Admin", "Candidate"], label="Select User Role", value="Candidate")
proceed_button = gr.Button("👉 Proceed")
candidate_ui = gr.Column(visible=False)
admin_ui = gr.Column(visible=False)
with candidate_ui:
gr.Markdown("## 🚀 Candidate Interview")
candidate_app = launch_candidate_app()
with admin_ui:
gr.Markdown("## 🔒 Admin Panel")
gr.Markdown("Admin operations and question generation will go here.")
def show_selected_ui(role):
if role == "Candidate":
return gr.update(visible=True), gr.update(visible=False)
elif role == "Admin":
return gr.update(visible=False), gr.update(visible=True)
else:
return gr.update(visible=False), gr.update(visible=False)
proceed_button.click(show_selected_ui, inputs=[user_role], outputs=[candidate_ui, admin_ui])
return manager_app
def cleanup():
for audio_file in interview_state.temp_audio_files:
if os.path.exists(audio_file):
os.unlink(audio_file)
if __name__ == "__main__":
manager_app = create_manager_app()
try:
manager_app.launch(server_name="0.0.0.0", server_port=7860, debug=True)
finally:
cleanup()