AI Recruiter — Digital Worker for Hiring
AI Recruiter automates full recruitment cycle: job description generation, multi-platform posting, incoming resume screening, interview scheduling, personalized candidate communication, ATS pipeline management. Enables one recruiter to manage significantly larger volume of openings without quality loss.
Job Description Generator
from openai import AsyncOpenAI
from pydantic import BaseModel
client = AsyncOpenAI()
class JobDescription(BaseModel):
title: str
department: str
level: str
responsibilities: list[str]
required_skills: list[str]
preferred_skills: list[str]
about_team: str
compensation_range: str
async def generate_job_description(
hiring_manager_brief: str,
similar_jd_examples: list[str] = None,
) -> str:
"""Generates JD from brief hiring manager input"""
examples_context = ""
if similar_jd_examples:
examples_context = f"\nExamples of similar positions for style:\n{similar_jd_examples[0][:500]}"
response = await client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "system",
"content": f"""You are an experienced HR. Write job descriptions that attract strong candidates.
Avoid: lists of 20 requirements, unrealistic demands, corporate jargon.
Focus: what the person will do, why it's interesting, who we need.{examples_context}"""
}, {
"role": "user",
"content": f"Hiring manager brief:\n{hiring_manager_brief}",
}],
temperature=0.4,
)
return response.choices[0].message.content
Multi-Platform Publishing
class JobBoardPublisher:
async def publish_to_all_boards(self, jd: JobDescription) -> dict:
"""Publishes job to all platforms simultaneously"""
results = await asyncio.gather(
self.publish_hh(jd),
self.publish_avito_rabota(jd),
self.publish_linkedin(jd),
self.publish_superjob(jd),
return_exceptions=True,
)
return {
"hh.ru": results[0] if not isinstance(results[0], Exception) else str(results[0]),
"avito": results[1] if not isinstance(results[1], Exception) else str(results[1]),
"linkedin": results[2] if not isinstance(results[2], Exception) else str(results[2]),
"superjob": results[3] if not isinstance(results[3], Exception) else str(results[3]),
}
async def publish_hh(self, jd: JobDescription) -> str:
response = await hh_api.post(
endpoint="/vacancies",
data={
"name": jd.title,
"area": {"id": "1"}, # Moscow
"description": format_for_hh(jd),
"key_skills": [{"name": s} for s in jd.required_skills[:10]],
"employment": {"id": "full"},
"schedule": {"id": "fullDay"},
}
)
return response["id"]
Candidate Screening and Ranking
class CandidateScreener:
async def screen_batch(
self,
candidates: list[dict],
job_description: JobDescription,
required_skills: list[str],
) -> list[dict]:
"""Parallel candidate screening"""
semaphore = asyncio.Semaphore(10)
async def screen_one(candidate: dict) -> dict:
async with semaphore:
return await self._screen_single(
candidate, job_description, required_skills
)
results = await asyncio.gather(*[screen_one(c) for c in candidates])
return sorted(results, key=lambda x: -x["score"])
async def _screen_single(
self,
candidate: dict,
jd: JobDescription,
required_skills: list[str],
) -> dict:
from pydantic import BaseModel
from typing import Literal
class ScreeningResult(BaseModel):
score: int # 0-100
recommendation: Literal["strong_yes", "yes", "maybe", "no"]
required_skills_match: int # match percentage
experience_match: str
red_flags: list[str]
green_flags: list[str]
personalized_question: str # Interview question
result = await client.beta.chat.completions.parse(
model="gpt-4o",
messages=[{
"role": "system",
"content": f"""Evaluate candidate objectively. Required skills: {required_skills}.
DO NOT assume hidden skills. Consider ONLY explicitly stated experience."""
}, {
"role": "user",
"content": f"Position:\n{jd.title}\n\nResume:\n{candidate['resume_text']}"
}],
response_format=ScreeningResult,
temperature=0,
)
return {
"candidate_id": candidate["id"],
"name": candidate["name"],
"email": candidate["email"],
**result.choices[0].message.parsed.model_dump(),
}
Communication Automation
class CandidateCommunicator:
async def send_invite(self, candidate: dict, vacancy: dict) -> None:
invite_text = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "system",
"content": "Write interview invitation. Tone: respectful, specific. 3-4 sentences."
}, {
"role": "user",
"content": f"Candidate: {candidate['name']}, position: {vacancy['title']}, company: {vacancy['company']}"
}],
)
await email_service.send(
to=candidate["email"],
subject=f"Interview Invitation — {vacancy['title']} at {vacancy['company']}",
body=invite_text.choices[0].message.content + f"\n\nTo schedule: {CALENDLY_URL}",
)
async def send_rejection(self, candidate: dict, reason: str) -> None:
"""Personalized rejection with specific reason"""
rejection = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "system",
"content": "Write polite rejection. No clichés. State specific reason (respectfully). 2-3 sentences."
}, {
"role": "user",
"content": f"Candidate: {candidate['name']}, rejection reason: {reason}"
}],
)
await email_service.send(
to=candidate["email"],
subject="Application Status",
body=rejection.choices[0].message.content,
)
Case Study: IT Company, 30 Open Positions
Situation: 3 recruiters, 30 active openings, 200+ resumes/day, time-to-hire 52 days.
Automated by AI Recruiter:
- Resume parsing from job boards every 2 hours
- Screening and scoring (primary filtering 70%)
- Automatic rejections for irrelevant candidates
- Invites for top 30% with Calendly link
- Reminders to candidates who didn't respond to interview request
Results:
- Time-to-hire: 52 → 31 days
- Response time to candidate: 3.2 days → 40 minutes
- Recruiters focus on: interviews, offers, onboarding
- Concordance (AI vs recruiter on 300 jointly rated): 87%
Timeline
- JD generator and posting: 1–2 weeks
- Screening and ranking: 2–3 weeks
- Communication templates and email integration: 1 week
- ATS integration (HH, Huntflow, etc.): 1–2 weeks
- Total: 5–8 weeks