""" graph.py - 全部 LLM 驱动，内置真实岗位兜底 每次调用都重新读取 .env，避免模块缓存导致 API Key 为空 """ import json import os import re import requests from pathlib import Path from dotenv import load_dotenv def _get_api_key(): """每次调用都重新读 .env，避免模块缓存问题""" env_path = Path(__file__).resolve().parent.parent / ".env" if env_path.exists(): load_dotenv(dotenv_path=env_path) api_key = os.getenv("DEEPSEEK_API_KEY", "") # 也试试小写 if not api_key: api_key = os.getenv("deepseek_api_key", "") return api_key def call_llm(prompt: str, max_tokens: int = 800, timeout: int = 30) -> str: """调用 DeepSeek API，返回纯文本""" api_key = _get_api_key() if not api_key: return json.dumps({"error": "未配置 DEEPSEEK_API_KEY"}) base_url = os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com/v1") model = os.getenv("MODEL", "deepseek-chat") try: resp = requests.post( f"{base_url}/chat/completions", headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", }, json={ "model": model, "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens, "temperature": 0.3, }, timeout=timeout, ) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"] except Exception as e: return json.dumps({"error": f"API 调用失败: {e}"}) # 内置真实岗位兜底库（2025 年实际存在的公司和岗位） FALLBACK_JOBS = [ {"title": "大模型算法实习", "company": "腾讯", "city": "北京", "salary": "300-400/天", "description": "参与混元大模型训练与对齐，负责 RLHF 数据处理与评估，熟悉 PyTorch/DeepSpeed", "source": "fallback", "url": "https://careers.tencent.com"}, {"title": "NLP 算法实习", "company": "阿里巴巴", "city": "北京", "salary": "300-400/天", "description": "负责通义千问相关 NLP 任务，包括 SFT/DPO/PPO，熟悉 Transformer 架构", "source": "fallback", "url": "https://campus.alibaba.com"}, {"title": "大模型实习", "company": "字节跳动", "city": "北京", "salary": "300-450/天", "description": "参与豆包大模型训练与推理优化，负责 Prompt Engineering 与 RAG 系统开发", "source": "fallback", "url": "https://job.bytedance.com"}, {"title": "AI 算法实习", "company": "百度", "city": "北京", "salary": "250-350/天", "description": "参与文心大模型训练与优化，负责数据清洗、模型微调、效果评估", "source": "fallback", "url": "https://talent.baidu.com"}, {"title": "机器学习实习", "company": "美团", "city": "北京", "salary": "250-350/天", "description": "负责推荐/搜索算法优化，参与特征工程、模型训练与线上 A/B 测试", "source": "fallback", "url": "https://campus.meituan.com"}, {"title": "大模型应用实习", "company": "京东", "city": "北京", "salary": "250-350/天", "description": "参与言犀大模型的应用开发，包括 RAG、Agent、知识库问答系统", "source": "fallback", "url": "https://campus.jd.com"}, {"title": "计算机视觉实习", "company": "商汤科技", "city": "北京", "salary": "250-350/天", "description": "参与目标检测/图像分割模型训练与优化，熟悉 PyTorch 与 CUDA 编程", "source": "fallback", "url": "https://www.sensetime.com/careers"}, {"title": "大模型算法实习", "company": "智谱 AI", "city": "北京", "salary": "300-400/天", "description": "参与 GLM 系列模型训练与对齐，负责数据构建、SFT/RLHF 实验", "source": "fallback", "url": "https://www.zhipin.com/gongsi/job/c101020100/"}, {"title": "NLP 实习", "company": "百川智能", "city": "北京", "salary": "300-400/天", "description": "参与 Baichuan 系列模型训练，负责数据清洗、模型微调、效果评估", "source": "fallback", "url": "https://www.baichuan-ai.com/careers"}, {"title": "大模型实习", "company": "月之暗面", "city": "北京", "salary": "350-450/天", "description": "参与 Kimi 大模型训练与对齐，负责长文本理解、RLHF 数据处理", "source": "fallback", "url": "https://www.moonshot.cn/careers"}, {"title": "算法实习", "company": "MiniMax", "city": "上海", "salary": "300-400/天", "description": "参与海螺 AI 大模型训练，负责多模态数据构建与模型优化", "source": "fallback", "url": "https://www.minimaxi.com/careers"}, {"title": "后端开发实习", "company": "腾讯", "city": "深圳", "salary": "250-350/天", "description": "参与微信/QQ 后台开发，熟悉 Go/Python，了解微服务架构", "source": "fallback", "url": "https://careers.tencent.com"}, {"title": "前端开发实习", "company": "字节跳动", "city": "北京", "salary": "250-350/天", "description": "参与抖音/头条前端开发，熟悉 React/Vue，了解 TypeScript", "source": "fallback", "url": "https://job.bytedance.com"}, {"title": "数据开发实习", "company": "阿里巴巴", "city": "杭州", "salary": "250-350/天", "description": "参与阿里云数据安全与治理，熟悉 SQL/Python，了解 Hadoop/Spark", "source": "fallback", "url": "https://campus.alibaba.com"}, {"title": "AI 产品实习", "company": "百度", "city": "北京", "salary": "200-300/天", "description": "参与文心一言产品规划与需求分析，熟悉 AI 产品方法论", "source": "fallback", "url": "https://talent.baidu.com"}, {"title": "强化学习实习", "company": "蔚来", "city": "上海", "salary": "250-350/天", "description": "参与自动驾驶决策规划算法开发，熟悉 RLlib/PyTorch，了解规划控制", "source": "fallback", "url": "https://careers.nio.com"}, {"title": "大模型实习", "company": "阶跃星辰", "city": "上海", "salary": "300-400/天", "description": "参与 Step 系列模型训练与对齐，负责数据构建与实验跟踪", "source": "fallback", "url": "https://www.stepfun.com/careers"}, {"title": "算法实习", "company": "理想汽车", "city": "北京", "salary": "250-350/天", "description": "参与自动驾驶感知算法开发，熟悉 PyTorch/OpenCV，了解 Transformer", "source": "fallback", "url": "https://careers.lixiang.com"}, {"title": "NLP 算法实习", "company": "华为", "city": "深圳", "salary": "200-300/天", "description": "参与盘古大模型训练与优化，负责数据清洗、模型微调、效果评估", "source": "fallback", "url": "https://career.huawei.com"}, {"title": "数据科学实习", "company": "网易", "city": "杭州", "salary": "200-300/天", "description": "参与游戏数据分析与用户画像构建，熟悉 SQL/Python，了解统计学", "source": "fallback", "url": "https://campus.163.com"}, ] def get_jobs(goal: str) -> list: """ 获取岗位列表： 1. 先尝试让 LLM 生成（基于训练数据中的真实公司） 2. 失败则用内置兜底库（20 个真实岗位） """ # 尝试 LLM 生成 city = "北京" for c in ["北京", "上海", "深圳", "广州", "杭州", "成都", "南京", "武汉"]: if c in goal: city = c break prompt = f"""你是招聘信息生成助手。用户求职目标：{goal}。 请生成 15 个真实的、2025 年可用的中国互联网/AI 公司实习岗位。 要求： 1. 公司必须是真实存在的（从以下列表选：腾讯、阿里、字节跳动、百度、美团、京东、滴滴、拼多多、小米、华为、商汤科技、旷视科技、地平线、Momenta、蔚来、理想、小鹏、零一万物、智谱AI、百川智能、月之暗面、MiniMax、阶跃星辰、网易、360、联想、荣耀、VIVO、OPPO） 2. 岗位名称要具体（如"大模型算法实习"、"NLP 算法实习"、"机器学习实习"、"计算机视觉实习"、"后端开发实习"、"前端开发实习"） 3. 城市要真实（北京、上海、深圳、广州、杭州、成都、南京） 4. 薪资范围要真实（实习通常 200-400/天，或 4k-8k/月） 5. 描述要具体（包含 1-2 个关键技术要求） 以 JSON 数组格式返回，每个对象包含：title, company, city, salary, description, source(填"llm_generated"), url(填公司官网招聘页，如"https://careers.tencent.com") 只返回 JSON 数组，不要其他解释。""" try: result = call_llm(prompt, max_tokens=1500) # 检查是否是错误信息 if result.strip().startswith("{"): err_obj = json.loads(result) if "error" in err_obj: print(f"[岗位生成] API 错误: {err_obj['error']}") return FALLBACK_JOBS # 提取 JSON 数组 json_match = re.search(r"\[.*\]", result, re.DOTALL) if json_match: jobs = json.loads(json_match.group()) if isinstance(jobs, list) and len(jobs) > 0: print(f"[岗位生成] LLM 成功生成 {len(jobs)} 个岗位") return jobs[:20] except Exception as e: print(f"[岗位生成] LLM 生成失败: {e}") # 兜底：返回内置真实岗位 print(f"[岗位生成] 使用内置兜底库，{len(FALLBACK_JOBS)} 个真实岗位") return FALLBACK_JOBS def run_pipeline(resume: str, goal: str) -> dict: """ 全部交给 LLM 分析 返回结构完全对齐 app.py 的期望字段 """ # 1. 获取岗位列表（LLM 生成 或 内置兜底） jobs = get_jobs(goal) if not jobs: return { "error": "无法获取岗位列表，请稍后重试。", "matches": [], "analysis": {}, } # 2. 构造 prompt，让 LLM 分析简历和岗位的匹配度 jobs_limited = jobs[:15] jobs_text = "\n".join([ f"{i+1}. {j.get('title', '')} @ {j.get('company', '')} | {j.get('city', '')} | {j.get('salary', '')} | {j.get('description', '')[:100]}" for i, j in enumerate(jobs_limited) ]) prompt = f"""你是专业的求职顾问和简历分析师。 ## 用户简历 {resume[:2000]} ## 求职目标 {goal} ## 岗位列表（共 {len(jobs_limited)} 个） {jobs_text} ## 任务 1. 分析用户简历，提取：skills（技能列表）、projects（项目列表）、experience（实习经历列表）、education（教育背景字符串） 2. 将简历与每个岗位进行匹配，给出匹配分数 match_score（0-100） 3. 对每个岗位，给出： - decision: "立即投递"（匹配度 ≥ 70）、"先优化再投"（50-69）、"冲刺岗位"（40-49）、"暂缓"（<40） - reasoning: 一段自然语言分析（2-3 句话），说明为什么匹配/不匹配 - evidence: 简历中哪些经历匹配这个岗位（数组，每个元素是一段话，格式："你在XX项目的YY经验，正好匹配该岗位要求的ZZ能力"） - gaps: 简历缺少什么，导致不匹配（数组，每个元素是一段话） - suggestions: 如何改进简历以提高匹配度（数组，每个元素是一段话） 4. 给出 what_if 分析（数组，最多 3 个）： - expected_gain: 预期提升（如"+15 分"） - skill: 需要补充的技能/经历 - action: 具体行动 - time_needed: 需要的时间 - impact: 预期影响 5. 给出 action_plan（对象，包含 today 数组和 this_week 数组）： - today: 今天应该做什么（数组） - this_week: 本周应该做什么（数组） ## 输出格式 以 JSON 格式返回，结构如下（注意字段名必须完全一致，不要多余逗号）： {{ "analysis": {{ "skills": ["技能1", "技能2"], "projects": ["项目1", "项目2"], "experience": ["经历1", "经历2"], "education": "教育背景" }}, "matches": [ {{ "title": "岗位名称", "company": "公司名称", "city": "城市", "salary": "薪资", "match_score": 85, "decision": "立即投递", "reasoning": "你在腾讯AI Lab的RAG项目使用了FAISS+LangChain，正好匹配该岗位要求的向量检索能力", "evidence": ["你在腾讯AI Lab的RAG项目使用了FAISS+LangChain，正好匹配该岗位要求的向量检索能力", "你会用 PyTorch 微调模型，符合岗位要求的模型优化能力"], "gaps": ["缺少大规模分布式训练经验", "没有生产环境部署经验"], "suggestions": ["在简历里加一段分布式训练的项目描述", "补充模型部署的相关经验"] }} ], "what_if": [ {{ "expected_gain": "+15 分", "skill": "补充 RAG 项目经验", "action": "做一个基于 LangChain + FAISS 的 RAG 项目，部署到 HuggingFace Spaces", "time_needed": "2 周", "impact": "匹配度从 60 → 75，进入稳投区间" }} ], "action_plan": {{ "today": ["优化简历，突出 RAG 项目经验", "投递 2 个稳投岗位"], "this_week": ["完成一个新项目并写到简历里", "投递 5 个冲刺岗位", "准备面试：复习 PyTorch 和 Transformer"] }} }} 只返回 JSON，不要其他解释。确保 JSON 格式正确，字段名完全一致。""" # 3. 调用 LLM 分析 result = call_llm(prompt, max_tokens=4000, timeout=90) # 调试：保存原始返回 import time debug_dir = Path(__file__).resolve().parent.parent / "debug" debug_dir.mkdir(exist_ok=True) with open(debug_dir / f"llm_raw_{int(time.time())}.txt", "w", encoding="utf-8") as f: f.write(result) # 4. 解析结果 try: # 检查是否是错误信息 if result.strip().startswith("{"): err_obj = json.loads(result) if "error" in err_obj: return { "error": err_obj["error"], "matches": [], "analysis": {}, } # 提取 JSON（支持 ```json 代码块 或 裸 JSON） json_match = re.search(r"```json\s*([\s\S]*?)\s*```", result) if not json_match: json_match = re.search(r"\{[\s\S]*\}", result) if json_match: json_str = json_match.group(1) if json_match.lastindex else json_match.group() # 清理可能的 markdown 代码块标记 json_str = re.sub(r"^```json\s*", "", json_str) json_str = re.sub(r"\s*```$", "", json_str) report = json.loads(json_str) else: return { "error": "LLM 返回格式错误，无法解析 JSON", "raw": result[:500], "matches": [], "analysis": {}, } # 确保有必要的字段 if "matches" not in report: report["matches"] = [] if "analysis" not in report: report["analysis"] = {} if "what_if" not in report: report["what_if"] = [] if "action_plan" not in report: report["action_plan"] = {"today": [], "this_week": []} # 把岗位信息合并进 matches for i, m in enumerate(report.get("matches", [])): if i < len(jobs_limited): m["city"] = m.get("city") or jobs_limited[i].get("city", "") m["salary"] = m.get("salary") or jobs_limited[i].get("salary", "") if "company" not in m or not m["company"]: m["company"] = jobs_limited[i].get("company", "") return report except json.JSONDecodeError as e: return { "error": f"解析 LLM 返回失败: {e}", "raw": result[:500], "matches": [], "analysis": {}, } except Exception as e: return { "error": f"处理 LLM 返回失败: {e}", "raw": result[:500], "matches": [], "analysis": {}, } if __name__ == "__main__": test_resume = "张三 | 大模型应用算法工程师\n2022.09-2026.06 XX大学 计算机科学\n实习: LoRA微调Qwen2.5 + RAG知识库(FAISS)\n项目: Offer捕手 9Agent匹配系统 NDCG@10=0.87\n技能: Python PyTorch Transformers LangChain FAISS" test_goal = "大模型算法实习，北京" print("开始测试 run_pipeline...") result = run_pipeline(test_resume, test_goal) print(json.dumps(result, ensure_ascii=False, indent=2))