from __future__ import annotations from pydantic import model_validator, Field # type: ignore from typing import ( Dict, Literal, ClassVar, Any, AsyncGenerator, List, Optional # override ) # NOTE: You must import Optional for late binding to work import inspect import re import json import asyncio import time import asyncio import numpy as np # type: ignore from .base import Agent, agent_registry, LLMMessage from models import ApiActivityType, ApiMessage, Candidate, ChatMessage, ChatMessageError, ChatMessageMetaData, ApiMessageType, ChatMessageStatus, ChatMessageStreaming, ChatMessageUser, ChatOptions, ChatSenderType, ApiStatusType, Job, JobRequirements, JobRequirementsMessage, Tunables import model_cast from logger import logger import defines import backstory_traceback as traceback class JobRequirementsAgent(Agent): agent_type: Literal["job_requirements"] = "job_requirements" # type: ignore _agent_type: ClassVar[str] = agent_type # Add this for registration # Stage 1A: Job Analysis Implementation def create_job_analysis_prompt(self, job_description: str) -> tuple[str, str]: """Create the prompt for job requirements analysis.""" logger.info(f"{self.agent_type} - {inspect.stack()[0].function}") system_prompt = """ You are an objective job requirements analyzer. Your task is to extract and categorize the specific skills, experiences, and qualifications required in a job description WITHOUT any reference to any candidate. ## INSTRUCTIONS: 1. Analyze ONLY the job description provided. 2. Extract company information, job title, and all requirements. 3. If a requirement is compound (e.g., "5+ years experience with React, Node.js and MongoDB" or "FastAPI/Django/React"), break it down into individual components. 4. Categorize requirements into: - Technical skills (required and preferred) - Experience requirements (required and preferred) - Education requirements - Soft skills - Industry knowledge - Responsibilities - Company values 5. Extract and categorize all requirements and preferences. 6. DO NOT consider any candidate information - this is a pure job analysis task. 7. Provide the output in a structured JSON format as specified below. ## OUTPUT FORMAT: ```json { "company_name": "Company Name", "job_title": "Job Title", "job_summary": "Brief summary of the job", "job_requirements": { "technical_skills": { "required": ["skill1", "skill2"], "preferred": ["skill1", "skill2"] }, "experience_requirements": { "required": ["exp1", "exp2"], "preferred": ["exp1", "exp2"] }, "soft_skills": ["skill1", "skill2"], "experience": ["exp1", "exp2"], "education": ["req1", "req2"], "certifications": ["knowledge1", "knowledge2"], "preferred_attributes": ["resp1", "resp2"], "company_values": ["value1", "value2"] } } ``` Be specific and detailed in your extraction. If a requirement can be broken down into several separate requirements, split them. For example, the technical_skill of "Python/Django/FastAPI" should be separated into different requirements: Python, Django, and FastAPI. For example, if the job description mentions: "Python/Django/FastAPI", you should extract it as: "technical_skills": { "required": [ "Python", "Django", "FastAPI" ] }, Avoid vague categorizations and be precise about whether skills are explicitly required or just preferred. """ prompt = f"Job Description:\n{job_description}" return system_prompt, prompt async def analyze_job_requirements( self, llm: Any, model: str, session_id: str, prompt: str ) -> AsyncGenerator[ChatMessageStreaming | ChatMessage | ChatMessageError | ChatMessageStatus, None]: """Analyze job requirements from job description.""" system_prompt, prompt = self.create_job_analysis_prompt(prompt) status_message = ChatMessageStatus( session_id=session_id, content="Analyzing job requirements", activity=ApiActivityType.THINKING ) yield status_message logger.info(f"🔍 {status_message.content}") generated_message = None async for generated_message in self.llm_one_shot(llm, model, session_id=session_id, prompt=prompt, system_prompt=system_prompt): if generated_message.status == ApiStatusType.ERROR: yield generated_message return if generated_message.status != ApiStatusType.DONE: yield generated_message if not generated_message: error_message = ChatMessageError( session_id=session_id, content="Job requirements analysis failed to generate a response.") logger.error(f"⚠️ {error_message.content}") yield error_message return yield generated_message return def gather_requirements(self, reqs: JobRequirements) -> Dict[str, Any]: # technical_skills: Requirements = Field(..., alias="technicalSkills") # experience_requirements: Requirements = Field(..., alias="experienceRequirements") # soft_skills: Optional[List[str]] = Field(default_factory=list, alias="softSkills") # experience: Optional[List[str]] = [] # education: Optional[List[str]] = [] # certifications: Optional[List[str]] = [] # preferred_attributes: Optional[List[str]] = Field(None, alias="preferredAttributes") # company_values: Optional[List[str]] = Field(None, alias="companyValues") """Gather and format job requirements for display.""" display = { "technical_skills": { "required": reqs.technical_skills.required, "preferred": reqs.technical_skills.preferred }, "experience_requirements": { "required": reqs.experience_requirements.required, "preferred": reqs.experience_requirements.preferred }, "soft_skills": reqs.soft_skills, "experience": reqs.experience, "education": reqs.education, "certifications": reqs.certifications, "preferred_attributes": reqs.preferred_attributes, "company_values": reqs.company_values } return display async def generate( self, llm: Any, model: str, session_id: str, prompt: str, tunables: Optional[Tunables] = None, temperature=0.7 ) -> AsyncGenerator[ApiMessage, None]: if not self.user: error_message = ChatMessageError( session_id=session_id, content="User is not set for this agent." ) logger.error(f"⚠️ {error_message.content}") yield error_message return # Stage 1A: Analyze job requirements status_message = ChatMessageStatus( session_id=session_id, content = f"Analyzing job requirements", activity=ApiActivityType.THINKING ) yield status_message generated_message = None async for generated_message in self.analyze_job_requirements(llm, model, session_id, prompt): if generated_message.status == ApiStatusType.ERROR: yield generated_message return if generated_message.status != ApiStatusType.DONE: yield generated_message if not generated_message: error_message = ChatMessageError( session_id=session_id, content="Job requirements analysis failed to generate a response." ) logger.error(f"⚠️ {error_message.content}") yield error_message return requirements = None job_requirements_data = "" company = "" summary = "" title = "" try: json_str = self.extract_json_from_text(generated_message.content) requirements_json = json.loads(json_str) company = requirements_json.get("company_name", "") title = requirements_json.get("job_title", "") summary = requirements_json.get("job_summary", "") job_requirements_data = requirements_json.get("job_requirements", None) requirements = JobRequirements.model_validate(job_requirements_data) except json.JSONDecodeError as e: status_message.status = ApiStatusType.ERROR status_message.content = f"Failed to parse job requirements JSON: {str(e)}\n\n{job_requirements_data}" logger.error(f"⚠️ {status_message.content}") yield status_message return except ValueError as e: status_message.status = ApiStatusType.ERROR status_message.content = f"Job requirements validation error: {str(e)}\n\n{job_requirements_data}" logger.error(f"⚠️ {status_message.content}") logger.error(f"Content: {prompt}") yield status_message return except Exception as e: status_message.status = ApiStatusType.ERROR status_message.content = f"Unexpected error processing job requirements: {str(e)}\n\n{job_requirements_data}" logger.error(traceback.format_exc()) logger.error(f"⚠️ {status_message.content}") yield status_message return # Gather and format requirements for display display = self.gather_requirements(requirements) logger.info(f"📋 Job requirements extracted: {json.dumps(display, indent=2)}") job = Job( owner_id=self.user.id, owner_type=self.user.user_type, company=company, title=title, summary=summary, requirements=requirements, description=prompt, ) job_requirements_message = JobRequirementsMessage( session_id=session_id, status=ApiStatusType.DONE, job=job, ) yield job_requirements_message logger.info(f"✅ Job requirements analysis completed successfully.") return # Register the base agent agent_registry.register(JobRequirementsAgent._agent_type, JobRequirementsAgent)