Implement coordinated multi-model AI system with HF endpoint monitoring

core/coordinator.py

@@ -0,0 +1,218 @@
+import asyncio
+import logging
+from typing import List, Dict, Optional, Tuple
+from core.llm_factory import llm_factory
+from core.session import session_manager
+from services.hf_endpoint_monitor import hf_monitor
+from services.weather import weather_service
+try:
+    from tavily import TavilyClient
+    TAVILY_AVAILABLE = True
+except ImportError:
+    TavilyClient = None
+    TAVILY_AVAILABLE = False
+import os
+
+logger = logging.getLogger(__name__)
+
+class AICoordinator:
+    """Coordinate multiple AI models and external services"""
+    
+    def __init__(self):
+        self.tavily_client = None
+        if TAVILY_AVAILABLE and os.getenv("TAVILY_API_KEY"):
+            self.tavily_client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
+
+    async def coordinate_response(self, user_id: str, user_query: str) -> Dict:
+        """
+        Coordinate Ollama (fast) and HF (deep) responses
+        
+        Returns:
+            Dict with 'immediate_response' and 'final_response'
+        """
+        try:
+            # Get conversation history
+            session = session_manager.get_session(user_id)
+            conversation_history = session.get("conversation", [])
+            
+            # Step 1: Gather external data with Ollama
+            logger.info("Step 1: Gathering external data...")
+            external_data = await self._gather_external_data(user_query)
+            
+            # Step 2: Get immediate Ollama response (fast)
+            logger.info("Step 2: Getting immediate Ollama response...")
+            immediate_response = await self._get_ollama_response(
+                user_query, conversation_history, external_data
+            )
+            
+            # Step 3: Initialize HF endpoint in background
+            logger.info("Step 3: Initializing HF endpoint...")
+            hf_task = asyncio.create_task(self._initialize_and_get_hf_response(
+                user_query, conversation_history, external_data, immediate_response
+            ))
+            
+            # Return immediate response while HF processes
+            return {
+                'immediate_response': immediate_response,
+                'hf_task': hf_task,  # Background task for HF processing
+                'external_data': external_data
+            }
+            
+        except Exception as e:
+            logger.error(f"Coordination failed: {e}")
+            # Fallback to simple Ollama response
+            immediate_response = await self._get_ollama_response(
+                user_query, conversation_history, {}
+            )
+            return {
+                'immediate_response': immediate_response,
+                'hf_task': None,
+                'external_data': {}
+            }
+
+    async def _gather_external_data(self, query: str) -> Dict:
+        """Gather external data from various sources"""
+        data = {}
+        
+        # Tavily/DuckDuckGo search
+        if self.tavily_client:
+            try:
+                search_result = self.tavily_client.search(query, max_results=3)
+                data['search_results'] = search_result.get('results', [])
+            except Exception as e:
+                logger.warning(f"Tavily search failed: {e}")
+        
+        # Weather data (if location mentioned)
+        if 'weather' in query.lower() or 'temperature' in query.lower():
+            try:
+                # Extract location from query or use default
+                location = self._extract_location(query) or "New York"
+                weather = weather_service.get_current_weather(location)
+                if weather:
+                    data['weather'] = weather
+            except Exception as e:
+                logger.warning(f"Weather data failed: {e}")
+        
+        # Current date/time
+        from datetime import datetime
+        data['current_datetime'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        
+        return data
+
+    async def _get_ollama_response(self, query: str, history: List, external_data: Dict) -> str:
+        """Get fast response from Ollama"""
+        try:
+            # Enhance query with external data
+            enhanced_query = self._enhance_query_with_data(query, external_data)
+            
+            # Get Ollama provider
+            ollama_provider = llm_factory.get_provider('ollama')
+            if not ollama_provider:
+                raise Exception("Ollama provider not available")
+            
+            # Prepare conversation with external context
+            enhanced_history = history.copy()
+            if external_data:
+                context_message = {
+                    "role": "system",
+                    "content": f"External context: {str(external_data)}"
+                }
+                enhanced_history.insert(0, context_message)
+            
+            enhanced_history.append({"role": "user", "content": enhanced_query})
+            
+            # Generate response
+            response = ollama_provider.generate(enhanced_query, enhanced_history)
+            return response or "I'm processing your request..."
+            
+        except Exception as e:
+            logger.error(f"Ollama response failed: {e}")
+            return "I'm thinking about your question..."
+
+    async def _initialize_and_get_hf_response(self, query: str, history: List,
+                                             external_data: Dict, ollama_response: str) -> Optional[str]:
+        """Initialize HF endpoint and get deep analysis"""
+        try:
+            # Check if HF endpoint is available
+            hf_status = hf_monitor.check_endpoint_status()
+            
+            if not hf_status['available']:
+                logger.info("HF endpoint not available, attempting to warm up...")
+                # Try to warm up the endpoint
+                warmup_success = hf_monitor.warm_up_endpoint()
+                if not warmup_success:
+                    return None
+            
+            # Get HF provider
+            hf_provider = llm_factory.get_provider('huggingface')
+            if not hf_provider:
+                return None
+            
+            # Prepare enhanced conversation for HF
+            enhanced_history = history.copy()
+            
+            # Add Ollama's initial response for HF to consider
+            enhanced_history.append({
+                "role": "assistant",
+                "content": f"Initial response (for reference): {ollama_response}"
+            })
+            
+            # Add external data context
+            if external_data:
+                context_message = {
+                    "role": "system",
+                    "content": f"Additional context data: {str(external_data)}"
+                }
+                enhanced_history.insert(0, context_message)
+            
+            # Add HF's role instruction
+            enhanced_history.append({
+                "role": "system",
+                "content": "You are providing deep analysis and second opinions. Consider the initial response and enhance it with deeper insights."
+            })
+            
+            enhanced_history.append({"role": "user", "content": query})
+            
+            # Generate deep response
+            deep_response = hf_provider.generate(query, enhanced_history)
+            return deep_response
+            
+        except Exception as e:
+            logger.error(f"HF response failed: {e}")
+            return None
+
+    def _enhance_query_with_data(self, query: str, data: Dict) -> str:
+        """Enhance query with gathered external data"""
+        if not data:
+            return query
+        
+        context_parts = []
+        
+        if 'search_results' in data:
+            context_parts.append("Recent information:")
+            for result in data['search_results'][:2]:  # Limit to 2 results
+                context_parts.append(f"- {result.get('title', 'Result')}: {result.get('content', '')[:100]}...")
+        
+        if 'weather' in data:
+            weather = data['weather']
+            context_parts.append(f"Current weather: {weather.get('temperature', 'N/A')}°C in {weather.get('city', 'Unknown')}")
+        
+        if 'current_datetime' in data:
+            context_parts.append(f"Current time: {data['current_datetime']}")
+        
+        if context_parts:
+            return f"{query}\n\nContext: {' '.join(context_parts)}"
+        
+        return query
+
+    def _extract_location(self, query: str) -> Optional[str]:
+        """Extract location from query (simple implementation)"""
+        # This could be enhanced with NER or more sophisticated parsing
+        locations = ['New York', 'London', 'Tokyo', 'Paris', 'Berlin', 'Sydney']
+        for loc in locations:
+            if loc.lower() in query.lower():
+                return loc
+        return None
+
+# Global coordinator instance
+coordinator = AICoordinator()

core/session.py

@@ -3,6 +3,7 @@ import time
 from typing import Dict, Any, Optional
 from core.memory import load_user_state, save_user_state
 import logging
+from datetime import datetime
 
 # Set up logging
 logging.basicConfig(level=logging.INFO)
@@ -91,6 +92,51 @@ class SessionManager:
             logger.error(f"Error updating session for user {user_id}: {e}")
             return False
 
+    def update_session_with_ai_coordination(self, user_id: str, ai_data: Dict) -> bool:
+        """Update session with AI coordination data"""
+        try:
+            # Get existing session
+            session = self.get_session(user_id)
+            
+            # Add AI coordination tracking
+            if 'ai_coordination' not in session:
+                session['ai_coordination'] = {
+                    'requests_processed': 0,
+                    'ollama_responses': 0,
+                    'hf_responses': 0,
+                    'last_coordination': None
+                }
+            
+            coord_data = session['ai_coordination']
+            coord_data['requests_processed'] += 1
+            coord_data['last_coordination'] = datetime.now().isoformat()
+            
+            # Track response types
+            if 'immediate_response' in ai_data:
+                coord_data['ollama_responses'] += 1
+            if ai_data.get('hf_response'):
+                coord_data['hf_responses'] += 1
+            
+            # Convert complex data to JSON strings for Redis
+            redis_data = {}
+            for key, value in session.items():
+                if isinstance(value, (dict, list)):
+                    redis_data[key] = json.dumps(value)
+                else:
+                    redis_data[key] = value
+            
+            # Save updated session
+            result = save_user_state(user_id, redis_data)
+            if result:
+                logger.debug(f"Successfully updated coordination session for user {user_id}")
+            else:
+                logger.warning(f"Failed to save coordination session for user {user_id}")
+            
+            return result
+        except Exception as e:
+            logger.error(f"Error updating coordination session for user {user_id}: {e}")
+            return False
+
     def clear_session(self, user_id: str) -> bool:
         """Clear user session data
         Args:

demo_coordinated_ai.py

@@ -0,0 +1,84 @@
+import asyncio
+import sys
+from pathlib import Path
+
+# Add project root to path
+project_root = Path(__file__).parent
+sys.path.append(str(project_root))
+
+from core.coordinator import coordinator
+from core.session import session_manager
+from services.hf_endpoint_monitor import hf_monitor
+
+async def demo_coordinated_response():
+    """Demonstrate the coordinated AI response system"""
+    print("=== AI Life Coach Coordinated Response Demo ===")
+    print()
+    
+    user_id = "demo_user"
+    user_query = "What's the weather like in New York today and how should I plan my day?"
+    
+    print(f"User Query: {user_query}")
+    print()
+    
+    # Show HF endpoint status
+    print("HF Endpoint Status:")
+    print(hf_monitor.get_status_summary())
+    print()
+    
+    # Coordinate responses
+    print("Coordinating AI responses...")
+    coordination_result = await coordinator.coordinate_response(user_id, user_query)
+    
+    # Show immediate response
+    print("Immediate Response (Ollama):")
+    print(coordination_result['immediate_response'])
+    print()
+    
+    # Show external data gathered
+    print("External Data Gathered:")
+    for key, value in coordination_result['external_data'].items():
+        print(f"  {key}: {value}")
+    print()
+    
+    # Update session with coordination data
+    session_manager.update_session_with_ai_coordination(user_id, {
+        'immediate_response': coordination_result['immediate_response'],
+        'external_data': coordination_result['external_data']
+    })
+    
+    # If HF task is available, wait for it
+    hf_task = coordination_result.get('hf_task')
+    if hf_task:
+        print("Waiting for deep analysis from HF endpoint...")
+        try:
+            hf_response = await hf_task
+            if hf_response:
+                print("Deep Analysis Response (HF Endpoint):")
+                print(hf_response)
+                print()
+                
+                # Update session with HF response
+                session_manager.update_session_with_ai_coordination(user_id, {
+                    'hf_response': hf_response
+                })
+            else:
+                print("HF Endpoint did not provide a response (may still be initializing)")
+        except Exception as e:
+            print(f"Error getting HF response: {e}")
+    
+    # Show session coordination data
+    session = session_manager.get_session(user_id)
+    if 'ai_coordination' in session:
+        coord_data = session['ai_coordination']
+        print("AI Coordination Statistics:")
+        print(f"  Requests Processed: {coord_data['requests_processed']}")
+        print(f"  Ollama Responses: {coord_data['ollama_responses']}")
+        print(f"  HF Responses: {coord_data['hf_responses']}")
+        print(f"  Last Coordination: {coord_data['last_coordination']}")
+    
+    print()
+    print("🎉 Demo completed successfully!")
+
+if __name__ == "__main__":
+    asyncio.run(demo_coordinated_response())

services/hf_endpoint_monitor.py

@@ -0,0 +1,109 @@
+import requests
+import time
+import logging
+from typing import Dict, Optional
+from utils.config import config
+
+logger = logging.getLogger(__name__)
+
+class HFEndpointMonitor:
+    """Monitor Hugging Face endpoint status and health"""
+    
+    def __init__(self):
+        self.endpoint_url = config.hf_api_url
+        self.hf_token = config.hf_token
+        self.is_initialized = False
+        self.last_check = 0
+        self.check_interval = 60  # Check every minute
+
+    def check_endpoint_status(self) -> Dict:
+        """Check if HF endpoint is available and initialized"""
+        try:
+            # Check if endpoint exists and is responsive
+            headers = {"Authorization": f"Bearer {self.hf_token}"}
+            
+            # Simple model list check (doesn't trigger initialization)
+            response = requests.get(
+                f"{self.endpoint_url}/models",
+                headers=headers,
+                timeout=10
+            )
+            
+            status_info = {
+                'available': response.status_code == 200,
+                'status_code': response.status_code,
+                'initialized': self._is_endpoint_initialized(response),
+                'timestamp': time.time()
+            }
+            
+            logger.info(f"HF Endpoint Status: {status_info}")
+            return status_info
+            
+        except Exception as e:
+            logger.error(f"HF endpoint check failed: {e}")
+            return {
+                'available': False,
+                'status_code': None,
+                'initialized': False,
+                'error': str(e),
+                'timestamp': time.time()
+            }
+
+    def _is_endpoint_initialized(self, response) -> bool:
+        """Determine if endpoint is fully initialized"""
+        # If we get a model list, it's likely initialized
+        try:
+            data = response.json()
+            return 'data' in data or 'models' in data
+        except:
+            return False
+
+    def warm_up_endpoint(self) -> bool:
+        """Send a warm-up request to initialize the endpoint"""
+        try:
+            logger.info("Warming up HF endpoint...")
+            headers = {
+                "Authorization": f"Bearer {self.hf_token}",
+                "Content-Type": "application/json"
+            }
+            
+            # Simple test request to trigger initialization
+            payload = {
+                "model": "meta-llama/Llama-2-7b-chat-hf",  # Adjust as needed
+                "messages": [{"role": "user", "content": "Hello"}],
+                "max_tokens": 10
+            }
+            
+            response = requests.post(
+                f"{self.endpoint_url}/chat/completions",
+                headers=headers,
+                json=payload,
+                timeout=30
+            )
+            
+            success = response.status_code in [200, 201]
+            if success:
+                self.is_initialized = True
+                logger.info("✅ HF endpoint warmed up successfully")
+            else:
+                logger.warning(f"⚠️ HF endpoint warm-up response: {response.status_code}")
+            
+            return success
+            
+        except Exception as e:
+            logger.error(f"HF endpoint warm-up failed: {e}")
+            return False
+
+    def get_status_summary(self) -> str:
+        """Get human-readable status summary"""
+        status = self.check_endpoint_status()
+        if status['available']:
+            if status.get('initialized', False):
+                return "🟢 HF Endpoint: Available and Initialized"
+            else:
+                return "🟡 HF Endpoint: Available but Initializing"
+        else:
+            return "🔴 HF Endpoint: Unavailable"
+
+# Global instance
+hf_monitor = HFEndpointMonitor()

test_hf_monitor.py

@@ -0,0 +1,42 @@
+import sys
+from pathlib import Path
+
+# Add project root to path
+project_root = Path(__file__).parent
+sys.path.append(str(project_root))
+
+from services.hf_endpoint_monitor import hf_monitor
+
+def test_hf_monitor():
+    """Test the HF endpoint monitor"""
+    print("=== HF Endpoint Monitor Test ===")
+    print()
+    
+    # Show current status
+    print("Current HF Endpoint Status:")
+    status = hf_monitor.check_endpoint_status()
+    print(f"  Available: {status['available']}")
+    print(f"  Status Code: {status['status_code']}")
+    print(f"  Initialized: {status.get('initialized', 'Unknown')}")
+    if 'error' in status:
+        print(f"  Error: {status['error']}")
+    print()
+    
+    # Show human-readable status
+    print("Human-Readable Status:")
+    print(hf_monitor.get_status_summary())
+    print()
+    
+    # Try to warm up endpoint if not available
+    if not status['available']:
+        print("Attempting to warm up endpoint...")
+        success = hf_monitor.warm_up_endpoint()
+        print(f"Warm-up result: {'Success' if success else 'Failed'}")
+        print()
+        
+        # Check status again
+        print("Status after warm-up attempt:")
+        print(hf_monitor.get_status_summary())
+
+if __name__ == "__main__":
+    test_hf_monitor()