413 lines
14 KiB
Python
413 lines
14 KiB
Python
"""AI service for CV parsing, matching, and generation using GLM-5.2 via Ollama Cloud."""
|
|
import json
|
|
import re
|
|
from datetime import datetime, date
|
|
from dateutil.relativedelta import relativedelta
|
|
from openai import OpenAI
|
|
from config import get_settings
|
|
|
|
settings = get_settings()
|
|
|
|
# LLM client
|
|
_client = None
|
|
|
|
def get_client():
|
|
global _client
|
|
if _client is None:
|
|
_client = OpenAI(
|
|
api_key=settings.llm_api_key,
|
|
base_url=settings.llm_base_url,
|
|
)
|
|
return _client
|
|
|
|
|
|
def llm_chat(messages, temperature=0.3, max_tokens=4000, response_format=None):
|
|
"""Call the LLM with messages and return the response text."""
|
|
client = get_client()
|
|
kwargs = {
|
|
"model": settings.llm_model,
|
|
"messages": messages,
|
|
"temperature": temperature,
|
|
"max_tokens": max_tokens,
|
|
}
|
|
if response_format:
|
|
kwargs["response_format"] = response_format
|
|
resp = client.chat.completions.create(**kwargs)
|
|
return resp.choices[0].message.content
|
|
|
|
|
|
def extract_json(text):
|
|
"""Extract JSON from LLM response, handling markdown code blocks and truncation."""
|
|
json_str = text.strip()
|
|
if json_str.startswith("```"):
|
|
json_str = re.sub(r'^```(?:json)?\s*', '', json_str)
|
|
json_str = re.sub(r'\s*```$', '', json_str)
|
|
|
|
try:
|
|
return json.loads(json_str)
|
|
except json.JSONDecodeError:
|
|
# Try to fix truncated JSON by closing open braces/brackets
|
|
open_b = json_str.count('{')
|
|
close_b = json_str.count('}')
|
|
open_arr = json_str.count('[')
|
|
close_arr = json_str.count(']')
|
|
if open_b > close_b or open_arr > close_arr:
|
|
# Remove any trailing partial content after last complete element
|
|
# then close the structures
|
|
json_str = json_str.rstrip()
|
|
# Remove trailing incomplete key-value (e.g. cut off mid-string)
|
|
json_str = re.sub(r'[\s,]*"[^"]*":\s*"?[^",}\]]*$', '', json_str)
|
|
json_str += '}' * max(0, open_b - json_str.count('}'))
|
|
json_str += ']' * max(0, open_arr - json_str.count(']'))
|
|
return json.loads(json_str)
|
|
raise
|
|
|
|
|
|
# ============================================================
|
|
# CV PARSING
|
|
# ============================================================
|
|
|
|
PARSE_PROMPT = """You are a CV/resume parser. Analyze the following CV text and extract structured information.
|
|
|
|
Return ONLY a valid JSON object with this exact structure:
|
|
{
|
|
"first_name": "",
|
|
"last_name": "",
|
|
"email": "",
|
|
"phone": "",
|
|
"address": "",
|
|
"linkedin": "",
|
|
"github": "",
|
|
"website": "",
|
|
"summary": "",
|
|
"skills": [
|
|
{
|
|
"skill_name": "",
|
|
"skill_category": "Programming|Database|Cloud|DevOps|Tools|Framework|Language|Soft Skill|Other",
|
|
"proficiency": "Expert|Advanced|Intermediate|Beginner",
|
|
"start_date": "YYYY-MM-DD or YYYY-01-01 if only year is known",
|
|
"end_date": "YYYY-MM-DD or null if still active/present"
|
|
}
|
|
],
|
|
"experience": [
|
|
{
|
|
"company": "",
|
|
"position": "",
|
|
"location": "",
|
|
"start_date": "YYYY-MM-DD or null",
|
|
"end_date": "YYYY-MM-DD or null if current",
|
|
"description": "",
|
|
"achievements": ["bullet point 1", "bullet point 2"],
|
|
"skills_used": ["skill1", "skill2"]
|
|
}
|
|
],
|
|
"education": [
|
|
{
|
|
"institution": "",
|
|
"degree": "",
|
|
"field_of_study": "",
|
|
"start_date": "YYYY-MM-DD or null",
|
|
"end_date": "YYYY-MM-DD or null",
|
|
"grade": "",
|
|
"description": ""
|
|
}
|
|
],
|
|
"certifications": [
|
|
{
|
|
"name": "",
|
|
"issuer": "",
|
|
"issue_date": "YYYY-MM-DD or null",
|
|
"expiry_date": "YYYY-MM-DD or null",
|
|
"credential_id": ""
|
|
}
|
|
]
|
|
}
|
|
|
|
Rules:
|
|
- Extract dates as accurately as possible. If only a year is mentioned, use YYYY-01-01.
|
|
- If a skill or job is described as "present" or "current", set end_date to null.
|
|
- For skills, infer the start_date from when they first appear in experience/education.
|
|
- Be thorough — extract ALL skills, experience entries, education, and certifications.
|
|
- If a field is not found, use empty string "" or null as appropriate.
|
|
- Return ONLY the JSON, no other text.
|
|
|
|
CV Text to parse:
|
|
---
|
|
__CV_TEXT__
|
|
---"""
|
|
|
|
def parse_cv(cv_text: str) -> dict:
|
|
"""Parse a CV's raw text into structured data using AI."""
|
|
max_chars = 12000
|
|
if len(cv_text) > max_chars:
|
|
cv_text = cv_text[:max_chars] + "\n[...truncated...]"
|
|
|
|
prompt = PARSE_PROMPT.replace("__CV_TEXT__", cv_text)
|
|
messages = [
|
|
{"role": "system", "content": "You are a precise CV parser that outputs only valid JSON."},
|
|
{"role": "user", "content": prompt}
|
|
]
|
|
resp = llm_chat(messages, temperature=0.1, max_tokens=4000)
|
|
return extract_json(resp)
|
|
|
|
|
|
# ============================================================
|
|
# DYNAMIC EXPERIENCE CALCULATION
|
|
# ============================================================
|
|
|
|
def calculate_years_experience(start_date, end_date, reference_date=None):
|
|
"""Calculate years of experience dynamically based on reference date.
|
|
|
|
If end_date is None (still active), use reference_date (or today) as the end.
|
|
Returns a float representing years of experience.
|
|
"""
|
|
if reference_date is None:
|
|
reference_date = date.today()
|
|
elif isinstance(reference_date, str):
|
|
reference_date = datetime.strptime(reference_date, "%Y-%m-%d").date()
|
|
|
|
if isinstance(start_date, str):
|
|
start_date = datetime.strptime(start_date, "%Y-%m-%d").date()
|
|
|
|
if end_date is None:
|
|
end_date = reference_date
|
|
elif isinstance(end_date, str):
|
|
end_date = datetime.strptime(end_date, "%Y-%m-%d").date()
|
|
|
|
# Cap end_date at reference_date (don't count future time)
|
|
if end_date > reference_date:
|
|
end_date = reference_date
|
|
|
|
if start_date > end_date:
|
|
return 0.0
|
|
|
|
delta = relativedelta(end_date, start_date)
|
|
years = delta.years + delta.months / 12.0
|
|
return round(years, 1)
|
|
|
|
|
|
def calculate_skill_years(skill_start, skill_end, reference_date=None):
|
|
"""Calculate years for a specific skill."""
|
|
return calculate_years_experience(skill_start, skill_end, reference_date)
|
|
|
|
|
|
# ============================================================
|
|
# CV MATCHING
|
|
# ============================================================
|
|
|
|
MATCH_PROMPT = """You are a recruitment AI that matches candidates to job requirements.
|
|
|
|
Given a candidate's structured CV data and a list of position requirements, determine how well
|
|
the candidate matches each position.
|
|
|
|
Return ONLY a valid JSON array:
|
|
[
|
|
{
|
|
"position_title": "",
|
|
"match_score": 0-100,
|
|
"matching_skills": ["skill1", "skill2"],
|
|
"missing_skills": ["skill3"],
|
|
"experience_analysis": "Brief analysis of experience relevance",
|
|
"recommendation": "strong_match|moderate_match|weak_match|no_match",
|
|
"reasoning": "Detailed explanation"
|
|
}
|
|
]
|
|
|
|
Calculate experience years dynamically as of __REFERENCE_DATE__.
|
|
|
|
Candidate CV Data (JSON):
|
|
__CANDIDATE_DATA__
|
|
|
|
Position Requirements (JSON):
|
|
__REQUIREMENTS__
|
|
|
|
Return ONLY the JSON array, no other text."""
|
|
|
|
def match_candidate_to_requirements(candidate_data: dict, requirements: list, reference_date: date = None) -> list:
|
|
"""Match a candidate against a list of position requirements."""
|
|
ref_str = (reference_date or date.today()).isoformat()
|
|
|
|
prompt = (MATCH_PROMPT
|
|
.replace("__REFERENCE_DATE__", ref_str)
|
|
.replace("__CANDIDATE_DATA__", json.dumps(candidate_data, default=str))
|
|
.replace("__REQUIREMENTS__", json.dumps(requirements, default=str)))
|
|
|
|
messages = [
|
|
{"role": "system", "content": "You are a precise recruitment matching AI that outputs only valid JSON."},
|
|
{"role": "user", "content": prompt}
|
|
]
|
|
resp = llm_chat(messages, temperature=0.2, max_tokens=3000)
|
|
return extract_json(resp)
|
|
|
|
|
|
# ============================================================
|
|
# CV GENERATION (regenerate CV aligned to requirement)
|
|
# ============================================================
|
|
|
|
GENERATE_PROMPT = """You are an expert CV writer who regenerates CVs to align with specific job requirements.
|
|
|
|
IMPORTANT RULES:
|
|
1. DO NOT invent or fabricate skills, experience, or qualifications the candidate does not have.
|
|
2. DO reorder and emphasize relevant experience that matches the requirement.
|
|
3. DO rephrase descriptions to highlight relevant aspects without lying.
|
|
4. DO use dynamic experience calculations based on the generation date: __GENERATION_DATE__
|
|
5. DO format the CV according to the template structure provided.
|
|
6. DO ensure all dates and durations are accurate and dynamically calculated.
|
|
7. DO NOT change job titles to something the candidate didn't do. You can rephrase but not invent.
|
|
|
|
For each skill, calculate years of experience dynamically:
|
|
- If a skill started in 2020-01-01 and is still active (end_date=null),
|
|
and the generation date is __GENERATION_DATE__, the experience is calculated from start to generation date.
|
|
- If a skill has both start and end dates, calculate the duration between them.
|
|
|
|
Template Structure:
|
|
__TEMPLATE_STRUCTURE__
|
|
|
|
Candidate Data (with full CV):
|
|
__CANDIDATE_DATA__
|
|
|
|
Target Position Requirement:
|
|
__REQUIREMENT__
|
|
|
|
Generate the CV content. Return a JSON object:
|
|
{
|
|
"generated_content": "The full formatted CV as HTML or structured text",
|
|
"generated_data": {
|
|
"skills_with_years": [
|
|
{
|
|
"skill_name": "",
|
|
"years": 0.0,
|
|
"start_date": "",
|
|
"end_date": "",
|
|
"calculated_as_of": "__GENERATION_DATE__"
|
|
}
|
|
],
|
|
"experience_highlights": ["relevant achievement 1", "relevant achievement 2"],
|
|
"summary_aligned": "A summary paragraph aligned to the requirement"
|
|
},
|
|
"changes_made": "List of what was changed/reordered/emphasized (for transparency)"
|
|
}
|
|
|
|
Return ONLY the JSON, no other text."""
|
|
|
|
def generate_aligned_cv(
|
|
candidate_data: dict,
|
|
requirement: dict,
|
|
template_structure: dict,
|
|
generation_date: date = None
|
|
) -> dict:
|
|
"""Generate a CV aligned to a specific requirement."""
|
|
gen_date_str = (generation_date or date.today()).isoformat()
|
|
|
|
prompt = (GENERATE_PROMPT
|
|
.replace("__GENERATION_DATE__", gen_date_str)
|
|
.replace("__TEMPLATE_STRUCTURE__", json.dumps(template_structure, default=str))
|
|
.replace("__CANDIDATE_DATA__", json.dumps(candidate_data, default=str))
|
|
.replace("__REQUIREMENT__", json.dumps(requirement, default=str)))
|
|
|
|
messages = [
|
|
{"role": "system", "content": "You are an expert CV writer that outputs only valid JSON."},
|
|
{"role": "user", "content": prompt}
|
|
]
|
|
resp = llm_chat(messages, temperature=0.4, max_tokens=4000)
|
|
return extract_json(resp)
|
|
|
|
|
|
# ============================================================
|
|
# TEMPLATE GENERATION
|
|
# ============================================================
|
|
|
|
TEMPLATE_GEN_PROMPT = """You are a CV template designer. Based on the user's description, create a CV template structure.
|
|
|
|
The template structure defines the sections and layout of a CV. Return a JSON object:
|
|
{
|
|
"sections": [
|
|
{
|
|
"name": "Section Name",
|
|
"type": "header|summary|experience|education|skills|certifications|projects|custom",
|
|
"order": 1,
|
|
"fields": ["field1", "field2"],
|
|
"styling": {
|
|
"show_years": true,
|
|
"group_by_category": true,
|
|
"show_proficiency": false,
|
|
"max_items": null
|
|
}
|
|
}
|
|
],
|
|
"styling": "CSS string for the template, dark or light theme"
|
|
}
|
|
|
|
User description of desired template:
|
|
__DESCRIPTION__
|
|
|
|
If the user wants a specific industry or role focus, tailor the template accordingly.
|
|
Return ONLY the JSON, no other text."""
|
|
|
|
def generate_template(description: str) -> dict:
|
|
"""Generate a CV template from a text description using AI."""
|
|
prompt = TEMPLATE_GEN_PROMPT.replace("__DESCRIPTION__", description)
|
|
messages = [
|
|
{"role": "system", "content": "You are a CV template designer that outputs only valid JSON."},
|
|
{"role": "user", "content": prompt}
|
|
]
|
|
|
|
# Retry with increasing max_tokens if JSON parsing fails
|
|
for attempt, max_tok in enumerate([2000, 3000, 4000]):
|
|
try:
|
|
resp = llm_chat(messages, temperature=0.5, max_tokens=max_tok)
|
|
return extract_json(resp)
|
|
except json.JSONDecodeError as e:
|
|
if attempt < 2:
|
|
# Retry with more tokens
|
|
continue
|
|
# Last attempt failed — try to salvage partial JSON
|
|
try:
|
|
# Attempt to fix truncated JSON by closing braces
|
|
json_str = resp.strip()
|
|
if json_str.startswith("```"):
|
|
json_str = re.sub(r'^```(?:json)?\s*', '', json_str)
|
|
json_str = re.sub(r'\s*```$', '', json_str)
|
|
# Count open vs close braces
|
|
open_b = json_str.count('{')
|
|
close_b = json_str.count('}')
|
|
open_arr = json_str.count('[')
|
|
close_arr = json_str.count(']')
|
|
# Append missing closers
|
|
json_str += '}' * (open_b - close_b)
|
|
json_str += ']' * (open_arr - close_arr)
|
|
return json.loads(json_str)
|
|
except:
|
|
raise ValueError(f"AI returned invalid JSON (truncated at {max_tok} tokens). Please try again with a shorter description.")
|
|
|
|
|
|
# ============================================================
|
|
# CHAT (general AI interaction)
|
|
# ============================================================
|
|
|
|
SYSTEM_PROMPT = """You are an AI assistant for a CV/Candidate management system. You help users with:
|
|
1. Analyzing CVs in the database
|
|
2. Matching candidates to job requirements
|
|
3. Creating and modifying CV templates
|
|
4. Generating tailored CVs for specific positions
|
|
5. General questions about the candidate database
|
|
|
|
You have access to the system's database. When the user asks about candidates or requirements,
|
|
use the provided context to answer accurately.
|
|
|
|
Be concise and direct. When suggesting actions, format them clearly."""
|
|
|
|
def chat(user_message: str, conversation_history: list = None, context: str = "") -> str:
|
|
"""General AI chat with optional context."""
|
|
messages = [{"role": "system", "content": SYSTEM_PROMPT}]
|
|
|
|
if context:
|
|
messages.append({"role": "system", "content": f"Context:\n{context}"})
|
|
|
|
if conversation_history:
|
|
for msg in conversation_history[-10:]: # last 10 messages
|
|
messages.append({"role": msg["role"], "content": msg["content"]})
|
|
|
|
messages.append({"role": "user", "content": user_message})
|
|
|
|
return llm_chat(messages, temperature=0.5, max_tokens=2000) |