Spaces:
Runtime error
Runtime error
| from typing import Literal, Tuple, Dict, Optional | |
| import os | |
| import time | |
| import json | |
| import requests | |
| import PyPDF2 | |
| from datetime import datetime, timedelta | |
| import pytz | |
| import streamlit as st | |
| from phi.agent import Agent | |
| from phi.model.openai import OpenAIChat | |
| from phi.tools.email import EmailTools | |
| from phi.tools.zoom import ZoomTool | |
| from phi.utils.log import logger | |
| from streamlit_pdf_viewer import pdf_viewer | |
| class CustomZoomTool(ZoomTool): | |
| def __init__(self, *, account_id: Optional[str] = None, client_id: Optional[str] = None, client_secret: Optional[str] = None, name: str = "zoom_tool"): | |
| super().__init__(account_id=account_id, client_id=client_id, client_secret=client_secret, name=name) | |
| self.token_url = "https://zoom.us/oauth/token" | |
| self.access_token = None | |
| self.token_expires_at = 0 | |
| def get_access_token(self) -> str: | |
| if self.access_token and time.time() < self.token_expires_at: | |
| return str(self.access_token) | |
| headers = {"Content-Type": "application/x-www-form-urlencoded"} | |
| data = {"grant_type": "account_credentials", "account_id": self.account_id} | |
| try: | |
| response = requests.post(self.token_url, headers=headers, data=data, auth=(self.client_id, self.client_secret)) | |
| response.raise_for_status() | |
| token_info = response.json() | |
| self.access_token = token_info["access_token"] | |
| expires_in = token_info["expires_in"] | |
| self.token_expires_at = time.time() + expires_in - 60 | |
| self._set_parent_token(str(self.access_token)) | |
| return str(self.access_token) | |
| except requests.RequestException as e: | |
| logger.error(f"Error fetching access token: {e}") | |
| return "" | |
| def _set_parent_token(self, token: str) -> None: | |
| """Helper method to set the token in the parent ZoomTool class""" | |
| if token: | |
| self._ZoomTool__access_token = token | |
| # Role requirements as a constant dictionary | |
| ROLE_REQUIREMENTS: Dict[str, str] = { | |
| "ai_ml_engineer": """ | |
| Required Skills: | |
| - Python, PyTorch/TensorFlow | |
| - Machine Learning algorithms and frameworks | |
| - Deep Learning and Neural Networks | |
| - Data preprocessing and analysis | |
| - MLOps and model deployment | |
| - RAG, LLM, Finetuning and Prompt Engineering | |
| """, | |
| "frontend_engineer": """ | |
| Required Skills: | |
| - React/Vue.js/Angular | |
| - HTML5, CSS3, JavaScript/TypeScript | |
| - Responsive design | |
| - State management | |
| - Frontend testing | |
| """, | |
| "backend_engineer": """ | |
| Required Skills: | |
| - Python/Java/Node.js | |
| - REST APIs | |
| - Database design and management | |
| - System architecture | |
| - Cloud services (AWS/GCP/Azure) | |
| - Kubernetes, Docker, CI/CD | |
| """ | |
| } | |
| def init_session_state() -> None: | |
| """Initialize only necessary session state variables.""" | |
| defaults = { | |
| 'candidate_email': "", 'openai_api_key': "", 'resume_text': "", 'analysis_complete': False, | |
| 'is_selected': False, 'zoom_account_id': "", 'zoom_client_id': "", 'zoom_client_secret': "", | |
| 'email_sender': "", 'email_passkey': "", 'company_name': "", 'current_pdf': None | |
| } | |
| for key, value in defaults.items(): | |
| if key not in st.session_state: | |
| st.session_state[key] = value | |
| def create_resume_analyzer() -> Agent: | |
| """Creates and returns a resume analysis agent.""" | |
| if not st.session_state.openai_api_key: | |
| st.error("Please enter your OpenAI API key first.") | |
| return None | |
| return Agent( | |
| model=OpenAIChat( | |
| id="gpt-4o", | |
| api_key=st.session_state.openai_api_key | |
| ), | |
| description="You are an expert technical recruiter who analyzes resumes.", | |
| instructions=[ | |
| "Analyze the resume against the provided job requirements", | |
| "Be lenient with AI/ML candidates who show strong potential", | |
| "Consider project experience as valid experience", | |
| "Value hands-on experience with key technologies", | |
| "Return a JSON response with selection decision and feedback" | |
| ], | |
| markdown=True | |
| ) | |
| def create_email_agent() -> Agent: | |
| return Agent( | |
| model=OpenAIChat( | |
| id="gpt-4o", | |
| api_key=st.session_state.openai_api_key | |
| ), | |
| tools=[EmailTools( | |
| receiver_email=st.session_state.candidate_email, | |
| sender_email=st.session_state.email_sender, | |
| sender_name=st.session_state.company_name, | |
| sender_passkey=st.session_state.email_passkey | |
| )], | |
| description="You are a professional recruitment coordinator handling email communications.", | |
| instructions=[ | |
| "Draft and send professional recruitment emails", | |
| "Act like a human writing an email", | |
| "Maintain a friendly yet professional tone", | |
| "Always end emails with exactly: 'Best,\nthe ClutchAi recruiting team'", | |
| "Never include the sender's or receiver's name in the signature", | |
| f"The name of the company is '{st.session_state.company_name}'" | |
| ], | |
| markdown=True, | |
| show_tool_calls=True | |
| ) | |
| def create_scheduler_agent() -> Agent: | |
| zoom_tools = CustomZoomTool( | |
| account_id=st.session_state.zoom_account_id, | |
| client_id=st.session_state.zoom_client_id, | |
| client_secret=st.session_state.zoom_client_secret | |
| ) | |
| return Agent( | |
| name="Interview Scheduler", | |
| model=OpenAIChat( | |
| id="gpt-4o", | |
| api_key=st.session_state.openai_api_key | |
| ), | |
| tools=[zoom_tools], | |
| description="You are an interview scheduling coordinator.", | |
| instructions=[ | |
| "You are an expert at scheduling technical interviews using Zoom.", | |
| "Schedule interviews during business hours (9 AM - 5 PM EST)", | |
| "Create meetings with proper titles and descriptions", | |
| "Ensure all meeting details are included in responses", | |
| "Use ISO 8601 format for dates", | |
| "Handle scheduling errors gracefully" | |
| ], | |
| markdown=True, | |
| show_tool_calls=True | |
| ) | |
| def extract_text_from_pdf(pdf_file) -> str: | |
| try: | |
| pdf_reader = PyPDF2.PdfReader(pdf_file) | |
| text = "" | |
| for page in pdf_reader.pages: | |
| text += page.extract_text() | |
| return text | |
| except Exception as e: | |
| st.error(f"Error extracting PDF text: {str(e)}") | |
| return "" | |
| def extract_email_from_resume(resume_text: str, analyzer: Agent) -> Optional[str]: | |
| """ | |
| Uses GPT-4 to extract an email address from the resume text. | |
| """ | |
| try: | |
| response = analyzer.run( | |
| f"""Extract the email address from the following text: | |
| Resume Text: | |
| {resume_text} | |
| Return the email address only, or 'None' if not found.""" | |
| ) | |
| assistant_message = next((msg.content for msg in response.messages if msg.role == 'assistant'), None) | |
| return assistant_message.strip() if assistant_message and assistant_message != 'None' else None | |
| except Exception as e: | |
| logger.error(f"Error extracting email from resume: {str(e)}") | |
| return None | |
| def analyze_resume( | |
| resume_text: str, | |
| role: Literal["ai_ml_engineer", "frontend_engineer", "backend_engineer"], | |
| analyzer: Agent | |
| ) -> Tuple[bool, str,int]: | |
| try: | |
| response = analyzer.run( | |
| f"""Act as a ATS tracking agent, and score the resume first and Please analyze this resume against the following requirements and provide your response in valid JSON format also provide a score out of 100.: | |
| Role Requirements: | |
| {ROLE_REQUIREMENTS[role]} | |
| Resume Text: | |
| {resume_text} | |
| Your response must be a valid JSON object like this: | |
| {{ | |
| "selected": true/false, | |
| "feedback": "Detailed feedback explaining the decision", | |
| "matching_skills": ["skill1", "skill2"], | |
| "missing_skills": ["skill3", "skill4"], | |
| "experience_level": "junior/mid/senior" | |
| }} | |
| Evaluation criteria: | |
| 1. Match at least 70% of required skills | |
| 2. Consider both theoretical knowledge and practical experience | |
| 3. Value project experience and real-world applications | |
| 4. Consider transferable skills from similar technologies | |
| 5. Look for evidence of continuous learning and adaptability | |
| Important: Return ONLY the JSON object without any markdown formatting or backticks. | |
| """ | |
| ) | |
| assistant_message = next((msg.content for msg in response.messages if msg.role == 'assistant'), None) | |
| if not assistant_message: | |
| raise ValueError("No assistant message found in response.") | |
| result = json.loads(assistant_message.strip()) | |
| if not isinstance(result, dict) or not all(k in result for k in ["selected", "feedback"]): | |
| raise ValueError("Invalid response format") | |
| return result["selected"], result["feedback"] | |
| except (json.JSONDecodeError, ValueError) as e: | |
| st.error(f"Error processing response: {str(e)}") | |
| return False, f"Error analyzing resume: {str(e)}" | |
| # The modified sections include analyze_resume_with_score and main logic. | |
| def analyze_resume_with_score( | |
| resume_text: str, | |
| role: Literal["ai_ml_engineer", "frontend_engineer", "backend_engineer"], | |
| analyzer: Agent | |
| ) -> Tuple[bool, str, int]: | |
| try: | |
| response = analyzer.run( | |
| f"""Act as an ATS tracking agent. Please score this resume and analyze it against the following requirements: | |
| Role Requirements: | |
| {ROLE_REQUIREMENTS[role]} | |
| Resume Text: | |
| {resume_text} | |
| Your response must be a valid JSON object like this: | |
| {{ | |
| "selected": true/false, | |
| "feedback": "Detailed feedback explaining the decision", | |
| "ats_score": # ATS score as an integer percentage | |
| "matching_skills": ["skill1", "skill2"], | |
| "missing_skills": ["skill3", "skill4"], | |
| "experience_level": "junior/mid/senior" | |
| }} | |
| Evaluation criteria: | |
| 1. Match at least 70% of required skills | |
| 2. Consider both theoretical knowledge and practical experience | |
| 3. Value project experience and real-world applications | |
| 4. Consider transferable skills from similar technologies | |
| 5. Look for evidence of continuous learning and adaptability | |
| Important: Return ONLY the JSON object without any markdown formatting or backticks. | |
| """ | |
| ) | |
| assistant_message = next((msg.content for msg in response.messages if msg.role == 'assistant'), None) | |
| # DEBUG: Print the raw response | |
| print("DEBUG: Raw assistant message:", assistant_message) | |
| if not assistant_message: | |
| raise ValueError("No assistant message found in response.") | |
| result = json.loads(assistant_message.strip()) | |
| # DEBUG: Print parsed JSON response | |
| print("DEBUG: Parsed JSON response:", result) | |
| if not isinstance(result, dict) or not all(k in result for k in ["selected", "feedback", "ats_score"]): | |
| raise ValueError("Invalid response format") | |
| return result["selected"], result["feedback"], result["ats_score"] | |
| except (json.JSONDecodeError, ValueError) as e: | |
| st.error(f"Error processing response: {str(e)}") | |
| return False, f"Error analyzing resume: {str(e)}", 0 | |
| def send_selection_email(email_agent: Agent, to_email: str, role: str) -> None: | |
| email_agent.run( | |
| f""" | |
| Send an email to {to_email} regarding their selection for the {role} position. | |
| The email should: | |
| 1. Congratulate them on being selected | |
| 2. Explain the next steps in the process | |
| 3. Mention that they will receive interview details shortly | |
| 4. The name of the company is 'ClutchAI Recruiting Team' | |
| """ | |
| ) | |
| def send_rejection_email(email_agent: Agent, to_email: str, role: str, feedback: str) -> None: | |
| """ | |
| Send a rejection email with constructive feedback. | |
| """ | |
| email_agent.run( | |
| f""" | |
| Send an email to {to_email} regarding their application for the {role} position. | |
| Use this specific style: | |
| 2. be empathetic and human | |
| 3. mention specific feedback from: {feedback} | |
| 4. encourage them to upskill and try again | |
| 5. suggest some learning resources based on missing skills | |
| 6. end the email with exactly: | |
| Best, | |
| The Clutchai recruiting team | |
| Do not include any names in the signature. | |
| The tone should be like a human writing a quick but thoughtful email. | |
| """ | |
| ) | |
| def schedule_interview(scheduler: Agent, candidate_email: str, email_agent: Agent, role: str) -> None: | |
| """ | |
| Schedule interviews during business hours (9 AM - 5 PM IST). | |
| """ | |
| try: | |
| # Get current time in IST | |
| ny_tz= pytz.timezone('US/Eastern') | |
| current_time_ist = datetime.now(ny_tz) | |
| tomorrow_ist = current_time_ist + timedelta(days=1) | |
| interview_time = tomorrow_ist.replace(hour=11, minute=0, second=0, microsecond=0) | |
| formatted_time = interview_time.strftime('%Y-%m-%dT%H:%M:%S') | |
| meeting_response = scheduler.run( | |
| f"""Schedule a 60-minute technical interview with these specifications: | |
| - Title: '{role} Technical Interview' | |
| - Date: {formatted_time} | |
| - Timezone: EST (Eastern Standard Time) | |
| - Attendee: {candidate_email} | |
| Important Notes: | |
| - The meeting must be between 9 AM - 5 PM IST | |
| - Use EST (UTC+5:30) timezone for all communications | |
| - Include timezone information in the meeting details | |
| """ | |
| ) | |
| email_agent.run( | |
| f"""Send an interview confirmation email with these details: | |
| - Role: {role} position | |
| - Meeting Details: {meeting_response} | |
| Important: | |
| - Clearly specify that the time is in EST (Eastern Standard Time) | |
| - Ask the candidate to join 5 minutes early | |
| - Include timezone conversion link if possible | |
| - Ask him to be confident and not so nervous and prepare well for the interview | |
| """ | |
| ) | |
| st.success("Interview scheduled successfully!") | |
| except Exception as e: | |
| logger.error(f"Error scheduling interview: {str(e)}") | |
| st.error("Unable to schedule interview. Please try again.") | |
| def main() -> None: | |
| st.set_page_config(page_title="AI Recruitment System", page_icon="🤖", layout="wide") | |
| # Inject custom CSS for dark blue background | |
| st.markdown( | |
| """ | |
| <style> | |
| .stApp { | |
| background-color: #001f3f; /* Dark blue */ | |
| color: white; /* White text for contrast */ | |
| } | |
| </style> | |
| """, | |
| unsafe_allow_html=True | |
| ) | |
| st.title("🌟 AI Recruitment System") | |
| st.markdown( | |
| """ | |
| Welcome to the AI Recruitment System! | |
| Streamline your hiring process with AI-powered resume analysis and interview scheduling. | |
| """ | |
| ) | |
| st.markdown("---") | |
| init_session_state() | |
| with st.sidebar: | |
| st.header("Configuration") | |
| # OpenAI Configuration | |
| st.subheader("OpenAI Settings") | |
| api_key = st.text_input("OpenAI API Key", type="password", value=st.session_state.openai_api_key, help="Get your API key from platform.openai.com") | |
| if api_key: st.session_state.openai_api_key = api_key | |
| st.subheader("Zoom Settings") | |
| zoom_account_id = st.text_input("Zoom Account ID", type="password", value=st.session_state.zoom_account_id) | |
| zoom_client_id = st.text_input("Zoom Client ID", type="password", value=st.session_state.zoom_client_id) | |
| zoom_client_secret = st.text_input("Zoom Client Secret", type="password", value=st.session_state.zoom_client_secret) | |
| st.subheader("Email Settings") | |
| email_sender = st.text_input("Sender Email", value=st.session_state.email_sender, help="Email address to send from") | |
| email_passkey = st.text_input("Email App Password", type="password", value=st.session_state.email_passkey, help="App-specific password for email") | |
| company_name = st.text_input("Company Name", value=st.session_state.company_name, help="Name to use in email communications") | |
| if zoom_account_id: st.session_state.zoom_account_id = zoom_account_id | |
| if zoom_client_id: st.session_state.zoom_client_id = zoom_client_id | |
| if zoom_client_secret: st.session_state.zoom_client_secret = zoom_client_secret | |
| if email_sender: st.session_state.email_sender = email_sender | |
| if email_passkey: st.session_state.email_passkey = email_passkey | |
| if company_name: st.session_state.company_name = company_name | |
| required_configs = {'OpenAI API Key': st.session_state.openai_api_key, 'Zoom Account ID': st.session_state.zoom_account_id, | |
| 'Zoom Client ID': st.session_state.zoom_client_id, 'Zoom Client Secret': st.session_state.zoom_client_secret, | |
| 'Email Sender': st.session_state.email_sender, 'Email Password': st.session_state.email_passkey, | |
| 'Company Name': st.session_state.company_name} | |
| missing_configs = [k for k, v in required_configs.items() if not v] | |
| if missing_configs: | |
| st.warning(f"Please configure the following in the sidebar: {', '.join(missing_configs)}") | |
| return | |
| if not st.session_state.openai_api_key: | |
| st.warning("Please enter your OpenAI API key in the sidebar to continue.") | |
| return | |
| role = st.selectbox("Select the role you're applying for:", ["ai_ml_engineer", "frontend_engineer", "backend_engineer"]) | |
| with st.expander("View Required Skills", expanded=True): st.markdown(ROLE_REQUIREMENTS[role]) | |
| # Add a "New Application" button before the resume upload | |
| if st.button("📝 New Application"): | |
| # Clear only the application-related states | |
| keys_to_clear = ['resume_text', 'analysis_complete', 'is_selected', 'candidate_email', 'current_pdf'] | |
| for key in keys_to_clear: | |
| if key in st.session_state: | |
| st.session_state[key] = None if key == 'current_pdf' else "" | |
| st.rerun() | |
| resume_file = st.file_uploader("Upload your resume (PDF)", type=["pdf"], key="resume_uploader") | |
| if resume_file is not None and resume_file != st.session_state.get('current_pdf'): | |
| st.session_state.current_pdf = resume_file | |
| st.session_state.resume_text = "" | |
| st.session_state.analysis_complete = False | |
| st.session_state.is_selected = False | |
| st.rerun() | |
| if resume_file: | |
| st.subheader("Uploaded Resume") | |
| col1, col2 = st.columns([4, 1]) | |
| with col1: | |
| import tempfile, os | |
| with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp_file: | |
| tmp_file.write(resume_file.read()) | |
| tmp_file_path = tmp_file.name | |
| resume_file.seek(0) | |
| try: pdf_viewer(tmp_file_path) | |
| finally: os.unlink(tmp_file_path) | |
| with col2: | |
| st.download_button(label="📥 Download", data=resume_file, file_name=resume_file.name, mime="application/pdf") | |
| # Process the resume text | |
| if not st.session_state.resume_text: | |
| with st.spinner("Processing your resume..."): | |
| resume_text = extract_text_from_pdf(resume_file) | |
| if resume_text: | |
| st.session_state.resume_text = resume_text | |
| st.success("Resume processed successfully!") | |
| # Create a resume analyzer to extract email | |
| resume_analyzer = create_resume_analyzer() | |
| if resume_analyzer: | |
| email_from_resume = extract_email_from_resume(resume_text, resume_analyzer) | |
| if email_from_resume: | |
| st.session_state.candidate_email = email_from_resume | |
| st.success(f"Email ID extracted: {email_from_resume}") | |
| else: | |
| st.warning("Could not extract an email ID from the resume.") | |
| else: | |
| st.error("Could not process the PDF. Please try again.") | |
| # Email input with session state | |
| email = st.text_input( | |
| "Candidate's email address", | |
| value=st.session_state.candidate_email, | |
| key="email_input" | |
| ) | |
| st.session_state.candidate_email = email | |
| # Analysis and next steps | |
| if st.session_state.resume_text and email and not st.session_state.analysis_complete: | |
| if st.button("Analyze Resume"): | |
| with st.spinner("Analyzing your resume..."): | |
| resume_analyzer = create_resume_analyzer() | |
| email_agent = create_email_agent() # Create email agent here | |
| if resume_analyzer and email_agent: | |
| print("DEBUG: Starting resume analysis") | |
| is_selected, feedback, ats_score = analyze_resume_with_score( | |
| st.session_state.resume_text, | |
| role, | |
| resume_analyzer | |
| ) | |
| print(f"DEBUG: Analysis complete - Selected: {is_selected}, Feedback: {feedback}") | |
| # Display resume text | |
| st.subheader("Resume Contents") | |
| st.text_area("Extracted Resume Text", st.session_state.resume_text, height=300, disabled=True) | |
| # Display ATS score | |
| st.subheader("ATS Score") | |
| st.metric(label="ATS Match Score", value=f"{ats_score}%", help="Percentage match with the job requirements") | |
| if is_selected: | |
| st.success("Great! Candiate's skills match our requirements.") | |
| st.session_state.analysis_complete = True | |
| st.session_state.is_selected = True | |
| st.rerun() | |
| else: | |
| st.warning("Unfortunately, your skills don't match our requirements.") | |
| st.write(f"Feedback: {feedback}") | |
| # Send rejection email | |
| with st.spinner("Sending feedback email..."): | |
| try: | |
| send_rejection_email( | |
| email_agent=email_agent, | |
| to_email=email, | |
| role=role, | |
| feedback=feedback | |
| ) | |
| st.info("We've sent you an email with detailed feedback.") | |
| except Exception as e: | |
| logger.error(f"Error sending rejection email: {e}") | |
| st.error("Could not send feedback email. Please try again.") | |
| if st.session_state.get('analysis_complete') and st.session_state.get('is_selected', False): | |
| st.success("Congratulations! Your skills match our requirements.") | |
| st.info("Click 'Proceed with Application' to continue with the interview process.") | |
| if st.button("Proceed with Application", key="proceed_button"): | |
| print("DEBUG: Proceed button clicked") # Debug | |
| with st.spinner("🔄 Processing your application..."): | |
| try: | |
| print("DEBUG: Creating email agent") # Debug | |
| email_agent = create_email_agent() | |
| print(f"DEBUG: Email agent created: {email_agent}") # Debug | |
| print("DEBUG: Creating scheduler agent") # Debug | |
| scheduler_agent = create_scheduler_agent() | |
| print(f"DEBUG: Scheduler agent created: {scheduler_agent}") # Debug | |
| # 3. Send selection email | |
| with st.status("📧 Sending confirmation email...", expanded=True) as status: | |
| print(f"DEBUG: Attempting to send email to {st.session_state.candidate_email}") # Debug | |
| send_selection_email( | |
| email_agent, | |
| st.session_state.candidate_email, | |
| role | |
| ) | |
| print("DEBUG: Email sent successfully") # Debug | |
| status.update(label="✅ Confirmation email sent!") | |
| # 4. Schedule interview | |
| with st.status("📅 Scheduling interview...", expanded=True) as status: | |
| print("DEBUG: Attempting to schedule interview") # Debug | |
| schedule_interview( | |
| scheduler_agent, | |
| st.session_state.candidate_email, | |
| email_agent, | |
| role | |
| ) | |
| print("DEBUG: Interview scheduled successfully") # Debug | |
| status.update(label="✅ Interview scheduled!") | |
| print("DEBUG: All processes completed successfully") # Debug | |
| st.success(""" | |
| 🎉 Application Successfully Processed! | |
| Tasks Done: | |
| 1. Selection confirmation ✅ | |
| 2. Interview details with Zoom link 🔗 | |
| Next steps: | |
| 1. Join the interview 5 minutes early | |
| 2. Evaluate the candiate | |
| 3. Report the result to the HR | |
| """) | |
| except Exception as e: | |
| print(f"DEBUG: Error occurred: {str(e)}") # Debug | |
| print(f"DEBUG: Error type: {type(e)}") # Debug | |
| import traceback | |
| print(f"DEBUG: Full traceback: {traceback.format_exc()}") # Debug | |
| st.error(f"An error occurred: {str(e)}") | |
| st.error("Please try again or contact support.") | |
| # Reset button | |
| if st.sidebar.button("Reset Application"): | |
| for key in st.session_state.keys(): | |
| if key != 'openai_api_key': | |
| del st.session_state[key] | |
| st.rerun() | |
| if __name__ == "__main__": | |
| main() |