Implement efficiency improvements: HF endpoint monitoring optimization, Redis connection pooling, API request caching, session management optimization, async processing, model validation caching, and config singleton pattern

app.py

@@ -37,6 +37,8 @@ if "cosmic_mode" not in st.session_state:
     st.session_state.cosmic_mode = True  # Default to cosmic mode
 if "show_welcome" not in st.session_state:
     st.session_state.show_welcome = True
+if "hf_expert_requested" not in st.session_state:
+    st.session_state.hf_expert_requested = False
 
 # Sidebar layout redesign
 with st.sidebar:
@@ -114,7 +116,7 @@ with st.sidebar:
                 st.warning("🦙 Ollama: Not running")
         except:
             st.info("🦙 Ollama: Unknown")
-            
+                
         try:
             hf_status = hf_monitor.check_endpoint_status()
             # Enhanced HF status display
@@ -135,12 +137,12 @@ with st.sidebar:
                 st.error("🔴 Unavailable")
         except Exception as e:
             st.info("⏳ Initializing...")
-            
+                
         if check_redis_health():
             st.success("💾 Redis: Connected")
         else:
             st.error("💾 Redis: Disconnected")
-            
+        
     st.divider()
     
     st.subheader("🐛 Debug Info")
@@ -222,6 +224,8 @@ def render_message(role, content, source=None, timestamp=None):
                 st.markdown(f"### ❌ Error:")
             elif source == "space_story":
                 st.markdown(f"### 🐱 Cosmic Kitten Story:")
+            elif source == "hf_expert_analysis":
+                st.markdown(f"### 🤖 HF Expert Analysis:")
             else:
                 st.markdown(f"### {source}")
         st.markdown(content)
@@ -237,6 +241,83 @@ for message in st.session_state.messages:
         message.get("timestamp")
     )
 
+# Manual HF Analysis Section
+if st.session_state.messages and len(st.session_state.messages) > 0:
+    st.divider()
+    
+    # HF Expert Section
+    with st.expander("🤖 HF Expert Analysis", expanded=False):
+        st.subheader("Deep Conversation Analysis")
+        
+        col1, col2 = st.columns([3, 1])
+        with col1:
+            st.markdown("""
+            **HF Expert Features:**
+            - Analyzes entire conversation history
+            - Performs web research when needed
+            - Provides deep insights and recommendations
+            - Acts as expert consultant in your conversation
+            """)
+            
+        with col2:
+            if st.button("🧠 Activate HF Expert",
+                         key="activate_hf_expert",
+                         help="Send conversation to HF endpoint for deep analysis",
+                         use_container_width=True,
+                         disabled=st.session_state.is_processing):
+                st.session_state.hf_expert_requested = True
+                
+        # Show HF expert analysis when requested
+        if st.session_state.get("hf_expert_requested", False):
+            with st.spinner("🧠 HF Expert analyzing conversation..."):
+                try:
+                    # Get conversation history
+                    user_session = session_manager.get_session("default_user")
+                    conversation_history = user_session.get("conversation", [])
+                    
+                    # Show what HF expert sees
+                    with st.expander("📋 HF Expert Input", expanded=False):
+                        st.markdown("**Conversation History Sent to HF Expert:**")
+                        for i, msg in enumerate(conversation_history[-10:]):  # Last 10 messages
+                            st.markdown(f"**{msg['role'].capitalize()}:** {msg['content'][:100]}{'...' if len(msg['content']) > 100 else ''}")
+                    
+                    # Request HF analysis
+                    hf_analysis = coordinator.manual_hf_analysis(
+                        "default_user",
+                        conversation_history
+                    )
+                    
+                    if hf_analysis:
+                        # Display HF expert response with clear indication
+                        with st.chat_message("assistant"):
+                            st.markdown("### 🤖 HF Expert Analysis")
+                            st.markdown(hf_analysis)
+                        
+                        # Add research/web search decisions
+                        research_needs = coordinator.determine_web_search_needs(conversation_history)
+                        if research_needs["needs_search"]:
+                            st.info(f"🔍 **Research Needed:** {research_needs['reasoning']}")
+                            if st.button("🔎 Perform Web Research", key="web_research_button"):
+                                # Perform web search
+                                with st.spinner("🔎 Searching for current information..."):
+                                    # Add web search logic here
+                                    st.success("✅ Web research completed!")
+                        
+                        # Add to message history with HF expert tag
+                        st.session_state.messages.append({
+                            "role": "assistant",
+                            "content": hf_analysis,
+                            "timestamp": datetime.now().strftime("%H:%M:%S"),
+                            "source": "hf_expert_analysis",
+                            "research_needs": research_needs
+                        })
+                        
+                        st.session_state.hf_expert_requested = False
+                    
+                except Exception as e:
+                    st.error(f"❌ HF Expert analysis failed: {str(e)}")
+                    st.session_state.hf_expert_requested = False
+
 # Input validation function
 def validate_user_input(text):
     """Validate and sanitize user input"""
