import streamlit as st from groq import Groq import requests import re import numpy as np from bs4 import BeautifulSoup import PyPDF2 import docx from io import StringIO import csv import json from datetime import datetime import pytz import time import hashlib from collections import defaultdict # ============================================ # FILE PROCESSING FUNCTIONS # ============================================ def extract_text_from_pdf(file): try: file.seek(0) pdf_reader = PyPDF2.PdfReader(file) text = "" for page_num, page in enumerate(pdf_reader.pages): page_text = page.extract_text() if page_text and page_text.strip(): text += f"\n--- Page {page_num + 1} ---\n" text += page_text.strip() + "\n" return text[:5000] if text.strip() else "No extractable text in PDF" except Exception as e: return f"Error reading PDF: {str(e)}" def extract_text_from_docx(file): try: file.seek(0) doc = docx.Document(file) text = "" for para in doc.paragraphs: if para.text and para.text.strip(): text += para.text.strip() + "\n\n" for table in doc.tables: for row in table.rows: row_text = [] for cell in row.cells: if cell.text and cell.text.strip(): row_text.append(cell.text.strip()) if row_text: text += " | ".join(row_text) + "\n" return text[:5000] if text.strip() else "No extractable text in document" except Exception as e: return f"Error reading Word document: {str(e)}" def extract_text_from_txt(file): try: file.seek(0) content = file.read().decode('utf-8') return content[:5000] if content.strip() else "File is empty" except UnicodeDecodeError: try: file.seek(0) content = file.read().decode('latin-1') return content[:5000] except: return "Error decoding text file" except Exception as e: return f"Error reading text file: {str(e)}" def extract_text_from_csv(file): try: file.seek(0) content = file.read().decode('utf-8') csv_reader = csv.reader(StringIO(content)) text = "CSV Data:\n\n" rows = list(csv_reader) if rows: text += "Headers: " + " | ".join(rows[0]) + "\n\n" for i, row in enumerate(rows[1:11], 1): text += f"Row {i}: " + " | ".join(row) + "\n" if len(rows) > 11: text += f"\n... and {len(rows) - 11} more rows" return text[:5000] if text.strip() else "CSV file appears empty" except Exception as e: return f"Error reading CSV: {str(e)}" def extract_text_from_json(file): try: file.seek(0) content = file.read().decode('utf-8') data = json.loads(content) formatted = json.dumps(data, indent=2) if len(formatted) > 3000: text = "JSON Data Summary:\n\n" text += f"Type: {type(data).__name__}\n" if isinstance(data, dict): text += f"Keys: {', '.join(list(data.keys())[:10])}\n" elif isinstance(data, list): text += f"Length: {len(data)}\n" text += "\nFull JSON (truncated):\n" + formatted[:3000] else: text = formatted return text[:5000] except Exception as e: return f"Error reading JSON: {str(e)}" def process_uploaded_file(uploaded_file): file_type = uploaded_file.type file_name = uploaded_file.name.lower() if file_type == "application/pdf" or file_name.endswith('.pdf'): return extract_text_from_pdf(uploaded_file) elif file_type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document" or file_name.endswith('.docx'): return extract_text_from_docx(uploaded_file) elif file_type == "text/plain" or file_name.endswith('.txt'): return extract_text_from_txt(uploaded_file) elif file_type == "text/csv" or file_name.endswith('.csv'): return extract_text_from_csv(uploaded_file) elif file_type == "application/json" or file_name.endswith('.json'): return extract_text_from_json(uploaded_file) else: return f"Unsupported file type: {file_type}" # ============================================ # CONFIG # ============================================ TEMPERATURE = 0 MAX_TOKENS = 800 st.set_page_config(page_title="MozeAI", page_icon="šŸ§ ", layout="wide") # ============================================ # CSS - FIXED CHAT INPUT AT BOTTOM # ============================================ st.markdown(""" """, unsafe_allow_html=True) # ============================================ # GROQ CLIENT # ============================================ client = Groq(api_key=st.secrets.get("GROQ_API_KEY")) def get_current_datetime(): tz = pytz.timezone('Asia/Seoul') now = datetime.now(tz) return f"""Current Information: ā€¢ Date: {now.strftime('%B %d, %Y')} ā€¢ Time: {now.strftime('%I:%M %p')} ā€¢ Day: {now.strftime('%A')} ā€¢ Timezone: Asia/Seoul""" # ============================================ # LIGHTWEIGHT MEMORY SYSTEM (No torch needed) # ============================================ class LightweightMemory: def __init__(self): self.memories = [] self.keyword_index = defaultdict(list) def add_memory(self, text, metadata=None): if len(text) < 50: return words = set(re.findall(r'\b[a-z]{3,}\b', text.lower())) memory = { "text": text, "keywords": words, "metadata": metadata or {}, "timestamp": time.time() } self.memories.append(memory) for word in words: self.keyword_index[word].append(len(self.memories) - 1) if len(self.memories) > 50: self.memories = self.memories[-50:] self._rebuild_index() def _rebuild_index(self): self.keyword_index = defaultdict(list) for idx, memory in enumerate(self.memories): for word in memory["keywords"]: self.keyword_index[word].append(idx) def retrieve(self, query, top_k=3): if not self.memories: return [] query_words = set(re.findall(r'\b[a-z]{3,}\b', query.lower())) scored = [] for idx, memory in enumerate(self.memories): matches = len(query_words & memory["keywords"]) if matches > 0: scored.append((memory["text"], matches)) scored.sort(key=lambda x: x[1], reverse=True) return [text for text, score in scored[:top_k]] def get_context(self, query): results = self.retrieve(query) if results: context = "RELEVANT PAST CONVERSATIONS:\n\n" for i, result in enumerate(results): context += f"[{i+1}] {result}\n\n" return context return "" # Initialize memory memory = LightweightMemory() def store_memory(text): memory.add_memory(text) def retrieve_memory(query): return memory.get_context(query) # ============================================ # LLM FUNCTION WITH MULTIPLE MODEL FALLBACKS # ============================================ def llm_with_fallback(messages, max_retries=2): models_to_try = [ "llama-3.3-70b-versatile", "llama-3.1-70b-versatile", "mixtral-8x7b-32768", "llama-3.1-8b-instant", "gemma2-9b-it" ] for model in models_to_try: for attempt in range(max_retries): try: completion = client.chat.completions.create( model=model, temperature=TEMPERATURE, max_tokens=MAX_TOKENS, messages=messages, timeout=30 ) st.session_state.last_model_used = model return completion.choices[0].message.content.strip() except Exception as e: if attempt < max_retries - 1: time.sleep(2 ** attempt) continue return "AI service temporarily unavailable. Please try again." def llm(messages): return llm_with_fallback(messages) # ============================================ # SYSTEM PROMPT # ============================================ SYSTEM_PROMPT = """ You are MozeAI, an advanced AI assistant with REAL-TIME internet access and file analysis capabilities. CREATOR INFORMATION (ONLY mention when asked directly): - Created by Mukiibi Moses, a Computer Engineering student at Kyungdong University, South Korea. YOUR CAPABILITIES: - REAL-TIME web search for current information - File analysis for PDF, DOCX, TXT, CSV, JSON files - File comparison (compare multiple documents) - Memory of past conversations - Image generation and editing - Calculator and news **CRITICAL RULES FOR QUESTIONS:** 1. When a user ASKS "can you generate images?" or "do you generate images?" ā†’ ANSWER WITH TEXT: "Yes, I can generate images! Just tell me what you want, for example: 'generate image of a cat' or 'draw a beautiful sunset'" 2. When a user SAYS "generate image of X" or "draw Y" ā†’ THEN generate the image 3. Questions about your capabilities should NEVER trigger actual image generation 4. Only generate an image when the user explicitly COMMANDS you to, not when they ASK if you can CRITICAL RULES: 1. For questions about PEOPLE, PLACES, EVENTS, or ANY topic not related to your creator, USE SEARCH RESULTS 2. ONLY mention your creator when specifically asked 3. Use the conversation history and retrieved memory for context 4. Answer concisely and accurately 5. Be conversational and friendly """ # ============================================ # SESSION STATE # ============================================ if "chat_history" not in st.session_state: st.session_state.chat_history = [] if "uploaded_files" not in st.session_state: st.session_state.uploaded_files = {} if "file_context" not in st.session_state: st.session_state.file_context = "" if "last_search_query" not in st.session_state: st.session_state.last_search_query = None if "last_search_results" not in st.session_state: st.session_state.last_search_results = None if "last_response" not in st.session_state: st.session_state.last_response = None if "last_topic" not in st.session_state: st.session_state.last_topic = None if "last_image_prompt" not in st.session_state: st.session_state.last_image_prompt = None if "code_search_cache" not in st.session_state: st.session_state.code_search_cache = {} if "is_resetting" not in st.session_state: st.session_state.is_resetting = False if "last_model_used" not in st.session_state: st.session_state.last_model_used = None # ============================================ # SEARCH FUNCTIONS # ============================================ def internet_search(query): try: clean_query = query.strip() url = "https://html.duckduckgo.com/html/" params = {"q": clean_query} headers = {"User-Agent": "Mozilla/5.0"} response = requests.post(url, data=params, headers=headers, timeout=10) if response.status_code == 200: results = re.findall(r'([^<]+)', response.text) snippets = re.findall(r']*>([^<]+)', response.text) if results: context = f"SEARCH RESULTS for '{clean_query}':\n\n" for i in range(min(3, len(results))): context += f"ā€¢ {results[i]}\n" if i < len(snippets): snippet = re.sub(r'<[^>]+>', '', snippets[i]) context += f" {snippet[:300]}...\n\n" return context[:2000] return "" except: return "" def get_current_news(): try: url = "https://rss2json.com/api.json?rss_url=https://feeds.bbci.co.uk/news/rss.xml" response = requests.get(url, timeout=10) if response.status_code == 200: data = response.json() items = data.get("items", [])[:3] news_text = "LATEST NEWS HEADLINES:\n\n" for item in items: news_text += f"ā€¢ {item.get('title', '')}\n" news_text += f" {item.get('description', '')[:150]}...\n\n" return news_text[:1000] except: pass return "" # ============================================ # CALCULATOR # ============================================ def calculator(query): try: expression = query.lower() expression = expression.replace("Ɨ", "*").replace("x", "*") numbers = re.findall(r"[0-9\+\-\*\/\.\(\) ]+", expression) if numbers: result = eval(numbers[0]) return str(result) except: return None # ============================================ # WEB SCRAPING # ============================================ def scrape_webpage(url): try: headers = {"User-Agent": "Mozilla/5.0"} response = requests.get(url, headers=headers, timeout=15) if response.status_code == 200: soup = BeautifulSoup(response.content, 'html.parser') for element in soup(["script", "style", "nav", "footer"]): element.decompose() text = soup.get_text() lines = (line.strip() for line in text.splitlines()) text = ' '.join(line for line in lines if line) return text[:3000] if len(text) > 200 else None except: pass return None def extract_urls_from_query(query): url_pattern = r'https?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+[^\s]*' return re.findall(url_pattern, query) # ============================================ # CODING SEARCH FUNCTIONS # ============================================ def search_coding_solution(query): search_queries = [ f"{query} stack overflow", f"{query} example code", f"{query} best practice", f"{query} github" ] all_results = "" for search_q in search_queries[:2]: result = internet_search(search_q) if result: all_results += result + "\n\n" return all_results def search_coding_solution_cached(query): cache_key = query.lower().strip() if cache_key in st.session_state.code_search_cache: return st.session_state.code_search_cache[cache_key] result = search_coding_solution(query) st.session_state.code_search_cache[cache_key] = result return result def coding_assistant_with_search(query, context=""): with st.spinner("šŸ” Searching the internet for the best solution..."): search_results = search_coding_solution_cached(query) coding_prompt = f""" You are an expert programmer. Generate the best possible code based on the user's request. USER REQUEST: {query} ## INTERNET SEARCH RESULTS (Use these as reference): {search_results[:3000]} ## REQUIREMENTS: - Code must be complete and runnable - Include all imports - Add comments - Handle edge cases Generate the best possible code now: """ messages = [ {"role": "system", "content": "You are an expert programming assistant. Use search results to find the best solution."}, {"role": "user", "content": coding_prompt} ] return clean_answer(llm(messages)) # ============================================ # ROUTER FUNCTION # ============================================ def route(query): q = query.lower() if extract_urls_from_query(query): return "scrape_url" file_keywords = ["document", "file", "upload", "pdf", "docx", "txt", "csv", "json", "what is this", "summarize"] if any(x in q for x in file_keywords): return "file_task" comparison_keywords = ["compare", "comparison", "difference", "similarities"] if any(x in q for x in comparison_keywords): return "compare_files" if any(x in q for x in ["generate image", "create image", "draw", "make an image", "picture of", "image of"]): return "generate_image" if any(x in q for x in ["make it", "make the", "change it", "change the", "turn it", "add a", "remove", "edit image", "modify image"]): return "edit_image" if any(x in q for x in ["who is", "tell me about", "what is", "weather", "temperature", "rain", "snow", "news", "headlines"]): return "search" if any(x in q for x in ["+", "-", "*", "/", "calculate"]): return "calculator" if any(x in q for x in ["time", "date", "today"]): return "datetime" coding_keywords = ["code", "python", "javascript", "html", "css", "react", "tkinter", "function", "class", "import", "algorithm", "debug", "fix", "write a program", "create a script"] if any(x in q for x in coding_keywords): return "coding_with_search" factual_keywords = ["president", "current", "elected", "prime minister", "leader", "ceo of"] if any(x in q for x in factual_keywords): return "search" return "reason" # ============================================ # CLEAN ANSWER # ============================================ def clean_answer(text): text = text.split("šŸ§ ")[0] text = text.split("Plan:")[0] text = text.split("Thinking:")[0] return text.strip() # ============================================ # REASONING FUNCTION # ============================================ def reason(question, context): memory_context = retrieve_memory(question) enhanced_context = context if memory_context: enhanced_context += "\n" + memory_context history_text = "" if st.session_state.chat_history: history_text = "PREVIOUS CONVERSATION:\n" last_exchanges = st.session_state.chat_history[-8:] if len(st.session_state.chat_history) > 8 else st.session_state.chat_history for role, msg in last_exchanges: if role == "user": history_text += f"User: {msg}\n" else: history_text += f"Assistant: {msg}\n" history_text += "\n" messages = [ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": f""" {history_text} {enhanced_context[:3000]} USER QUESTION: {question} ANSWER: """} ] return clean_answer(llm(messages)) # ============================================ # FILE FUNCTIONS # ============================================ def compare_files(query, file_context, filenames): prompt = f"Files: {filenames}\n\nContent: {file_context[:4000]}\n\nQuestion: {query}\n\nCompare the files." messages = [{"role": "system", "content": "You compare files."}, {"role": "user", "content": prompt}] return clean_answer(llm(messages)) def analyze_uploaded_files(query, file_context, filenames): prompt = f"Files: {filenames}\n\nContent: {file_context[:6000]}\n\nQuestion: {query}\n\nAnswer based on file content." messages = [{"role": "system", "content": "You analyze files."}, {"role": "user", "content": prompt}] return clean_answer(llm(messages)) def evaluate_work(question, file_context): prompt = f"Content: {file_context[:3000]}\n\nRequest: {question}\n\nProvide assessment." messages = [{"role": "system", "content": "You evaluate work."}, {"role": "user", "content": prompt}] return clean_answer(llm(messages)) # ============================================ # IMAGE GENERATION FUNCTIONS # ============================================ def generate_image(prompt): try: encoded_prompt = requests.utils.quote(prompt) timestamp = int(time.time()) image_url = f"https://image.pollinations.ai/prompt/{encoded_prompt}?width=1024&height=1024&nologo=true&_={timestamp}" return image_url except: return None def generate_and_display_image(prompt, is_edit=False): image_url = generate_image(prompt) if image_url: if is_edit: return f"šŸŽØ **Edited Image - New Prompt:** '{prompt}'\n\n![Generated Image]({image_url})\n\n*Image generated by AI*" else: return f"šŸŽØ **Generated Image for:** '{prompt}'\n\n![Generated Image]({image_url})\n\n*Image generated by AI*" else: return "āŒ Sorry, I couldn't generate an image right now." # ============================================ # RUN AGENT FUNCTION # ============================================ def run_agent(query): q = query.lower().strip() reset_phrases = ["leave the document", "clear context", "forget the file", "start fresh", "clear files", "new chat"] if any(phrase in q for phrase in reset_phrases): st.session_state.file_context = "" st.session_state.uploaded_files = {} st.session_state.last_search_query = None st.session_state.last_search_results = None st.session_state.last_topic = None st.session_state.last_image_prompt = None return "āœ… Context cleared! How can I help you today?" # ============================================ # DIRECT RESPONSE FOR CAPABILITY QUESTIONS # ============================================ if q in ["can you generate images", "do you generate images", "can you create images", "can you draw", "are you able to generate images"]: return "Yes, I can generate images! Just tell me what you want, for example: 'generate image of a cat' or 'draw a beautiful sunset'" # ============================================ # DIRECT EDIT CHECK - MUST BE BEFORE ROUTER # ============================================ edit_triggers = ["make it", "make the", "change it", "change the", "turn it", "add a", "remove"] if any(phrase in q for phrase in edit_triggers) and st.session_state.get("last_image_prompt"): last_prompt = st.session_state.last_image_prompt # Extract edit instruction edit_text = query for word in edit_triggers: if word in edit_text.lower(): edit_text = re.sub(re.escape(word), "", edit_text.lower(), flags=re.IGNORECASE).strip() break edit_text = ' '.join(edit_text.split()) new_prompt = f"{last_prompt}, {edit_text}" st.session_state.last_image_prompt = new_prompt with st.spinner("Editing image..."): return generate_and_display_image(new_prompt, is_edit=True) # ============================================ # DIRECT RESPONSES # ============================================ if any(phrase in q for phrase in ["who are you", "who is this", "what are you"]): return "I'm MozeAI, your AI assistant! Created by Mukiibi Moses, a Computer Engineering student at Kyungdong University. I can search the web, analyze files, generate images, and help with coding. What can I do for you?" if any(phrase in q for phrase in ["mukiibi moses", "who is moses", "your maker", "your creator", "who created you"]): return "Mukiibi Moses is my creator, a Computer Engineering student at Kyungdong University in South Korea, specializing in AI and machine learning. His portfolio: https://moze12432.github.io/" tool = route(query) context = "" follow_up = ["tell me more", "more about", "continue", "elaborate"] is_follow_up = any(phrase in q for phrase in follow_up) if is_follow_up: if st.session_state.last_search_results: context = st.session_state.last_search_results context += "\n\n Provide MORE information." elif st.session_state.last_response: context = f"Previous: {st.session_state.last_response[:500]}\n\nContinue." else: context = get_current_datetime() if tool == "compare_files" and st.session_state.file_context and len(st.session_state.uploaded_files) >= 2: filenames = "\n".join(st.session_state.uploaded_files.keys()) with st.spinner("Comparing files..."): response = compare_files(query, st.session_state.file_context, filenames) st.session_state.last_response = response store_memory(response) return response elif tool == "file_task" and st.session_state.file_context: filenames = "\n".join(st.session_state.uploaded_files.keys()) with st.spinner("Reading files..."): response = analyze_uploaded_files(query, st.session_state.file_context, filenames) st.session_state.last_response = response store_memory(response) return response elif tool == "evaluate" and st.session_state.file_context: with st.spinner("Evaluating..."): response = evaluate_work(query, st.session_state.file_context) st.session_state.last_response = response store_memory(response) return response elif tool == "scrape_url": urls = extract_urls_from_query(query) scraped = "" for url in urls: content = scrape_webpage(url) if content: scraped += f"\nContent from {url}:\n{content}\n" if scraped: context = scraped else: return "I couldn't read that link." elif tool == "calculator": result = calculator(query) if result: return f"Result: {result}" elif tool == "datetime": context += get_current_datetime() elif tool == "generate_image": with st.spinner("Generating image..."): image_prompt = query for phrase in ["generate image of", "generate an image of", "create image of", "draw a", "picture of", "image of"]: if phrase in image_prompt.lower(): image_prompt = re.sub(re.escape(phrase), "", image_prompt.lower(), flags=re.IGNORECASE).strip() break image_prompt = ' '.join(image_prompt.split()) if not image_prompt: image_prompt = query st.session_state.last_image_prompt = image_prompt return generate_and_display_image(image_prompt) elif tool == "edit_image": with st.spinner("Editing image..."): last_prompt = st.session_state.get("last_image_prompt", "") if not last_prompt: return "No previous image. Generate one first with 'generate image of...'" edit_text = query for word in ["make it", "change it", "turn it", "add a", "edit image"]: if word in edit_text.lower(): edit_text = re.sub(re.escape(word), "", edit_text.lower(), flags=re.IGNORECASE).strip() break edit_text = ' '.join(edit_text.split()) new_prompt = f"{last_prompt}, {edit_text}" st.session_state.last_image_prompt = new_prompt return generate_and_display_image(new_prompt, is_edit=True) elif tool == "coding_with_search": response = coding_assistant_with_search(query) st.session_state.last_response = response store_memory(response) return response else: search_result = internet_search(query) if search_result: context += "\n" + search_result st.session_state.last_search_query = query st.session_state.last_search_results = search_result else: context += get_current_datetime() answer = reason(query, context) st.session_state.last_response = answer if not is_follow_up: store_memory(answer) return answer # ============================================ # UI - MAIN DISPLAY # ============================================ st.markdown('

