Create handler.py

handler.py

@@ -0,0 +1,51 @@
+import torch
+from PIL import Image
+from transformers import AutoModel, AutoTokenizer
+import requests
+from io import BytesIO
+import base64
+import ssl
+import urllib3
+
+urllib3.disable_warnings()
+ssl._create_default_https_context = ssl._create_unverified_context
+
+class EndpointHandler:
+    def __init__(self, path=""):
+        model_name = "openbmb/MiniCPM-o-2_6"
+        self.tokenizer = AutoTokenizer.from_pretrained(model_name, trust_remote_code=True)
+        self.model = AutoModel.from_pretrained(model_name, trust_remote_code=True).eval()
+
+    def __call__(self, data):
+        image_input = data.get("image")
+        question = data.get("question", "What is in this image?")
+        if not image_input:
+            return {"error": "Image is required."}
+
+        try:
+            if image_input.startswith("http"):
+                response = requests.get(image_input, verify=False)
+                image = Image.open(BytesIO(response.content)).convert("RGB")
+            else:
+                image = Image.open(BytesIO(base64.b64decode(image_input))).convert("RGB")
+        except Exception as e:
+            return {"error": f"Failed to load image: {e}"}
+
+        msgs = [{"role": "user", "content": question}]
+        result_text = ""
+
+        try:
+            with torch.no_grad():
+                for chunk in self.model.chat(
+                    image=image,
+                    msgs=msgs,
+                    tokenizer=self.tokenizer,
+                    stream=True,
+                    max_new_tokens=128,
+                    temperature=0.3
+                ):
+                    result_text += chunk
+        except Exception as e:
+            return {"error": f"Model inference failed: {e}"}
+
+        return {"output": result_text}