Create intelligence_extractor.py

src/ai/services/intelligence_extractor.py

@@ -0,0 +1,171 @@
+"""
+src/ai/services/intelligence_extractor.py
+Comprehensive business intelligence extraction from call transcripts
+"""
+import logging
+from typing import Dict, Any, List
+from datetime import datetime
+
+logger = logging.getLogger(__name__)
+
+class IntelligenceExtractor:
+    """Extract comprehensive business intelligence from interaction transcripts"""
+    
+    def __init__(self, llm_service):
+        self.llm_service = llm_service
+        self.setup_extraction_schema()
+    
+    def setup_extraction_schema(self):
+        """Setup the comprehensive extraction schema"""
+        self.extraction_schema = {
+            "business_relationships": {
+                "stakeholders": {
+                    "mentioned_contacts": List[Dict],  # Names, titles, influence levels
+                    "new_contacts": List[Dict],        # Previously unknown contacts
+                    "relationship_changes": List[Dict], # Changes in relationships/roles
+                    "decision_makers": List[Dict],     # Identified decision makers
+                    "influence_network": Dict          # Relationship mapping
+                },
+                "team_intelligence": {
+                    "internal_stakeholders": List[Dict], # Our team members mentioned
+                    "role_changes": List[Dict],         # Team role updates
+                    "expertise_needed": List[str]       # Required expertise identified
+                }
+            },
+            "opportunities": {
+                "new_opportunities": List[Dict],      # New business opportunities
+                "existing_updates": List[Dict],       # Updates to existing opportunities
+                "cross_sell": List[Dict],            # Cross-sell opportunities
+                "upsell": List[Dict],                # Upsell opportunities
+                "risk_factors": List[Dict],          # Identified risks
+                "competition_intel": List[Dict]       # Competitive intelligence
+            },
+            "project_intelligence": {
+                "active_projects": List[Dict],        # Current project updates
+                "project_health": List[Dict],         # Project status/health
+                "resource_needs": List[Dict],         # Resource requirements
+                "timeline_updates": List[Dict],       # Schedule changes
+                "success_metrics": List[Dict]         # Performance indicators
+            },
+            "client_intelligence": {
+                "pain_points": List[Dict],           # Client challenges
+                "strategic_initiatives": List[Dict],  # Client's strategic plans
+                "budget_cycles": Dict,                # Budget timing/constraints
+                "technology_stack": List[str],        # Tech infrastructure
+                "organizational_changes": List[Dict]  # Org changes at client
+            },
+            "market_intelligence": {
+                "industry_trends": List[Dict],        # Market trends mentioned
+                "competitor_mentions": List[Dict],    # Competitor information
+                "regulatory_updates": List[Dict],     # Compliance/regulatory info
+                "market_challenges": List[Dict]       # Industry challenges
+            },
+            "follow_up_actions": {
+                "meetings": List[Dict],               # Scheduled/requested meetings
+                "action_items": List[Dict],           # Specific tasks
+                "deliverables": List[Dict],           # Expected deliverables
+                "proposals_needed": List[Dict],       # Required proposals
+                "approvals_required": List[Dict]      # Needed approvals
+            },
+            "sentiment_analysis": {
+                "overall_sentiment": float,           # Call sentiment score
+                "topic_sentiment": Dict,              # Sentiment by topic
+                "risk_signals": List[Dict],           # Potential issues
+                "opportunity_signals": List[Dict]     # Positive indicators
+            },
+            "technical_requirements": {
+                "integration_needs": List[Dict],      # Integration requirements
+                "technical_challenges": List[Dict],   # Technical issues
+                "infrastructure_updates": List[Dict], # Infrastructure needs
+                "security_requirements": List[Dict]   # Security considerations
+            }
+        }
+
+    async def extract_intelligence(self, 
+                                transcript: str, 
+                                context: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Extract comprehensive intelligence from transcript
+        
+        Args:
+            transcript: Call transcript text
+            context: Additional context (account history, etc.)
+            
+        Returns:
+            Dict containing extracted intelligence
+        """
+        try:
+            # Extract intelligence using LLM
+            prompt = self._generate_extraction_prompt(transcript, context)
+            analysis = await self.llm_service.analyze_with_context(prompt, self.extraction_schema)
+            
+            # Post-process and link entities
+            processed_results = self._process_extracted_intelligence(analysis, context)
+            
+            return processed_results
+            
+        except Exception as e:
+            logger.error(f"Intelligence extraction failed: {str(e)}")
+            raise
+
+    def _generate_extraction_prompt(self, transcript: str, context: Dict) -> str:
+        """Generate context-aware extraction prompt"""
+        return f"""
+        Analyze this sales interaction transcript and extract comprehensive business intelligence.
+        Consider all aspects of account management, relationships, opportunities, and market dynamics.
+        
+        Account Context:
+        - Client: {context.get('account_name')}
+        - Industry: {context.get('industry')}
+        - Relationship Status: {context.get('relationship_status')}
+        
+        Focus on:
+        1. Relationship mapping and stakeholder influence
+        2. Opportunity identification and updates
+        3. Project intelligence and health
+        4. Market and competitive insights
+        5. Follow-up actions and next steps
+        6. Technical and integration requirements
+        
+        Transcript:
+        {transcript}
+        """
+
+    def _process_extracted_intelligence(self, 
+                                    analysis: Dict[str, Any], 
+                                    context: Dict[str, Any]) -> Dict[str, Any]:
+        """Process and link extracted intelligence"""
+        try:
+            # Enhance with relationship links
+            self._link_stakeholders(analysis, context)
+            
+            # Map to existing opportunities
+            self._map_opportunities(analysis, context)
+            
+            # Update project intelligence
+            self._update_project_status(analysis, context)
+            
+            # Generate action recommendations
+            self._generate_recommendations(analysis)
+            
+            return analysis
+            
+        except Exception as e:
+            logger.error(f"Intelligence processing failed: {str(e)}")
+            raise
+
+    def _link_stakeholders(self, analysis: Dict, context: Dict) -> None:
+        """Link mentioned stakeholders with CRM data"""
+        # Implementation for stakeholder linking
+
+    def _map_opportunities(self, analysis: Dict, context: Dict) -> None:
+        """Map extracted opportunities to existing pipeline"""
+        # Implementation for opportunity mapping
+
+    def _update_project_status(self, analysis: Dict, context: Dict) -> None:
+        """Update project status based on new intelligence"""
+        # Implementation for project updates
+
+    def _generate_recommendations(self, analysis: Dict) -> None:
+        """Generate action recommendations based on intelligence"""
+        # Implementation for recommendation generation