Create handler.py

handler.py

@@ -0,0 +1,82 @@
+from typing import Dict, Any
+import torch
+from transformers import LlavaNextProcessor, LlavaNextForConditionalGeneration, BitsAndBytesConfig
+from PIL import Image
+import requests
+from io import BytesIO
+import base64
+
+class EndpointHandler:
+    def __init__(self, path=""):
+        # Configuració de la quantització
+        quantization_config = BitsAndBytesConfig(
+            load_in_4bit=True,
+            bnb_4bit_quant_type="nf4",
+            bnb_4bit_compute_dtype=torch.float16,
+        )
+
+        # Carrega el processador i model de forma global
+        self.processor = LlavaNextProcessor.from_pretrained("llava-hf/llava-v1.6-mistral-7b-hf")
+        self.model = LlavaNextForConditionalGeneration.from_pretrained(
+            "llava-hf/llava-v1.6-mistral-7b-hf",
+            quantization_config=quantization_config,
+            device_map="auto"
+        )
+
+    def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]:
+        logs = []
+        logs.append("Iniciant processament de la petició.")
+        
+        inputs = data.get("inputs")
+        if not inputs:
+            logs.append("Format d'entrada invàlid. Manca la clau 'inputs'.")
+            return {"error": "Invalid input format. 'inputs' key is missing.", "logs": logs}
+
+        image_url = inputs.get("url")
+        image_data = inputs.get("image_data")
+        prompt = inputs.get("prompt")
+        max_tokens = inputs.get("max_tokens", 100)
+        
+        if not prompt:
+            logs.append("S'ha de proporcionar 'prompt' en 'inputs'.")
+            return {"error": "The 'prompt' must be provided in 'inputs'.", "logs": logs}
+
+        if not image_url and not image_data:
+            logs.append("S'ha de proporcionar 'url' o 'image_data' en 'inputs'.")
+            return {"error": "Either 'url' or 'image_data' must be provided in 'inputs'.", "logs": logs}
+
+        logs.append(f"Processant entrada: url={image_url}, image_data={'present' if image_data else 'absent'}, prompt={prompt}")
+
+        try:
+            if image_url:
+                logs.append(f"Carregant imatge des de URL: {image_url}")
+                response = requests.get(image_url, stream=True)
+                image = Image.open(response.raw)
+            elif image_data:
+                logs.append("Carregant imatge des de dades d'imatge en brut.")
+                image = Image.open(BytesIO(base64.b64decode(image_data)))
+            
+            if image.format == 'PNG':
+                logs.append("Convertint imatge PNG a JPG.")
+                image = image.convert('RGB')
+                buffer = BytesIO()
+                image.save(buffer, format="JPEG")
+                buffer.seek(0)
+                image = Image.open(buffer)
+
+        except Exception as e:
+            logs.append(f"Error carregant imatge: {str(e)}")
+            return {"error": str(e), "logs": logs}
+
+        try:
+            logs.append("Processant imatge amb el model.")
+            inputs = self.processor(prompt, image, return_tensors="pt").to("cuda")
+            output = self.model.generate(**inputs, max_new_tokens=max_tokens)
+            result = self.processor.decode(output[0], skip_special_tokens=True)
+            logs.append("Processament complet.")
+            return {"input_prompt": prompt, "model_output": result, "logs": logs}
+
+        except Exception as e:
+            logs.append(f"Error processant el model: {str(e)}")
+            return {"error": str(e), "logs": logs}
+