How to Build an AI Recruitment Agent That Screens Resumes and Shortlists Candidates

ai-recruitment-agent/
├── app/
│   ├── __init__.py
│   ├── anonymizer.py
│   ├── schemas.py
│   └── scorer.py
├── tests/
│   └── test_recruitment_agent.py
└── requirements.txt

import re
from typing import Tuple, Dict

EMAIL_REGEX = re.compile(r"[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+")
PHONE_REGEX = re.compile(r"(\+?\d{1,3}[\s-]?)?\(?\d{3,4}\)?[\s.-]?\d{3}[\s.-]?\d{4}")
PRONOUN_REGEX = re.compile(r"\b(he/him|she/her|they/them|mr\.|mrs\.|ms\.)\b", re.IGNORECASE)

def anonymize_resume_text(raw_text: str, candidate_name: str = "") -> Tuple[str, Dict[str, str]]:
    """Masks contact PII, candidate name, and gendered titles before LLM evaluation."""
    pii_vault: Dict[str, str] = {}
    emails = EMAIL_REGEX.findall(raw_text)
    if emails:
        pii_vault["email"] = emails[0]

    redacted = EMAIL_REGEX.sub("[REDACTED_EMAIL]", raw_text)
    redacted = PHONE_REGEX.sub("[REDACTED_PHONE]", redacted)
    redacted = PRONOUN_REGEX.sub("[REDACTED_TITLE]", redacted)

    if candidate_name:
        for part in candidate_name.split():
            if len(part) > 2:
                redacted = re.sub(rf"\b{re.escape(part)}\b", "[CANDIDATE]", redacted, flags=re.IGNORECASE)

    return redacted, pii_vault

from typing import List, Dict
from pydantic import BaseModel, Field

class SkillEvidence(BaseModel):
    canonical_skill: str = Field(..., description="Normalized skill name, e.g. 'PostgreSQL', 'FastAPI'")
    years_used: float = Field(..., ge=0.0)
    verbatim_quote: str = Field(..., description="Exact quote from resume proving this skill.")

class NormalizedCandidateProfile(BaseModel):
    candidate_id: str
    total_relevant_experience_years: float = Field(..., ge=0.0)
    skills: List[SkillEvidence]
    highest_degree: str
    production_projects_count: int = Field(..., ge=0)

class JobRequirementRubric(BaseModel):
    job_id: str
    title: str
    minimum_experience_years: float
    mandatory_skills: List[str]
    preferred_skills_weights: Dict[str, float] = Field(
        ..., description="Mapping of canonical skill name to point weight (0-100 scale)"
    )

class CandidateEvaluationReport(BaseModel):
    candidate_id: str
    passed_mandatory_knockout: bool
    missing_mandatory_skills: List[str]
    total_score: float
    score_breakdown: Dict[str, float]
    evidence_citations: Dict[str, str]

from typing import Dict
from app.schemas import (
    NormalizedCandidateProfile,
    JobRequirementRubric,
    CandidateEvaluationReport,
)

SKILL_SYNONYM_MAP: Dict[str, str] = {
    "postgres": "postgresql",
    "pg": "postgresql",
    "k8s": "kubernetes",
    "py": "python",
    "reactjs": "react",
    "fastapi": "fastapi",
}

def canonicalize_skill(name: str) -> str:
    cleaned = name.strip().lower()
    return SKILL_SYNONYM_MAP.get(cleaned, cleaned)

