feat: add gap analysis, LLM explainer via Groq, and all API routes

backend/src/matching/llm_explainer.py

@@ -0,0 +1,84 @@
+from groq import AsyncGroq
+from ..config import get_settings
+
+
+SYSTEM_PROMPT = """You are a senior technical recruiter with deep engineering knowledge.
+You analyze candidate profiles against job descriptions and provide concise, honest assessments.
+Always ground your analysis in the specific data provided — never hallucinate skills or experience.
+Structure your response in exactly two parts:
+1. FIT ANALYSIS (2-3 sentences on why the candidate is or isn't a good match)
+2. KEY GAPS (1-2 sentences addressing specific missing requirements)"""
+
+
+def _build_prompt(jd: dict, candidate: dict, gaps: list[dict]) -> str:
+    gaps_text = ""
+    if gaps:
+        gap_lines = []
+        for g in gaps:
+            gap_lines.append(f"- {g['type'].replace('_', ' ').title()}: {g['detail']}")
+        gaps_text = "Pre-computed gaps:\n" + "\n".join(gap_lines)
+
+    skills_text = ", ".join(
+        (candidate.get("programming_languages") or [])
+        + (candidate.get("backend_frameworks") or [])
+        + (candidate.get("frontend_technologies") or [])
+    )
+
+    return f"""JOB DESCRIPTION:
+Title: {jd.get('title', 'Unknown')}
+Required Skills: {', '.join(jd.get('required_skills') or [])}
+Min Experience: {jd.get('min_yoe', 'Not specified')} years
+Engineer Type: {jd.get('engineer_type', 'Not specified')}
+Location: {jd.get('location', 'Not specified')}
+
+CANDIDATE PROFILE:
+Summary: {candidate.get('parsed_summary') or 'Not provided'}
+Skills: {candidate.get('parsed_skills') or skills_text or 'Not provided'}
+Experience: {candidate.get('years_of_experience', 'Unknown')} years
+Company: {candidate.get('most_recent_company') or 'Not provided'}
+Growth Velocity Score: {candidate.get('growth_velocity', 0.5):.2f} / 1.0
+
+{gaps_text}
+
+Provide your assessment:"""
+
+
+async def generate_explanation(jd: dict, candidate: dict, gaps: list[dict]) -> str:
+    settings = get_settings()
+    client = AsyncGroq(api_key=settings.groq_api_key)
+
+    try:
+        response = await client.chat.completions.create(
+            model=settings.groq_model,
+            messages=[
+                {"role": "system", "content": SYSTEM_PROMPT},
+                {"role": "user", "content": _build_prompt(jd, candidate, gaps)},
+            ],
+            temperature=0.3,
+            max_tokens=400,
+        )
+        return response.choices[0].message.content.strip()
+    except Exception:
+        return _fallback_explanation(jd, candidate, gaps)
+
+
+def _fallback_explanation(jd: dict, candidate: dict, gaps: list[dict]) -> str:
+    skill_gaps = [g["detail"] for g in gaps if g["type"] == "missing_skill"]
+    yoe_gaps = [g["detail"] for g in gaps if g["type"] == "yoe_gap"]
+
+    parts = []
+    skills_text = ", ".join(
+        (candidate.get("programming_languages") or [])[:5]
+    )
+    if skills_text:
+        parts.append(f"FIT ANALYSIS: Candidate brings experience in {skills_text}")
+    else:
+        parts.append("FIT ANALYSIS: Candidate profile available for review.")
+
+    if skill_gaps or yoe_gaps:
+        gap_list = (skill_gaps[:3] + yoe_gaps)[:3]
+        parts.append(f"KEY GAPS: Missing requirements include: {', '.join(gap_list)}.")
+    else:
+        parts.append("KEY GAPS: No critical gaps identified from structured data.")
+
+    return "\n".join(parts)

backend/src/routers/__init__.py

@@ -0,0 +1,5 @@
+from .jds import router as jds_router
+from .candidates import router as candidates_router
+from .matching import router as matching_router
+
+__all__ = ["jds_router", "candidates_router", "matching_router"]

backend/src/routers/candidates.py

