Spaces:
Running
Running
| import streamlit as st | |
| from crewai import Crew | |
| from jd_generator import jd_generator, create_jd_task | |
| from resume_ranker import resume_ranker, create_resume_rank_task | |
| from email_automation import email_automation, create_email_task, simulate_email_api | |
| from interview_scheduler import interview_scheduler, create_schedule_task, simulate_calendar_api, create_schedule_summary_task | |
| from interview_agent import interview_agent, create_interview_task, evaluate_response_task | |
| from hire_recommendation import hire_recommendation_agent, create_hire_recommendation_task | |
| from sentiment_analyzer import sentiment_analyzer, create_sentiment_task | |
| from datetime import datetime | |
| import os | |
| import logging | |
| from streamlit_extras.add_vertical_space import add_vertical_space | |
| # Setup logging | |
| logging.basicConfig(filename="Logs/app.log", level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") | |
| st.set_page_config(page_title="AI Recruitment System", layout="wide") | |
| st.title("AI Recruitment System") | |
| # MCP Context | |
| if "mcp_context" not in st.session_state: | |
| st.session_state.mcp_context = { | |
| "job_description": None, | |
| "ranked_resumes": None, | |
| "scheduled_time": None, | |
| "interview_transcript": None | |
| } | |
| # Sidebar for transparency | |
| st.sidebar.markdown(""" | |
| ### AI Capabilities & Limitations | |
| - **Powered by llama3-70b-8192**: Generates human-like text but may produce inaccuracies. | |
| - **RAG**: Enhances outputs with web data, limited by search quality. | |
| - **Simulation**: Email and calendar APIs are simulated. | |
| ### Ethical Hiring | |
| Data is processed in-memory and not stored unless saved by the user. | |
| """) | |
| tabs = st.tabs([ | |
| "JD Generator", "Resume Ranker", "Email Automation", "Interview Scheduler", | |
| "Interview Agent", "Hire Recommendation", "Sentiment Analyzer" | |
| ]) | |
| # Tab 1: JD Generator | |
| with tabs[0]: | |
| st.header("JD Generator") | |
| with st.expander("Template & Inputs", expanded=True): | |
| st.markdown("**Upload a JD template** *(optional, defaults to Templates/jd_template.txt)*") | |
| template_file = st.file_uploader("Upload JD Template (.txt)", type=["txt"], key="jd_template") | |
| if template_file: | |
| with open("Templates/jd_template.txt", "wb") as f: | |
| f.write(template_file.read()) | |
| job_title = st.text_input("Job Title", "e.g., Senior Python Developer", key="jd_job_title", help="Enter the job title") | |
| skills = st.text_area("Required Skills", "e.g., Python, Flask, SQL, AWS", key="jd_skills") | |
| experience_level = st.text_input("Experience Level", "e.g., 5+ years", key="jd_experience") | |
| if st.button("Generate Job Description", key="jd_button", help="Generate a detailed job description"): | |
| if job_title and skills and experience_level: | |
| with st.spinner("Generating detailed JD..."): | |
| try: | |
| jd_task = create_jd_task(job_title, skills, experience_level) | |
| crew = Crew(agents=[jd_generator], tasks=[jd_task], verbose=True) | |
| result = crew.kickoff() | |
| st.session_state.mcp_context["job_description"] = result | |
| st.subheader("Generated Job Description") | |
| st.markdown(result) # Use markdown to render formatted output | |
| logging.info(f"JD generated for {job_title}") | |
| except Exception as e: | |
| st.error(f"Error: {str(e)}. See Logs/app.log for details.") | |
| logging.error(f"JD generation failed: {str(e)}") | |
| else: | |
| st.error("Fill in all fields.") | |
| logging.warning("JD generation attempted with missing fields") | |
| # Tab 2: Resume Ranker | |
| with tabs[1]: | |
| st.header("Resume Ranker") | |
| with st.expander("Inputs", expanded=True): | |
| job_desc = st.text_area("Job Description", value=st.session_state.mcp_context["job_description"] or "", key="resume_job_desc", help="Paste or enter job description") | |
| dir_path = st.text_input("Directory Path", "e.g., D:/resumes", key="dir_path") | |
| uploaded_files = st.file_uploader("Upload Resume PDFs", type=["pdf"], accept_multiple_files=True, key="resume_files") | |
| if st.button("Rank Resumes", key="resume_button", help="Rank uploaded or directory resumes"): | |
| if job_desc: | |
| if (dir_path and os.path.isdir(dir_path)) or uploaded_files: | |
| with st.spinner("Ranking resumes..."): | |
| try: | |
| task = create_resume_rank_task(job_desc, dir_path, uploaded_files) | |
| if task: | |
| crew = Crew(agents=[resume_ranker], tasks=[task], verbose=True) | |
| result = crew.kickoff() | |
| st.session_state.mcp_context["ranked_resumes"] = result | |
| st.subheader("Ranked Resumes") | |
| st.write(result) | |
| logging.info("Resumes ranked") | |
| else: | |
| st.error("No valid resumes found.") | |
| except Exception as e: | |
| st.error(f"Error: {str(e)}. See Logs/app.log for details.") | |
| logging.error(f"Resume ranking failed: {str(e)}") | |
| else: | |
| st.error("Provide a directory path or upload resumes.") | |
| else: | |
| st.error("Provide a job description.") | |
| # Tab 3: Email Automation (Personalized AI-Generated Content) | |
| with tabs[2]: | |
| st.header("Email Automation") | |
| with st.expander("Email Details", expanded=True): | |
| st.markdown("**Enter details for a personalized email**") | |
| candidate_name = st.text_input("Candidate Name", "e.g., John Doe", key="email_candidate", help="Candidate's full name") | |
| job_title_email = st.text_input("Job Title", "e.g., Senior Python Developer", key="email_job_title", help="Job title for the email") | |
| email_type = st.selectbox("Email Type", ["interview_invite", "hiring_team_update"], key="email_type", help="Choose email purpose") | |
| details = st.text_area("Details", value=st.session_state.mcp_context["scheduled_time"] or "e.g., March 25, 2025, 10 AM", key="email_details", help="e.g., interview time or update details") | |
| recipient_email = st.text_input("Recipient Email", "e.g., [email protected]", key="email_recipient", help="Recipient's email address") | |
| if st.button("Send Email", key="email_button", help="Generate and simulate sending a personalized email"): | |
| if candidate_name and job_title_email and details and recipient_email: | |
| with st.spinner("Generating personalized email..."): | |
| try: | |
| task = create_email_task(candidate_name, email_type, job_title_email, details, recipient_email) | |
| crew = Crew(agents=[email_automation], tasks=[task], verbose=True) | |
| email_content = crew.kickoff() | |
| result = simulate_email_api(email_content, recipient_email) | |
| st.subheader("Email Content") | |
| st.write(email_content) | |
| st.subheader("API Response") | |
| st.write(result) | |
| logging.info(f"Email simulated for {recipient_email}") | |
| except Exception as e: | |
| st.error(f"Error: {str(e)}. See Logs/app.log for details.") | |
| logging.error(f"Email generation failed: {str(e)}") | |
| else: | |
| st.error("Fill in all fields.") | |
| # Tab 4: Interview Scheduler (Candidate Availability Only) | |
| with tabs[3]: | |
| st.header("Interview Scheduler") | |
| with st.expander("Scheduling Details", expanded=True): | |
| st.markdown("**Enter candidate availability** ") | |
| candidate_name_sched = st.text_input("Candidate Name", "e.g., John Doe", key="sched_candidate", help="Candidate's full name") | |
| job_title_sched = st.text_input("Job Title", "e.g., Senior Python Developer", key="sched_job_title", help="Job title for the interview") | |
| candidate_avail = st.text_area("Candidate Availability", "e.g., March 25, 2025, 9 AM - 12 PM", key="sched_candidate_avail", help="e.g., March 25, 2025, 9 AM - 12 PM") | |
| if st.button("Schedule Interview", key="sched_button", help="Schedule the interview based on candidate availability"): | |
| if candidate_name_sched and job_title_sched and candidate_avail: | |
| with st.spinner("Scheduling interview..."): | |
| try: | |
| time_task = create_schedule_task(job_title_sched, candidate_name_sched, candidate_avail) | |
| crew = Crew(agents=[interview_scheduler], tasks=[time_task], verbose=True) | |
| scheduled_time_str = crew.kickoff() | |
| scheduled_time = datetime.strptime(scheduled_time_str, "%B %d, %Y, %I:%M %p") | |
| calendar_result = simulate_calendar_api(candidate_name_sched, job_title_sched, scheduled_time) | |
| st.session_state.mcp_context["scheduled_time"] = scheduled_time_str | |
| summary_task = create_schedule_summary_task(candidate_name_sched, job_title_sched, scheduled_time_str) | |
| crew = Crew(agents=[interview_scheduler], tasks=[summary_task], verbose=True) | |
| summary = crew.kickoff() | |
| st.subheader("Scheduled Time") | |
| st.write(scheduled_time_str) | |
| st.subheader("Calendar Response") | |
| st.write(calendar_result) | |
| st.subheader("Interview Summary") | |
| st.write(summary) | |
| logging.info(f"Interview scheduled for {candidate_name_sched} with summary") | |
| except Exception as e: | |
| st.error(f"Error: {str(e)}. See Logs/app.log for details.") | |
| logging.error(f"Scheduling failed: {str(e)}") | |
| else: | |
| st.error("Fill in all fields.") | |
| logging.warning("Scheduling attempted with missing fields") | |
| # Tab 5: Interview Agent | |
| with tabs[4]: | |
| st.header("Interview Agent") | |
| with st.expander("Job Description", expanded=True): | |
| st.markdown("**Enter the job description** *(e.g., Senior Python Developer requiring...)*") | |
| job_desc_interview = st.text_area("Job Description", value=st.session_state.mcp_context["job_description"] or "", key="interview_job_desc") | |
| if "interview_history" not in st.session_state: | |
| st.session_state.interview_history = [] | |
| if "current_question" not in st.session_state: | |
| st.session_state.current_question = None | |
| if st.button("Start Interview", key="start_interview", help="Begin the interview"): | |
| if job_desc_interview: | |
| with st.spinner("Generating Initial Question..."): | |
| try: | |
| task = create_interview_task(job_desc_interview) | |
| crew = Crew(agents=[interview_agent], tasks=[task], verbose=True) | |
| question = crew.kickoff() | |
| st.session_state.current_question = question | |
| st.session_state.interview_history = [{"role": "agent", "content": str(question)}] | |
| logging.info("Interview started") | |
| except Exception as e: | |
| st.error(f"Error: {str(e)}. See Logs/app.log for details.") | |
| logging.error(f"Interview start failed: {str(e)}") | |
| else: | |
| st.error("Provide a job description.") | |
| if st.session_state.interview_history: | |
| st.subheader("Conversation History") | |
| for message in st.session_state.interview_history: | |
| st.write(f"**{message['role'].capitalize()}**: {message['content']}") | |
| if st.session_state.current_question: | |
| candidate_response = st.text_area("Your Response", key="candidate_response", value="", height=100) | |
| if st.button("Submit Response", key="submit_response", help="Submit your answer"): | |
| if candidate_response: | |
| with st.spinner("Generating Follow-up..."): | |
| try: | |
| st.session_state.interview_history.append({"role": "candidate", "content": candidate_response}) | |
| eval_task = evaluate_response_task(job_desc_interview, st.session_state.interview_history, candidate_response) | |
| crew = Crew(agents=[interview_agent], tasks=[eval_task], verbose=True) | |
| follow_up = crew.kickoff() | |
| st.session_state.current_question = follow_up | |
| st.session_state.interview_history.append({"role": "agent", "content": str(follow_up)}) | |
| st.session_state.mcp_context["interview_transcript"] = "\n".join([f"{m['role']}: {m['content']}" for m in st.session_state.interview_history]) | |
| st.experimental_rerun() | |
| logging.info("Follow-up question generated") | |
| except Exception as e: | |
| st.error(f"Error: {str(e)}. See Logs/app.log for details.") | |
| logging.error(f"Follow-up generation failed: {str(e)}") | |
| else: | |
| st.error("Provide a response.") | |
| # Tab 6: Hire Recommendation | |
| with tabs[5]: | |
| st.header("Hire Recommendation Agent") | |
| with st.expander("Transcript Input", expanded=True): | |
| transcript_default = st.session_state.mcp_context["interview_transcript"] or "" | |
| transcript_file = st.file_uploader("Upload Transcript (.txt)", type=["txt"], key="hire_transcript") | |
| transcript_text = st.text_area("Or Paste Transcript", value=transcript_default, key="hire_transcript_text") | |
| if st.button("Generate Recommendation", key="hire_button", help="Analyze transcript for hiring decision"): | |
| transcript = transcript_file.read().decode("utf-8") if transcript_file else transcript_text | |
| if transcript: | |
| with st.spinner("Analyzing Transcript..."): | |
| try: | |
| task = create_hire_recommendation_task(transcript) | |
| crew = Crew(agents=[hire_recommendation_agent], tasks=[task], verbose=True) | |
| result = crew.kickoff() | |
| st.subheader("Hiring Recommendation") | |
| st.write(result) | |
| logging.info("Hiring recommendation generated") | |
| except Exception as e: | |
| st.error(f"Error: {str(e)}. See Logs/app.log for details.") | |
| logging.error(f"Hire recommendation failed: {str(e)}") | |
| else: | |
| st.error("Upload a file or paste a transcript.") | |
| # Tab 7: Sentiment Analyzer | |
| with tabs[6]: | |
| st.header("Sentiment Analyzer") | |
| with st.expander("Transcript Input", expanded=True): | |
| sentiment_default = st.session_state.mcp_context["interview_transcript"] or "" | |
| sentiment_file = st.file_uploader("Upload Transcript (.txt)", type=["txt"], key="sentiment_transcript") | |
| sentiment_text = st.text_area("Or Paste Transcript", value=sentiment_default, key="sentiment_transcript_text") | |
| if st.button("Analyze Sentiment", key="sentiment_button", help="Analyze transcript sentiment"): | |
| transcript = sentiment_file.read().decode("utf-8") if sentiment_file else sentiment_text | |
| if transcript: | |
| with st.spinner("Analyzing Sentiment..."): | |
| try: | |
| task = create_sentiment_task(transcript) | |
| crew = Crew(agents=[sentiment_analyzer], tasks=[task], verbose=True) | |
| result = crew.kickoff() | |
| st.subheader("Sentiment Analysis") | |
| st.write(result) | |
| logging.info("Sentiment analysis completed") | |
| except Exception as e: | |
| st.error(f"Error: {str(e)}. See Logs/app.log for details.") | |
| logging.error(f"Sentiment analysis failed: {str(e)}") | |
| else: | |
| st.error("Upload a file or paste a transcript.") | |
| add_vertical_space(2) |