Update src/ai/services/intelligence_extractor.py

src/ai/services/intelligence_extractor.py

@@ -1,171 +1,356 @@
 """
-src/ai/services/intelligence_extractor.py
-Comprehensive business intelligence extraction from call transcripts
+Enhanced interaction analysis using LangGraph for agent orchestration
 """
-import logging
-from typing import Dict, Any, List
+from typing import Dict, List, Any, Optional, Annotated
 from datetime import datetime
+import uuid
+import logging
+from langgraph.graph import Graph, MessageGraph
+from langgraph.prebuilt import ToolMessage
+from langgraph.graph.message import MessageState
+import json
 
 logger = logging.getLogger(__name__)
 
-class IntelligenceExtractor:
-    """Extract comprehensive business intelligence from interaction transcripts"""
+class InteractionAnalysisGraph:
+    """
+    Orchestrates interaction analysis using LangGraph
+    """
+    def __init__(self, db_service, llm_service):
+        self.db = db_service
+        self.llm = llm_service
+        self.setup_tools()
+        self.build_graph()
     
-    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
-            }
+    def setup_tools(self):
+        """Setup tools available to agents"""
+        self.tools = {
+            # Contact Management Tools
+            'find_contact': self._create_tool(
+                self._find_contact,
+                "Find existing contact in database",
+                {"name": str, "company": str}
+            ),
+            'create_contact': self._create_tool(
+                self._create_contact,
+                "Create new contact record",
+                {"name": str, "title": str, "company": str}
+            ),
+            'update_contact': self._create_tool(
+                self._update_contact,
+                "Update existing contact",
+                {"id": str, "updates": dict}
+            ),
+            
+            # Opportunity Tools
+            'find_opportunity': self._create_tool(
+                self._find_opportunity,
+                "Find existing opportunity",
+                {"name": str, "account_id": str}
+            ),
+            'create_opportunity': self._create_tool(
+                self._create_opportunity,
+                "Create new opportunity",
+                {"name": str, "account_id": str, "value": float}
+            ),
+            'update_opportunity': self._create_tool(
+                self._update_opportunity,
+                "Update opportunity details",
+                {"id": str, "updates": dict}
+            ),
+            
+            # Follow-up Tools
+            'create_follow_up': self._create_tool(
+                self._create_follow_up,
+                "Create follow-up action",
+                {"title": str, "due_date": str, "assignee": str}
+            ),
+            'schedule_calendar': self._create_tool(
+                self._schedule_calendar,
+                "Schedule calendar event",
+                {"title": str, "date": str, "duration": int}
+            )
         }
+    
+    def build_graph(self):
+        """Build the LangGraph processing graph"""
+        workflow = Graph()
 
-    async def extract_intelligence(self, 
-                                transcript: str, 
-                                context: Dict[str, Any]) -> Dict[str, Any]:
-        """
-        Extract comprehensive intelligence from transcript
+        # Define nodes
+        workflow.add_node("extract_intelligence", self.extract_intelligence_node)
+        workflow.add_node("process_contacts", self.process_contacts_node)
+        workflow.add_node("process_opportunities", self.process_opportunities_node)
+        workflow.add_node("process_follow_ups", self.process_follow_ups_node)
+        workflow.add_node("generate_summary", self.generate_summary_node)
+
+        # Define edges
+        workflow.add_edge("extract_intelligence", "process_contacts")
+        workflow.add_edge("process_contacts", "process_opportunities")
+        workflow.add_edge("process_opportunities", "process_follow_ups")
+        workflow.add_edge("process_follow_ups", "generate_summary")
+
+        # Set entry point
+        workflow.set_entry_point("extract_intelligence")
         
-        Args:
-            transcript: Call transcript text
-            context: Additional context (account history, etc.)
-            
-        Returns:
-            Dict containing extracted intelligence
+        self.workflow = workflow.compile()
+    
+    async def process_interaction(self, interaction_data: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Process interaction through the graph
         """
         try:
-            # Extract intelligence using LLM
-            prompt = self._generate_extraction_prompt(transcript, context)
-            analysis = await self.llm_service.analyze_with_context(prompt, self.extraction_schema)
+            # Initialize state
+            state = MessageState(
+                messages=[],
+                metadata={
+                    "interaction": interaction_data,
+                    "processed_at": datetime.now().isoformat(),
+                    "results": {}
+                }
+            )
             
