159 lines
8.1 KiB
Python
159 lines
8.1 KiB
Python
"""求职助手 Agent 岗位简历优化 Service
|
|
|
|
主要功能：针对岗位并发优化简历；岗位匹配度评分。
|
|
依赖：resume_loader（简历统一查询）、customize_resume_store（定制简历存取+构建）、job_agent.resume_optimizer（岗位简历优化）、job_agent.match_scorer（匹配评分）
|
|
使用表：bg_user_resume + 5张子表（通过 resume_loader 查询）、bg_job（查岗位）、bg_user_job_customize_resume（定制简历）
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import time
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.ai.job_agent.resume_optimizer import optimize_summary, optimize_experience_record
|
|
from app.ai.job_agent.match_scorer import score_match as ai_score_match
|
|
from app.core.logger import log
|
|
from app.models.job import Job
|
|
from app.schemas.customize_resume import CustomizeResume, Education, Work, Internship, Project, Competition
|
|
from app.services.resume_loader import load_resume_detail
|
|
from app.services import customize_resume_store
|
|
|
|
|
|
class JobAgentChatService:
|
|
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def score_match(self, user_id: int, job_id: int, customize_resume_id: int | None = None, resume_id: int | None = None) -> int:
|
|
"""岗位匹配评分：查定制简历或原始简历 + 岗位 → 序列化简历 → 调AI评分，返回 0-100 整数"""
|
|
if customize_resume_id:
|
|
cr = await customize_resume_store.get_by_id(user_id, customize_resume_id)
|
|
if cr is None:
|
|
raise ValueError("定制简历不存在")
|
|
else:
|
|
detail = await load_resume_detail(self.session, resume_id, user_id)
|
|
cr = customize_resume_store.build_from_detail(detail)
|
|
job = await self._get_job(job_id)
|
|
resume_text = self._build_resume_text_from_cr(cr)
|
|
skill_tags = "、".join(job.skill_tags) if job.skill_tags else ""
|
|
job_desc = f"{job.description or ''}\n{job.requirement or ''}"
|
|
return await ai_score_match(job.title or "", skill_tags, job_desc, resume_text)
|
|
|
|
@staticmethod
|
|
def _build_resume_text_from_cr(cr: CustomizeResume) -> str:
|
|
"""将定制简历序列化为文本供 AI 评估使用"""
|
|
r = cr.resume
|
|
parts: list[str] = []
|
|
if r.name:
|
|
parts.append(f"姓名：{r.name}")
|
|
if r.skills:
|
|
parts.append(f"技能：{'、'.join(r.skills)}")
|
|
if r.certificates:
|
|
parts.append(f"证书：{'、'.join(r.certificates)}")
|
|
if r.summary:
|
|
parts.append(f"个人概述：{r.summary}")
|
|
if cr.education:
|
|
parts.append("教育经历：")
|
|
for e in cr.education:
|
|
parts.append(f" - {e.school} {e.major} {e.degree} {e.study_type}".rstrip())
|
|
if cr.work:
|
|
parts.append("工作经历：")
|
|
for w in cr.work:
|
|
parts.append(f" - {w.company_name} {w.position}".rstrip())
|
|
parts.extend(f" {p.text}" for p in w.description if p.text)
|
|
if cr.internship:
|
|
parts.append("实习经历：")
|
|
for i in cr.internship:
|
|
parts.append(f" - {i.company_name} {i.position}".rstrip())
|
|
parts.extend(f" {p.text}" for p in i.description if p.text)
|
|
if cr.project:
|
|
parts.append("项目经历：")
|
|
for p in cr.project:
|
|
parts.append(f" - {p.project_name} {p.role}".rstrip())
|
|
parts.extend(f" {seg.text}" for seg in p.description if seg.text)
|
|
if cr.competition:
|
|
parts.append("竞赛经历：")
|
|
for c in cr.competition:
|
|
parts.append(f" - {c.competition_name} {c.award}".rstrip())
|
|
return "\n".join(parts) if parts else "暂无简历信息"
|
|
|
|
async def optimize_resume(self, user_id: int, resume_id: int, job_id: int) -> dict:
|
|
"""针对岗位优化简历：查简历+岗位 → 构建定制简历 → 按单条记录并发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. 构建并发任务（按单条记录粒度）
|
|
tasks: list[tuple[str, int, object]] = []
|
|
job_desc = f"{job.description or ''}\n{job.requirement or ''}"
|
|
if 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, resume_id=detail.resume.id, resume_name=f"{detail.resume.resume_name or ''}-定制")
|
|
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)]:
|
|
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:
|
|
"""查岗位"""
|
|
result = await self.session.execute(select(Job).where(Job.id == job_id))
|
|
job = result.scalar_one_or_none()
|
|
if not job:
|
|
raise ValueError("岗位不存在")
|
|
return job
|
|
|
|
@staticmethod
|
|
def _apply_optimize_result(cr: CustomizeResume, key: str, idx: int, result) -> None:
|
|
"""将 AI 优化结果应用到定制简历（单条记录粒度）"""
|
|
if key == "summary" and isinstance(result, str):
|
|
cr.resume.summary = 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__}")
|