← Back to DevBytes

Building AI Agents for Recruiters: Screening and Scheduling

Building AI Agents for Recruiters: Screening and Scheduling

Recruitment teams are drowning in a sea of resumes while trying to coordinate interviews across multiple time zones, calendars, and stakeholders. AI agents—autonomous software programs that can reason, plan, and execute tasks—are transforming these two core recruiting functions from manual drudgery into streamlined, intelligent workflows. This tutorial walks you through building two production-ready AI agents: one for resume screening and candidate matching, and another for automated interview scheduling. By the end, you'll have a unified pipeline that can process hundreds of applicants and book interviews with zero human intervention.

What Are AI Recruiting Agents?

AI recruiting agents are specialized LLM-powered programs that operate within a defined domain to perform recruitment tasks autonomously. Unlike simple chatbots or static rule engines, these agents use large language models to understand unstructured data (resumes, job descriptions, email threads), reason about qualifications and availability, and act by producing structured outputs, sending emails, or updating calendars.

Our system consists of two collaborating agents:

Both agents share a common memory store (a lightweight database) so the screening agent's shortlist feeds directly into the scheduling agent's outreach queue.

Why AI Recruiting Agents Matter

The numbers speak for themselves. A single corporate job posting attracts an average of 250 resumes. Recruiters spend roughly 23 hours screening per role, and another 2–5 hours on scheduling logistics per candidate pipeline. For agencies handling dozens of roles simultaneously, the math becomes unsustainable.

AI agents deliver four concrete benefits:

Importantly, these agents are designed to augment human decision-making, not replace it. The screening agent provides explainable scores and lets recruiters adjust weights; the scheduling agent defers to human override on final slot confirmation when confidence is low.

Core Architecture Overview

Our system uses a modular architecture with clear separation of concerns. Here's the high-level data flow:

We'll build this using Python, OpenAI's API (for LLM and embedding calls), FastAPI for optional HTTP endpoints, and libraries like pdfplumber, icalendar, and smtplib. All code is self-contained and can run on any machine with Python 3.10+.

How to Build the Resume Screening Agent

Step 1: Project Setup and Dependencies

Create a new project directory and install the required packages:

mkdir recruiting-agents && cd recruiting-agents
python -m venv .venv && source .venv/bin/activate
pip install openai pdfplumber python-docx pydantic numpy sqlite3
pip install fastapi uvicorn icalendar aiosmtplib python-dotenv

Create a .env file for your API keys:

OPENAI_API_KEY=sk-your-key-here
FROM_EMAIL=recruiter@yourcompany.com
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USER=your-email@yourcompany.com
SMTP_PASS=your-app-password

Step 2: Data Models for Structured Extraction

We define Pydantic models that the LLM will populate from unstructured text. These models are the contract between extraction and matching:

# models.py
from pydantic import BaseModel, Field
from typing import List, Optional
from datetime import date
from enum import Enum

class SeniorityLevel(str, Enum):
    JUNIOR = "junior"
    MID = "mid"
    SENIOR = "senior"
    LEAD = "lead"
    EXECUTIVE = "executive"

class EmploymentType(str, Enum):
    FULL_TIME = "full_time"
    PART_TIME = "part_time"
    CONTRACT = "contract"
    INTERNSHIP = "internship"

class Skill(BaseModel):
    name: str
    years_of_experience: Optional[float] = None
    proficiency: Optional[str] = None  # beginner, intermediate, advanced, expert

class Education(BaseModel):
    degree: str
    institution: str
    graduation_year: Optional[int] = None
    field_of_study: Optional[str] = None

class WorkExperience(BaseModel):
    company: str
    title: str
    start_date: Optional[date] = None
    end_date: Optional[date] = None
    description: str
    achievements: List[str] = Field(default_factory=list)

