AI в HR: рекрутинг, скрининг и онбординг с нейросетями

import anthropic
import json
from pathlib import Path


RESUME_PARSER_PROMPT = """Извлеки структурированные данные из резюме.

Ответь строго JSON:
{{
  "name": "Имя Фамилия",
  "email": "email@example.com",
  "phone": "+7...",
  "location": "Город",
  "experience_years": N,
  "current_role": "текущая должность",
  "skills": {{
    "languages": ["Python", "JavaScript"],
    "frameworks": ["FastAPI", "React"],
    "databases": ["PostgreSQL"],
    "tools": ["Docker", "Git"],
    "soft_skills": ["лидерство", "коммуникация"]
  }},
  "experience": [
    {{
      "company": "Компания",
      "role": "Должность",
      "period": "2023-2025",
      "highlights": ["ключевое достижение 1", "достижение 2"]
    }}
  ],
  "education": [
    {{
      "institution": "Вуз",
      "degree": "Степень/специальность",
      "year": 2020
    }}
  ],
  "languages": ["русский (родной)", "английский (B2)"]
}}

Резюме:
{resume_text}"""


def parse_resume(resume_text: str) -> dict:
    """Parse resume into structured data using LLM."""
    client = anthropic.Anthropic()

    response = client.messages.create(
        model="claude-haiku-4-20250514",
        max_tokens=2048,
        messages=[{
            "role": "user",
            "content": RESUME_PARSER_PROMPT.format(
                resume_text=resume_text,
            ),
        }],
    )

    text = response.content[0].text
    start = text.index("{")
    end = text.rindex("}") + 1
    return json.loads(text[start:end])

SCORING_PROMPT = """Оцени кандидата по соответствию вакансии.

Вакансия:
{job_description}

Данные кандидата (JSON):
{candidate_json}

Оцени по критериям (каждый 0-10):
1. HARD_SKILLS: соответствие технических навыков требованиям
2. EXPERIENCE: релевантность и длительность опыта
3. SENIORITY: соответствие уровню позиции
4. CULTURE_FIT: soft skills, язык, локация
5. GROWTH_POTENTIAL: карьерная траектория, обучаемость

Ответь JSON:
{{
  "scores": {{
    "hard_skills": N,
    "experience": N,
    "seniority": N,
    "culture_fit": N,
    "growth_potential": N
  }},
  "total": N,
  "match_percentage": N,
  "must_have_met": ["навык1", "навык2"],
  "must_have_missing": ["навык3"],
  "nice_to_have_met": ["навык4"],
  "strengths": ["сильная сторона 1"],
  "concerns": ["потенциальная проблема 1"],
  "recommendation": "INVITE | MAYBE | REJECT",
  "summary": "Краткое обоснование в 2-3 предложениях"
}}"""


def score_candidate(
    job_description: str,
    candidate_data: dict,
    model: str = "claude-sonnet-4-20250514",
) -> dict:
    """Score candidate against job description."""
    client = anthropic.Anthropic()

    response = client.messages.create(
        model=model,
        max_tokens=1024,
        system=(
            "Ты объективный HR-аналитик. Оценивай только навыки "
            "и опыт. Игнорируй имя, пол, возраст, национальность, "
            "фото. Фокусируйся на skills-match и потенциале."
        ),
        messages=[{
            "role": "user",
            "content": SCORING_PROMPT.format(
                job_description=job_description,
                candidate_json=json.dumps(candidate_data, ensure_ascii=False),
            ),
        }],
    )

    text = response.content[0].text
    start = text.index("{")
    end = text.rindex("}") + 1
    return json.loads(text[start:end])


def rank_candidates(
    job_description: str,
    candidates: list[dict],
) -> list[dict]:
    """Score and rank multiple candidates."""
    scored = []
    for candidate in candidates:
        result = score_candidate(job_description, candidate)
        result["candidate_name"] = candidate.get("name", "N/A")
        scored.append(result)

    scored.sort(key=lambda x: x["total"], reverse=True)
    return scored

import statistics


def bias_audit(
    job_description: str,
    scoring_fn,
    test_resumes: list[dict],
) -> dict:
    """Audit AI scoring for demographic bias.

    Each test resume should have 'demographic_group' field
    with values like 'male', 'female', 'group_a', 'group_b'.
    """
    group_scores: dict[str, list[float]] = {}

    for resume in test_resumes:
        group = resume.get("demographic_group", "unknown")
        result = scoring_fn(job_description, resume)
        score = result["total"]

        if group not in group_scores:
            group_scores[group] = []
        group_scores[group].append(score)

    # Calculate metrics per group
    report = {"groups": {}, "disparate_impact": {}}
    for group, scores in group_scores.items():
        report["groups"][group] = {
            "count": len(scores),
            "mean": round(statistics.mean(scores), 2),
            "median": round(statistics.median(scores), 2),
            "stdev": round(statistics.stdev(scores), 2) if len(scores) > 1 else 0,
        }

    # Four-Fifths Rule (disparate impact analysis)
    groups = list(report["groups"].keys())
    max_rate = max(g["mean"] for g in report["groups"].values())

    for group in groups:
        group_mean = report["groups"][group]["mean"]
        ratio = group_mean / max_rate if max_rate > 0 else 1.0
        report["disparate_impact"][group] = {
            "ratio": round(ratio, 3),
            "passes_four_fifths": ratio >= 0.8,
        }

    report["overall_fair"] = all(
        d["passes_four_fifths"]
        for d in report["disparate_impact"].values()
    )

    return report

from dataclasses import dataclass


@dataclass
class ScreeningConfig:
    job_description: str
    must_have_skills: list[str]
    nice_to_have_skills: list[str]
    min_experience_years: int
    invite_threshold: float = 7.0  # min score to invite
    maybe_threshold: float = 5.0   # min score for "maybe"


def screening_pipeline(
    config: ScreeningConfig,
    resumes: list[str],
) -> dict:
    """Full screening pipeline: parse → score → rank → report."""
    # Step 1: Parse all resumes
    parsed = []
    for resume_text in resumes:
        try:
            data = parse_resume(resume_text)
            parsed.append(data)
        except (json.JSONDecodeError, ValueError):
            parsed.append({"error": "Failed to parse", "raw": resume_text[:200]})

    # Step 2: Pre-filter by must-have (quick check before LLM scoring)
    qualified = []
    disqualified = []
    for candidate in parsed:
        if "error" in candidate:
            disqualified.append(candidate)
            continue

        exp = candidate.get("experience_years", 0)
        if exp < config.min_experience_years:
            candidate["rejection_reason"] = "Insufficient experience"
            disqualified.append(candidate)
            continue

        qualified.append(candidate)

    # Step 3: AI scoring for qualified candidates
    ranked = rank_candidates(config.job_description, qualified)

    # Step 4: Categorize
    invite = [c for c in ranked if c["total"] >= config.invite_threshold]
    maybe = [c for c in ranked if config.maybe_threshold <= c["total"] < config.invite_threshold]
    reject = [c for c in ranked if c["total"] < config.maybe_threshold]

    return {
        "total_resumes": len(resumes),
        "parsed_successfully": len(parsed) - len([p for p in parsed if "error" in p]),
        "invite": invite,
        "maybe": maybe,
        "reject": reject,
        "disqualified_early": disqualified,
        "summary": {
            "invite_count": len(invite),
            "maybe_count": len(maybe),
            "reject_count": len(reject),
            "disqualified_count": len(disqualified),
        },
    }