Set up model

app.py

@@ -0,0 +1,157 @@
+import gradio as gr
+import torch
+import os
+import requests
+from io import BytesIO
+from PIL import Image
+from transformers import AutoModel, AutoProcessor, CLIPModel
+from thai2transformers.preprocess import process_transformers
+from qdrant_client import QdrantClient
+from qdrant_client.http import models
+from qdrant_client.http.models import Filter, FieldCondition, MatchValue
+
+device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+
+# Load models
+image_processor = AutoProcessor.from_pretrained("openai/clip-vit-base-patch32")
+image_model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32").to(device)
+
+text_processor = AutoProcessor.from_pretrained("openthaigpt/CLIPTextCamembertModelWithProjection", trust_remote_code=True)
+text_model = AutoModel.from_pretrained("openthaigpt/CLIPTextCamembertModelWithProjection", trust_remote_code=True).to(device)
+
+# Qdrant setup
+
+url = os.environ.get("QDRANT_URL")
+api_key = os.environ.get("QDRANT_API_KEY")
+qdrant_client = QdrantClient(url=url, api_key=api_key)
+
+from PIL import ImageDraw, ImageFont
+
+def generate_error_image(message="No image"):
+    img = Image.new("RGB", (256, 256), color=(240, 240, 240))
+    draw = ImageDraw.Draw(img)
+    try:
+        font = ImageFont.truetype("arial.ttf", 18)
+    except:
+        font = ImageFont.load_default()
+    draw.text((10, 120), message[:30], fill="red", font=font)
+    return img
+
+
+def get_image_embedding(image: Image.Image):
+    inputs = image_processor(images=image, return_tensors="pt")
+    inputs = {k: v.to(device) for k, v in inputs.items()}
+    with torch.no_grad():
+        image_embeddings = image_model.get_image_features(**inputs)
+    image_embeddings /= image_embeddings.norm(dim=1, keepdim=True)
+    return image_embeddings[0].cpu().numpy().tolist()
+
+def get_text_embedding(text: str):
+    try:
+        processed = process_transformers(text)
+    except Exception as e:
+        return [(None, f"Preprocessing error: {str(e)}")]
+    inputs = text_processor(text=processed, return_tensors="pt", padding=True)
+    inputs = {k: v.to(device) for k, v in inputs.items()}
+    with torch.no_grad():
+        text_embeddings = text_model(**inputs).text_embeds
+    text_embeddings /= text_embeddings.norm(dim=1, keepdim=True)
+    return text_embeddings[0].cpu().numpy().tolist()
+
+def retrieve_from_qdrant(query_vector, modality, limit=10):
+    return qdrant_client.query_points(
+        collection_name="thai2transformers_clip",
+        query=query_vector,
+        with_payload=True,
+        query_filter=Filter(
+            must=[
+                FieldCondition(
+                    key="modality",
+                    match=MatchValue(value=modality)
+                )
+            ]
+        ),
+        limit=limit
+    ).points
+
+def load_image_from_url(url):
+    try:
+        # Fetch image from URL
+        request = requests.get(url)
+        request.raise_for_status()  # Raise an error for bad responses
+        return Image.open(BytesIO(request.content)).convert("RGB")
+    except Exception as e:
+        print(f"Error fetching image from {url}: {e}")
+        return None
+
+def multimodal_query(text_input, text_target, image_input, image_url, mode):
+    if mode == "image-to-image":
+        image = image_input or (load_image_from_url(image_url) if image_url else None)
+        if image is None:
+            return [(Image.new("RGB", (256, 256), color="white"), "❌ No image provided.")]
+
+        query_vector = get_image_embedding(image)
+        modality = "image"
+
+    elif mode == "image-to-text":
+        image = image_input or (load_image_from_url(image_url) if image_url else None)
+        if image is None:
+            return [(Image.new("RGB", (256, 256), color="white"), "❌ No image provided.")]
+        query_vector = get_image_embedding(image)
+        modality = "text"
+
+    elif mode == "text-to-image":
+        if not text_input:
+            return [(None, "No text provided.")]
+        query_vector = get_text_embedding(text_input)
+        modality = "image"
+
+    elif mode == "text-to-text":
+        if not text_input:
+            return [(None, "No text provided.")]
+        query_vector = get_text_embedding(text_input)
+        modality = "text"
+
+    else:
+        return [(None, "Invalid mode selected.")]
+
+    results = retrieve_from_qdrant(query_vector, modality)
+    outputs = []
+
+    for res in results:
+        try:
+            img_url = res.payload.get("image_url")
+            caption = res.payload.get("name", "")
+            if img_url:
+                img = Image.open(BytesIO(requests.get(img_url).content)).resize((256, 256))
+            else:
+                img = generate_error_image("No image URL")
+            outputs.append((img, caption[:40]))
+        except Exception as e:
+            fallback_img = generate_error_image("Error loading image")
+            outputs.append((fallback_img, f"Error: {str(e)[:30]}"))
+
+    return outputs
+
+
+# Gradio UI
+with gr.Blocks(title="🔄 Multimodal Query System") as demo:
+    gr.Markdown("## 🔎 ค้นหาด้วยรูปภาพและข้อความ (CLIP + Qdrant)\nรองรับทั้ง image-to-image, image-to-text, text-to-image และ text-to-text")
+
+    with gr.Row():
+        with gr.Column():
+            mode = gr.Dropdown(label="โหมดการค้นหา", choices=["image-to-image", "image-to-text", "text-to-image", "text-to-text"], value="image-to-image")
+            image_input = gr.Image(type="pil", label="อัปโหลดภาพ")
+            image_url = gr.Textbox(label="หรือใส่ URL ของภาพ")
+            text_input = gr.Textbox(label="ข้อความที่ใช้ค้นหา (text input)")
+            text_target = gr.Textbox(label="เป้าหมาย (ใช้ในบางกรณี)", visible=False)
+            search_btn = gr.Button("🔍 ค้นหา")
+
+        with gr.Column():
+            gallery = gr.Gallery(label="ผลลัพธ์", columns=5, height=600)
+
+    search_btn.click(
+        fn=multimodal_query,
+        inputs=[text_input, text_target, image_input, image_url, mode],
+        outputs=gallery
+    )

