Example 2 encoder

app.py

@@ -1,80 +1,157 @@
-import gradio as gr
-from transformers import AutoProcessor, AutoModelForZeroShotImageClassification
-from PIL import Image
+# import gradio as gr
+# from transformers import AutoProcessor, AutoModelForZeroShotImageClassification
+# from PIL import Image
+# import requests
+# import torch
+
+# # Load the FashionCLIP processor and model
+# processor = AutoProcessor.from_pretrained("patrickjohncyh/fashion-clip")
+# model = AutoModelForZeroShotImageClassification.from_pretrained("patrickjohncyh/fashion-clip")
+
+# # Define the function to process both text and image inputs
+# def generate_embeddings(input_text=None, input_image_url=None):
+#     try:
+#         if input_image_url:
+#             # Process image with accompanying text
+#             response = requests.get(input_image_url, stream=True)
+#             response.raise_for_status()
+#             image = Image.open(response.raw)
+
+#             # Use a default text if none is provided
+#             if not input_text:
+#                 input_text = "this is an image"
+
+#             # Prepare inputs for the model
+#             inputs = processor(
+#                 text=[input_text],
+#                 images=image,
+#                 return_tensors="pt",
+#                 padding=True
+#             )
+
+#             with torch.no_grad():
+#                 outputs = model(**inputs)
+
+#             image_embedding = outputs.logits_per_image.cpu().numpy().tolist()
+#             return {
+#                 "type": "image_embedding",
+#                 "input_image_url": input_image_url,
+#                 "input_text": input_text,
+#                 "embedding": image_embedding
+#             }
+
+#         elif input_text:
+#             # Process text input only
+#             inputs = processor(
+#                 text=[input_text],
+#                 images=None,
+#                 return_tensors="pt",
+#                 padding=True
+#             )
+#             with torch.no_grad():
+#                 outputs = model(**inputs)
+
+#             text_embedding = outputs.logits_per_text.cpu().numpy().tolist()
+#             return {
+#                 "type": "text_embedding",
+#                 "input_text": input_text,
+#                 "embedding": text_embedding
+#             }
+#         else:
+#             return {"error": "Please provide either a text query or an image URL."}
+
+#     except Exception as e:
+#         return {"error": str(e)}
+
+# # Create the Gradio interface
+# interface = gr.Interface(
+#     fn=generate_embeddings,
+#     inputs=[
+#         gr.Textbox(label="Text Query (Optional)", placeholder="e.g., red dress (used with image or for text embedding)"),
+#         gr.Textbox(label="Image URL", placeholder="e.g., https://example.com/image.jpg (used with or without text query)")
+#     ],
+#     outputs="json",
+#     title="FashionCLIP Combined Embedding API",
+#     description="Provide a text query and/or an image URL to compute embeddings for vector search."
+# )
+
+# # Launch the app
+# if __name__ == "__main__":
+#     interface.launch()
+# print(generate_embeddings("red dress"))
+
+
+
+import uuid
 import requests
-import torch
+from PIL import Image
+import numpy as np
+import gradio as gr
+from encoder import FashionCLIPEncoder
 
-# Load the FashionCLIP processor and model
-processor = AutoProcessor.from_pretrained("patrickjohncyh/fashion-clip")
-model = AutoModelForZeroShotImageClassification.from_pretrained("patrickjohncyh/fashion-clip")
+# Constants
+REQUESTS_HEADERS = {
+    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
+}
 
-# Define the function to process both text and image inputs
-def generate_embeddings(input_text=None, input_image_url=None):
-    try:
-        if input_image_url:
-            # Process image with accompanying text
-            response = requests.get(input_image_url, stream=True)
-            response.raise_for_status()
-            image = Image.open(response.raw)
-
-            # Use a default text if none is provided
-            if not input_text:
-                input_text = "this is an image"
-
-            # Prepare inputs for the model
-            inputs = processor(
-                text=[input_text],
-                images=image,
-                return_tensors="pt",
-                padding=True
-            )
-
-            with torch.no_grad():
-                outputs = model(**inputs)
-
-            image_embedding = outputs.logits_per_image.cpu().numpy().tolist()
-            return {
-                "type": "image_embedding",
-                "input_image_url": input_image_url,
-                "input_text": input_text,
-                "embedding": image_embedding
-            }
-
-        elif input_text:
-            # Process text input only
-            inputs = processor(
-                text=[input_text],
-                images=None,
-                return_tensors="pt",
-                padding=True
-            )
-            with torch.no_grad():
-                outputs = model(**inputs)
-
-            text_embedding = outputs.logits_per_text.cpu().numpy().tolist()
-            return {
-                "type": "text_embedding",
-                "input_text": input_text,
-                "embedding": text_embedding
-            }
-        else:
-            return {"error": "Please provide either a text query or an image URL."}
+# Initialize encoder
+encoder = FashionCLIPEncoder()
 
+# Helper function to download images
+def download_image_as_pil(url: str, timeout: int = 10) -> Image.Image:
+    try:
+        response = requests.get(url, stream=True, headers=REQUESTS_HEADERS, timeout=timeout)
+        if response.status_code == 200:
+            return Image.open(response.raw).convert("RGB")  # Ensure consistent format
+        return None
     except Exception as e:
