Update app.py

app.py

@@ -1,158 +1,337 @@
 import gradio as gr
 import requests
 import json
+import os
+import base64
+from PIL import Image
+import soundfile as sf
+import mimetypes
+import logging
+from io import BytesIO
+import tempfile
 
-# Default model
-MODEL = "ZeppelinCorp/Charm_15"
+# Set up logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
 
-# Maximum number of messages to keep in chat history
-MAX_HISTORY_MESSAGES = 5
+# Hugging Face API configuration
+HF_API_URL = os.getenv("HF_API_URL")  # URL of your Hugging Face model endpoint
+HF_API_TOKEN = os.getenv("HF_API_TOKEN")  # Hugging Face API token
 
-# Generate response using Hugging Face Inference API
-def generate_response(prompt, model_name, temperature, top_k, top_p, chat_history):
-    global MODEL
-    
-    if model_name != MODEL:
-        MODEL = model_name
-    
-    # Truncate chat history to the last MAX_HISTORY_MESSAGES messages
-    if len(chat_history) > MAX_HISTORY_MESSAGES:
-        chat_history = chat_history[-MAX_HISTORY_MESSAGES:]
+# Default parameter values
+default_max_tokens = 4096
+default_temperature = 0.7
+default_top_p = 0.9
+default_presence_penalty = 0.0
+default_frequency_penalty = 0.0
+
+# Initialize MIME types
+mimetypes.init()
+
+def call_hf_endpoint(payload, api_url, api_token, params=None):
+    """Call Hugging Face Inference API with the given payload."""
+    # Set parameters from the UI inputs or use defaults
+    if params is None:
+        params = {
+            "max_tokens": default_max_tokens,
+            "temperature": default_temperature,
+            "top_p": default_top_p,
+            "presence_penalty": default_presence_penalty,
+            "frequency_penalty": default_frequency_penalty
+        }
     
-    # Combine chat history into a single prompt
-    full_prompt = "\n".join([f"User: {msg['content']}" if msg['role'] == 'user' else f"Bot: {msg['content']}" for msg in chat_history])
-    full_prompt += f"\nUser: {prompt}\nBot:"
+    # Add parameters to the payload
+    if "parameters" not in payload:
+        payload["parameters"] = params
     