configuration_clip_camembert.py

@@ -0,0 +1,51 @@
+from transformers import CamembertConfig
+
+
+class CLIPTextCamembertConfig(CamembertConfig):
+    # ref : https://huggingface.co/airesearch/wangchanberta-base-att-spm-uncased/blob/main/config.json
+    model_type = "clip_text_camembert"
+
+    def __init__(
+        self,
+        vocab_size=25005,
+        hidden_size=768,
+        intermediate_size=3072,
+        projection_dim=512,
+        num_hidden_layers=12,
+        num_attention_heads=12,
+        max_position_embeddings=512,
+        hidden_act="gelu",
+        layer_norm_eps=1e-12,
+        attention_dropout=0.1,
+        initializer_range=0.02,
+        initializer_factor=1.0,
+        pad_token_id=1,
+        bos_token_id=0,
+        eos_token_id=2,
+        type_vocab_size=1,
+        **kwargs,
+    ):
+        super().__init__(
+            pad_token_id=pad_token_id,
+            bos_token_id=bos_token_id,
+            eos_token_id=eos_token_id,
+            **kwargs,
+        )
+
+        self.vocab_size = vocab_size
+        self.hidden_size = hidden_size
+        self.intermediate_size = intermediate_size
+        self.projection_dim = projection_dim
+        self.num_hidden_layers = num_hidden_layers
+        self.num_attention_heads = num_attention_heads
+        self.max_position_embeddings = max_position_embeddings
+        self.layer_norm_eps = layer_norm_eps
+        self.hidden_act = hidden_act
+        self.initializer_range = initializer_range
+        self.initializer_factor = initializer_factor
+        self.attention_dropout = attention_dropout
+        self.type_vocab_size = type_vocab_size
+        self.auto_map = {
+            "AutoConfig": "configuration_clip_camembert.CLIPTextCamembertConfig",
+            "AutoModel": "modeling_clip_camembert.CLIPTextCamembertModelWithProjection",
+        }

modeling_clip_camembert.py

@@ -0,0 +1,71 @@
+from .configuration_clip_camembert import CLIPTextCamembertConfig
+from transformers import (
+    CamembertModel,
+    CLIPTextModelWithProjection,
+)
+from transformers.models.clip.modeling_clip import CLIPTextModelOutput
+import torch
+from torch import nn
+from typing import Any, Optional, Tuple, Union
+
+
+class CLIPTextCamembertModelWithProjection(CLIPTextModelWithProjection):
+    config_class = CLIPTextCamembertConfig
+
+    def __init__(self, config: CLIPTextCamembertConfig):
+        super().__init__(config)
+
+        self.text_model = CamembertModel(config)
+
+        self.text_projection = nn.Linear(
+            config.hidden_size, config.projection_dim, bias=False
+        )
+        # Initialize weights and apply final processing
+        self.post_init()
+
+    def forward(
+        self,
+        input_ids: Optional[torch.Tensor] = None,
+        attention_mask: Optional[torch.Tensor] = None,
+        position_ids: Optional[torch.Tensor] = None,
+        output_attentions: Optional[bool] = None,
+        output_hidden_states: Optional[bool] = None,
+        return_dict: Optional[bool] = None,
+    ) -> Union[Tuple, CLIPTextModelOutput]:
+        return_dict = (
+            return_dict if return_dict is not None else self.config.use_return_dict
+        )
+
+        text_outputs = self.text_model(
+            input_ids=input_ids,
+            attention_mask=attention_mask,
+            position_ids=position_ids,
+            output_attentions=output_attentions,
+            output_hidden_states=output_hidden_states,
+            return_dict=return_dict,
+        )
+
+        pooled_output = text_outputs[1]
+
+        text_embeds = self.text_projection(pooled_output)
+
+        if not return_dict:
+            outputs = (text_embeds, text_outputs[0]) + text_outputs[2:]
+            return tuple(output for output in outputs if output is not None)
+
+        return CLIPTextModelOutput(
+            text_embeds=text_embeds,
+            last_hidden_state=text_outputs.last_hidden_state,
+            hidden_states=text_outputs.hidden_states,
+            attentions=text_outputs.attentions,
+        )
+
+    def converter_weight(
+        self, path_model="airesearch/wangchanberta-base-att-spm-uncased"
+    ):
+        r"""
+        converter weight from airesearch/wangchanberta-base-att-spm-uncased
+        """
+        pretrained_state_dict = CamembertModel.from_pretrained(path_model).state_dict()
+        # Load the new state dictionary into the custom model
+        self.text_model.load_state_dict(pretrained_state_dict)

requirements.txt

@@ -0,0 +1,13 @@
+torch
+torchvision
+torchaudio
+thai2transformers==0.1.2
+pythainlp
+transformers
+pillow
+qdrant-client
+requests
+gradio
+numpy
+matplotlib
+ftfy