🎨 Mimic GenAIOrchestration

app.py

@@ -19,7 +19,9 @@ def healthcheck():
 async def inference(payload: PayloadModel, token: str = Depends(authenticate_token)):
     try:
         model_response = await internvl_inference(model, payload)
-        return JSONResponse(status_code=200, content={"status": "ok", "response": model_response})
+        model_response = "True" if model_response else "False"
+        final_response = {"1":{"query_status": model_response}}
+        return JSONResponse(status_code=200, content={"final_response": final_response})
     except Exception as e:
         print(f"Error: {e}")
         return JSONResponse(status_code=500, content={"status": "error", "message": str(e)})

internvl_utils.py

@@ -6,7 +6,7 @@ from torchvision.transforms.functional import InterpolationMode
 from transformers import AutoConfig
 from models import InternVL3
 from payload_model import PayloadModel
-from models.misc_utils import convert_base64_to_pil
+from models.misc_utils import get_images_using_bbox, get_whole_image
 
 IMAGENET_MEAN = (0.485, 0.456, 0.406)
 IMAGENET_STD = (0.229, 0.224, 0.225)
@@ -75,9 +75,10 @@ def dynamic_preprocess(image, min_num=1, max_num=12, image_size=448, use_thumbna
         processed_images.append(thumbnail_img)
     return processed_images
 
-def load_image(image, input_size=448, max_num=12):
-    # image = Image.open(image_file).convert('RGB')
-    pil_image = convert_base64_to_pil(image)
+def load_image(pil_image, input_size=448, max_num=12):
+    pil_image = pil_image.convert('RGB')
+
+    # pil_image = convert_base64_to_pil(image)
     transform = build_transform(input_size=input_size)
     images = dynamic_preprocess(pil_image, image_size=input_size, use_thumbnail=True, max_num=max_num)
     pixel_values = [transform(image) for image in images]
@@ -114,4 +115,15 @@ def split_model(model_name):
     return device_map
 
 async def internvl_inference(model: InternVL3, payload: PayloadModel):
-    return await model(payload)
+    try:
+        if payload.input_utilization_mode == "bbox":
+            images = get_images_using_bbox(payload)
+        else:
+            images = get_whole_image(payload)
+
+    except Exception as e:
+        raise Exception(f"Error: {e}")
+
+    prompt_keyword = payload.prompt
+    prompt_eval_mode = payload.prompt_eval_mode
+    return await model(images, prompt_keyword, prompt_eval_mode)

models/InternVL3/intervl3.py

@@ -4,7 +4,7 @@ from payload_model import PayloadModel
 from internvl_utils import load_image
 from pydantic import BaseModel, Field
 from typing import Optional
-
+import PIL
 class InternVL3(BaseModel):
     model_name: str
     model: Optional[AutoModel] = None
@@ -43,11 +43,11 @@ respond with "Yes" else respond with "No". Limit your response to either "Yes" o
             query_prompt = None
         return query_prompt
 
-    def predict(self, payload: PayloadModel):
-        pixel_values = load_image(payload.image)
-        query_prompt = self.get_query_prompt(payload.prompt_keyword)
+    def predict(self, pil_image: PIL.Image.Image, prompt_keyword: str):
+        pixel_values = load_image(pil_image)
+        query_prompt = self.get_query_prompt(prompt_keyword)
         if query_prompt is None:
-            model_response = f"Invalid prompt keyword: {payload.prompt_keyword}"
+            model_response = f"Invalid prompt keyword: {prompt_keyword}"
         else:
             model_response = self.model.chat(
                 self.tokenizer,
@@ -58,11 +58,37 @@ respond with "Yes" else respond with "No". Limit your response to either "Yes" o
 
         return model_response
     
+    def eval_or(self, images: list[PIL.Image.Image], prompt_keyword: str):
+        model_responses = []
+        for image in images:
+            model_response = self.predict(image, prompt_keyword)
+            model_responses.append(model_response)
+            if self.extract_model_response(model_response):
+                return True, model_responses
+        return False, model_responses
+
+    def eval_and(self, images: list[PIL.Image.Image], prompt_keyword: str):
+        model_responses = []
+        for image in images:
+            model_response = self.predict(image, prompt_keyword)
+            model_responses.append(model_response)
+            if not self.extract_model_response(model_response):
+                return False, model_responses
+        return True, model_responses
+
     def extract_model_response(self, model_response: str):
         return "Yes" in model_response
 
-    async def __call__(self, payload: PayloadModel):
-        model_response = self.predict(payload)
-        extracted_response = self.extract_model_response(model_response)
-        return extracted_response
+    async def __call__(self, images: list[PIL.Image.Image], prompt_keyword: str, prompt_eval_mode: str):
+        overall_response = False
+        if prompt_eval_mode == "or":
+            overall_response, model_responses = self.eval_or(images, prompt_keyword)
+        elif prompt_eval_mode == "and":
+            overall_response, model_responses = self.eval_and(images, prompt_keyword)
+        else:
+            raise ValueError(f"Invalid prompt eval mode: {prompt_eval_mode}")
+        
+        print(f"Model responses: {model_responses}")
+        
+        return overall_response
 

models/misc_utils.py

@@ -2,7 +2,7 @@ import cv2
 import numpy as np
 import base64
 from PIL import Image
-
+from payload_model import PayloadModel
 
 def convert_base64_to_cv2(base64_string: str):
     return cv2.imdecode(np.frombuffer(base64.b64decode(base64_string), np.uint8), cv2.IMREAD_COLOR)
@@ -13,3 +13,24 @@ def convert_cv2_to_pil(image: np.ndarray):
 def convert_base64_to_pil(base64_string: str):
     return convert_cv2_to_pil(convert_base64_to_cv2(base64_string))
 
+def get_images_using_bbox(payload: PayloadModel):
+    images = []
+    # Forcing that only a single image is received
+    cv2_image = convert_base64_to_cv2(payload.input_data[0])
+    print(f"Bbox: {payload.bbox}")
+    images_bboxes = payload.bbox
+    image_bboxes = images_bboxes[0]
+    for idx, bbox in enumerate(image_bboxes):
+        x1, y1, x2, y2 = bbox
+        image = cv2_image[y1:y2, x1:x2]
+        cv2.imwrite(f"image_{idx}.png", image)
+        pil_image = convert_cv2_to_pil(image)
+        images.append(pil_image)
+    return images
+
+def get_whole_image(payload: PayloadModel):
+    images = []
+    # Forcing that only a single image is received
+    pil_image = convert_base64_to_pil(payload.input_data[0])
+    images.append(pil_image)
+    return images

payload_model.py

@@ -2,5 +2,8 @@ from pydantic import BaseModel
 
 class PayloadModel(BaseModel):
     """Type check for payload parameters"""
-    image: str
-    prompt_keyword: str
+    input_data: list[str]
+    prompt: str
+    prompt_eval_mode: str
+    input_utilization_mode: str
+    bbox: list[list[list[int]]]