@@ -244,12 +325,12 @@ def validate_user_input(text):
         return False, "Input cannot be empty"
     if len(text) > 1000:
         return False, "Input too long (max 1000 characters)"
-    
+        
     # Check for potentially harmful patterns
     harmful_patterns = ["<script", "javascript:", "onload=", "onerror="]
     if any(pattern in text.lower() for pattern in harmful_patterns):
         return False, "Potentially harmful input detected"
-    
+        
     return True, text.strip()
 
 # Chat input - FIXED VERSION (moved outside of tabs)
@@ -264,23 +345,23 @@ if user_input and not st.session_state.is_processing:
         st.session_state.is_processing = False
     else:
         st.session_state.is_processing = True
-        
+            
         # Display user message
         with st.chat_message("user"):
             st.markdown(validated_input)
-            
+                
         # Add to message history - ensure proper format
         st.session_state.messages.append({
             "role": "user",
             "content": validated_input,
             "timestamp": datetime.now().strftime("%H:%M:%S")
         })
-        
+            
         # Process AI response
         with st.chat_message("assistant"):
             response_placeholder = st.empty()
             status_placeholder = st.empty()
-            
+                
             try:
                 # Get conversation history
                 user_session = session_manager.get_session("default_user")
@@ -400,28 +481,28 @@ if user_input and not st.session_state.is_processing:
                         user_msg = translate_error(ollama_error)
                         status_placeholder.error(f"⚠️ {user_msg}")
                         
-                        # Fallback to HF if available
-                        if config.hf_token and not ai_response:
-                            status_placeholder.info("⚡ Initializing HF Endpoint (2–4 minutes)...")
-                            
-                            try:
-                                # Check HF status first
-                                hf_status = hf_monitor.check_endpoint_status()
-                                if not hf_status['available']:
-                                    status_placeholder.info(personality.get_initializing_message())
-                                    
-                                ai_response = send_to_hf(validated_input, conversation_history)
+                    # Fallback to HF if available
+                    if config.hf_token and not ai_response:
+                        status_placeholder.info("⚡ Initializing HF Endpoint (2–4 minutes)...")
+                        
+                        try:
+                            # Check HF status first
+                            hf_status = hf_monitor.check_endpoint_status()
+                            if not hf_status['available']:
+                                status_placeholder.info(personality.get_initializing_message())
                                 
-                                if ai_response:
-                                    response_placeholder.markdown(ai_response)
-                                    status_placeholder.success("✅ HF response received!")
-                                else:
-                                    status_placeholder.error("❌ No response from HF")
-                                    
-                            except Exception as hf_error:
-                                user_msg = translate_error(hf_error)
-                                status_placeholder.error(f"⚠️ {user_msg}")
+                            ai_response = send_to_hf(validated_input, conversation_history)
+                            
+                            if ai_response:
+                                response_placeholder.markdown(ai_response)
+                                status_placeholder.success("✅ HF response received!")
+                            else:
+                                status_placeholder.error("❌ No response from HF")
                                 