-            # Post-process and link entities
-            processed_results = self._process_extracted_intelligence(analysis, context)
+            # Run workflow
+            final_state = await self.workflow.ainvoke(state)
             
-            return processed_results
+            return final_state.metadata["results"]
             
         except Exception as e:
-            logger.error(f"Intelligence extraction failed: {str(e)}")
+            logger.error(f"Graph processing 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')}
+    async def extract_intelligence_node(self, state: MessageState) -> MessageState:
+        """Extract structured intelligence from interaction"""
+        interaction = state.metadata["interaction"]
         
-        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)
+            # Extract using LLM
+            extracted = await self.llm.analyze_interaction(
+                interaction["transcript"],
+                self.intelligence_schema
+            )
             
-            # Generate action recommendations
-            self._generate_recommendations(analysis)
+            # Update state
+            state.metadata["extracted"] = extracted
+            state.messages.append(
+                ToolMessage(
+                    content="Intelligence extracted successfully",
+                    tool_name="extract_intelligence",
+                    tool_output=extracted
+                )
+            )
             
-            return analysis
+            return state
             
         except Exception as e:
-            logger.error(f"Intelligence processing failed: {str(e)}")
+            logger.error(f"Intelligence extraction failed: {str(e)}")
             raise
 
-    def _link_stakeholders(self, analysis: Dict, context: Dict) -> None:
-        """Link mentioned stakeholders with CRM data"""
-        # Implementation for stakeholder linking
+    async def process_contacts_node(self, state: MessageState) -> MessageState:
+        """Process and update contacts"""
+        extracted = state.metadata["extracted"]
+        contacts = extracted.get("contacts", [])
+        results = {"contacts": {"new": [], "updated": []}}
+        
+        for contact in contacts:
+            try:
+                # Try to find existing contact
+                existing = await self.tools["find_contact"](
+                    name=contact["name"],
+                    company=contact["company"]
+                )
+                
+                if existing:
+                    # Update existing
+                    if self._should_update_contact(contact, existing):
+                        updated = await self.tools["update_contact"](
+                            id=existing["id"],
+                            updates=contact
+                        )
+                        results["contacts"]["updated"].append(updated)
+                else:
+                    # Create new
+                    if self._should_create_contact(contact):
+                        new_contact = await self.tools["create_contact"](
+                            name=contact["name"],
+                            title=contact["title"],
+                            company=contact["company"]
+                        )
+                        results["contacts"]["new"].append(new_contact)
+                
+            except Exception as e:
+                logger.error(f"Contact processing failed: {str(e)}")
+                continue
+        
+        state.metadata["results"].update(results)
+        return state
+
+    async def process_opportunities_node(self, state: MessageState) -> MessageState:
+        """Process and update opportunities"""
+        extracted = state.metadata["extracted"]
+        opportunities = extracted.get("opportunities", [])
+        results = {"opportunities": {"new": [], "updated": []}}
+        
+        for opp in opportunities:
+            try:
+                # Try to find existing opportunity
+                existing = await self.tools["find_opportunity"](
+                    name=opp["name"],
+                    account_id=state.metadata["interaction"]["account_id"]
+                )
+                
+                if existing:
+                    # Update existing
+                    if self._should_update_opportunity(opp, existing):
+                        updated = await self.tools["update_opportunity"](
+                            id=existing["id"],
+                            updates=opp
+                        )
+                        results["opportunities"]["updated"].append(updated)
+                else:
+                    # Create new
+                    if self._should_create_opportunity(opp):
+                        new_opp = await self.tools["create_opportunity"](
+                            name=opp["name"],
+                            account_id=state.metadata["interaction"]["account_id"],
+                            value=opp.get("value", 0)
+                        )
+                        results["opportunities"]["new"].append(new_opp)
+                
+            except Exception as e:
+                logger.error(f"Opportunity processing failed: {str(e)}")
+                continue
+        
+        state.metadata["results"].update(results)
+        return state
+
+    async def process_follow_ups_node(self, state: MessageState) -> MessageState:
+        """Process follow-ups and calendar events"""
+        extracted = state.metadata["extracted"]
+        follow_ups = extracted.get("follow_ups", [])
+        results = {"follow_ups": [], "calendar_events": []}
+        
+        for follow_up in follow_ups:
+            try:
+                # Create follow-up
+                new_follow_up = await self.tools["create_follow_up"](
+                    title=follow_up["title"],
+                    due_date=follow_up["due_date"],
+                    assignee=follow_up["assignee"]
+                )
+                results["follow_ups"].append(new_follow_up)
+                
+                # Schedule calendar event if needed
+                if follow_up.get("needs_calendar", False):
+                    calendar_event = await self.tools["schedule_calendar"](
+                        title=follow_up["title"],
+                        date=follow_up["due_date"],
+                        duration=follow_up.get("duration", 30)
+                    )
+                    results["calendar_events"].append(calendar_event)
+                
+            except Exception as e:
+                logger.error(f"Follow-up processing failed: {str(e)}")
+                continue
+        
+        state.metadata["results"].update(results)
+        return state
+
+    async def generate_summary_node(self, state: MessageState) -> MessageState:
+        """Generate final summary of all updates"""
+        results = state.metadata["results"]
+        
+        summary = {
+            "changes_made": {
+                "contacts": len(results["contacts"]["new"]) + len(results["contacts"]["updated"]),
+                "opportunities": len(results["opportunities"]["new"]) + len(results["opportunities"]["updated"]),
+                "follow_ups": len(results["follow_ups"])
+            },
+            "needs_attention": self._identify_attention_items(results),
+            "next_steps": self._generate_next_steps(results)
+        }
+        
+        state.metadata["results"]["summary"] = summary
+        return state
+
+    def _should_update_contact(self, new_data: Dict, existing: Dict) -> bool:
+        """Determine if contact should be updated"""
+        # Compare relevant fields and return True if update needed
+        # Add user confirmation logic here
+        return True  # Placeholder
 
