Spaces:
Configuration error
Configuration error
| #!/usr/bin/env python3 | |
| """AI Lead Scoring Agent for B2B prospect qualification""" | |
| import asyncio | |
| import json | |
| import os | |
| from datetime import datetime | |
| from typing import Dict, List, Optional | |
| # Hermes tools available | |
| try: | |
| from hermes_tools import web_search, terminal | |
| except ImportError: | |
| pass | |
| # Lead scoring weights | |
| WEIGHTS = { | |
| "job_seniority": 0.25, | |
| "company_size": 0.20, | |
| "digital_footprint": 0.15, | |
| "engagement_velocity": 0.15, | |
| "historical_data": 0.15, | |
| "industry_fit": 0.10 | |
| } | |
| def score_job_seniority(title: str) -> float: | |
| """Score job title 0-1""" | |
| high = ['CEO', 'CTO', 'CMO', 'VP', 'Director', 'Head'] | |
| mid = ['Manager', 'Lead', 'Specialist', 'Coordinator'] | |
| title_upper = title.upper() | |
| if any(h in title_upper for h in high): | |
| return 1.0 | |
| elif any(m in title_upper for m in mid): | |
| return 0.6 | |
| return 0.3 | |
| def score_company_size(revenue: Optional[str] = None, employees: Optional[int] = None) -> float: | |
| """Score company size 0-1""" | |
| if revenue: | |
| try: | |
| rev = float(revenue.replace('$', '').replace(',', '')) | |
| if rev >= 10000000: # $10M+ | |
| return 1.0 | |
| elif rev >= 1000000: # $1M-$10M | |
| return 0.8 | |
| elif rev >= 100000: # $100K-$1M | |
| return 0.6 | |
| except: | |
| pass | |
| if employees: | |
| if employees >= 500: | |
| return 1.0 | |
| elif employees >= 50: | |
| return 0.7 | |
| elif employees >= 10: | |
| return 0.4 | |
| return 0.2 | |
| def score_industry_fit(industry: str, target_industries: List[str]) -> float: | |
| """Score how well industry matches targets""" | |
| industry_lower = industry.lower() | |
| matches = sum(1 for t in target_industries if t.lower() in industry_lower) | |
| return min(1.0, matches * 0.3) | |
| def calculate_total_score(lead_data: Dict) -> int: | |
| """Calculate total lead score 0-100""" | |
| job_score = score_job_seniority(lead_data.get('job_title', '')) | |
| company_score = score_company_size( | |
| lead_data.get('revenue'), | |
| lead_data.get('employees') | |
| ) | |
| industry_score = score_industry_fit( | |
| lead_data.get('industry', ''), | |
| lead_data.get('target_industries', []) | |
| ) | |
| # Engagement signals | |
| engagement = lead_data.get('engagement', {}) | |
| engagement_score = min(1.0, ( | |
| engagement.get('website_visits', 0) * 0.2 + | |
| engagement.get('email_opens', 0) * 0.15 + | |
| engagement.get('content_downloads', 0) * 0.25 + | |
| engagement.get('social_engagement', 0) * 0.1 | |
| )) | |
| # Historical data | |
| historical = lead_data.get('historical_data', {}) | |
| historical_score = min(1.0, ( | |
| historical.get('previous_interest', 0) * 0.5 + | |
| historical.get('demo_requests', 0) * 0.8 + | |
| historical.get('last_purchase_days', 365) / 365 * -0.5 + 0.5 # Recent = good | |
| )) | |
| # Digital footprint | |
| tech_stack = lead_data.get('tech_stack', []) | |
| digital_score = min(1.0, len(tech_stack) * 0.1 + | |
| (0 if 'none' in tech_stack else 0.3)) | |
| # Weighted average | |
| total = ( | |
| job_score * WEIGHTS['job_seniority'] + | |
| company_score * WEIGHTS['company_size'] + | |
| digital_score * WEIGHTS['digital_footprint'] + | |
| engagement_score * WEIGHTS['engagement_velocity'] + | |
| historical_score * WEIGHTS['historical_data'] + | |
| industry_score * WEIGHTS['industry_fit'] | |
| ) * 100 | |
| return round(max(0, min(100, total))) | |
| def generate_prediction(score: int, industry: str) -> Dict: | |
| """Generate prediction metadata""" | |
| if score >= 80: | |
| action = "Immediate outreach - scheduling calls" | |
| timeline = "0-7 days" | |
| elif score >= 65: | |
| action = "Send personalized content package" | |
| timeline = "7-14 days" | |
| elif score >= 50: | |
| action = "Monitor for engagement signals" | |
| timeline = "14-30 days" | |
| else: | |
| action = "Nurture with educational content" | |
| timeline = "30+ days" | |
| return { | |
| "recommended_action": action, | |
| "timeline": timeline, | |
| "confidence": "high" if score > 70 else "medium" if score > 50 else "low" | |
| } | |
| async def process_lead(lead_data: Dict) -> Dict: | |
| """Process single lead and return score""" | |
| score = calculate_total_score(lead_data) | |
| prediction = generate_prediction(score, lead_data.get('industry', '')) | |
| return { | |
| "lead_id": lead_data.get('id'), | |
| "company": lead_data.get('company'), | |
| "job_title": lead_data.get('job_title'), | |
| "lead_score": score, | |
| "prediction": prediction, | |
| "processed_at": datetime.utcnow().isoformat() | |
| } | |
| async def main(): | |
| """Example usage""" | |
| sample_leads = [ | |
| { | |
| "id": 1, | |
| "company": "TechStart Inc", | |
| "job_title": "Marketing Director", | |
| "industry": "SaaS", | |
| "revenue": "$25000000", | |
| "target_industries": ["SaaS", "Technology", "Software"], | |
| "engagement": { | |
| "website_visits": 5, | |
| "email_opens": 3 | |
| } | |
| }, | |
| { | |
| "id": 2, | |
| "company": "Local Bakery", | |
| "job_title": "Owner", | |
| "industry": "Food & Beverage", | |
| "employees": 15, | |
| "engagement": { | |
| "website_visits": 1 | |
| } | |
| } | |
| ] | |
| for lead in sample_leads: | |
| result = await process_lead(lead) | |
| print(json.dumps(result, indent=2)) | |
| if __name__ == "__main__": | |
| asyncio.run(main()) |