def evaluate_candidate_deterministically(
    profile: NormalizedCandidateProfile,
    rubric: JobRequirementRubric,
) -> CandidateEvaluationReport:
    """Computes an auditable, deterministic score from extracted skill evidence."""
    candidate_skill_map = {
        canonicalize_skill(s.canonical_skill): s for s in profile.skills
    }

    missing_mandatory = []
    for req_skill in rubric.mandatory_skills:
        canon_req = canonicalize_skill(req_skill)
        if canon_req not in candidate_skill_map:
            missing_mandatory.append(req_skill)

    exp_passed = profile.total_relevant_experience_years >= rubric.minimum_experience_years
    passed_knockout = (len(missing_mandatory) == 0) and exp_passed

    breakdown: Dict[str, float] = {}
    citations: Dict[str, str] = {}

    # 1. Experience score (up to 30 points)
    if rubric.minimum_experience_years > 0:
        exp_ratio = min(1.5, profile.total_relevant_experience_years / rubric.minimum_experience_years)
        breakdown["experience_points"] = round((exp_ratio / 1.5) * 30.0, 2)
    else:
        breakdown["experience_points"] = 30.0

    # 2. Mandatory skills (30 points if all present, proportional otherwise)
    if rubric.mandatory_skills:
        matched_count = len(rubric.mandatory_skills) - len(missing_mandatory)
        breakdown["mandatory_skills_points"] = round(
            (matched_count / len(rubric.mandatory_skills)) * 30.0, 2
        )
        for req_skill in rubric.mandatory_skills:
            c_skill = canonicalize_skill(req_skill)
            if c_skill in candidate_skill_map:
                citations[req_skill] = candidate_skill_map[c_skill].verbatim_quote
    else:
        breakdown["mandatory_skills_points"] = 30.0

    # 3. Preferred skills weighted points (up to 40 points)
    pref_points = 0.0
    for pref_skill, weight in rubric.preferred_skills_weights.items():
        c_pref = canonicalize_skill(pref_skill)
        if c_pref in candidate_skill_map:
            pref_points += weight
            citations[pref_skill] = candidate_skill_map[c_pref].verbatim_quote
    breakdown["preferred_skills_points"] = round(min(40.0, pref_points), 2)

    total_score = round(sum(breakdown.values()), 2)
    return CandidateEvaluationReport(
        candidate_id=profile.candidate_id,
        passed_mandatory_knockout=passed_knockout,
        missing_mandatory_skills=missing_mandatory,
        total_score=total_score,
        score_breakdown=breakdown,
        evidence_citations=citations,
    )

from app.anonymizer import anonymize_resume_text
from app.schemas import NormalizedCandidateProfile, SkillEvidence, JobRequirementRubric
from app.scorer import evaluate_candidate_deterministically

def test_blind_resume_anonymizer():
    raw = "John Doe (he/him) | john.doe@gmail.com | +1-555-234-5678 | Built Postgres APIs."
    redacted, vault = anonymize_resume_text(raw, candidate_name="John Doe")
    assert "john.doe@gmail.com" not in redacted
    assert "John" not in redacted
    assert "he/him" not in redacted
    assert vault["email"] == "john.doe@gmail.com"

def test_deterministic_candidate_scoring_and_synonyms():
    profile = NormalizedCandidateProfile(
        candidate_id="cand_01",
        total_relevant_experience_years=4.5,
        highest_degree="B.Sc. Data Analytics",
        production_projects_count=3,
        skills=[
            SkillEvidence(canonical_skill="Python", years_used=4.0, verbatim_quote="Built FastAPI services in Python 3.12"),
            SkillEvidence(canonical_skill="Postgres", years_used=3.0, verbatim_quote="Designed Postgres transactional outbox"),
            SkillEvidence(canonical_skill="K8s", years_used=2.0, verbatim_quote="Deployed microservices on K8s"),
        ],
    )
    rubric = JobRequirementRubric(
        job_id="job_ai_eng",
        title="Senior Backend AI Engineer",
        minimum_experience_years=3.0,
        mandatory_skills=["Python", "PostgreSQL"],
        preferred_skills_weights={"Kubernetes": 20.0, "LangGraph": 20.0},
    )
    report = evaluate_candidate_deterministically(profile, rubric)
    assert report.passed_mandatory_knockout is True
    assert report.missing_mandatory_skills == []
    assert report.total_score == 80.0
    assert "Designed Postgres transactional outbox" in report.evidence_citations["PostgreSQL"]