from __future__ import annotations import hashlib import json import threading from typing import Any, Dict, List, Optional, Set, Tuple import numpy as np import torch from cachetools import TTLCache from pydantic import BaseModel, Field from sentence_transformers import SentenceTransformer from services.processor_utils import ( DEFAULT_SKILL_ALIASES, DEFAULT_CATEGORY_SKILLS, SENIORITY_ORDER, ) # --------------------------------------------------------------------------- # Requires (for the ONNX path): pip install "optimum[onnxruntime]" onnxruntime # If these aren't installed, model loading below automatically falls back to # the plain PyTorch backend -- nothing crashes, it just runs at pre-ONNX speed. # --------------------------------------------------------------------------- try: import onnxruntime as ort _ONNXRUNTIME_AVAILABLE = True except ImportError: _ONNXRUNTIME_AVAILABLE = False # --------------------------------------------------------------------------- # Pin CPU thread count to the box's vCPU count. Left unpinned, PyTorch's # (and separately, ONNX Runtime's) default intra-op threading can # oversubscribe a 2-vCPU host and cause contention between concurrent # encode() calls. # --------------------------------------------------------------------------- torch.set_num_threads(2) # --------------------------------------------------------------------------- # Pydantic models # --------------------------------------------------------------------------- class JobInput(BaseModel): """Strictly typed job posting for JobMatcher.predict().""" job_id: str job_title: str company_name: str description: str skills_canonical: List[str] = Field(default_factory=list) skills_technical: List[str] = Field(default_factory=list) skills_soft: List[str] = Field(default_factory=list) category: str = "Other" seniority: str = "mid" min_years_experience: int = 0 class ScoreBreakdown(BaseModel): skill_score: float semantic_score: float title_score: float category_score: float seniority_score: float experience_score: float category_modifier: float = 0.0 class MismatchAnalysis(BaseModel): reason: str matched_technical_skills: List[str] missing_technical_skills: List[str] matched_soft_skills: List[str] missing_soft_skills: List[str] cv_extra_strengths: List[str] seniority_fit: str category_fit: str improvement_advice: List[str] class MatchPrediction(BaseModel): job_id: str job_title: str company_name: str match_score: str match_level: str cv_title: str cv_seniority: str cv_category: str cv_skills: List[str] cv_years_experience: int score_breakdown: ScoreBreakdown mismatch_analysis: MismatchAnalysis # --------------------------------------------------------------------------- # Pure scoring helpers # --------------------------------------------------------------------------- def _cosine_similarity(a: np.ndarray, b: np.ndarray) -> float: norm_a = np.linalg.norm(a) norm_b = np.linalg.norm(b) if norm_a == 0 or norm_b == 0: return 0.0 return float(np.dot(a, b) / (norm_a * norm_b)) def _seniority_fit(cv_seniority: str, job_seniority: str) -> Tuple[float, str]: cv_rank = SENIORITY_ORDER.get(cv_seniority, 1) job_rank = SENIORITY_ORDER.get(job_seniority, 2) diff = cv_rank - job_rank if diff == 0: return 1.0, f"Good seniority fit ({cv_seniority} → {job_seniority})." if diff == -1: return ( 0.65, f"Slightly under the requested level ({cv_seniority} → {job_seniority}).", ) if diff < -1: return 0.25, f"Seniority mismatch ({cv_seniority} CV for {job_seniority} role)." return ( 0.85, f"Candidate may be more experienced than required ({cv_seniority} → {job_seniority}).", ) def _experience_fit(cv_years: int, required_years: int) -> float: if required_years <= 0: return 1.0 if cv_years <= 0: return 0.55 if required_years <= 1 else 0.35 return max(0.1, min(1.0, cv_years / required_years)) def _soft_category_score( cv_skills: Set[str], job_category: str, category_skills: Dict[str, Set[str]], ) -> float: """ Jaccard-based soft category match instead of binary 1.0 / 0.30. Scores range from 0.30 (no overlap) to 1.0 (exact match). """ job_cat_skills = category_skills.get(job_category, set()) if not job_cat_skills: return 0.30 intersection = len(cv_skills & job_cat_skills) union = len(cv_skills | job_cat_skills) if union == 0: return 0.30 jaccard = intersection / union return max(0.30, min(1.0, 0.30 + jaccard * 0.70)) def _match_level(score: float) -> str: if score >= 80: return "Excellent Match" if score >= 70: return "Strong Match" if score >= 55: return "Good Match" if score >= 40: return "Moderate Match" if score >= 25: return "Weak Match" return "Very Weak Match" def _build_advice( missing_technical: List[str], job: JobInput, cv_category: str, ) -> List[str]: advice: List[str] = [] if missing_technical: advice.append( "Add evidence for these technical skills if you have them: " + ", ".join(missing_technical[:5]) + "." ) if cv_category != job.category: advice.append( f"This CV looks closer to {cv_category}; tailor the summary and " f"projects toward {job.category} for this job." ) if job.seniority == "senior": advice.append( "For senior roles, highlight leadership responsibilities, team size " "managed, and measurable achievements." ) if not advice: advice.append( "The CV is well aligned; strengthen it with measurable achievements " "and recent work examples." ) return advice[:4] # --------------------------------------------------------------------------- # Main class # --------------------------------------------------------------------------- class JobMatcher: """ CV-to-job matcher using sentence embeddings + skill taxonomy scoring. Single-job usage ---------------- matcher = JobMatcher() result = matcher.predict(job_dict, cv_markdown) Batch usage (same CV, many jobs — CV processed only once) --------------------------------------------------------- results = matcher.predict_batch(list_of_job_dicts, cv_markdown) Performance notes (batched implementation) ------------------------------------------- All model.encode() calls are batched ACROSS jobs, not per-job: - 1 call embeds every job's semantic text - 1 call embeds every job's title - 1 call embeds the (deduplicated) set of missing technical skills across *all* jobs, used for semantic skill fallback matching - The CV title vector is computed once and cached (was previously recomputed on every job). This means a request with N jobs makes a fixed ~4 model.encode() calls total instead of up to 4*N + 1, regardless of N. Scoring math and output are unchanged — this is purely a batching optimization. Model choice ------------ Semantic model is BAAI/bge-small-en-v1.5 (33M params, standard BERT encoder architecture, 384-dim, 512-token context). This replaces jina-embeddings-v2-base-en (137M, custom ALiBi remote-code arch). The jina model's custom architecture does not reliably export via optimum's ONNX path, so on most boxes it was silently falling back to full PyTorch fp32 despite use_onnx=True. bge-small has published quantized ONNX weights matching the _ONNX_FILE_CANDIDATES naming below, ships an order of magnitude fewer parameters, and stays within a couple of points of jina-base on semantic similarity benchmarks — it does not meaningfully change match_score behavior for this use case since semantic similarity is only 23% of the final weighted score to begin with. Title model is UNCHANGED (TechWolf/JobBERT-v3, falling back to v2) — title matching is cheap (short strings) and this model is domain-specific for job titles, so there's no reason to swap it. Truncation was reduced from 4000 to 1500 chars (job text) / 2000 chars (CV) — most of the matching signal lives in the first few hundred words, and this cuts attention cost roughly in half again on top of the smaller model, with negligible score impact in practice. Text is no longer manually chunked into overlapping ~180-word windows before embedding; the smaller model's 512-token window comfortably covers the reduced truncation length in a single forward pass. Both models are loaded via the ONNX Runtime backend (with int8 quantization where a quantized export is available on the model repo) for faster CPU inference, with automatic fallback to plain PyTorch if ONNX packages or ONNX weights aren't available — see _load_model(). The active backend is logged explicitly after load so a silent fallback to PyTorch is visible instead of assumed away. CV caching ---------- Parsing + embedding a CV (cv_vec, cv_title_vec, individual_skills_vecs) is the fixed per-request cost that doesn't shrink with batching. It's cached in-process for a configurable TTL, keyed by a hash of the CV inputs, using cachetools.TTLCache (pure in-memory, no external service required, thread-safe via an internal lock). A repeat request for the same CV — e.g. pagination across job results — skips CV parsing and all three CV-side encode() calls entirely. """ # Candidate ONNX file names to try, in preference order (most # aggressively quantized first). `None` means "let sentence-transformers # auto-select / auto-export the default ONNX model". Not every model # repo ships every variant, so we try each and move on if one 404s. _ONNX_FILE_CANDIDATES: List[Optional[str]] = [ "onnx/model_qint8_avx512_vnni.onnx", "onnx/model_quantized.onnx", "onnx/model_qint8.onnx", "onnx/model_int8.onnx", "onnx/model.onnx", None, ] # Truncation lengths — kept as class constants so they're easy to tune # without hunting through the method bodies. _JOB_TEXT_MAX_CHARS: int = 1500 _CV_TEXT_MAX_CHARS: int = 2000 def __init__( self, semantic_model_name: str = "BAAI/bge-small-en-v1.5", title_model_name: str = "TechWolf/JobBERT-v2", skill_aliases: Optional[Dict[str, List[str]]] = None, category_skills: Optional[Dict[str, Set[str]]] = None, use_onnx: bool = True, cv_cache_maxsize: int = 256, cv_cache_ttl_seconds: int = 600, ) -> None: # bge-small does not need trust_remote_code — it's a standard # BERT-family architecture, which is exactly why it exports to # ONNX reliably (unlike jina's custom ALiBi remote code). self._model = self._load_model( semantic_model_name, use_onnx=use_onnx, trust_remote_code=False ) try: self._title_model = self._load_model( "TechWolf/JobBERT-v3", use_onnx=use_onnx, trust_remote_code=False ) except Exception: self._title_model = self._load_model( title_model_name, use_onnx=use_onnx, trust_remote_code=False ) self._aliases: Dict[str, List[str]] = skill_aliases or DEFAULT_SKILL_ALIASES self._category_skills: Dict[str, Set[str]] = ( category_skills or DEFAULT_CATEGORY_SKILLS ) # In-process CV embedding cache. No external service, no setup — # just a bounded, TTL'd dict living in this process's memory. # Guarded by a lock since TTLCache itself is not thread-safe. self._cv_cache: TTLCache = TTLCache( maxsize=cv_cache_maxsize, ttl=cv_cache_ttl_seconds ) self._cv_cache_lock = threading.RLock() # ------------------------------------------------------------------ # Internal: load a SentenceTransformer via ONNX Runtime if possible, # trying quantized variants first, falling back to plain PyTorch. # ------------------------------------------------------------------ @classmethod def _load_model( cls, model_name: str, use_onnx: bool = True, trust_remote_code: bool = False, ) -> SentenceTransformer: if use_onnx and _ONNXRUNTIME_AVAILABLE: session_options = None try: session_options = ort.SessionOptions() session_options.intra_op_num_threads = 2 session_options.inter_op_num_threads = 1 except Exception: session_options = None for file_name in cls._ONNX_FILE_CANDIDATES: try: model_kwargs: Dict[str, Any] = {} if file_name: model_kwargs["file_name"] = file_name if session_options is not None: model_kwargs["session_options"] = session_options model = SentenceTransformer( model_name, backend="onnx", trust_remote_code=trust_remote_code, model_kwargs=model_kwargs, ) print( f"[JobMatcher] Loaded '{model_name}' via ONNX Runtime " f"(file={file_name or 'auto-export'})." ) cls._log_active_backend(model, model_name) return model except Exception as exc: print( f"[JobMatcher] ONNX load attempt failed for " f"'{model_name}' (file={file_name or 'auto-export'}): {exc}" ) elif use_onnx and not _ONNXRUNTIME_AVAILABLE: print( "[JobMatcher] onnxruntime/optimum not installed — " f"falling back to PyTorch backend for '{model_name}'. " 'Install with: pip install "optimum[onnxruntime]" onnxruntime' ) print(f"[JobMatcher] Loading '{model_name}' via PyTorch backend.") model = SentenceTransformer(model_name, trust_remote_code=trust_remote_code) cls._log_active_backend(model, model_name) return model @staticmethod def _log_active_backend(model: SentenceTransformer, model_name: str) -> None: """ Explicitly verify + log which backend actually ended up active. The try/except ladder in _load_model can silently fall through to PyTorch even when use_onnx=True, so this makes that visible instead of assumed. """ backend = getattr(model, "backend", None) or "pytorch" print(f"[JobMatcher] '{model_name}' ACTIVE backend = {backend}") if backend != "onnx": print( f"[JobMatcher] NOTE: '{model_name}' is running on PyTorch, not ONNX. " "Check that 'optimum[onnxruntime]' and 'onnxruntime' are installed " "and that the model repo has quantized ONNX weights." ) # ------------------------------------------------------------------ # Public: single-job prediction (backward-compatible) # ------------------------------------------------------------------ def predict( self, job_json: Dict[str, Any], cv_markdown: Optional[str] = None, cv_title: Optional[str] = None, cv_years: Optional[int] = None, cv_seniority: Optional[str] = None, cv_skills_canonical: Optional[List[str]] = None, cv_skills_technical: Optional[List[str]] = None, cv_skills_soft: Optional[List[str]] = None, cv_category: Optional[str] = None, ) -> Dict[str, Any]: """ Predict the match between one job and one CV. Implemented as a thin wrapper around predict_batch([job]) so there is a single scoring code path to keep in sync. """ results = self.predict_batch( jobs=[job_json], cv_markdown=cv_markdown or "", cv_title=cv_title or "Unknown", cv_years=cv_years or 0, cv_seniority=cv_seniority or "mid", cv_skills_canonical=cv_skills_canonical or [], cv_skills_technical=cv_skills_technical or [], cv_skills_soft=cv_skills_soft or [], cv_category=cv_category or "Other", ) return results[0] # ------------------------------------------------------------------ # Public: batch prediction (one CV vs. N jobs) # ------------------------------------------------------------------ def predict_batch( self, jobs: List[Dict[str, Any]], cv_markdown: str, cv_title: str, cv_years: int, cv_seniority: str, cv_skills_canonical: List[str], cv_skills_technical: List[str], cv_skills_soft: List[str], cv_category: str, ) -> List[Dict[str, Any]]: """ Match one CV against multiple jobs. The CV is parsed and embedded **once** (or fetched from cache if an identical CV was matched recently). All job-level embeddings (semantic text, title, missing-skill fallback) are computed in batched encode() calls across every job in the request, then scoring is done purely in numpy with no further model calls. """ cv_data = self._get_or_parse_cv( cv_markdown=cv_markdown, cv_title=cv_title, cv_years=cv_years, cv_seniority=cv_seniority, cv_skills_canonical=cv_skills_canonical, cv_skills_technical=cv_skills_technical, cv_skills_soft=cv_skills_soft, cv_category=cv_category, ) job_objs = [JobInput(**j) for j in jobs] n = len(job_objs) if n == 0: return [] # ---- Batch 1: job semantic-text embeddings, ONE encode() call ---- job_texts = [self._build_job_index_text(j) for j in job_objs] job_vecs = self._embed_batch_texts(job_texts) # ---- Batch 2: job title embeddings, ONE encode() call ---- job_title_vecs = self._embed_title_batch([j.job_title for j in job_objs]) # ---- Precompute exact skill matches + collect missing technical # skills across ALL jobs for a single batched semantic pass ---- cv_skills_technical_set = cv_data["cv_skills_technical"] per_job_technical: List[Set[str]] = [] per_job_soft: List[Set[str]] = [] per_job_matched_technical: List[List[str]] = [] per_job_missing_technical: List[List[str]] = [] all_missing_skills: Set[str] = set() for job in job_objs: job_technical = { s.strip().lower() for s in job.skills_technical if s.strip() } job_soft = {s.strip().lower() for s in job.skills_soft if s.strip()} matched = sorted(cv_skills_technical_set & job_technical) missing = sorted(job_technical - cv_skills_technical_set) per_job_technical.append(job_technical) per_job_soft.append(job_soft) per_job_matched_technical.append(matched) per_job_missing_technical.append(missing) all_missing_skills.update(missing) # ---- Batch 3: semantic skill-matching embeddings, ONE encode() # call for the deduplicated union of missing skills across # every job in the request ---- missing_skill_vecs: Dict[str, np.ndarray] = {} individual_skills_vecs = cv_data["individual_skills_vecs"] if all_missing_skills and len(individual_skills_vecs) > 0: unique_missing = sorted(all_missing_skills) contextualized = [f"software development skill {s}" for s in unique_missing] vecs = self._model.encode( contextualized, normalize_embeddings=True, show_progress_bar=False, batch_size=32, ) missing_skill_vecs = dict(zip(unique_missing, np.asarray(vecs))) # ---- Score each job — pure numpy from here, no model calls ---- results = [ self._score_job( job=job_objs[i], cv_data=cv_data, job_vec=job_vecs[i], job_title_vec=job_title_vecs[i], job_technical=per_job_technical[i], job_soft=per_job_soft[i], matched_technical=per_job_matched_technical[i], missing_technical=per_job_missing_technical[i], missing_skill_vecs=missing_skill_vecs, ) for i in range(n) ] results.sort( key=lambda x: float(x["match_score"].rstrip("%")), reverse=True, ) return results # ------------------------------------------------------------------ # Internal: CV cache lookup / population # ------------------------------------------------------------------ def _cv_cache_key( self, cv_markdown: str, cv_title: str, cv_years: int, cv_seniority: str, cv_skills_canonical: List[str], cv_skills_technical: List[str], cv_skills_soft: List[str], cv_category: str, ) -> str: """ Deterministic hash of every input that affects CV parsing/embedding. Truncate the markdown to the same length actually used for embedding so two CVs differing only past the truncation point still hit the same cache entry. """ payload = { "cv_markdown": (cv_markdown or "")[: self._CV_TEXT_MAX_CHARS], "cv_title": cv_title or "", "cv_years": cv_years, "cv_seniority": cv_seniority, "cv_skills_canonical": sorted(cv_skills_canonical or []), "cv_skills_technical": sorted(cv_skills_technical or []), "cv_skills_soft": sorted(cv_skills_soft or []), "cv_category": cv_category or "", } blob = json.dumps(payload, sort_keys=True, separators=(",", ":")) return hashlib.sha256(blob.encode("utf-8")).hexdigest() def _get_or_parse_cv( self, cv_markdown: str, cv_title: str, cv_years: int, cv_seniority: str, cv_skills_canonical: List[str], cv_skills_technical: List[str], cv_skills_soft: List[str], cv_category: str, ) -> Dict[str, Any]: cache_key = self._cv_cache_key( cv_markdown=cv_markdown, cv_title=cv_title, cv_years=cv_years, cv_seniority=cv_seniority, cv_skills_canonical=cv_skills_canonical, cv_skills_technical=cv_skills_technical, cv_skills_soft=cv_skills_soft, cv_category=cv_category, ) with self._cv_cache_lock: cached = self._cv_cache.get(cache_key) if cached is not None: return cached cv_data = self._parse_cv( cv_markdown=cv_markdown, cv_title=cv_title, cv_years=cv_years, cv_seniority=cv_seniority, cv_skills_canonical=cv_skills_canonical, cv_skills_technical=cv_skills_technical, cv_skills_soft=cv_skills_soft, cv_category=cv_category, ) with self._cv_cache_lock: self._cv_cache[cache_key] = cv_data return cv_data # ------------------------------------------------------------------ # Internal: parse + embed CV (done once per unique CV, then cached) # ------------------------------------------------------------------ def _parse_cv( self, cv_markdown: str, cv_title: str, cv_years: int, cv_seniority: str, cv_skills_canonical: List[str], cv_skills_technical: List[str], cv_skills_soft: List[str], cv_category: str, ) -> Dict[str, Any]: """ Return a dict holding all CV-level data needed by _score_job. """ cv_skills_set = set(cv_skills_canonical) cv_skills_technical_set = set(cv_skills_technical) cv_skills_soft_set = set(cv_skills_soft) cv_vec = ( self._embed(cv_markdown[: self._CV_TEXT_MAX_CHARS]) if cv_markdown else np.zeros(384, dtype="float32") ) # Step 1: compute the CV title vector once here instead of # recomputing it inside every per-job title-similarity call. cv_title_vec = ( self._embed_title(cv_title) if cv_title and cv_title != "Unknown" else None ) individual_skills_list = list(cv_skills_technical_set | cv_skills_soft_set) if individual_skills_list: contextualized_cv = [ f"software development skill {s}" for s in individual_skills_list ] individual_skills_vecs = self._model.encode( contextualized_cv, normalize_embeddings=True, show_progress_bar=False, batch_size=32, ) else: individual_skills_vecs = np.array([]) return { "cv_title": cv_title, "cv_title_vec": cv_title_vec, "cv_years": cv_years, "cv_seniority": cv_seniority, "cv_skills": cv_skills_set, "cv_skills_technical": cv_skills_technical_set, "cv_skills_soft": cv_skills_soft_set, "cv_category": cv_category, "cv_vec": cv_vec, "individual_skills_list": individual_skills_list, "individual_skills_vecs": individual_skills_vecs, } # ------------------------------------------------------------------ # Internal: score one job against pre-parsed CV data + precomputed # job-level vectors (no model calls happen in this method). # ------------------------------------------------------------------ def _score_job( self, job: JobInput, cv_data: Dict[str, Any], job_vec: np.ndarray, job_title_vec: np.ndarray, job_technical: Set[str], job_soft: Set[str], matched_technical: List[str], missing_technical: List[str], missing_skill_vecs: Dict[str, np.ndarray], ) -> Dict[str, Any]: cv_title = cv_data["cv_title"] cv_title_vec = cv_data["cv_title_vec"] cv_years = cv_data["cv_years"] cv_seniority = cv_data["cv_seniority"] cv_skills = cv_data["cv_skills"] cv_skills_soft = cv_data["cv_skills_soft"] cv_category = cv_data["cv_category"] cv_vec = cv_data["cv_vec"] individual_skills_vecs = cv_data["individual_skills_vecs"] # ── Semantic similarity (job embedding precomputed in the batch) ── semantic = max(0.0, min(1.0, _cosine_similarity(cv_vec, job_vec))) # ── Skill scoring ─────────────────────────────────────────────── matched_technical = list(matched_technical) missing_technical = list(missing_technical) # Semantic fallback for missing technical skills, using the # precomputed (batched, deduplicated) missing-skill vector lookup. if missing_technical and len(individual_skills_vecs) > 0 and missing_skill_vecs: semantic_matches = self._semantic_skill_matches_precomputed( missing_technical, missing_skill_vecs, individual_skills_vecs, threshold=0.88, ) if semantic_matches: matched_technical = sorted(set(matched_technical) | semantic_matches) missing_technical = sorted(set(missing_technical) - semantic_matches) matched_soft = sorted(cv_skills_soft & job_soft) missing_soft = sorted(job_soft - cv_skills_soft) extra = sorted(cv_skills - job_technical - job_soft)[:8] tech_denom = min(len(job_technical), 12) if job_technical else 1 soft_denom = min(len(job_soft), 3) if job_soft else 1 tech_score = ( min(1.0, len(matched_technical) / tech_denom) if job_technical else 0.50 ) soft_score = min(1.0, len(matched_soft) / soft_denom) if job_soft else 0.50 skill = min(1.0, (tech_score * 0.80) + (soft_score * 0.20)) # ── Component scores ───────────────────────────────────────────── cv_cat_lower = cv_category.strip().lower() if cv_category else "" job_cat_lower = job.category.strip().lower() if job.category else "" if cv_cat_lower and job_cat_lower and cv_cat_lower == job_cat_lower: cat_score = 1.0 else: cat_score = _soft_category_score( cv_skills, job.category, self._category_skills ) seniority, sen_reason = _seniority_fit(cv_seniority, job.seniority) title = self._title_similarity_precomputed( cv_title, cv_title_vec, job_title_vec ) experience = _experience_fit(cv_years, job.min_years_experience) # ── Weighted aggregation ───────────────────────────────────────── base_final = ( 0.42 * skill + 0.23 * semantic + 0.12 * title + 0.11 * cat_score + 0.07 * seniority + 0.05 * experience ) * 100 # Smooth Modifier applies universally if cat_score < 0.75: cat_modifier = -20.0 + ((cat_score - 0.30) / 0.45) * 20.0 cat_modifier = max(-20.0, min(0.0, cat_modifier)) else: cat_modifier = ((cat_score - 0.75) / 0.25) * 10.0 cat_modifier = max(0.0, min(10.0, cat_modifier)) final = base_final + cat_modifier # Guardrails if job_technical and not matched_technical: final = min(final, 34.0) if cv_cat_lower != job_cat_lower and skill < 0.30: final = min(final, 42.0) if ( job.seniority == "senior" and SENIORITY_ORDER.get(cv_seniority, 1) < SENIORITY_ORDER["mid"] ): final = min(final, 45.0) final = round(max(0.0, min(100.0, final)), 2) level = _match_level(final) # ── Reason string ──────────────────────────────────────────────── if matched_technical or matched_soft: reason = ( f"{level}: matched {len(matched_technical)}/{len(job_technical)} technical skill(s) and " f"{len(matched_soft)}/{len(job_soft)} soft skill(s)." ) else: reason = ( f"{level}: the CV does not show the main technical skills for this " f"role. It looks closer to {cv_category}." ) prediction = MatchPrediction( job_id=job.job_id, job_title=job.job_title, company_name=job.company_name, match_score=f"{final}%", match_level=level, cv_title=cv_title, cv_seniority=cv_seniority, cv_category=cv_category, cv_skills=sorted(cv_skills), cv_years_experience=cv_years, score_breakdown=ScoreBreakdown( skill_score=round(skill * 100, 2), semantic_score=round(semantic * 100, 2), title_score=round(title * 100, 2), category_score=round(cat_score * 100, 2), seniority_score=round(seniority * 100, 2), experience_score=round(experience * 100, 2), category_modifier=round(cat_modifier, 2), ), mismatch_analysis=MismatchAnalysis( reason=reason, matched_technical_skills=matched_technical, missing_technical_skills=missing_technical[:10], matched_soft_skills=matched_soft, missing_soft_skills=missing_soft[:10], cv_extra_strengths=extra, seniority_fit=sen_reason, category_fit=f"CV category: {cv_category}; Job category: {job.category}.", improvement_advice=_build_advice(missing_technical, job, cv_category), ), ) return prediction.model_dump() # ------------------------------------------------------------------ # Private helpers # ------------------------------------------------------------------ def _build_job_index_text(self, job: JobInput) -> str: """Build the semantic index text for a job (same as before, factored out).""" text = ( f"{job.job_title}\nCompany: {job.company_name}\n{job.description}\n" f"Technical skills: {', '.join(job.skills_technical)}\n" f"Soft skills: {', '.join(job.skills_soft)}" ) return text[: self._JOB_TEXT_MAX_CHARS] def _embed_batch_texts(self, texts: List[str]) -> np.ndarray: """ Embed a list of texts in a SINGLE batched encode() call. No manual word-chunking: bge-small-en-v1.5 supports sequences up to 512 tokens, and job text is already truncated to _JOB_TEXT_MAX_CHARS chars upstream, so it fits in one forward pass per item. """ if not texts: return np.zeros((0, 384), dtype="float32") safe_texts = [t if t else "" for t in texts] vectors = self._model.encode( safe_texts, normalize_embeddings=True, show_progress_bar=False, batch_size=32, ) return np.asarray(vectors, dtype="float32") def _embed(self, text: str) -> np.ndarray: """Encode a single text → normalised float32 vector (single forward pass).""" return self._embed_batch_texts([text])[0] def _embed_title_batch(self, titles: List[str]) -> np.ndarray: """Encode a list of titles in ONE batched encode() call with the title model.""" if not titles: return np.zeros((0, 384), dtype="float32") vectors = self._title_model.encode( titles, normalize_embeddings=True, show_progress_bar=False, batch_size=32 ) return np.asarray(vectors).astype("float32") def _embed_title(self, text: str) -> np.ndarray: """Encode a single title using the title model (used once per CV).""" vector = self._title_model.encode( text, normalize_embeddings=True, show_progress_bar=False ) return vector.astype("float32") def _semantic_skill_matches_precomputed( self, missing_skills: List[str], missing_skill_vecs: Dict[str, np.ndarray], cv_skills_vecs: np.ndarray, threshold: float = 0.72, ) -> Set[str]: """ Same semantics as the original _semantic_skill_matches, but takes precomputed vectors for the missing skills (built once per request in a single batched encode() call in predict_batch) instead of encoding them again per job. """ if not missing_skills or len(cv_skills_vecs) == 0 or not missing_skill_vecs: return set() matched: Set[str] = set() for skill in missing_skills: vec = missing_skill_vecs.get(skill) if vec is None: continue sims = np.dot(cv_skills_vecs, vec) if np.max(sims) >= threshold: matched.add(skill) return matched def _title_similarity_precomputed( self, cv_title: str, cv_title_vec: Optional[np.ndarray], job_title_vec: np.ndarray, ) -> float: """ Same semantics as the original _title_similarity, but uses the CV title vector cached once in cv_data and the job title vector computed once per request in the batched title encode() call — no model calls happen here. """ if not cv_title or cv_title == "Unknown" or cv_title_vec is None: return 0.35 return max(0.2, min(1.0, _cosine_similarity(cv_title_vec, job_title_vec)))