app.py

import base64
import requests
from io import BytesIO
from PIL import Image
import gradio as gr

def encode_image(img):
    """
    Encodes a PIL Image to a base64 string in PNG format.
    """
    buffered = BytesIO()
    img.save(buffered, format="PNG")
    encoded_string = base64.b64encode(buffered.getvalue()).decode("utf-8")
    return encoded_string

def get_api_response(api_key, user_inputs):
    """
    Sends the user message and image to the Hyperbolic API and retrieves the response.
    """
    if not api_key:
        return {"error": "API key is required."}
    
    if not user_inputs.get("text") and not user_inputs.get("image"):
        return {"error": "Please provide a text message, an image, or both."}
    
    try:
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {api_key}",
        }
        
        messages = []
        
        if user_inputs.get("text"):
            messages.append({
                "type": "text",
                "text": user_inputs["text"]
            })
        
        if user_inputs.get("image"):
            # Open the uploaded image (already a PIL Image)
            base64_img = encode_image(user_inputs["image"])
            messages.append({
                "type": "image_url",
                "image_url": {"url": f"data:image/png;base64,{base64_img}"}
            })
        
        payload = {
            "messages": [
                {
                    "role": "user",
                    "content": messages,
                }
            ],
            "model": "Qwen/Qwen2-VL-72B-Instruct",
            "max_tokens": 2048,
            "temperature": 0.7,
            "top_p": 0.9,
        }
        
        api_endpoint = "https://api.hyperbolic.xyz/v1/chat/completions"
        response = requests.post(api_endpoint, headers=headers, json=payload)
        
        # Check if the request was successful
        if response.status_code == 200:
            api_response = response.json()
            # Extract the AI's reply (assuming the response structure)
            ai_reply = api_response.get("choices", [{}])[0].get("message", {}).get("content", "No response content.")
            return {"response": ai_reply}
        else:
            return {"error": f"API Error: {response.status_code} - {response.text}"}
    
    except Exception as e:
        return {"error": str(e)}

def chatbot_response(api_key, user_inputs, history):
    """
    Handles the chatbot interaction by updating the conversation history.
    """
    user_text = user_inputs.get("text")
    user_image = user_inputs.get("image")
    
    # Append the user's message to the history
    if user_text or user_image:
        history.append(("User", user_text, user_image))
    
    # Get the API response
    api_result = get_api_response(api_key, user_inputs)
    
    if "error" in api_result:
        ai_message = f"Error: {api_result['error']}"
    else:
        ai_message = api_result["response"]
    
    # Append the AI's response to the history
    history.append(("AI", ai_message, None))
    
    return history, history

# Define the Gradio interface
with gr.Blocks() as demo:
    gr.Markdown(
        """
        # 🖼️ Qwen2-VL-72B-Instruct with Hyperbolic API

        Engage in a conversation with the AI by sending text messages and/or uploading images. Enter your Hyperbolic API key to get started.
        """
    )
    
    with gr.Row():
        api_key_input = gr.Textbox(
            label="🔑 Hyperbolic API Key",
            type="password",
            placeholder="Enter your API key here",
            interactive=True
        )
    
    chatbot = gr.Chatbot(label="💬 Chatbot")  # Removed `.style()` method
    
    with gr.Row():
        chat_input = gr.MultimodalTextbox(
            label="Your Input",
            placeholder="Type your message and/or upload an image...",
            file_count="multiple",  # Allows multiple files if needed
            interactive=True
        )
    
    send_button = gr.Button("📤 Send")
    
    # Hidden state to keep track of the conversation history
    state = gr.State([])
    
    send_button.click(
        fn=chatbot_response,
        inputs=[api_key_input, chat_input, state],
        outputs=[chatbot, state]
    )
    
    gr.Markdown(
        """
        ---
        **Note:** Your API key is used only for this session and is not stored. Ensure you trust the environment in which you're running this application.
        """
    )

# Launch the Gradio app
if __name__ == "__main__":
    demo.launch()