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 traceback import asyncio import time import asyncio import numpy as np # type: ignore from .base import Agent, agent_registry, LLMMessage from models import Candidate, ChatMessage, ChatMessageBase, ChatMessageMetaData, ChatMessageType, ChatMessageUser, ChatOptions, ChatSenderType, ChatStatusType, JobRequirements import model_cast from logger import logger import defines 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. Extract and categorize all requirements and preferences. 4. DO NOT consider any candidate information - this is a pure job analysis task. ## 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"] }, "education_requirements": ["req1", "req2"], "soft_skills": ["skill1", "skill2"], "industry_knowledge": ["knowledge1", "knowledge2"], "responsibilities": ["resp1", "resp2"], "company_values": ["value1", "value2"] } } ``` Be specific and detailed in your extraction. Break down compound requirements into individual components. For example, "5+ years experience with React, Node.js and MongoDB" should be separated into: - Experience: "5+ years software development" - Technical skills: "React", "Node.js", "MongoDB" 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, user_message: ChatMessage ) -> AsyncGenerator[ChatMessage, None]: """Analyze job requirements from job description.""" system_prompt, prompt = self.create_job_analysis_prompt(user_message.content) analyze_message = user_message.model_copy() analyze_message.content = prompt generated_message = None async for generated_message in self.llm_one_shot(llm, model, system_prompt=system_prompt, user_message=analyze_message): if generated_message.status == ChatStatusType.ERROR: generated_message.content = "Error analyzing job requirements." yield generated_message return if not generated_message: status_message = ChatMessage( session_id=user_message.session_id, sender=ChatSenderType.AGENT, status = ChatStatusType.ERROR, type = ChatMessageType.ERROR, content = "Job requirements analysis failed to generate a response.") yield status_message return generated_message.status = ChatStatusType.DONE generated_message.type = ChatMessageType.RESPONSE yield generated_message return async def generate( self, llm: Any, model: str, user_message: ChatMessageUser, user: Candidate | None, temperature=0.7 ) -> AsyncGenerator[ChatMessage, None]: # Stage 1A: Analyze job requirements status_message = ChatMessage( session_id=user_message.session_id, sender=ChatSenderType.AGENT, status=ChatStatusType.STATUS, type=ChatMessageType.THINKING, content = f"Analyzing job requirements") yield status_message generated_message = None async for generated_message in self.analyze_job_requirements(llm, model, user_message): if generated_message.status == ChatStatusType.ERROR: status_message.status = ChatStatusType.ERROR status_message.content = generated_message.content yield status_message return if not generated_message: status_message.status = ChatStatusType.ERROR status_message.content = "Job requirements analysis failed." yield status_message return json_str = self.extract_json_from_text(generated_message.content) job_requirements : JobRequirements | None = None job_requirements_data = "" company_name = "" job_summary = "" job_title = "" try: job_requirements_data = json.loads(json_str) job_requirements_data = job_requirements_data.get("job_requirements", None) job_title = job_requirements_data.get("job_title", "") company_name = job_requirements_data.get("company_name", "") job_summary = job_requirements_data.get("job_summary", "") job_requirements = JobRequirements.model_validate(job_requirements_data) if not job_requirements: raise ValueError("Job requirements data is empty or invalid.") except json.JSONDecodeError as e: status_message.status = ChatStatusType.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 = ChatStatusType.ERROR status_message.content = f"Job requirements validation error: {str(e)}\n\n{job_requirements_data}" logger.error(f"⚠️ {status_message.content}") yield status_message return except Exception as e: status_message.status = ChatStatusType.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 status_message.status = ChatStatusType.DONE status_message.type = ChatMessageType.RESPONSE job_data = { "company": company_name, "title": job_title, "summary": job_summary, "requirements": job_requirements.model_dump(mode="json", exclude_unset=True) } status_message.content = json.dumps(job_data) yield status_message logger.info(f"✅ Job requirements analysis completed successfully.") return # Register the base agent agent_registry.register(JobRequirementsAgent._agent_type, JobRequirementsAgent)