-        return {"error": str(e)}
-
-# Create the Gradio interface
-interface = gr.Interface(
-    fn=generate_embeddings,
-    inputs=[
-        gr.Textbox(label="Text Query (Optional)", placeholder="e.g., red dress (used with image or for text embedding)"),
-        gr.Textbox(label="Image URL", placeholder="e.g., https://example.com/image.jpg (used with or without text query)")
+        print(f"Error downloading image: {e}")
+        return None
+
+# Embedding function for a batch of images
+def batch_process_images(image_urls: list):
+    embeddings = []
+    results = []
+    for url in image_urls:
+        try:
+            # Download image
+            image = download_image_as_pil(url)
+            if not image:
+                results.append({"image_url": url, "error": "Failed to download image"})
+                continue
+            
+            # Generate embedding
+            embedding = encoder.encode_images([image])[0]
+            
+            # Normalize embedding
+            embedding_normalized = embedding / np.linalg.norm(embedding)
+            
+            # Append results
+            results.append({
+                "image_url": url,
+                "embedding_preview": embedding_normalized[:5].tolist(),  # First 5 values for preview
+                "success": True
+            })
+        except Exception as e:
+            results.append({"image_url": url, "error": str(e)})
+    return results
+
+# Gradio Interface
+iface = gr.Interface(
+    fn=batch_process_images,
+    inputs=gr.Textbox(
+        lines=5,
+        placeholder="Enter image URLs separated by commas",
+        label="Batch Image URLs",
+    ),
+    outputs=gr.JSON(label="Embedding Results"),
+    title="Batch Fashion CLIP Embedding API",
+    description="Enter multiple image URLs (separated by commas) to generate embeddings for the batch. Each embedding preview includes the first 5 values.",
+    examples=[
+        ["https://cdn.shopify.com/s/files/1/0522/2239/4534/files/CT21355-22_1024x1024.webp, https://cdn.shopify.com/s/files/1/0522/2239/4534/files/00907857-C6B0-4D2A-8AEA-688BDE1E67D7_1024x1024.jpg"]
     ],
-    outputs="json",
-    title="FashionCLIP Combined Embedding API",
-    description="Provide a text query and/or an image URL to compute embeddings for vector search."
 )
 
-# Launch the app
+# Launch Gradio App
 if __name__ == "__main__":
-    interface.launch()
+    iface.launch()

encoder.py

@@ -0,0 +1,89 @@
+from typing import List, Dict, Optional
+import torch
+from PIL.Image import Image
+from torch.utils.data import DataLoader
+from datasets import Dataset
+from transformers import AutoModel, AutoProcessor
+
+MODEL_NAME = "Marqo/marqo-fashionCLIP"
+
+
+class FashionCLIPEncoder:
+    def __init__(self):
+        self.processor = AutoProcessor.from_pretrained(
+            MODEL_NAME, trust_remote_code=True
+        )
+        self.model = AutoModel.from_pretrained(MODEL_NAME, trust_remote_code=True)
+
+        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        self.model.to(self.device)
+        self.model.eval()
+
+    def encode_images(
+        self, images: List[Image], batch_size: Optional[int] = None
+    ) -> List[List[float]]:
+        if batch_size is None:
+            batch_size = min(len(images), 32)  # Default to a safe batch size
+
+        def transform_fn(el: Dict):
+            return self.processor(
+                images=[content for content in el["image"]], return_tensors="pt"
+            )
+
+        dataset = Dataset.from_dict({"image": images})
+        dataset.set_format("torch")
+        dataset.set_transform(transform_fn)
+        dataloader = DataLoader(dataset, batch_size=batch_size)
+
+        image_embeddings = []
+
+        with torch.no_grad():
+            for batch in dataloader:
+                try:
+                    batch = {k: v.to(self.device) for k, v in batch.items()}
+                    embeddings = self._encode_images(batch)
+                    image_embeddings.extend(embeddings)
+                except Exception as e:
+                    print(f"Error encoding image batch: {e}")
+
+        return image_embeddings
+
+    def encode_text(
+        self, text: List[str], batch_size: Optional[int] = None
+    ) -> List[List[float]]:
+        if batch_size is None:
+            batch_size = min(len(text), 32)  # Default to a safe batch size
+
+        def transform_fn(el: Dict):
+            kwargs = {
+                "padding": "max_length",
+                "return_tensors": "pt",
+                "truncation": True,
+            }
+            return self.processor(text=el["text"], **kwargs)
+
+        dataset = Dataset.from_dict({"text": text})
+        dataset = dataset.map(
+            function=transform_fn, batched=True, remove_columns=["text"]
+        )
+        dataset.set_format("torch")
+        dataloader = DataLoader(dataset, batch_size=batch_size)
+
+        text_embeddings = []
+
+        with torch.no_grad():
+            for batch in dataloader:
+                try:
+                    batch = {k: v.to(self.device) for k, v in batch.items()}
+                    embeddings = self._encode_text(batch)
+                    text_embeddings.extend(embeddings)
+                except Exception as e:
+                    print(f"Error encoding text batch: {e}")
+
+        return text_embeddings
+
+    def _encode_images(self, batch: Dict) -> List:
+        return self.model.get_image_features(**batch).detach().cpu().numpy().tolist()
+
+    def _encode_text(self, batch: Dict) -> List:
+        return self.model.get_text_features(**batch).detach().cpu().numpy().tolist()

test_local.py

@@ -0,0 +1,14 @@
+from app import generate_embeddings  # Import the function from app.py
+
+
+# Test with a text query only
+result_text = generate_embeddings(input_text="red dress")
+print("Text Embedding Result:", result_text)
+
+# Test with an image URL only
+result_image = generate_embeddings(input_image_url="https://vacier.com/cdn/shop/files/Unisize_Ring_db777381-c510-457f-b8c9-5812665d094b.jpg?v=1731838123&width=1080")
+print("Image Embedding Result:", result_image)
+
+# Test with both text and image
+result_both = generate_embeddings(input_text="red dress", input_image_url="https://vacier.com/cdn/shop/files/Unisize_Ring_db777381-c510-457f-b8c9-5812665d094b.jpg?v=1731838123&width=1080")
+print("Text and Image Embedding Result:", result_both)