""" lib/domain.py — V5 Dynamic Domain Taxonomy Builds a 3-tier skill taxonomy dynamically from the JD understanding. No hardcoded "search_spec_score" — the domain engine derives importance weights from what the JD actually asks for. Tier 1: Skills explicitly mentioned in "absolutely need" / required section Tier 2: Skills mentioned in "nice to have" / preferred section Tier 3: Skills that are adjacent/related but not mentioned in JD This drives: - skill_coverage scoring (Tier 1 weighted more than Tier 2) - domain_specialization scoring (how deep in the JD's domain) - evidence extraction prioritization (look for Tier 1 evidence first) """ from __future__ import annotations from lib.jd_parser import JDUnderstanding, get_jd from dataclasses import dataclass, field @dataclass class DomainTaxonomy: """Dynamic 3-tier taxonomy derived from JD.""" tier1: dict[str, list[str]] = field(default_factory=dict) tier2: dict[str, list[str]] = field(default_factory=dict) tier3: dict[str, list[str]] = field(default_factory=dict) skill_tier: dict[str, int] = field(default_factory=dict) skill_domain: dict[str, str] = field(default_factory=dict) domain_importance: dict[str, float] = field(default_factory=dict) class DomainEngine: """ Builds a dynamic domain taxonomy from JD understanding. The engine takes the parsed JD and creates a 3-tier skill taxonomy that drives all downstream scoring. """ # Adjacency map: which domains are related to which DOMAIN_ADJACENCY = { "search": ["ranking", "embeddings", "vector_db", "nlp"], "ranking": ["search", "embeddings", "evaluation", "nlp"], "embeddings": ["vector_db", "llm", "nlp", "search"], "vector_db": ["embeddings", "search", "infrastructure"], "llm": ["embeddings", "nlp", "engineering"], "evaluation": ["ranking", "search", "llm"], "engineering": ["infrastructure", "llm", "embeddings"], "infrastructure": ["engineering", "vector_db"], "hr_tech": ["ranking", "search"], "open_source": [], "nlp": ["search", "ranking", "embeddings", "llm"], "cv": [], "speech": [], "robotics": [], } # Extended skill lists for Tier 3 (related but not in JD) EXTENDED_SKILLS = { "search": ["lucene", "solr", "query understanding", "spell correction", "facet", "filter", "re-ranking", "query expansion", "relevance feedback", "learning to rank", "click model"], "ranking": [" ctr model", "conversion rate", "personalization", "feature engineering for ranking", "position bias", "counterfactual", "offline evaluation", "online evaluation"], "embeddings": ["word2vec", "glove", "fasttext", "cohere embedding", "voyage embedding", "instructor", "colbert", "cross-encoder", "bi-encoder", "late interaction"], "vector_db": ["hnsw", "ivf", "pq", "product quantization", "ann", "approximate nearest neighbor", "index management", "sharding", "replication", "consistency level"], "llm": ["gpt", "bert", "t5", "llama", "mistral", "transformer", "attention", "decoder", "encoder", "sequence to sequence", "instruction tuning", "rlhf", "dpo", "constitutional ai"], "evaluation": ["precision", "recall", "f1", "hit rate", "mrr", "ndcg", "map", "coverage", "diversity", "novelty", "calibration", "fairness", "A/B testing"], "engineering": ["ci/cd", "git", "unit test", "integration test", "code review", "agile", "scrum", "technical design", "api design", "rest", "graphql", "microservice", "observability", "monitoring", "logging"], "infrastructure": ["aws", "gcp", "azure", "docker", "kubernetes", "terraform", "helm", "kafka", "redis", "postgresql", "mongodb", "cassandra", "elasticsearch cluster", "load balancing", "auto-scaling", "cdn"], "hr_tech": ["ats", "applicant tracking", "resume parsing", "candidate scoring", "talent pool", "sourcing", "recruitment funnel", "interview scheduling"], "open_source": ["github", "gitlab", "pull request", "code contribution", "library", "package", "pypi", "npm"], "nlp": ["tokenization", "lemmatization", "stemming", "pos tagging", "named entity recognition", "sentiment analysis", "text classification", "language model", "sequence labeling", "summarization", "translation", "question answering"], } def __init__(self, jd: JDUnderstanding | None = None): self.jd = jd or get_jd() self._taxonomy: DomainTaxonomy | None = None def build(self) -> DomainTaxonomy: """Build the 3-tier taxonomy from JD.""" if self._taxonomy is not None: return self._taxonomy tier1 = dict(self.jd.required_skills) # deep copy tier2 = dict(self.jd.preferred_skills) tier3: dict[str, list[str]] = {} # Collect all Tier 1+2 skills and domains all_skills = set() all_domains = set() for t in [tier1, tier2]: for domain, skills in t.items(): all_skills.update(skills) all_domains.add(domain) # Build Tier 3: related skills not in Tier 1 or 2 for domain in all_domains: related_domains = self.DOMAIN_ADJACENCY.get(domain, []) for rd in related_domains: for skill in self.EXTENDED_SKILLS.get(rd, []): if skill.lower() not in all_skills and rd not in tier1 and rd not in tier2: if rd not in tier3: tier3[rd] = [] tier3[rd].append(skill.lower()) # Also add extended skills for Tier 1/2 domains that weren't fully listed for domain in all_domains: for skill in self.EXTENDED_SKILLS.get(domain, []): if skill.lower() not in all_skills: if domain not in tier3: tier3[domain] = [] tier3[domain].append(skill.lower()) # Build flat lookup maps skill_tier: dict[str, int] = {} skill_domain: dict[str, str] = {} for domain, skills in tier1.items(): for s in skills: skill_tier[s] = 1 skill_domain[s] = domain for domain, skills in tier2.items(): for s in skills: skill_tier[s] = 2 skill_domain[s] = domain for domain, skills in tier3.items(): for s in skills: if s not in skill_tier: # Don't override higher tiers skill_tier[s] = 3 skill_domain[s] = domain # Domain importance: weighted by number of Tier 1 skills + bonus for Tier 2 domain_importance: dict[str, float] = {} for domain, skills in tier1.items(): domain_importance[domain] = len(skills) * 1.0 for domain, skills in tier2.items(): domain_importance[domain] = domain_importance.get(domain, 0) + len(skills) * 0.5 # Normalize max_imp = max(domain_importance.values()) if domain_importance else 1.0 for d in domain_importance: domain_importance[d] /= max_imp self._taxonomy = DomainTaxonomy( tier1=tier1, tier2=tier2, tier3=tier3, skill_tier=skill_tier, skill_domain=skill_domain, domain_importance=domain_importance, ) return self._taxonomy def get_skill_weight(self, skill: str) -> float: """Get weight for a skill based on its tier. 1.0 for Tier 1, 0.5 for Tier 2, 0.1 for Tier 3.""" tax = self.build() tier = tax.skill_tier.get(skill.lower(), 3) return {1: 1.0, 2: 0.5, 3: 0.1}.get(tier, 0.05) def get_all_skills_flat(self) -> list[tuple[str, int, str]]: """Return all skills as (skill, tier, domain) tuples.""" tax = self.build() result = [] for skill, tier in tax.skill_tier.items(): domain = tax.skill_domain.get(skill, "unknown") result.append((skill, tier, domain)) return sorted(result, key=lambda x: (x[1], x[0])) # Singleton def get_taxonomy() -> DomainTaxonomy: """Get the domain taxonomy (cached after first build).""" if not hasattr(get_taxonomy, "_cache"): engine = DomainEngine() get_taxonomy._cache = engine.build() return get_taxonomy._cache