from groq import AsyncGroq from ..config import get_settings, get_revolving_groq_key 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=get_revolving_groq_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)