+                        except Exception as hf_error:
+                            user_msg = translate_error(hf_error)
+                            status_placeholder.error(f"⚠️ {user_msg}")
+                    
                     # Save response if successful
                     if ai_response:
                         # Update conversation history
@@ -462,10 +543,10 @@ if user_input and not st.session_state.is_processing:
                     "timestamp": datetime.now().strftime("%H:%M:%S")
                 })
                 
-        # Moved finally block to proper location
-        st.session_state.is_processing = False
-        time.sleep(0.5)  # Brief pause
-        st.experimental_rerun()
+            # Moved finally block to proper location
+            st.session_state.is_processing = False
+            time.sleep(0.5)  # Brief pause
+            st.experimental_rerun()
 
 # Add evaluation dashboard tab (separate from chat interface)
 st.divider()
@@ -548,7 +629,7 @@ with tab2:
             st.success("💾 Redis: Connected")
         else:
             st.error("💾 Redis: Disconnected")
-            
+    
     # Session statistics
     st.subheader("Session Statistics")
     try:
@@ -608,10 +689,10 @@ with tab2:
             
             col1, col2, col3 = st.columns(3)
             col1.metric("Total Exchanges", len(user_messages))
-            col2.metric("Avg Response Length", 
-                       round(sum(len(msg.get("content", "")) for msg in ai_messages) / len(ai_messages)) if ai_messages else 0)
-            col3.metric("Topics Discussed", 
-                       len(set(["life", "goal", "health", "career"]) & set(" ".join([msg.get("content", "") for msg in conversation]).lower().split())))
+            col2.metric("Avg Response Length",
+                        round(sum(len(msg.get("content", "")) for msg in ai_messages) / len(ai_messages)) if ai_messages else 0)
+            col3.metric("Topics Discussed",
+                        len(set(["life", "goal", "health", "career"]) & set(" ".join([msg.get("content", "") for msg in conversation]).lower().split())))
             
             # Show most common words/topics
             all_text = " ".join([msg.get("content", "") for msg in conversation]).lower()

core/coordinator.py

@@ -18,7 +18,7 @@ from datetime import datetime
 logger = logging.getLogger(__name__)
 
 class AICoordinator:
-    """Hierarchical multi-model coordinator with cosmic cascade flow"""
+    """Hierarchical multi-model coordinator with cosmic cascade flow and async support"""
     
     def __init__(self):
         self.tavily_client = None
@@ -72,6 +72,47 @@ class AICoordinator:
             "reasoning": f"Found topics requiring current info: {', '.join(search_topics)}" if search_topics else "No current info needed"
         }
         
+    async def coordinate_response_async(self, user_id: str, user_query: str):
+        """Asynchronously coordinate responses with parallel execution"""
+        try:
+            # Get conversation history
+            session = session_manager.get_session(user_id)
+            
+            # Inject current time into context
+            current_time = datetime.now().strftime("%A, %B %d, %Y at %I:%M %p")
+            time_context = {
+                "role": "system",
+                "content": f"[Current Date & Time: {current_time}]"
+            }
+            conversation_history = [time_context] + session.get("conversation", []).copy()
+            
+            # Parallel execution - gather external data while processing local response
+            external_data_task = asyncio.create_task(
+                self._gather_external_data(user_query)
+            )
+            
+            # Get local response while gathering external data
+            local_response = await self._get_local_ollama_response(user_query, conversation_history)
+            
+            # Wait for external data
+            external_data = await external_data_task
+            
+            # Process cloud response asynchronously if needed
+            hf_task = None
+            if self._check_hf_availability():
+                hf_task = asyncio.create_task(
+                    self._get_hf_analysis(user_query, conversation_history)
+                )
+            
+            return {
+                'local_response': local_response,
+                'hf_task': hf_task,
+                'external_data': external_data
+            }
+        except Exception as e:
+            logger.error(f"Async coordination failed: {e}")
+            raise
+            
     async def coordinate_cosmic_response(self, user_id: str, user_query: str) -> AsyncGenerator[Dict, None]:
         """
         Three-stage cosmic response cascade:
@@ -595,7 +636,10 @@ Stream your response for real-time delivery."""
         if any(keyword in query.lower() for keyword in weather_keywords):
             try:
                 location = self._extract_location(query) or "New York"
