coderound-bkl / app.py
cloud450's picture
Update app.py
112ba22 verified
"""
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("""
<div class="app-header">
<h1>⚡ AI Recruitment Agent</h1>
<p>5-stage hybrid pipeline · Groq LLM · Pinecone embeddings · Deterministic reranking</p>
</div>
<div style="display:flex; gap:8px; padding:12px 32px; background:#0c0c14; border-bottom:1px solid #1e1e2e;">
<span class="pipeline-step">① Normalize</span>
<span style="color:#64748b;align-self:center">→</span>
<span class="pipeline-step">② Embed</span>
<span style="color:#64748b;align-self:center">→</span>
<span class="pipeline-step">③ Rerank</span>
<span style="color:#64748b;align-self:center">→</span>
<span class="pipeline-step">④ Deep Review</span>
<span style="color:#64748b;align-self:center">→</span>
<span class="pipeline-step">⑤ Shortlist</span>
</div>
""")
# ── Main Layout ────────────────────────────────────────
with gr.Row(equal_height=False):
# Left column — inputs
with gr.Column(scale=4, min_width=360):
gr.HTML('<div style="height:16px"></div>')
# 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('<div style="height:12px"></div>')
# CSV upload
csv_upload = gr.File(
label="📂 Upload Candidates CSV",
file_types=[".csv"],
elem_classes=["panel"],
)
# Candidate count badge
candidate_count = gr.HTML(
'<div style="color:#64748b; font-family:\'IBM Plex Mono\',monospace; font-size:0.75rem; padding:6px 0;">No candidates loaded</div>'
)
gr.HTML('<div style="height:12px"></div>')
# 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('<div style="height:16px"></div>')
# 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('<div style="height:16px"></div>')
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(
'<div style="color:#64748b;font-family:\'IBM Plex Mono\',monospace;font-size:0.8rem;padding:8px 0;">Run evaluation to see results.</div>'
)
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 (
[],
'<div style="color:#64748b;font-family:\'IBM Plex Mono\',monospace;font-size:0.75rem;padding:6px 0;">No candidates loaded</div>',
gr.update(visible=False),
pd.DataFrame(),
)
candidates, df, err = parse_csv_to_candidates(file.name)
if err:
return (
[],
f'<div style="color:#f87171;font-family:\'IBM Plex Mono\',monospace;font-size:0.75rem;padding:6px 0;">⚠ {err}</div>',
gr.update(visible=False),
pd.DataFrame(),
)
count = len(candidates)
badge_color = "#22c55e" if count > 0 else "#f87171"
badge = f'<div style="color:{badge_color};font-family:\'IBM Plex Mono\',monospace;font-size:0.75rem;padding:6px 0;">✓ {count} candidates loaded from CSV</div>'
# 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._",
"{}",
'<div style="color:#f87171;font-size:0.8rem;">Job description required.</div>',
None,
)
return
if not candidates:
yield (
"⚠ Please upload a CSV file with candidates first.",
gr.update(),
"_No results yet._",
"{}",
'<div style="color:#f87171;font-size:0.8rem;">No candidates loaded.</div>',
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..._",
"{}",
'<div style="color:#818cf8;font-size:0.8rem;font-family:\'IBM Plex Mono\',monospace;">⏳ Evaluating candidates...</div>',
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'<div style="color:#f87171;font-size:0.8rem;">❌ {error_holder[0]}</div>',
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'''
<div style="display:flex;gap:16px;align-items:center;padding:8px 0;">
<div style="color:#22c55e;font-family:'IBM Plex Mono',monospace;font-size:0.8rem;">
✓ Evaluation complete · {n} candidates shortlisted
</div>
<div style="color:#64748b;font-family:'IBM Plex Mono',monospace;font-size:0.8rem;">
Top pick: <span style="color:#e2e8f0">{top_name}</span> {emoji}
</div>
</div>
'''
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._",
"{}",
'<div style="color:#64748b;font-family:\'IBM Plex Mono\',monospace;font-size:0.75rem;padding:6px 0;">No candidates loaded</div>',
gr.update(visible=False),
pd.DataFrame(),
'<div style="color:#64748b;font-size:0.8rem;font-family:\'IBM Plex Mono\',monospace;">Run evaluation to see results.</div>',
)
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("""
<div style="text-align:center;padding:20px;color:#334155;font-family:'IBM Plex Mono',monospace;font-size:0.7rem;border-top:1px solid #1e1e2e;margin-top:24px;">
AI Recruitment Agent · Groq + Pinecone + SentenceTransformers · Gradio 4.16.0
</div>
""")
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,
)