Create app.py

app.py

@@ -0,0 +1,565 @@
+import gradio as gr
+import json
+import os
+import time
+from datetime import datetime
+from typing import List, Dict, Any, Optional, Tuple
+import tempfile
+import base64
+from pathlib import Path
+
+# Core dependencies
+try:
+    from together import Together
+    import PyPDF2
+    from PIL import Image
+    import speech_recognition as sr
+    import io
+    import subprocess
+    import sys
+except ImportError as e:
+    print(f"Missing dependency: {e}")
+    print("Install with: pip install together PyPDF2 pillow speechrecognition pyaudio")
+    sys.exit(1)
+
+class ConversationMemory:
+    """Manages conversation context and memory across sessions"""
+    
+    def __init__(self):
+        self.conversations = []
+        self.context_graph = {}
+        self.session_data = {}
+    
+    def add_interaction(self, input_type: str, content: str, response: str, metadata: Dict = None):
+        interaction = {
+            "timestamp": datetime.now().isoformat(),
+            "input_type": input_type,
+            "content": content[:500] + "..." if len(content) > 500 else content,  # Truncate for memory
+            "response": response[:1000] + "..." if len(response) > 1000 else response,
+            "metadata": metadata or {}
+        }
+        self.conversations.append(interaction)
+    
+    def get_relevant_context(self, query: str, limit: int = 3) -> List[Dict]:
+        # Simple relevance scoring - in production, use embeddings
+        relevant = []
+        query_lower = query.lower()
+        
+        for conv in reversed(self.conversations[-10:]):  # Check last 10 interactions
+            score = 0
+            content_lower = conv["content"].lower()
+            response_lower = conv["response"].lower()
+            
+            # Simple keyword matching
+            for word in query_lower.split():
+                if len(word) > 3:  # Skip short words
+                    if word in content_lower or word in response_lower:
+                        score += 1
+            
+            if score > 0:
+                relevant.append((score, conv))
+        
+        # Sort by relevance and return top results
+        relevant.sort(key=lambda x: x[0], reverse=True)
+        return [conv for score, conv in relevant[:limit]]
+
+class NexusAI:
+    """Main AI processing class"""
+    
+    def __init__(self, api_key: str = None):
+        self.api_key = api_key
+        self.client = None
+        self.memory = ConversationMemory()
+        
+        if api_key:
+            self.initialize_client(api_key)
+    
+    def initialize_client(self, api_key: str):
+        """Initialize Together AI client"""
+        try:
+            self.client = Together(api_key=api_key)
+            self.api_key = api_key
+            return True, "API key initialized successfully!"
+        except Exception as e:
+            return False, f"Failed to initialize API key: {str(e)}"
+    
+    def extract_text_from_pdf(self, pdf_path: str) -> str:
+        """Extract text from PDF file"""
+        try:
+            with open(pdf_path, 'rb') as file:
+                pdf_reader = PyPDF2.PdfReader(file)
+                text = ""
+                for page in pdf_reader.pages:
+                    text += page.extract_text() + "\n"
+                return text.strip()
+        except Exception as e:
+            return f"Error reading PDF: {str(e)}"
+    
+    def analyze_image(self, image_path: str) -> str:
+        """Analyze image and return description"""
+        try:
+            with Image.open(image_path) as img:
+                # Basic image analysis - in production, use vision models
+                width, height = img.size
+                mode = img.mode
+                format_type = img.format
+                
+                description = f"Image Analysis:\n"
+                description += f"- Dimensions: {width}x{height} pixels\n"
+                description += f"- Color mode: {mode}\n"
+                description += f"- Format: {format_type}\n"
+                
+                # Simple color analysis
+                if mode == "RGB":
+                    # Get dominant colors (simplified)
+                    img_small = img.resize((50, 50))
+                    colors = img_small.getcolors(2500)
+                    if colors:
+                        dominant_color = max(colors, key=lambda x: x[0])[1]
+                        description += f"- Dominant color (RGB): {dominant_color}\n"
+                
+                return description
+        except Exception as e:
+            return f"Error analyzing image: {str(e)}"
+    
+    def transcribe_audio(self, audio_path: str) -> str:
+        """Transcribe audio to text"""
+        try:
+            r = sr.Recognizer()
+            with sr.AudioFile(audio_path) as source:
+                audio_data = r.record(source)
+                text = r.recognize_google(audio_data)
+                return text
+        except Exception as e:
+            return f"Error transcribing audio: {str(e)}"
+    
+    def execute_code(self, code: str, language: str = "python") -> str:
+        """Execute code safely (basic implementation)"""
+        try:
+            if language.lower() == "python":
+                # Create a temporary file
+                with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
+                    f.write(code)
+                    temp_file = f.name
+                
+                # Execute with timeout
+                try:
+                    result = subprocess.run([sys.executable, temp_file], 
+                                          capture_output=True, text=True, timeout=10)
+                    output = result.stdout
+                    if result.stderr:
+                        output += f"\nErrors:\n{result.stderr}"
+                    return output
+                except subprocess.TimeoutExpired:
+                    return "Code execution timed out (10s limit)"
+                finally:
+                    os.unlink(temp_file)
+            else:
+                return f"Language '{language}' not supported yet. Only Python is available."
+        except Exception as e:
+            return f"Error executing code: {str(e)}"
+    
+    def build_context_messages(self, user_input: str, input_type: str, extracted_content: str = "") -> List[Dict]:
+        """Build context messages for the AI model"""
+        messages = []
+        
+        # Add system message
+        system_msg = """You are Nexus AI, a creative multimodal assistant that helps users across different types of content. 
+        You excel at connecting insights across text, documents, images, voice, and code. Always provide helpful, 
+        contextual responses that build on previous interactions when relevant."""
+        
+        messages.append({"role": "system", "content": system_msg})
+        
+        # Add relevant conversation history
+        relevant_context = self.memory.get_relevant_context(user_input)
+        for context in relevant_context:
+            messages.append({
+                "role": "assistant", 
+                "content": f"[Previous {context['input_type']} interaction] {context['response'][:200]}..."
+            })
+        
+        # Build current user message
+        current_content = f"Input Type: {input_type}\n\n"
+        
+        if extracted_content:
+            current_content += f"Extracted Content:\n{extracted_content[:2000]}...\n\n" if len(extracted_content) > 2000 else f"Extracted Content:\n{extracted_content}\n\n"
+        
+        current_content += f"User Query: {user_input}"
+        
+        messages.append({"role": "user", "content": current_content})
+        
+        return messages
+    
+    def generate_response(self, user_input: str, input_type: str, extracted_content: str = "") -> str:
+        """Generate AI response using AFM-4.5B model"""
+        if not self.client:
+            return "❌ Please initialize your Together AI API key first!"
+        
+        try:
+            messages = self.build_context_messages(user_input, input_type, extracted_content)
+            
+            response = self.client.chat.completions.create(
+                model="arcee-ai/AFM-4.5B-Preview",
+                messages=messages,
+                max_tokens=1024,
+                temperature=0.7
+            )
+            
+            ai_response = response.choices[0].message.content
+            
+            # Store interaction in memory
+            self.memory.add_interaction(
+                input_type=input_type,
+                content=user_input + ("\n" + extracted_content if extracted_content else ""),
+                response=ai_response
+            )
+            
+            return ai_response
+            
+        except Exception as e:
+            return f"❌ Error generating response: {str(e)}"
+
+# Initialize the AI assistant
+nexus_ai = NexusAI()
+
+def initialize_api_key(api_key: str) -> Tuple[str, str]:
+    """Initialize the API key"""
+    if not api_key.strip():
+        return "❌ Please enter a valid API key", "error"
+    
+    success, message = nexus_ai.initialize_client(api_key.strip())
+    status = "success" if success else "error"
+    return message, status
+
+def process_text_input(user_input: str, api_key_status: str) -> str:
+    """Process text input"""
+    if api_key_status != "success":
+        return "❌ Please initialize your Together AI API key first!"
+    
+    if not user_input.strip():
+        return "Please enter some text to get started!"
+    
+    return nexus_ai.generate_response(user_input, "text")
+
+def process_pdf_input(pdf_file, user_question: str, api_key_status: str) -> str:
+    """Process PDF input with question"""
+    if api_key_status != "success":
+        return "❌ Please initialize your Together AI API key first!"
+    
+    if pdf_file is None:
+        return "Please upload a PDF file first!"
+    
+    # Extract text from PDF
+    extracted_text = nexus_ai.extract_text_from_pdf(pdf_file.name)
+    
+    if user_question.strip():
+        return nexus_ai.generate_response(user_question, "pdf", extracted_text)
+    else:
+        return nexus_ai.generate_response("Please summarize this document", "pdf", extracted_text)
+
+def process_image_input(image_file, user_question: str, api_key_status: str) -> str:
+    """Process image input with question"""
+    if api_key_status != "success":
+        return "❌ Please initialize your Together AI API key first!"
+    
+    if image_file is None:
+        return "Please upload an image file first!"
+    
+    # Analyze image
+    image_analysis = nexus_ai.analyze_image(image_file.name)
+    
+    if user_question.strip():
+        return nexus_ai.generate_response(user_question, "image", image_analysis)
+    else:
+        return nexus_ai.generate_response("What can you tell me about this image?", "image", image_analysis)
+
+def process_audio_input(audio_file, user_question: str, api_key_status: str) -> str:
+    """Process audio input with question"""
+    if api_key_status != "success":
+        return "❌ Please initialize your Together AI API key first!"
+    
+    if audio_file is None:
+        return "Please upload an audio file first!"
+    
+    # Transcribe audio
+    transcribed_text = nexus_ai.transcribe_audio(audio_file.name)
+    
+    if user_question.strip():
+        combined_input = f"Transcribed audio: '{transcribed_text}'\n\nUser question: {user_question}"
+        return nexus_ai.generate_response(combined_input, "audio", transcribed_text)
+    else:
+        return nexus_ai.generate_response("Please help me with this audio content", "audio", transcribed_text)
+
+def process_code_input(code_input: str, language: str, action: str, api_key_status: str) -> str:
+    """Process code input"""
+    if api_key_status != "success":
+        return "❌ Please initialize your Together AI API key first!"
+    
+    if not code_input.strip():
+        return "Please enter some code first!"
+    
+    result = ""
+    
+    if action == "Execute Code":
+        execution_result = nexus_ai.execute_code(code_input, language)
+        result = f"**Code Execution Result:**\n```\n{execution_result}\n```\n\n"
+    
+    ai_response = nexus_ai.generate_response(
+        f"Please analyze this {language} code and provide insights:\n\n{code_input}", 
+        "code", 
+        result
+    )
+    
+    return result + ai_response
+
+def show_conversation_history() -> str:
+    """Show recent conversation history"""
+    if not nexus_ai.memory.conversations:
+        return "No conversation history yet. Start chatting to build your knowledge base!"
+    
+    history = "## 📚 Recent Conversation History\n\n"
+    for i, conv in enumerate(nexus_ai.memory.conversations[-5:], 1):  # Show last 5
+        timestamp = datetime.fromisoformat(conv["timestamp"]).strftime("%H:%M:%S")
+        history += f"**{i}. [{conv['input_type'].upper()}] {timestamp}**\n"
+        history += f"Input: {conv['content'][:100]}{'...' if len(conv['content']) > 100 else ''}\n"
+        history += f"Response: {conv['response'][:150]}{'...' if len(conv['response']) > 150 else ''}\n\n"
+    
+    return history
+
+# Create the Gradio interface
+def create_nexus_interface():
+    with gr.Blocks(
+        theme=gr.themes.Soft(),
+        title="Nexus AI Assistant",
+        css="""
+        .gradio-container {
+            max-width: 1200px !important;
+        }
+        .api-key-box {
+            border: 2px solid #e1e5e9;
+            border-radius: 8px;
+            padding: 15px;
+            margin-bottom: 20px;
+            background-color: #f8f9fa;
+        }
+        """
+    ) as app:
+        
+        # Header
+        gr.HTML("""
+        <div style="text-align: center; padding: 20px; background: linear-gradient(135deg, #667eea 0%, #764ba2 100%); border-radius: 10px; margin-bottom: 20px;">
+            <h1 style="color: white; margin: 0; font-size: 2.5em; font-weight: bold;">🚀 Nexus AI Assistant</h1>
+            <p style="color: white; margin: 10px 0 0 0; font-size: 1.2em;">Creative Multimodal AI Powered by AFM-4.5B</p>
+        </div>
+        """)
+        
+        # API Key Section
+        with gr.Group(elem_classes=["api-key-box"]):
+            gr.HTML("<h3>🔑 API Configuration</h3>")
+            with gr.Row():
+                api_key_input = gr.Textbox(
+                    label="Together AI API Key",
+                    type="password",
+                    placeholder="Enter your Together AI API key here...",
+                    scale=3
+                )
+                api_key_btn = gr.Button("Initialize API Key", variant="primary", scale=1)
+            
+            api_key_status = gr.Textbox(
+                label="Status",
+                interactive=False,
+                value="Please enter your API key to get started"
+            )
+            
+            # Hidden state to track API key status
+            api_key_state = gr.State(value="not_initialized")
+        
+        # Main Interface Tabs
+        with gr.Tabs():
+            
+            # Text Chat Tab
+            with gr.Tab("💬 Text Chat"):
+                with gr.Column():
+                    text_input = gr.Textbox(
+                        label="Your Message",
+                        placeholder="Ask me anything! I can help with creative tasks, analysis, problem-solving...",
+                        lines=3
+                    )
+                    text_btn = gr.Button("Send Message", variant="primary")
+                    text_output = gr.Textbox(
+                        label="Nexus AI Response",
+                        lines=8,
+                        interactive=False
+                    )
+            
+            # PDF Analysis Tab
+            with gr.Tab("📄 PDF Analysis"):
+                with gr.Row():
+                    with gr.Column(scale=1):
+                        pdf_file = gr.File(
+                            label="Upload PDF",
+                            file_types=[".pdf"]
+                        )
+                        pdf_question = gr.Textbox(
+                            label="Question about PDF (optional)",
+                            placeholder="What would you like to know about this document?",
+                            lines=2
+                        )
+                        pdf_btn = gr.Button("Analyze PDF", variant="primary")
+                    
+                    with gr.Column(scale=1):
+                        pdf_output = gr.Textbox(
+                            label="Analysis Result",
+                            lines=12,
+                            interactive=False
+                        )
+            
+            # Image Analysis Tab
+            with gr.Tab("🖼️ Image Analysis"):
+                with gr.Row():
+                    with gr.Column(scale=1):
+                        image_file = gr.Image(
+                            label="Upload Image",
+                            type="filepath"
+                        )
+                        image_question = gr.Textbox(
+                            label="Question about Image (optional)",
+                            placeholder="What would you like to know about this image?",
+                            lines=2
+                        )
+                        image_btn = gr.Button("Analyze Image", variant="primary")
+                    
+                    with gr.Column(scale=1):
+                        image_output = gr.Textbox(
+                            label="Analysis Result",
+                            lines=12,
+                            interactive=False
+                        )
+            
+            # Voice Processing Tab
+            with gr.Tab("🎤 Voice Processing"):
+                with gr.Row():
+                    with gr.Column(scale=1):
+                        audio_file = gr.Audio(
+                            label="Upload Audio",
+                            type="filepath"
+                        )
+                        audio_question = gr.Textbox(
+                            label="Additional Question (optional)",
+                            placeholder="Any specific question about the audio content?",
+                            lines=2
+                        )
+                        audio_btn = gr.Button("Process Audio", variant="primary")
+                    
+                    with gr.Column(scale=1):
+                        audio_output = gr.Textbox(
+                            label="Processing Result",
+                            lines=12,
+                            interactive=False
+                        )
+            
+            # Code Executor Tab
+            with gr.Tab("⚡ Code Executor"):
+                with gr.Row():
+                    with gr.Column(scale=1):
+                        code_input = gr.Code(
+                            label="Code Input",
+                            language="python",
+                            lines=10
+                        )
+                        with gr.Row():
+                            language_select = gr.Dropdown(
+                                choices=["python", "javascript", "java", "cpp"],
+                                value="python",
+                                label="Language",
+                                scale=1
+                            )
+                            code_action = gr.Radio(
+                                choices=["Execute Code", "Analyze Only"],
+                                value="Execute Code",
+                                label="Action",
+                                scale=1
+                            )
+                        code_btn = gr.Button("Process Code", variant="primary")
+                    
+                    with gr.Column(scale=1):
+                        code_output = gr.Textbox(
+                            label="Result & Analysis",
+                            lines=15,
+                            interactive=False
+                        )
+            
+            # Memory & History Tab
+            with gr.Tab("🧠 Memory & History"):
+                with gr.Column():
+                    gr.HTML("<h3>Conversation Memory</h3>")
+                    gr.HTML("<p>Nexus AI remembers your interactions and can connect insights across different input types.</p>")
+                    
+                    history_btn = gr.Button("Show Recent History", variant="secondary")
+                    history_output = gr.Textbox(
+                        label="Conversation History",
+                        lines=15,
+                        interactive=False
+                    )
+        
+        # Event handlers
+        api_key_btn.click(
+            fn=initialize_api_key,
+            inputs=[api_key_input],
+            outputs=[api_key_status, api_key_state]
+        )
+        
+        text_btn.click(
+            fn=process_text_input,
+            inputs=[text_input, api_key_state],
+            outputs=[text_output]
+        )
+        
+        pdf_btn.click(
+            fn=process_pdf_input,
+            inputs=[pdf_file, pdf_question, api_key_state],
+            outputs=[pdf_output]
+        )
+        
+        image_btn.click(
+            fn=process_image_input,
+            inputs=[image_file, image_question, api_key_state],
+            outputs=[image_output]
+        )
+        
+        audio_btn.click(
+            fn=process_audio_input,
+            inputs=[audio_file, audio_question, api_key_state],
+            outputs=[audio_output]
+        )
+        
+        code_btn.click(
+            fn=process_code_input,
+            inputs=[code_input, language_select, code_action, api_key_state],
+            outputs=[code_output]
+        )
+        
+        history_btn.click(
+            fn=show_conversation_history,
+            outputs=[history_output]
+        )
+        
+        # Footer
+        gr.HTML("""
+        <div style="text-align: center; padding: 20px; margin-top: 30px; border-top: 1px solid #e1e5e9;">
+            <p style="color: #666;">🚀 <strong>Nexus AI Assistant</strong> - Powered by AFM-4.5B | Built with ❤️ using Gradio</p>
+            <p style="color: #888; font-size: 0.9em;">Multi-modal AI assistant for creative and analytical tasks</p>
+        </div>
+        """)
+    
+    return app
+
+# Launch the application
+if __name__ == "__main__":
+    app = create_nexus_interface()
+    app.launch(
+        server_name="0.0.0.0",
+        server_port=7860,
+        share=True,
+        debug=True
+    )