import numpy as np # from sentence_transformers import SentenceTransformer import warnings import requests import os from dotenv import load_dotenv import logging # Load environment variables load_dotenv() # Fix NumPy 2.0 compatibility issue if not hasattr(np, "float_"): np.float_ = np.float64 # Suppress NumPy warnings warnings.filterwarnings("ignore", message=".*np.float_.*") # Setup logging logger = logging.getLogger(__name__) class Embedder: def __init__(self): """Initialize embedder with OpenAI API configuration""" self.openai_api_key = os.getenv("OPENAI_API_KEY") self.embedding_model = os.getenv("OPENAI_EMBEDDING_MODEL", "text-embedding-3-small") def generate_embedding(self, text: str): """Generate embeddings using OpenAI API""" try: if not self.openai_api_key: logger.error("OpenAI API key not configured") return None url = "https://api.openai.com/v1/embeddings" headers = { "Authorization": f"Bearer {self.openai_api_key}", "Content-Type": "application/json" } # Use text-embedding-3-small with 768 dimensions to match Qdrant collection # Default is 1536, but we need 768 to match existing SentenceTransformer embeddings payload = { "input": text, "model": "text-embedding-3-small", "dimensions": 768 # Match existing Qdrant collection dimensions } response = requests.post(url, json=payload, headers=headers) response.raise_for_status() result = response.json() embedding = result['data'][0]['embedding'] logger.debug(f"Generated embedding with {len(embedding)} dimensions") return embedding except Exception as e: logger.error(f"Error generating embedding with OpenAI: {str(e)}") if hasattr(e, 'response') and hasattr(e.response, 'text'): logger.error(f"OpenAI API response: {e.response.text}") return None def embed_with_qdrant(self, text: str): """Generate embeddings using OpenAI API (wrapper for compatibility)""" return self.generate_embedding(text) def store_in_qdrant( self, jobseeker_id: str, text: str, embedding: list = None, country_name: str = "", country_header_code: str = "", ): """Store resume data in Qdrant collection with country information""" try: if embedding is None: embedding = self.embed_with_qdrant(text) collection_name = os.getenv("QDRANT_COLLECTION_NAME") QDRANT_SEARCH_URL = os.getenv("QDRANT_SEARCH_URL") QDRANT_API_KEY = os.getenv("QDRANT_API_KEY") if not all([collection_name, QDRANT_SEARCH_URL, QDRANT_API_KEY]): logger.warning( "Qdrant configuration not complete in environment variables" ) return False url = f"{QDRANT_SEARCH_URL}/{collection_name}/points" headers = { "api-key": QDRANT_API_KEY, "Content-Type": "application/json", } # Convert string ID to integer for Qdrant compatibility try: point_id = int(jobseeker_id) except ValueError: # If jobseeker_id is not a valid integer, use hash point_id = hash(jobseeker_id) % (2**32) # Keep it as 32-bit integer point_payload = { "points": [ { "id": point_id, # Use integer ID "vector": embedding, "payload": { "resume_id": jobseeker_id, # Keep original ID in payload "text": text[:1000], # Limit text size "country_name": country_name, # Add country name "country_header_code": country_header_code, # Add country header code }, } ] } response = requests.put(url, json=point_payload, headers=headers) response.raise_for_status() logger.info(f"Successfully stored jobseeker {jobseeker_id} in Qdrant") return True except Exception as e: logger.error( f"Error storing in Qdrant for jobseeker {jobseeker_id}: {str(e)}" ) return False def store_in_qdrant_with_details( self, jobseeker_id: str, text: str, embedding: list = None, payload_data: dict = None, ): """Store resume data in Qdrant collection with detailed payload information""" try: if embedding is None: embedding = self.embed_with_qdrant(text) collection_name = os.getenv("QDRANT_COLLECTION_NAME") QDRANT_SEARCH_URL = os.getenv("QDRANT_SEARCH_URL") QDRANT_API_KEY = os.getenv("QDRANT_API_KEY") if not all([collection_name, QDRANT_SEARCH_URL, QDRANT_API_KEY]): logger.warning( "Qdrant configuration not complete in environment variables" ) return False url = f"{QDRANT_SEARCH_URL}/{collection_name}/points" headers = { "api-key": QDRANT_API_KEY, "Content-Type": "application/json", } # Convert string ID to integer for Qdrant compatibility try: point_id = int(jobseeker_id) except ValueError: # If jobseeker_id is not a valid integer, use hash point_id = hash(jobseeker_id) % (2**32) # Keep it as 32-bit integer # Create comprehensive payload payload = { "resume_id": jobseeker_id, # Keep original ID in payload "text": text[:1000], # Limit text size for performance } # Add detailed payload data if provided if payload_data: payload.update( { "country_name": payload_data.get("country_name", ""), "country_header_code": payload_data.get( "country_header_code", "" ), "email": payload_data.get("email", ""), "first_name": payload_data.get("first_name", ""), "last_name": payload_data.get("last_name", ""), "profile_summary": payload_data.get("profile_summary", "")[ :500 ], # Limit size "skills": payload_data.get("skills", [])[ :20 ], # Limit to 20 skills max "experience_years": payload_data.get("experience_years", 0), "highest_education": payload_data.get("highest_education", ""), "current_designation": payload_data.get( "current_designation", "" ), "languages": payload_data.get("languages", [])[ :10 ], # Limit to 10 languages max } ) point_payload = { "points": [ { "id": point_id, # Use integer ID "vector": embedding, "payload": payload, } ] } response = requests.put(url, json=point_payload, headers=headers) response.raise_for_status() logger.info( f"Successfully stored jobseeker {jobseeker_id} in Qdrant with detailed payload" ) return True except Exception as e: logger.error( f"Error storing detailed data in Qdrant for jobseeker {jobseeker_id}: {str(e)}" ) return False def search_with_qdrant(self, query: str, top_k: int = 20): """Search resumes in Qdrant using vector similarity""" try: QDRANT_SEARCH_URL = os.getenv("QDRANT_SEARCH_URL") QDRANT_API_KEY = os.getenv("QDRANT_API_KEY") collection_name = os.getenv("QDRANT_COLLECTION_NAME") if not all([QDRANT_SEARCH_URL, QDRANT_API_KEY, collection_name]): logger.warning( "Qdrant configuration not complete in environment variables" ) return {"result": []} # Generate embedding for query embedding = self.embed_with_qdrant(query) url = f"{QDRANT_SEARCH_URL}/{collection_name}/points/search" headers = { "api-key": QDRANT_API_KEY, "Content-Type": "application/json", } payload = {"vector": embedding, "top": top_k, "with_payload": True} response = requests.post(url, json=payload, headers=headers) response.raise_for_status() return response.json() except Exception as e: logger.error(f"Error searching Qdrant: {str(e)}") return {"result": []} def search_filtered_candidates( self, query: str, filtered_resume_ids: list, top_k: int = 20, country_code: str = None ): """ Search specific candidates in Qdrant using vector similarity Uses client-side filtering for better performance with large ID lists Optionally filters by country code """ try: QDRANT_SEARCH_URL = os.getenv("QDRANT_SEARCH_URL") QDRANT_API_KEY = os.getenv("QDRANT_API_KEY") collection_name = os.getenv("QDRANT_COLLECTION_NAME") if not all([QDRANT_SEARCH_URL, QDRANT_API_KEY, collection_name]): logger.warning( "Qdrant configuration not complete in environment variables" ) return {"result": []} if not filtered_resume_ids: logger.info("No filtered resume IDs provided for vector search") return {"result": []} # Generate embedding for query embedding = self.embed_with_qdrant(query) if not embedding: logger.error("Failed to generate embedding for query") return {"result": []} url = f"{QDRANT_SEARCH_URL}/{collection_name}/points/search" headers = { "api-key": QDRANT_API_KEY, "Content-Type": "application/json", } # NO LIMIT - Process all available candidates without restrictions # Get ALL candidates from the pool for comprehensive matching num_candidates = len(filtered_resume_ids) # Remove all limits - search ALL candidates search_limit = num_candidates # Process every single candidate logger.info(f"Vector search: searching ALL {num_candidates} candidates (no limit)") payload = { "vector": embedding, "limit": search_limit, # Use 'limit' instead of 'top' (Qdrant API standard) "with_payload": True } # Add country filter if provided if country_code: payload["filter"] = { "must": [ { "key": "country_header_code", "match": {"value": country_code} } ] } logger.info(f"Vector search with country filter: {country_code}") response = requests.post(url, json=payload, headers=headers) # Log error details if request fails if response.status_code != 200: logger.error(f"Qdrant API error: {response.status_code} - {response.text}") response.raise_for_status() result = response.json() # Client-side filtering: keep only results whose resume_id is in filtered_resume_ids filtered_resume_ids_set = set(str(rid) for rid in filtered_resume_ids) # Convert to set for O(1) lookup filtered_results = [] for item in result.get('result', []): resume_id = item.get('payload', {}).get('resume_id') if resume_id and str(resume_id) in filtered_resume_ids_set: filtered_results.append(item) # Stop if we have enough results if len(filtered_results) >= top_k: break logger.info( f"Vector search: retrieved {len(result.get('result', []))} total, " f"filtered to {len(filtered_results)} from {len(filtered_resume_ids)} target IDs" ) return {"result": filtered_results} except Exception as e: logger.error(f"Error searching filtered candidates in Qdrant: {str(e)}") return {"result": []} def search_with_filters(self, query: str, filters: dict = None, top_k: int = 20): """Search resumes in Qdrant with advanced filtering by skills, country, etc.""" try: QDRANT_SEARCH_URL = os.getenv("QDRANT_SEARCH_URL") QDRANT_API_KEY = os.getenv("QDRANT_API_KEY") collection_name = os.getenv("QDRANT_COLLECTION_NAME") if not all([QDRANT_SEARCH_URL, QDRANT_API_KEY, collection_name]): logger.warning( "Qdrant configuration not complete in environment variables" ) return {"result": []} # Generate embedding for query embedding = self.embed_with_qdrant(query) url = f"{QDRANT_SEARCH_URL}/{collection_name}/points/search" headers = { "api-key": QDRANT_API_KEY, "Content-Type": "application/json", } payload = {"vector": embedding, "top": top_k, "with_payload": True} # Build filters if provided if filters: filter_conditions = [] # Country filter if filters.get("country_header_code"): filter_conditions.append( { "key": "country_header_code", "match": {"value": filters["country_header_code"]}, } ) # Skills filter (any of the specified skills) if filters.get("skills") and isinstance(filters["skills"], list): filter_conditions.append( {"key": "skills", "match": {"any": filters["skills"]}} ) # Education filter if filters.get("education"): filter_conditions.append( { "key": "highest_education", "match": {"value": filters["education"]}, } ) # Experience minimum years filter if filters.get("min_experience_years"): filter_conditions.append( { "key": "experience_years", "range": {"gte": filters["min_experience_years"]}, } ) if filter_conditions: payload["filter"] = {"must": filter_conditions} response = requests.post(url, json=payload, headers=headers) response.raise_for_status() result = response.json() logger.info( f"Qdrant search with filters returned {len(result.get('result', []))} results" ) return result except Exception as e: logger.error(f"Error searching with filters in Qdrant: {str(e)}") return {"result": []}