-    def _map_opportunities(self, analysis: Dict, context: Dict) -> None:
-        """Map extracted opportunities to existing pipeline"""
-        # Implementation for opportunity mapping
+    def _should_create_contact(self, contact_data: Dict) -> bool:
+        """Determine if new contact should be created"""
+        # Add validation and user confirmation logic here
+        return True  # Placeholder
 
-    def _update_project_status(self, analysis: Dict, context: Dict) -> None:
-        """Update project status based on new intelligence"""
-        # Implementation for project updates
+    def _should_update_opportunity(self, new_data: Dict, existing: Dict) -> bool:
+        """Determine if opportunity should be updated"""
+        # Compare relevant fields and return True if update needed
+        # Add user confirmation logic here
+        return True  # Placeholder
 
-    def _generate_recommendations(self, analysis: Dict) -> None:
-        """Generate action recommendations based on intelligence"""
-        # Implementation for recommendation generation
+    def _should_create_opportunity(self, opp_data: Dict) -> bool:
+        """Determine if new opportunity should be created"""
+        # Add validation and user confirmation logic here
+        return True  # Placeholder
+
+    def _identify_attention_items(self, results: Dict) -> List[Dict]:
+        """Identify items needing user attention"""
+        attention_items = []
+        
+        # Add logic to identify items needing review/confirmation
+        
+        return attention_items
+
+    def _generate_next_steps(self, results: Dict) -> List[Dict]:
+        """Generate recommended next steps"""
+        next_steps = []
+        
+        # Add logic to generate recommended actions
+        
+        return next_steps
+
+    @property
+    def intelligence_schema(self) -> Dict:
+        """Schema for intelligence extraction"""
+        return {
+            "contacts": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "name": {"type": "string"},
+                        "title": {"type": "string"},
+                        "company": {"type": "string"},
+                        "department": {"type": "string"},
+                        "influence_level": {"type": "string"}
+                    }
+                }
+            },
+            "opportunities": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "name": {"type": "string"},
+                        "type": {"type": "string"},
+                        "value": {"type": "number"},
+                        "stage": {"type": "string"},
+                        "next_steps": {"type": "string"}
+                    }
+                }
+            },
+            "follow_ups": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "title": {"type": "string"},
+                        "type": {"type": "string"},
+                        "due_date": {"type": "string"},
+                        "assignee": {"type": "string"},
+                        "needs_calendar": {"type": "boolean"}
+                    }
+                }
+            }
+        }