AI Agent HR and Recruitment Automation Complete Guide 2026

class ResumeScreeningAgent:
    """AI agent for intelligent resume screening."""
    
    def __init__(self, criteria_model, embedding_model):
        self.criteria = criteria_model
        self.embeddings = embedding_model
        self.job_requirements = {}
    
    def configure_job(self, job_id, requirements):
        """Configure screening criteria for a job posting."""
        self.job_requirements[job_id] = requirements
        
        # Create embedding of requirements for semantic matching
        requirements_text = self._format_requirements(requirements)
        requirements_embedding = self.embeddings.encode(requirements_text)
        self.job_requirements[job_id + "_embedding"] = requirements_embedding
    
    def screen_resume(self, job_id, resume_data):
        """Screen a resume against job requirements."""
        requirements = self.job_requirements.get(job_id)
        if not requirements:
            raise ValueError(f"Job {job_id} not configured")
        
        results = {
            "resume_id": resume_data["id"],
            "job_id": job_id,
            "scores": {},
            "recommendation": None,
            "flags": []
        }
        
        # Technical skills matching
        technical_score = self._match_skills(
            resume_data.get("skills", []),
            requirements.get("required_skills", []),
            requirements.get("preferred_skills", [])
        )
        results["scores"]["technical"] = technical_score
        
        # Experience matching
        experience_score = self._evaluate_experience(
            resume_data.get("experience_years", 0),
            requirements.get("min_years", 0),
            requirements.get("max_years", None)
        )
        results["scores"]["experience"] = experience_score
        
        # Education matching
        education_score = self._evaluate_education(
            resume_data.get("education", ""),
            requirements.get("required_education", [])
        )
        results["scores"]["education"] = education_score
        
        # Semantic relevance using embeddings
        semantic_score = self._calculate_semantic_similarity(
            resume_data.get("resume_text", ""),
            job_id
        )
        results["scores"]["semantic"] = semantic_score
        
        # Calculate overall score
        weights = requirements.get("scoring_weights", {
            "technical": 0.35,
            "experience": 0.25,
            "education": 0.15,
            "semantic": 0.25
        })
        
        overall_score = sum(
            results["scores"][key] * weight 
            for key, weight in weights.items()
        )
        results["overall_score"] = overall_score
        
        # Generate recommendation
        threshold = requirements.get("passing_threshold", 0.7)
        if overall_score >= threshold:
            results["recommendation"] = "advance"
        elif overall_score >= threshold * 0.7:
            results["recommendation"] = "review"
        else:
            results["recommendation"] = "not_advance"
        
        # Check for red flags
        results["flags"] = self._check_red_flags(resume_data, requirements)
        
        return results
    
    def _match_skills(self, candidate_skills, required, preferred):
        """Match candidate skills against requirements."""
        candidate_lower = [s.lower() for s in candidate_skills]
        required_lower = [s.lower() for s in required]
        preferred_lower = [s.lower() for s in preferred]
        
        required_matches = sum(1 for s in required_lower if any(s in c or c in s for c in candidate_lower))
        preferred_matches = sum(1 for s in preferred_lower if any(s in c or c in s for c in candidate_lower))
        
        required_score = required_matches / len(required) if required else 1.0
        preferred_bonus = (preferred_matches / len(preferred) * 0.2) if preferred else 0
        
        return min(1.0, required_score + preferred_bonus)
    
    def _evaluate_experience(self, years, min_years, max_years):
        """Evaluate years of experience."""
        if years >= min_years:
            if max_years and years > max_years:
                # Overqualified
                return max(0.5, 1.0 - (years - max_years) * 0.1)
            return 1.0
        return years / min_years if min_years > 0 else 0
    
    def _calculate_semantic_similarity(self, resume_text, job_id):
        """Calculate semantic similarity using embeddings."""
        resume_embedding = self.embeddings.encode(resume_text)
        job_embedding = self.job_requirements.get(job_id + "_embedding")
        
        if job_embedding is None:
            return 0.5
        
        # Cosine similarity
        import numpy as np
        cosine_sim = np.dot(resume_embedding, job_embedding) / (
            np.linalg.norm(resume_embedding) * np.linalg.norm(job_embedding)
        )
        return (cosine_sim + 1) / 2  # Normalize to 0-1
    
    def _check_red_flags(self, resume_data, requirements):
        """Check for potential red flags."""
        flags = []
        
        # Check for gaps
        if resume_data.get("employment_gaps", []):
            for gap in resume_data["employment_gaps"]:
                if gap.get("months", 0) > 12:
                    flags.append({
                        "type": "employment_gap",
                        "severity": "medium",
                        "details": f"Employment gap of {gap['months']} months"
                    })
        
        # Check for frequent job changes
        if resume_data.get("job_changes_per_year", 0) > 2:
            flags.append({
                "type": "frequent_job_changes",
                "severity": "low",
                "details": "High frequency of job changes"
            })
        
        return flags

