""" AI Recruitment Matching Agent — Gradio 4.16.0 UI Run: python gradio_app.py """ import os import asyncio import uuid import io import json import threading from typing import List, Optional from dotenv import load_dotenv load_dotenv() import pandas as pd import gradio as gr from app.models.schemas import Candidate, EvaluationResponse from app.services.evaluation_service import perform_hybrid_evaluation # ───────────────────────────────────────────────────────────────────────────── # Helpers # ───────────────────────────────────────────────────────────────────────────── VERDICT_EMOJI = { "strong hire": "🟢", "hire": "🟡", "consider": "🟠", "reject": "🔴", } DECISION_COLOR = { "strong hire": "#22c55e", "hire": "#eab308", "consider": "#f97316", "reject": "#ef4444", } SAMPLE_JD = """Backend Engineer — SaaS Platform We are seeking a Backend Engineer to design and build the core infrastructure of our SaaS platform. The role involves developing scalable microservices, building APIs, and managing IoT data pipelines. Core Requirements: - Minimum 3 years of experience in backend development - Strong proficiency in Node.js - Experience with FastAPI, Django, or Express - Strong understanding of RESTful APIs and microservices - Experience with relational and/or NoSQL databases Preferred: - Experience with AWS, GCP, or Azure - Docker, Kubernetes, CI/CD pipelines - Redis, Kafka or RabbitMQ - Startup experience Skills: Backend Engineer, Node.js, AWS, Microservices, IoT, SaaS, Serverless, API Development""" def parse_csv_to_candidates(filepath: str) -> tuple[List[Candidate], pd.DataFrame, str]: """Parse uploaded file (CSV/XLSX) into Candidate objects.""" try: # ✅ Handle CSV + encoding fallback if filepath.endswith(".csv"): try: df = pd.read_csv(filepath, encoding="utf-8") except UnicodeDecodeError: df = pd.read_csv(filepath, encoding="latin-1") # ✅ Handle Excel elif filepath.endswith(".xlsx"): df = pd.read_excel(filepath) else: return [], pd.DataFrame(), "Unsupported file type. Use CSV or XLSX." df = df.fillna("") candidates = [] # Smart column detection col_map = {col.lower().strip(): col for col in df.columns} def get_col(candidates_list): for c in candidates_list: if c in col_map: return col_map[c] return None name_col = get_col(["name", "full_name", "candidate_name"]) email_col = get_col(["email", "email_address"]) skills_col = get_col(["skills", "parsed_skills", "technical_skills"]) exp_col = get_col(["experience", "parsed_work_experience", "work_experience", "years_of_experience"]) proj_col = get_col(["projects", "parsed_projects"]) edu_col = get_col(["education", "parsed_metadata_education", "education_status"]) resume_col = get_col(["resume_text", "parsed_summary", "summary", "resume"]) for _, row in df.iterrows(): candidates.append(Candidate( id=str(uuid.uuid4()), name=str(row[name_col]) if name_col else "Unknown", email=str(row[email_col]) if email_col else "", skills=str(row[skills_col]) if skills_col else "", experience=str(row[exp_col]) if exp_col else "", projects=str(row[proj_col]) if proj_col else "", education=str(row[edu_col]) if edu_col else "", resume_text=str(row[resume_col]) if resume_col else "", )) return candidates, df, "" except Exception as e: return [], pd.DataFrame(), f"Error parsing file: {e}" def build_shortlist_table(response: EvaluationResponse) -> pd.DataFrame: rows = [] for rank in response.shortlist: detail = response.details.get(rank.candidate_id, {}) emoji = VERDICT_EMOJI.get(rank.decision.lower(), "⚪") rows.append({ "Rank": rank.rank, "Name": rank.name, "Decision": f"{emoji} {rank.decision.title()}", "Confidence": f"{int(detail.get('confidence', 0) * 100)}%", "Why": rank.reason, "Strengths": " | ".join(detail.get("strengths", [])), "Risks": " | ".join(detail.get("risks", [])), "Signal": detail.get("hidden_signal", ""), }) return pd.DataFrame(rows) def build_detail_md(response: EvaluationResponse, shortlist_df: pd.DataFrame) -> str: md_parts = [] for rank in response.shortlist: detail = response.details.get(rank.candidate_id, {}) emoji = VERDICT_EMOJI.get((detail.get("verdict") or rank.decision).lower(), "⚪") verdict = (detail.get("verdict") or rank.decision).title() confidence_pct = int(detail.get("confidence", 0) * 100) md_parts.append(f""" ### {rank.rank}. {rank.name} {emoji} {verdict} **Why:** {detail.get("why", rank.reason)} **Confidence:** {confidence_pct}% **Strengths:** {chr(10).join(f"- {s}" for s in detail.get("strengths", []))} **Risks:** {chr(10).join(f"- {r}" for r in detail.get("risks", []))} **Hidden Signal:** _{detail.get("hidden_signal", "—")}_ --- """) return "\n".join(md_parts) if md_parts else "_No results yet._" # ───────────────────────────────────────────────────────────────────────────── # Core async runner # ───────────────────────────────────────────────────────────────────────────── def run_evaluation_sync(jd: str, candidates: List[Candidate], log_queue: list): """Run async pipeline in a thread-safe way.""" def progress_cb(msg: str): log_queue.append(msg) loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) try: result = loop.run_until_complete( perform_hybrid_evaluation(jd, candidates, progress_cb=progress_cb) ) return result, None except Exception as e: return None, str(e) finally: loop.close() # ───────────────────────────────────────────────────────────────────────────── # Gradio App # ───────────────────────────────────────────────────────────────────────────── CSS = """ /* ── Root & Typography ── */ @import url('https://fonts.googleapis.com/css2?family=IBM+Plex+Mono:wght@400;600&family=Syne:wght@400;700;800&display=swap'); :root { --bg: #0a0a0f; --surface: #12121a; --border: #1e1e2e; --accent: #6ee7b7; --accent2: #818cf8; --warn: #fbbf24; --danger: #f87171; --text: #e2e8f0; --muted: #64748b; --radius: 8px; } body, .gradio-container { background: var(--bg) !important; font-family: 'Syne', sans-serif !important; color: var(--text) !important; } /* Header */ .app-header { background: linear-gradient(135deg, #0f172a 0%, #1e1b4b 50%, #0f172a 100%); border-bottom: 1px solid var(--accent2); padding: 24px 32px; margin-bottom: 0; } .app-header h1 { font-family: 'Syne', sans-serif; font-weight: 800; font-size: 2rem; color: var(--accent); margin: 0; letter-spacing: -0.5px; } .app-header p { color: var(--muted); font-family: 'IBM Plex Mono', monospace; font-size: 0.78rem; margin: 4px 0 0; } /* Panels */ .panel { background: var(--surface); border: 1px solid var(--border); border-radius: var(--radius); padding: 20px; } /* Labels */ label span { font-family: 'IBM Plex Mono', monospace !important; font-size: 0.72rem !important; color: var(--accent2) !important; text-transform: uppercase; letter-spacing: 0.08em; } /* Textboxes */ textarea, input[type="text"] { background: #0d0d16 !important; border: 1px solid var(--border) !important; border-radius: var(--radius) !important; color: var(--text) !important; font-family: 'IBM Plex Mono', monospace !important; font-size: 0.82rem !important; } textarea:focus, input:focus { border-color: var(--accent2) !important; box-shadow: 0 0 0 2px rgba(129, 140, 248, 0.15) !important; } /* Buttons */ button.primary { background: linear-gradient(135deg, #4f46e5 0%, #7c3aed 100%) !important; color: white !important; border: none !important; border-radius: var(--radius) !important; font-family: 'Syne', sans-serif !important; font-weight: 700 !important; font-size: 0.95rem !important; padding: 12px 28px !important; transition: all 0.2s ease !important; letter-spacing: 0.02em; } button.primary:hover { transform: translateY(-1px) !important; box-shadow: 0 4px 20px rgba(124, 58, 237, 0.4) !important; } button.secondary { background: transparent !important; border: 1px solid var(--border) !important; color: var(--muted) !important; border-radius: var(--radius) !important; font-family: 'IBM Plex Mono', monospace !important; font-size: 0.8rem !important; } button.secondary:hover { border-color: var(--accent2) !important; color: var(--accent2) !important; } /* Log box */ .log-box textarea { font-family: 'IBM Plex Mono', monospace !important; font-size: 0.75rem !important; color: var(--accent) !important; background: #050508 !important; border-color: #1a1a2e !important; line-height: 1.6; } /* Dataframe */ .dataframe th { background: #1a1a2e !important; color: var(--accent2) !important; font-family: 'IBM Plex Mono', monospace !important; font-size: 0.72rem !important; text-transform: uppercase; letter-spacing: 0.06em; } .dataframe td { font-family: 'IBM Plex Mono', monospace !important; font-size: 0.8rem !important; color: var(--text) !important; border-color: var(--border) !important; } /* Status badge */ .status-badge { display: inline-flex; align-items: center; gap: 6px; padding: 4px 12px; border-radius: 20px; font-family: 'IBM Plex Mono', monospace; font-size: 0.75rem; font-weight: 600; } /* Tabs */ .tab-nav button { font-family: 'IBM Plex Mono', monospace !important; font-size: 0.8rem !important; color: var(--muted) !important; border-bottom: 2px solid transparent !important; background: transparent !important; } .tab-nav button.selected { color: var(--accent) !important; border-bottom-color: var(--accent) !important; } /* Markdown output */ .markdown-body { font-family: 'Syne', sans-serif; color: var(--text); line-height: 1.7; } .markdown-body h3 { color: var(--accent2); font-size: 1.05rem; margin-top: 24px; } .markdown-body strong { color: var(--accent); } .markdown-body hr { border-color: var(--border); } /* Pipeline steps */ .pipeline-step { display: inline-block; padding: 3px 10px; margin: 2px; border-radius: 4px; font-family: 'IBM Plex Mono', monospace; font-size: 0.7rem; background: #1a1a2e; color: var(--accent2); border: 1px solid #2d2d5e; } /* Accent divider */ .divider { height: 2px; background: linear-gradient(90deg, var(--accent2), transparent); margin: 16px 0; border: none; } """ def create_app(): with gr.Blocks( css=CSS, title="AI Recruitment Agent", theme=gr.themes.Base( primary_hue="violet", neutral_hue="slate", ), ) as app: # ── State ────────────────────────────────────────────── candidates_state = gr.State([]) response_state = gr.State(None) # ── Header ───────────────────────────────────────────── gr.HTML("""

