Spaces:
Sleeping
Sleeping
Integrate DataEnrichmentManager into lead_scoring_agent - enrich profiles before scoring
5b9ce2d verified | """ | |
| lead_scoring_agent.py | |
| --- | |
| ML/transformer+rule agent that ingests full lead/profile asset and outputs: | |
| - intent_score: Float | |
| - readiness: Category (e.g., 'ready', 'researching', 'cold') | |
| - segment: Category (e.g., 'SMB', 'Enterprise', 'Freelancer') | |
| - rationale/feature_importance: Dict[str, float] (reasons/score contributions) | |
| Integrates with enrichment, research, outreach, analytics, validation agents. See README for pipeline, hooks, and agent call documentation. | |
| """ | |
| from transformers import pipeline | |
| import numpy as np | |
| from data_enrichment import DataEnrichmentManager | |
| class LeadScoringAgent: | |
| def __init__(self, username: str = 'default', model_name: str = 'distilbert-base-uncased'): self.classifier = pipeline('text-classification', model=model_name) | |
| self.classifier = pipeline('text-classification', model=model_name) | |
| self.username = username | |
| self.enrichment_manager = DataEnrichmentManager(username) | |
| def score_lead(self, profile: dict) -> dict: | |
| # ENRICHMENT STEP: Optionally enrich profile before scoring for better accuracy | |
| try: | |
| enriched_data = self.enrichment_manager.enrich_contact_full(profile) | |
| if enriched_data: | |
| # Merge enriched data into profile for richer scoring | |
| for key, value in enriched_data.items(): | |
| if key not in profile or not profile.get(key): | |
| profile[key] = value | |
| except Exception: | |
| pass # Continue with available data if enrichment fails | |
| text = self.profile_to_text(profile) | |
| model_result = self.classifier(text)[0] | |
| intent_score = float(model_result['score']) | |
| label = model_result['label'].lower() | |
| # Rule-based logic (customize as needed) | |
| readiness = 'ready' if intent_score > 0.8 and 'purchase' in label else 'researching' if intent_score > 0.5 else 'cold' | |
| segment = self.segment_from_profile(profile) | |
| feature_importance = self.estimate_feature_importance(profile) | |
| rationale = {k: v for k, v in feature_importance.items() if abs(v) > 0.1} | |
| return { | |
| 'intent_score': intent_score, | |
| 'readiness': readiness, | |
| 'segment': segment, | |
| 'rationale': rationale, | |
| 'feature_importance': feature_importance | |
| } | |
| def profile_to_text(self, profile: dict) -> str: | |
| # Join relevant fields into text | |
| fields = ['bio', 'recent_activity', 'interests', 'company', 'role'] | |
| return ' '.join([str(profile.get(f, '')) for f in fields]) | |
| def segment_from_profile(self, profile: dict) -> str: | |
| size = profile.get('company_size', '').lower() | |
| if 'enterprise' in size or profile.get('employees', 0) > 1000: | |
| return 'Enterprise' | |
| elif 'small' in size or profile.get('employees', 0) < 50: | |
| return 'SMB' | |
| return 'Freelancer' if 'freelancer' in profile.get('role', '').lower() else 'SMB' | |
| def estimate_feature_importance(self, profile: dict) -> dict: | |
| imp = {} | |
| imp['bio_len'] = len(profile.get('bio', '')) / 100.0 | |
| imp['activity'] = len(profile.get('recent_activity', '')) / 50.0 | |
| imp['interest_match'] = 1.0 if 'AI' in profile.get('interests', '') else 0.0 | |
| imp['company_rep'] = 1.0 if 'Google' in profile.get('company', '') else 0.2 | |
| return imp | |
| # Example usage: | |
| if __name__ == '__main__': | |
| agent = LeadScoringAgent() | |
| profile = { | |
| 'bio': 'Experienced AI practitioner looking to purchase cloud solutions.', | |
| 'recent_activity': 'Demo requested.', | |
| 'interests': 'AI, Machine Learning, Cloud', | |
| 'company': 'Google', | |
| 'role': 'Manager', | |
| 'company_size': 'enterprise', | |
| 'employees': 1200 | |
| } | |
| print(agent.score_lead(profile)) | |