import streamlit as st import time import os import sys import json import asyncio from datetime import datetime from pathlib import Path sys.path.append(str(Path(__file__).parent)) from utils.config import config from core.llm import send_to_ollama, send_to_hf from core.session import session_manager from core.memory import check_redis_health from core.coordinator import coordinator from core.errors import translate_error import logging # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) st.set_page_config(page_title="CosmicCat AI Assistant", page_icon="🐱", layout="wide") # Initialize session state safely at the top of app.py if "messages" not in st.session_state: st.session_state.messages = [] if "last_error" not in st.session_state: st.session_state.last_error = "" if "is_processing" not in st.session_state: st.session_state.is_processing = False if "ngrok_url_temp" not in st.session_state: st.session_state.ngrok_url_temp = st.session_state.get("ngrok_url", "https://7bcc180dffd1.ngrok-free.app") if "hf_expert_requested" not in st.session_state: st.session_state.hf_expert_requested = False if "cosmic_mode" not in st.session_state: st.session_state.cosmic_mode = True # Default to cosmic mode # Sidebar layout redesign with st.sidebar: st.title("🐱 CosmicCat AI Assistant") st.markdown("Your personal AI-powered life development assistant") # PRIMARY ACTIONS st.subheader("đŸ’Ŧ Primary Actions") model_options = { "Mistral 7B (Local)": "mistral:latest", "Llama 2 7B (Local)": "llama2:latest", "OpenChat 3.5 (Local)": "openchat:latest" } selected_model_name = st.selectbox( "Select Model", options=list(model_options.keys()), index=0, key="sidebar_model_select" ) st.session_state.selected_model = model_options[selected_model_name] # Toggle for cosmic mode st.session_state.cosmic_mode = st.toggle("Enable Cosmic Cascade", value=st.session_state.cosmic_mode) st.divider() # CONFIGURATION st.subheader("âš™ī¸ Configuration") ngrok_url_input = st.text_input( "Ollama Server URL", value=st.session_state.ngrok_url_temp, help="Enter your ngrok URL", key="sidebar_ngrok_url" ) if ngrok_url_input != st.session_state.ngrok_url_temp: st.session_state.ngrok_url_temp = ngrok_url_input st.success("✅ URL updated!") if st.button("📡 Test Connection"): try: import requests headers = { "ngrok-skip-browser-warning": "true", "User-Agent": "CosmicCat-Test" } with st.spinner("Testing connection..."): response = requests.get( f"{ngrok_url_input}/api/tags", headers=headers, timeout=15 ) if response.status_code == 200: st.success("✅ Connection successful!") else: st.error(f"❌ Failed: {response.status_code}") except Exception as e: st.error(f"❌ Error: {str(e)[:50]}...") if st.button("đŸ—‘ī¸ Clear History"): st.session_state.messages = [] st.success("History cleared!") st.divider() # ADVANCED FEATURES with st.expander("🔍 Advanced Features", expanded=False): st.subheader("📊 System Monitor") try: from services.ollama_monitor import check_ollama_status ollama_status = check_ollama_status() if ollama_status.get("running"): st.success("đŸĻ™ Ollama: Running") else: st.warning("đŸĻ™ Ollama: Not running") except: st.info("đŸĻ™ Ollama: Unknown") try: from services.hf_endpoint_monitor import hf_monitor hf_status = hf_monitor.check_endpoint_status() if hf_status['available']: st.success("🤗 HF: Available") else: st.warning("🤗 HF: Not available") except: st.info("🤗 HF: Unknown") if check_redis_health(): st.success("💾 Redis: Connected") else: st.error("💾 Redis: Disconnected") st.divider() st.subheader("🤖 HF Expert Analysis") 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 """) if st.button("🧠 Activate HF Expert", key="activate_hf_expert_sidebar", 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 st.divider() st.subheader("🐛 Debug Info") # Show current configuration st.markdown(f"**Environment:** {'HF Space' if config.is_hf_space else 'Local'}") st.markdown(f"**Model:** {st.session_state.selected_model}") st.markdown(f"**Ollama URL:** {st.session_state.ngrok_url_temp}") st.markdown(f"**Cosmic Mode:** {'Enabled' if st.session_state.cosmic_mode else 'Disabled'}") # Show active features features = [] if config.hf_token: features.append("HF Expert") if os.getenv("TAVILY_API_KEY"): features.append("Web Search") if config.openweather_api_key: features.append("Weather") st.markdown(f"**Active Features:** {', '.join(features) if features else 'None'}") # Main interface st.title("🐱 CosmicCat AI Assistant") st.markdown("Ask me anything about personal development, goal setting, or life advice!") # Consistent message rendering function with cosmic styling def render_message(role, content, source=None, timestamp=None): """Render chat messages with consistent styling""" with st.chat_message(role): if source: if source == "local_kitty": st.markdown(f"### 🐱 Cosmic Kitten Says:") elif source == "orbital_station": st.markdown(f"### đŸ›°ī¸ Orbital Station Reports:") elif source == "cosmic_summary": st.markdown(f"### 🌟 Final Cosmic Summary:") elif source == "error": st.markdown(f"### ❌ Error:") elif source == "hf_expert": st.markdown(f"### 🤖 HF Expert Analysis:") else: st.markdown(f"### {source}") st.markdown(content) if timestamp: st.caption(f"🕒 {timestamp}") # Display messages for message in st.session_state.messages: render_message( message["role"], message["content"], message.get("source"), 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 enhanced visual indication 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 """) # Show conversation preview for HF expert st.markdown("**Conversation Preview for HF Expert:**") st.markdown("---") for i, msg in enumerate(st.session_state.messages[-5:]): # Last 5 messages role = "👤 You" if msg["role"] == "user" else "🤖 Assistant" st.markdown(f"**{role}:** {msg['content'][:100]}{'...' if len(msg['content']) > 100 else ''}") st.markdown("---") # Show web search determination try: user_session = session_manager.get_session("default_user") conversation_history = user_session.get("conversation", []) research_needs = coordinator.determine_web_search_needs(conversation_history) if research_needs["needs_search"]: st.info(f"🔍 **Research Needed:** {research_needs['reasoning']}") else: st.success("✅ No research needed for this conversation") except Exception as e: st.warning("âš ī¸ Could not determine research needs") 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 (outside of the expander) 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 in a separate expander 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", "research_needs": research_needs }) st.session_state.hf_expert_requested = False except Exception as e: user_msg = translate_error(e) st.error(f"❌ HF Expert analysis failed: {user_msg}") st.session_state.hf_expert_requested = False # Input validation function def validate_user_input(text): """Validate and sanitize user input""" if not text or not text.strip(): 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 = ["