import math import re from typing import List, Tuple, Dict, Any from tech_radar.db.models import JobPosting class SemanticVectorStore: """ Lightweight, high-performance Vector & Semantic Search Engine for TechRadar-MCP across all software domains. """ def __init__(self): self.jobs: List[JobPosting] = [] self.doc_vectors: List[Dict[str, float]] = [] self.idf: Dict[str, float] = {} def _tokenize(self, text: str) -> List[str]: words = re.findall(r'\b[a-zA-Z0-9+#\.-]+\b', text.lower()) return [w for w in words if len(w) > 1] def index_jobs(self, jobs: List[JobPosting]): self.jobs = jobs self.doc_vectors = [] doc_count = len(jobs) doc_freq = {} raw_docs = [] for job in jobs: text = f"{job.title} {job.company} {job.tech_domain} {job.city} {job.area} {' '.join(job.tech_stack)} {job.requirements}" tokens = self._tokenize(text) raw_docs.append(tokens) unique_tokens = set(tokens) for token in unique_tokens: doc_freq[token] = doc_freq.get(token, 0) + 1 self.idf = { token: math.log((doc_count + 1) / (freq + 1)) + 1.0 for token, freq in doc_freq.items() } for tokens in raw_docs: tf = {} for t in tokens: tf[t] = tf.get(t, 0) + 1 length = len(tokens) or 1 vec = { term: (freq / length) * self.idf.get(term, 1.0) for term, freq in tf.items() } self.doc_vectors.append(vec) def _cosine_similarity(self, vec1: Dict[str, float], vec2: Dict[str, float]) -> float: intersection = set(vec1.keys()) & set(vec2.keys()) numerator = sum(vec1[x] * vec2[x] for x in intersection) sum1 = sum(val ** 2 for val in vec1.values()) sum2 = sum(val ** 2 for val in vec2.values()) denominator = math.sqrt(sum1) * math.sqrt(sum2) if not denominator: return 0.0 return float(numerator / denominator) def search_semantic( self, query: str, domain: str = None, city: str = None, top_k: int = 15 ) -> List[Tuple[JobPosting, float]]: if not self.doc_vectors or not self.jobs: return [] q_tokens = self._tokenize(query) tf = {} for t in q_tokens: tf[t] = tf.get(t, 0) + 1 q_length = len(q_tokens) or 1 q_vec = { term: (freq / q_length) * self.idf.get(term, 1.0) for term, freq in tf.items() } results = [] for idx, job in enumerate(self.jobs): if city and city.lower() != "all" and job.city.lower() != city.lower(): continue if domain and domain.lower() != "all" and job.tech_domain.lower() != domain.lower(): continue sim = self._cosine_similarity(q_vec, self.doc_vectors[idx]) tech_match_bonus = sum( 0.15 for t in q_tokens if any(t.lower() == stack.lower() for stack in job.tech_stack) ) final_score = round(min(1.0, sim + tech_match_bonus), 3) if final_score > 0.01: results.append((job, final_score)) results.sort(key=lambda x: x[1], reverse=True) return results[:top_k] def match_resume_to_jd(self, resume_text: str, job: JobPosting) -> float: r_tokens = self._tokenize(resume_text) jd_text = f"{job.title} {job.tech_domain} {' '.join(job.tech_stack)} {job.requirements}" jd_tokens = self._tokenize(jd_text) r_set = set(r_tokens) jd_set = set(jd_tokens) tech_matches = [t for t in job.tech_stack if any(t.lower() == r.lower() for r in r_set)] tech_ratio = len(tech_matches) / (len(job.tech_stack) or 1) common_vocab = r_set & jd_set vocab_ratio = len(common_vocab) / (len(jd_set) or 1) overall_score = (tech_ratio * 0.6) + (vocab_ratio * 0.4) return round(min(99.0, overall_score * 100.0), 1)