Redrob-hackathon / lib /jd_parser.py
Mohit0708's picture
Initial commit
7b833a7
Raw
History Blame Contribute Delete
28.4 kB
"""
lib/jd_parser.py — V5 Universal JD Parser
Replaces the hardcoded jd_requirements.py with a dynamic parser that extracts
structured requirements from ANY job description text. No LLM calls needed —
uses pattern matching, keyword dictionaries, and section detection.
The parsed output drives ALL downstream modules: features, evidence, scoring,
retrieval, and reasoning. Changing the JD means re-running the parser, not
rewriting code.
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from typing import Optional
# ---------------------------------------------------------------------------
# Full JD text — extracted from job_description.docx
# In production, this would be read from a file or API.
# ---------------------------------------------------------------------------
JD_FULL_TEXT = """
Job Description: Senior AI Engineer — Founding Team
Company: Redrob AI (Series A AI-native talent intelligence platform)
Location: Pune/Noida, India (Hybrid) | Open to relocation candidates from Tier-1 Indian cities
Employment Type: Full-time
Experience Required: 5–9 years
We need someone who is simultaneously comfortable with:
Deep technical depth in modern ML systems — embeddings, retrieval, ranking, LLMs, fine-tuning.
Scrappy product-engineering attitude — willing to ship a working ranker in a week.
Own the intelligence layer: ranking, retrieval, and matching systems.
Things you absolutely need:
Production experience with embeddings-based retrieval systems (sentence-transformers, OpenAI embeddings, BGE, E5, or similar) deployed to real users. Handling embedding drift, index refresh, retrieval-quality regression in production.
Production experience with vector databases or hybrid search infrastructure — Pinecone, Weaviate, Qdrant, Milvus, OpenSearch, Elasticsearch, FAISS, or something similar.
Strong Python. We care about code quality.
Hands-on experience designing evaluation frameworks for ranking systems — NDCG, MRR, MAP, offline-to-online correlation, A/B test interpretation.
Things we'd like you to have but won't reject you for:
LLM fine-tuning experience (LoRA, QLoRA, PEFT)
Experience with learning-to-rank models (XGBoost-based or neural)
Prior exposure to HR-tech, recruiting tech, or marketplace products
Background in distributed systems or large-scale inference optimization
Open-source contributions in the AI/ML space
Disqualifiers:
Pure research environments without production deployment.
Recent (under 12 months) LangChain-only AI experience without substantial pre-LLM ML production experience.
Senior engineer who hasn't written production code in the last 18 months (architecture/tech lead drift).
Do NOT want:
Title-chasers switching companies every 1.5 years.
Framework enthusiasts (LangChain tutorials, not systems thinkers).
People who have only worked at consulting firms (TCS, Infosys, Wipro, Accenture, Cognizant, Capgemini, etc.) in their entire career.
People whose primary expertise is computer vision, speech, or robotics without significant NLP/IR exposure.
People whose work has been entirely on closed-source proprietary systems for 5+ years without external validation.
Location: Pune/Noida preferred. Hyderabad, Mumbai, Delhi NCR welcome.
Notice period: sub-30-day preferred. 30+ day notice candidates are still in scope but the bar gets higher.
Ideal candidate:
6-8 years total experience, 4-5 in applied ML/AI roles at product companies (not pure services).
Shipped at least one end-to-end ranking, search, or recommendation system to real users at meaningful scale.
Strong opinions about retrieval (hybrid vs dense), evaluation (offline vs online), and LLM integration (when to fine-tune vs prompt).
Located in or willing to relocate to Noida or Pune.
Active on Redrob platform or has clear signal of being in the job market.
"""
@dataclass
class JDUnderstanding:
"""Structured output of JD parsing. All downstream modules consume this."""
# Core requirements
required_skills: dict[str, list[str]] = field(default_factory=dict)
preferred_skills: dict[str, list[str]] = field(default_factory=dict)
red_flags: dict[str, list[str]] = field(default_factory=dict)
# Experience
yoe_low: int = 5
yoe_high: int = 9
yoe_ideal_low: int = 6
yoe_ideal_high: int = 8
yoe_domain_low: int = 4
yoe_domain_high: int = 5
seniority: str = "senior"
# Domain
domain: str = "unknown"
domain_label: str = "AI/ML Engineer"
# Location
preferred_locations: list[str] = field(default_factory=list)
welcome_locations: list[str] = field(default_factory=list)
# Behavioural expectations
notice_preferred_days: int = 30
requires_product_company: bool = True
requires_production_code: bool = True
# Production evidence vocabulary
production_evidence: list[str] = field(default_factory=list)
# Pre-LLM signals
pre_llm_keywords: list[str] = field(default_factory=list)
pre_llm_cutoff_year: int = 2022
post_llm_markers: list[str] = field(default_factory=list)
# Domain detection keywords
non_target_domains: list[str] = field(default_factory=list)
non_target_rescue: list[str] = field(default_factory=list)
# Title patterns
research_only_titles: list[str] = field(default_factory=list)
architect_titles: list[str] = field(default_factory=list)
bad_title_patterns: list[str] = field(default_factory=list)
consulting_firms: list[str] = field(default_factory=list)
consulting_industries: list[str] = field(default_factory=list)
# Ideal candidate text (for embedding similarity)
ideal_text: str = ""
# Raw JD text
raw_text: str = ""
class JDParser:
"""
Universal JD parser. Extracts structured requirements from JD text
using pattern matching and keyword dictionaries. No LLM needed.
Design: The parser uses a combination of:
1. Section detection (required/preferred/red-flags sections)
2. Keyword extraction with domain knowledge
3. Pattern matching for numbers (years, scale, locations)
4. Domain classification from keyword clusters
"""
# Comprehensive skill→domain mapping for dynamic taxonomy
SKILL_DOMAINS = {
# Search / Retrieval / Ranking
"bm25": "search", "elasticsearch": "search", "opensearch": "search",
"solr": "search", "search ranking": "search", "search relevance": "search",
"query understanding": "search", "information retrieval": "search",
"ranking model": "ranking", "learning to rank": "ranking",
"learning-to-rank": "ranking", "ltr model": "ranking",
"lambdamart": "ranking", "xgboost": "ranking", "neural ranking": "ranking",
"ranking system": "ranking", "recommendation system": "ranking",
"recommender system": "ranking", "collaborative filtering": "ranking",
"click-through": "ranking", "ctr model": "ranking",
# Embeddings / Vectors
"embedding": "embeddings", "sentence-transformers": "embeddings",
"sentence transformers": "embeddings", "openai embedding": "embeddings",
"bge": "embeddings", "e5 embedding": "embeddings", "dense retrieval": "embeddings",
"semantic search": "embeddings", "retrieval-augmented": "embeddings",
"rag pipeline": "embeddings", "rag system": "embeddings",
"text embedding": "embeddings",
# Vector DB / Infrastructure
"pinecone": "vector_db", "weaviate": "vector_db", "qdrant": "vector_db",
"milvus": "vector_db", "faiss": "vector_db", "vector database": "vector_db",
"vector db": "vector_db", "hybrid search": "vector_db",
"hybrid retrieval": "vector_db", "chroma": "vector_db",
"chromadb": "vector_db", "annoy": "vector_db", "scaNN": "vector_db",
# LLM / Fine-tuning
"lora": "llm", "qlora": "llm", "peft": "llm", "fine-tun": "llm",
"finetun": "llm", "langchain": "llm", "llamaindex": "llm",
"chatgpt": "llm", "gpt-4": "llm", "claude": "llm", "gemini": "llm",
"openai api": "llm", "anthropic": "llm", "llm": "llm",
"prompt engineering": "llm", "re-ranking": "llm", "reranking": "llm",
# Evaluation
"ndcg": "evaluation", "mrr": "evaluation", "map@": "evaluation",
"mean average precision": "evaluation", "precision@": "evaluation",
"offline evaluation": "evaluation", "online evaluation": "evaluation",
"a/b test": "evaluation", "ab test": "evaluation",
"offline-to-online": "evaluation", "evaluation framework": "evaluation",
"evaluation pipeline": "evaluation", "recall@": "evaluation",
# Python / Engineering
"python": "engineering", "scikit-learn": "engineering", "sklearn": "engineering",
"pytorch": "engineering", "tensorflow": "engineering", "jax": "engineering",
"fastapi": "engineering", "flask": "engineering", "docker": "engineering",
"kubernetes": "engineering", "k8s": "engineering",
# Distributed / Scale
"distributed system": "infrastructure", "large-scale inference": "infrastructure",
"low latency": "infrastructure", "high throughput": "infrastructure",
"horizontal scaling": "infrastructure", "inference optimization": "infrastructure",
"model serving": "infrastructure", "kafka": "infrastructure",
"spark": "infrastructure", "airflow": "infrastructure",
# HR-Tech / Marketplace
"recruiting": "hr_tech", "hr tech": "hr_tech", "hrtech": "hr_tech",
"talent platform": "hr_tech", "marketplace": "hr_tech",
"job search": "hr_tech", "candidate matching": "hr_tech",
"hiring platform": "hr_tech",
# Open Source
"open source": "open_source", "open-source": "open_source",
"published a paper": "open_source", "conference talk": "open_source",
"blog post": "open_source", "github.com": "open_source",
"oss contribut": "open_source",
# NLP
"nlp": "nlp", "natural language": "nlp",
"text classification": "nlp", "named entity": "nlp",
"tokenization": "nlp", "transformer": "nlp",
"attention mechanism": "nlp", "bert": "nlp", "gpt": "nlp",
# Computer Vision (non-target for this JD)
"computer vision": "cv", "image classification": "cv",
"object detection": "cv", "segmentation": "cv",
# Speech (non-target)
"speech recognition": "speech", "tts": "speech", "asr": "speech",
"speech-to-text": "speech", "text-to-speech": "speech",
# Robotics (non-target)
"robotics": "robotics", "robot": "robotics",
}
# Domain label mapping
DOMAIN_LABELS = {
"search": "Search & Information Retrieval",
"ranking": "Learning to Rank & Recommendation",
"embeddings": "Embeddings & Dense Retrieval",
"vector_db": "Vector Databases & Hybrid Search",
"llm": "LLM & Fine-tuning",
"evaluation": "Evaluation Frameworks",
"engineering": "ML Engineering",
"infrastructure": "Distributed Systems & Infrastructure",
"hr_tech": "HR-Tech & Marketplace",
"open_source": "Open Source & External Validation",
"nlp": "Natural Language Processing",
"cv": "Computer Vision",
"speech": "Speech Processing",
"robotics": "Robotics",
}
def __init__(self, jd_text: str | None = None):
self.raw = (jd_text or JD_FULL_TEXT).lower()
self.raw_original = jd_text or JD_FULL_TEXT
self._parsed: Optional[JDUnderstanding] = None
def parse(self) -> JDUnderstanding:
"""Parse the JD and return structured understanding."""
if self._parsed is not None:
return self._parsed
u = JDUnderstanding(raw_text=self.raw_original)
# 1. Extract required/preferred skills with domain classification
self._extract_skills(u)
# 2. Extract experience requirements
self._extract_experience(u)
# 3. Detect domain
self._detect_domain(u)
# 4. Extract locations
self._extract_locations(u)
# 5. Extract red flags and disqualifiers
self._extract_red_flags(u)
# 6. Build production evidence vocabulary
self._build_production_evidence(u)
# 7. Build pre-LLM signals
self._build_pre_llm_signals(u)
# 8. Extract ideal candidate text
self._extract_ideal_text(u)
self._parsed = u
return u
def _extract_skills(self, u: JDUnderstanding) -> None:
"""Extract required and preferred skills, classified by domain."""
text = self.raw
# --- Required skills (section: "absolutely need" -> "like you to have") ---
required_section = self._extract_section(text, [
"things you absolutely need", "must have", "required skills",
"requirements", "essential",
], stop_at=[
"things we'd like you to have", "things we explicitly do not want",
"like you to have", "nice to have", "won't reject you for",
"do not want", "disqualif", "location:", "ideal candidate",
])
# Build required skills grouped by domain
required = {}
for skill, domain in self.SKILL_DOMAINS.items():
if skill in required_section:
if domain not in required:
required[domain] = []
required[domain].append(skill)
# Ensure at least the core domains are represented
if not required:
required = self._fallback_required_skills(text)
u.required_skills = required
# --- Preferred skills (section: "like you to have" -> "do not want") ---
preferred_section = self._extract_section(text, [
"like you to have", "nice to have", "preferred", "bonus",
"won't reject you for",
], stop_at=[
"things we explicitly do not want", "do not want",
"disqualif", "location:", "ideal candidate",
])
preferred = {}
for skill, domain in self.SKILL_DOMAINS.items():
if skill in preferred_section and not any(
skill in v for v in required.values()
):
if domain not in preferred:
preferred[domain] = []
preferred[domain].append(skill)
u.preferred_skills = preferred
def _extract_section(self, text: str, section_markers: list[str],
stop_at: list[str] | None = None) -> str:
"""Extract text from a section identified by markers."""
_DEFAULT_STOPS = [
"things we explicitly do not want", "do not want",
"disqualif", "red flag", "location:", "notice period",
"ideal candidate", "how to read", "final note",
"the vibe check", "what we mean by",
]
stops = stop_at or _DEFAULT_STOPS
for marker in section_markers:
idx = text.find(marker)
if idx >= 0:
best_end = len(text)
for em in stops:
eidx = text.find(em, idx + len(marker))
if 0 < eidx < best_end:
best_end = eidx
return text[idx:best_end]
return ""
def _in_required_context(self, text: str, skill: str) -> bool:
"""Check if a skill appears in a required-sounding context."""
# Look for skill within 200 chars of requirement indicators
for indicator in ["need:", "require", "must have", "essential", "production experience with"]:
idx = text.find(indicator)
while idx >= 0:
context = text[idx:idx + 300]
if skill in context:
return True
idx = text.find(indicator, idx + 1)
return False
def _fallback_required_skills(self, text: str) -> dict[str, list[str]]:
"""Fallback skill extraction when section detection fails."""
required = {}
# Check for the most important domains
key_skills = [
("embeddings", "embedding", "sentence-transformers", "dense retrieval",
"semantic search", "retrieval-augmented", "rag pipeline"),
("vector_db", "pinecone", "weaviate", "qdrant", "milvus", "faiss",
"vector database", "hybrid search", "opensearch", "elasticsearch"),
("evaluation", "ndcg", "mrr", "map@", "a/b test", "evaluation framework"),
("engineering", "python"),
]
for domain, *skills in key_skills:
found = [s for s in skills if s in text]
if found:
required[domain] = found
return required
def _extract_experience(self, u: JDUnderstanding) -> None:
"""Extract experience requirements from JD text."""
text = self.raw
# Try to find "X-Y years" pattern
yoe_patterns = [
r"(\d+)\s*[-–]\s*(\d+)\s*years",
r"(\d+)\s*to\s*(\d+)\s*years",
r"(\d+)\s*-\s*(\d+)\s*year",
]
for pat in yoe_patterns:
m = re.search(pat, text)
if m:
u.yoe_low = int(m.group(1))
u.yoe_high = int(m.group(2))
break
# Look for ideal range ("6-8 years")
ideal_patterns = [
r"ideal.*?(\d+)\s*[-–]\s*(\d+)\s*years",
r"(\d+)\s*[-–]\s*(\d+)\s*years.*?total",
r"(\d+)[-–](\d+)\s*years total",
]
for pat in ideal_patterns:
m = re.search(pat, text)
if m:
u.yoe_ideal_low = int(m.group(1))
u.yoe_ideal_high = int(m.group(2))
break
# Look for domain experience ("4-5 in applied ML")
domain_patterns = [
r"(\d+)\s*[-–]\s*(\d+)\s*(?:years?|yrs?)\s*(?:in|of)\s*(?:applied|relevant|domain)",
r"(\d+)\s*[-–]\s*(\d+)\s*(?:are|in)\s*(?:applied|relevant)",
]
for pat in domain_patterns:
m = re.search(pat, text)
if m:
u.yoe_domain_low = int(m.group(1))
u.yoe_domain_high = int(m.group(2))
break
# Seniority detection
text_lower = text
def _detect_domain(self, u: JDUnderstanding) -> None:
"""Detect the primary job domain from JD text."""
text = self.raw
required_section = self._extract_section(text, [
"things you absolutely need", "must have",
], stop_at=["like you to have", "do not want"])
domain_scores: dict[str, int] = {}
for skill, domain in self.SKILL_DOMAINS.items():
if skill in required_section:
domain_scores[domain] = domain_scores.get(domain, 0) + 2
elif skill in text:
domain_scores[domain] = domain_scores.get(domain, 0) + 1
if domain_scores:
u.domain = max(domain_scores, key=domain_scores.get)
u.domain_label = self.DOMAIN_LABELS.get(u.domain, u.domain)
def _extract_experience(self, u: JDUnderstanding) -> None:
"""Extract experience requirements from JD text."""
text = self.raw
text_lower = text
# Try to find "X-Y years" pattern
yoe_patterns = [
r"(\d+)\s*[-–]\s*(\d+)\s*years",
r"(\d+)\s*to\s*(\d+)\s*years",
r"(\d+)\s*-\s*(\d+)\s*year",
]
for pat in yoe_patterns:
m = re.search(pat, text)
if m:
u.yoe_low = int(m.group(1))
u.yoe_high = int(m.group(2))
break
# Look for ideal range ("6-8 years")
ideal_patterns = [
r"ideal.*?(\d+)\s*[-–]\s*(\d+)\s*years",
r"(\d+)\s*[-–]\s*(\d+)\s*years.*?total",
r"(\d+)[-–](\d+)\s*years total",
]
for pat in ideal_patterns:
m = re.search(pat, text)
if m:
u.yoe_ideal_low = int(m.group(1))
u.yoe_ideal_high = int(m.group(2))
break
# Look for domain experience ("4-5 in applied ML")
domain_patterns = [
r"(\d+)\s*[-–]\s*(\d+)\s*(?:years?|yrs?)\s*(?:in|of)\s*(?:applied|relevant|domain)",
]
for pat in domain_patterns:
m = re.search(pat, text)
if m:
u.yoe_domain_low = int(m.group(1))
u.yoe_domain_high = int(m.group(2))
break
# Seniority detection
if any(w in text_lower for w in ["senior", "staff", "principal", "lead"]):
if "principal" in text_lower:
u.seniority = "principal"
elif "staff" in text_lower:
u.seniority = "staff"
else:
u.seniority = "senior"
elif any(w in text_lower for w in ["junior", "entry", "fresher"]):
u.seniority = "junior"
else:
u.seniority = "mid"
def _detect_domain(self, u: JDUnderstanding) -> None:
"""Detect the primary job domain from JD text."""
text = self.raw
# Count domain keyword hits
domain_scores: dict[str, int] = {}
for skill, domain in self.SKILL_DOMAINS.items():
if skill in text:
domain_scores[domain] = domain_scores.get(domain, 0) + 1
if domain_scores:
u.domain = max(domain_scores, key=domain_scores.get)
u.domain_label = self.DOMAIN_LABELS.get(u.domain, u.domain)
def _extract_locations(self, u: JDUnderstanding) -> None:
"""Extract location preferences."""
text = self.raw
# Preferred locations
preferred_markers = ["preferred", "ideal location", "based in"]
for marker in preferred_markers:
idx = text.find(marker)
if idx >= 0:
context = text[idx:idx + 200]
# Extract city names
cities = ["pune", "noida", "bangalore", "bengaluru", "hyderabad",
"mumbai", "delhi", "ncr", "gurgaon", "gurugram",
"chennai", "kolkata", "bangalore"]
for city in cities:
if city in context:
if city in ("pune", "noida"):
if city not in u.preferred_locations:
u.preferred_locations.append(city)
elif city not in u.welcome_locations:
u.welcome_locations.append(city)
# Fallback if no preferred found
if not u.preferred_locations:
for city in ["pune", "noida"]:
if city in text:
u.preferred_locations.append(city)
def _extract_red_flags(self, u: JDUnderstanding) -> None:
"""Extract disqualifiers and red flags from JD."""
text = self.raw
# Research-only
u.research_only_titles = [
"research scientist", "research engineer", "research fellow",
"postdoctoral", "phd researcher", "academic researcher",
]
# Architect/manager drift
u.architect_titles = ["architect", "tech lead", "engineering manager", "head of"]
# Non-engineering titles
u.bad_title_patterns = [
"customer support", "customer success", "marketing manager", "marketing director",
"content writer", "hr manager", "human resources", "graphic designer",
"ui designer", "ux designer", "sales manager", "account manager",
"civil engineer", "mechanical engineer", "electrical engineer",
"accountant", "recruiter", "talent acquisition", "operations manager",
"android developer", "ios developer", "mobile developer",
"seo specialist", "social media manager", "business analyst",
"project manager", "product manager",
]
# Consulting firms
u.consulting_firms = [
"tcs", "tata consultancy services", "infosys", "wipro", "accenture",
"cognizant", "capgemini", "hcl", "tech mahindra", "mindtree",
"l&t infotech", "lti", "mphasis", "hexaware", "persistent systems",
"zensar", "birlasoft", "niit", "cyient", "mastek", "sonata software",
"genpact", "wns", "firstsource",
]
u.consulting_industries = ["it services", "consulting", "staffing", "bpo"]
# Non-target domains
u.non_target_domains = ["computer vision", "speech recognition", "robotics", "cv engineer"]
u.non_target_rescue = [
"nlp", "natural language", "retrieval", "search", "ranking",
"recommendation", "embeddings", "text classification",
]
# Red flags as structured dict
u.red_flags = {
"research_only": ["pure research", "academic", "no production"],
"framework_enthusiast": ["langchain tutorial", "framework enthusiast", "demo"],
"consulting_only": ["consulting firms", "entire career", "pure services"],
"wrong_domain": ["computer vision", "speech", "robotics", "without nlp"],
"title_chaser": ["switching companies", "1.5 years", "title-chaser"],
"no_external_validation": ["closed-source", "proprietary", "5+ years without"],
"architect_drift": ["hasn't written", "production code", "18 months"],
}
# Notice period preference
notice_match = re.search(r"sub-(\d+)-day", text)
if notice_match:
u.notice_preferred_days = int(notice_match.group(1))
# Product company requirement
u.requires_product_company = "product compan" in text
u.requires_production_code = "production code" in text or "production deployment" in text
def _build_production_evidence(self, u: JDUnderstanding) -> None:
"""Build vocabulary for detecting production deployment evidence."""
u.production_evidence = [
"production", "deployed", "shipped", "live traffic",
"real users", "at scale", "latency", "throughput",
"a/b test", "recall improvement", "ranking quality",
"rollout", "launched", "owned the", "on-call",
"end-to-end", "serving", "p99", "qps",
]
def _build_pre_llm_signals(self, u: JDUnderstanding) -> None:
"""Build pre-LLM IR vocabulary and post-LLM markers."""
u.pre_llm_keywords = [
"search ranking", "information retrieval", "recommendation system",
"recommender system", "learning to rank", "click-through", "ctr model",
"collaborative filtering", "search relevance", "query understanding",
"ranking algorithm", "elasticsearch", "solr", "bm25",
]
u.post_llm_markers = [
"langchain", "llamaindex", "rag pipeline", "chatgpt", "gpt-4", "llama 2",
"claude", "gemini", "openai api", "anthropic",
]
def _extract_ideal_text(self, u: JDUnderstanding) -> None:
"""Extract the ideal candidate description for embedding similarity."""
text = self.raw
# Look for "ideal candidate" section
ideal_markers = ["ideal candidate", "ideal candidate we're imagining",
"what we're looking for", "who we want"]
for marker in ideal_markers:
idx = text.find(marker)
if idx >= 0:
# Take a generous chunk after the marker
end = len(text)
for stop in ["final note", "how to read", "good luck"]:
sidx = text.find(stop, idx)
if 0 < sidx < end:
end = sidx
u.ideal_text = text[idx:end].strip()
return
# Fallback: use the entire JD as ideal text
u.ideal_text = text
# ---------------------------------------------------------------------------
# Singleton: parse the JD once at import time
# ---------------------------------------------------------------------------
def get_jd() -> JDUnderstanding:
"""Get the parsed JD understanding (cached after first parse)."""
if not hasattr(get_jd, "_cache"):
parser = JDParser()
get_jd._cache = parser.parse()
return get_jd._cache