from dataclasses import dataclass from typing import List, Optional, Dict from datetime import datetime import numpy as np from termcolor import colored import json from encoder import create_encoders, FIELD_MAPPING, LegacyFieldEncoder from embeddings import EmbeddingManager, MatchResult, Skill @dataclass class Skill: skill_name: str @dataclass class JobPosting: # Essential matching fields (from API) title: str role_description: str company_description: str primary_skills: List[Skill] secondary_skills: List[Skill] # Additional API fields (with defaults) job_posting_id: str = "test_id" status: str = "active" location: str = "Test Location" workplace_model: str = "hybrid" job_engagement: str = "contract-to-hire" min_years_of_experience: int = 0 max_years_of_experience: int = 0 project_duration_from: datetime = datetime.now() project_duration_to: datetime = datetime.now() hourly_bill_rate_min: float = 50.0 hourly_bill_rate_max: float = 100.0 annual_salary_min: float = 100000.0 annual_salary_max: float = 150000.0 day_to_day_job_responsibilities: str = "" reason_for_hire: str = "" application_of_skills: str = "" company_id: str = "test_company" @dataclass class IndependentJobSeekerAssessmentRDS: # Essential matching fields (from API) primary_skills: List[str] secondary_skills: List[str] experiences: List[dict] educations: List[dict] certifications: List[dict] @dataclass class JobseekerInfoRDS: summary: str def transform_jobseeker_to_opensearch(jobseeker: IndependentJobSeekerAssessmentRDS, jobseeker_id: str) -> Dict: """Transform jobseeker data to OpenSearch format""" return { "jobseeker_id": jobseeker_id, "primary_skills": jobseeker.primary_skills, "secondary_skills": jobseeker.secondary_skills, "experiences": jobseeker.experiences, "educations": jobseeker.educations, "certifications": jobseeker.certifications, } def create_test_data(): """Create test data matching actual API fields""" print("\nCreating test data...") # Create test job posting with only relevant fields job_posting = JobPosting( title="Senior Software Engineer - ML/Cloud", role_description="Leading backend development team in cloud infrastructure projects. " "Focus on machine learning systems and scalable architectures. " "Responsible for ML pipeline optimization and team mentorship.", company_description="Tech company specializing in AI solutions", primary_skills=[ Skill("Python"), Skill("AWS"), Skill("Kubernetes"), Skill("TensorFlow"), Skill("PyTorch") ], secondary_skills=[ Skill("Docker"), Skill("CI/CD"), Skill("Agile"), Skill("Team Leadership") ] ) # Create matching job seeker (should show high similarity) matching_seeker = IndependentJobSeekerAssessmentRDS( primary_skills=[ "Python", "AWS", "Kubernetes", "TensorFlow", "PyTorch" ], secondary_skills=[ "Docker", "CI/CD", "Agile", "Team Leadership" ], experiences=[{ "title": "Senior Software Engineer", "company": "AI Tech Corp", "duration": "4 years", "description": "Led machine learning infrastructure team, developed scalable ML pipelines, " "optimized cloud resources, mentored junior engineers" }], educations=[{ "degree": "Master's", "field": "Computer Science", "institution": "Tech University" }], certifications=[{ "name": "AWS Solutions Architect Professional", "organization": "AWS", "start_date": "2023-01", "end_date": "2026-01" }] ) matching_info = JobseekerInfoRDS( summary="Senior ML engineer specialized in building scalable AI systems and leading engineering teams" ) # Create partial matching job seeker - more realistic partial match partial_matching_seeker = IndependentJobSeekerAssessmentRDS( primary_skills=[ "Python", "AWS", # Has the basic skills "Java", "TensorFlow" # Missing PyTorch, just has TensorFlow ], secondary_skills=[ "Docker", # Has some but not all secondary skills "Git", # Different version control instead of CI/CD "Scrum" # Basic Agile framework but no team leadership ], experiences=[{ "title": "Data Analyst", "company": "Tech Solutions Inc", "duration": "2 years", "description": "Worked on machine learning projects using TensorFlow. " "Maintained AWS infrastructure and helped with basic Kubernetes deployments. " "Member of an agile team developing ML-powered features." }], educations=[{ "degree": "Bachelor's", # Lower education level "field": "Computer Science", "institution": "Tech University" }], certifications=[{ "name": "AWS Cloud Practitioner", # Entry-level cert "organization": "AWS", "start_date": "2022-01", "end_date": "2025-01" }] ) partial_matching_info = JobseekerInfoRDS( summary="Data analyst working on graphical analysis and budget forecasting." ) # Create non-matching job seeker non_matching_seeker = IndependentJobSeekerAssessmentRDS( primary_skills=[ "Java", "Spring", "Oracle" ], secondary_skills=[ "Hibernate", "JSP", "Struts" ], experiences=[{ "title": "Java Developer", "company": "Enterprise Corp", "duration": "5 years", "description": "Built enterprise banking applications using Java stack, " "implemented transaction processing systems" }], educations=[{ "degree": "Bachelor's", "field": "Information Systems", "institution": "Business School" }], certifications=[{ "name": "Oracle Certified Professional", "organization": "Oracle", "start_date": "2022-01", "end_date": "2025-01" }] ) non_matching_info = JobseekerInfoRDS( summary="Experienced Java developer specialized in enterprise banking applications" ) return ( job_posting, matching_seeker, matching_info, partial_matching_seeker, partial_matching_info, non_matching_seeker, non_matching_info ) def analyze_match_result(match_result: MatchResult, candidate_type: str = "matching"): """Analyze and display match results""" print(f"\nAnalyzing match results for {candidate_type} candidate:") print("=" * 60) # Print field-by-field analysis print("\nField-by-Field Analysis:") print("-" * 40) # Define the order we want to display fields field_order = [ 'title_summary', 'primary_skills_primary_skills', 'secondary_skills_secondary_skills', 'role_description_experience', 'role_description_certifications' # Added certifications ] # Print scores in specified order for field_pair in field_order: if field_pair in match_result.field_scores: score = match_result.field_scores[field_pair] score_color = "green" if score > 0.85 else "yellow" if score > 0.7 else "red" print(f"{field_pair:35} | {colored(f'{score:.3f}', score_color)}") # Print overall similarity print("\nOverall Match Analysis:") print("-" * 40) score_color = "green" if match_result.similarity_score > 0.8 else \ "yellow" if match_result.similarity_score > 0.65 else "red" print(f"Match Score: {colored(f'{match_result.similarity_score:.3f}', score_color)}") # Print interpretation print("\nMatch Interpretation:") if match_result.similarity_score > 0.8: print(colored("Strong Match", "green"), "- Highly relevant candidate") print("Key Strengths:") print(match_result.explanation) elif match_result.similarity_score > 0.65: print(colored("Moderate Match", "yellow"), "- Potentially suitable candidate") print("Analysis:") print(match_result.explanation) else: print(colored("Weak Match", "red"), "- May not be suitable") print("Gaps:") print(match_result.explanation) return match_result.similarity_score def run_model_comparison_tests(manager: EmbeddingManager): """Run comprehensive comparison tests""" print("\nInitializing embedding manager...") # Get test data (job_posting, matching_seeker, matching_info, partial_matching_seeker, partial_matching_info, non_matching_seeker, non_matching_info) = create_test_data() print("\n" + "="*80) print("Testing with matching candidate (should show high similarity)") print("="*80) # Get embeddings and match result for matching candidate job_embeddings = manager.embed_jobposting(job_posting) matching_embeddings = manager.embed_jobseeker(matching_seeker, matching_info) matching_result = manager.calculate_similarity(job_embeddings, matching_embeddings) matching_similarity = analyze_match_result(matching_result, "matching") print("\n" + "="*80) print("Testing with partially matching candidate (should show moderate similarity)") print("="*80) # Get embeddings and match result for partial match partial_embeddings = manager.embed_jobseeker(partial_matching_seeker, partial_matching_info) partial_result = manager.calculate_similarity(job_embeddings, partial_embeddings) partial_similarity = analyze_match_result(partial_result, "partial matching") print("\n" + "="*80) print("Testing with non-matching candidate (should show low similarity)") print("="*80) # Get embeddings and match result for non-match non_matching_embeddings = manager.embed_jobseeker(non_matching_seeker, non_matching_info) non_matching_result = manager.calculate_similarity(job_embeddings, non_matching_embeddings) non_matching_similarity = analyze_match_result(non_matching_result, "non-matching") # Print comparative analysis print("\nComparative Analysis:") print("="*40) # Similarity differences print("\nSimilarity Differences:") match_vs_partial = matching_similarity - partial_similarity match_vs_non = matching_similarity - non_matching_similarity partial_vs_non = partial_similarity - non_matching_similarity print(f"Matching vs Partial: {colored(f'{match_vs_partial:>8.3f}', 'blue')}") print(f"Matching vs Non-Match: {colored(f'{match_vs_non:>8.3f}', 'blue')}") print(f"Partial vs Non-Match: {colored(f'{partial_vs_non:>8.3f}', 'blue')}") # Discrimination ratios print("\nDiscrimination Ratios:") match_partial_ratio = matching_similarity / max(partial_similarity, 0.001) match_non_ratio = matching_similarity / max(non_matching_similarity, 0.001) ratio_color = "green" if match_partial_ratio > 1.5 else "yellow" if match_partial_ratio > 1.2 else "red" print(f"Matching/Partial Ratio: {colored(f'{match_partial_ratio:>8.2f}x', ratio_color)}") ratio_color = "green" if match_non_ratio > 2.0 else "yellow" if match_non_ratio > 1.5 else "red" print(f"Matching/Non-Match Ratio:{colored(f'{match_non_ratio:>8.2f}x', ratio_color)}") # Quality assessment print("\nModel Quality Assessment:") print("-" * 40) discrimination_score = (match_vs_partial + match_vs_non) / 2 discrimination_color = "green" if discrimination_score > 0.3 else \ "yellow" if discrimination_score > 0.2 else "red" print(f"Discrimination Score: {colored(f'{discrimination_score:.3f}', discrimination_color)}") if discrimination_score > 0.3: print("Model shows good discrimination between candidate types") elif discrimination_score > 0.2: print("Model shows moderate discrimination - may need tuning") else: print("Model shows poor discrimination - consider adjusting weights or thresholds") def run_comparison_tests(job_encoder, seeker_encoder, legacy_encoder): """Run tests comparing new field-specific vs legacy approach""" print("\nRunning comparison tests between field-specific and legacy approaches...") # Get test data (job_posting, matching_seeker, matching_info, partial_matching_seeker, partial_matching_info, non_matching_seeker, non_matching_info) = create_test_data() # Test new field-specific approach print("\n" + "="*80) print("TESTING FIELD-SPECIFIC APPROACH") print("="*80) manager = EmbeddingManager(job_encoder, seeker_encoder) run_model_comparison_tests(manager) # Test legacy approach print("\n" + "="*80) print("TESTING LEGACY APPROACH") print("="*80) # Create legacy embeddings print("\nGenerating legacy embeddings...") job_emb = legacy_encoder.encode_jobposting(job_posting) match_emb = legacy_encoder.encode_jobseeker(matching_seeker, matching_info) partial_emb = legacy_encoder.encode_jobseeker(partial_matching_seeker, partial_matching_info) non_match_emb = legacy_encoder.encode_jobseeker(non_matching_seeker, non_matching_info) print("\nCalculating legacy similarities...") def calc_legacy_sim(emb1, emb2): """Calculate cosine similarity between two embeddings""" # Ensure embeddings are normalized emb1_norm = emb1 / (np.linalg.norm(emb1) + 1e-9) emb2_norm = emb2 / (np.linalg.norm(emb2) + 1e-9) # Calculate cosine similarity sim = np.dot(emb1_norm, emb2_norm) # Debug prints print(f"DEBUG: Embedding norms: {np.linalg.norm(emb1):.3f}, {np.linalg.norm(emb2):.3f}") print(f"DEBUG: Raw similarity: {sim:.3f}") return sim # Calculate similarities with extra debug info print("\nMatching candidate:") match_sim = (calc_legacy_sim(job_emb, match_emb) + 1) / 2 print("\nPartial matching candidate:") partial_sim = (calc_legacy_sim(job_emb, partial_emb) + 1) / 2 print("\nNon-matching candidate:") non_match_sim = (calc_legacy_sim(job_emb, non_match_emb) + 1) / 2 print(f"\nLegacy Approach Results:") print(f"Job embedding shape: {job_emb.shape}") print(f"Matching embedding shape: {match_emb.shape}") print(f"Matching candidate similarity: {match_sim:.3f}") print(f"Partial matching similarity: {partial_sim:.3f}") print(f"Non-matching similarity: {non_match_sim:.3f}") print("\nLegacy Discrimination Analysis:") print(f"Match vs Partial diff: {(match_sim - partial_sim):.3f}") print(f"Match vs Non-match diff: {(match_sim - non_match_sim):.3f}") print(f"Match/Non-match ratio: {(match_sim / non_match_sim):.2f}x") # Compare embedding statistics print("\nEmbedding Statistics:") print(f"Job embedding mean/std: {np.mean(job_emb):.3f}/{np.std(job_emb):.3f}") print(f"Match embedding mean/std: {np.mean(match_emb):.3f}/{np.std(match_emb):.3f}") print(f"Partial embedding mean/std: {np.mean(partial_emb):.3f}/{np.std(partial_emb):.3f}") print(f"Non-match embedding mean/std: {np.mean(non_match_emb):.3f}/{np.std(non_match_emb):.3f}") def main(): """Main test function with both approaches""" print("Creating encoders...") # Get both encoders for the field-specific approach field_encoder, seeker_encoder = create_encoders('all-mpnet-base-v2') # Create legacy encoder using local Qwen2 legacy_encoder = LegacyFieldEncoder("/Users/sebastian_a/jobposting-embedding") # Pass both encoders to comparison run_comparison_tests(field_encoder, seeker_encoder, legacy_encoder) if __name__ == "__main__": main()