""" Report Generator — Markdown 求职策略报告导出模块 一键生成包含岗位排序、简历优化建议、7 天投递策略、面试准备的完整报告。 """ from __future__ import annotations import json from datetime import date from pathlib import Path from typing import Optional # --------------------------------------------------------------------------- # 报告生成主函数 # --------------------------------------------------------------------------- def generate_report( profile: dict, ranked_jobs: list[dict], strategy: dict, gaps: Optional[list[str]] = None, rewrites: Optional[list[dict]] = None, top_n: int = 5, ) -> str: """ 生成完整 Markdown 求职策略报告。 参数： - profile: 学生画像（来自 Resume Parser Agent） - ranked_jobs: 已排序岗位列表（来自 Application Ranker Agent） - strategy: 策略包（来自 Strategy Planner Agent） - gaps: 能力缺口列表（可选，来自 Gap Diagnosis Agent） - rewrites: 简历改写建议（可选，来自 Resume Conversion Agent） - top_n: 报告中展示的岗位数量（默认 5） 返回： - Markdown 格式字符串 """ parts = [] # ---- 标题 + 基本信息 ---- parts.append(_build_header(profile)) parts.append("") # ---- 一、岗位投递优先级榜单 ---- parts.append("## 一、岗位投递优先级榜单") parts.append("") parts.append(_build_job_table(ranked_jobs[:top_n])) parts.append("") # ---- 二、今日优先投递 Top3 ---- parts.append("## 二、今日优先投递建议（Top 3）") parts.append("") parts.append(_build_priority_top3(strategy.get("priority_top3", []))) parts.append("") # ---- 三、稳妥 / 平衡 / 冲刺岗位组合 ---- parts.append("## 三、岗位组合建议") parts.append("") parts.append(_build_job_combo(strategy.get("job_combo", {}))) parts.append("") # ---- 四、能力缺口诊断 ---- parts.append("## 四、能力缺口诊断") parts.append("") parts.append(_build_gaps(ranked_jobs[:top_n], gaps)) parts.append("") # ---- 五、简历改写建议 ---- parts.append("## 五、简历改写建议") parts.append("") parts.append(_build_rewrites(ranked_jobs[:top_n], rewrites)) parts.append("") # ---- 六、7 天投递计划 ---- parts.append("## 六、7 天投递执行计划") parts.append("") parts.append(_build_apply_plan(strategy.get("apply_plan_7day", []))) parts.append("") # ---- 七、面试准备计划 ---- parts.append("## 七、7 天面试准备计划") parts.append("") parts.append(_build_interview_plan(strategy.get("interview_plan_7day", []))) parts.append("") # ---- 八、需要先改简历再投的岗位 ---- resume_advice = strategy.get("resume_advice", []) if resume_advice: parts.append("") parts.append("## 八、需优先优化简历的岗位") parts.append("") parts.append(_build_resume_advice(resume_advice)) parts.append("") # ---- 附录：学生画像摘要 ---- parts.append("") parts.append("## 附录：学生画像摘要") parts.append("") parts.append(_build_profile_summary(profile)) parts.append("") parts.append("---") parts.append(f"*报告生成时间：{date.today().strftime('%Y-%m-%d')} | 由 Offer 捕手 Agent 决策驾驶舱自动生成*") return "\n".join(parts) # --------------------------------------------------------------------------- # 各小节构建函数 # --------------------------------------------------------------------------- def _build_header(profile: dict) -> str: skills = "、".join(profile.get("skills", [])[:8]) or "待补充" return f"""# 📊 Offer 捕手 · 求职策略报告 **技能标签：** {skills} """ def _build_job_table(ranked_jobs: list[dict]) -> str: if not ranked_jobs: return "_暂无匹配岗位，请先运行匹配。_" lines = [] lines.append("| 排名 | 岗位 | 公司 | 城市 | Match | Pass | Risk | Growth | 优先级 | 推荐动作 |") lines.append("|------|------|------|------|-------|------|------|---------|--------|----------|") action_map = {} for job in ranked_jobs: pass_s = job.get("pass_score", 50) risk_s = job.get("risk_score", 50) if pass_s >= 70 and risk_s <= 30: action = "✅ 立即投递" elif pass_s >= 55: action = "✏️ 先优化再投" elif job.get("growth_score", 50) >= 65: action = "🚀 冲刺岗位" else: action = "⏳ 暂缓" action_map[id(job)] = action for i, job in enumerate(ranked_jobs, 1): title = job.get("title", "未知岗位") company = job.get("company", "") city = job.get("city", "") match = job.get("match_score", job.get("score", 0)) pass_s = job.get("pass_score", "-") risk_s = job.get("risk_score", "-") growth = job.get("growth_score", "-") priority = job.get("apply_priority", "-") action = action_map.get(id(job), "-") lines.append(f"| Top{i} | {title} | {company} | {city} | {match} | {pass_s} | {risk_s} | {growth} | {priority} | {action} |") return "\n".join(lines) def _build_priority_top3(top3: list[dict]) -> str: if not top3: return "_请先运行匹配获取岗位推荐。_" lines = [] for item in top3: lines.append(f"**Top {item['rank']}：{item['title']}（{item['company']}）**") lines.append(f"- 推荐动作：{item['apply_action']}") lines.append(f"- 理由：{item['reason']}") lines.append(f"- ApplyPriority = {item.get('apply_priority', '-')}，PassScore = {item.get('pass_score', '-')}，RiskScore = {item.get('risk_score', '-')}") lines.append("") return "\n".join(lines) def _build_job_combo(combo: dict) -> str: if not combo: return "_暂无组合建议。_" lines = [] emoji_map = {"稳妥岗（高通过率）": "🛡️", "平衡岗（综合推荐）": "⚖️", "冲刺岗（高成长价值）": "🚀"} for category, jobs in combo.items(): emoji = emoji_map.get(category, "📌") lines.append(f"**{emoji} {category}**") if not jobs: lines.append(" （暂无）") for j in jobs: lines.append(f" - {j.get('title', '')}（{j.get('company', '')}）| Match={j.get('match_score', '-')} Pass={j.get('pass_score', '-')} Risk={j.get('risk_score', '-')}") lines.append("") return "\n".join(lines) def _build_gaps(ranked_jobs: list[dict], gaps: Optional[list[str]]) -> str: lines = [] if gaps: for g in gaps[:6]: lines.append(f"- ⚠️ {g}") elif ranked_jobs: # 从第一个岗位的 gaps 字段取 first_gaps = ranked_jobs[0].get("gaps", []) if first_gaps: for g in first_gaps: lines.append(f"- ⚠️ {g}") else: lines.append("_岗位能力缺口已纳入评分，暂无高风险项。_") else: lines.append("_请先运行匹配。_") return "\n".join(lines) if lines else "_暂无数据。_" def _build_rewrites(ranked_jobs: list[dict], rewrites: Optional[list[dict]]) -> str: lines = [] # 优先用传入的 rewrites，否则从 Top1 岗位取 items = rewrites or (ranked_jobs[0].get("rewrites", []) if ranked_jobs else []) if not items: return "_暂无改写建议，简历与岗位匹配度较好。_" for item in items: before = item.get("before", "") after = item.get("after", "") lines.append(f"**原表达：**") lines.append(f"> {before}") lines.append(f"**建议改写：**") lines.append(f"> {after}") lines.append("") return "\n".join(lines) def _build_apply_plan(plan: list[str]) -> str: if not plan: return "_暂无投递计划。_" return "\n".join(f"- {p}" for p in plan) def _build_interview_plan(plan: list[str]) -> str: if not plan: return "_暂无面试准备计划。_" return "\n".join(f"- {p}" for p in plan) def _build_resume_advice(advice: list[dict]) -> str: lines = [] for item in advice[:5]: lines.append(f"**{item['title']}（{item['company']}）** — 紧急度：{item.get('urgency', '中')}") for tip in item.get("advice", []): lines.append(f" - {tip}") lines.append("") return "\n".join(lines) if lines else "_所有岗位简历匹配度较好，可直接投递。_" def _build_profile_summary(profile: dict) -> str: lines = [] skills = "、".join(profile.get("skills", [])) or "未检测到明显技能词" project_sigs = "、".join(profile.get("project_signals", profile.get("project_signals", []))) or "未检测到项目信号" lines.append(f"- **技能：** {skills}") lines.append(f"- **项目信号：** {project_sigs}") lines.append(f"- **量化指标：** {'有 ✅' if profile.get('has_metrics') else '缺少 ⚠️'}（建议在简历中补充 NDCG/HitRate 等具体数字）") lines.append(f"- **LLM 项目信号：** {'有 ✅' if profile.get('has_llm_project') else '无 ⚠️'}（大模型方向建议补充 RAG/Agent 项目描述）") lines.append(f"- **推荐项目信号：** {'有 ✅' if profile.get('has_rec_project') else '无 ⚠️'}（推荐方向建议补充召回/排序项目描述）") return "\n".join(lines) # --------------------------------------------------------------------------- # 保存报告到文件 # --------------------------------------------------------------------------- def save_report(md_text: str, output_path: str | Path) -> Path: """ 将 Markdown 报告保存到指定路径，返回实际保存的 Path 对象。 """ path = Path(output_path) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(md_text, encoding="utf-8") return path # --------------------------------------------------------------------------- # CLI 入口（方便本地测试） # --------------------------------------------------------------------------- if __name__ == "__main__": # 构造一个假数据结构用于测试报告格式 dummy_profile = { "skills": ["Python", "PyTorch", "Transformer", "RAG", "Agent", "Embedding"], "project_signals": ["RAG", "Agent", "检索", "推荐", "Semantic ID"], "has_metrics": True, "has_llm_project": True, "has_rec_project": True, } dummy_jobs = [ { "title": "大模型应用算法实习生", "company": "腾讯云智能", "city": "深圳", "direction": "大模型应用算法", "score": 92, "match_score": 92, "pass_score": 78, "risk_score": 22, "growth_score": 85, "apply_priority": 88, "gaps": ["建议补充 Agent 工具调用失败兜底方案", "简历缺少多轮对话评估指标"], "rewrites": [ {"before": "做过 RAG Demo。", "after": "基于 bge embedding + FAISS 构建企业知识库问答，Top5 命中率提升 18%，并设计 Agent 工具调用链路。"} ], }, { "title": "LLM 推荐算法实习生", "company": "内容平台事业群", "city": "北京", "direction": "推荐算法", "score": 85, "match_score": 85, "pass_score": 65, "risk_score": 40, "growth_score": 90, "apply_priority": 76, "gaps": [], "rewrites": [], }, ] dummy_strategy = { "priority_top3": [ {"rank": 1, "title": "大模型应用算法实习生", "company": "腾讯云智能", "apply_action": "先优化再投", "reason": "PassScore=78，建议补齐 Agent 项目证据再投。", "apply_priority": 88, "pass_score": 78, "risk_score": 22}, ], "job_combo": { "稳妥岗（高通过率）": [{"title": "大模型应用算法实习生", "company": "腾讯云智能", "match_score": 92, "pass_score": 78, "risk_score": 22, "growth_score": 85}], "平衡岗（综合推荐）": [{"title": "LLM 推荐算法实习生", "company": "内容平台事业群", "match_score": 85, "pass_score": 65, "risk_score": 40, "growth_score": 90}], "冲刺岗（高成长价值）": [], }, "apply_plan_7day": [f"第 {i} 天：投递执行 + 简历优化。" for i in range(1, 8)], "interview_plan_7day": [f"第 {i} 天：面试准备。" for i in range(1, 8)], "resume_advice": [], } md = generate_report(dummy_profile, dummy_jobs, dummy_strategy) out = Path(__file__).resolve().parent.parent / "reports" saved = save_report(md, out / "求职策略报告_测试.md") print(f"测试报告已保存到：{saved}")