""" langgraph_workflow.py — 基于 LangGraph StateGraph 的多 Agent 求职决策工作流。 9 个 Agent 通过有向无环图编排，共享 AgentState，支持条件路由。 """ from pathlib import Path import json from typing import Optional from langgraph.graph import StateGraph, END try: from .agent_state import AgentState from .llm_client import LLMClient from .agents import ( CareerIntentAgent, JobScoutAgent, JDAnalystAgent, ResumeEvidenceAgent, MatchReasoningAgent, CounterfactualPlanningAgent, ResumeCoachAgent, InterviewCoachAgent, StrategyPlannerAgent, ) except ImportError: # Allows direct execution from the src directory. from agent_state import AgentState from llm_client import LLMClient from agents import ( CareerIntentAgent, JobScoutAgent, JDAnalystAgent, ResumeEvidenceAgent, MatchReasoningAgent, CounterfactualPlanningAgent, ResumeCoachAgent, InterviewCoachAgent, StrategyPlannerAgent, ) # Singleton LLMClient _llm: Optional[LLMClient] = None _use_online: bool = False # 设置为 True 才调 DeepSeek API def _get_llm() -> Optional[LLMClient]: """Return LLMClient if use_online is True, else None (force fallback mode).""" global _llm, _use_online if not _use_online: return None # 所有 Agent 走规则 fallback，不调 API if _llm is None: _llm = LLMClient() return _llm # ============================================================================ # Node functions — 每个 Agent 包装为一个 LangGraph node # ============================================================================ def _node_career_intent(state: AgentState) -> AgentState: agent = CareerIntentAgent(llm_client=_get_llm()) state.intent = agent.run(state.resume_text, state.user_goal) state.agent_trace.append(f"[CareerIntent] direction={state.intent.direction} stage={state.intent.stage}") return state def _node_job_scout(state: AgentState) -> AgentState: corpus = _load_curated_demo_jobs() scout = JobScoutAgent(llm_client=_get_llm()) user_jds = [] if state.user_jd_text and state.user_jd_text.strip(): user_jds.append({ "title": "用户粘贴目标岗位", "company": "用户提供", "city": state.intent.target_cities[0] if state.intent and state.intent.target_cities else "不限", "salary": "", "jd_text": state.user_jd_text.strip(), "source_url": "用户粘贴", }) state.jds = scout.scout(state.intent, user_jds=user_jds, local_corpus=corpus) if hasattr(scout, "queries") and scout.queries: state.search_queries = scout.queries state.agent_trace.append(f"[JobScout] Generated queries: {scout.queries}") state.agent_trace.append(f"[JobScout] found {len(state.jds)} jobs") return state def _node_jd_analyst(state: AgentState) -> AgentState: analyst = JDAnalystAgent(llm_client=_get_llm()) for i, jd in enumerate(state.jds): if not jd.hard_skills and jd.jd_text: try: analyzed = analyst.analyze(jd.jd_text, { "title": jd.title, "company": jd.company, "city": jd.city, "salary": jd.salary, "source_url": jd.source_url, }) state.jds[i] = analyzed except Exception: pass state.agent_trace.append(f"[JDAnalyst] analyzed {len(state.jds)} JDs") return state def _node_resume_evidence(state: AgentState) -> AgentState: agent = ResumeEvidenceAgent(llm_client=_get_llm()) direction = state.intent.direction if state.intent else "" state.resume_evidence = agent.run(state.resume_text, direction) ev = state.resume_evidence state.agent_trace.append(f"[Evidence] skills={len(ev.skill_evidence)} gaps={len(ev.gap_areas)}") return state def _node_match_reasoning(state: AgentState) -> AgentState: agent = MatchReasoningAgent(llm_client=_get_llm()) state.match_results = [] for jd in state.jds[:15]: result = agent.evaluate(jd, state.resume_evidence) intent_text = (state.intent.direction if state.intent else "").lower() jd_text = " ".join([jd.title or "", jd.company or "", jd.direction or "", jd.jd_text or ""]).lower() if any(k in intent_text for k in ["agent", "llm", "大模型", "应用"]): if any(k in jd_text for k in ["agent", "rag", "llm", "langgraph", "langchain", "大模型"]): result.match_score = min(float(result.match_score) + 8, 100) if any(k in jd_text for k in ["游戏", "图像", "计算机视觉", "cv"]): result.match_score = max(float(result.match_score) - 15, 0) if jd.source_url == "用户粘贴": result.title = result.title or jd.title or "用户粘贴目标岗位" result.company = result.company or jd.company or "用户提供" if result.evidence_based_reasoning: result.evidence_based_reasoning = "用户提供的目标 JD，优先进行深度诊断。" + result.evidence_based_reasoning else: result.evidence_based_reasoning = "用户提供的目标 JD，优先进行深度诊断。" state.match_results.append(result) state.match_results.sort( key=lambda x: ( 0 if (x.company == "用户提供" or x.title.startswith("用户粘贴")) else 1, -x.match_score, ) ) state.agent_trace.append(f"[Match] {len(state.match_results)} matched") return state def _node_counterfactual(state: AgentState) -> AgentState: agent = CounterfactualPlanningAgent(llm_client=_get_llm()) top_matches = state.match_results[:5] if len(state.match_results) >= 5 else state.match_results state.counterfactual = agent.plan(state.resume_evidence, top_matches) cf = state.counterfactual state.agent_trace.append(f"[Counterfactual] {len(cf.top3_payoffs)} suggestions") return state def _node_resume_coach(state: AgentState) -> AgentState: agent = ResumeCoachAgent(llm_client=_get_llm()) target = state.jds[0] if state.jds else None state.coach = agent.coach(state.resume_text, state.resume_evidence, target) c = state.coach state.agent_trace.append(f"[ResumeCoach] rewrite={len(c.can_rewrite)} need_first={len(c.need_project_first)}") return state def _node_interview_coach(state: AgentState) -> AgentState: agent = InterviewCoachAgent(llm_client=_get_llm()) top_matches = state.match_results[:3] state.interview_prep = agent.prepare(top_matches, state.resume_evidence) state.agent_trace.append(f"[Interview] {len(state.interview_prep.likely_questions)} Qs") return state def _node_strategy_planner(state: AgentState) -> AgentState: agent = StrategyPlannerAgent(llm_client=_get_llm()) state.strategy = agent.plan(state.match_results) s = state.strategy state.agent_trace.append(f"[Strategy] safe={len(s.safe_jobs)} stretch={len(s.stretch_jobs)} skip={len(s.skip_jobs)}") return state # ============================================================================ # Conditional routing # ============================================================================ def _route_after_job_scout(state: AgentState) -> str: """如果 JD 不足 3 个，仍继续（demo 稳定性优先）。""" return "jd_analyst" def _route_after_match(state: AgentState) -> str: """如果所有匹配都低分，直接走 strategy（跳过 counterfactual + coach）。""" if state.match_results and all(m.match_score < 30 for m in state.match_results[:5]): state.agent_trace.append("[Route] all low scores → skip to strategy") return "strategy_planner" return "counterfactual" # ============================================================================ # Graph builder # ============================================================================ def build_offer_catcher_graph() -> StateGraph: """构建 LangGraph StateGraph。""" workflow = StateGraph(AgentState) # 添加 9 个节点 workflow.add_node("career_intent", _node_career_intent) workflow.add_node("job_scout", _node_job_scout) workflow.add_node("jd_analyst", _node_jd_analyst) workflow.add_node("resume_evidence", _node_resume_evidence) workflow.add_node("match_reasoning", _node_match_reasoning) workflow.add_node("counterfactual", _node_counterfactual) workflow.add_node("resume_coach", _node_resume_coach) workflow.add_node("interview_coach", _node_interview_coach) workflow.add_node("strategy_planner", _node_strategy_planner) # 主顺序边 workflow.set_entry_point("career_intent") workflow.add_edge("career_intent", "job_scout") workflow.add_conditional_edges("job_scout", _route_after_job_scout, {"jd_analyst": "jd_analyst"}) workflow.add_edge("jd_analyst", "resume_evidence") workflow.add_edge("resume_evidence", "match_reasoning") workflow.add_conditional_edges("match_reasoning", _route_after_match, { "counterfactual": "counterfactual", "strategy_planner": "strategy_planner", }) workflow.add_edge("counterfactual", "resume_coach") workflow.add_edge("resume_coach", "interview_coach") workflow.add_edge("interview_coach", "strategy_planner") workflow.add_edge("strategy_planner", END) return workflow # ============================================================================ # Public API # ============================================================================ _graph = None # 缓存编译后的图 def run_full_pipeline(resume: str, goal: str = "", use_online: bool = False, user_jd_text: str = ""): """ 一站式执行：编译 LangGraph 图，运行 9 Agent 工作流，返回 FinalDecisionReport。 """ global _graph, _use_online, _llm # P2: 将 use_online 注入到全局 flag，_get_llm() 会根据此标志决定是否返回 LLMClient _use_online = use_online if use_online: _llm = LLMClient() # 每次在线运行重新读取 provider/model/env，支持 UI 动态切换 elif not use_online: pass # _get_llm() 将返回 None，所有 Agent 走 fallback if _graph is None: _graph = build_offer_catcher_graph().compile() # 构建初始状态 state = AgentState(resume_text=resume, user_goal=goal, user_jd_text=user_jd_text) # 如果启用 LLM，注入 client（当前 demo 默认规则版） # 注意：这里 LLM client 暂不通过 graph 注入，agent 内部自行 fallback # 执行图（返回 dict） result_dict = _graph.invoke(state) # 从 dict 重建 AgentState（LangGraph 返回 TypedDict） final_state = AgentState( resume_text=result_dict.get("resume_text", resume), user_goal=result_dict.get("user_goal", goal), user_jd_text=result_dict.get("user_jd_text", user_jd_text), intent=result_dict.get("intent"), search_queries=result_dict.get("search_queries", []), jds=result_dict.get("jds", []), resume_evidence=result_dict.get("resume_evidence"), match_results=result_dict.get("match_results", []), counterfactual=result_dict.get("counterfactual"), coach=result_dict.get("coach"), interview_prep=result_dict.get("interview_prep"), strategy=result_dict.get("strategy"), agent_trace=result_dict.get("agent_trace", []), ) # 构建报告 try: from .final_report import ReportBuilder except ImportError: from final_report import ReportBuilder builder = ReportBuilder() return builder.build(final_state) def _load_curated_demo_jobs() -> list[dict]: """加载精选 Demo 岗位。 不再读取公开聚合语料。 这些数据源质量不可控，容易混入海外高管岗、IT 管理岗、非学生岗， 会破坏多 Agent 决策结果。主流程只使用人工精选岗位兜底； 真实岗位应来自用户粘贴 JD 或后续联网搜索工具。 """ root = Path(__file__).resolve().parent.parent path = root / "data" / "jobs.json" if path.exists(): try: jobs = json.loads(path.read_text(encoding="utf-8")) except Exception: return [] return [job for job in jobs if _is_curated_student_algorithm_job(job)] return [] def _is_curated_student_algorithm_job(job: dict) -> bool: """主流程候选岗位硬过滤，宁可少也不要脏。""" text = " ".join( str(job.get(k, "")) for k in ("title", "company", "direction", "stage", "jd", "jd_text", "description") ).lower() title = str(job.get("title", "")).lower() direction = str(job.get("direction", "")).lower() stage = str(job.get("stage", job.get("recruit_type", ""))).lower() positive = [ "算法", "大模型", "llm", "agent", "rag", "推荐", "搜索", "nlp", "计算机视觉", "cv", "机器学习", "深度学习", ] student_markers = ["实习", "校招", "应届", "intern", "campus"] negative = [ "chief", "manager", "director", "lead", "principal", "senior", "infrastructure manager", "asset manager", "human resource", "hr", "sales", "marketing", "finance", "consultant", "contract", "top secret", "clearance", "u.s.", "united states", ] if any(word in text for word in negative): return False if not any(word in text for word in positive): return False if not any(word in text for word in student_markers): return False if "算法" not in title and not any(word in direction for word in positive): return False return True