class CandidateCommunicationAgent:
    """AI agent for candidate communications."""
    
    def __init__(self, llm, crm_integration):
        self.llm = llm
        self.crm = crm_integration
        self.templates = self._load_templates()
    
    def handle_inbound_message(self, candidate_id, message):
        """Process incoming candidate message."""
        
        # Get candidate context
        candidate = self.crm.get_candidate(candidate_id)
        
        # Classify intent
        intent = self._classify_intent(message)
        
        # Generate appropriate response
        if intent == "status_inquiry":
            response = self._generate_status_update(candidate)
        elif intent == "question":
            response = self._answer_question(candidate, message)
        elif intent == "withdrawal":
            response = self._handle_withdrawal(candidate)
        elif intent == "scheduling":
            response = self._handle_scheduling_request(candidate, message)
        else:
            response = self._generate_escalation(candidate, message)
        
        # Log interaction
        self.crm.log_interaction(candidate_id, {
            "type": "inbound",
            "message": message,
            "intent": intent,
            "response": response
        })
        
        return response
    
    def send_scheduled_updates(self, job_id):
        """Send scheduled status updates to candidates."""
        candidates = self.crm.get_candidates_for_job(job_id)
        
        for candidate in candidates:
            # Check if update is due
            if self._should_send_update(candidate):
                status = self._get_current_status(candidate, job_id)
                message = self._format_status_update(candidate, status)
                
                self.crm.send_message(candidate["id"], message)
    
    def _classify_intent(self, message):
        """Classify the intent of a candidate message."""
        intent_classifier = """
        Classify this candidate message into one of:
        - status_inquiry: asking about application status
        - question: asking a specific question
        - withdrawal: withdrawing application
        - scheduling: requesting interview scheduling
        - other: anything else
        
        Message: {message}
        
        Intent:
        """
        
        response = self.llm.generate(intent_classifier.format(message=message))
        return response.strip().lower().replace(" ", "_")
    
    def _generate_status_update(self, candidate):
        """Generate personalized status update."""
        status = candidate.get("current_status", "application_received")
        
        templates = {
            "application_received": "Thank you for applying! We've received your application and are reviewing it. We'll be in touch within 5-7 business days.",
            "screening": "Great news! Your application has passed initial screening. We'll be in touch soon about next steps.",
            "interview": "Congratulations! We'd like to invite you for an interview. Please check your email for scheduling details.",
            "offer": "We're pleased to extend you an offer! Please check your email for details.",
            "rejected": "Thank you for your interest. After careful consideration, we've decided to move forward with other candidates."
        }
        
        return templates.get(status, "Thank you for your continued interest.")

