Create app.py

app.py

@@ -0,0 +1,530 @@
+import streamlit as st
+import torch
+from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
+from huggingface_hub import login
+import random
+import time
+import requests
+import json
+import asyncio
+import aiohttp
+from typing import Optional, List, Dict
+import nest_asyncio
+
+# Apply nest_asyncio for async operations in Streamlit
+nest_asyncio.apply()
+
+# Hugging Face token setup
+HUGGINGFACE_TOKEN = st.secrets.get("HUGGINGFACE_TOKEN", "")
+
+if HUGGINGFACE_TOKEN:
+    try:
+        login(token=HUGGINGFACE_TOKEN)
+        st.success("✅ Successfully connected to Hugging Face")
+    except Exception as e:
+        st.error(f"❌ Hugging Face login failed: {e}")
+
+# Custom CSS for silver/sky blue theme
+st.markdown("""
+<style>
+    .main {
+        background-color: #f8f9fa;
+    }
+    .stChatMessage {
+        padding: 16px;
+        border-radius: 12px;
+        margin: 8px 0;
+    }
+    .stChatMessage[data-testid="stChatMessage"] {
+        border: 1px solid #e1e5e9;
+    }
+    /* User message styling */
+    .stChatMessage:has(div[data-testid="stChatMessageContent"] > div:first-child > div:first-child) {
+        background-color: #e3f2fd;
+        border-left: 4px solid #2196f3;
+    }
+    /* Assistant message styling */
+    .stChatMessage:has(div[data-testid="stChatMessageContent"] > div:first-child > div:last-child) {
+        background-color: #ffffff;
+        border-left: 4px solid #78909c;
+    }
+    .stTextInput > div > div > input {
+        border: 2px solid #90caf9;
+        border-radius: 20px;
+        padding: 12px 16px;
+    }
+    .stButton > button {
+        background-color: #2196f3;
+        color: white;
+        border: none;
+        border-radius: 20px;
+        padding: 10px 24px;
+        font-weight: 500;
+    }
+    .stButton > button:hover {
+        background-color: #1976d2;
+        color: white;
+    }
+    .sidebar .sidebar-content {
+        background-color: #eceff1;
+    }
+    .api-status {
+        padding: 8px;
+        border-radius: 6px;
+        margin: 5px 0;
+        font-size: 0.8em;
+    }
+    .api-success {
+        background-color: #e8f5e8;
+        border-left: 4px solid #4caf50;
+    }
+    .api-error {
+        background-color: #ffebee;
+        border-left: 4px solid #f44336;
+    }
+</style>
+""", unsafe_allow_html=True)
+
+# Personal and Company Information
+PERSONAL_INFO = {
+    "name": "Adedoyin Ifeoluwa James",
+    "role": "Entrepreneur, Founder & CEO of YAH Tech",
+    "goals": [
+        "Build profitable systems that matter",
+        "Reshaping the economical world",
+        "Use software development to create flexibility and contribute to economic growth"
+    ]
+}
+
+COMPANY_INFO = {
+    "name": "YAH Tech",
+    "type": "Venture studio / app development company focused on solving problems with futuristic solutions",
+    "purpose": "Build and launch technology-driven ventures that generate profit and societal value",
+    "primary_activity": "App development and creation of scalable business systems",
+    "philosophy": "Learn, understand, create, and evaluate",
+    "stage": "Growth and development, focused on building sustainable processes and systems"
+}
+
+# API Configuration
+API_CONFIG = {
+    "news_api": {
+        "url": "https://newsapi.org/v2/top-headlines",
+        "params": {
+            "country": "us",
+            "category": "technology",
+            "pageSize": 5,
+            "apiKey": st.secrets.get("NEWS_API_KEY", "demo_key")
+        }
+    },
+    "weather_api": {
+        "url": "https://api.openweathermap.org/data/2.5/weather",
+        "params": {
+            "q": "Lagos,NG",
+            "appid": st.secrets.get("WEATHER_API_KEY", "demo_key"),
+            "units": "metric"
+        }
+    },
+    "financial_api": {
+        "url": "https://www.alphavantage.co/query",
+        "params": {
+            "function": "GLOBAL_QUOTE",
+            "symbol": "AAPL",
+            "apikey": st.secrets.get("ALPHA_VANTAGE_KEY", "demo_key")
+        }
+    },
+    "yah_tech_internal": {
+        "url": "http://localhost:8000"  # For internal FastAPI server
+    }
+}
+
+class APIClient:
+    def __init__(self):
+        self.session = None
+    
+    async def __aenter__(self):
+        self.session = aiohttp.ClientSession()
+        return self
+    
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        await self.session.close()
+    
+    async def make_async_request(self, url: str, params: dict = None, headers: dict = None):
+        """Make asynchronous API request"""
+        try:
+            async with self.session.get(url, params=params, headers=headers) as response:
+                if response.status == 200:
+                    return await response.json()
+                else:
+                    return {"error": f"API returned status code {response.status}"}
+        except Exception as e:
+            return {"error": str(e)}
+    
+    def make_sync_request(self, url: str, params: dict = None, headers: dict = None):
+        """Make synchronous API request"""
+        try:
+            response = requests.get(url, params=params, headers=headers, timeout=10)
+            if response.status_code == 200:
+                return response.json()
+            else:
+                return {"error": f"API returned status code {response.status_code}"}
+        except Exception as e:
+            return {"error": str(e)}
+
+class ExternalAPIService:
+    def __init__(self):
+        self.api_client = APIClient()
+    
+    async def get_tech_news(self):
+        """Get latest technology news"""
+        config = API_CONFIG["news_api"]
+        # For demo purposes, return mock data if no API key
+        if config["params"]["apiKey"] == "demo_key":
+            return {
+                "articles": [
+                    {"title": "AI Technology Advances in 2024", "description": "Latest breakthroughs in artificial intelligence..."},
+                    {"title": "Venture Studio Model Gains Popularity", "description": "More entrepreneurs adopting the venture studio approach..."},
+                    {"title": "Economic Impact of Tech Startups", "description": "How technology startups are reshaping economies..."}
+                ]
+            }
+        
+        result = await self.api_client.make_async_request(config["url"], config["params"])
+        return result
+    
+    async def get_weather(self, city: str = "Lagos"):
+        """Get weather information"""
+        config = API_CONFIG["weather_api"]
+        config["params"]["q"] = city
+        
+        if config["params"]["appid"] == "demo_key":
+            return {
+                "weather": [{"description": "clear sky"}],
+                "main": {"temp": 28, "humidity": 65},
+                "name": city
+            }
+        
+        result = await self.api_client.make_async_request(config["url"], config["params"])
+        return result
+    
+    async def get_stock_info(self, symbol: str = "AAPL"):
+        """Get stock market information"""
+        config = API_CONFIG["financial_api"]
+        config["params"]["symbol"] = symbol
+        
+        if config["params"]["apikey"] == "demo_key":
+            return {
+                "Global Quote": {
+                    "05. price": "185.50",
+                    "09. change": "+1.25",
+                    "10. change percent": "+0.68%"
+                }
+            }
+        
+        result = await self.api_client.make_async_request(config["url"], config["params"])
+        return result
+    
+    def get_crypto_data(self, coin: str = "bitcoin"):
+        """Get cryptocurrency data (synchronous for simplicity)"""
+        url = f"https://api.coingecko.com/api/v3/simple/price"
+        params = {
+            "ids": coin,
+            "vs_currencies": "usd",
+            "include_24hr_change": "true"
+        }
+        
+        result = self.api_client.make_sync_request(url, params)
+        return result
+
+# Knowledge base for the AI (updated with API capabilities)
+KNOWLEDGE_BASE = {
+    "personal": {
+        "who are you": f"I'm {PERSONAL_INFO['name']}, {PERSONAL_INFO['role']}.",
+        "what are your goals": f"My goals include: {', '.join(PERSONAL_INFO['goals'])}.",
+        "what do you do": f"I'm the CEO of YAH Tech, focused on {PERSONAL_INFO['goals'][2]}.",
+        "tell me about yourself": f"I'm {PERSONAL_INFO['name']}, {PERSONAL_INFO['role']}. My mission is to {PERSONAL_INFO['goals'][0].lower()} and {PERSONAL_INFO['goals'][1].lower()}.",
+    },
+    "company": {
+        "what is yah tech": f"{COMPANY_INFO['name']} is a {COMPANY_INFO['type']}. Our purpose is to {COMPANY_INFO['purpose'].lower()}.",
+        "what does yah tech do": f"We specialize in {COMPANY_INFO['primary_activity'].lower()}. {COMPANY_INFO['purpose']}.",
+        "company philosophy": f"Our philosophy is: '{COMPANY_INFO['philosophy']}'.",
+        "company stage": f"We're currently in the {COMPANY_INFO['stage']} phase.",
+        "venture studio": "As a venture studio, we don't just build apps - we build complete businesses with scalable systems and futuristic solutions.",
+    },
+    "api_capabilities": {
+        "what can you do": "I can fetch real-time data like tech news, weather, stock prices, and cryptocurrency data. I also have knowledge about YAH Tech and Adedoyin James.",
+        "get news": "I can fetch the latest technology news for you.",
+        "weather": "I can check weather information for any city.",
+        "stock": "I can get stock market data for companies.",
+        "crypto": "I can fetch cryptocurrency prices and changes.",
+    }
+}
+
+class YAHBot:
+    def __init__(self):
+        self.model_name = "google/flan-t5-small"
+        self.tokenizer = None
+        self.model = None
+        self.api_service = ExternalAPIService()
+        self._load_model()
+    
+    def _load_model(self):
+        """Load the Hugging Face model"""
+        try:
+            with st.spinner("🔄 Loading AI model..."):
+                self.tokenizer = AutoTokenizer.from_pretrained(
+                    "Adedoyinjames/YAH-Tech-Chat-Bot", 
+                    use_auth_token=HUGGINGFACE_TOKEN if HUGGINGFACE_TOKEN else True
+                )
+                self.model = AutoModelForSeq2SeqLM.from_pretrained(
+                    "Adedoyinjames/YAH-Tech-Chat-Bot",
+                    use_auth_token=HUGGINGFACE_TOKEN if HUGGINGFACE_TOKEN else True,
+                    torch_dtype=torch.float16 if torch.cuda.is_available() else torch.float32
+                )
+            st.success("✅ AI model loaded successfully!")
+        except Exception as e:
+            st.error(f"❌ Failed to load custom model: {e}")
+            st.info("🔄 Falling back to standard FLAN-T5-small model...")
+            try:
+                self.tokenizer = AutoTokenizer.from_pretrained(self.model_name)
+                self.model = AutoModelForSeq2SeqLM.from_pretrained(self.model_name)
+                st.success("✅ Standard model loaded successfully!")
+            except Exception as e2:
+                st.error(f"❌ Failed to load standard model: {e2}")
+    
+    def _get_knowledge_response(self, user_input):
+        """Check if the input matches any knowledge base entries"""
+        user_input_lower = user_input.lower()
+        
+        # Check all knowledge bases
+        for category in KNOWLEDGE_BASE.values():
+            for key, response in category.items():
+                if key in user_input_lower:
+                    return response
+        
+        # Specific patterns
+        if any(word in user_input_lower for word in ["adedoyin", "ifeoluwa", "james"]):
+            return KNOWLEDGE_BASE["personal"]["who are you"]
+        
+        if any(word in user_input_lower for word in ["yah tech", "your company"]):
+            return KNOWLEDGE_BASE["company"]["what is yah tech"]
+        
+        return None
+    
+    async def _handle_api_requests(self, user_input):
+        """Handle API-based queries"""
+        user_input_lower = user_input.lower()
+        
+        if any(word in user_input_lower for word in ["news", "headlines", "tech news"]):
+            news_data = await self.api_service.get_tech_news()
+            if "articles" in news_data:
+                articles = news_data["articles"][:3]  # Show top 3
+                response = "📰 **Latest Tech News:**\n\n"
+                for i, article in enumerate(articles, 1):
+                    response += f"{i}. **{article.get('title', 'No title')}**\n"
+                    response += f"   {article.get('description', 'No description available')}\n\n"
+                return response
+            else:
+                return "❌ Could not fetch news at the moment. Please try again later."
+        
+        elif any(word in user_input_lower for word in ["weather", "temperature"]):
+            # Extract city name if mentioned
+            city = "Lagos"
+            for word in user_input_lower.split():
+                if word not in ["weather", "temperature", "what's", "the"] and len(word) > 2:
+                    city = word.capitalize()
+                    break
+            
+            weather_data = await self.api_service.get_weather(city)
+            if "weather" in weather_data:
+                temp = weather_data["main"]["temp"]
+                desc = weather_data["weather"][0]["description"]
+                humidity = weather_data["main"]["humidity"]
+                return f"🌤️ **Weather in {weather_data.get('name', city)}:**\n\nTemperature: {temp}°C\nConditions: {desc}\nHumidity: {humidity}%"
+            else:
+                return "❌ Could not fetch weather data. Please try again later."
+        
+        elif any(word in user_input_lower for word in ["stock", "price", "apple", "aapl"]):
+            stock_data = await self.api_service.get_stock_info()
+            if "Global Quote" in stock_data:
+                quote = stock_data["Global Quote"]
+                price = quote.get("05. price", "N/A")
+                change = quote.get("09. change", "N/A")
+                change_percent = quote.get("10. change percent", "N/A")
+                return f"📈 **Apple Stock (AAPL):**\n\nPrice: ${price}\nChange: {change} ({change_percent})"
+            else:
+                return "❌ Could not fetch stock data. Please try again later."
+        
+        elif any(word in user_input_lower for word in ["crypto", "bitcoin", "ethereum"]):
+            crypto_data = self.api_service.get_crypto_data()
+            if "bitcoin" in crypto_data:
+                btc_data = crypto_data["bitcoin"]
+                price = btc_data.get("usd", "N/A")
+                change = btc_data.get("usd_24h_change", "N/A")
+                return f"₿ **Bitcoin Price:**\n\nCurrent: ${price:,.2f}\n24h Change: {change:.2f}%"
+            else:
+                return "❌ Could not fetch cryptocurrency data. Please try again later."
+        
+        return None
+    
+    async def generate_response(self, user_input, conversation_history):
+        """Generate response using knowledge base, API calls, or AI model"""
+        # First check knowledge base
+        knowledge_response = self._get_knowledge_response(user_input)
+        if knowledge_response:
+            return knowledge_response
+        
+        # Check for API-based queries
+        api_response = await self._handle_api_requests(user_input)
+        if api_response:
+            return api_response
+        
+        # Use AI model for other queries
+        if self.model and self.tokenizer:
+            try:
+                # Prepare input with context
+                context = "You are YAH Bot, an AI assistant for YAH Tech. Be helpful, professional and focus on technology, entrepreneurship, and business development."
+                full_input = f"{context}\n\nConversation:\n" + "\n".join([f"Human: {msg['content']}\nAI: {msg['content']}" 
+                                                                        for msg in conversation_history[-3:]]) + f"\nHuman: {user_input}\nAI:"
+                
+                inputs = self.tokenizer.encode(full_input, return_tensors="pt", max_length=512, truncation=True)
+                
+                with torch.no_grad():
+                    outputs = self.model.generate(
+                        inputs,
+                        max_length=150,
+                        num_return_sequences=1,
+                        temperature=0.7,
+                        do_sample=True,
+                        pad_token_id=self.tokenizer.eos_token_id
+                    )
+                
+                response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
+                return response
+                
+            except Exception as e:
+                return f"I encountered an error: {str(e)}. However, I can tell you that at YAH Tech, we're focused on building scalable systems that drive economic growth."
+        
+        # Fallback responses
+        fallback_responses = [
+            "That's an interesting question! At YAH Tech, we believe in learning, understanding, and creating solutions that matter.",
+            "I'd be happy to discuss that. As a venture studio, we focus on building technology-driven ventures that create both profit and societal value.",
+            "Great question! Our approach at YAH Tech is to develop futuristic solutions through careful evaluation and systematic creation.",
+            f"From my perspective as {PERSONAL_INFO['name']}, I focus on building systems that reshape the economic landscape through technology."
+        ]
+        return random.choice(fallback_responses)
+
+def initialize_session_state():
+    """Initialize session state variables"""
+    if "messages" not in st.session_state:
+        st.session_state.messages = []
+    if "bot" not in st.session_state:
+        st.session_state.bot = YAHBot()
+    if "conversation_started" not in st.session_state:
+        st.session_state.conversation_started = False
+    if "api_status" not in st.session_state:
+        st.session_state.api_status = {}
+
+def display_api_status():
+    """Display API status in sidebar"""
+    st.sidebar.subheader("API Status")
+    
+    # Mock API status for demo
+    statuses = {
+        "News API": "✅ Connected" if API_CONFIG["news_api"]["params"]["apiKey"] != "demo_key" else "⚠️ Demo Mode",
+        "Weather API": "✅ Connected" if API_CONFIG["weather_api"]["params"]["appid"] != "demo_key" else "⚠️ Demo Mode",
+        "Financial API": "✅ Connected" if API_CONFIG["financial_api"]["params"]["apikey"] != "demo_key" else "⚠️ Demo Mode",
+        "Crypto API": "✅ Connected",
+        "Internal API": "🔌 Not Connected"  # FastAPI server would be separate
+    }
+    
+    for api, status in statuses.items():
+        color = "🟢" if "✅" in status else "🟡" if "⚠️" in status else "🔴"
+        st.sidebar.write(f"{color} {api}: {status}")
+
+def main():
+    # Page configuration
+    st.set_page_config(
+        page_title="YAH Tech Chatbot",
+        page_icon="🤖",
+        layout="centered"
+    )
+    
+    # Initialize session state
+    initialize_session_state()
+    
+    # Header
+    col1, col2 = st.columns([1, 4])
+    with col1:
+        st.image("https://via.placeholder.com/80/2196f3/ffffff?text=YT", width=80)
+    with col2:
+        st.title("YAH Tech Assistant")
+        st.caption("Powered by AI • Built for Innovation • API-Enabled")
+    
+    st.markdown("---")
+    
+    # Sidebar
+    with st.sidebar:
+        st.header("About YAH Tech")
+        st.markdown(f"""
+        **Founder**: {PERSONAL_INFO['name']}  
+        **Role**: {PERSONAL_INFO['role']}  
+        
+        **Company**: {COMPANY_INFO['name']}  
+        **Focus**: {COMPANY_INFO['type']}  
+        **Stage**: {COMPANY_INFO['stage']}
+        
+        **Philosophy**: *{COMPANY_INFO['philosophy']}*
+        """)
+        
+        # API Status
+        display_api_status()
+        
+        st.markdown("---")
+        st.subheader("API Capabilities")
+        st.markdown("""
+        Ask me to:
+        - **Get tech news** 📰
+        - **Check weather** 🌤️
+        - **Stock prices** 📈
+        - **Crypto data** ₿
+        - **YAH Tech info** 🏢
+        - **Business advice** 💼
+        """)
+        
+        st.markdown("---")
+        st.subheader("Quick Actions")
+        if st.button("🔄 Test APIs"):
+            st.info("API testing feature would be implemented here")
+        
+        if st.button("🗑️ Clear Chat"):
+            st.session_state.messages = []
+            st.session_state.conversation_started = False
+            st.rerun()
+    
+    # Chat interface
+    if not st.session_state.conversation_started:
+        st.info(f"👋 Hello! I'm YAH Bot, your AI assistant for everything related to {PERSONAL_INFO['name']} and {COMPANY_INFO['name']}. I can also fetch real-time data through APIs! How can I help you today?")
+        st.session_state.conversation_started = True
+    
+    # Display chat messages
+    for message in st.session_state.messages:
+        with st.chat_message(message["role"]):
+            st.markdown(message["content"])
+    
+    # Chat input
+    if prompt := st.chat_input("Ask me about YAH Tech, get news, check weather, or anything else..."):
+        # Add user message to chat history
+        st.session_state.messages.append({"role": "user", "content": prompt})
+        with st.chat_message("user"):
+            st.markdown(prompt)
+        
+        # Generate and display assistant response
+        with st.chat_message("assistant"):
+            with st.spinner("🤔 Thinking..."):
+                # Use asyncio to handle async API calls
+                response = asyncio.run(st.session_state.bot.generate_response(prompt, st.session_state.messages))
+                st.markdown(response)
+        
+        st.session_state.messages.append({"role": "assistant", "content": response})
+
+if __name__ == "__main__":
+    main()