import json import math import os from collections import Counter from pathlib import Path from src.resume_parser import contains try: from src.conversion import attach_conversion_scores except Exception: # pragma: no cover — conversion 模块不存在时回退 attach_conversion_scores = None # --------------------------------------------------------------------------- # 语义匹配器（可选，sentence_transformers 未安装时回退到 token 相似度） # --------------------------------------------------------------------------- _SEMANTIC_MATCHER = None _ENABLE_EMBEDDING_CLIENT = os.getenv("ENABLE_EMBEDDING_CLIENT", "").lower() in {"1", "true", "yes"} _ENABLE_LOCAL_SEMANTIC = os.getenv("ENABLE_LOCAL_SEMANTIC_MATCHER", "").lower() in {"1", "true", "yes"} _TRY_LOAD_SEMANTIC = False # 默认不加载，避免自动下载模型 # 主排序里的 embedding 能力必须显式开启，避免 API key 或缓存模型改变 core eval 结果。 if _ENABLE_LOCAL_SEMANTIC and os.getenv("SEMANTIC_MODEL_PATH"): _TRY_LOAD_SEMANTIC = True def _get_semantic_matcher(): """Lazy-load SemanticMatcher（避免导入时即失败）。""" global _SEMANTIC_MATCHER, _TRY_LOAD_SEMANTIC if not _TRY_LOAD_SEMANTIC: return _SEMANTIC_MATCHER _TRY_LOAD_SEMANTIC = False try: from src.semantic_matcher import SemanticMatcher _SEMANTIC_MATCHER = SemanticMatcher() print("[OK] SemanticMatcher loaded — using embedding-based semantic recall.") except Exception as exc: print(f"[WARN] SemanticMatcher not available ({exc}) — falling back to token cosine.") _SEMANTIC_MATCHER = None return _SEMANTIC_MATCHER def semantic_similarity(resume_text: str, job_text: str) -> float: """语义相似度（优先 API/client，回退 embedding，最后 token cosine）。""" # 1) EmbeddingClient 默认关闭，避免误用 API 或自动加载未声明的 embedding 模型。 if _ENABLE_EMBEDDING_CLIENT: try: from src.embedding_client import get_embedding_client _CLIENT = get_embedding_client() if _CLIENT.is_available(): sim = _CLIENT.semantic_similarity(resume_text, job_text) if sim is not None: return sim except Exception: pass # 2) SemanticMatcher（本地 sentence-transformers） matcher = _get_semantic_matcher() if matcher is not None: try: return matcher.compute_similarity(resume_text, job_text) except Exception: pass # 3) fallback：token cosine return cosine_similarity(resume_text, job_text) WEIGHTS = { "技能匹配": 0.28, "项目匹配": 0.25, "经历匹配": 0.10, "方向匹配": 0.27, "成长潜力": 0.10, } def clamp_score(value: float) -> int: return max(0, min(100, round(value))) def tokenize(text: str) -> list[str]: text = (text or "").lower() tokens = [] buff = [] for char in text: if char.isascii() and char.isalnum(): buff.append(char) else: if buff: tokens.append("".join(buff)) buff = [] if "\u4e00" <= char <= "\u9fff": tokens.append(char) if buff: tokens.append("".join(buff)) return tokens def cosine_similarity(left: str, right: str) -> float: left_counter = Counter(tokenize(left)) right_counter = Counter(tokenize(right)) if not left_counter or not right_counter: return 0.0 keys = set(left_counter) | set(right_counter) dot = sum(left_counter[k] * right_counter[k] for k in keys) left_norm = math.sqrt(sum(v * v for v in left_counter.values())) right_norm = math.sqrt(sum(v * v for v in right_counter.values())) return dot / (left_norm * right_norm) def load_jobs(jobs_path: Path) -> list[dict]: with jobs_path.open("r", encoding="utf-8") as file: return json.load(file) def overlap(required: list[str], actual: list[str], resume_text: str) -> list[str]: return [ term for term in required if term in actual or contains(resume_text, term) ] def score_job(job: dict, profile: dict, resume_text: str, target_role: str, target_city: str, stage: str) -> dict: matched_skills = overlap(job["skills"], profile["skills"], resume_text) missing_skills = [skill for skill in job["skills"] if skill not in matched_skills] matched_project = overlap(job["project_signals"], profile["project_signals"], resume_text) semantic_recall = semantic_similarity(resume_text, f"{job['title']} {job['jd']} {' '.join(job['skills'])}") skill_score = clamp_score((len(matched_skills) / max(len(job["skills"]), 1)) * 100) project_score = clamp_score( (len(matched_project) / max(len(job["project_signals"]), 1)) * 100 + (8 if profile["has_metrics"] else 0) + semantic_recall * 12 ) experience_score = 86 if job["stage"] == stage else 68 direction_score = 92 if job["direction"] == target_role else 40 city_score = 88 if target_city == "不限" else 96 if job["city"] == target_city else 62 preference_score = clamp_score(0.72 * direction_score + 0.28 * city_score) growth_score = clamp_score( 48 + min(len(profile["skills"]), 10) * 3 + (12 if profile["has_llm_project"] else 0) + (8 if profile["has_rec_project"] else 0) ) dimensions = { "技能匹配": skill_score, "项目匹配": project_score, "经历匹配": experience_score, "方向匹配": preference_score, "成长潜力": growth_score, } match_score = clamp_score(sum(dimensions[name] * weight for name, weight in WEIGHTS.items())) # direction match bonus: 目标方向岗位获得额外加分 if job["direction"] == target_role: match_score = clamp_score(match_score + 15) # title match bonus: 精确匹配 target_role 的岗位加分最高；含额外方向词（如 Agent）的次之 import re as _re title_lower = job["title"].lower() target_lower = target_role.lower() title_exact_match = False if target_lower in title_lower: # 精确匹配：title 去掉「实习生/工程师」后缀后与 target_role 高度一致 title_clean = _re.sub(r"(实习生|工程师|生|员)$", "", job["title"]).lower() target_clean = _re.sub(r"(实习生|工程师|生|员)$", "", target_role).lower() if target_clean in title_clean or title_clean in target_clean: match_score = clamp_score(match_score + 20) title_exact_match = True else: # 模糊匹配（如 target=大模型应用算法，title=大模型 Agent 应用实习生） match_score = clamp_score(match_score + 6) else: title_exact_match = False # 迁移型强匹配补偿：当用户目标方向较宽，但某个非目标方向岗位在技能和项目上 # 明显更贴合简历时，不能只因 direction 字段不同被平台/泛算法岗位压下去。 if job["direction"] != target_role and skill_score >= 75 and project_score >= 70: match_score = clamp_score(match_score + 5) # 基础返回值 scored = dict(job) scored.update( { "match_score": match_score, "score": match_score, # 保留兼容 "growth_score": growth_score, "semantic_recall": round(semantic_recall, 4), "target_role": target_role, "target_city": target_city, "target_stage": stage, "matched_skills": matched_skills, "missing_skills": missing_skills, "dimensions": dimensions, "strengths": build_strengths(job, matched_skills, profile), "gaps": build_gaps(job, missing_skills, profile), "rewrites": build_rewrites(job, matched_skills, missing_skills), "interview_plan": build_interview_plan(job, missing_skills), } ) # 附加 conversion 分数（PassScore / RiskScore / KeywordCoverage / GrowthScore） if attach_conversion_scores is not None: try: conv = attach_conversion_scores( profile=profile, job=scored, resume_text=resume_text, target_role=target_role, target_city=target_city, stage=stage, ) # 安全赋值：过滤 None 值（conv 返回值可能包含显式 None） _conv_map = { "pass_score": 50, "risk_score": 50, "growth_score": growth_score, "keyword_coverage": 0.0, } for key, default in _conv_map.items(): val = conv.get(key) if isinstance(conv, dict) else None # 关键：val is None 时用 default（不仅 key 不存在，值本身为 None 也要覆盖） scored[key] = default if val is None else val except Exception as e: import sys print(f"[WARN] attach_conversion_scores failed: {e}", file=sys.stderr) scored.setdefault("pass_score", 50) scored.setdefault("risk_score", 50) scored.setdefault("growth_score", growth_score) scored.setdefault("keyword_coverage", 0.0) else: scored.setdefault("pass_score", 50) scored.setdefault("risk_score", 50) scored.setdefault("keyword_coverage", 0.0) # 计算 ApplyPriority（确保 growth_score 不为 None） ps = scored.get("pass_score", 50) rs = scored.get("risk_score", 50) gs = scored.get("growth_score") if gs is None: gs = growth_score # 使用 score_job 本地计算的 growth_score scored["growth_score"] = gs # 写回字典，防止后续读者读到 None ms = scored.get("match_score", match_score) apply_priority = round(0.40 * ms + 0.30 * ps - 0.15 * rs + 0.15 * gs, 2) # title_exact_match bonus：精确匹配 target_role 的岗位在 apply_priority 上加权重， # 防止 match_score 被 clamp_score 封顶后无法区分精确/模糊匹配 if title_exact_match: apply_priority = round(apply_priority + 1.5, 2) scored["apply_priority"] = apply_priority # 附加证据链（Evidence Chain） try: from src.evidence import attach_evidence scored = attach_evidence(scored, resume_text, profile) except Exception: pass # 标记 title 精确匹配，供排序破 tie 使用 scored["title_exact_match"] = title_exact_match return scored def rank_jobs( resume_text: str, profile: dict, target_role: str, target_city: str, stage: str, top_k: int, jobs_path: Path, ) -> list[dict]: """ 返回全部打分后的岗位列表（按 apply_priority 降序）。 调用方可根据需要用 sorted() 构造 by_match / by_priority 两个榜单。 """ jobs = load_jobs(jobs_path) scored_jobs = [ score_job(job, profile, resume_text, target_role, target_city, stage) for job in jobs ] # 默认按 apply_priority 降序排列，title_exact_match=True 的岗位排在同分前 scored_jobs.sort( key=lambda item: ( item.get("apply_priority", item.get("score", 0)), int(item.get("title_exact_match", False)), ), reverse=True, ) return scored_jobs[:top_k] def rank_job_list( resume_text: str, profile: dict, target_role: str, target_city: str, stage: str, top_k: int, jobs: list[dict], ) -> list[dict]: """ 对传入的 jobs list 打分排序（不读取文件）。 与 rank_jobs 参数兼容，jobs_path 替换为 jobs 列表。 """ scored_jobs = [ score_job(job, profile, resume_text, target_role, target_city, stage) for job in jobs ] scored_jobs.sort( key=lambda item: ( item.get("apply_priority", item.get("score", 0)), int(item.get("title_exact_match", False)), ), reverse=True, ) return scored_jobs[:top_k] def build_strengths(job: dict, matched_skills: list[str], profile: dict) -> list[str]: strengths = [] if matched_skills: strengths.append(f"命中核心技能：{'、'.join(matched_skills[:6])}，具备进入岗位精排池的基础。") if profile["has_llm_project"] and "大模型" in job["direction"]: strengths.append("简历中已有 LLM/RAG/Agent 信号，能对齐大模型应用算法岗位。") if profile["has_rec_project"] and ("推荐" in job["direction"] or "匹配" in job["jd"]): strengths.append("推荐系统、召回排序、语义表征经历可以迁移到人岗匹配推荐。") if profile["has_metrics"]: strengths.append("项目描述包含 NDCG、HitRate、TopK 等指标，便于证明算法效果。") return strengths or ["当前简历具备基础技术栈，但需要补充更明确的项目证据。"] def build_gaps(job: dict, missing_skills: list[str], profile: dict) -> list[str]: gaps = [] if missing_skills: gaps.append(f"岗位还要求 {'、'.join(missing_skills[:5])}，建议补充真实项目或技能证据。") if not profile["has_metrics"]: gaps.append("项目缺少可量化指标，容易被认为只是功能实现。") if "Agent" in job["jd"] and "Agent" not in profile["project_signals"]: gaps.append("Agent 工作流表达不够突出，需要说明任务拆解、工具调用和失败兜底。") return gaps or ["暂无明显硬伤，建议继续压缩表达并突出岗位关键词。"] def build_rewrites(job: dict, matched_skills: list[str], missing_skills: list[str]) -> list[dict]: skills = job.get("skills") or ["岗位相关技能"] main_skill = matched_skills[0] if matched_skills else skills[0] missing = missing_skills[0] if missing_skills else skills[-1] return [ { "before": "负责推荐系统建模和模型结果分析。", "after": f"围绕 {job['direction']} 场景，使用 {main_skill} 构建候选召回与精排链路，并通过 NDCG@10 / TopK 命中率评估模型收益。", }, { "before": "做过一个 LLM 求职助手 Demo。", "after": "设计多 Agent 求职匹配流程，将简历解析、JD 理解、岗位排序、能力缺口诊断和简历优化拆成可解释节点。", }, { "before": "熟悉大模型相关技术。", "after": f"熟悉 {'、'.join(skills[:4])}，可独立完成 Prompt 模板、结构化输出、检索增强和结果评估配置。", }, { "before": "技能栏可以继续补充。", "after": f"若确有实践，建议补充 {missing}，并用一次实验、接口调用或离线评估证明不是仅停留在概念层。", }, ] def build_interview_plan(job: dict, missing_skills: list[str]) -> list[str]: themes = job.get("interview_themes", ["算法基础", "项目深挖", "系统设计", "业务理解"]) if not themes or len(themes) < 2: themes = ["算法基础", "项目深挖", "系统设计", "业务理解"] skills = job.get("skills") or ["岗位相关技能"] focus = missing_skills[:3] or skills[:3] if not focus: focus = ["岗位相关技能"] return [ f"第 1 天：复盘 {job['direction']} 岗位 JD，整理关键词和项目证据。", f"第 2 天：准备 {themes[0]}，讲清输入、模型、输出和指标。", f"第 3 天：补齐 {focus[0]} 的实践案例，写成可追问的简历 bullet。", f"第 4 天：准备 {themes[1]}，重点讲失败 case 和优化方案。", f"第 5 天：针对 {job['company']} 场景模拟 3 个算法方案题。", "第 6 天：做一次 20 分钟项目深挖 mock，覆盖数据、模型、评估、落地。", "第 7 天：压缩成 1 分钟自我介绍、3 分钟项目介绍和 5 个高频追问答案。", ]