class InterviewSchedulingAgent:
    """AI agent for interview scheduling."""
    
    def __init__(self, calendar_integration, candidate_portal):
        self.calendar = calendar_integration
        self.portal = candidate_portal
        self.timezone_handler = TimezoneHandler()
    
    def handle_scheduling_request(self, request):
        """Process interview scheduling request."""
        
        # Parse request details
        job_id = request.get("job_id")
        candidate_id = request.get("candidate_id")
        interviewer_ids = request.get("interviewers", [])
        interview_types = request.get("types", ["technical"])
        
        # Get interviewer availability
        availability = self._get_collective_availability(
            interviewer_ids,
            request.get("date_range", "next_two_weeks")
        )
        
        # Get candidate preferences
        candidate = self.portal.get_candidate(candidate_id)
        candidate_availability = candidate.get("availability", {})
        
        # Find matching slots
        matching_slots = self._find_matching_slots(
            availability,
            candidate_availability,
            request.get("duration", 60)
        )
        
        if not matching_slots:
            return {
                "status": "no_availability",
                "message": "No matching times found. Would you like to provide additional availability?"
            }
        
        # Present options to candidate
        options = self._format_time_options(matching_slots[:5], candidate.get("timezone"))
        
        # Store pending scheduling
        self.portal.send_scheduling_options(candidate_id, options)
        
        return {
            "status": "options_sent",
            "options": options
        }
    
    def process_candidate_selection(self, candidate_id, slot_index, job_id):
        """Process candidate's slot selection."""
        
        # Get the selected slot
        pending = self.portal.get_pending_scheduling(candidate_id)
        selected_slot = pending["options"][slot_index]
        
        # Create calendar events
        events = []
        for interviewer_id in pending["interviewers"]:
            event = self.calendar.create_event(
                organizer=interviewer_id,
                title=f"Interview - {job_id}",
                start_time=selected_slot["start"],
                end_time=selected_slot["end"],
                attendees=[candidate_id, interviewer_id]
            )
            events.append(event)
        
        # Send confirmations
        self._send_confirmations(candidate_id, pending["interviewers"], selected_slot, events)
        
        return {
            "status": "scheduled",
            "details": selected_slot
        }
    
    def handle_rescheduling(self, candidate_id, job_id, new_preference):
        """Handle interview rescheduling."""
        
        # Cancel existing events
        existing_events = self.calendar.get_events_for_interview(candidate_id, job_id)
        for event in existing_events:
            self.calendar.cancel_event(event["id"])
        
        # Request new scheduling
        return self.handle_scheduling_request({
            "candidate_id": candidate_id,
            "job_id": job_id,
            "date_range": new_preference
        })

class HRPolicyAgent:
    """AI agent for employee HR questions."""
    
    def __init__(self, llm, knowledge_base, user_directory):
        self.llm = llm
        self.kb = knowledge_base
        self.users = user_directory
        self.conversation_history = {}
    
    def handle_query(self, user_id, query):
        """Process employee query."""
        
        # Get user context
        user = self.users.get_user(user_id)
        
        # Check conversation history
        history = self.conversation_history.get(user_id, [])
        
        # Search knowledge base
        relevant_docs = self.kb.search(query, top_k=3)
        
        # Check for policy-specific queries
        policy_answer = self._check_policy_docs(query, relevant_docs)
        
        if policy_answer:
            # Use direct policy answer
            response = policy_answer
            source = "policy"
        else:
            # Generate answer using LLM with context
            response = self._generate_answer(query, relevant_docs, user, history)
            source = "generated"
        
        # Log for analytics
        self._log_query(user_id, query, response, source)
        
        # Update conversation history
        history.append({"query": query, "response": response})
        self.conversation_history[user_id] = history[-10:]  # Keep last 10
        
        return {
            "response": response,
            "source": source,
            "references": [doc["id"] for doc in relevant_docs] if source == "generated" else None
        }
    
    def _check_policy_docs(self, query, relevant_docs):
        """Check if query has direct policy answer."""
        
        # Look for exact or near-match policy documents
        for doc in relevant_docs:
            if doc.get("type") == "policy" and doc.get("confidence", 0) > 0.9:
                return self._format_policy_answer(doc)
        
        return None
    
    def _generate_answer(self, query, docs, user, history):
        """Generate answer using LLM with context."""
        
        context = self._build_context(query, docs, user, history)
        
        prompt = f"""You are an HR assistant helping employees with their questions.
        
        Context:
        {context}
        
        Employee Question: {query}
        
        Instructions:
        - Provide accurate, helpful answers based on the context
        - If information is not available, say so honestly
        - Be concise and friendly
        - Use a professional but warm tone
        - Do not make up information
        """
        
        return self.llm.generate(prompt)
    
    def _build_context(self, query, docs, user, history):
        """Build context for answer generation."""
        
        user_info = f"Employee: {user['name']}, Department: {user.get('department', 'N/A')}"
        
        doc_context = "\n\n".join([
            f"Document: {doc['title']}\n{doc['content'][:500]}"
            for doc in docs
        ])
        
        history_context = ""
        if history:
            history_context = "Recent conversation:\n"
            for turn in history[-3:]:
                history_context += f"Q: {turn['query']}\nA: {turn['response'][:200]}...\n"
        
        return f"{user_info}\n\n{doc_context}\n\n{history_context}"

