File size: 8,841 Bytes
49b13c6 518f864 49b13c6 518f864 49b13c6 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 |
import asyncio
import json
from pathlib import Path
from typing import Any
from pydantic import BaseModel
from vsp.app.main import VspDataEnrichment
from vsp.app.model.linkedin.linkedin_models import LinkedinProfile
from vsp.app.model.vsp.vsp_models import VSPProfile
from vsp.shared import logger_factory
logger = logger_factory.get_logger(__name__)
class ComparisonResult(BaseModel):
profile_name: str
correct_enums: int
total_enums: int
accuracy: float
additional_enums_in_actual: int
comparisons: dict[str, Any]
def load_profiles() -> dict[str, tuple[LinkedinProfile, VSPProfile]]:
sample_profiles_dir = Path("tests/test_data/sample_profiles")
classified_profiles_dir = Path("tests/test_data/sample_profiles_classified")
profiles = {}
for profile_file in sample_profiles_dir.glob("*.json"):
name = profile_file.stem
with profile_file.open() as f:
linkedin_data = json.load(f)
linkedin_profile = LinkedinProfile.model_validate(linkedin_data)
classified_file = classified_profiles_dir / f"{name}.json"
if classified_file.exists():
with classified_file.open() as f:
classified_data = json.load(f)
classified_profile = VSPProfile.model_validate(classified_data)
profiles[name] = (linkedin_profile, classified_profile)
return profiles
async def compare_profiles(linkedin_profile: LinkedinProfile, classified_profile: VSPProfile) -> ComparisonResult:
vsp_enrichment = VspDataEnrichment()
result = await vsp_enrichment.process_linkedin_profile(linkedin_profile)
comparisons = {}
correct_enums = 0
total_enums = 0
additional_enums_in_actual = 0
# Compare educations
for classified_edu, result_edu in zip(classified_profile.education.education_history, result.classified_educations):
comparisons[f"Education: {classified_edu.school}"] = {
"expected": classified_edu.degree_characterization,
"actual": result_edu.classification.output.value,
"confidence": result_edu.classification.confidence,
"reasoning": result_edu.classification.reasoning,
}
if classified_edu.degree_characterization is not None:
total_enums += 1
if classified_edu.degree_characterization == result_edu.classification.output.value:
correct_enums += 1
elif result_edu.classification.output.value is not None:
additional_enums_in_actual += 1
# Compare work experiences
for classified_exp, result_exp in zip(
classified_profile.professional_experience.experience_history, result.classified_work_experiences
):
exp_key = f"Job: {classified_exp.title} at {classified_exp.company}"
comparisons[exp_key] = {
"primary_job_type": {
"expected": classified_exp.primary_job_type,
"actual": result_exp.work_experience_classification.primary_job_type.value,
"confidence": result_exp.work_experience_classification.confidence,
"reasoning": result_exp.work_experience_classification.reasoning,
},
"secondary_job_type": {
"expected": classified_exp.secondary_job_type,
"actual": result_exp.work_experience_classification.secondary_job_type.value,
"confidence": result_exp.work_experience_classification.confidence,
"reasoning": result_exp.work_experience_classification.reasoning,
},
}
if classified_exp.primary_job_type is not None:
total_enums += 1
if classified_exp.primary_job_type == result_exp.work_experience_classification.primary_job_type.value:
correct_enums += 1
elif result_exp.work_experience_classification.primary_job_type.value is not None:
additional_enums_in_actual += 1
if classified_exp.secondary_job_type is not None:
total_enums += 1
if classified_exp.secondary_job_type == result_exp.work_experience_classification.secondary_job_type.value:
correct_enums += 1
elif result_exp.work_experience_classification.secondary_job_type.value is not None:
additional_enums_in_actual += 1
# Add comparisons for investment banking, investing focus, etc. if available
if result_exp.investment_banking_classification:
comparisons[exp_key]["investment_banking_group"] = {
"expected": classified_exp.investment_banking_focus,
"actual": result_exp.investment_banking_classification.investment_banking_group.value,
"confidence": result_exp.investment_banking_classification.confidence,
"reasoning": result_exp.investment_banking_classification.reasoning,
}
if classified_exp.investment_banking_focus is not None:
total_enums += 1
if (
classified_exp.investment_banking_focus
== result_exp.investment_banking_classification.investment_banking_group.value
):
correct_enums += 1
elif result_exp.investment_banking_classification.investment_banking_group.value is not None:
additional_enums_in_actual += 1
if result_exp.investing_focus_asset_class_classification:
comparisons[exp_key]["investing_focus_asset_class"] = {
"expected": classified_exp.investing_focus_stage,
"actual": result_exp.investing_focus_asset_class_classification.investing_focus_asset_class.value,
"confidence": result_exp.investing_focus_asset_class_classification.confidence,
"reasoning": result_exp.investing_focus_asset_class_classification.reasoning,
}
if classified_exp.investing_focus_stage is not None:
total_enums += 1
if (
classified_exp.investing_focus_stage
== result_exp.investing_focus_asset_class_classification.investing_focus_asset_class.value
):
correct_enums += 1
elif result_exp.investing_focus_asset_class_classification.investing_focus_asset_class.value is not None:
additional_enums_in_actual += 1
if result_exp.investing_focus_sector_classification:
comparisons[exp_key]["investing_focus_sector"] = {
"expected": classified_exp.investing_focus_sector,
"actual": result_exp.investing_focus_sector_classification.investing_focus_sector.value,
"confidence": result_exp.investing_focus_sector_classification.confidence,
"reasoning": result_exp.investing_focus_sector_classification.reasoning,
}
if classified_exp.investing_focus_sector is not None:
total_enums += 1
if (
classified_exp.investing_focus_sector
== result_exp.investing_focus_sector_classification.investing_focus_sector.value
):
correct_enums += 1
elif result_exp.investing_focus_sector_classification.investing_focus_sector.value is not None:
additional_enums_in_actual += 1
accuracy = correct_enums / total_enums if total_enums > 0 else 0
return ComparisonResult(
profile_name=f"{linkedin_profile.first_name} {linkedin_profile.last_name}",
correct_enums=correct_enums,
total_enums=total_enums,
accuracy=accuracy,
additional_enums_in_actual=additional_enums_in_actual,
comparisons=comparisons,
)
async def run_tests() -> None:
profiles = load_profiles()
results = []
for _, (linkedin_profile, classified_profile) in profiles.items():
result = await compare_profiles(linkedin_profile, classified_profile)
results.append(result)
logger.info(
f"Processed {result.profile_name}: Accuracy {result.accuracy:.2%}, "
f"Additional enums in actual: {result.additional_enums_in_actual}"
)
overall_accuracy = sum(r.accuracy for r in results) / len(results)
total_additional_enums = sum(r.additional_enums_in_actual for r in results)
logger.info(f"Overall accuracy: {overall_accuracy:.2%}")
logger.info(f"Total additional enums in actual: {total_additional_enums}")
# Save detailed results to a JSON file
with open("enum_classifier_results.json", "w") as f:
json.dump([r.model_dump() for r in results], f, indent=2)
logger.info("Detailed results saved to enum_classifier_results.json")
if __name__ == "__main__":
asyncio.run(run_tests())
|