""" lib/evidence.py — V5 Structured Evidence Engine Extracts STRUCTURED evidence from candidate records — not raw snippets, not keyword counts, but typed evidence objects with company, role, metric, ownership, impact, and time context. Each evidence piece is scored independently, then ranked. The top-3 evidence pieces drive the reasoning output. Evidence types: - IMPACT: Quantified outcomes ("improved X by Y%", "reduced latency from A to B") - SCALE: System scale ("X million users", "YK QPS", "Z TB data") - RETRIEVAL: Retrieval/search system work with ownership context - PRODUCTION: Deployment to production with ownership - EVALUATION: Evaluation framework design or usage - PRE_LLM: Pre-2022 IR/search/ranking experience - DEPTH: Career depth in relevant domain - OWNERSHIP: Ownership verbs (architected, owned, led, built) """ from __future__ import annotations import re import hashlib from dataclasses import dataclass, field from lib import schema from lib.jd_parser import get_jd from lib.constants import REFERENCE_DATE @dataclass class Evidence: """A single structured evidence piece.""" type: str # impact, scale, retrieval, production, evaluation, pre_llm, depth, ownership score: float # 0-20, how strong this evidence is company: str = "" role: str = "" metric: str = "" # The actual metric/value extracted ownership: str = "" # What ownership verb was detected context: str = "" # Generated summary (NEVER raw text) year_range: str = "" # e.g. "2020-2023" domain: str = "" # Which JD domain this maps to # --------------------------------------------------------------------------- # Ownership hierarchy weights # --------------------------------------------------------------------------- _OWNERSHIP_TIERS = { "architected": 1.00, "architecture": 0.98, "designed the architecture": 1.00, "spearheaded": 0.95, "pioneered": 0.94, "owned": 0.85, "owner of": 0.85, "sole owner": 0.88, "led": 0.82, "led the": 0.82, "team lead": 0.80, "designed": 0.75, "designed and": 0.76, "built": 0.70, "built the": 0.70, "built a": 0.68, "developed": 0.65, "developed a": 0.64, "implemented": 0.55, "implemented a": 0.54, "created": 0.52, "created the": 0.52, "contributed": 0.30, "contributed to": 0.30, "worked on": 0.25, "part of": 0.22, "used": 0.15, "utilized": 0.15, "worked with": 0.15, "experienced in": 0.12, "familiar with": 0.10, "exposure to": 0.08, "knowledge of": 0.08, "aware of": 0.05, } # Pre-compile ownership patterns (longest first to avoid partial matches) _OWNERSHIP_PATTERNS = sorted(_OWNERSHIP_TIERS.keys(), key=len, reverse=True) _OWNERSHIP_RE = re.compile( r'\b(' + '|'.join(re.escape(p) for p in _OWNERSHIP_PATTERNS) + r')\b', re.IGNORECASE, ) # Impact regex patterns (numbers-first) _IMPACT_PATTERNS = [ # "improved X by Y%" (r'improved\s+\w+\s+by\s+(\d+(?:\.\d+)?)\s*%', 0.9, "improvement_pct"), # "reduced latency from Xms to Yms" (r'reduced\s+latency\s+from\s+(\d+)\s*ms\s+to\s+(\d+)\s*ms', 1.0, "latency_reduction"), # "Nms p99" or "p99 latency of Nms" (r'(?:p99|p95|tail)\s*(?:latency\s*)?(?:of\s*)?(\d+)\s*ms', 0.85, "tail_latency"), # "NDCG at 0.XX" or "NDCG improved to 0.XX" (r'ndcg.*?(0\.\d{2,3})', 0.95, "ndcg_score"), # "Recall@K improved to X%" (r'recall@?\d+\s*(?:improved\s*)?(?:to\s*)?(\d+(?:\.\d+)?)\s*%', 0.90, "recall_pct"), # "X million users" or "X million daily active" (r'(\d+(?:\.\d+)?)\s*million\s+(?:daily\s+)?(?:active\s+)?users', 0.80, "user_scale_m"), # "XK QPS" or "XK requests per second" (r'(\d+(?:\.\d+)?)\s*k?\s*(?:qps|rps|requests?\s*per\s*sec)', 0.85, "qps"), # "X% increase in Y" (r'(\d+(?:\.\d+)?)\s*%\s*increase\s+in\s+\w+', 0.80, "increase_pct"), # "Xx throughput" or "X times throughput" (r'(\d+(?:\.\d+)?)\s*x\s+(?:throughput|improvement|speedup)', 0.85, "throughput_x"), # "served X requests" or "handling X requests" (r'(?:served|handling|processing)\s+(\d[\d,]*)\s*(?:requests?|queries?|events?)', 0.70, "request_volume"), # "X TB of data" or "X GB" (r'(\d+)\s*(?:TB|GB)\s+of\s+data', 0.60, "data_scale"), ] # Retrieval system keywords _RETRIEVAL_KEYWORDS = [ "bm25", "elasticsearch", "opensearch", "faiss", "pinecone", "weaviate", "qdrant", "milvus", "vector database", "vector db", "hybrid search", "hybrid retrieval", "semantic search", "dense retrieval", "sparse retrieval", "ann", "approximate nearest neighbor", "embedding index", "inverted index", "query understanding", "re-ranking", "reranking", "cross-encoder", "bi-encoder", "colbert", "hnsw", "ivf", "product quantization", ] # Evaluation keywords _EVALUATION_KEYWORDS = [ "ndcg", "mrr", "map@", "precision@", "recall@", "a/b test", "ab test", "offline evaluation", "online evaluation", "evaluation framework", "evaluation pipeline", "offline-to-online", "ranking quality", "relevance judgment", "annotation", ] # Production deployment keywords _PRODUCTION_KEYWORDS = [ "production", "deployed", "shipped", "launched", "live traffic", "real users", "at scale", "rollout", "on-call", "serving", "end-to-end", "p99", "sla", ] def extract_all_evidence(c: dict) -> list[Evidence]: """ Extract ALL evidence pieces from a candidate, score each, return sorted by score descending. """ jd = get_jd() ch = schema.career_history(c) all_evidence: list[Evidence] = [] for role in ch: company = role.get("company", "") title = role.get("title", "") desc = (role.get("description") or "").lower() start = schema.parse_date(role.get("start_date")) end = schema.parse_date(role.get("end_date")) or REFERENCE_DATE year_range = f"{start.year if start else '?'}-{end.year}" is_current = role.get("is_current", False) duration = role.get("duration_months", 0) or 0 role_text = f"{title} {desc}" # 1. IMPACT evidence (score 15-20) impact_ev = _extract_impact(role_text, company, title, year_range) all_evidence.extend(impact_ev) # 2. SCALE evidence (score 11-15) scale_ev = _extract_scale(role_text, company, title, year_range) all_evidence.extend(scale_ev) # 3. RETRIEVAL evidence (score 12-18) retrieval_ev = _extract_retrieval(role_text, company, title, year_range, jd) all_evidence.extend(retrieval_ev) # 4. PRODUCTION evidence (score 9-15) prod_ev = _extract_production(role_text, company, title, year_range) all_evidence.extend(prod_ev) # 5. EVALUATION evidence (score 10-13) eval_ev = _extract_evaluation(role_text, company, title, year_range) all_evidence.extend(eval_ev) # 6. PRE-LLM evidence (score 8-12) if start and start.year < jd.pre_llm_cutoff_year: # Check not contaminated by post-LLM markers if not any(m in desc for m in jd.post_llm_markers): pre_llm_ev = _extract_pre_llm(role_text, company, title, year_range, jd) all_evidence.extend(pre_llm_ev) # 7. OWNERSHIP evidence (score 8-14) owner_ev = _extract_ownership(role_text, company, title, year_range) all_evidence.extend(owner_ev) # 8. CAREER DEPTH evidence (score 8-12) depth_ev = _extract_career_depth(c, jd) all_evidence.extend(depth_ev) # Sort by score descending, take top evidence all_evidence.sort(key=lambda e: e.score, reverse=True) return all_evidence def get_top_evidence(c: dict, n: int = 3) -> list[Evidence]: """Get top-N evidence pieces for reasoning.""" return extract_all_evidence(c)[:n] def get_evidence_summary(c: dict) -> dict: """Get aggregated evidence statistics for scoring.""" all_ev = extract_all_evidence(c) if not all_ev: return { "best_score": 0, "count": 0, "has_impact": False, "has_scale": False, "has_retrieval": False, "has_production": False, "has_evaluation": False, "has_pre_llm": False, "has_ownership": False, "total_score": 0, "top3_avg": 0, } types_present = {ev.type for ev in all_ev} top3 = all_ev[:3] return { "best_score": all_ev[0].score, "count": len(all_ev), "has_impact": "impact" in types_present, "has_scale": "scale" in types_present, "has_retrieval": "retrieval" in types_present, "has_production": "production" in types_present, "has_evaluation": "evaluation" in types_present, "has_pre_llm": "pre_llm" in types_present, "has_ownership": "ownership" in types_present, "total_score": sum(e.score for e in all_ev), "top3_avg": sum(e.score for e in top3) / len(top3), "top_evidence": top3, } # --------------------------------------------------------------------------- # Individual extractors # --------------------------------------------------------------------------- def _extract_impact(text: str, company: str, role: str, year_range: str) -> list[Evidence]: """Extract quantified impact metrics.""" evidence_list = [] for pattern, weight, metric_type in _IMPACT_PATTERNS: m = re.search(pattern, text, re.IGNORECASE) if m: groups = [g for g in m.groups() if g is not None] metric_val = " / ".join(groups) # Generate a human-readable summary summaries = { "improvement_pct": f"Achieved {metric_val}% improvement in a key metric", "latency_reduction": f"Reduced latency from {groups[0]}ms to {groups[1]}ms" if len(groups) >= 2 else f"Improved latency to {groups[0]}ms", "tail_latency": f"Operating at {metric_val}ms tail latency", "ndcg_score": f"Attained NDCG of {metric_val}", "recall_pct": f"Reached {metric_val}% recall in retrieval evaluation", "user_scale_m": f"Served {metric_val} million users", "qps": f"Handled {metric_val}K queries per second", "increase_pct": f"Drove {metric_val}% increase in a core metric", "throughput_x": f"Achieved {metric_val}x throughput improvement", "request_volume": f"Processed {metric_val} requests", "data_scale": f"Managed {metric_val} of data at scale", } ctx = summaries.get(metric_type, f"Measured {metric_val} impact") # Detect ownership verb ownership = _detect_ownership(text) bonus = 1.0 + (0.15 if ownership and _OWNERSHIP_TIERS.get(ownership.lower(), 0) >= 0.75 else 0) evidence_list.append(Evidence( type="impact", score=15 * weight * bonus, company=company, role=role, metric=metric_val, ownership=ownership or "", context=ctx, year_range=year_range, domain="general", )) # Deduplicate: keep highest score per metric_type seen = {} for ev in evidence_list: mt = ev.metric if mt not in seen or ev.score > seen[mt].score: seen[mt] = ev return list(seen.values()) def _extract_scale(text: str, company: str, role: str, year_range: str) -> list[Evidence]: """Extract evidence of system scale.""" evidence_list = [] # Scale patterns scale_patterns = [ (r'(\d+(?:\.\d+)?)\s*million\s+(?:monthly|daily)?\s*(?:active\s+)?users?', 0.90, "user_scale"), (r'(\d[\d,]*)\s*(?:employees?|engineers?|team members?)', 0.50, "team_scale"), (r'(\d+)(?:\s*-\s*\d+)?\s*(?:tb|gb|pb)\s+(?:of\s+)?(?:data|storage)', 0.70, "data_scale"), (r'(\d[\d,]*)\s*(?:micro)?services?', 0.55, "service_count"), (r'(\d+)\s*(?:servers?|nodes?|instances?)', 0.50, "infra_scale"), ] for pattern, weight, scale_type in scale_patterns: m = re.search(pattern, text, re.IGNORECASE) if m: metric_val = m.group(1) summaries = { "user_scale": f"System serving {metric_val}M+ users", "team_scale": f"Engineering org of {metric_val}+", "data_scale": f"Data infrastructure at {metric_val} scale", "service_count": f"{metric_val}+ microservices managed", "infra_scale": f"Infrastructure of {metric_val}+ nodes", } evidence_list.append(Evidence( type="scale", score=11 + 4 * weight, company=company, role=role, metric=metric_val, ownership=_detect_ownership(text) or "", context=summaries.get(scale_type, f"Scale: {metric_val}"), year_range=year_range, )) return evidence_list[:2] # Cap at 2 scale evidence pieces def _extract_retrieval(text: str, company: str, role: str, year_range: str, jd) -> list[Evidence]: """Extract retrieval/search system evidence.""" found_kw = [kw for kw in _RETRIEVAL_KEYWORDS if kw in text] if not found_kw: return [] ownership = _detect_ownership(text) ownership_weight = _OWNERSHIP_TIERS.get(ownership.lower(), 0.3) if ownership else 0.3 # Summarize retrieval context (never raw text) tech_list = found_kw[:3] if len(found_kw) > 3: tech_list.append(f"+{len(found_kw)-3} more") # Classify retrieval type has_vector = any(kw in found_kw for kw in ["faiss", "pinecone", "weaviate", "qdrant", "milvus", "vector db", "vector database"]) has_hybrid = any(kw in found_kw for kw in ["hybrid search", "hybrid retrieval", "bm25", "elasticsearch", "opensearch"]) has_dense = any(kw in found_kw for kw in ["dense retrieval", "semantic search", "embedding index", "ann"]) if has_hybrid and has_vector: rtype = "hybrid vector search" elif has_hybrid: rtype = "hybrid text search" elif has_vector or has_dense: rtype = "dense vector retrieval" else: rtype = "search infrastructure" action = _ownership_action(ownership) ctx = f"{action} a {rtype} system leveraging {', '.join(tech_list)}" score = 12 + 6 * ownership_weight if has_hybrid and has_vector: score += 2 # Bonus for full stack return [Evidence( type="retrieval", score=min(20, score), company=company, role=role, metric=", ".join(found_kw[:3]), ownership=ownership or "", context=ctx, year_range=year_range, domain="search", )] def _extract_production(text: str, company: str, role: str, year_range: str) -> list[Evidence]: """Extract production deployment evidence.""" found_kw = [kw for kw in _PRODUCTION_KEYWORDS if kw in text] if not found_kw: return [] ownership = _detect_ownership(text) ownership_weight = _OWNERSHIP_TIERS.get(ownership.lower(), 0.3) if ownership else 0.3 # Summarize what was deployed prod_type = "production system" if "end-to-end" in text: prod_type = "end-to-end production system" elif "live traffic" in text: prod_type = "system serving live user traffic" elif "at scale" in text: prod_type = "system at production scale" action = _ownership_action(ownership) ctx = f"{action} a {prod_type}" score = 9 + 6 * ownership_weight return [Evidence( type="production", score=min(18, score), company=company, role=role, metric=", ".join(found_kw[:3]), ownership=ownership or "", context=ctx, year_range=year_range, )] def _extract_evaluation(text: str, company: str, role: str, year_range: str) -> list[Evidence]: """Extract evaluation framework evidence.""" found_kw = [kw for kw in _EVALUATION_KEYWORDS if kw in text] if not found_kw: return [] ownership = _detect_ownership(text) ownership_weight = _OWNERSHIP_TIERS.get(ownership.lower(), 0.3) if ownership else 0.3 has_ndcg = "ndcg" in found_kw has_ab = any("a/b" in kw or "ab" in kw for kw in found_kw) eval_type = "evaluation framework" if has_ndcg and has_ab: eval_type = "comprehensive evaluation framework (NDCG + A/B testing)" elif has_ndcg: eval_type = "NDCG-based evaluation framework" elif has_ab: eval_type = "A/B testing evaluation" action = _ownership_action(ownership) ctx = f"{action} an {eval_type}" score = 10 + 3 * ownership_weight if has_ndcg and has_ab: score += 2 return [Evidence( type="evaluation", score=min(15, score), company=company, role=role, metric=", ".join(found_kw[:3]), ownership=ownership or "", context=ctx, year_range=year_range, domain="evaluation", )] def _extract_pre_llm(text: str, company: str, role: str, year_range: str, jd) -> list[Evidence]: """Extract pre-LLM IR/search experience.""" found_kw = [kw for kw in jd.pre_llm_keywords if kw in text] if not found_kw: return [] ownership = _detect_ownership(text) ownership_weight = _OWNERSHIP_TIERS.get(ownership.lower(), 0.3) if ownership else 0.3 action = _ownership_action(ownership) ctx = f"{action} pre-2022 search/ranking infrastructure ({', '.join(found_kw[:2])})" score = 8 + 4 * ownership_weight return [Evidence( type="pre_llm", score=min(14, score), company=company, role=role, metric=", ".join(found_kw[:2]), ownership=ownership or "", context=ctx, year_range=year_range, domain="search", )] def _extract_ownership(text: str, company: str, role: str, year_range: str) -> list[Evidence]: """Extract high-ownership evidence.""" m = _OWNERSHIP_RE.search(text) if not m: return [] verb = m.group(1).lower() weight = _OWNERSHIP_TIERS.get(verb, 0.3) if weight < 0.70: # Only report strong ownership return [] action = _ownership_action(verb) ctx = f"{action} a significant technical initiative" return [Evidence( type="ownership", score=8 + 6 * weight, company=company, role=role, metric=verb, ownership=verb, context=ctx, year_range=year_range, )] def _extract_career_depth(c: dict, jd) -> list[Evidence]: """Extract career depth in relevant domain.""" ch = schema.career_history(c) total_months = sum((r.get("duration_months") or 0) for r in ch) or 1 # Build relevant keyword set from JD relevant_kw = set() for skills in jd.required_skills.values(): relevant_kw.update(skills) for skills in jd.preferred_skills.values(): relevant_kw.update(skills) relevant_kw.update(jd.pre_llm_keywords) relevant_months = 0 for r in ch: role_text = f"{r.get('title','')} {r.get('description','')}".lower() if any(kw in role_text for kw in relevant_kw): relevant_months += (r.get("duration_months") or 0) ratio = relevant_months / total_months if ratio < 0.3: return [] years_relevant = relevant_months / 12 ctx = f"Spent {years_relevant:.1f} of {total_months/12:.1f} total career years in the JD's core domain" return [Evidence( type="depth", score=8 + 4 * ratio, company="", role="", metric=f"{ratio:.0%} of career in domain", context=ctx, year_range="", )] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _detect_ownership(text: str) -> str | None: """Detect the strongest ownership verb in text.""" m = _OWNERSHIP_RE.search(text) if m: return m.group(1).lower() return None def _ownership_action(verb: str | None) -> str: """Convert ownership verb to action phrase for reasoning.""" if not verb: return "Worked on" v = verb.lower() weight = _OWNERSHIP_TIERS.get(v, 0.3) if weight >= 0.90: return "Architected and led" elif weight >= 0.75: return "Designed and built" elif weight >= 0.60: return "Led development of" elif weight >= 0.45: return "Built and shipped" elif weight >= 0.30: return "Contributed to" return "Worked on"