⚡ AI Recruitment Agent

5-stage hybrid pipeline · Groq LLM · Pinecone embeddings · Deterministic reranking

① Normalize → ② Embed → ③ Rerank → ④ Deep Review → ⑤ Shortlist
""") # ── Main Layout ──────────────────────────────────────── with gr.Row(equal_height=False): # Left column — inputs with gr.Column(scale=4, min_width=360): gr.HTML('
') # JD input jd_input = gr.Textbox( label="📋 Job Description", placeholder="Paste the full job description here...", lines=14, value=SAMPLE_JD, elem_classes=["panel"], ) gr.HTML('
') # CSV upload csv_upload = gr.File( label="📂 Upload Candidates CSV", file_types=[".csv"], elem_classes=["panel"], ) # Candidate count badge candidate_count = gr.HTML( '
No candidates loaded
' ) gr.HTML('
') # Preview table preview_table = gr.Dataframe( label="👥 Candidate Preview", headers=["Name", "Email", "Skills Preview"], datatype=["str", "str", "str"], visible=False, wrap=True, elem_classes=["panel"], ) gr.HTML('
') # Action buttons with gr.Row(): run_btn = gr.Button( "🚀 Run Evaluation", variant="primary", scale=3, ) clear_btn = gr.Button( "↺ Reset", variant="secondary", scale=1, ) # Right column — outputs with gr.Column(scale=6, min_width=500): gr.HTML('
') with gr.Tabs(elem_classes=["tab-nav"]): # Tab 1 — Live Log with gr.Tab("📡 Live Pipeline Log"): log_output = gr.Textbox( label="", lines=18, interactive=False, placeholder="Pipeline logs will appear here...", elem_classes=["log-box"], ) # Tab 2 — Results Table with gr.Tab("🏆 Shortlist"): status_html = gr.HTML( '
Run evaluation to see results.
' ) results_table = gr.Dataframe( label="Final Shortlist", wrap=True, elem_classes=["panel"], ) # Tab 3 — Deep Reviews with gr.Tab("🔍 Deep Reviews"): detail_output = gr.Markdown( value="_Run evaluation to see candidate deep reviews._", ) # Tab 4 — Raw JSON with gr.Tab("{ } Raw JSON"): raw_json_output = gr.Code( language="json", label="Full API Response", lines=30, ) # ── Event Handlers ────────────────────────────────────── def on_csv_upload(file): if file is None: return ( [], '
No candidates loaded
', gr.update(visible=False), pd.DataFrame(), ) candidates, df, err = parse_csv_to_candidates(file.name) if err: return ( [], f'
⚠ {err}
', gr.update(visible=False), pd.DataFrame(), ) count = len(candidates) badge_color = "#22c55e" if count > 0 else "#f87171" badge = f'
✓ {count} candidates loaded from CSV
' # Build preview preview_rows = [] for c in candidates[:10]: skills_preview = (c.skills or "")[:80] + ("..." if len(c.skills or "") > 80 else "") preview_rows.append([c.name, c.email or "—", skills_preview]) preview_df = pd.DataFrame(preview_rows, columns=["Name", "Email", "Skills Preview"]) return candidates, badge, gr.update(visible=True), preview_df csv_upload.change( fn=on_csv_upload, inputs=[csv_upload], outputs=[candidates_state, candidate_count, preview_table, preview_table], ) def on_run(jd: str, candidates: list): if not jd.strip(): yield ( "⚠ Please enter a Job Description.", gr.update(), "_No results yet._", "{}", '
Job description required.
', None, ) return if not candidates: yield ( "⚠ Please upload a CSV file with candidates first.", gr.update(), "_No results yet._", "{}", '
No candidates loaded.
', None, ) return log_queue = [] result_holder = [None] error_holder = [None] # Run in thread def run(): res, err = run_evaluation_sync(jd, candidates, log_queue) result_holder[0] = res error_holder[0] = err thread = threading.Thread(target=run) thread.start() # Stream logs while running import time last_log_len = 0 while thread.is_alive(): time.sleep(0.5) if len(log_queue) > last_log_len: last_log_len = len(log_queue) log_text = "\n".join(log_queue) yield ( log_text, gr.update(), "_Processing..._", "{}", '
⏳ Evaluating candidates...
', None, ) thread.join() final_logs = "\n".join(log_queue) if error_holder[0]: yield ( final_logs + f"\n\n❌ ERROR: {error_holder[0]}", gr.update(), "_Evaluation failed._", "{}", f'
❌ {error_holder[0]}
', None, ) return response: EvaluationResponse = result_holder[0] # Build outputs shortlist_df = build_shortlist_table(response) detail_md = build_detail_md(response, shortlist_df) raw_json = json.dumps(response.model_dump(), indent=2) n = len(response.shortlist) top = response.shortlist[0] if response.shortlist else None top_name = top.name if top else "—" top_decision = top.decision if top else "—" emoji = VERDICT_EMOJI.get((top_decision or "").lower(), "⚪") status = f'''
✓ Evaluation complete · {n} candidates shortlisted
Top pick: {top_name} {emoji}
''' yield ( final_logs + "\n\n✅ Evaluation complete.", shortlist_df, detail_md, raw_json, status, response, ) run_btn.click( fn=on_run, inputs=[jd_input, candidates_state], outputs=[ log_output, results_table, detail_output, raw_json_output, status_html, response_state, ], ) def on_clear(): return ( [], SAMPLE_JD, None, "", pd.DataFrame(), "_No results yet._", "{}", '
No candidates loaded
', gr.update(visible=False), pd.DataFrame(), '
Run evaluation to see results.
', ) clear_btn.click( fn=on_clear, outputs=[ candidates_state, jd_input, csv_upload, log_output, results_table, detail_output, raw_json_output, candidate_count, preview_table, preview_table, status_html, ], ) # ── Footer ───────────────────────────────────────────── gr.HTML("""
AI Recruitment Agent · Groq + Pinecone + SentenceTransformers · Gradio 4.16.0
""") return app if __name__ == "__main__": share = os.getenv("GRADIO_SHARE", "false").lower() == "true" port = int(os.getenv("GRADIO_PORT", "7860")) print(f"\n{'='*50}") print(" AI Recruitment Agent") print(f" Starting on http://0.0.0.0:{port}") print(f" Public share: {share}") print(f"{'='*50}\n") app = create_app() app.queue().launch( server_name="0.0.0.0", server_port=port, share=share, show_error=True, )