@@ -0,0 +1,89 @@
+import io
+import json
+import pandas as pd
+from fastapi import APIRouter, Depends, HTTPException, UploadFile, File
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, func
+
+from ..database import get_db
+from ..models.candidate import Candidate
+from ..schemas.candidate import CandidateResponse, UploadResponse, TaskStatusResponse
+from ..workers.celery_app import celery_app
+from ..workers.ingest import ingest_candidates_batch
+
+router = APIRouter()
+
+BATCH_SIZE = 100
+
+
+@router.post("/upload", response_model=UploadResponse)
+async def upload_candidates(file: UploadFile = File(...)):
+    content = await file.read()
+    filename = file.filename or ""
+
+    if filename.endswith(".csv"):
+        df = pd.read_csv(io.BytesIO(content))
+        rows = df.where(pd.notna(df), None).to_dict(orient="records")
+    elif filename.endswith(".json") or filename.endswith(".jsonl"):
+        try:
+            data = json.loads(content)
+            rows = data if isinstance(data, list) else [data]
+        except json.JSONDecodeError:
+            rows = [json.loads(line) for line in content.decode().splitlines() if line.strip()]
+    else:
+        raise HTTPException(status_code=400, detail="Only CSV or JSON files are supported")
+
+    if not rows:
+        raise HTTPException(status_code=400, detail="File contains no data rows")
+
+    task_ids = []
+    for i in range(0, len(rows), BATCH_SIZE):
+        batch = rows[i: i + BATCH_SIZE]
+        serializable = []
+        for r in batch:
+            clean = {}
+            for k, v in r.items():
+                if v is None or (isinstance(v, float) and pd.isna(v)):
+                    clean[k] = None
+                else:
+                    clean[k] = v
+            serializable.append(clean)
+        task = ingest_candidates_batch.delay(serializable)
+        task_ids.append(task.id)
+
+    return UploadResponse(
+        task_id=task_ids[0] if task_ids else "",
+        queued=len(rows),
+        message=f"Queued {len(rows)} candidates across {len(task_ids)} batches",
+    )
+
+
+@router.get("/status/{task_id}", response_model=TaskStatusResponse)
+async def task_status(task_id: str):
+    result = celery_app.AsyncResult(task_id)
+    return TaskStatusResponse(
+        task_id=task_id,
+        status=result.status,
+        result=result.result if result.ready() else None,
+    )
+
+
+@router.get("/count")
+async def candidate_count(db: AsyncSession = Depends(get_db)):
+    result = await db.execute(select(func.count()).select_from(Candidate))
+    count = result.scalar()
+    return {"count": count}
+
+
+@router.get("/{candidate_id}", response_model=CandidateResponse)
+async def get_candidate(candidate_id: str, db: AsyncSession = Depends(get_db)):
+    import uuid as _uuid
+    try:
+        cid = _uuid.UUID(candidate_id)
+    except ValueError:
+        raise HTTPException(status_code=400, detail="Invalid candidate ID")
+    result = await db.execute(select(Candidate).where(Candidate.id == cid))
+    cand = result.scalar_one_or_none()
+    if not cand:
+        raise HTTPException(status_code=404, detail="Candidate not found")
+    return cand

backend/src/routers/jds.py

@@ -0,0 +1,43 @@
+import uuid
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select
+
+from ..database import get_db
+from ..models.jd import JobDescription
+from ..schemas.jd import JDCreate, JDResponse, JDListItem
+from ..workers.ingest import ingest_jd
+
+router = APIRouter()
+
+
+@router.post("", response_model=JDResponse, status_code=201)
+async def create_jd(payload: JDCreate, db: AsyncSession = Depends(get_db)):
+    jd = JobDescription(
+        id=uuid.uuid4(),
+        title=payload.title,
+        raw_text=payload.raw_text,
+        status="processing",
+    )
+    db.add(jd)
+    await db.commit()
+    await db.refresh(jd)
+
+    ingest_jd.delay(str(jd.id), payload.raw_text, payload.title)
+
+    return jd
+
+
+@router.get("", response_model=list[JDListItem])
+async def list_jds(db: AsyncSession = Depends(get_db)):
+    result = await db.execute(select(JobDescription).order_by(JobDescription.created_at.desc()).limit(50))
+    return result.scalars().all()
+
+
+@router.get("/{jd_id}", response_model=JDResponse)
+async def get_jd(jd_id: uuid.UUID, db: AsyncSession = Depends(get_db)):
+    result = await db.execute(select(JobDescription).where(JobDescription.id == jd_id))
+    jd = result.scalar_one_or_none()
+    if not jd:
+        raise HTTPException(status_code=404, detail="JD not found")
+    return jd