class CandidateProfile(BaseModel):
    full_name: str
    email: Optional[str] = None
    phone: Optional[str] = None
    location: Optional[str] = None
    current_title: Optional[str] = None
    seniority_estimate: Optional[SeniorityLevel] = None
    skills: List[Skill] = Field(default_factory=list)
    education: List[Education] = Field(default_factory=list)
    work_experience: List[WorkExperience] = Field(default_factory=list)
    years_of_total_experience: Optional[float] = None
    certifications: List[str] = Field(default_factory=list)
    languages: List[str] = Field(default_factory=list)
    summary: Optional[str] = None

class JobRequirements(BaseModel):
    title: str
    department: Optional[str] = None
    required_skills: List[str] = Field(default_factory=list)
    nice_to_have_skills: List[str] = Field(default_factory=list)
    minimum_years_experience: Optional[float] = None
    required_education: Optional[str] = None
    seniority_level: Optional[SeniorityLevel] = None
    employment_type: Optional[EmploymentType] = None
    location_requirements: Optional[str] = None
    responsibilities: List[str] = Field(default_factory=list)
    salary_range: Optional[str] = None
    company_summary: Optional[str] = None

Step 3: Resume Parsing and Structured Extraction

The extraction function sends raw resume text to the LLM with a carefully engineered prompt that requests structured JSON output matching our Pydantic schema:

# extractor.py
import json
import pdfplumber
import docx
from openai import OpenAI
from typing import Dict, Any
from models import CandidateProfile, JobRequirements

client = OpenAI()

def parse_resume_file(file_path: str) -> str:
    """Extract raw text from PDF or DOCX resume files."""
    if file_path.endswith('.pdf'):
        with pdfplumber.open(file_path) as pdf:
            text = "\n".join(
                page.extract_text() or "" for page in pdf.pages
            )
        return text.strip()
    elif file_path.endswith('.docx'):
        doc = docx.Document(file_path)
        text = "\n".join(paragraph.text for paragraph in doc.paragraphs)
        return text.strip()
    else:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read().strip()

EXTRACTION_SYSTEM_PROMPT = """You are an expert resume parser. Extract ALL information from the resume
into a structured JSON object following the exact schema below. Be exhaustive.

Required fields and their types:
- full_name: string
- email: string or null
- phone: string or null
- location: string or null
- current_title: string or null
- seniority_estimate: "junior" | "mid" | "senior" | "lead" | "executive" | null
- skills: array of {name: string, years_of_experience: float|null, proficiency: string|null}
- education: array of {degree: string, institution: string, graduation_year: int|null, field_of_study: string|null}
- work_experience: array of {company: string, title: string, start_date: "YYYY-MM-DD"|null, end_date: "YYYY-MM-DD"|null, description: string, achievements: [string]}
- years_of_total_experience: float or null
- certifications: [string]
- languages: [string]
- summary: string or null

Rules:
1. Infer years_of_experience from work history dates even if not explicitly stated.
2. Estimate seniority_estimate based on titles, years, and scope of responsibilities.
3. For each skill, estimate years_of_experience from work history context.
4. If a field cannot be determined, use null. Never make up data.
5. Return ONLY valid JSON, no markdown, no commentary."""

