File size: 4,602 Bytes
8e13475
 
 
 
 
 
 
 
3691ca2
8e13475
 
 
 
 
 
9a26341
8e13475
3691ca2
8e13475
 
 
 
9a26341
3691ca2
8e13475
 
 
 
 
 
 
3691ca2
 
9a26341
3691ca2
 
9a26341
3691ca2
 
9a26341
8bafeab
9a26341
3691ca2
 
 
 
 
8e13475
 
 
 
3691ca2
8e13475
 
 
 
 
 
 
 
3691ca2
8e13475
 
 
 
3691ca2
 
 
 
8e13475
 
 
 
 
 
9a26341
3691ca2
 
 
9a26341
 
 
3691ca2
9a26341
 
3691ca2
 
9a26341
3691ca2
 
 
 
 
 
 
81e2b71
3691ca2
 
 
8e13475
 
 
 
6c2df0f
8e13475
3691ca2
8e13475
 
 
 
 
3691ca2
8e13475
 
 
 
 
8bafeab
8e13475
 
9a26341
 
 
 
 
 
3691ca2
 
8e13475
3691ca2
 
8e13475
3691ca2
 
9a26341
3691ca2
8e13475
 
 
 
 
3691ca2
8e13475
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
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()