backend/src/routers/matching.py

@@ -0,0 +1,318 @@
+import uuid
+from datetime import datetime, timezone
+from fastapi import APIRouter, Depends, HTTPException, Request
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy import select, delete
+
+from ..database import get_db
+from ..models.jd import JobDescription
+from ..models.candidate import Candidate
+from ..models.match_result import MatchResult
+from ..schemas.match import MatchResponse, MatchedCandidate, ComponentScores, GapItem, CandidateDetailResponse, ReRankRequest
+from ..matching.stage1 import stage1_retrieve
+from ..matching.stage2 import stage2_rerank
+from ..matching.llm_explainer import generate_explanation
+from ..matching.scorer import rerank_with_weights
+
+router = APIRouter()
+
+
+def _get_qdrant(request: Request):
+    return request.app.state.qdrant
+
+
+async def _load_jd(jd_id: uuid.UUID, db: AsyncSession) -> JobDescription:
+    result = await db.execute(select(JobDescription).where(JobDescription.id == jd_id))
+    jd = result.scalar_one_or_none()
+    if not jd:
+        raise HTTPException(status_code=404, detail="JD not found")
+    if jd.status == "processing":
+        raise HTTPException(status_code=202, detail="JD is still being processed, try again shortly")
+    return jd
+
+
+@router.post("/{jd_id}", response_model=MatchResponse)
+async def trigger_match(
+    jd_id: uuid.UUID,
+    request: Request,
+    db: AsyncSession = Depends(get_db),
+):
+    jd = await _load_jd(jd_id, db)
+    qdrant = _get_qdrant(request)
+
+    jd_dict = {
+        "id": str(jd.id),
+        "title": jd.title,
+        "raw_text": jd.raw_text,
+        "required_skills": jd.required_skills or [],
+        "min_yoe": jd.min_yoe,
+        "max_yoe": jd.max_yoe,
+        "role_type": jd.role_type,
+        "engineer_type": jd.engineer_type,
+        "location": jd.location,
+        "remote_allowed": jd.remote_allowed,
+    }
+
+    shortlist = await stage1_retrieve(jd_dict, db, qdrant)
+    final_ranked = await stage2_rerank(jd_dict, shortlist)
+
+    await db.execute(delete(MatchResult).where(MatchResult.jd_id == jd_id))
+
+    match_records = []
+    for i, item in enumerate(final_ranked):
+        mr = MatchResult(
+            id=uuid.uuid4(),
+            jd_id=jd_id,
+            candidate_id=uuid.UUID(item["candidate_id"]),
+            rank=i + 1,
+            stage1_score=item.get("stage1_score", 0),
+            stage2_score=item.get("stage2_score"),
+            final_score=item.get("final_score", 0),
+            component_scores=item.get("component_scores", {}),
+            gaps=item.get("gaps", []),
+        )
+        match_records.append(mr)
+        db.add(mr)
+
+    await db.commit()
+
+    results = []
+    for i, item in enumerate(final_ranked):
+        results.append(
+            MatchedCandidate(
+                candidate_id=uuid.UUID(item["candidate_id"]),
+                rank=i + 1,
+                name=item.get("name"),
+                email=item.get("email"),
+                role_type=item.get("role_type"),
+                engineer_type=item.get("engineer_type"),
+                years_of_experience=item.get("years_of_experience"),
+                most_recent_company=item.get("most_recent_company"),
+                parsed_summary=item.get("parsed_summary"),
+                programming_languages=item.get("programming_languages") or [],
+                growth_velocity=item.get("growth_velocity", 0.5),
+                stage1_score=item.get("stage1_score", 0),
+                stage2_score=item.get("stage2_score"),
+                final_score=item.get("final_score", 0),
+                component_scores=ComponentScores(**item.get("component_scores", {})),
+                gaps=[GapItem(**g) for g in item.get("gaps", [])],
+            )
+        )
+
+    return MatchResponse(
+        jd_id=jd_id,
+        jd_title=jd.title,
+        jd_quality=jd.jd_quality or {},
+        total_matched=len(results),
+        results=results,
+        weights_used={"semantic": 0.20, "skill": 0.35, "yoe": 0.15, "company": 0.10, "growth": 0.10, "education": 0.10},
+    )
+
+
+@router.get("/{jd_id}", response_model=MatchResponse)
+async def get_match_results(jd_id: uuid.UUID, db: AsyncSession = Depends(get_db)):
+    jd = await _load_jd(jd_id, db)
+
+    result = await db.execute(
+        select(MatchResult, Candidate)
+        .join(Candidate, MatchResult.candidate_id == Candidate.id)
+        .where(MatchResult.jd_id == jd_id)
+        .order_by(MatchResult.rank)
+    )
+    rows = result.all()
+
+    if not rows:
+        raise HTTPException(status_code=404, detail="No match results found. Run POST /api/match/{jd_id} first.")
+
+    results = []
+    for mr, cand in rows:
+        results.append(
+            MatchedCandidate(
+                candidate_id=cand.id,
+                rank=mr.rank or 0,
+                name=cand.name,
+                email=cand.email,
+                role_type=cand.role_type,
+                engineer_type=cand.engineer_type,
+                years_of_experience=cand.years_of_experience,
+                most_recent_company=cand.most_recent_company,
+                parsed_summary=cand.parsed_summary,
+                programming_languages=cand.programming_languages or [],
+                growth_velocity=cand.growth_velocity,
+                stage1_score=mr.stage1_score,
+                stage2_score=mr.stage2_score,
+                final_score=mr.final_score,
+                component_scores=ComponentScores(**(mr.component_scores or {})),
+                gaps=[GapItem(**g) for g in (mr.gaps or [])],
+            )
+        )
+
+    return MatchResponse(
+        jd_id=jd_id,
+        jd_title=jd.title,
+        jd_quality=jd.jd_quality or {},
+        total_matched=len(results),
+        results=results,
+    )
+
+
+@router.get("/{jd_id}/{candidate_id}", response_model=CandidateDetailResponse)
+async def get_candidate_detail(
+    jd_id: uuid.UUID,
+    candidate_id: uuid.UUID,
+    db: AsyncSession = Depends(get_db),
+):
+    jd = await _load_jd(jd_id, db)
+
+    mr_result = await db.execute(
+        select(MatchResult).where(
+            MatchResult.jd_id == jd_id,
+            MatchResult.candidate_id == candidate_id,
+        )
+    )
+    mr = mr_result.scalar_one_or_none()
+    if not mr:
+        raise HTTPException(status_code=404, detail="Match result not found for this JD/candidate pair")
+
+    cand_result = await db.execute(select(Candidate).where(Candidate.id == candidate_id))
+    cand = cand_result.scalar_one_or_none()
+    if not cand:
+        raise HTTPException(status_code=404, detail="Candidate not found")
+
+    if not mr.explanation:
+        jd_dict = {
+            "id": str(jd.id),
+            "title": jd.title,
+            "raw_text": jd.raw_text,
+            "required_skills": jd.required_skills or [],
+            "min_yoe": jd.min_yoe,
+            "engineer_type": jd.engineer_type,
+            "location": jd.location,
+            "remote_allowed": jd.remote_allowed,
+        }
+        cand_dict = {
+            "parsed_summary": cand.parsed_summary,
+            "parsed_skills": cand.parsed_skills,
+            "years_of_experience": cand.years_of_experience,
+            "programming_languages": cand.programming_languages or [],
+            "backend_frameworks": cand.backend_frameworks or [],
+            "frontend_technologies": cand.frontend_technologies or [],
+            "most_recent_company": cand.most_recent_company,
+            "growth_velocity": cand.growth_velocity,
+        }
+        explanation = await generate_explanation(jd_dict, cand_dict, mr.gaps or [])
+        mr.explanation = explanation
+        mr.explanation_generated_at = datetime.now(timezone.utc)
+        await db.commit()
+
+    return CandidateDetailResponse(
+        jd_id=jd_id,
+        candidate_id=candidate_id,
+        rank=mr.rank,
+        final_score=mr.final_score,
+        component_scores=ComponentScores(**(mr.component_scores or {})),
+        gaps=[GapItem(**g) for g in (mr.gaps or [])],
+        explanation=mr.explanation,
+        candidate={
+            "name": cand.name,
+            "email": cand.email,
+            "role_type": cand.role_type,
+            "engineer_type": cand.engineer_type,
+            "years_of_experience": cand.years_of_experience,
+            "most_recent_company": cand.most_recent_company,
+            "parsed_summary": cand.parsed_summary,
+            "parsed_skills": cand.parsed_skills,
+            "parsed_work_experience": cand.parsed_work_experience or [],
+            "programming_languages": cand.programming_languages or [],
+            "backend_frameworks": cand.backend_frameworks or [],
+            "gen_ai_experience": cand.gen_ai_experience,
+            "growth_velocity": cand.growth_velocity,
+            "looking_for": cand.looking_for,
+            "open_to_working_at": cand.open_to_working_at,
+            "is_actively_or_passively_looking": cand.is_actively_or_passively_looking,
+            "most_recent_company_is_funded": cand.most_recent_company_is_funded,
+            "most_recent_company_is_product_company": cand.most_recent_company_is_product_company,
+            "most_recent_company_total_funding": cand.most_recent_company_total_funding,
+        },
+        jd={
+            "title": jd.title,
+            "required_skills": jd.required_skills or [],
+            "min_yoe": jd.min_yoe,
+            "role_type": jd.role_type,
+            "engineer_type": jd.engineer_type,
+            "location": jd.location,
+        },
+    )
+
+
+@router.post("/{jd_id}/rerank", response_model=MatchResponse)
+async def rerank_results(
+    jd_id: uuid.UUID,
+    payload: ReRankRequest,
+    db: AsyncSession = Depends(get_db),
+):
+    jd = await _load_jd(jd_id, db)
+
+    result = await db.execute(
+        select(MatchResult, Candidate)
+        .join(Candidate, MatchResult.candidate_id == Candidate.id)
+        .where(MatchResult.jd_id == jd_id)
+        .order_by(MatchResult.rank)
+    )
+    rows = result.all()
+
+    if not rows:
+        raise HTTPException(status_code=404, detail="No match results found.")
+
+    items = []
+    for mr, cand in rows:
+        items.append({
+            "candidate_id": str(cand.id),
+            "name": cand.name,
+            "email": cand.email,
+            "role_type": cand.role_type,
+            "engineer_type": cand.engineer_type,
+            "years_of_experience": cand.years_of_experience,
+            "most_recent_company": cand.most_recent_company,
+            "parsed_summary": cand.parsed_summary,
+            "programming_languages": cand.programming_languages or [],
+            "growth_velocity": cand.growth_velocity,
+            "stage1_score": mr.stage1_score,
+            "stage2_score": mr.stage2_score,
+            "final_score": mr.final_score,
+            "component_scores": mr.component_scores or {},
+            "gaps": mr.gaps or [],
+        })
+
+    reranked = rerank_with_weights(items, payload.weights)
+
+    results = [
+        MatchedCandidate(
+            candidate_id=uuid.UUID(item["candidate_id"]),
+            rank=item["rank"],
+            name=item.get("name"),
+            email=item.get("email"),
+            role_type=item.get("role_type"),
+            engineer_type=item.get("engineer_type"),
+            years_of_experience=item.get("years_of_experience"),
+            most_recent_company=item.get("most_recent_company"),
+            parsed_summary=item.get("parsed_summary"),
+            programming_languages=item.get("programming_languages") or [],
+            growth_velocity=item.get("growth_velocity", 0.5),
+            stage1_score=item.get("stage1_score", 0),
+            stage2_score=item.get("stage2_score"),
+            final_score=item.get("final_score", 0),
+            component_scores=ComponentScores(**(item.get("component_scores") or {})),
+            gaps=[GapItem(**g) for g in item.get("gaps", [])],
+        )
+        for item in reranked
+    ]
+
+    return MatchResponse(
+        jd_id=jd_id,
+        jd_title=jd.title,
+        jd_quality=jd.jd_quality or {},
+        total_matched=len(results),
+        results=results,
+        weights_used=payload.weights,
+    )