def extract_candidate_profile(resume_text: str) -> CandidateProfile:
    """Use LLM to extract structured candidate data from resume text."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": EXTRACTION_SYSTEM_PROMPT},
            {"role": "user", "content": f"Parse this resume:\n\n{resume_text[:8000]}"}
        ],
        response_format={"type": "json_object"},
        temperature=0.1,
        max_tokens=2000
    )
    raw_json = json.loads(response.choices[0].message.content)
    return CandidateProfile(**raw_json)

def extract_job_requirements(job_description: str) -> JobRequirements:
    """Extract structured job requirements from a job posting."""
    system_prompt = """You are an expert at parsing job descriptions.
    Extract structured requirements as JSON following this schema:
    {
      "title": string,
      "department": string|null,
      "required_skills": [string],
      "nice_to_have_skills": [string],
      "minimum_years_experience": float|null,
      "required_education": string|null,
      "seniority_level": "junior"|"mid"|"senior"|"lead"|"executive"|null,
      "employment_type": "full_time"|"part_time"|"contract"|"internship"|null,
      "location_requirements": string|null,
      "responsibilities": [string],
      "salary_range": string|null,
      "company_summary": string|null
    }
    Return ONLY valid JSON."""
    
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": f"Parse this job description:\n\n{job_description[:6000]}"}
        ],
        response_format={"type": "json_object"},
        temperature=0.1,
        max_tokens=1500
    )
    raw_json = json.loads(response.choices[0].message.content)
    return JobRequirements(**raw_json)

Step 4: Semantic Matching and Scoring Engine

The matching engine combines embedding-based similarity with deterministic rule checks. This hybrid approach gives both nuance (semantic understanding) and precision (hard gates on must-have requirements):

# matcher.py
import numpy as np
from openai import OpenAI
from typing import List, Tuple
from models import CandidateProfile, JobRequirements

client = OpenAI()

def compute_embedding(text: str) -> List[float]:
    """Get OpenAI embedding vector for a text block."""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text[:8000]  # truncate to context limit
    )
    return response.data[0].embedding

def cosine_similarity(a: List[float], b: List[float]) -> float:
    """Compute cosine similarity between two vectors."""
    a_np = np.array(a)
    b_np = np.array(b)
    return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))

def build_candidate_text(profile: CandidateProfile) -> str:
    """Build a representative text blob from structured profile for embedding."""
    parts = []
    if profile.summary:
        parts.append(profile.summary)
    if profile.current_title:
        parts.append(f"Current Title: {profile.current_title}")
    parts.append("Skills: " + ", ".join(
        f"{s.name}" + (f" ({s.years_of_experience}y)" if s.years_of_experience else "")
        for s in profile.skills
    ))
    parts.append("Experience: " + " | ".join(
        f"{exp.title} at {exp.company}: {exp.description[:200]}"
        for exp in profile.work_experience[:5]
    ))
    if profile.education:
        parts.append("Education: " + ", ".join(
            f"{ed.degree} from {ed.institution}"
            for ed in profile.education
        ))
    return "\n".join(parts)

def build_job_text(requirements: JobRequirements) -> str:
    """Build a representative text blob from job requirements for embedding."""
    parts = [f"Title: {requirements.title}"]
    if requirements.company_summary:
        parts.append(requirements.company_summary)
    parts.append("Required Skills: " + ", ".join(requirements.required_skills))
    if requirements.nice_to_have_skills:
        parts.append("Nice-to-have: " + ", ".join(requirements.nice_to_have_skills))
    parts.append("Responsibilities: " + "; ".join(requirements.responsibilities))
    return "\n".join(parts)

def check_hard_requirements(
    profile: CandidateProfile,
    requirements: JobRequirements
) -> Tuple[bool, List[str]]:
    """Check non-negotiable requirements. Returns (passed, [reasons])."""
    failures = []
    candidate_skill_names = {s.name.lower() for s in profile.skills}
    
    # Check required skills
    for req_skill in requirements.required_skills:
        if req_skill.lower() not in candidate_skill_names:
            # Check for partial matches
            found = any(req_skill.lower() in cs or cs in req_skill.lower() 
                       for cs in candidate_skill_names)
            if not found:
                failures.append(f"Missing required skill: {req_skill}")
    
    # Check minimum years experience
    if requirements.minimum_years_experience:
        if profile.years_of_total_experience is None:
            failures.append("Cannot verify minimum experience — no total years extracted")
        elif profile.years_of_total_experience < requirements.minimum_years_experience:
            failures.append(
                f"Experience: {profile.years_of_total_experience}y vs required {requirements.minimum_years_experience}y"
            )
    
    # Check seniority match
    if requirements.seniority_level:
        seniority_order = {"junior": 1, "mid": 2, "senior": 3, "lead": 4, "executive": 5}
        if profile.seniority_estimate:
            profile_level = seniority_order.get(profile.seniority_estimate.value, 0)
            required_level = seniority_order.get(requirements.seniority_level.value, 0)
            if profile_level < required_level - 1:
                failures.append(
                    f"Seniority mismatch: candidate is {profile.seniority_estimate.value}, "
                    f"role requires {requirements.seniority_level.value}"
                )
    
    return len(failures) == 0, failures

def score_candidate(
    profile: CandidateProfile,
    requirements: JobRequirements,
    job_embedding: List[float] = None
) -> dict:
    """
    Score a candidate profile against job requirements.
    Returns a dict with detailed scoring breakdown.
    """
    # Pre-compute job embedding if not provided
    if job_embedding is None:
        job_text = build_job_text(requirements)
        job_embedding = compute_embedding(job_text)
    
    # Compute semantic similarity score (0-100)
    candidate_text = build_candidate_text(profile)
    candidate_embedding = compute_embedding(candidate_text)
    semantic_score = cosine_similarity(candidate_embedding, job_embedding) * 100
    
    # Check hard requirements
    hard_pass, hard_failures = check_hard_requirements(profile, requirements)
    
    # Skill match bonus (0-50)
    required_set = {s.lower() for s in requirements.required_skills}
    nice_set = {s.lower() for s in requirements.nice_to_have_skills}
    candidate_skills = {s.name.lower() for s in profile.skills}
    
    required_match = len(required_set & candidate_skills) / max(len(required_set), 1)
    nice_match = len(nice_set & candidate_skills) / max(len(nice_set), 1)
    skill_bonus = (required_match * 30) + (nice_match * 20)
    
    # Experience bonus (0-20)
    exp_bonus = 0.0
    if requirements.minimum_years_experience and profile.years_of_total_experience:
        ratio = profile.years_of_total_experience / requirements.minimum_years_experience
        exp_bonus = min(ratio * 10, 20)
    
    # Composite score
    composite_score = semantic_score + skill_bonus + exp_bonus
    # Cap at 100
    composite_score = min(composite_score, 100.0)
    
    return {
        "candidate_name": profile.full_name,
        "candidate_email": profile.email,
        "semantic_score": round(semantic_score, 2),
        "skill_bonus": round(skill_bonus, 2),
        "experience_bonus": round(exp_bonus, 2),
        "composite_score": round(composite_score, 2),
        "hard_requirements_pass": hard_pass,
        "hard_failures": hard_failures,
        "seniority_estimate": profile.seniority_estimate.value if profile.seniority_estimate else None,
        "years_of_experience": profile.years_of_total_experience,
    }

def rank_candidates(
    profiles: List[CandidateProfile],
    requirements: JobRequirements
) -> List[dict]:
    """Score and rank all candidates for a job, returning sorted list."""
    job_embedding = compute_embedding(build_job_text(requirements))
    scored = []
    for profile in profiles:
        result = score_candidate(profile, requirements, job_embedding=job_embedding)
        scored.append(result)
    
    # Sort: hard-pass first, then by composite score descending
    scored.sort(key=lambda r: (not r["hard_requirements_pass"], -r["composite_score"]))
    return scored

Step 5: The Screening Agent Pipeline

This orchestrator ties parsing, extraction, and matching together into a single runnable pipeline:

# screening_agent.py
import os
import json
import sqlite3
from typing import List
from extractor import parse_resume_file, extract_candidate_profile, extract_job_requirements
from matcher import rank_candidates
from models import CandidateProfile, JobRequirements

class ScreeningAgent:
    def __init__(self, db_path: str = "recruiting.db"):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        """Create tables for storing screening results."""
        conn = sqlite3.connect(self.db_path)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS screening_results (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                job_title TEXT,
                candidate_email TEXT,
                candidate_name TEXT,
                composite_score REAL,
                hard_pass INTEGER,
                hard_failures TEXT,
                raw_profile TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        conn.execute("""
            CREATE TABLE IF NOT EXISTS job_descriptions (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT,
                raw_text TEXT,
                structured_requirements TEXT,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        """)
        conn.commit()
        conn.close()
    
    def process_resumes(
        self,
        resume_paths: List[str],
        job_description_text: str
    ) -> List[dict]:
        """
        Main pipeline: parse resumes, extract profiles, score against job.
        Returns ranked list of candidate scores.
        """
        # Extract job requirements once
        print("Extracting job requirements...")
        job_reqs = extract_job_requirements(job_description_text)
        self._store_job(job_description_text, job_reqs)
        
        profiles = []
        for path in resume_paths:
            print(f"Processing: {path}")
            raw_text = parse_resume_file(path)
            profile = extract_candidate_profile(raw_text)
            profiles.append(profile)
            print(f"  -> Extracted profile for {profile.full_name}")
        
        # Rank all candidates
        print(f"Scoring {len(profiles)} candidates...")
        rankings = rank_candidates(profiles, job_reqs)
        
        # Persist results
        self._store_results(job_reqs.title, rankings)
        
        return rankings
    
    def _store_job(self, raw_text: str, requirements: JobRequirements):
        conn = sqlite3.connect(self.db_path)
        conn.execute(
            "INSERT INTO job_descriptions (title, raw_text, structured_requirements) VALUES (?, ?, ?)",
            (requirements.title, raw_text, requirements.model_dump_json())
        )
        conn.commit()
        conn.close()
    
    def _store_results(self, job_title: str, rankings: List[dict]):
        conn = sqlite3.connect(self.db_path)
        for r in rankings:
            conn.execute(
                "INSERT INTO screening_results (job_title, candidate_email, candidate_name, "
                "composite_score, hard_pass, hard_failures, raw_profile) VALUES (?, ?, ?, ?, ?, ?, ?)",
                (
                    job_title,
                    r.get("candidate_email"),
                    r["candidate_name"],
                    r["composite_score"],
                    1 if r["hard_requirements_pass"] else 0,
                    json.dumps(r.get("hard_failures", [])),
                    json.dumps(r)
                )
            )
        conn.commit()
        conn.close()
    
    def get_shortlist(self, job_title: str, min_score: float = 60.0) -> List[dict]:
        """Retrieve candidates above a score threshold for a given job."""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.execute(
            "SELECT candidate_name, candidate_email, composite_score, hard_pass, hard_failures "
            "FROM screening_results WHERE job_title = ? AND composite_score >= ? AND hard_pass = 1 "
            "ORDER BY composite_score DESC",
            (job_title, min_score)
        )
        results = [
            {
                "name": row[0],
                "email": row[1],
                "score": row[2],
                "hard_pass": bool(row[3]),
                "hard_failures": json.loads(row[4]) if row[4] else []
            }
            for row in cursor.fetchall()
        ]
        conn.close()
        return results

# --- Example usage ---
if __name__ == "__main__":
    agent = ScreeningAgent()
    resumes = [
        "./resumes/alice_smith.pdf",
        "./resumes/bob_jones.docx",
        "./resumes/charlie_lee.pdf",
    ]
    job_desc = """
    Senior Software Engineer - Backend
    Required: 5+ years Python, AWS experience, REST API design
    Nice-to-have: Kubernetes, PostgreSQL optimization
    Location: Remote (US time zones)
    """
    shortlist = agent.process_resumes(resumes, job_desc)
    for i, candidate in enumerate(shortlist[:5]):
        print(f"{i+1}. {candidate['candidate_name']} — Score: {candidate['composite_score']:.1f}")

How to Build the Interview Scheduling Agent

Step 1: Calendar Integration Layer

The scheduling agent needs to read and write calendar events. We'll build a clean abstraction that works with both Google Calendar API and local ICS file-based calendars:

# calendar_manager.py
from datetime import datetime, timedelta, time
from typing import List, Dict, Optional
from dataclasses import dataclass
from enum import Enum

class AvailabilitySlot:
    """Represents a free block of time on a calendar."""
    def __init__(self, start: datetime, end: datetime):
        self.start = start
        self.end = end
    
    def duration_minutes(self) -> int:
        return int((self.end - self.start).total_seconds() / 60)
    
    def to_dict(self) -> dict:
        return {
            "start": self.start.isoformat(),
            "end": self.end.isoformat(),
            "duration_minutes": self.duration_minutes()
        }
    
    def __repr__(self):
        return f"Slot({self.start.strftime('%Y-%m-%d %H:%M')} -> {self.end.strftime('%H:%M')})"

@dataclass
class CalendarEvent:
    summary: str
    start: datetime
    end: datetime
    attendees: List[str]
    location: Optional[str] = None
    description: Optional[str] = None

class CalendarManager:
    """
    Manages recruiter calendar — finding free slots, booking events.
    In production, this would wrap Google Calendar API.
    Here we use a simulated in-memory calendar with typical business hours.
    """
    
    def __init__(self, timezone_offset: int = -5):
        """
        Args:
            timezone_offset: Hours offset from UTC (default US Eastern: -5)
        """
        self.timezone_offset = timezone_offset
        self.events: List[CalendarEvent] = []
        # Default business hours: 9 AM to 5 PM, Monday-Friday
        self.business_hours_start = time(9, 0)
        self.business_hours_end = time(17, 0)
        self.business_days = {0, 1, 2, 3, 4}  # Monday=0 through Friday=4
    
    def add_event(self, event: CalendarEvent):
        """Add an existing event (e.g., from Google Calendar sync)."""
        self.events.append(event)
    
    def is_business_hours(self, dt: datetime) -> bool:
        """Check if a datetime falls within business hours."""
        if dt.weekday() not in self.business_days:
            return False
        slot_time = dt.time()
        return self.business_hours_start <= slot_time <= self.business_hours_end
    
    def get_free_slots(
        self,
        start_date: datetime,
        end_date: datetime,
        duration_minutes: int = 60,
        gap_minutes: int = 15
    ) -> List[AvailabilitySlot]:
        """
        Find all available time slots in the given date range.
        
        Args:
            start_date: Beginning of search window
            end_date: End of search window
            duration_minutes: Required length of each slot
            gap_minutes: Buffer time between meetings
        """
        free_slots = []
        current = start_date.replace(minute=0, second=0, microsecond=0)
        
        while current < end_date:
            # Skip non-business hours
            if not self.is_business_hours(current):
                current += timedelta(hours=1)
                continue
            
            slot_end = current + timedelta(minutes=duration_minutes)
            
            # Check slot doesn't exceed business hours for that day
            business_end_today = current.replace(
                hour=self.business_hours_end.hour,
                minute=self.business_hours_end.minute
            )
            if slot_end > business_end_today:
                current = (current + timedelta(days=1)).replace(
                    hour=self.business_hours_start.hour,
                    minute=self.business_hours_start.minute
                )
                continue
            
            # Check no conflicts with existing events (with gap buffer)
            conflict = False
            buffered_start = current - timedelta(minutes=gap_minutes)
            buffered_end = slot_end + timedelta(minutes=gap_minutes)
            for evt in self.events:
                if buffered_start < evt.end and buffered_end > evt.start:
                    conflict = True
                    break
            
            if not conflict:
                free_slots.append(AvailabilitySlot(current, slot_end))
            
            current += timedelta(minutes=30)  # Check every 30 min increment
        
        return free_slots
    
    def book_event(self, slot: AvailabilitySlot, candidate_name: str, candidate_email: str) -> CalendarEvent:
        """Book a confirmed interview slot."""
        event = CalendarEvent(
            summary=f"Interview: {candidate_name}",
            start=slot.start,
            end=slot.end,
            attendees=[candidate_email, "recruiter@company.com"],
            description=f"Candidate interview for {candidate_name}",
            location="Video Call (link to be sent)"
        )
        self.events.append(event)
        return event
    
    def cancel_event(self, start_time: datetime, candidate_email: str) -> bool:
        """Cancel an interview by matching start time and attendee."""
        for evt in self.events:
            if evt.start == start_time and candidate_email in evt.attendees:
                self.events.remove(evt)
                return True
        return False

Step 2: Email Automation for Candidate Communication

The agent communicates with candidates via email. We build a templated email system with LLM-powered response interpretation:

# email_manager.py
import smtplib
import os
import re
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from datetime import datetime
from typing import Optional, List
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()
client

— Ad —

Google AdSense will appear here after approval

← Back to all articles