import os import base64 import logging from typing import List, Dict, Any, Optional, Union from pathlib import Path import json from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage from langchain_anthropic import ChatAnthropic from langchain_groq import ChatGroq # Handle langchain_google_genai import error try: from langchain_google_genai import ChatGoogleGenerativeAI GEMINI_AVAILABLE = True except ImportError: # Gemini model is not available ChatGoogleGenerativeAI = None GEMINI_AVAILABLE = False from langchain_core.messages.base import BaseMessage # Fix import paths try: from app.utils.logging_utils import time_it, setup_logger from app.core.config import settings except ImportError: # Try relative imports for running from project root from behavior_backend.app.utils.logging_utils import time_it, setup_logger # Mock settings for testing class Settings: def __init__(self): self.OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY", "") self.ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY", "") self.GROQ_API_KEY = os.environ.get("GROQ_API_KEY", "") self.GOOGLE_API_KEY = os.environ.get("GOOGLE_API_KEY", "") settings = Settings() # Configure logging logger = setup_logger(__name__) class AIFaceAnalyzer: """Service for analyzing candidate profile pictures using Visual LLMs.""" def __init__(self, provider: str = "openai"): """ Initialize the AI face analyzer service. Args: provider: The LLM provider to use ('openai', 'anthropic', 'groq', or 'gemini') """ # If Gemini isn't available and that's the requested provider, fall back to OpenAI if provider.lower() == "gemini" and not GEMINI_AVAILABLE: logger.warning("Gemini provider requested but langchain_google_genai is not installed. Falling back to OpenAI.") provider = "openai" self.provider = provider.lower() self._init_model() def _init_model(self): """Initialize the LLM model based on the selected provider.""" if self.provider == "openai": api_key = os.environ.get("OPENAI_API_KEY") or getattr(settings, "OPENAI_API_KEY", "") if not api_key: raise ValueError("OPENAI_API_KEY not found in environment or settings") self.model = ChatOpenAI( model="gpt-4o-mini", max_tokens=4096, temperature=0.2, api_key=api_key ) elif self.provider == "anthropic": api_key = os.environ.get("ANTHROPIC_API_KEY") or getattr(settings, "ANTHROPIC_API_KEY", "") if not api_key: raise ValueError("ANTHROPIC_API_KEY not found in environment or settings") self.model = ChatAnthropic( model="claude-3-sonnet-20240229", temperature=0.2, max_tokens=4096, api_key=api_key ) elif self.provider == "groq": api_key = os.environ.get("GROQ_API_KEY") or getattr(settings, "GROQ_API_KEY", "") if not api_key: raise ValueError("GROQ_API_KEY not found in environment or settings") self.model = ChatGroq( model="qwen-2.5-32b", temperature=0.2, api_key=api_key ) logger.warning("Groq doesn't currently support vision models. Falling back to text-only analysis.") elif self.provider == "gemini": # This should not happen due to the check in __init__, but just in case if not GEMINI_AVAILABLE: logger.error("Gemini provider selected but langchain_google_genai is not installed!") logger.info("Falling back to OpenAI provider") self.provider = "openai" return self._init_model() api_key = os.environ.get("GOOGLE_API_KEY") or getattr(settings, "GOOGLE_API_KEY", "") if not api_key: raise ValueError("GOOGLE_API_KEY not found in environment or settings") self.model = ChatGoogleGenerativeAI( model="gemini-1.5-pro", temperature=0.2, max_tokens=4096, timeout=None, max_retries=2, api_key=api_key, #convert_system_message_to_human=True # Gemini requires converting system messages to human ) else: raise ValueError(f"Unsupported provider: {self.provider}. Use 'openai', 'anthropic', 'groq', or 'gemini'.") def _encode_image_to_base64(self, image_path: Union[str, Path]) -> str: """ Encode an image to base64. Args: image_path: Path to the image file Returns: Base64 encoded image """ image_path = Path(image_path) if not image_path.exists(): raise FileNotFoundError(f"Image file not found: {image_path}") with open(image_path, "rb") as image_file: return base64.b64encode(image_file.read()).decode("utf-8") def _prepare_messages(self, image_paths: List[Union[str, Path]], job_title: Optional[str] = None) -> List[BaseMessage]: """ Prepare messages for the LLM with images. Args: image_paths: List of paths to the images job_title: Optional job title for context Returns: List of messages for the LLM """ system_prompt = """You are an expert in professional appearance analysis for job interviews. Analyze the candidate's picture(s) randonly selected from a video (interview/self-introduction/etc) and provide an assessment of their professional appearance, focusing on: 1. Overall professional impression 2. Attire and dress code appropriateness 3. Facial expressions and perceived attitude 4. Background and setting appropriateness 5. Visual cues that might indicate personality traits relevant for professional settings 6. Areas of improvement for professional presentation Ouput: it must be a valid and structured JSON object. Provide your analysis in a structured JSON format with these keys: - professionalImpression: Overall analysis of how professional the candidate appears - attireAssessment: Analysis of clothing and accessories - facialExpressionAnalysis: Analysis of facial expressions, perceived emotions and attitude - backgroundAssessment: Analysis of the photo background and setting - personalityIndicators: Potential personality traits inferred from visual cues - recommendationsForImprovement: Specific recommendations for improving professional appearance - overallScore: A score from 1-10 on professional appearance suitability""" system_message = SystemMessage(content=system_prompt) # Create the content for the human message content = [] # Add text content text_content = "Please analyze this candidate's profile picture" if job_title: text_content += f" for a {job_title} position" text_content += "." content.append(text_content) # Add image content for image_path in image_paths: try: base64_image = self._encode_image_to_base64(image_path) if self.provider == "openai": content.append({ "type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{base64_image}"} }) elif self.provider == "anthropic": content.append({ "type": "image", "source": { "type": "base64", "media_type": "image/jpeg", "data": base64_image } }) elif self.provider == "gemini" and GEMINI_AVAILABLE: content.append({ "type": "image_url", "image_url": f"data:image/jpeg;base64,{base64_image}" }) except Exception as e: logger.error(f"Error encoding image {image_path}: {e}") human_message = HumanMessage(content=content) # For Gemini, we need to combine system message with human message if self.provider == "gemini" and GEMINI_AVAILABLE: return [system_message, human_message] else: return [system_message, human_message] @time_it def analyze_profile_pictures( self, image_paths: List[Union[str, Path]], job_title: Optional[str] = None ) -> Dict[str, Any]: """ Analyze candidate profile pictures using the configured LLM. Args: image_paths: List of paths to the profile pictures (1-3 images) job_title: Optional job title for context Returns: Dictionary with analysis results """ if not image_paths: logger.warning("No images provided for analysis") return self._generate_empty_analysis() # Limit to max 3 images if len(image_paths) > 3: logger.warning(f"Too many images provided ({len(image_paths)}). Using only the first 3.") image_paths = image_paths[:3] try: logger.info(f"Analyzing {len(image_paths)} profile pictures with {self.provider}") # Prepare messages with images messages = self._prepare_messages(image_paths, job_title) # Get response from LLM response = self.model.invoke(messages) # Extract and parse the response content response_content = response.content.replace("```json", "").replace("```", "") # Try to parse JSON from the response try: # First, try to extract JSON if it's wrapped in markdown code blocks if "```json" in response_content and "```" in response_content.split("```json", 1)[1]: json_str = response_content.split("```json", 1)[1].split("```", 1)[0].strip() result = json.loads(json_str) elif "```" in response_content and "```" in response_content.split("```", 1)[1]: json_str = response_content.split("```", 1)[1].split("```", 1)[0].strip() result = json.loads(json_str) else: # If not in code blocks, try parsing the whole response result = json.loads(response_content) except json.JSONDecodeError: logger.warning(f"Failed to parse JSON from response: {response_content}") # Create a formatted response manually result = { "professionalImpression": "Could not parse structured analysis", "rawResponse": response_content } # Add metadata result["provider"] = self.provider result["imageCount"] = len(image_paths) return result except Exception as e: logger.error(f"Error analyzing profile pictures: {e}") return self._generate_empty_analysis() def _generate_empty_analysis(self) -> Dict[str, Any]: """ Generate an empty analysis result when analysis fails. Returns: Empty analysis dictionary """ return { "professionalImpression": "No analysis available", "attireAssessment": "No analysis available", "facialExpressionAnalysis": "No analysis available", "backgroundAssessment": "No analysis available", "personalityIndicators": "No analysis available", "recommendationsForImprovement": "No analysis available", "overallScore": 0, "error": "Failed to analyze profile pictures", "provider": self.provider, "imageCount": 0 }