""" graph.py - 技术方向匹配分析（重新实现） 不再生成虚假岗位，专注分析简历适合的技术方向 """ 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 = 2000, timeout: int = 90) -> 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}"}) def run_pipeline(resume: str, goal: str) -> dict: """ 核心技术方向匹配分析 返回：技术方向匹配度、优势、不足、改进建议 """ prompt = f"""你是专业的求职顾问和技术方向分析师。 ## 用户简历 {resume[:3000]} ## 求职目标 {goal} ## 任务 分析这份简历，找出它适合的技术方向，并给出详细的匹配分析和改进建议。 ## 技术要求 你需要分析以下常见技术方向（根据简历内容选择最相关的 3-5 个方向）： 1. AI/LLM 算法（大模型、NLP、Transformer、PyTorch） 2. 推荐算法（推荐系统、CTR 预估、排序算法） 3. 数据分析（SQL、Python、统计分析、可视化） 4. 前端开发（React、Vue、TypeScript、HTML/CSS） 5. 后端开发（Python、Java、Go、微服务） 6. 计算机视觉（CV、目标检测、图像分割、OpenCV） 7. 强化学习（RL、游戏 AI、机器人） 8. 数据工程（大数据、Spark、Hadoop、数据仓库） ## 输出格式 以 JSON 格式返回，结构如下（注意字段名必须完全一致，确保 JSON 格式正确）： ```json {{ "resume_analysis": {{ "skills": ["技能1", "技能2"], "projects": ["项目1", "项目2"], "experience": ["经历1", "经历2"], "education": "教育背景字符串", "strengths": ["优势1", "优势2"], "weaknesses": ["不足1", "不足2"] }}, "direction_matches": [ {{ "direction": "技术方向名称（如：AI/LLM 算法）", "match_score": 85, "level": "高匹配/中匹配/低匹配", "description": "这个方向的一般要求（1-2句话）", "your_advantages": ["你的优势1", "你的优势2"], "your_gaps": ["你的不足1", "你的不足2"], "how_to_improve": [ {{ "action": "具体行动（如：做一个 RAG 项目）", "time_needed": "需要的时间（如：2 周）", "expected_gain": "预期提升（如：+15 分）" }} ], "reference_resources": ["学习资源1", "学习资源2"] }} ], "overall_suggestions": [ "总体建议1", "总体建议2" ], "action_plan": {{ "today": ["今天应该做什么"], "this_week": ["本周应该做什么"], "this_month": ["本月应该做什么"] }} }} ``` ## 要求 1. direction_matches 应该包含 3-5 个最相关的技术方向 2. 每个方向都要有具体的改进建议（actionable） 3. 改进建议要具体、可执行（不要说"学习 XX"，要说"做一个 XX 项目"） 4. 只返回 JSON，不要其他解释 5. 确保 JSON 格式正确，不要有 trailing commas """ # 调用 LLM 分析 result = call_llm(prompt, max_tokens=4000, timeout=120) # 调试：保存原始返回 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) # 解析结果 try: # 检查是否是错误信息 if result.strip().startswith("{"): err_obj = json.loads(result) if "error" in err_obj: return { "error": err_obj["error"], "direction_matches": [], "resume_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) # 尝试修复常见的 JSON 错误 json_str = re.sub(r",\s*\}", "}", json_str) json_str = re.sub(r",\s*\]", "]", json_str) report = json.loads(json_str) else: return { "error": "LLM 返回格式错误，无法解析 JSON", "raw": result[:500], "direction_matches": [], "resume_analysis": {}, } # 确保有必要的字段 if "direction_matches" not in report: report["direction_matches"] = [] if "resume_analysis" not in report: report["resume_analysis"] = {} if "overall_suggestions" not in report: report["overall_suggestions"] = [] if "action_plan" not in report: report["action_plan"] = {"today": [], "this_week": [], "this_month": []} return report except json.JSONDecodeError as e: return { "error": f"解析 LLM 返回失败: {e}", "raw": result[:500], "direction_matches": [], "resume_analysis": {}, } except Exception as e: return { "error": f"处理 LLM 返回失败: {e}", "raw": result[:500], "direction_matches": [], "resume_analysis": {}, } if __name__ == "__main__": test_resume = """张三 北京邮电大学 - 计算机科学与技术 - 本科 - 2026 届 手机：13800000000 | 邮箱：zhangsan@bupt.edu.cn 教育背景 2022.09-2026.06 北京邮电大学 计算机科学与技术 本科 GPA：3.8/4.0 专业排名：15/200 实习经历 2025.06-2025.09 腾讯 大模型算法实习生 - 参与混元大模型训练，负责 RLHF 数据处理 - 使用 PyTorch 实现 DPO 算法，提升模型效果 5% - 负责数据清洗与质量评估，处理 10 万+ 条训练数据 项目经历 2024.09-2025.06 基于 RAG 的简历匹配系统 - 使用 LangChain + DeepSeek API 构建简历-岗位匹配系统 - 实现向量检索（FAISS），召回准确率达 85% - 部署 Streamlit 应用，支持 PDF 简历解析 技能 Python, PyTorch, LangChain, DeepSeek API, Git, Linux """ test_goal = "北京 AI 大模型算法实习" print("开始测试 run_pipeline（新版本：技术方向匹配分析）...") result = run_pipeline(test_resume, test_goal) print("\n" + "="*50) print("测试结果：") print(json.dumps(result, ensure_ascii=False, indent=2)) if "direction_matches" in result and result["direction_matches"]: print("\n" + "="*50) print("技术方向匹配分析：") for i, dm in enumerate(result["direction_matches"], 1): print(f"\n{i}. {dm.get('direction')} - 匹配度: {dm.get('match_score')}%") print(f" 级别: {dm.get('level')}") print(f" 你的优势: {', '.join(dm.get('your_advantages', []))}") print(f" 你的不足: {', '.join(dm.get('your_gaps', []))}") print(f" 改进建议: ") for imp in dm.get('how_to_improve', []): print(f" - {imp.get('action')} ({imp.get('time_needed')}) → {imp.get('expected_gain')}")