|
|
|
|
|
|
|
|
import torch |
|
|
import os |
|
|
|
|
|
from modelscope.models.base import TorchModel |
|
|
from modelscope.preprocessors.base import Preprocessor |
|
|
from modelscope.pipelines.base import Model, Pipeline |
|
|
from modelscope.utils.config import Config |
|
|
from modelscope.pipelines.builder import PIPELINES |
|
|
from modelscope.preprocessors.builder import PREPROCESSORS |
|
|
from modelscope.models.builder import MODELS |
|
|
from modelscope.preprocessors.image import load_image |
|
|
|
|
|
|
|
|
from vlmo.utils.beit_utils import load_from_config |
|
|
|
|
|
|
|
|
@PIPELINES.register_module( |
|
|
"multi-modal-embeddings", module_name="multi-modal-embedding-pipeline" |
|
|
) |
|
|
class MyCustomPipeline(Pipeline): |
|
|
"""Give simple introduction to this pipeline. |
|
|
|
|
|
Examples: |
|
|
|
|
|
>>> from modelscope.pipelines import pipeline |
|
|
>>> input = "Hello, ModelScope!" |
|
|
>>> my_pipeline = pipeline('my-task', 'my-model-id') |
|
|
>>> result = my_pipeline(input) |
|
|
|
|
|
""" |
|
|
|
|
|
def __init__(self, model, preprocessor=None, **kwargs): |
|
|
""" |
|
|
use `model` and `preprocessor` to create a custom pipeline for prediction |
|
|
Args: |
|
|
model: model id on modelscope hub. |
|
|
preprocessor: the class of method be init_preprocessor |
|
|
""" |
|
|
super().__init__(model=model, auto_collate=False) |
|
|
self.model_dir = model |
|
|
self._device = "cuda" if torch.cuda.is_available() else "cpu" |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
model_config = { |
|
|
"loss_names": {"itc": 1}, |
|
|
"beit_version": "large", |
|
|
"encoder_embed_dim": 1024, |
|
|
"out_embed_dim": 1024, |
|
|
"encoder_layers": 21, |
|
|
"beit3_vl_layers": 3, |
|
|
|
|
|
"visual_mask_size": 14, |
|
|
"tokenizer_type": "GLMChineseTokenizer", |
|
|
"tokenizer": os.path.join(self.model_dir, "./vlmo/tokenizer"), |
|
|
"vocab_size": 115244, |
|
|
"whole_word_masking": False, |
|
|
"precision": 32, |
|
|
"test_only": True, |
|
|
"flash_attn": True, |
|
|
"model_path": os.path.join(self.model_dir, "m2_encoder_1B.ckpt"), |
|
|
"modelscope": { |
|
|
"model_id": "M2Cognition/M2_Encoder_Large" |
|
|
}, |
|
|
"model_file": "m2_encoder_1B.ckpt" |
|
|
} |
|
|
model, processors = load_from_config(model_config) |
|
|
self.model = model |
|
|
self.model.to(self._device).eval() |
|
|
self._tokenizer, self._img_processor = processors |
|
|
|
|
|
def _sanitize_parameters(self, **pipeline_parameters): |
|
|
""" |
|
|
this method should sanitize the keyword args to preprocessor params, |
|
|
forward params and postprocess params on '__call__' or '_process_single' method |
|
|
considered to be a normal classmethod with default implementation / output |
|
|
|
|
|
Default Returns: |
|
|
Dict[str, str]: preprocess_params = {} |
|
|
Dict[str, str]: forward_params = {} |
|
|
Dict[str, str]: postprocess_params = pipeline_parameters |
|
|
""" |
|
|
return {}, pipeline_parameters, {} |
|
|
|
|
|
def _check_input(self, inputs): |
|
|
pass |
|
|
|
|
|
def _check_output(self, outputs): |
|
|
pass |
|
|
|
|
|
def forward(self, forward_params): |
|
|
"""Provide default implementation using self.model and user can reimplement it""" |
|
|
|
|
|
labels = forward_params.get("label_list", "") |
|
|
labels = labels.split(",") |
|
|
if len(labels) > 1 and labels[0] != "": |
|
|
txt_encoding = self._tokenizer( |
|
|
labels, |
|
|
padding="max_length", |
|
|
truncation=True, |
|
|
max_length=self.model.hparams.config["max_text_len"], |
|
|
return_special_tokens_mask=True, |
|
|
) |
|
|
txt_data = { |
|
|
"text_ids": torch.tensor(txt_encoding["input_ids"]).to(self._device), |
|
|
"text_masks": torch.tensor(txt_encoding["attention_mask"]).to( |
|
|
self._device |
|
|
), |
|
|
"text_labels": None, |
|
|
} |
|
|
txt_feats = self.model.infer_text(txt_data)["cls_vlffn_feats"] |
|
|
image = forward_params["image"] |
|
|
image = load_image(image) |
|
|
img = self._img_processor(image).unsqueeze(0) |
|
|
img_data = {"image": [img.to(self._device)]} |
|
|
img_feats = self.model.infer_image(img_data)["cls_vlffn_feats"] |
|
|
logits_per_image = self.model.logit_scale.exp() * img_feats @ txt_feats.t() |
|
|
probs = logits_per_image.softmax(dim=-1).detach().cpu() |
|
|
index = probs.max(dim=-1)[1][0] |
|
|
label = labels[index] |
|
|
return {"text": label, "scores": probs.numpy().tolist()[0]} |
|
|
else: |
|
|
rets = {} |
|
|
if "text" in forward_params: |
|
|
text = forward_params.get("text") |
|
|
txt_encoding = self._tokenizer( |
|
|
text, |
|
|
padding="max_length", |
|
|
truncation=True, |
|
|
max_length=self.model.hparams.config["max_text_len"], |
|
|
return_special_tokens_mask=True, |
|
|
) |
|
|
txt_data = { |
|
|
"text_ids": torch.tensor(txt_encoding["input_ids"]).to( |
|
|
self._device |
|
|
), |
|
|
"text_masks": torch.tensor(txt_encoding["attention_mask"]).to( |
|
|
self._device |
|
|
), |
|
|
"text_labels": None, |
|
|
} |
|
|
txt_feats = self.model.infer_text(txt_data)["cls_vlffn_feats"] |
|
|
rets.update({"text_embedding": txt_feats.detach()}) |
|
|
if "img" in forward_params: |
|
|
input_img = forward_params["img"] |
|
|
img = self._img_processor(input_img).unsqueeze(0) |
|
|
img_data = {"image": [img.to(self._device)]} |
|
|
img_feats = self.model.infer_image(img_data)["cls_vlffn_feats"] |
|
|
rets.update({"img_embedding": img_feats.detach()}) |
|
|
|
|
|
return rets |
|
|
|
|
|
def preprocess(self, inputs): |
|
|
return inputs |
|
|
|
|
|
def postprocess(self, inputs): |
|
|
"""If current pipeline support model reuse, common postprocess |
|
|
code should be write here. |
|
|
|
|
|
Args: |
|
|
inputs: input data |
|
|
|
|
|
Return: |
|
|
dict of results: a dict containing outputs of model, each |
|
|
output should have the standard output name. |
|
|
""" |
|
|
return inputs |
|
|
|
|
|
|
|
|
""" |
|
|
# Tips: usr_config_path is the temporary save configuration location， after upload modelscope hub, it is the model_id |
|
|
usr_config_path = "/tmp/snapdown/" |
|
|
config = Config( |
|
|
{ |
|
|
"framework": "pytorch", |
|
|
"task": "multi-modal-embeddings", |
|
|
"model": {"type": "m2-encoder"}, |
|
|
"pipeline": {"type": "multi-modal-embedding-pipeline"}, |
|
|
"allow_remote": True, |
|
|
} |
|
|
) |
|
|
config.dump("/tmp/snapdown/" + "configuration.json") |
|
|
""" |
|
|
|
|
|
if __name__ == "__main__": |
|
|
from modelscope.pipelines import pipeline |
|
|
from modelscope.preprocessors.image import load_image |
|
|
|
|
|
model = "M2Cognition/M2-Encoder" |
|
|
pipe = pipeline("multi-modal-embeddings", model=model) |
|
|
input = { |
|
|
"image": "https://clip-cn-beijing.oss-cn-beijing.aliyuncs.com/pokemon.jpeg", |
|
|
"label_list": "杰尼龟,妙蛙种子,小火龙,皮卡丘", |
|
|
} |
|
|
demo = pipe(input) |
|
|
print("demo output", demo) |
|
|
inputs = {"text": ["杰尼龟", "妙蛙种子", "小火龙", "皮卡丘"]} |
|
|
output = pipe(inputs) |
|
|
print("text output", output) |
|
|
input_img = load_image( |
|
|
"https://clip-cn-beijing.oss-cn-beijing.aliyuncs.com/pokemon.jpeg" |
|
|
) |
|
|
inputs = {"img": input_img} |
|
|
img_embedding = pipe(inputs) |
|
|
print("image output", img_embedding) |
|
|
|
