ruCLIP Large [vit-large-patch14-224]

Бесплатно

import os
import json
from collections import OrderedDict
from typing import Dict
import kfserving
import requests
import torch
import numpy as np
import io
from io import BytesIO
import base64
from PIL import Image
import re

from ruclip import CLIP, RuCLIPProcessor

def open_images_base64(img_strs):
    return [Image.open(BytesIO(base64.b64decode(img_str))) for img_str in img_strs]

def open_image_link(links):
    imgs = []
    for img_link in links:
        response = requests.get(img_link)
        imgs.append(Image.open(BytesIO(response.content)))
    return imgs

def create_image(sim_plt):
    my_stringIObytes = io.BytesIO()
    sim_plt.savefig(my_stringIObytes, format="jpg")
    my_stringIObytes.seek(0)
    my_base64 = base64.b64encode(my_stringIObytes.read())
    return my_base64

class KFServingRuClipModel(kfserving.KFModel):
    def __init__(self, title: str, model_path="./ruclip-vit-large-patch14-224"):
        super().__init__(name)
        self.name = name
        self.ready = False
        self.model_path = model_path

    def load(self):
        self.device = "cuda"
        self.clip = CLIP.from_pretrained(self.model_path).eval().to(self.device)
        self.clip_processor = RuCLIPProcessor.from_pretrained(self.model_path)
        self.ready = True
        
    def get_text_latents(self, texts):
        with torch.no_grad():
            inputs = self.clip_processor(text=texts, images=None)
            text_latents = self.clip.encode_text(
                input_ids=inputs["input_ids"].to(self.device), 
            )
            text_latents = text_latents / text_latents.norm(dim=-1, keepdim=True)
        return text_latents
        
    def get_logits(self, text_latents, pil_images):
        with torch.no_grad():
            inputs = self.clip_processor(text=None, images=pil_images)
            image_latents = self.clip.encode_image(
                pixel_values=inputs["pixel_values"].to(self.device)
            )
            image_latents = image_latents / image_latents.norm(dim=-1, keepdim=True)
            logits_per_text = torch.matmul(text_latents, image_latents.t())
            logits_per_image = logits_per_text.t()
        return logits_per_text, logits_per_image
    
    def get_similarity_scores(self, texts, images):
        """
        Find the most similar image to text.
        `texts`: array of texts or one text ["some_desc"]
        `images`: array of images.
        """
        text_latents = self.get_text_latents(texts)
        results = []
        for pil_image in images:
            _, logits_per_image = self.get_logits(text_latents, [pil_image])
            probs_raw = (logits_per_image * self.clip.logit_scale.exp().detach()).softmax(dim=-1)[0]
            label_id = probs_raw.argmax().item()
            confidence = probs_raw.max().item()
            
            probs = []
            for i in range(len(texts)):
                probs.append({texts[i]: probs_raw[i].item()})

            buffered = BytesIO()
            pil_image.save(buffered, format="JPEG")
            img = base64.b64encode(buffered.getvalue()).decode("utf-8")
            results.append({"image": img, "text": texts[label_id], "confidence" : confidence, "all_res": probs})
        
        return results
    
    def predict(self, request: Dict) -> Dict:
        texts = request["instances"][0]["texts"]
        
        img_strs = request["instances"][0].get("images", None)
        if img_strs is not None:
            images = open_images_base64(img_strs)
        
        images_links = request["instances"][0].get("image_links", None)

        if images_links is not None:
            images = open_image_link(images_links)
          
        error_msg = None
        predictions = []
        try:
            predictions = self.get_similarity_scores(texts, images)
        except Exception as ex:
            print(ex)
            error_msg = ex

        if error_msg is not None:
            return {"predictions": predictions, "error_message": str(error_msg)}
        else:
            return {"predictions": predictions}

     predictions: [
				{
					"image": "base64 image", 
					"text": "наиболее подходящий текст для картинки" ,
          "confidence": "мера близости для лучшей картинки", 
          "all_res": [{"менее вероятный текст": 0.31}, {"другой текст": 0.33},} ...]
				}
     ]

!pip install -r requirements.txt
from kfserving_ru_CLIP import KFServingRuClipModel

model = KFServingRuClipModel("kfserving-clip")
model.load()

# zero-shot and links
url_cat = "https://cs11.livemaster.ru/storage/topic/NxN/2c/9b/9cf0a41d13ecb11439e6145dff576315df83op.jpg?h=3KvOPndE06tlraLLSmkHPQ"
url_cat2 = "https://pbs.twimg.com/profile_images/560798448962633728/rDEdUfV_.jpeg"
url_dog = "https://ichef.bbci.co.uk/news/640/cpsprodpb/475B/production/_98776281_gettyimages-521697453.jpg"
result = model.predict({"instances": [{
        "texts": ["собачка", "кошка", "мышка", "машина", "стол", "дом", "жидкость"],
        "image_links": [url_dog, url_cat, url_cat2]
    }]
})

"Res: ", result

`{"predictions": "[{"image": b"/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAAgGBgcGBQ...", "text": "собачка", "confidence": 0.99822598695755, "all_res": [\{"собачка": 0.99822598695755}, \{"кошка": 0.0016664331778883934}, \{"мышка": 1.3638593372888863e-05}, \{"машина": 1.954195431608241e-05}, \{"стол": 3.68908649761579e-06}, \{"дом": 3.0324985345941968e-05}, \{"жидкость": 4.039051418658346e-05}]}, ... ]", "error_message": None}`