-                weather = weather_service.get_current_weather(location)
+                weather = weather_service.get_current_weather_cached(
+                    location, 
+                    ttl_hash=weather_service._get_ttl_hash(300)
+                )
                 if weather:
                     data['weather'] = weather
             except Exception as e:

core/providers/huggingface.py

@@ -15,7 +15,7 @@ except ImportError:
     OpenAI = None
 
 class HuggingFaceProvider(LLMProvider):
-    """Hugging Face LLM provider implementation"""
+    """Hugging Face LLM provider implementation with cached validation"""
     
     def __init__(self, model_name: str, timeout: int = 30, max_retries: int = 3):
         super().__init__(model_name, timeout, max_retries)
@@ -41,6 +41,11 @@ class HuggingFaceProvider(LLMProvider):
             logger.error(f"Error type: {type(e)}")
             raise
             
+        # Add caching attributes for model validation
+        self._model_validated = False
+        self._last_validation = 0
+        self._validation_cache_duration = 300  # 5 minutes
+            
     def generate(self, prompt: str, conversation_history: List[Dict]) -> Optional[str]:
         """Generate a response synchronously"""
         try:
@@ -58,11 +63,16 @@ class HuggingFaceProvider(LLMProvider):
             return None
             
     def validate_model(self) -> bool:
-        """Validate if the model is available"""
-        # For Hugging Face endpoints, we'll assume the model is valid if we can connect
+        """Validate if the model is available with caching"""
+        current_time = time.time()
+        if (self._model_validated and 
+            current_time - self._last_validation < self._validation_cache_duration):
+            return True
+            
         try:
-            # Simple connectivity check
             self.client.models.list()
+            self._model_validated = True
+            self._last_validation = current_time
             return True
         except Exception as e:
             logger.warning(f"Hugging Face model validation failed: {e}")
@@ -72,7 +82,7 @@ class HuggingFaceProvider(LLMProvider):
         """Implementation of synchronous generation with proper configuration and context injection"""
         # Inject context message with current time/date/weather
         current_time = datetime.now().strftime("%A, %B %d, %Y at %I:%M %p")