backend/src/schemas/__init__.py

@@ -0,0 +1 @@
+# schemas package

backend/src/schemas/candidate.py

@@ -0,0 +1,37 @@
+from uuid import UUID
+from typing import Any
+from pydantic import BaseModel
+
+
+class CandidateResponse(BaseModel):
+    id: UUID
+    name: str | None = None
+    email: str | None = None
+    role_type: str | None = None
+    engineer_type: str | None = None
+    years_of_experience: float | None = None
+    most_recent_company: str | None = None
+    parsed_summary: str | None = None
+    parsed_skills: str | None = None
+    programming_languages: list[str] = []
+    backend_frameworks: list[str] = []
+    frontend_technologies: list[str] = []
+    growth_velocity: float = 0.5
+    gen_ai_experience: bool | None = None
+    looking_for: str | None = None
+    open_to_working_at: str | None = None
+    is_actively_or_passively_looking: str | None = None
+
+    model_config = {"from_attributes": True}
+
+
+class UploadResponse(BaseModel):
+    task_id: str
+    queued: int
+    message: str
+
+
+class TaskStatusResponse(BaseModel):
+    task_id: str
+    status: str
+    result: Any | None = None

backend/src/schemas/jd.py

@@ -0,0 +1,35 @@
+from uuid import UUID
+from datetime import datetime
+from typing import Any
+from pydantic import BaseModel, Field
+
+
+class JDCreate(BaseModel):
+    title: str
+    raw_text: str
+
+
+class JDResponse(BaseModel):
+    id: UUID
+    title: str
+    raw_text: str
+    status: str
+    min_yoe: float | None = None
+    role_type: str | None = None
+    engineer_type: str | None = None
+    location: str | None = None
+    required_skills: list[str] = []
+    jd_quality: dict[str, Any] = {}
+    created_at: datetime
+
+    model_config = {"from_attributes": True}
+
+
+class JDListItem(BaseModel):
+    id: UUID
+    title: str
+    status: str
+    jd_quality: dict[str, Any] = {}
+    created_at: datetime
+
+    model_config = {"from_attributes": True}

