| | |
| |
|
| | 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) |
| |
|