class BenefitsAgent:
    """AI agent for benefits administration."""
    
    def __init__(self, benefits_system, llm):
        self.benefits = benefits_system
        self.llm = llm
    
    def handle_enrollment(self, user_id, enrollment_type):
        """Guide employee through benefits enrollment."""
        
        user = self.benefits.get_employee(user_id)
        current_benefits = user.get("current_benefits", {})
        
        if enrollment_type == "open":
            return self._handle_open_enrollment(user)
        elif enrollment_type == "new_hire":
            return self._handle_new_hire_enrollment(user)
        elif enrollment_type == "life_event":
            return self._handle_life_event_enrollment(user)
        else:
            return {"error": "Unknown enrollment type"}
    
    def answer_benefit_question(self, user_id, question):
        """Answer specific benefits question."""
        
        user = self.benefits.get_employee(user_id)
        benefits_info = self.benefits.get_plan_details(user.get("plans", []))
        
        # Search for relevant information
        relevant = self._search_benefits_info(question, benefits_info)
        
        # Generate answer
        answer = self.llm.generate(f"""
            Employee question about benefits: {question}
            
            Employee's current plans: {benefits_info['summary']}
            
            Relevant information: {relevant}
            
            Provide a helpful, accurate answer about their benefits.
        """)
        
        return {
            "answer": answer,
            "relevant_plans": relevant.get("plans", [])
        }

class OnboardingAgent:
    """AI agent for new hire onboarding."""
    
    def __init__(self, onboarding_workflow, it_systems, hr_systems):
        self.workflow = onboarding_workflow
        self.it = it_systems
        self.hr = hr_systems
        self.new_hires = {}
    
    def initiate_onboarding(self, employee_id, start_date, department):
        """Start onboarding process for new hire."""
        
        # Create onboarding checklist
        checklist = self.workflow.generate_checklist(department, "full_time")
        
        # Initialize tracking
        self.new_hires[employee_id] = {
            "start_date": start_date,
            "department": department,
            "checklist": checklist,
            "completed": [],
            "pending": checklist
        }
        
        # Trigger parallel tasks
        self._setup_it_access(employee_id, department)
        self._setup_hr_records(employee_id)
        self._send_welcome materials(employee_id)
        
        # Schedule first-day tasks
        self._schedule_day_one_tasks(employee_id, start_date)
        
        return {
            "status": "initiated",
            "checklist": checklist,
            "start_date": start_date
        }
    
    def handle_task_completion(self, employee_id, task_id):
        """Process task completion and trigger next steps."""
        
        hire = self.new_hires.get(employee_id)
        if not hire:
            return {"error": "Employee not found in onboarding"}
        
        # Mark complete
        hire["completed"].append(task_id)
        hire["pending"] = [t for t in hire["pending"] if t["id"] != task_id]
        
        # Check for dependent tasks
        completed_task = next(t for t in hire["completed"] if t["id"] == task_id)
        
        if completed_task.get("triggers"):
            for triggered_task in completed_task["triggers"]:
                if triggered_task not in hire["completed"]:
                    self._execute_task(employee_id, triggered_task)
        
        # Update progress
        progress = len(hire["completed"]) / len(hire["checklist"])
        
        return {
            "status": "updated",
            "progress": progress,
            "pending_count": len(hire["pending"])
        }
    
    def answer_onboarding_question(self, employee_id, question):
        """Answer new hire questions during onboarding."""
        
        hire = self.new_hires.get(employee_id)
        department = hire["department"] if hire else None
        
        # Check if it's a process question
        if "how do i" in question.lower() or "how to" in question.lower():
            return self._find_process_answer(question, department)
        
        # Check if it's an IT question
        if any(word in question.lower() for word in ["password", "email", "access", "login"]):
            return self._handle_it_question(employee_id, question)
        
        # General onboarding question
        return self._answer_general_question(question, department)