backend/src/schemas/match.py

@@ -0,0 +1,71 @@
+from uuid import UUID
+from typing import Any
+from pydantic import BaseModel, Field
+
+
+class ComponentScores(BaseModel):
+    semantic: float = 0.0
+    skill: float = 0.0
+    yoe: float = 0.0
+    company: float = 0.0
+    growth: float = 0.0
+    education: float = 0.0
+
+
+class GapItem(BaseModel):
+    type: str
+    detail: str
+    mitigated_by_remote: bool | None = None
+
+
+class MatchedCandidate(BaseModel):
+    candidate_id: UUID
+    rank: int
+    name: str | None = None
+    email: str | None = None
+    role_type: str | None = None
+    engineer_type: str | None = None
+    years_of_experience: float | None = None
+    most_recent_company: str | None = None
+    parsed_summary: str | None = None
+    programming_languages: list[str] = []
+    growth_velocity: float = 0.5
+    stage1_score: float
+    stage2_score: float | None = None
+    final_score: float
+    component_scores: ComponentScores
+    gaps: list[GapItem] = []
+
+
+class MatchResponse(BaseModel):
+    jd_id: UUID
+    jd_title: str
+    jd_quality: dict[str, Any] = {}
+    total_matched: int
+    results: list[MatchedCandidate]
+    weights_used: dict[str, float] = {}
+
+
+class CandidateDetailResponse(BaseModel):
+    jd_id: UUID
+    candidate_id: UUID
+    rank: int | None = None
+    final_score: float
+    component_scores: ComponentScores
+    gaps: list[GapItem] = []
+    explanation: str | None = None
+    candidate: dict[str, Any] = {}
+    jd: dict[str, Any] = {}
+
+
+class ReRankRequest(BaseModel):
+    weights: dict[str, float] = Field(
+        default={
+            "semantic": 0.20,
+            "skill": 0.35,
+            "yoe": 0.15,
+            "company": 0.10,
+            "growth": 0.10,
+            "education": 0.10,
+        }
+    )