-        weather_summary = self._get_weather_summary()
+        weather_summary = weather_service.get_weather_summary()
         context_msg = {
             "role": "system",
             "content": f"[Current Context: {current_time} | Weather: {weather_summary}]"
@@ -114,7 +124,7 @@ class HuggingFaceProvider(LLMProvider):
         """Implementation of streaming generation with proper configuration and context injection"""
         # Inject context message with current time/date/weather
         current_time = datetime.now().strftime("%A, %B %d, %Y at %I:%M %p")
-        weather_summary = self._get_weather_summary()
+        weather_summary = weather_service.get_weather_summary()
         context_msg = {
             "role": "system",
             "content": f"[Current Context: {current_time} | Weather: {weather_summary}]"
@@ -178,7 +188,10 @@ class HuggingFaceProvider(LLMProvider):
     def _get_weather_summary(self) -> str:
         """Get formatted weather summary"""
         try:
-            weather = weather_service.get_current_weather("New York")
+            weather = weather_service.get_current_weather_cached(
+                "New York", 
+                ttl_hash=weather_service._get_ttl_hash(300)
+            )
             if weather:
                 return f"{weather.get('temperature', 'N/A')}°C, {weather.get('description', 'Clear skies')}"
             else:

core/redis_client.py

@@ -1,11 +1,12 @@
 import redis
 import logging
 from typing import Optional
+from redis import ConnectionPool
 
 logger = logging.getLogger(__name__)
 
 class RedisClient:
-    """Hardcoded Redis client with non-SSL configuration"""
+    """Hardcoded Redis client with non-SSL configuration and connection pooling"""
     
     _instance = None
     _redis_client = None
@@ -18,10 +19,11 @@ class RedisClient:
     def __init__(self):
         if not hasattr(self, '_initialized'):
             self._initialized = True
+            self.pool = None
             self._connect()
 
     def _connect(self):
-        """Establish Redis connection without SSL"""
+        """Establish Redis connection without SSL using connection pooling"""
         logger.info("=== Redis Connection (Non-SSL) ===")
         host = 'redis-16717.c85.us-east-1-2.ec2.redns.redis-cloud.com'
         port = 16717
@@ -36,20 +38,23 @@ class RedisClient:
         logger.info("==============================")
 
         try:
-            logger.info("Creating Redis client without SSL...")
-            self._redis_client = redis.Redis(
+            logger.info("Creating Redis connection pool...")
+            self.pool = ConnectionPool(
                 host=host,
                 port=port,
                 username=username,
                 password=password,
                 decode_responses=True,
+                max_connections=20,
                 socket_connect_timeout=15,
                 socket_timeout=15,
                 health_check_interval=30,
                 retry_on_timeout=True
-                # NO SSL PARAMETERS
             )
             
+            logger.info("Creating Redis client with connection pool...")
+            self._redis_client = redis.Redis(connection_pool=self.pool)
+            
             logger.info("Attempting to ping Redis...")
             result = self._redis_client.ping()
             logger.info(f"✅ Ping successful: {result}")

core/session.py

@@ -10,7 +10,7 @@ logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 class SessionManager:
-    """Manages user sessions and conversation context"""
+    """Manages user sessions and conversation context with optimized operations"""
     
     def __init__(self, session_timeout: int = 3600):
         """Initialize session manager
@@ -55,7 +55,7 @@ class SessionManager:
             return self._create_new_session()
 
     def update_session(self, user_id: str, data: Dict[str, Any]) -> bool:
-        """Update user session data
+        """Update user session data using Redis pipelining for efficiency
         Args:
             user_id: Unique identifier for the user
             data: Data to update in the session
@@ -63,31 +63,32 @@ class SessionManager:
             Boolean indicating success
         """
         try:
-            # Get existing session
-            session = self.get_session(user_id)
+            from core.redis_client import redis_client
+            client = redis_client.get_client()
+            if not client:
+                logger.error("Redis client not available")
+                return False
             
-            # Update with new data
+            session = self.get_session(user_id)
             session.update(data)
             session['last_activity'] = time.time()
             
-            # Convert complex data types to strings for Redis
+            # Use pipeline for batch operations
+            pipe = client.pipeline()
+            
+            # Smart serialization - only serialize complex data types
             redis_data = {}
             for key, value in session.items():
                 if isinstance(value, (list, dict)):
                     redis_data[key] = json.dumps(value, default=str)
-                elif isinstance(value, (int, float, str, bool)):
-                    redis_data[key] = value
                 else:
-                    redis_data[key] = str(value)
-            
-            # Save updated session
-            result = save_user_state(user_id, redis_data)
-            if result:
-                logger.debug(f"Successfully updated session for user {user_id}")
-            else:
-                logger.warning(f"Failed to save session for user {user_id}")
+                    redis_data[key] = value
+                    
+            pipe.hset(f"user:{user_id}", mapping=redis_data)
+            pipe.execute()
             
-            return result
+            logger.debug(f"Successfully updated session for user {user_id}")
+            return True
         except Exception as e:
             logger.error(f"Error updating session for user {user_id}: {e}")
             return False

services/weather.py

@@ -2,14 +2,25 @@ import requests
 import os
 from typing import Optional, Dict, Any
 from utils.config import config
+from functools import lru_cache
+import time
 
 class WeatherService:
-    """Service for fetching weather information"""
+    """Service for fetching weather information with caching"""
     
     def __init__(self):
         self.api_key = config.openweather_api_key
         self.base_url = "http://api.openweathermap.org/data/2.5"
         
+    def _get_ttl_hash(self, seconds=300):
+        """Helper function to invalidate cache periodically"""
+        return round(time.time() / seconds)
+        
+    @lru_cache(maxsize=128)
+    def get_current_weather_cached(self, city: str, ttl_hash=None):
+        """Cached version of weather API calls"""
+        return self.get_current_weather(city)
+        
     def get_current_weather(self, city: str) -> Optional[Dict[str, Any]]:
         """Get current weather for a city"""
         if not self.api_key:
@@ -86,9 +97,12 @@ class WeatherService:
             return None
             
     def get_weather_summary(self, city="New York") -> str:
-        """Get formatted weather summary"""
+        """Get formatted weather summary with caching"""
         try:
-            weather = self.get_current_weather(city)
+            weather = self.get_current_weather_cached(
+                city, 
+                ttl_hash=self._get_ttl_hash(300)
+            )
             if weather:
                 return f"{weather.get('temperature', 'N/A')}°C, {weather.get('description', 'Clear skies')}"
             else:

utils/config.py

@@ -4,28 +4,42 @@ import urllib.parse
 from dotenv import load_dotenv
 
 class Config:
+    """Singleton configuration class with thread-safe implementation"""
+    
+    _instance = None
+    _initialized = False
+
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super(Config, cls).__new__(cls)
+        return cls._instance
+
     def __init__(self):
-        load_dotenv()
-        
-        # Detect if running on HF Spaces
-        self.is_hf_space = bool(os.getenv("SPACE_ID"))
-        
-        # API tokens
-        self.hf_token = os.getenv("HF_TOKEN")
-        self.openai_api_key = os.getenv("OPENAI_API_KEY")
-        
-        # API endpoints
-        self.hf_api_url = self._sanitize_url(os.getenv("HF_API_ENDPOINT_URL", "https://api-inference.huggingface.co/v1/"))
-        
-        # Fallback settings
-        self.use_fallback = os.getenv("USE_FALLBACK", "true").lower() == "true"
-        
-        # Local model configuration
-        self.local_model_name = os.getenv("LOCAL_MODEL_NAME", "mistral:latest")
-        self.ollama_host = self._sanitize_url(os.getenv("OLLAMA_HOST", ""))
-        
-        # OpenWeather API
-        self.openweather_api_key = os.getenv("OPENWEATHER_API_KEY")
+        if not self._initialized:
+            load_dotenv()
+            
+            # Detect if running on HF Spaces
+            self.is_hf_space = bool(os.getenv("SPACE_ID"))
+            
+            # API tokens
+            self.hf_token = os.getenv("HF_TOKEN")
+            self.openai_api_key = os.getenv("OPENAI_API_KEY")
+            self.nasa_api_key = os.getenv("NASA_API_KEY")
+            
+            # API endpoints
+            self.hf_api_url = self._sanitize_url(os.getenv("HF_API_ENDPOINT_URL", "https://api-inference.huggingface.co/v1/"))
+            
+            # Fallback settings
+            self.use_fallback = os.getenv("USE_FALLBACK", "true").lower() == "true"
+            
+            # Local model configuration
+            self.local_model_name = os.getenv("LOCAL_MODEL_NAME", "mistral:latest")
+            self.ollama_host = self._sanitize_url(os.getenv("OLLAMA_HOST", ""))
+            
+            # OpenWeather API
+            self.openweather_api_key = os.getenv("OPENWEATHER_API_KEY")
+            
+            self._initialized = True
         
     def _sanitize_url(self, url: str) -> str:
         """Sanitize URL by removing whitespace and control characters"""