Rezi AI智能简历生成产品设计

import jsonimport refrom typing import Any, Dict, List, Optional, Tuple  class JDParser:    """JD结构化解析：提取技能要求、职责描述和任职资格"""     def __init__(self, llm_call=None):        self.llm_call = llm_call or self._mock_llm     def parse(self, jd_text: str) -> Dict[str, Any]:        """将非结构化JD文本解析为结构化画像"""        prompt = self._build_parse_prompt(jd_text)        result = self.llm_call(prompt)        return self._validate_schema(result)     def _build_parse_prompt(self, jd_text: str) -> str:        return json.dumps({            "instruction": "解析JD，按JSON Schema输出",            "schema": {                "skills": "技能要求列表",                "responsibilities": "职责描述列表",                "qualifications": "任职资格列表"            },            "jd_text": jd_text[:2000]        }, ensure_ascii=False)     def _validate_schema(self, result: Dict) -> Dict:        for key in ("skills", "responsibilities", "qualifications"):            if key not in result:                result[key] = []        return result     def _mock_llm(self, prompt: str) -> Dict:        return {            "skills": ["Kubernetes", "微服务架构", "高并发", "分布式系统"],            "responsibilities": ["负责后端服务架构设计", "支撑高并发请求"],            "qualifications": ["5年后端开发经验", "熟悉云原生技术栈"]        }  class SemanticMatcher:    """用户经历与JD要求的语义匹配度计算"""     def __init__(self, similarity_fn=None):        self.similarity_fn = similarity_fn or self._keyword_similarity     def match(self, experiences: List[str], jd_profile: Dict) -> List[Dict]:        """计算每条经历与JD的匹配度，按匹配度排序"""        scored = []        jd_keywords = self._flatten_jd(jd_profile)        for exp in experiences:            score = self._calc_match_score(exp, jd_keywords)            scored.append({"experience": exp, "score": score})        scored.sort(key=lambda x: x["score"], reverse=True)        return scored     def _calc_match_score(self, exp: str, keywords: List[str]) -> float:        total = 0.0        for kw in keywords:            total += self.similarity_fn(exp, kw)        return round(total / max(len(keywords), 1), 3)     def _keyword_similarity(self, text: str, keyword: str) -> float:        return 1.0 if keyword.lower() in text.lower() else 0.0     def _flatten_jd(self, profile: Dict) -> List[str]:        kws = []        for v in profile.values():            if isinstance(v, list):                kws.extend(v)        return kws  class ATSOptimizer:    """ATS关键词覆盖率检测与等价替换优化"""     def __init__(self, llm_call=None):        self.llm_call = llm_call or self._mock_replace     def analyze_coverage(self, resume_text: str, keywords: List[str]) -> Dict:        """检测简历对JD关键词的覆盖率"""        hit, missing = [], []        for kw in keywords:            if kw.lower() in resume_text.lower():                hit.append(kw)            else:                missing.append(kw)        coverage = round(len(hit) / max(len(keywords), 1), 2)        return {"coverage": coverage, "hit": hit, "missing": missing}     def optimize(self, resume_text: str, missing: List[str]) -> str:        """对缺失关键词进行语义等价替换"""        optimized = resume_text        for kw in missing:            replacement = self.llm_call(optimized, kw)            if replacement:                optimized = replacement        return optimized     def _mock_replace(self, text: str, keyword: str) -> str:        if keyword == "Kafka" and "数据处理" in text:            return text.replace("数据处理", "基于Kafka的数据处理")        if keyword == "高并发" and "后端开发" in text:            return text.replace("后端开发", "高并发后端开发")        return text  class ResumeScorer:    """简历质量多维评估：量化度、具体度、连贯度、规范度"""     VAGUE_PATTERNS = [r"参与了多个", r"负责相关工作", r"进行了优化"]     def score(self, resume: Dict) -> Dict[str, Any]:        quant = self._score_quantification(resume)        specific = self._score_specificity(resume)        coherent = self._score_coherence(resume)        standard = self._score_standardization(resume)        total = round(quant * 0.35 + specific * 0.25 +                      coherent * 0.2 + standard * 0.2)        return {            "quantification": quant, "specificity": specific,            "coherence": coherent, "standardization": standard,            "total": total        }     def _score_quantification(self, resume: Dict) -> int:        experiences = resume.get("experiences", [])        has_num = sum(1 for e in experiences                      if re.search(r"\d+[%万千亿]|\d+人|\d+个", e))        ratio = has_num / max(len(experiences), 1)        return int(ratio * 100)     def _score_specificity(self, resume: Dict) -> int:        experiences = resume.get("experiences", [])        vague_count = sum(1 for e in experiences                          if any(re.search(p, e) for p in self.VAGUE_PATTERNS))        ratio = 1 - vague_count / max(len(experiences), 1)        return int(ratio * 100)     def _score_coherence(self, resume: Dict) -> int:        dates = resume.get("dates", [])        if len(dates) < 2:            return 100        gaps = sum(1 for i in range(1, len(dates))                   if dates[i][0] - dates[i - 1][1] > 180)        ratio = 1 - gaps / max(len(dates) - 1, 1)        return int(ratio * 100)     def _score_standardization(self, resume: Dict) -> int:        score = 100        if not resume.get("email"):            score -= 20        if not resume.get("phone"):            score -= 20        if not resume.get("education"):            score -= 15        return max(score, 0)  def demo_rezi_pipeline():    """演示Rezi AI简历生成全流程"""    jd_text = """高级后端工程师，要求精通微服务架构和分布式系统，    熟悉Kubernetes容器化部署，能支撑高并发场景。5年后端经验。"""     user_resume = {        "experiences": [            "参与后端开发工作",            "参与数据处理pipeline开发",            "带领3人小组完成项目"        ],        "dates": [(2020, 2021), (2021, 2023), (2023, 2024)],        "email": "user@example.com", "phone": "1234567890",        "education": "BS Computer Science"    }     # Step 1: JD解析    parser = JDParser()    jd_profile = parser.parse(jd_text)    print(f"JD解析结果: {jd_profile}")     # Step 2: 语义匹配    matcher = SemanticMatcher()    matches = matcher.match(user_resume["experiences"], jd_profile)    print(f"经历匹配排序: {matches}")     # Step 3: ATS优化    optimizer = ATSOptimizer()    resume_text = " ".join(user_resume["experiences"])    keywords = jd_profile["skills"]    coverage = optimizer.analyze_coverage(resume_text, keywords)    print(f"关键词覆盖率: {coverage['coverage']}, 缺失: {coverage['missing']}")    optimized = optimizer.optimize(resume_text, coverage["missing"])    new_cov = optimizer.analyze_coverage(optimized, keywords)    print(f"优化后覆盖率: {new_cov['coverage']}")     # Step 4: 弱点分析    scorer = ResumeScorer()    scores = scorer.score(user_resume)    print(f"简历评分: {scores}")  if __name__ == "__main__":    demo_rezi_pipeline()