trying to get api working but it is not working yet

experimental/clip_api_app.py

@@ -1,12 +1,9 @@
-# File name: model.py
-import json
-import os
+from typing import List
 import numpy as np
 import torch
-from starlette.requests import Request
-from PIL import Image
 import ray
 from ray import serve
+from PIL import Image
 from clip_retrieval.load_clip import load_clip, get_tokenizer 
 # from clip_retrieval.clip_client import ClipClient, Modality
 
@@ -24,11 +21,14 @@ class CLIPTransform:
 
         print ("using device", self.device)        
 
-    def text_to_embeddings(self, prompt):
-        text = self.tokenizer([prompt]).to(self.device)
+    @serve.batch(max_batch_size=32)
+    # def text_to_embeddings(self, prompts: List[str]) -> torch.Tensor:
+    def text_to_embeddings(self, prompts: List[str]) -> List[np.ndarray]:
+        text = self.tokenizer(prompts).to(self.device)
         with torch.no_grad():
             prompt_embededdings = self.model.encode_text(text)
         prompt_embededdings /= prompt_embededdings.norm(dim=-1, keepdim=True)
+        prompt_embededdings = prompt_embededdings.cpu().numpy().tolist()
         return(prompt_embededdings)
     
     def image_to_embeddings(self, input_im):
@@ -45,31 +45,31 @@ class CLIPTransform:
         image_embeddings /= image_embeddings.norm(dim=-1, keepdim=True)
         return(image_embeddings)    
 
-    async def __call__(self, http_request: Request) -> str:
-        request = await http_request.json()
-        # print(type(request))
-        # print(str(request))
-        # switch based if we are using text or image
-        embeddings = None
-        if "text" in request:
-            prompt = request["text"]
-            embeddings = self.text_to_embeddings(prompt)
-        elif "image" in request:
-            image_url = request["image_url"]
-            # download image from url
-            import requests
-            from io import BytesIO
-            input_image = Image.open(BytesIO(image_url))
-            input_image = input_image.convert('RGB')
-            input_image = np.array(input_image)
-            embeddings = self.image_to_embeddings(input_image)
-        elif "preprocessed_image" in request:
-            prepro = request["preprocessed_image"]
-            # create torch tensor on the device
-            prepro = torch.tensor(prepro).to(self.device)
-            embeddings = self.preprocessed_image_to_emdeddings(prepro)
-        else:
-            raise Exception("Invalid request")
-        return embeddings.cpu().numpy().tolist()
+    # async def __call__(self, http_request: Request) -> str:
+    #     request = await http_request.json()
+    #     # print(type(request))
+    #     # print(str(request))
+    #     # switch based if we are using text or image
+    #     embeddings = None
+    #     if "text" in request:
+    #         prompt = request["text"]
+    #         embeddings = self.text_to_embeddings(prompt)
+    #     elif "image" in request:
+    #         image_url = request["image_url"]
+    #         # download image from url
+    #         import requests
+    #         from io import BytesIO
+    #         input_image = Image.open(BytesIO(image_url))
+    #         input_image = input_image.convert('RGB')
+    #         input_image = np.array(input_image)
+    #         embeddings = self.image_to_embeddings(input_image)
+    #     elif "preprocessed_image" in request:
+    #         prepro = request["preprocessed_image"]
+    #         # create torch tensor on the device
+    #         prepro = torch.tensor(prepro).to(self.device)
+    #         embeddings = self.preprocessed_image_to_emdeddings(prepro)
+    #     else:
+    #         raise Exception("Invalid request")
+    #     return embeddings.cpu().numpy().tolist()
 
 deployment_graph = CLIPTransform.bind()

experimental/clip_api_app_client.py

@@ -1,10 +1,10 @@
-# File name: graph_client.py
-from concurrent.futures import ThreadPoolExecutor
-import json
-import os
-import requests
-from concurrent.futures import ThreadPoolExecutor, as_completed
+import ray
+from ray import serve
 import time
+import asyncio
+
+# Create a Semaphore object
+semaphore = asyncio.Semaphore(10)
 
 test_image_url = "https://static.wixstatic.com/media/4d6b49_42b9435ce1104008b1b5f7a3c9bfcd69~mv2.jpg/v1/fill/w_454,h_333,fp_0.50_0.50,q_90/4d6b49_42b9435ce1104008b1b5f7a3c9bfcd69~mv2.jpg"
 english_text = (
@@ -12,38 +12,44 @@ english_text = (
     "of wisdom, it was the age of foolishness, it was the epoch of belief"
 )
 
+async def send_text_request(serve_client, number):
+    async with semaphore:
+        # async_handle = serve_client.get_handle("CLIPTransform", sync=False)
+        async_handle = serve.get_deployment("CLIPTransform").get_handle(sync=False)
+        # async_handle = serve.get_deployment("CLIPTransform").get_handle()
+        embeddings = ray.get(await async_handle.text_to_embeddings.remote(english_text))
+        # embeddings = await async_handle.text_to_embeddings.remote(english_text)
+        # embeddings = async_handle.text_to_embeddings.remote(english_text)
+        # embeddings = await ray.get(embeddings)
+        return number, embeddings
 
-def send_text_request(number):
-    json = {"text": english_text}
-    url = os.environ.get("HTTP_ADDRESS", "http://127.0.0.1:8000/")
-    response = requests.post(url, json=json)
-    embeddings = response.text
-    return number, embeddings
-
-def process_text(numbers, max_workers=10):
-    with ThreadPoolExecutor(max_workers=max_workers) as executor:
-        futures = [executor.submit(send_text_request, number) for number in numbers]
-        for future in as_completed(futures):
-            n_result, result = future.result()
-            result = json.loads(result)
-            print (f"{n_result} : {len(result[0])}")
-
-# def process_text(numbers, max_workers=10):
-#     for n in numbers:
-#         n_result, result = send_text_request(n)
-#         result = json.loads(result)
-#         print (f"{n_result} : {len(result[0])}")
+# def process_text(server_client, numbers, max_workers=10):
+#     with ThreadPoolExecutor(max_workers=max_workers) as executor:
+#         futures = [executor.submit(send_text_request, server_client, number) for number in numbers]
+#         for future in as_completed(futures):
+#             n_result, result = future.result()
+#             print (f"{n_result} : {len(result[0])}")
+async def process_text(server_client, numbers):
+    tasks = [send_text_request(server_client, number) for number in numbers]
+    for future in asyncio.as_completed(tasks):
+        n_result, result = await future
+        print (f"{n_result} : {len(result[0])}")
 
 if __name__ == "__main__":
     # n_calls = 100000
-    n_calls = 10000
+    n_calls = 1
     numbers = list(range(n_calls))
+    ray.init()
+    server_client = serve.start(detached=True) 
     start_time = time.monotonic()
-    process_text(numbers)
+
+    # Run the async function
+    asyncio.run(process_text(server_client, numbers))
+
     end_time = time.monotonic()
     total_time = end_time - start_time
     avg_time_ms = total_time / n_calls * 1000
     calls_per_sec = n_calls / total_time
     print(f"Average time taken: {avg_time_ms:.2f} ms")
     print(f"Number of calls per second: {calls_per_sec:.2f}")    
-
+    ray.shutdown()