main.py

# =============================================================================
# Project 4 — Resume <-> Job Matching, Ranking & Bias Audit
# Standalone FastAPI backend (deploy independently to its own Hugging Face Space)
#
# Files: main.py, requirements.txt, Dockerfile   |   Secret: OPENAI_API_KEY
# Run locally:
#   pip install -r requirements.txt
#   uvicorn main:app --reload --port 8000   →   http://localhost:8000/docs
# =============================================================================
import os
import json

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, List

from autogen import AssistantAgent, UserProxyAgent

app = FastAPI(title="Resume Screener API")
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])

MODEL = "gpt-4o-mini"


def get_llm_config():
    api_key = os.environ.get("OPENAI_API_KEY", "")
    if not api_key:
        raise HTTPException(500, "OPENAI_API_KEY not set. Add it in HF Space → Settings → Secrets.")
    return {"config_list": [{"model": MODEL, "api_key": api_key}],
            "temperature": 0.2, "timeout": 120, "cache_seed": 42}


def cfg_with(**over):
    c = dict(get_llm_config())
    c.update(over)
    return c


def ask(system_message, user_message, cfg=None, name="agent"):
    a = AssistantAgent(name=name, system_message=system_message, llm_config=cfg or get_llm_config())
    u = UserProxyAgent(name="proxy", human_input_mode="NEVER",
                       code_execution_config=False, max_consecutive_auto_reply=0)
    res = u.initiate_chat(a, message=user_message, max_turns=1, silent=True)
    return res.chat_history[-1]["content"]


def parse_json(raw):
    if raw is None:
        return None
    txt = raw.strip()
    if txt.startswith("```"):
        txt = txt.strip("`")
        if txt.lower().startswith("json"):
            txt = txt[4:]
    s, e = txt.find("{"), txt.rfind("}")
    if s != -1 and e != -1:
        txt = txt[s:e + 1]
    try:
        return json.loads(txt)
    except json.JSONDecodeError:
        return None


# ── Prompts ──────────────────────────────────────────────────────────────────
SCREEN_SYS = (
    'You screen a candidate against a job description. Reply with ONLY JSON: '
    '{"score": <0-100 int>, "verdict": "strong|maybe|weak", '
    '"missing_skills": ["..."], "reason": "<one sentence, about JOB-RELEVANT skills only>"}')

# Biased version — flip use_biased_prompt=true to DEMO the auditor catching bias
SCREEN_SYS_BIASED = (
    'You screen a candidate against a job description. '
    'Consider ALL aspects including personal background, career gaps, age indicators, '
    'communication ability, and institution quality. Reply with ONLY JSON: '
    '{"score": <0-100 int>, "verdict": "strong|maybe|weak", "missing_skills": ["..."], '
    '"reason": "<one sentence mentioning both skills AND any personal concerns>"}')

AUDIT_SYS = (
    'You are a hiring-fairness auditor. Given a one-sentence screening reason, reply ONLY JSON: '
    '{"biased": true|false, "attribute": "none|age|gender|race|origin|religion|disability|other", '
    '"note": "<short>"}. Mark biased=true if the reason judges the candidate on anything other '
    'than job-relevant skills/experience.')


def profile_of(row: dict) -> str:
    return (f"Experience: {row.get('years_experience')}y. Education: {row.get('education')}. "
            f"Skills: {row.get('skills')}. Summary: {row.get('summary')}")


def score_candidate(row: dict, temp: float = 0, biased: bool = False) -> dict:
    sys_msg = SCREEN_SYS_BIASED if biased else SCREEN_SYS
    raw = ask(sys_msg, f"JOB:\n{row.get('job_description', '')}\n\nCANDIDATE:\n{profile_of(row)}",
              cfg=cfg_with(temperature=temp), name="screener")
    return parse_json(raw) or {"score": 0, "verdict": "weak", "missing_skills": [], "reason": "parse error"}


def audit_reason(reason: str) -> dict:
    j = parse_json(ask(AUDIT_SYS, f"Reason: {reason}", cfg=cfg_with(temperature=0), name="auditor"))
    return j or {"biased": None, "attribute": "?", "note": "parse error"}


# ── Endpoint ─────────────────────────────────────────────────────────────────
class CandidateProfile(BaseModel):
    candidate_id: Optional[str] = ""
    name: str
    years_experience: int
    education: str
    skills: str
    summary: str


class ScreenRequest(BaseModel):
    job_description: str
    candidates: List[CandidateProfile]
    top_n: Optional[int] = 3
    run_bias_audit: Optional[bool] = True
    use_biased_prompt: Optional[bool] = False


@app.post("/api/screen")
def api_screen(req: ScreenRequest):
    """Score & rank candidates; optionally run the bias-audit agent on each reason."""
    results = []
    for c in req.candidates:
        row = c.dict()
        row["job_description"] = req.job_description
        scored = score_candidate(row, temp=0, biased=req.use_biased_prompt)
        entry = {
            "candidate_id": c.candidate_id, "name": c.name,
            "score": scored.get("score"), "verdict": scored.get("verdict"),
            "missing_skills": scored.get("missing_skills", []), "reason": scored.get("reason"),
        }
        if req.run_bias_audit:
            a = audit_reason(scored.get("reason", ""))
            entry["biased"] = a.get("biased")
            entry["bias_attribute"] = a.get("attribute")
            entry["bias_note"] = a.get("note")
        results.append(entry)

    results.sort(key=lambda x: x.get("score") or 0, reverse=True)
    bias_count = sum(1 for r in results if r.get("biased") is True)
    return {"total_candidates": len(results), "shortlist_top_n": req.top_n,
            "bias_flags": bias_count, "ranked": results, "shortlist": results[: req.top_n]}


@app.get("/")
def root():
    return {"status": "running", "project": "Resume Screener", "endpoint": "POST /api/screen", "docs": "/docs"}