""" final_report.py — 统一可信决策报告 FinalDecisionReport：9 Agent 输出被整合为一个结构化、可验真的决策报告。 每个字段都有来源约束，不是大模型的自由发挥。 """ from dataclasses import dataclass, field from datetime import datetime # --------------------------------------------------------------------------- # Sub-schemas # --------------------------------------------------------------------------- @dataclass class IntentSummary: """Agent 判断的求职意图摘要。""" target_role: str = "" # 如 "LLM应用算法" stage: str = "" # "实习" / "校招" / "提前批" city_preference: list[str] = field(default_factory=list) # ["深圳","北京"] reasoning: list[str] = field(default_factory=list) # 推理步骤 confidence: float = 0.5 # 0～1，基于简历信息量估算 @dataclass class JDSource: """一个有来源可查的 JD。""" title: str = "" company: str = "" city: str = "" salary: str = "" source_type: str = "" # "公开爬取" / "用户粘贴" / "内置语料" / "联网搜索" source_url: str = "" # 原始链接（公开爬取必须非空） fetched_at: str = "" # ISO 8601 时间戳 raw_snippet: str = "" # JD 原文片段（至少 50 字符） parsed_requirements: list[str] = field(default_factory=list) # 抽取的需求列表 def has_url(self) -> bool: return bool(self.source_url and self.source_url.strip()) def is_verifiable(self) -> bool: """无来源 URL 的 JD 不可进入推荐列表。""" return self.source_type == "用户粘贴" or self.has_url() @dataclass class ResumeAction: """一条简历优化动作，必须绑定证据。""" action_type: str = "" # "rewrite" / "add_project" / "add_skill" / "quantify" target_text: str = "" # 改写目标或新增内容 original_fragment: str = "" # 简历原文片段（rewrite 时必填） jd_requirement_fragment: str = "" # JD 要求片段 evidence_based: bool = False # True = 有简历证据，False = 需要先补经历 reason: str = "" # 为什么这样改 @dataclass class JobDecision: """对一个岗位的完整决策。""" job_id: str = "" # 唯一标识（如 url hash） title: str = "" company: str = "" city: str = "" salary: str = "" decision: str = "" # "稳投" / "冲刺" / "暂缓" match_score: float = 0.0 pass_likelihood: float = 0.0 risk_level: str = "中" # "低" / "中" / "高" jd_evidence: list[str] = field(default_factory=list) # JD 中的具体要求 resume_evidence: list[str] = field(default_factory=list) # 简历中已有的证据 missing_evidence: list[str] = field(default_factory=list) # 明显缺口 why_this_decision: list[str] = field(default_factory=list) # 推理链 resume_actions: list[ResumeAction] = field(default_factory=list) # 简历优化 interview_actions: list[str] = field(default_factory=list) # 面试准备 supporting_evidence: list[str] = field(default_factory=list) # 支撑证据 can_rewrite: bool = True need_new_project: bool = False def is_valid(self) -> bool: """合约验证：每个决策至少 1 条 jd_evidence。""" return len(self.jd_evidence) >= 1 @dataclass class WhatIfItem: """反事实规划中的一条补强建议。""" action: str = "" # 要做什么 expected_gain: str = "" # 预期收益（文字描述） why: str = "" # 原因为什么重要 needed_time: str = "" # 预计耗时 @dataclass class Portfolio: """投递组合。""" safe: list[JobDecision] = field(default_factory=list) stretch: list[JobDecision] = field(default_factory=list) hold: list[JobDecision] = field(default_factory=list) # --------------------------------------------------------------------------- # Top-level report # --------------------------------------------------------------------------- @dataclass class FinalDecisionReport: """9 Agent 协作产出的最终决策报告。""" # 第一屏 intent_summary: IntentSummary = field(default_factory=IntentSummary) # JD 来源（所有被检索到的） jd_sources: list[JDSource] = field(default_factory=list) # 投递组合 portfolio: Portfolio = field(default_factory=Portfolio) # 每个岗位的详细决策 job_decisions: list[JobDecision] = field(default_factory=list) # 反事实补强规划 what_if_plan: list[WhatIfItem] = field(default_factory=list) # 生成时间 generated_at: str = field(default_factory=lambda: datetime.now().isoformat()) # 测试用：所有 agent trace trace: list[str] = field(default_factory=list) # Job Search queries search_queries: list[str] = field(default_factory=list) def all_jobs_have_source(self) -> bool: """所有 jd_sources 中'公开爬取'类型的必须有 URL。""" for s in self.jd_sources: if s.source_type == "公开爬取" and not s.is_verifiable(): return False return True def all_resume_actions_tagged(self) -> list[str]: """检查所有 resume_action：evidence_based=True 的 rewrite 必须有原始片段。""" issues = [] for jd in self.job_decisions: for i, action in enumerate(jd.resume_actions): if action.action_type == "rewrite" and action.evidence_based and not action.original_fragment: issues.append( f"[{jd.company}-{jd.title}] resume_action[{i}] evidence_based rewrite 缺少 original_fragment") return issues def validate_contract(self) -> tuple[bool, list[str]]: """运行完整合约检查。返回 (通过, 问题列表)。""" issues = [] # 1. 无来源 JD 不进推荐 for jd in self.jd_sources: if jd.source_type == "公开爬取" and not jd.is_verifiable(): issues.append(f"[{jd.company}-{jd.title}] 公开爬取 JD 无 source_url") # 2. 每个 job_decision 至少 1 条 jd_evidence for i, jd in enumerate(self.job_decisions): if not jd.is_valid(): issues.append(f"job_decisions[{i}] ({jd.company}-{jd.title}) 缺少 jd_evidence") # 3. 每个 resume_action 标记 evidence_based 或 need_new_experience issues.extend(self.all_resume_actions_tagged()) # 4. 不出现在 portfolio 的 JD 不应在 job_decisions 中 # （放宽松：允许） passed = len(issues) == 0 return passed, issues def to_markdown(self) -> str: """将结构化的决策报告序列化为 Markdown 字符串。""" def _fmt_val(val, pct=False): try: f_val = float(val) if pct: # 如果 pass_likelihood 是 0.0-1.0 范围，显示为百分比；如果是 0-100 范围，除以 100 if f_val > 1.0: f_val = f_val / 100.0 return f"{f_val:.1%}" else: return f"{f_val:.1f}" except Exception: return str(val) parts = [] parts.append("# 📊 Offer 捕手 · 求职决策报告") parts.append("") # 1. Intent Summary parts.append("## 一、求职意向诊断") intent = self.intent_summary parts.append(f"- **目标岗位方向**: {intent.target_role or '待评估'}") parts.append(f"- **期望求职通道**: {intent.stage or '待评估'}") city_pref = "、".join(intent.city_preference) if intent.city_preference else "不限" parts.append(f"- **期望工作城市**: {city_pref}") parts.append(f"- **意向评估置信度**: {_fmt_val(intent.confidence, pct=True)}") if intent.reasoning: parts.append("\n**意向分析依据:**") for r in intent.reasoning: parts.append(f"- {r}") if hasattr(self, "search_queries") and self.search_queries: parts.append("\n**Job Search Agent 动态检索词:**") for q in self.search_queries: parts.append(f"- `{q}`") parts.append("") # 2. Portfolio parts.append("## 二、推荐投递组合") parts.append("") def _render_combo_list(jobs, label): lines = [f"### 📌 {label}"] if not jobs: lines.append("_无匹配岗位_") else: for j in jobs: lines.append(f"- **{j.title}** @ **{j.company}** | Match: `{_fmt_val(j.match_score)}` | Pass: `{_fmt_val(j.pass_likelihood, pct=True)}` | Risk: `{j.risk_level}`") return "\n".join(lines) parts.append(_render_combo_list(self.portfolio.safe, "建议立即投递岗")) parts.append("") parts.append(_render_combo_list(self.portfolio.stretch, "冲刺与简历优化岗")) parts.append("") parts.append(_render_combo_list(self.portfolio.hold, "暂缓考虑岗")) parts.append("") # 3. Detailed decisions parts.append("## 三、精细人岗匹配诊断与简历改写") parts.append("") for idx, jd in enumerate(self.job_decisions, 1): parts.append(f"### {idx}. {jd.title} @ {jd.company}") parts.append(f"- **决策建议**: `{jd.decision}`") parts.append(f"- **Match Score**: `{_fmt_val(jd.match_score)}` | **Pass Likelihood**: `{_fmt_val(jd.pass_likelihood, pct=True)}` | **Risk Level**: `{jd.risk_level}`") can_rw_str = "✅ 是" if getattr(jd, "can_rewrite", True) else "❌ 否" need_pj_str = "⚠️ 是 (需先补项目)" if getattr(jd, "need_new_project", False) else "✅ 否" parts.append(f"- **是否可改写**: {can_rw_str} | **是否需要先做Demo补强**: {need_pj_str}") parts.append("") # Evidence parts.append("#### 🔍 匹配证据链与缺口对齐") if jd.jd_evidence: parts.append("**JD 核心要求:**") for e in jd.jd_evidence: parts.append(f" - {e}") if hasattr(jd, "supporting_evidence") and jd.supporting_evidence: parts.append("**支撑证据 (Resume Evidence):**") for e in jd.supporting_evidence: parts.append(f" - {e}") elif jd.resume_evidence: parts.append("**简历匹配要点:**") for e in jd.resume_evidence: parts.append(f" - {e}") if jd.missing_evidence: parts.append("**缺失/不足证据 (Gaps):**") for e in jd.missing_evidence: parts.append(f" - ⚠️ {e}") parts.append("") # Why this decision if jd.why_this_decision: parts.append("**决策依据推理:**") for r in jd.why_this_decision: parts.append(f"> {r}") parts.append("") # Resume actions if jd.resume_actions: parts.append("#### 📝 简历句式优化建议") for action in jd.resume_actions: parts.append(f"- **优化动作**: `{action.action_type}`") if action.jd_requirement_fragment: parts.append(f" - **对应岗位需求**: {action.jd_requirement_fragment}") if action.original_fragment: parts.append(f" - **原简历描述**: _\"{action.original_fragment}\"_") parts.append(f" - **推荐修改为**: **\"{action.target_text}\"**") if action.reason: parts.append(f" - **修改理由**: {action.reason}") parts.append("") # Interview preparation if jd.interview_actions: parts.append("#### 🧠 针对性面试考察点") for q in jd.interview_actions: parts.append(f"- {q}") parts.append("") parts.append("---") # 4. What-if Plan if self.what_if_plan: parts.append("## 四、反事实生涯强化规划") parts.append("如果针对以下缺失点进行短期补强，预期能够带来显著的匹配度收益：") parts.append("") for item in self.what_if_plan: parts.append(f"- **补强动作**: {item.action}") parts.append(f" - **预期收益**: {item.expected_gain}") parts.append(f" - **依据/原因**: {item.why}") parts.append(f" - **预估成本**: {item.needed_time}") parts.append("") parts.append(f"*报告生成时间：{self.generated_at[:19]} | 由 Offer 捕手 LangGraph Multi-Agent 协同引擎自动生成*") return "\n".join(parts) # --------------------------------------------------------------------------- # Builder # --------------------------------------------------------------------------- try: from .agent_state import AgentState, CareerIntent, JDProfile, ResumeEvidence, MatchResult from .agent_state import CounterfactualPlan, CoachOutput, InterviewPrep, StrategyOutput except ImportError: # Allows direct execution from the src directory. from agent_state import AgentState, CareerIntent, JDProfile, ResumeEvidence, MatchResult from agent_state import CounterfactualPlan, CoachOutput, InterviewPrep, StrategyOutput class ReportBuilder: """从 AgentState 构建 FinalDecisionReport。""" def build(self, state: AgentState) -> FinalDecisionReport: report = FinalDecisionReport() # 1. Intent summary if state.intent: report.intent_summary = self._build_intent(state.intent) # 2. Search queries report.search_queries = state.search_queries # 3. JD sources report.jd_sources = self._build_jd_sources(state.jds) # 4. Job decisions report.job_decisions = self._build_job_decisions( state.match_results, state.resume_evidence, state.coach, state.interview_prep) # 5. Portfolio report.portfolio = self._build_portfolio(state.strategy, report.job_decisions) # 6. What-if plan report.what_if_plan = self._build_what_if(state.counterfactual) # 7. Trace report.trace = state.agent_trace return report def _build_intent(self, intent: CareerIntent) -> IntentSummary: return IntentSummary( target_role=intent.direction or "待确认", stage=intent.stage or "校招", city_preference=intent.target_cities or ["深圳", "北京"], reasoning=[intent.reasoning] if intent.reasoning else [], confidence=0.7 if intent.direction and intent.direction != "待确认" else 0.4, ) def _build_jd_sources(self, jds: list[JDProfile]) -> list[JDSource]: sources = [] for jd in jds: source_url_val = (jd.source_url or "") if hasattr(jd, 'source_url') else "" source_type = "用户粘贴" if source_url_val == "用户粘贴" else ( "公开爬取" if source_url_val and source_url_val.startswith("http") else "Demo精选岗位") snippet = (jd.jd_text or "") if len(snippet) < 80: snippet = snippet + "（更多JD详情请联系系统管理员或查阅原始链接）" sources.append(JDSource( title=jd.title or "", company=jd.company or "", city=jd.city or "", salary=jd.salary or "面议", source_type=source_type, source_url=source_url_val if source_url_val.startswith("http") else "", fetched_at=datetime.now().isoformat()[:19], raw_snippet=snippet[:400], parsed_requirements=list(jd.hard_skills)[:8] if jd.hard_skills else [], )) return sources def _build_job_decisions( self, matches: list[MatchResult], evidence: ResumeEvidence, coach: CoachOutput, interview: InterviewPrep, ) -> list[JobDecision]: decisions = [] for i, m in enumerate(matches): # 构建 JD evidence（从 JD 中提取的具体要求） jd_evidence_list = [] if hasattr(m, 'jd') and m.jd and m.jd.hard_skills: jd_evidence_list = [f"要求：{s}" for s in m.jd.hard_skills[:5]] if not jd_evidence_list and m.missing_evidence: jd_evidence_list = [f"缺失要求：{x}" for x in m.missing_evidence[:3]] if not jd_evidence_list: jd_evidence_list = ["JD 结构化解析完成（规则提取）"] # 构建 Resume evidence（从简历中匹配到的证据） resume_evidence_list = [] if evidence and evidence.skill_evidence: for skill, ev_list in list(evidence.skill_evidence.items())[:5]: if ev_list: resume_evidence_list.append(f"技能 {skill}：{ev_list[0][:80]}") # 简历优化动作 actions = [] coach_actions = (coach.can_rewrite if coach else []) + (coach.need_project_first if coach else []) for ca in coach_actions[:4]: is_rewrite = ca in (coach.can_rewrite if coach else []) # 尝试从简历证据中提取原始片段 orig_fragment = "" if is_rewrite and evidence and evidence.skill_evidence: for skill, ev_list in evidence.skill_evidence.items(): if ev_list: orig_fragment = ev_list[0] break actions.append(ResumeAction( action_type="rewrite" if is_rewrite else "add_project", target_text=ca[:120], original_fragment=orig_fragment[:150] if is_rewrite and orig_fragment else "", jd_requirement_fragment=str(jd_evidence_list[0])[:120] if jd_evidence_list else "", evidence_based=is_rewrite and bool(orig_fragment), reason="基于简历原文改写" if (is_rewrite and orig_fragment) else ("需要先补真实项目经历再写入简历" if is_rewrite else "补充缺失经历"), )) d = JobDecision( job_id=f"job_{i}", title=m.title or "", company=m.company or "", decision=m.apply_action or "暂缓", match_score=float(m.match_score) if m.match_score else 0.0, pass_likelihood=float(m.pass_likelihood) if m.pass_likelihood else 0.0, risk_level=m.risk_level or "中", jd_evidence=jd_evidence_list, resume_evidence=resume_evidence_list, supporting_evidence=m.supporting_evidence if hasattr(m, 'supporting_evidence') else [], missing_evidence=list(m.missing_evidence)[:5] if m.missing_evidence else [], why_this_decision=[(m.evidence_based_reasoning or "")[:200]], resume_actions=actions, interview_actions=(interview.likely_questions[:3] if interview else []), can_rewrite=m.can_rewrite if hasattr(m, 'can_rewrite') else True, need_new_project=m.need_new_project if hasattr(m, 'need_new_project') else False, ) decisions.append(d) return decisions def _build_portfolio( self, strategy: StrategyOutput, all_decisions: list[JobDecision], ) -> Portfolio: if strategy is None: # Fallback: derive from decisions safe = [d for d in all_decisions if d.decision == "立即投递"] stretch = [d for d in all_decisions if d.decision in ("先优化再投", "冲刺岗位")] hold = [d for d in all_decisions if d.decision == "暂缓"] return Portfolio(safe=safe, stretch=stretch, hold=hold) # Map strategy MatchResult → JobDecision def _lookup(mr_list, all_d): result = [] for mr in mr_list: for d in all_d: if d.title == mr.title and d.company == mr.company: result.append(d) break else: # 不在 all_decisions 中，创建一个 result.append(JobDecision( title=mr.title, company=mr.company, decision="稳投")) return result return Portfolio( safe=_lookup(strategy.safe_jobs if strategy.safe_jobs else [], all_decisions), stretch=_lookup(strategy.stretch_jobs if strategy.stretch_jobs else [], all_decisions), hold=_lookup(strategy.skip_jobs if strategy.skip_jobs else [], all_decisions), ) def _build_what_if(self, cf: CounterfactualPlan) -> list[WhatIfItem]: if cf is None or not cf.top3_payoffs: return [] items = [] for p in cf.top3_payoffs: items.append(WhatIfItem( action=p.get("action", ""), expected_gain=f"匹配度预估提升 +{p.get('match_gain','?')}%", why=p.get("why", ""), needed_time=f"{p.get('effort_days','?')}天", )) return items