offerpai/offerpai_python_ai
6
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

添加agent优化简历接口

Browse Source
This commit is contained in:
zk
2026-04-24 20:13:17 +08:00
parent a4a17d90c8
commit 4c2627dbb2
7 changed files with 235 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/services/customize_resume_store.py
+54 -5
View File
@@ -1,11 +1,17 @@
"""定制简历 Redis 存取模块
"""定制简历 Redis 存取 + 数据转换模块
提供定制简历的保存（自动回滚备份）、查询、回滚能力。
提供定制简历的保存（自动回滚备份）、查询、回滚、从 ResumeDetail 构建能力。
各 Service 统一复用，不直接操作 Redis key。
"""
import random
import string
from app.core.redis import RedisManager
from app.schemas.customize_resume import CustomizeResume
from app.schemas.customize_resume import (
CustomizeResume, ResumeProfile, Education, Work, Internship, Project, Competition, Paragraph,
)
from app.services.resume_loader import ResumeDetail
# Redis 常量
KEY_PREFIX = "customize:resume:"
@@ -13,16 +19,59 @@ EXPIRE = 12 * 60 * 60 # 12小时
ROLLBACK_KEY_PREFIX = "customize:resume:rollback:"
ROLLBACK_EXPIRE = 30 * 60 # 30分钟
_CHARS = string.ascii_letters + string.digits
def _rand_id() -> str:
"""生成随机8位字符串标识"""
return "".join(random.choices(_CHARS, k=8))
def _build_paragraphs(description: list[dict] | None) -> list[Paragraph]:
"""将数据库 description [{id, text}] 转为 Paragraph 列表"""
if not description:
return []
return [Paragraph(id=_rand_id(), text=item.get("text", "")) for item in description if isinstance(item, dict)]
def build_from_detail(detail: ResumeDetail) -> CustomizeResume:
"""从 ResumeDetail 组装 CustomizeResume"""
resume = detail.resume
profile = ResumeProfile(
avatarUrl=resume.avatar_url or "", name=resume.name or "", email=resume.email or "",
mobileNumber=resume.mobile_number or "", city=resume.city or "",
wechatNumber=resume.wechat_number or "", portfolioUrl=resume.portfolio_url or "",
skills=resume.skills or [], certificates=resume.certificates or [],
summary=resume.summary or "",
)
return CustomizeResume(
resume=profile,
education=[Education(id=_rand_id(), school=r.school or "", major=r.major or "",
degree=r.degree or "", studyType=r.study_type or "",
startDate=r.start_date or "", endDate=r.end_date or "",
description=_build_paragraphs(r.description)) for r in detail.education],
work=[Work(id=_rand_id(), companyName=r.company_name or "", position=r.position or "",
startDate=r.start_date or "", endDate=r.end_date or "",
description=_build_paragraphs(r.description)) for r in detail.work],
internship=[Internship(id=_rand_id(), companyName=r.company_name or "", position=r.position or "",
startDate=r.start_date or "", endDate=r.end_date or "",
description=_build_paragraphs(r.description)) for r in detail.internship],
project=[Project(id=_rand_id(), companyName=r.company_name or "", projectName=r.project_name or "",
role=r.role or "", startDate=r.start_date or "", endDate=r.end_date or "",
description=_build_paragraphs(r.description)) for r in detail.project],
competition=[Competition(id=_rand_id(), competitionName=r.competition_name or "", award=r.award or "",
awardDate=r.award_date or "",
description=_build_paragraphs(r.description)) for r in detail.competition],
)
async def save(user_id: int, cr: CustomizeResume) -> None:
"""保存定制简历，自动备份旧版本到回滚 key"""
key = f"{KEY_PREFIX}{user_id}"
rollback_key = f"{ROLLBACK_KEY_PREFIX}{user_id}"
# 备份旧数据
old_data = await RedisManager.client.get(key)
if old_data:
await RedisManager.client.set(rollback_key, old_data, ex=ROLLBACK_EXPIRE)
# 存新数据
await RedisManager.client.set(key, cr.model_dump_json(by_alias=True), ex=EXPIRE)
app/services/job_agent_chat_service.py
+68 -4
View File
@@ -1,14 +1,23 @@
"""求职助手 Agent 对话 Service
"""求职助手 Agent 对话 + 岗位简历优化 Service
主要功能：查询简历数据，调用 AI 模块完成对话。
依赖：resume_loader（简历统一查询）、job_agent.chat AI 模块
使用表：bg_user_resume + 5张子表（通过 resume_loader 查询）
主要功能：查询简历数据，调用 AI 模块完成对话；针对岗位并发优化简历。
依赖：resume_loader（简历统一查询）、customize_resume_store（定制简历存取+构建）、job_agent.chat AI 模块、job_agent.resume_optimizer（岗位简历优化）
使用表：bg_user_resume + 5张子表（通过 resume_loader 查询）、bg_job（查岗位）
"""
import asyncio
import json
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.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 ResumeDetail, load_resume_detail
from app.services import customize_resume_store
class JobAgentChatService:
@@ -60,3 +69,58 @@ class JobAgentChatService:
for r in detail.competition:
parts.append(f" - {r.competition_name or ''} {r.award or ''}")
return "\n".join(parts) if parts else "暂无简历信息"
async def optimize_resume(self, user_id: int, resume_id: int, job_id: int) -> dict:
"""针对岗位优化简历：查简历+岗位 → 构建定制简历 → 并发AI优化 → 存Redis → 返回"""
# 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 = []
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)))
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. 存 Redis
await customize_resume_store.save(user_id, cr)
# 5. 返回
return cr.model_dump(by_alias=True)
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, 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]
app/services/skill_gap_service.py
+2 -49
View File
@@ -8,8 +8,6 @@
import asyncio
import json
import random
import string
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
@@ -21,15 +19,13 @@ from app.ai.skill_gap_analyzer.analyzer import (
from app.ai.skill_gap_analyzer.prompts import MODULE_SCHEMAS
from app.core.logger import log
from app.schemas.customize_resume import (
CustomizeResume, ResumeProfile, Education, Work, Internship, Project, Competition, Paragraph,
CustomizeResume, ResumeProfile, Education, Work, Internship, Project, Competition,
)
from app.models.job import Job
from app.models.user_resume import UserResume
from app.services.resume_loader import ResumeDetail, load_resume_detail, load_default_resume_detail
from app.services import customize_resume_store
_CHARS = string.ascii_letters + string.digits
# 模块名 → 中文标签映射
_MODULE_LABELS = {
"resume": "个人简介",
@@ -41,18 +37,6 @@ _MODULE_LABELS = {
}
def _rand_id() -> str:
"""生成随机8位字符串标识"""
return "".join(random.choices(_CHARS, k=8))
def _build_paragraphs(description: list[dict] | None) -> list[Paragraph]:
"""将数据库 description [{id, text}] 转为 Paragraph 列表，id 用随机8位替换"""
if not description:
return []
return [Paragraph(id=_rand_id(), text=item.get("text", "")) for item in description if isinstance(item, dict)]
def _build_resume_json(detail: ResumeDetail) -> str:
"""拼装简历 JSON 字符串供 AI 使用"""
resume = detail.resume
@@ -120,7 +104,7 @@ class SkillGapService:
detail = await load_resume_detail(self.session, resume_id, user_id)
job = await self._get_job(job_id)
# 2. 组装基础定制简历
cr = self._build_customize_resume(detail)
cr = customize_resume_store.build_from_detail(detail)
# 3. 并发 AI 优化
tasks = []
job_desc = f"{job.description or ''}\n{job.requirement or ''}"
@@ -339,34 +323,3 @@ class SkillGapService:
if not job:
raise ValueError("岗位不存在")
return job
def _build_customize_resume(self, detail: ResumeDetail) -> CustomizeResume:
"""从 ResumeDetail 组装 CustomizeResume"""
resume = detail.resume
profile = ResumeProfile(
avatarUrl=resume.avatar_url or "", name=resume.name or "", email=resume.email or "",
mobileNumber=resume.mobile_number or "", city=resume.city or "",
wechatNumber=resume.wechat_number or "", portfolioUrl=resume.portfolio_url or "",
skills=resume.skills or [], certificates=resume.certificates or [],
summary=resume.summary or "",
)
return CustomizeResume(
resume=profile,
education=[Education(id=_rand_id(), school=r.school or "", major=r.major or "",
degree=r.degree or "", studyType=r.study_type or "",
startDate=r.start_date or "", endDate=r.end_date or "",
description=_build_paragraphs(r.description)) for r in detail.education],
work=[Work(id=_rand_id(), companyName=r.company_name or "", position=r.position or "",
startDate=r.start_date or "", endDate=r.end_date or "",
description=_build_paragraphs(r.description)) for r in detail.work],
internship=[Internship(id=_rand_id(), companyName=r.company_name or "", position=r.position or "",
startDate=r.start_date or "", endDate=r.end_date or "",
description=_build_paragraphs(r.description)) for r in detail.internship],
project=[Project(id=_rand_id(), companyName=r.company_name or "", projectName=r.project_name or "",
role=r.role or "", startDate=r.start_date or "", endDate=r.end_date or "",
description=_build_paragraphs(r.description)) for r in detail.project],
competition=[Competition(id=_rand_id(), competitionName=r.competition_name or "", award=r.award or "",
awardDate=r.award_date or "",
description=_build_paragraphs(r.description)) for r in detail.competition],
)