Automate Sales Intelligence 🚀

# Production Pattern with LangGraph & Queue (1000+ accounts/day)
from langgraph.graph import Graph, END
from typing import TypedDict, List, Optional
import openai
import asyncio
import aiohttp
from datetime import datetime
import redis
import json

class SalesIntelState(TypedDict):
    account_id: str
    company_name: str
    enriched_data: Optional[dict]
    buying_signals: Optional[dict]
    talking_points: Optional[dict]
    overall_score: int
    errors: List[str]
    retry_count: int
    timestamp: str

class SalesIntelPipeline:
    def __init__(self):
        self.redis_client = redis.Redis(
            host='localhost',
            port=6379,
            decode_responses=True
        )
        self.graph = self.build_graph()
    
    def build_graph(self):
        """Build LangGraph workflow"""
        graph = Graph()
        
        # Add nodes
        graph.add_node("enrich", self.enrich_node)
        graph.add_node("detect_signals", self.detect_signals_node)
        graph.add_node("generate_talking_points", self.generate_talking_points_node)
        graph.add_node("update_crm", self.update_crm_node)
        graph.add_node("handle_error", self.handle_error_node)
        
        # Add edges
        graph.set_entry_point("enrich")
        graph.add_conditional_edges(
            "enrich",
            self.check_enrichment_success,
            {
                "success": "detect_signals",
                "retry": "enrich",
                "fail": "handle_error"
            }
        )
        graph.add_conditional_edges(
            "detect_signals",
            self.check_signals_success,
            {
                "success": "generate_talking_points",
                "retry": "detect_signals",
                "fail": "handle_error"
            }
        )
        graph.add_edge("generate_talking_points", "update_crm")
        graph.add_edge("update_crm", END)
        graph.add_edge("handle_error", END)
        
        return graph.compile()
    
    async def enrich_node(self, state: SalesIntelState) -> SalesIntelState:
        """Enrich account data from multiple sources"""
        try:
            # Parallel API calls
            async with aiohttp.ClientSession() as session:
                tasks = [
                    self.get_zoominfo_data(session, state['company_name']),
                    self.get_news_data(session, state['company_name']),
                    self.get_linkedin_data(session, state['company_name']),
                    self.get_tech_stack_data(session, state['company_name'])
                ]
                results = await asyncio.gather(*tasks, return_exceptions=True)
            
            # Filter out errors
            valid_results = [r for r in results if not isinstance(r, Exception)]
            
            if not valid_results:
                state['errors'].append("All enrichment sources failed")
                return state
            
            # Combine and structure with LLM
            combined_data = {
                'zoominfo': results[0] if not isinstance(results[0], Exception) else None,
                'news': results[1] if not isinstance(results[1], Exception) else None,
                'linkedin': results[2] if not isinstance(results[2], Exception) else None,
                'tech_stack': results[3] if not isinstance(results[3], Exception) else None
            }
            
            enrichment_prompt = f"""Extract and structure company intelligence from these sources.

Company: {state['company_name']}

Data Sources:
{json.dumps(combined_data, indent=2)}

Output as JSON with: company_name, industry, employee_count, headquarters, tech_stack, key_contacts, recent_funding, expansion_plans, competitors, recent_news_summary"""

            response = openai.chat.completions.create(
                model="gpt-4",
                messages=[{"role": "user", "content": enrichment_prompt}],
                temperature=0.3,
                timeout=30
            )
            
            state['enriched_data'] = json.loads(response.choices[0].message.content)
            state['errors'] = []  # Clear errors on success
            
        except Exception as e:
            state['errors'].append(f"Enrichment failed: {str(e)}")
        
        return state
    
    async def detect_signals_node(self, state: SalesIntelState) -> SalesIntelState:
        """Detect buying signals from enriched data"""
        try:
            signal_prompt = f"""Analyze this company data for buying signals.

Company Data:
{json.dumps(state['enriched_data'], indent=2)}

Identify signals: funding_event, hiring_surge, tech_stack_gap, executive_change, market_expansion, competitor_switch.

For each: type, strength (high/medium/low), description, relevance, source, date.
Include: overall_score (0-100), recommendation.

Output as JSON."""

            response = openai.chat.completions.create(
                model="gpt-4",
                messages=[{"role": "user", "content": signal_prompt}],
                temperature=0.3,
                timeout=30
            )
            
            signals = json.loads(response.choices[0].message.content)
            state['buying_signals'] = signals
            state['overall_score'] = signals.get('overall_score', 0)
            state['errors'] = []
            
        except Exception as e:
            state['errors'].append(f"Signal detection failed: {str(e)}")
        
        return state
    
    async def generate_talking_points_node(self, state: SalesIntelState) -> SalesIntelState:
        """Generate personalized talking points"""
        try:
            talking_points_prompt = f"""Generate personalized sales talking points.

Company: {json.dumps(state['enriched_data'], indent=2)}
Signals: {json.dumps(state['buying_signals'], indent=2)}

Generate:
1. talking_points (4-5 with category, point, transition)
2. email_subject
3. email_opening
4. meeting_agenda (4 topics)

Be conversational and specific. Output as JSON."""

            response = openai.chat.completions.create(
                model="gpt-4",
                messages=[{"role": "user", "content": talking_points_prompt}],
                temperature=0.7,
                timeout=30
            )
            
            state['talking_points'] = json.loads(response.choices[0].message.content)
            state['errors'] = []
            
        except Exception as e:
            state['errors'].append(f"Talking points generation failed: {str(e)}")
        
        return state
    
    async def update_crm_node(self, state: SalesIntelState) -> SalesIntelState:
        """Update Salesforce with results"""
        try:
            # Store in Redis for CRM sync
            redis_key = f"sales_intel:{state['account_id']}"
            self.redis_client.setex(
                redis_key,
                86400,  # 24 hour expiry
                json.dumps({
                    'enriched_data': state['enriched_data'],
                    'buying_signals': state['buying_signals'],
                    'talking_points': state['talking_points'],
                    'overall_score': state['overall_score'],
                    'timestamp': state['timestamp']
                })
            )
            
            # Queue for Salesforce update (async worker will process)
            self.redis_client.lpush('salesforce_update_queue', json.dumps({
                'account_id': state['account_id'],
                'score': state['overall_score'],
                'signals': state['buying_signals']['signals'][:3],
                'talking_points': state['talking_points']
            }))
            
        except Exception as e:
            state['errors'].append(f"CRM update failed: {str(e)}")
        
        return state
    
    async def handle_error_node(self, state: SalesIntelState) -> SalesIntelState:
        """Handle errors and log"""
        print(f"Error processing {state['account_id']}: {state['errors']}")
        
        # Store error in Redis for monitoring
        self.redis_client.lpush('sales_intel_errors', json.dumps({
            'account_id': state['account_id'],
            'errors': state['errors'],
            'timestamp': state['timestamp']
        }))
        
        return state
    
    def check_enrichment_success(self, state: SalesIntelState) -> str:
        """Route based on enrichment success"""
        if state['enriched_data']:
            return "success"
        elif state['retry_count'] < 3:
            state['retry_count'] += 1
            return "retry"
        else:
            return "fail"
    
    def check_signals_success(self, state: SalesIntelState) -> str:
        """Route based on signal detection success"""
        if state['buying_signals']:
            return "success"
        elif state['retry_count'] < 3:
            state['retry_count'] += 1
            return "retry"
        else:
            return "fail"
    
    async def get_zoominfo_data(self, session: aiohttp.ClientSession, company_name: str):
        """Fetch ZoomInfo data"""
        async with session.get(
            'https://api.zoominfo.com/lookup/company',
            headers={'Authorization': f'Bearer {ZOOMINFO_API_KEY}'},
            params={'companyName': company_name},
            timeout=10
        ) as response:
            return await response.json()
    
    async def get_news_data(self, session: aiohttp.ClientSession, company_name: str):
        """Fetch news data"""
        async with session.get(
            'https://newsapi.org/v2/everything',
            params={
                'q': company_name,
                'apiKey': NEWS_API_KEY,
                'sortBy': 'publishedAt',
                'pageSize': 5
            },
            timeout=10
        ) as response:
            data = await response.json()
            return data.get('articles', [])
    
    async def get_linkedin_data(self, session: aiohttp.ClientSession, company_name: str):
        """Fetch LinkedIn data (requires Sales Navigator API)"""
        # Placeholder - implement with LinkedIn Sales Nav API
        return {'company': company_name, 'employees': []}
    
    async def get_tech_stack_data(self, session: aiohttp.ClientSession, company_name: str):
        """Fetch tech stack data"""
        # Placeholder - implement with BuiltWith or Clearbit
        return {'technologies': []}
    
    async def process_account(self, account_id: str, company_name: str) -> dict:
        """Process a single account through the pipeline"""
        initial_state: SalesIntelState = {
            'account_id': account_id,
            'company_name': company_name,
            'enriched_data': None,
            'buying_signals': None,
            'talking_points': None,
            'overall_score': 0,
            'errors': [],
            'retry_count': 0,
            'timestamp': datetime.now().isoformat()
        }
        
        result = await self.graph.ainvoke(initial_state)
        return result

# Usage
pipeline = SalesIntelPipeline()

# Process single account
result = await pipeline.process_account(
    account_id='001XXXXXXXXXXXXXXX',
    company_name='Acme Corp'
)

print(f"Score: {result['overall_score']}/100")
print(f"Errors: {len(result['errors'])}")

# Process batch (1000+ accounts)
accounts = [
    ('001XXX1', 'Company A'),
    ('001XXX2', 'Company B'),
    # ... 1000 more
]

tasks = [pipeline.process_account(aid, name) for aid, name in accounts]
results = await asyncio.gather(*tasks)

print(f"Processed {len(results)} accounts")
print(f"High priority: {len([r for r in results if r['overall_score'] > 70])}")