Code Examples

import asyncio
import logging
import aiohttp
import json
from datetime import datetime
from dataclasses import dataclass
from typing import Dict, List, Optional

# Production-ready configuration
@dataclass
class SILEConfig:
    api_key: str
    base_url: str = "http://localhost:8000"
    timeout: int = 30
    max_retries: int = 3
    log_level: str = "INFO"

class IntelligentCustomerPlatform:
    """
    Production-ready intelligent customer platform powered by SILE
    Features: AI chat, persistent memory, sentiment analysis, auto-escalation
    """
    
    def __init__(self, config: SILEConfig):
        self.config = config
        self.session = None
        self.logger = self._setup_logging()
        self.headers = {
            'X-API-Key': config.api_key,
            'Content-Type': 'application/json'
        }
    
    def _setup_logging(self) -> logging.Logger:
        """Configure structured logging for production"""
        logging.basicConfig(
            level=getattr(logging, self.config.log_level),
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        return logging.getLogger('intelligent_platform')
    
    async def initialize(self):
        """Initialize async session and validate connection"""
        self.session = aiohttp.ClientSession(
            timeout=aiohttp.ClientTimeout(total=self.config.timeout),
            headers=self.headers
        )
        
        # Validate API connection
        try:
            async with self.session.get(f"{self.config.base_url}/api/health") as response:
                if response.status == 200:
                    self.logger.info("✅ SILE API connection validated")
                else:
                    raise ConnectionError(f"API health check failed: {response.status}")
        except Exception as e:
            self.logger.error(f"❌ Failed to connect to SILE API: {e}")
            raise
    
    async def process_customer_message(
        self, 
        customer_id: str, 
        message: str,
        context: Optional[Dict] = None
    ) -> Dict:
        """Process customer message with full AI intelligence"""
        self.logger.info(f"Processing message from customer {customer_id}")
        
        try:
            # Step 1: Get relevant memories for context
            memories = await self.search_customer_memories(customer_id, message)
            
            # Step 2: Process with AI chat system
            chat_payload = {
                "message": message,
                "context": {
                    "user_id": customer_id,
                    "session_id": f"support_{customer_id}_{datetime.now().timestamp()}",
                    "platform": "customer_platform",
                    "relevant_memories": memories,
                    **(context or {})
                }
            }
            
            async with self.session.post(
                f"{self.config.base_url}/api/ai/chat", 
                json=chat_payload
            ) as response:
                if response.status == 200:
                    result = await response.json()
                else:
                    raise Exception(f"AI chat failed: {response.status}")
            
            # Step 3: Store interaction in memory
            await self.store_interaction_memory(customer_id, message, result)
            
            # Step 4: Check for escalation needs
            if self._needs_escalation(result):
                await self.trigger_escalation(customer_id, result)
            
            self.logger.info(f"✅ Successfully processed message for {customer_id}")
            return {
                "success": True,
                "response": result["response"],
                "intent": result.get("intent"),
                "model_used": result.get("model_used"),
                "escalated": self._needs_escalation(result),
                "timestamp": datetime.now().isoformat()
            }
            
        except Exception as e:
            self.logger.error(f"❌ Error processing message: {e}")
            return {
                "success": False, 
                "error": str(e),
                "fallback_response": "I'm experiencing technical difficulties. Let me connect you with a human agent."
            }
    
    async def search_customer_memories(self, customer_id: str, query: str) -> List[Dict]:
        """Search for relevant customer memories to enhance context"""
        try:
            search_payload = {
                "query": f"customer:{customer_id} {query}",
                "limit": 5
            }
            
            async with self.session.post(
                f"{self.config.base_url}/api/memory/search",
                json=search_payload
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    return data.get("results", [])
                else:
                    self.logger.warning(f"Memory search failed: {response.status}")
                    return []
        except Exception as e:
            self.logger.warning(f"Memory search error: {e}")
            return []
    
    async def store_interaction_memory(self, customer_id: str, message: str, ai_result: Dict):
        """Store customer interaction for future context"""
        memory_payload = {
            "content": f"Customer: {message}\nAI: {ai_result.get('response', '')}",
            "context": f"customer_support_{customer_id}",
            "tags": ["support", "interaction", ai_result.get("intent", "general")],
            "metadata": {
                "customer_id": customer_id,
                "intent": ai_result.get("intent"),
                "model_used": ai_result.get("model_used"),
                "timestamp": datetime.now().isoformat()
            }
        }
        
        try:
            async with self.session.post(
                f"{self.config.base_url}/api/memory/memories",
                json=memory_payload
            ) as response:
                if response.status == 200:
                    self.logger.debug(f"Stored memory for customer {customer_id}")
        except Exception as e:
            self.logger.warning(f"Failed to store memory: {e}")
    
    def _needs_escalation(self, ai_result: Dict) -> bool:
        """Determine if interaction needs human escalation"""
        intent = ai_result.get("intent", "").lower()
        escalation_intents = ["frustrated", "angry", "complaint", "refund", "cancel"]
        return any(trigger in intent for trigger in escalation_intents)
    
    async def trigger_escalation(self, customer_id: str, ai_result: Dict):
        """Trigger escalation to human agent"""
        self.logger.info(f"🚨 Escalating customer {customer_id} to human agent")
        # In production: integrate with ticketing system, Slack, etc.
    
    async def get_memory_stats(self) -> Dict:
        """Get memory system statistics"""
        async with self.session.get(f"{self.config.base_url}/api/memory/stats") as response:
            return await response.json() if response.status == 200 else {}
    
    async def close(self):
        """Clean up resources"""
        if self.session:
            await self.session.close()
            self.logger.info("🔒 Platform shut down cleanly")

# Production usage example
async def main():
    """Example of production deployment"""
    config = SILEConfig(
        api_key="sile_nidgbNEmjc9zmPx6lo9VLwTKGR9qHmuPgphMpeh3ltQ",
        base_url="http://localhost:8000",
        timeout=30
    )
    
    platform = IntelligentCustomerPlatform(config)
    
    try:
        await platform.initialize()
        
        # Example customer interactions
        scenarios = [
            ("cust_001", "Hi, I'm having trouble logging into my account"),
            ("cust_002", "I'm really frustrated! My order is late and customer service is terrible!"),
            ("cust_001", "Actually, I figured out the login issue. Thanks!")
        ]
        
        for customer_id, message in scenarios:
            print(f"\n🎯 Processing: {customer_id}: {message}")
            result = await platform.process_customer_message(customer_id, message)
            
            if result["success"]:
                print(f"🤖 AI Response: {result['response']}")
                print(f"🎯 Intent: {result['intent']}")
                print(f"⚡ Model: {result['model_used']}")
                if result["escalated"]:
                    print(f"🚨 ESCALATED to human agent")
            else:
                print(f"❌ Error: {result['error']}")
                print(f"🆘 Fallback: {result['fallback_response']}")
        
        # Show memory statistics
        stats = await platform.get_memory_stats()
        print(f"\n📊 Memory Stats: {stats.get('total_memories', 0)} memories stored")
        
    finally:
        await platform.close()

if __name__ == "__main__":
    asyncio.run(main())