-    API_URL = f"https://api-inference.huggingface.co/models/{MODEL}"
-    payload = {
-        "inputs": full_prompt,
-        "parameters": {
-            "max_length": 50,
-            "num_return_sequences": 1,
-            "temperature": temperature,
-            "top_k": top_k,
-            "top_p": top_p
-        }
+    # Set up headers
+    headers = {
+        "Authorization": f"Bearer {api_token}",
+        "Content-Type": "application/json"
     }
     
     try:
-        response = requests.post(API_URL, json=payload)
-        if response.status_code == 200:
-            generated_text = response.json()[0]['generated_text']
-            chat_history.append({"role": "user", "content": prompt})
-            chat_history.append({"role": "assistant", "content": generated_text})
+        logger.info(f"Sending request to {api_url}")
+        logger.info(f"Using parameters: {params}")
+        response = requests.post(api_url, headers=headers, json=payload)
+        response.raise_for_status()
+        result = response.json()
+        logger.info("Received response successfully")
+        return result
+    except requests.exceptions.HTTPError as error:
+        logger.error(f"Request failed with status code: {error.response.status_code}")
+        logger.error(f"Error message: {error.response.text}")
+        return {"error": error.response.text}
+
+def improved_fetch_audio_from_url(url):
+    """Fetch audio data from URL and convert to base64."""
+    try:
+        logger.info(f"Fetching audio from URL: {url}")
+        response = requests.get(url, timeout=30)
+        response.raise_for_status()
+        
+        # Determine MIME type based on URL
+        file_extension = os.path.splitext(url)[1].lower()
+        mime_type = None
+        
+        if file_extension == '.wav':
+            mime_type = "audio/wav"
+        elif file_extension == '.mp3':
+            mime_type = "audio/mpeg"
+        elif file_extension == '.flac':
+            mime_type = "audio/flac"
+        elif file_extension in ['.m4a', '.aac']:
+            mime_type = "audio/aac"
+        elif file_extension == '.ogg':
+            mime_type = "audio/ogg"
         else:
-            chat_history.append({"role": "assistant", "content": f"Error: {response.status_code} - {response.text}"})
+            # Try to detect the MIME type from headers
+            content_type = response.headers.get('Content-Type', '')
+            if content_type.startswith('audio/'):
+                mime_type = content_type
+            else:
+                mime_type = "audio/wav"  # Default to WAV
+        
+        logger.info(f"Detected MIME type: {mime_type}")
+        
+        # Convert to base64
+        base64_audio = base64.b64encode(response.content).decode('utf-8')
+        logger.info(f"Successfully encoded audio to base64, length: {len(base64_audio)}")
+        
+        return mime_type, base64_audio
     except Exception as e:
-        chat_history.append({"role": "assistant", "content": f"Error generating response: {e}"})
-    
-    return chat_history
+        logger.error(f"Error fetching audio from URL: {e}", exc_info=True)
+        return None, None
+
+def fetch_image_from_url(url):
+    """Fetch image data from URL and convert to base64."""
+    try:
+        logger.info(f"Fetching image from URL: {url}")
+        response = requests.get(url)
+        response.raise_for_status()
+        
+        # Determine MIME type based on URL
+        file_extension = os.path.splitext(url)[1].lower()
+        if file_extension in ['.jpg', '.jpeg']:
+            mime_type = "image/jpeg"
+        elif file_extension == '.png':
+            mime_type = "image/png"
+        elif file_extension == '.gif':
+            mime_type = "image/gif"
+        elif file_extension in ['.bmp', '.tiff', '.webp']:
+            mime_type = f"image/{file_extension[1:]}"
+        else:
+            mime_type = "image/jpeg"  # Default to JPEG
+        
+        # Convert to base64
+        base64_image = base64.b64encode(response.content).decode('utf-8')
+        
+        logger.info(f"Successfully fetched and encoded image, mime type: {mime_type}")
+        return mime_type, base64_image
+    except Exception as e:
+        logger.error(f"Error fetching image from URL: {e}")
+        return None, None
 
-# File upload functions
-def upload_image(image):
-    return f"**Image Uploaded:** {image.name}"
+def encode_base64_from_file(file_path):
+    """Encode file content to base64 string and determine MIME type."""
+    file_extension = os.path.splitext(file_path)[1].lower()
+    
+    # Map file extensions to MIME types
+    if file_extension in ['.jpg', '.jpeg']:
+        mime_type = "image/jpeg"
+    elif file_extension == '.png':
+        mime_type = "image/png"
+    elif file_extension == '.gif':
+        mime_type = "image/gif"
+    elif file_extension in ['.bmp', '.tiff', '.webp']:
+        mime_type = f"image/{file_extension[1:]}"
+    elif file_extension == '.flac':
+        mime_type = "audio/flac"
+    elif file_extension == '.wav':
+        mime_type = "audio/wav"
+    elif file_extension == '.mp3':
+        mime_type = "audio/mpeg"
+    elif file_extension in ['.m4a', '.aac']:
+        mime_type = "audio/aac"
+    elif file_extension == '.ogg':
+        mime_type = "audio/ogg"
+    else:
+        mime_type = "application/octet-stream"
+    
+    # Read and encode file content
+    with open(file_path, "rb") as file:
+        encoded_string = base64.b64encode(file.read()).decode('utf-8')
+    
+    return encoded_string, mime_type
 
-def upload_audio(audio):
-    return f"**Audio Uploaded:** {audio.name}"
+def process_message(history, message, conversation_state):
+    """Process user message and update both history and internal state."""
+    # Extract text and files
+    text_content = message["text"] if message["text"] else ""
+    
+    image_files = []
+    audio_files = []
+    
+    # Create content array for internal state
+    content_items = []
+    
+    # Add text if available
+    if text_content:
+        content_items.append({"type": "text", "text": text_content})
+    
+    # Process and immediately convert files to base64
+    if message["files"] and len(message["files"]) > 0:
+        for file_path in message["files"]:
+            file_extension = os.path.splitext(file_path)[1].lower()
+            file_name = os.path.basename(file_path)
+            
+            # Convert the file to base64 immediately
+            base64_content, mime_type = encode_base64_from_file(file_path)
+            
+            # Add to content items for the API
+            if mime_type.startswith("image/"):
+                content_items.append({
+                    "type": "image_url",
+                    "image_url": {
+                        "url": f"data:{mime_type};base64,{base64_content}"
+                    }
+                })
+                image_files.append(file_path)
+            elif mime_type.startswith("audio/"):
+                content_items.append({
+                    "type": "audio_url",
+                    "audio_url": {
+                        "url": f"data:{mime_type};base64,{base64_content}"
+                    }
+                })
+                audio_files.append(file_path)
+    
+    # Only proceed if we have content
+    if content_items:
+        # Add to Gradio chatbot history (for display)
+        history.append({"role": "user", "content": text_content})
 
-# Reasoning analysis
-def analyze_reasoning(input_text):
-    return f"**Reasoning Analysis:** Analyzing input: {input_text}"
+        # Add file messages if present
+        for file_path in image_files + audio_files:
+            history.append({"role": "user", "content": {"path": file_path}})
+            
+        logger.info(f"Updated history with user message. Current conversation has {len(image_files)} images and {len(audio_files)} audio files")
+        
+        # Add to internal conversation state (with base64 data)
+        conversation_state.append({
+            "role": "user",
+            "content": content_items
+        })
+    
+    return history, gr.MultimodalTextbox(value=None, interactive=False), conversation_state
 
-# Switch model function
-def switch_model(new_model):
-    global MODEL
+def bot_response(history, conversation_state):
+    """Generate bot response based on conversation state."""
+    if not conversation_state:
+        return history, conversation_state
+    
+    # Create the payload
+    payload = {
+        "inputs": conversation_state
+    }
+    
+    # Log the payload for debugging (without base64 data)
+    debug_payload = json.loads(json.dumps(payload))
+    for item in debug_payload["inputs"]:
+        if "content" in item and isinstance(item["content"], list):
+            for content_item in item["content"]:
+                if "image_url" in content_item:
+                    parts = content_item["image_url"]["url"].split(",")
+                    if len(parts) > 1:
+                        content_item["image_url"]["url"] = parts[0] + ",[BASE64_DATA_REMOVED]"
+                if "audio_url" in content_item:
+                    parts = content_item["audio_url"]["url"].split(",")
+                    if len(parts) > 1:
+                        content_item["audio_url"]["url"] = parts[0] + ",[BASE64_DATA_REMOVED]"
+    
+    logger.info(f"Sending payload: {json.dumps(debug_payload, indent=2)}")
+    
+    # Call Hugging Face Inference API
+    response = call_hf_endpoint(payload, HF_API_URL, HF_API_TOKEN)
     
-    MODEL = new_model
-    return f"**Model Switched:** {new_model}"
+    # Extract text response from the Hugging Face API response
+    try:
+        if isinstance(response, dict):
+            if "generated_text" in response:
+                result = response["generated_text"]
+            elif "error" in response:
+                result = f"Error: {response['error']}"
+            else:
+                result = f"Received response: {json.dumps(response)}"
+        else:
+            result = str(response)
+    except Exception as e:
+        result = f"Error processing response: {str(e)}"
+    
+    # Add bot response to history
+    history.append({"role": "assistant", "content": result})
+    
+    # Add to conversation state
+    conversation_state.append({
+        "role": "assistant",
+        "content": [{"type": "text", "text": result}]
+    })
+    
+    return history, conversation_state
 
-# Clear chat function
-def clear_chat():
-    return []
+def enable_input():
+    """Re-enable the input box after bot responds."""
+    return gr.MultimodalTextbox(interactive=True)
 
-# Custom CSS for better UI
-custom_css = """
-.gradio-container {
-    font-family: Arial, sans-serif;
-}
-h1 {
-    color: #4CAF50;
-}
-.chat-box {
-    height: 400px;
-    overflow-y: auto;
-    border: 1px solid #ccc;
-    padding: 10px;
-    border-radius: 5px;
-    background-color: #f9f9f9;
-}
-.user-message {
-    background-color: #e3f2fd;
-    padding: 8px;
-    border-radius: 5px;
-    margin-bottom: 5px;
-    max-width: 70%;
-    margin-left: auto;
-}
-.bot-message {
-    background-color: #f5f5f5;
-    padding: 8px;
-    border-radius: 5px;
-    margin-bottom: 5px;
-    max-width: 70%;
-    margin-right: auto;
-}
-"""
+def update_debug(conversation_state):
+    """Update debug output with the last payload that would be sent."""
+    if not conversation_state:
+        return {}
+    
+    # Create a payload from the conversation
+    payload = {
+        "inputs": conversation_state
+    }
+    
+    # Remove base64 data to avoid cluttering the UI
+    sanitized_payload = json.loads(json.dumps(payload))
+    for item in sanitized_payload["inputs"]:
+        if "content" in item and isinstance(item["content"], list):
+            for content_item in item["content"]:
+                if "image_url" in content_item:
+                    parts = content_item["image_url"]["url"].split(",")
+                    if len(parts) > 1:
+                        content_item["image_url"]["url"] = parts[0] + ",[BASE64_DATA_REMOVED]"
+                if "audio_url" in content_item:
+                    parts = content_item["audio_url"]["url"].split(",")
+                    if len(parts) > 1:
+                        content_item["audio_url"]["url"] = parts[0] + ",[BASE64_DATA_REMOVED]"
+    
+    return sanitized_payload
 
-# Interface setup
-with gr.Blocks(css=custom_css) as demo:
-    gr.Markdown("# Chatbot UI")
+# Gradio interface setup
+with gr.Blocks() as demo:
+    gr.Markdown("# Chatbot with Hugging Face Models")
     
     with gr.Tab("Chat"):
         # Chat history display
-        chat_history = gr.Chatbot(label="Chat", elem_classes="chat-box", type="messages")
+        chat_history = gr.Chatbot(label="Chat")
         
         with gr.Row():
             with gr.Column():
-                prompt_input = gr.Textbox(label="Enter your prompt", placeholder="Type your message here...", lines=2)
-                model_dropdown = gr.Dropdown(
-                    ["ZeppelinCorp/Charm_15", "ZeppelinCorp/Smartbloom_1.1", "gpt2", "EleutherAI/gpt-neo-125M"],
-                    label="Select Model",
-                    value=MODEL
-                )
-                temperature = gr.Slider(minimum=0.1, maximum=1.0, value=0.7, label="Temperature")
-                top_k = gr.Slider(minimum=1, maximum=100, value=50, label="Top-K")
-                top_p = gr.Slider(minimum=0.1, maximum=1.0, value=0.9, label="Top-P")
+                prompt_input = gr.MultimodalTextbox(label="Enter your prompt", placeholder="Type your message here...", lines=2)
                 generate_button = gr.Button("Send")
-                clear_button = gr.Button("Clear Chat")
         
         # Chat interaction
         generate_button.click(
-            generate_response,
-            inputs=[prompt_input, model_dropdown, temperature, top_k, top_p, chat_history],
-            outputs=chat_history
+            process_message,
+            inputs=[chat_history, prompt_input],
+            outputs=[chat_history, prompt_input]
         )
-        clear_button.click(clear_chat, inputs=None, outputs=chat_history)
-    
-    with gr.Tab("Upload Image"):
-        image_input = gr.File(label="Upload Image", file_types=["image"])
-        image_output = gr.Textbox(label="Upload Status", interactive=False)
-        image_input.change(upload_image, inputs=image_input, outputs=image_output)
-    
-    with gr.Tab("Upload Audio"):
-        audio_input = gr.File(label="Upload Audio", file_types=["audio"])
-        audio_output = gr.Textbox(label="Upload Status", interactive=False)
-        audio_input.change(upload_audio, inputs=audio_input, outputs=audio_output)
-    
-    with gr.Tab("Reasoning Analysis"):
-        reasoning_input = gr.Textbox(label="Enter text for reasoning analysis", placeholder="Type your text here...")
-        reasoning_output = gr.Textbox(label="Analysis Result", interactive=False)
-        reasoning_input.change(analyze_reasoning, inputs=reasoning_input, outputs=reasoning_output)
-    
-    with gr.Tab("Switch Model"):
-        model_switch_input = gr.Textbox(label="Enter new model name", placeholder="Type the model name...")
-        model_switch_output = gr.Textbox(label="Switch Status", interactive=False)
-        model_switch_input.change(switch_model, inputs=model_switch_input, outputs=model_switch_output)
+    
+    # Debug output
+    debug_output = gr.JSON(label="Debug Output")
+    demo.load(update_debug, inputs=[chat_history], outputs=debug_output, every=1)
 
 # Launch the interface
 demo.launch()