Creating a handler.py file to support HF dedicated inference endpoints

Hi all,

Made a quick handler.py file for people to use. It works for me pretty well. Feel free to test it out :)

Here is an example for when making a request to the endpoint:

import requests
import base64
import json

# Endpoint URL
api_url = 'your API URL'

# Path to the image file you want to test with
image_path = '' # Replace with the path to your image file

# The question to send to the API
question = 'what is going on in this picture?'

# Open the image file in binary mode and encode it in base64
with open(image_path, 'rb') as img:
encoded_image = base64.b64encode(img.read()).decode('utf-8')

# Prepare the JSON payload
payload = {
'inputs': {
'image': encoded_image,
'question': question
}
}

# Set the headers
headers = {
"Accept" : "application/json",
"Authorization": "Bearer HUGGINGFACE_TOKEN",
"Content-Type": "application/json"
}

# Send the POST request
response = requests.post(api_url, headers=headers, data=json.dumps(payload))

# Print the response from the server
print(response.json().get('body', {}))

handler.py

@@ -0,0 +1,58 @@
+from transformers import AutoModelForCausalLM, AutoTokenizer
+from PIL import Image
+import torch
+from io import BytesIO
+import base64
+
+class EndpointHandler:
+    def __init__(self, model_dir):
+        self.model_id = "zesquirrelnator/moondream2-finetuneV2"
+        self.model = AutoModelForCausalLM.from_pretrained(self.model_id, trust_remote_code=True)
+        self.tokenizer = AutoTokenizer.from_pretrained("vikhyatk/moondream2", trust_remote_code=True)
+
+        # Check if CUDA (GPU support) is available and then set the device to GPU or CPU
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.model.to(self.device)
+
+    def preprocess_image(self, encoded_image):
+        """Decode and preprocess the input image."""
+        decoded_image = base64.b64decode(encoded_image)
+        img = Image.open(BytesIO(decoded_image)).convert("RGB")
+        return img
+
+    def __call__(self, data):
+        """Handle the incoming request."""
+        try:
+            # Extract the inputs from the data
+            inputs = data.pop("inputs", data)
+            input_image = inputs['image']
+            question = inputs.get('question', "move to the red ball")
+
+            # Preprocess the image
+            img = self.preprocess_image(input_image)
+
+            # Perform inference
+            enc_image = self.model.encode_image(img).to(self.device)
+            answer = self.model.answer_question(enc_image, question, self.tokenizer)
+
+            # If the output is a tensor, move it back to CPU and convert to list
+            if isinstance(answer, torch.Tensor):
+                answer = answer.cpu().numpy().tolist()
+
+            # Create the response
+            response = {
+                "statusCode": 200,
+                "body": {
+                    "answer": answer
+                }
+            }
+            return response
+        except Exception as e:
+            # Handle any errors
+            response = {
+                "statusCode": 500,
+                "body": {
+                    "error": str(e)
+                }
+            }
+            return response