How to Build an AI Agent for Recruitment Screening

import fitz  # PyMuPDF, for resume PDFs

def pull_new(ats, seen: set) -> list[dict]:
    out = []
    for app in ats.list_applications(status="new"):
        if app["id"] in seen:                 # already screened
            continue
        seen.add(app["id"])
        text = parse_resume(app["resume_url"])    # PDF / docx -> plain text
        out.append({"id": app["id"], "role": app["role_id"], "text": text})
    return out

def parse_resume(url: str) -> str:
    doc = fitz.open(stream=download(url), filetype="pdf")
    return "\n".join(page.get_text() for page in doc)
# Application id makes it idempotent: each candidate is screened once, even on a re-run.

from google import genai
from google.genai import types
from pydantic import BaseModel
client = genai.Client()

class Profile(BaseModel):
    skills: list[str]; years_experience: float
    qualifications: list[str]; relevant_titles: list[str]
    # Note: NO name, gender, age, photo, religion, or marital status.

def extract_profile(resume_text: str) -> Profile:
    res = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=["Extract only the job-relevant fields below. Do NOT include "
                  "name, gender, age, photo, religion, or marital status.\n" + resume_text],
        config=types.GenerateContentConfig(
            response_mime_type="application/json", response_schema=Profile))
    return res.parsed                # a typed Profile, protected fields excluded

from google import genai
from google.genai import types
from pydantic import BaseModel
client = genai.Client()

class CriterionScore(BaseModel):
    criterion: str; score: int; reason: str   # 0-5, with a written reason

class Scored(BaseModel):
    scores: list[CriterionScore]

def score(profile: dict, rubric: list[dict]) -> Scored:
    res = client.models.generate_content(
        model="gemini-2.5-flash",
        contents=[f"Score this profile against each rubric criterion 0-5 and give "
                  f"a one-line reason per criterion. Judge job-relevant evidence "
                  f"only.\nRubric: {rubric}\nProfile: {profile}"],
        config=types.GenerateContentConfig(
            response_mime_type="application/json", response_schema=Scored))
    return res.parsed                # per-criterion scores, each with a reason

def total_score(scores: list[dict], weights: dict[str, float]) -> float:
    # weighted sum of per-criterion scores; weights are set per role
    return sum(s["score"] * weights.get(s["criterion"], 1.0) for s in scores)

def rank(candidates: list[dict], weights: dict[str, float]) -> list[dict]:
    for c in candidates:
        c["total"] = total_score(c["scores"], weights)
    return sorted(candidates, key=lambda c: c["total"], reverse=True)
# Plain arithmetic: same inputs, same order, fully explainable. A recommendation,
# not a decision; a person reviews the ranked list.

PROTECTED = {"name", "gender", "age", "photo", "religion", "marital_status"}
PROXY_HINTS = ("pin code", "postcode", "graduated", "date of birth", "dob")

def guardrail_check(profile: dict, raw_text: str) -> list[str]:
    flags = []
    if PROTECTED & set(profile):               # a protected field leaked in
        flags.append("protected_attribute_present")
    if any(h in raw_text.lower() for h in PROXY_HINTS):
        flags.append("possible_proxy_signal")   # for human attention
    return flags

def disparate_impact(shortlisted: dict, applied: dict) -> dict:
    # rates compared across groups, using data held separately for monitoring only
    rates = {g: shortlisted.get(g, 0) / max(applied.get(g, 1), 1) for g in applied}
    top = max(rates.values()) or 1
    # flag any group whose selection rate is below 80% of the highest (a common check)
    return {g: r for g, r in rates.items() if r < 0.8 * top}
# Reduces bias and surfaces issues for review. It does not remove bias; people decide.

def route_for_review(ranked: list[dict], flags: dict[str, list]) -> None:
    for c in ranked:
        review_queue.add(candidate=c["id"], scores=c["scores"],
                         total=c["total"], flags=flags.get(c["id"], []))
    # The agent NEVER auto-rejects or auto-hires. It recommends; a person decides.

def on_human_decision(candidate_id: str, decision: str, reviewer: str) -> None:
    # the only place an outcome is set, and only by a named person
    audit.log("decision", candidate_id, {"decision": decision, "by": reviewer})
# No code path closes out a candidate without a human. Recommendations only.

def log_screening(candidate: dict, rubric: list, flags: list) -> None:
    audit.write({
        "candidate_id": candidate["id"],
        "rubric": rubric,                      # the exact criteria and weights used
        "scores": candidate["scores"],         # per-criterion score + reason
        "total": candidate["total"],
        "guardrail_flags": flags,
        "model": "gemini-2.5-flash",
        "at": now(),
    })   # immutable; the human decision is appended later, with the reviewer's name
# Every score, reason, and flag is recorded, so any shortlist can be explained.

from google.adk.agents import SequentialAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

screen_pipeline = SequentialAgent(name="screen_pipeline",
    sub_agents=[extractor, scorer, guardrail])   # each an LlmAgent with the tools

runner = Runner(agent=screen_pipeline, app_name="screening",
                session_service=InMemorySessionService())

def guard(app: dict) -> str | None:
    if store.seen(app["id"]):       return "already_screened"   # idempotent
    if PROTECTED & set(app.get("profile", {})):
        return "redaction_failed"   # never score with protected attributes present
    return None
# Run on each new batch via cron / Cloud Scheduler. The agent recommends; people decide.