From ed8663cc42002b329649234fbebcac6524002386 Mon Sep 17 00:00:00 2001 From: zk Date: Wed, 13 May 2026 15:07:41 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=20agent=20=E5=AE=9A=E5=88=B6?= =?UTF-8?q?=E7=AE=80=E5=8E=86=E7=94=9F=E6=88=90=E9=80=9F=E5=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/ai/job_agent/resume_optimizer.py | 30 +++++---- app/services/job_agent_chat_service.py | 90 ++++++++++++++++---------- 2 files changed, 76 insertions(+), 44 deletions(-) diff --git a/app/ai/job_agent/resume_optimizer.py b/app/ai/job_agent/resume_optimizer.py index e0a1a6b..94d0b17 100644 --- a/app/ai/job_agent/resume_optimizer.py +++ b/app/ai/job_agent/resume_optimizer.py @@ -1,9 +1,11 @@ """求职助手 - 岗位简历优化 AI 引擎 -针对目标岗位并发优化简历（summary + 经历子表）。 +针对目标岗位并发优化简历（summary + 经历子表，按单条记录粒度并发）。 依赖：LLM 枚举、job_agent/prompts、parse_llm_json """ +import time + from langchain_core.output_parsers import StrOutputParser from langchain_core.prompts import ChatPromptTemplate @@ -16,40 +18,46 @@ from app.tool.json_helper import parse_llm_json _summary_chain = ( ChatPromptTemplate.from_messages([("system", RESUME_SUMMARY_OPTIMIZE_PROMPT), ("human", "请开始优化。")]) - | LLM.ZM_GPT_5_4.create(temperature=0.3) + | LLM.ZM_GPT_5_4_MINI.create(temperature=0.3) | StrOutputParser() ) async def optimize_summary(job_title: str, job_description: str, original_summary: str) -> str: """针对岗位优化个人概述""" + t0 = time.monotonic() try: - return await _summary_chain.ainvoke({ + result = await _summary_chain.ainvoke({ "job_title": job_title, "job_description": job_description or "", "original_summary": original_summary or "暂无", }) + log.info(f"岗位简历summary优化完成 ({round(time.monotonic() - t0, 2)}s)") + return result except Exception as e: - log.warning(f"岗位简历summary优化失败: {e}") + log.warning(f"岗位简历summary优化失败: {e} ({round(time.monotonic() - t0, 2)}s)") return original_summary -# ===== 经历优化 ===== +# ===== 单条经历优化 ===== _experience_chain = ( ChatPromptTemplate.from_messages([("system", RESUME_EXPERIENCE_OPTIMIZE_PROMPT), ("human", "请开始优化。")]) - | LLM.ZM_GPT_5_4.create(temperature=0.3) + | LLM.ZM_GPT_5_4_MINI.create(temperature=0.3) | StrOutputParser() ) -async def optimize_experience(job_title: str, job_description: str, module_data: str) -> list | dict | None: - """针对岗位优化经历模块描述，返回修改后的完整模块数据""" +async def optimize_experience_record(job_title: str, job_description: str, record_json: str) -> dict | None: + """针对岗位优化单条经历记录，返回修改后的记录数据""" + t0 = time.monotonic() try: raw = await _experience_chain.ainvoke({ "job_title": job_title, "job_description": job_description or "", - "original_module_data": module_data, + "original_module_data": record_json, }) - return parse_llm_json(raw) + result = parse_llm_json(raw) + log.info(f"岗位简历经历记录优化完成 ({round(time.monotonic() - t0, 2)}s)") + return result except Exception as e: - log.warning(f"岗位简历经历优化失败: {e}") + log.warning(f"岗位简历经历记录优化失败: {e} ({round(time.monotonic() - t0, 2)}s)") return None diff --git a/app/services/job_agent_chat_service.py b/app/services/job_agent_chat_service.py index ab4759a..a8d3849 100644 --- a/app/services/job_agent_chat_service.py +++ b/app/services/job_agent_chat_service.py @@ -7,12 +7,13 @@ import asyncio import json +import time from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.ai.job_agent.chat import agent_chat -from app.ai.job_agent.resume_optimizer import optimize_summary, optimize_experience +from app.ai.job_agent.resume_optimizer import optimize_summary, optimize_experience_record from app.core.logger import log from app.models.job import Job from app.schemas.customize_resume import CustomizeResume, Education, Work, Internship, Project, Competition @@ -71,35 +72,47 @@ class JobAgentChatService: return "\n".join(parts) if parts else "暂无简历信息" async def optimize_resume(self, user_id: int, resume_id: int, job_id: int) -> dict: - """针对岗位优化简历：查简历+岗位 → 构建定制简历 → 并发AI优化 → 存Redis → 返回""" + """针对岗位优化简历：查简历+岗位 → 构建定制简历 → 按单条记录并发AI优化 → 存数据库 → 返回""" # 1. 查简历 + 岗位 detail = await load_resume_detail(self.session, resume_id, user_id) job = await self._get_job(job_id) # 2. 构建定制简历 cr = customize_resume_store.build_from_detail(detail) - # 3. 并发 AI 优化（summary + 5张子表经历） - tasks = [] + # 3. 构建并发任务（按单条记录粒度） + tasks: list[tuple[str, int, object]] = [] job_desc = f"{job.description or ''}\n{job.requirement or ''}" if cr.resume.summary: - tasks.append(("summary", optimize_summary(job.title or "", job_desc, cr.resume.summary))) + tasks.append(("summary", 0, optimize_summary(job.title or "", job_desc, cr.resume.summary))) + for mod_name, idx, record_json in self._experience_tasks(cr): + tasks.append((mod_name, idx, optimize_experience_record(job.title or "", job_desc, record_json))) + log.info(f"岗位简历优化开始: {len(tasks)}个并发任务 [job={job_id}, resume={resume_id}]") + # 4. 并发执行 + if tasks: + t0 = time.monotonic() + results = await asyncio.gather(*[t[2] for t in tasks], return_exceptions=True) + log.info(f"岗位简历优化全部完成, 总耗时={round(time.monotonic() - t0, 2)}s") + for (mod_name, idx, _), result in zip(tasks, results): + if isinstance(result, Exception): + log.warning(f"岗位简历优化[{mod_name}[{idx}]]失败: {result}") + continue + if result is None: + continue + self._apply_optimize_result(cr, mod_name, idx, result) + # 5. 存数据库 + await customize_resume_store.save(user_id, job_id, cr) + log.info(f"岗位简历优化已保存 [user={user_id}, job={job_id}]") + # 6. 返回 + return cr.model_dump(by_alias=True) + + @staticmethod + def _experience_tasks(cr: CustomizeResume) -> list[tuple[str, int, str]]: + """构建各子表的 AI 优化任务列表，按单条记录拆分""" + result: list[tuple[str, int, str]] = [] for name, items in [("education", cr.education), ("work", cr.work), ("internship", cr.internship), ("project", cr.project), ("competition", cr.competition)]: - if items: - rows_json = json.dumps([item.model_dump(by_alias=True) for item in items], ensure_ascii=False) - tasks.append((name, optimize_experience(job.title or "", job_desc, rows_json))) - # 执行并发 - if tasks: - keys = [t[0] for t in tasks] - results = await asyncio.gather(*[t[1] for t in tasks], return_exceptions=True) - for key, result in zip(keys, results): - if isinstance(result, Exception): - log.warning(f"岗位简历优化[{key}]失败: {result}") - continue - self._apply_optimize_result(cr, key, result) - # 4. 存数据库 - await customize_resume_store.save(user_id, job_id, cr) - # 5. 返回 - return cr.model_dump(by_alias=True) + for idx, item in enumerate(items or []): + result.append((name, idx, json.dumps(item.model_dump(by_alias=True), ensure_ascii=False))) + return result async def _get_job(self, job_id: int) -> Job: """查岗位""" @@ -110,17 +123,28 @@ class JobAgentChatService: return job @staticmethod - def _apply_optimize_result(cr: CustomizeResume, key: str, result) -> None: - """将 AI 优化结果应用到定制简历""" + def _apply_optimize_result(cr: CustomizeResume, key: str, idx: int, result) -> None: + """将 AI 优化结果应用到定制简历（单条记录粒度）""" if key == "summary" and isinstance(result, str): cr.resume.summary = result - elif key == "education" and isinstance(result, list): - cr.education = [Education.model_validate(item) for item in result] - elif key == "work" and isinstance(result, list): - cr.work = [Work.model_validate(item) for item in result] - elif key == "internship" and isinstance(result, list): - cr.internship = [Internship.model_validate(item) for item in result] - elif key == "project" and isinstance(result, list): - cr.project = [Project.model_validate(item) for item in result] - elif key == "competition" and isinstance(result, list): - cr.competition = [Competition.model_validate(item) for item in result] + return + model_map = {"education": Education, "work": Work, "internship": Internship, "project": Project, "competition": Competition} + list_map = {"education": cr.education, "work": cr.work, "internship": cr.internship, "project": cr.project, "competition": cr.competition} + model_cls = model_map.get(key) + items = list_map.get(key) + if model_cls is None or items is None: + log.warning(f"未知优化模块: {key}") + return + if isinstance(result, dict): + try: + items[idx] = model_cls.model_validate(result) + except (IndexError, Exception) as e: + log.warning(f"应用优化结果[{key}[{idx}]]失败: {e}") + elif isinstance(result, list) and len(result) == 1 and isinstance(result[0], dict): + # 兼容 LLM 偶尔返回单元素数组的情况 + try: + items[idx] = model_cls.model_validate(result[0]) + except (IndexError, Exception) as e: + log.warning(f"应用优化结果[{key}[{idx}]]失败(数组兼容): {e}") + else: + log.warning(f"优化结果格式异常[{key}[{idx}]]: type={type(result).__name__}")