Spaces:
Running
Running
| """Interview intelligence: tracks, question generation, answer evaluation, follow-ups, session summaries.""" | |
| import json | |
| import logging | |
| import os | |
| import tempfile | |
| from typing import Dict, List, Optional | |
| import librosa | |
| import numpy as np | |
| import process_interview as pi | |
| logger = logging.getLogger("evalbot.ai") | |
| # -------------------------------------------------------------------------------------- | |
| # Catalog | |
| # -------------------------------------------------------------------------------------- | |
| TRACKS = [ | |
| # id, English, Arabic, group | |
| ("data_engineering", "Data Engineering", "هندسة البيانات", "tech"), | |
| ("data_analysis", "Data Analysis / BI", "تحليل البيانات", "tech"), | |
| ("data_science", "Data Science", "علم البيانات", "tech"), | |
| ("ai_ml", "AI / Machine Learning", "الذكاء الاصطناعي وتعلم الآلة", "tech"), | |
| ("backend", "Backend Development", "تطوير الـ Backend", "tech"), | |
| ("frontend", "Frontend Development", "تطوير الـ Frontend", "tech"), | |
| ("fullstack", "Full-Stack Development", "تطوير Full-Stack", "tech"), | |
| ("mobile", "Mobile Development (Android / iOS / Flutter)", "تطوير تطبيقات الموبايل", "tech"), | |
| ("devops_cloud", "DevOps & Cloud", "DevOps والحوسبة السحابية", "tech"), | |
| ("cybersecurity", "Cybersecurity", "الأمن السيبراني", "tech"), | |
| ("qa_testing", "Software Testing / QA", "اختبار البرمجيات", "tech"), | |
| ("embedded", "Embedded Systems & IoT", "الأنظمة المدمجة وإنترنت الأشياء", "tech"), | |
| ("networking", "Networking & IT Infrastructure", "الشبكات والبنية التحتية", "tech"), | |
| ("it_support", "IT Support", "الدعم الفني", "tech"), | |
| ("ui_ux", "UI / UX Design", "تصميم واجهات وتجربة المستخدم", "design"), | |
| ("graphic_design", "Graphic Design", "التصميم الجرافيكي", "design"), | |
| ("product_management", "Product Management", "إدارة المنتجات", "business"), | |
| ("project_management", "Project Management", "إدارة المشروعات", "business"), | |
| ("business_analysis", "Business Analysis", "تحليل الأعمال", "business"), | |
| ("marketing", "Marketing & Digital Marketing", "التسويق والتسويق الرقمي", "business"), | |
| ("sales", "Sales & Business Development", "المبيعات وتطوير الأعمال", "business"), | |
| ("hr", "Human Resources", "الموارد البشرية", "business"), | |
| ("finance", "Accounting & Finance", "المحاسبة والمالية", "business"), | |
| ("customer_service", "Customer Service", "خدمة العملاء", "business"), | |
| ("operations", "Operations & Supply Chain", "العمليات وسلاسل الإمداد", "business"), | |
| ("content", "Content Writing & Communication", "كتابة المحتوى والتواصل", "business"), | |
| ("teaching", "Teaching & Training", "التدريس والتدريب", "other"), | |
| ("engineering", "Engineering (Civil / Mechanical / Electrical)", "الهندسة (مدني / ميكانيكا / كهرباء)", "other"), | |
| ("healthcare", "Healthcare", "الرعاية الصحية", "other"), | |
| ("general", "General / Any role", "عام / أي وظيفة", "other"), | |
| ] | |
| TRACK_NAMES = {t[0]: t[1] for t in TRACKS} | |
| LEVELS = {"intern": "Internship", "junior": "Junior (0-2 years)", "mid": "Mid-level (2-5 years)", | |
| "senior": "Senior (5+ years)", "lead": "Lead / Manager"} | |
| TYPES = {"mixed": "a balanced mix of technical, behavioral and motivation questions", | |
| "technical": "mostly technical / role-specific knowledge questions, plus one opening question", | |
| "behavioral": "behavioral questions (STAR) about past experiences, teamwork, conflict, failure, leadership", | |
| "hr": "HR / fit questions: motivation, career goals, culture fit, strengths and weaknesses, salary expectations", | |
| "case": "problem-solving / case / scenario questions where the candidate must reason out loud"} | |
| def catalog() -> Dict: | |
| return {"tracks": [{"id": i, "en": en, "ar": ar, "group": g} for i, en, ar, g in TRACKS], | |
| "levels": list(LEVELS), "types": list(TYPES)} | |
| def role_name(track: str, custom_role: str = "") -> str: | |
| return (custom_role or "").strip()[:80] or TRACK_NAMES.get(track, "General / Any role") | |
| # -------------------------------------------------------------------------------------- | |
| # Question generation | |
| # -------------------------------------------------------------------------------------- | |
| QUESTION_SCHEMA = { | |
| "type": "object", | |
| "properties": {"questions": {"type": "array", "items": {"type": "object", "properties": { | |
| "text": {"type": "string"}, | |
| "kind": {"type": "string", "enum": ["opening", "technical", "behavioral", "situational", "motivation", "closing"]}, | |
| "skill": {"type": "string", "description": "the competency this question tests, 1-3 words"}, | |
| "rubric": {"type": "string", "description": "what a strong answer includes, 1-2 sentences"}}, | |
| "required": ["text", "kind", "skill", "rubric"]}}}, | |
| "required": ["questions"], | |
| } | |
| def generate_questions(track: str, level: str, itype: str, count: int = 6, custom_role: str = "", | |
| cv_text: str = "", jd_text: str = "") -> List[Dict]: | |
| role = role_name(track, custom_role) | |
| ctx = "" | |
| if jd_text.strip(): | |
| ctx += f"\nJob description (tailor questions to it):\n{jd_text.strip()[:4000]}\n" | |
| if cv_text.strip(): | |
| ctx += f"\nCandidate CV (ask about their real experience where relevant):\n{cv_text.strip()[:4000]}\n" | |
| prompt = f"""You are an expert interviewer. Write {count} spoken interview questions for this interview: | |
| - Role: {role} | |
| - Level: {LEVELS.get(level, level)} | |
| - Interview style: {TYPES.get(itype, TYPES['mixed'])} | |
| {ctx} | |
| Rules: | |
| - Questions are asked out loud, so keep each one to one or two clear sentences; no code to read, no lists. | |
| - Calibrate difficulty to the level. Technical questions must be specific to the role (real tools, concepts, trade-offs). | |
| - Each question must be answerable in about 1-2 minutes of speaking. | |
| - Start with a short opening question unless the style is purely technical; avoid near-duplicates. | |
| - Write the questions in English. | |
| """ | |
| res = pi._gemini_json(prompt, QUESTION_SCHEMA, temperature=0.8) | |
| qs = [q for q in (res or {}).get("questions", []) if q.get("text", "").strip()][:count] | |
| if not qs: # offline fallback: built-in bank | |
| bank = pi.QUESTION_BANK["General"] + pi.QUESTION_BANK["Behavioral"] | |
| qs = [{"text": t, "kind": "behavioral", "skill": "communication", "rubric": ""} for t in bank[:count]] | |
| return qs | |
| # -------------------------------------------------------------------------------------- | |
| # Answer evaluation (one recorded answer) | |
| # -------------------------------------------------------------------------------------- | |
| SOFT_SKILLS = ["communication", "teamwork", "conflict_resolution", "time_management", "adaptability", "leadership", | |
| "problem_solving", "emotional_intelligence"] | |
| ANSWER_SCHEMA = { | |
| "type": "object", | |
| "properties": { | |
| "score": {"type": "integer", "description": "0-10"}, | |
| "relevance": {"type": "string", "enum": ["on_topic", "partially", "off_topic"]}, | |
| "summary": {"type": "string"}, | |
| "strengths": {"type": "array", "items": {"type": "string"}}, | |
| "improvements": {"type": "array", "items": {"type": "string"}}, | |
| "delivery_tip": {"type": "string"}, | |
| "improved_answer": {"type": "string"}, | |
| "followup_question": {"type": "string", "description": "a natural probing follow-up, or empty string if not needed"}, | |
| "rubric": {"type": "object", "properties": { | |
| "clarity": {"type": "integer", "description": "0-10, how clearly the thoughts are articulated"}, | |
| "example_quality": {"type": "integer", "description": "0-10, relevance and depth of real examples"}, | |
| "structure": {"type": "integer", "description": "0-10, logical flow (STAR where it fits)"}, | |
| "outcome": {"type": "integer", "description": "0-10, a concrete result, resolution or learning"}}, | |
| "required": ["clarity", "example_quality", "structure", "outcome"]}, | |
| "soft_skills": {"type": "array", "items": {"type": "string", "enum": SOFT_SKILLS}, | |
| "description": "soft skills this answer gives real evidence of (may be empty)"}, | |
| }, | |
| "required": ["score", "relevance", "summary", "strengths", "improvements", "delivery_tip", "improved_answer", | |
| "followup_question", "rubric", "soft_skills"], | |
| } | |
| def transcribe_answer(audio_bytes: bytes, suffix: str = ".webm") -> Dict: | |
| """Transcribe one answer and compute its voice metrics. Returns transcript, words, duration, voice.""" | |
| with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tmp: | |
| tmp.write(audio_bytes) | |
| src = tmp.name | |
| wav = None | |
| try: | |
| wav = pi.convert_to_wav(src) | |
| y, sr = librosa.load(wav, sr=16000) | |
| if len(y) < sr: | |
| return {"transcript": "", "duration_s": len(y) / sr, "voice": {"error": "too short"}, "silent": True} | |
| tr = pi.transcribe(wav, speakers_expected=None, diarize=True) | |
| words = tr.get("words") or [] | |
| text = (tr.get("text") or "").strip() | |
| if not words or not text: | |
| return {"transcript": "", "duration_s": len(y) / sr, "voice": {"error": "no speech"}, "silent": True} | |
| start, end = words[0]["start"], words[-1]["end"] | |
| utt = {"role": "Interviewee", "start": start, "end": end, "text": text, | |
| "prosodic_features": pi.extract_prosodic_features(y, sr, start, end)} | |
| voice = pi.analyze_interviewee_voice(y, sr, [utt]) | |
| # integrity: a second voice speaking a substantial part of the answer | |
| counts: Dict[str, int] = {} | |
| for w in words: | |
| if w.get("speaker"): | |
| counts[w["speaker"]] = counts.get(w["speaker"], 0) + 1 | |
| ranked = sorted(counts.values(), reverse=True) | |
| other = sum(ranked[1:]) | |
| voice["speakers"] = len(counts) or 1 | |
| voice["second_voice"] = bool(other >= 10 and other / max(1, len(words)) >= 0.25) | |
| return {"transcript": text, "duration_s": round((end - start) / 1000, 1), "voice": voice, "silent": False, | |
| "speech_start_ms": start, "speech_end_ms": end} | |
| finally: | |
| for p in (src, wav): | |
| if p and os.path.exists(p): | |
| os.remove(p) | |
| def evaluate_answer(question: Dict, transcript: str, voice: Dict, role: str, level: str, lang: str, | |
| allow_followup: bool) -> Dict: | |
| metrics = {k: voice.get(k) for k in ("speaking_rate", "filler_ratio", "interpretation")} if "error" not in voice else {} | |
| prompt = f"""You are a fair, encouraging but honest interview coach. A candidate for "{role}" ({LEVELS.get(level, level)}) | |
| answered this interview question out loud. Evaluate the answer. | |
| Question: {question['text']} | |
| What a strong answer includes: {question.get('rubric') or 'n/a'} | |
| Answer (speech-to-text transcript): {transcript} | |
| Voice metrics: {json.dumps(metrics)} | |
| - score 0-10 for content quality: relevance, correctness, depth, structure (STAR where it fits), specificity. | |
| - 2-4 strengths and 2-4 improvements, one sentence each, concrete and tied to what they said. | |
| - delivery_tip: one tip from the voice metrics. | |
| - improved_answer: a stronger version (4-6 sentences, first person) that keeps the candidate's own facts and uses | |
| placeholders like [number] for missing details. Keep it in English. | |
| - rubric: clarity, example_quality, structure and outcome, each 0-10 (for purely technical answers, judge example_quality | |
| by the concreteness of the examples and outcome by whether the answer reaches a clear conclusion). | |
| - soft_skills: which of {", ".join(SOFT_SKILLS)} the answer shows real evidence of (empty list if none). | |
| - followup_question: {"one short probing follow-up a real interviewer would ask about something specific the candidate said (in English), or an empty string if the answer was complete" if allow_followup else "always an empty string"}. | |
| {pi._language_rule(lang)}""" | |
| res = pi._gemini_json(prompt, ANSWER_SCHEMA) | |
| if not res: | |
| return {"score": None, "relevance": None, "summary": "", "strengths": [], "improvements": [], | |
| "delivery_tip": "", "improved_answer": "", "followup_question": "", "ai_generated": False} | |
| res["score"] = int(max(0, min(10, res.get("score", 0)))) | |
| res["ai_generated"] = True | |
| if not allow_followup: | |
| res["followup_question"] = "" | |
| return res | |
| # -------------------------------------------------------------------------------------- | |
| # Session summary | |
| # -------------------------------------------------------------------------------------- | |
| SUMMARY_SCHEMA = { | |
| "type": "object", | |
| "properties": { | |
| "headline": {"type": "string", "description": "one-sentence overall verdict"}, | |
| "executive_summary": {"type": "array", "items": {"type": "string"}}, | |
| "strengths": {"type": "array", "items": {"type": "object", "properties": { | |
| "competency": {"type": "string"}, "explanation": {"type": "string"}, "evidence": {"type": "string"}}, | |
| "required": ["competency", "explanation", "evidence"]}}, | |
| "weaknesses": {"type": "array", "items": {"type": "object", "properties": { | |
| "area": {"type": "string"}, "feedback": {"type": "string"}}, "required": ["area", "feedback"]}}, | |
| "role_fit": {"type": "array", "items": {"type": "string"}}, | |
| "recommendations": {"type": "array", "items": {"type": "object", "properties": { | |
| "category": {"type": "string", "enum": ["Communication Skills", "Content Delivery", "Professional Presentation"]}, | |
| "recommendation": {"type": "string"}, "how_to": {"type": "string"}}, | |
| "required": ["category", "recommendation", "how_to"]}}, | |
| "learning_plan": {"type": "array", "items": {"type": "object", "properties": { | |
| "topic": {"type": "string"}, "why": {"type": "string"}, "how": {"type": "string"}}, | |
| "required": ["topic", "why", "how"]}}, | |
| "next_steps": {"type": "array", "items": {"type": "string"}}, | |
| "hiring_signal": {"type": "string", "enum": ["strong_yes", "yes", "maybe", "no"]}, | |
| "soft_skills": {"type": "array", "items": {"type": "object", "properties": { | |
| "skill": {"type": "string", "enum": SOFT_SKILLS}, "score": {"type": "integer", "description": "0-10"}, | |
| "comment": {"type": "string", "description": "one sentence with evidence"}}, | |
| "required": ["skill", "score", "comment"]}, | |
| "description": "only the soft skills the interview gave evidence about"}, | |
| "nonverbal": {"type": "string", "description": "2-3 sentences on non-verbal behaviour from the camera metrics, or empty string if none"}, | |
| }, | |
| "required": ["headline", "executive_summary", "strengths", "weaknesses", "role_fit", "recommendations", | |
| "learning_plan", "next_steps", "hiring_signal", "soft_skills", "nonverbal"], | |
| } | |
| def summarize_session(role: str, level: str, lang: str, answers: List[Dict], voice: Dict, | |
| calibrated: Optional[Dict], for_recruiter: bool, camera: Optional[Dict] = None) -> Optional[Dict]: | |
| qa = "\n\n".join(f"[{i + 1}] Q: {a['question']}\nA: {a.get('transcript') or '(no answer)'}\n" | |
| f"Score: {a.get('score')}/10; rubric: {json.dumps((a.get('feedback') or {}).get('rubric') or {})}" | |
| for i, a in enumerate(answers)) | |
| cam = "" | |
| if camera: | |
| cam = ("Camera metrics (computed on the candidate's device): " + json.dumps( | |
| {k: camera.get(k) for k in ("face_visible", "eye_contact", "smile", "emotions", "dominant_emotion")}) + | |
| "\nUse them only for the 'nonverbal' field, cautiously: expressions are not reliable signs of ability.") | |
| cal = "" | |
| if calibrated: | |
| cal = "Human-calibrated delivery scores (percentile vs 138 rated interviews): " + ", ".join( | |
| f"{t['trait']} {t['percentile']}th" for t in calibrated["traits"]) | |
| vm = {k: voice.get(k) for k in ("speaking_rate", "filler_ratio", "interpretation")} if voice and "error" not in voice else {} | |
| audience = ("The report is read by the hiring team." if for_recruiter | |
| else "The report is read by the candidate, who wants to improve; next_steps are their practice steps.") | |
| prompt = f"""You are EvalBot, a senior interviewer and career coach. Write the final evaluation of this interview. | |
| Role: {role} ({LEVELS.get(level, level)}). {audience} | |
| - Base every point on the answers below (quote briefly as evidence) and the metrics. Do not invent facts. | |
| - 3 bullets executive summary, 3-6 strengths, 3-6 weaknesses, 3 role-fit bullets, exactly 3 recommendations | |
| (one per category), a 3-5 item learning_plan of topics to study for this role, and 3-5 next_steps. | |
| - hiring_signal: your recommendation for a {LEVELS.get(level, level)} {role} position. | |
| - soft_skills: score 0-10 only those of {", ".join(SOFT_SKILLS)} the answers gave evidence about, each with evidence. | |
| - nonverbal: {"2-3 sentences from the camera metrics" if camera else "an empty string (no camera)"}. | |
| {pi._language_rule(lang)} | |
| Voice metrics: {json.dumps(vm)} | |
| {cal} | |
| {cam} | |
| Interview: | |
| {qa} | |
| """ | |
| return pi._gemini_json(prompt, SUMMARY_SCHEMA) | |
| # -------------------------------------------------------------------------------------- | |
| # Session-level audio features for calibrated scores | |
| # -------------------------------------------------------------------------------------- | |
| def session_audio(answer_audio: List[bytes]) -> Optional[tuple]: | |
| """Concatenate answer recordings into one 16 kHz signal with candidate utterances.""" | |
| ys, utts, t = [], [], 0 | |
| for blob in answer_audio: | |
| with tempfile.NamedTemporaryFile(suffix=".audio", delete=False) as tmp: | |
| tmp.write(blob) | |
| src = tmp.name | |
| try: | |
| wav = pi.convert_to_wav(src) | |
| y, sr = librosa.load(wav, sr=16000) | |
| os.remove(wav) | |
| finally: | |
| os.remove(src) | |
| iv = librosa.effects.split(y, top_db=35) | |
| if len(iv): | |
| y = y[iv[0][0]: iv[-1][1]] # trim leading/trailing silence | |
| if len(y) < 16000: | |
| continue | |
| utts.append({"role": "Interviewee", "start": t, "end": t + int(len(y) / 16), "text": ""}) | |
| ys.append(y) | |
| gap = np.zeros(8000, dtype=y.dtype) | |
| ys.append(gap) | |
| t += int(len(y) / 16) + 500 | |
| if not ys: | |
| return None | |
| return np.concatenate(ys), 16000, utts | |