class HRAgentIntegration:
    """Integration architecture for HR AI agents."""
    
    def __init__(self):
        self.agents = {}
        self.integrations = {}
        self.event_bus = EventBus()
    
    def register_agent(self, agent_type, agent_instance):
        """Register an AI agent."""
        self.agents[agent_type] = agent_instance
    
    def connect_hr_system(self, system_name, connector):
        """Connect to HR systems."""
        self.integrations[system_name] = connector
    
    def setup_event_handlers(self):
        """Setup event handlers for cross-system workflows."""
        
        # Example: New hire triggers onboarding
        self.event_bus.subscribe(
            "employee.hire",
            self._handle_new_hire
        )
        
        # Example: Termination triggers exit process
        self.event_bus.subscribe(
            "employee.termination",
            self._handle_termination
        )
    
    def _handle_new_hire(self, event_data):
        """Handle new employee hire event."""
        
        employee_id = event_data["employee_id"]
        
        # Trigger onboarding agent
        onboarding_agent = self.agents.get("onboarding")
        if onboarding_agent:
            onboarding_agent.initiate_onboarding(
                employee_id=employee_id,
                start_date=event_data["start_date"],
                department=event_data["department"]
            )
        
        # Trigger IT provisioning
        it_agent = self.agents.get("it_provisioning")
        if it_agent:
            it_agent.setup_access(employee_id, event_data["department"])

class HRAgentSecurity:
    """Security for HR AI agents."""
    
    def __init__(self):
        self.access_control = AccessControlMatrix()
        self.audit_logger = AuditLogger()
        self.data_masker = DataMasker()
    
    def check_access(self, user_id, resource, action):
        """Check if user can access resource."""
        
        # Get user role
        user_role = self.access_control.get_user_role(user_id)
        
        # Check permission
        allowed = self.access_control.has_permission(user_role, resource, action)
        
        # Log attempt
        self.audit_logger.log({
            "user": user_id,
            "action": action,
            "resource": resource,
            "allowed": allowed
        })
        
        return allowed
    
    def mask_sensitive_data(self, data, user_role):
        """Mask sensitive data based on role."""
        
        if user_role in ["hr_admin", "executive"]:
            return data  # Full access
        
        if user_role == "manager":
            return self.data_masker.mask(data, ["salary", "ssn", "medical"])
        
        # Regular employee - limited access
        return self.data_masker.mask(data, ["salary", "ssn", "medical", "performance"])

class HRMetricsTracker:
    """Track HR AI agent metrics."""
    
    def __init__(self):
        self.metrics = {}
    
    def track_recruitment_metrics(self):
        """Track recruitment-specific metrics."""
        
        metrics = {
            "time_to_screen": self._avg_time_to_screen(),
            "screening_accuracy": self._calculate_accuracy(),
            "candidate_satisfaction": self._get_satisfaction_score(),
            "interview_scheduling_time": self._avg_scheduling_time(),
            "cost_per_hire": self._calculate_cost_per_hire()
        }
        
        return metrics
    
    def track_employee_service_metrics(self):
        """Track employee service metrics."""
        
        metrics = {
            "query_resolution_rate": self._resolution_rate(),
            "avg_response_time": self._avg_response_time(),
            "escalation_rate": self._escalation_rate(),
            "employee_satisfaction": self._get_escalation_satisfaction(),
            "self_service_adoption": self._adoption_rate()
        }
        
        return metrics
    
    def generate_report(self):
        """Generate comprehensive HR AI report."""
        
        recruitment = self.track_recruitment_metrics()
        service = self.track_employee_service_metrics()
        
        return {
            "recruitment": recruitment,
            "employee_service": service,
            "recommendations": self._generate_recommendations(recruitment, service)
        }