Spaces:
Sleeping
Sleeping
File size: 3,849 Bytes
9065d10 5b9ce2d 9065d10 5b9ce2d 9065d10 5b9ce2d 9065d10 5b9ce2d 9065d10 | 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 | """
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))
|