šŸ§  MozeAI

', unsafe_allow_html=True) st.markdown('

Intelligent AI Assistant

', unsafe_allow_html=True) st.markdown("---") with st.sidebar: st.markdown("### šŸ§  MozeAI") st.markdown("---") if st.button("šŸ”„ New Chat", key="new_chat_btn", use_container_width=True): if not st.session_state.get("is_resetting", False): st.session_state.is_resetting = True st.session_state.chat_history = [] st.session_state.uploaded_files = {} st.session_state.file_context = "" st.session_state.last_image_prompt = None st.session_state.last_search_query = None st.session_state.last_search_results = None st.session_state.last_response = None st.session_state.code_search_cache = {} st.session_state.is_resetting = False st.success("New chat started!") st.rerun() if st.button("šŸ—‘ļø Clear Files", key="clear_files_btn", use_container_width=True): st.session_state.uploaded_files = {} st.session_state.file_context = "" st.success("Files cleared!") st.rerun() st.markdown("---") st.markdown("### šŸ“¤ Upload Files") uploaded_files = st.file_uploader( "Choose files", type=['pdf', 'docx', 'txt', 'csv', 'json'], accept_multiple_files=True, key="sidebar_uploader", label_visibility="collapsed" ) if uploaded_files: for file in uploaded_files: if file.name not in st.session_state.uploaded_files: with st.spinner(f"Processing {file.name}..."): content = process_uploaded_file(file) if content and not content.startswith("Error"): st.session_state.uploaded_files[file.name] = content st.success(f"āœ… {file.name}") if st.session_state.uploaded_files: parts = [] for name, content in st.session_state.uploaded_files.items(): parts.append(f"\n{'='*50}\nšŸ“„ {name}\n{'='*50}\n{content}\n") st.session_state.file_context = "\n".join(parts) st.info(f"šŸ“ {len(st.session_state.uploaded_files)} file(s) loaded") st.markdown("---") st.markdown("### šŸŽØ Image Generation") st.markdown("**Generate:** `generate image of a cat`") st.markdown("**Edit:** `make it black` or `add a hat`") st.markdown("---") st.markdown("### ā„¹ļø About") st.markdown("**Creator:** Mukiibi Moses") st.markdown("**University:** Kyungdong University, South Korea") if st.session_state.last_model_used: st.caption(f"Model: {st.session_state.last_model_used}") # ============================================ # CHAT DISPLAY # ============================================ for role, msg in st.session_state.chat_history: with st.chat_message(role): st.write(msg) query = st.chat_input("Ask me anything - generate images, search the web, analyze files, code, and more!") if query: st.session_state.chat_history.append(("user", query)) with st.chat_message("user"): st.write(query) response = run_agent(query) with st.chat_message("assistant"): st.write(response) st.session_state.chat_history.append(("assistant", response)) st.rerun()