Upload folder

configuration_qualityv.py

@@ -0,0 +1,78 @@
+from transformers.configuration_utils import PretrainedConfig
+from transformers import AutoConfig
+from transformers.activations import ACT2FN
+
+
+class QualityLinearAdapterConfig(PretrainedConfig):
+    model_type = "QualityvForCausalLM"
+    adapter_type = "linear"
+    
+    def __init__(self, 
+                 in_hidden_size: int = 1024,
+                 num_layers: int = 2, 
+                 intermediate_size: int = 2048,
+                 out_hidden_size: int = 2028,
+                 act_fn: str = "gelu", 
+                 **kwargs,
+                 ) -> None:
+        super().__init__(**kwargs)
+        
+        self.in_hidden_size = in_hidden_size
+        self.num_layers = num_layers
+        self.intermediate_size = intermediate_size
+        self.out_hidden_size = out_hidden_size
+        self.act_fn = act_fn
+        
+        
+class QualityvConfig(PretrainedConfig):
+    model_type = "QualityvForCausalLM"
+    def __init__(self, 
+                 vision_model_name: str=None,
+                 audio_model_name: str=None,
+                 llm_model_name: str=None,
+                 image_token_id: int=None,
+                 video_token_id: int=None,
+                 audio_token_id: int=None,
+                 adapter_type: str="linear",
+                 num_adapter_layers: int=2,
+                 **kwargs,
+                 ) -> None:
+        super().__init__(**kwargs)
+        self.vision_model_name = vision_model_name
+        self.audio_model_name = audio_model_name
+        self.llm_model_name = llm_model_name
+        self.image_token_id = image_token_id
+        self.video_token_id = video_token_id
+        self.audio_token_id = audio_token_id
+        self.adapter_type = adapter_type
+        self.num_adapter_layers = num_adapter_layers
+        if llm_model_name is not None:
+            self.llm_config = AutoConfig.from_pretrained(llm_model_name)
+            for key, value in self.llm_config.to_dict().items():
+                setattr(self, key, value)
+            if vision_model_name is not None:
+                self.vision_config = AutoConfig.from_pretrained(vision_model_name)
+                self.vision_adapter_config = QualityLinearAdapterConfig(
+                    in_hidden_size=self.vision_config.hidden_size,
+                    intermediate_size=self.vision_config.hidden_size * 2,
+                    out_hidden_size=self.llm_config.hidden_size,
+                    num_layers=num_adapter_layers,
+                )
+            else:
+                self.vision_config = None
+            if audio_model_name is not None:
+                self.audio_config = AutoConfig.from_pretrained(audio_model_name)
+                self.audio_adapter_config = QualityLinearAdapterConfig(
+                    in_hidden_size=self.audio_config.hidden_size,
+                    intermediate_size=self.audio_config.hidden_size * 2,
+                    out_hidden_size=self.llm_config.hidden_size,
+                    num_layers=num_adapter_layers,
+                )
+            else:
+                self.audio_config = None
+                
+    def get_vocab_size(self):
+        return self.llm_config.vocab_size
+    
+    def get_text_config(self, **kwargs):
+        return self.llm_config.get_text_config(**kwargs)

modeling_qualityv.py

@@ -0,0 +1,241 @@
+from transformers import AutoModel, AutoModelForCausalLM
+from transformers.activations import ACT2FN
+from transformers.modeling_utils import PreTrainedModel
+from transformers.generation.utils import GenerationMixin
+from transformers.modeling_outputs import CausalLMOutputWithPast
+import torch
+from torch import nn
+from torch.nn import RMSNorm
+from typing import List, Optional
+
+from .configuration_qualityv import QualityvConfig, QualityLinearAdapterConfig
+
+
+class QualityLinearAdapter(nn.Module):
+    def __init__(self, config: QualityLinearAdapterConfig):
+        super().__init__()
+        self.config = config
+        self.norm = RMSNorm(config.in_hidden_size)
+        self.act_fn = ACT2FN[config.act_fn]
+        if config.num_layers == 1:
+            self.linears = nn.Linear(config.in_hidden_size, config.out_hidden_size)
+        else:
+            model_list = []
+            for _ in range(config.num_layers - 1):
+                model_list.append(nn.Linear(config.in_hidden_size, config.intermediate_size))
+                model_list.append(self.act_fn)
+            model_list.append(nn.Linear(config.intermediate_size, config.out_hidden_size))
+            self.linears = nn.Sequential(*model_list)
+        
+    def forward(self, x: torch.Tensor) -> torch.Tensor:
+        x = self.linears(self.norm(x))
+        return x
+    
+    
+class QualityvForCausalLM(PreTrainedModel, GenerationMixin):
+    
+    
+    def __init__(self, config: QualityvConfig, *args, **kwargs):
+        super().__init__(config, *args, **kwargs)
+        self.config = config
+        self.llm_model = AutoModelForCausalLM.from_pretrained(config.llm_model_name)
+        if config.vision_config is not None:
+            self.vision_model = AutoModel.from_pretrained(config.vision_model_name)
+            self.vision_adapter = QualityLinearAdapter(config.vision_adapter_config)
+        if config.audio_config is not None:
+            self.audio_model = AutoModel.from_pretrained(config.audio_model_name)
+            self.audio_adapter = QualityLinearAdapter(config.audio_adapter_config)
+            self.decoder_input_ids = torch.tensor([[1, 1,]]) * self.audio_model.config.decoder_start_token_id
+        self.post_init()
+        
+    def get_input_embeddings(self):
+        return self.llm_model.get_input_embeddings()
+    
+    def set_input_embeddings(self, value):
+        self.llm_model.set_input_embeddings(value)
+        
+    def get_output_embeddings(self):
+        return self.llm_model.get_output_embeddings()
+    
+    def set_output_embeddings(self, value):
+        self.llm_model.set_output_embeddings(value)
+
+    def set_decoder(self, decoder):
+        self.llm_model.set_decoder(decoder)
+        
+    def get_decoder(self):
+        return self.llm_model.get_decoder()
+    
+    def get_vision_model(self):
+        return self.vision_model
+    
+    def get_audio_model(self):
+        return self.audio_model
+    
+    def get_video_features(self, pixel_values_videos: torch.Tensor) -> torch.Tensor:
+        video_embeds = self.vision_model(pixel_values_videos).last_hidden_state
+        video_embeds = self.vision_adapter(video_embeds)
+        return video_embeds
+    
+    def get_audio_features(self, audio_values: torch.Tensor) -> torch.Tensor:
+        audio_embeds = self.audio_model.encoder(audio_values).last_hidden_state
+        audio_embeds = self.audio_adapter(audio_embeds)
+        return audio_embeds
+    
+    def get_image_features(self, pixel_values: torch.Tensor) -> torch.Tensor:
+        image_embeds = self.vision_model(pixel_values).last_hidden_state
+        image_embeds = self.vision_adapter(image_embeds)
+        return image_embeds
+    
+    def replace_multi_modal_embeddings(self, multi_modal_embeds: torch.Tensor,
+                                       input_embeds: torch.Tensor,
+                                       input_ids: torch.LongTensor,
+                                       multi_modal_token_id: int,
+                                       note: str="multi_modal"):
+        # multi_modal_embeds: batch_size * num_frames, hidden_steps, hidden_size
+        # input_embeds: batch_size, seq_length, hidden_size
+        # input_ids: batch_size, seq_length
+        # multi_modal_token_id: int
+        # note: str
+        hidden_size = multi_modal_embeds.shape[-1]
+        multi_modal_embeds = multi_modal_embeds.view(-1, hidden_size)
+        n_modal_tokens = (input_ids == multi_modal_token_id).sum()
+        n_modal_embeds = multi_modal_embeds.shape[0]
+        if n_modal_tokens != n_modal_embeds:
+            raise ValueError(f"The number of {note} tokens ({n_modal_tokens}) does not match the number of {note} embeddings ({n_modal_embeds}).")
+        mask = input_ids == multi_modal_token_id
+        mask_unsqueezed = mask.unsqueeze(-1)
+        mask_expanded = mask_unsqueezed.expand_as(input_embeds)
+        video_mask = mask_expanded.to(input_embeds.device)
+        multi_modal_embeds = multi_modal_embeds.to(input_embeds.device, dtype=input_embeds.dtype)
+        input_embeds = input_embeds.masked_scatter(video_mask, multi_modal_embeds)
+        return input_embeds
+        
+    
+    def forward(self, 
+                input_ids: torch.LongTensor = None,
+                attention_mask: Optional[torch.Tensor] = None,
+                position_ids: Optional[torch.LongTensor] = None,
+                past_key_values: Optional[List[torch.FloatTensor]] = None,
+                inputs_embeds: Optional[torch.FloatTensor] = None,
+                labels: Optional[torch.LongTensor] = None,
+                use_cache: Optional[bool] = None,
+                output_attentions: Optional[bool] = None,
+                output_hidden_states: Optional[bool] = None,
+                return_dict: Optional[bool] = None,
+                pixel_values: Optional[torch.Tensor] = None,
+                pixel_values_videos: Optional[torch.FloatTensor] = None,
+                audio_values: Optional[torch.FloatTensor] = None,
+                cache_position: Optional[torch.LongTensor] = None,
+                **kwargs
+                ):
+        output_attentions = output_attentions if output_attentions is not None else self.config.llm_config.output_attentions
+        output_hidden_states = (
+            output_hidden_states if output_hidden_states is not None else self.config.llm_config.output_hidden_states
+        )
+        return_dict = return_dict if return_dict is not None else self.config.llm_config.use_return_dict
+        
+        if inputs_embeds is None:
+            inputs_embeds = self.get_input_embeddings()(input_ids)
+        
+            if pixel_values_videos is not None:
+                video_features = self.get_video_features(pixel_values_videos)
+                inputs_embeds = self.replace_multi_modal_embeddings(video_features, inputs_embeds, input_ids, self.config.video_token_id, note="video")
+            
+            if pixel_values is not None:
+                image_features = self.get_image_features(pixel_values)
+                inputs_embeds = self.replace_multi_modal_embeddings(image_features, inputs_embeds, input_ids, self.config.image_token_id, note="image")
+            
+            if audio_values is not None:
+                audio_features = self.get_audio_features(audio_values)
+                inputs_embeds = self.replace_multi_modal_embeddings(audio_features, inputs_embeds, input_ids, self.config.audio_token_id, note="audio")
+            
+        outputs = self.llm_model(
+            input_ids=None,
+            attention_mask=attention_mask,
+            position_ids=position_ids,
+            past_key_values=past_key_values,
+            inputs_embeds=inputs_embeds,
+            labels=labels,
+            use_cache=use_cache,
+            output_attentions=output_attentions,
+            output_hidden_states=output_hidden_states,
+            return_dict=return_dict,
+            cache_position=cache_position,
+            **kwargs
+        )
+         
+        return outputs
+        
+        
+        
+    def prepare_inputs_for_generation(self, 
+                                      input_ids, 
+                                      past_key_values=None, 
+                                      attention_mask=None, 
+                                      use_cache=None,
+                                      pixel_values=None,
+                                      pixel_values_videos=None,
+                                      audio_values=None,
+                                      cache_position=None,
+                                      **kwargs):
+        model_inputs = super().prepare_inputs_for_generation(
+            input_ids=input_ids,
+            past_key_values=past_key_values,
+            attention_mask=attention_mask,
+            use_cache=use_cache,
+            pixel_values=pixel_values,
+            pixel_values_videos=pixel_values_videos,
+            audio_values=audio_values,
+            **kwargs
+        )
+        if cache_position[0] != 0:
+            model_inputs["pixel_values"] = None
+            model_inputs["pixel_values_videos"] = None
+        return model_inputs
+    
+    def _expand_inputs_for_generation(self, 
+                                      expand_size: int = 1,
+                                      is_encoder_decoder: bool = False,
+                                      input_ids: Optional[torch.LongTensor] = None,
+                                      **model_kwargs,
+                                      ):
+        """Expands input tensors for generation when using beam search or sampling.
+
+        Args:
+            expand_size (int, optional): The size to expand the inputs by. Defaults to 1.
+            is_encoder_decoder (bool, optional): Whether the model is an encoder-decoder model. Defaults to False.
+            input_ids (Optional[torch.LongTensor], optional): The input token IDs. Defaults to None.
+            **model_kwargs: Additional model-specific keyword arguments.
+
+        Returns:
+            Tuple[torch.LongTensor, Dict[str, torch.Tensor]]: The expanded input_ids and model_kwargs.
+        """
+        if input_ids is not None:
+            input_ids = input_ids.repeat_interleave(expand_size, dim=0)
+
+        # Expand attention mask if present
+        if "attention_mask" in model_kwargs:
+            model_kwargs["attention_mask"] = model_kwargs["attention_mask"].repeat_interleave(expand_size, dim=0)
+
+        # Expand position IDs if present
+        if "position_ids" in model_kwargs:
+            model_kwargs["position_ids"] = model_kwargs["position_ids"].repeat_interleave(expand_size, dim=0)
+
+        # Expand pixel values for images if present
+        if "pixel_values" in model_kwargs and model_kwargs["pixel_values"] is not None:
+            model_kwargs["pixel_values"] = model_kwargs["pixel_values"].repeat_interleave(expand_size, dim=0)
+
+        # Expand pixel values for videos if present
+        if "pixel_values_videos" in model_kwargs and model_kwargs["pixel_values_videos"] is not None:
+            model_kwargs["pixel_values_videos"] = model_kwargs["pixel_values_videos"].repeat_interleave(expand_size, dim=0)
+
+        # Expand audio values if present
+        if "audio_values" in model_kwargs and model_kwargs["audio_values"] is not None:
+            model_kwargs["audio_values"] = model_kwargs["audio_values"].repeat_interleave(expand_size, dim=0)
+
+        # Expand cache position if present
+        if "cache_position" in model_kwargs and model_kwargs["cache_position"] is not None:
+            model_kwargs["cache_position"] = model_kwargs["cache_position"].repeat_interleave(expand_size, dim=0)
+
+        return input_ids, model_kwargs

processing_qualityv.py

@@ -0,0 +1,312 @@
+from typing import Union, Optional, List, Dict, Tuple, Callable
+from transformers.processing_utils import (ProcessorMixin,
+                                           VideosKwargs,
+                                           AudioKwargs,
+                                           ImagesKwargs,
+                                           TextKwargs,
+                                           ProcessingKwargs,
+                                           Unpack)
+import numpy as np
+import decord
+import torch
+import PIL
+from transformers.audio_utils import load_audio
+from transformers.image_utils import load_image, load_video
+from transformers import AutoImageProcessor, AutoFeatureExtractor, AutoTokenizer
+
+
+def load_audio_str(audio_path_or_url: str, sampling_rate: int = 16000) -> np.ndarray:
+    audio = load_audio(audio_path_or_url, sampling_rate=sampling_rate)
+    return audio
+
+
+def load_video_str(video_path_or_url: str, num_frames: int = 4, fps: int = None) -> List[np.ndarray]:
+    video = load_video(video_path_or_url, num_frames=num_frames, fps=fps,
+                       backend="decord")
+    return video
+
+
+def load_image_str(image_path_or_url: str) -> List[np.ndarray]:
+    image = load_image(image_path_or_url)
+    return image
+
+
+ImageInput = Union[
+    # same as transformers.image_utils.ImageInput
+    "PIL.Image.Image", np.ndarray, "torch.Tensor", list["PIL.Image.Image"], list[np.ndarray], list["torch.Tensor"],
+    # image urls, or image_paths
+    str, list[str]
+]
+
+
+VideoInput = Union[
+    # same as transformers.image_utils.VideoInput
+    list["PIL.Image.Image"], "np.ndarray", "torch.Tensor", list["np.ndarray"],
+    list["torch.Tensor"], list[list["PIL.Image.Image"]], list[list["np.ndarray"]],
+    list[list["torch.Tensor"]],
+    # video urls, or video_paths
+    str, list[str], list[list[str]]
+]
+
+
+AudioInput = Union[
+    # same as transformers.audio_utils.AudioInput
+    np.ndarray, "torch.Tensor", List[np.ndarray], Tuple[np.ndarray], List["torch.Tensor"], Tuple["torch.Tensor"],  # noqa: F821
+    # audio urls, or audio_paths
+    str, list[str]
+]
+
+
+class QualityvImageKwargs(ImagesKwargs):
+    tokens_per_image: int = 197
+
+
+class QualityvVideoKwargs(VideosKwargs):
+    num_frames: Union[int, None] = 4
+    fps: Union[int, None] = None
+    tokens_per_frame: int = 197
+
+
+class QualityvAudioKwargs(AudioKwargs):
+    sampling_rate: Union[int, None] = 16000
+    tokens_per_audio: int = 1500
+    
+    
+class QualityvProcessingKwargs(ProcessingKwargs):
+    images_kwargs: QualityvImageKwargs
+    videos_kwargs: QualityvVideoKwargs
+    audio_kwargs: QualityvAudioKwargs
+    text_kwargs: TextKwargs
+    
+
+class QualityvProcessor(ProcessorMixin):
+    
+    attributes = ["image_processor", 
+                  "audio_processor",
+                  "tokenizer"]
+    image_processor_class = "AutoImageProcessor"
+    audio_processor_class = "AutoFeatureExtractor"
+    tokenizer_class = "AutoTokenizer"
+    
+    chat_template = """{% set image_count = namespace(value=0) %}{% set video_count = namespace(value=0) %}{% set audio_count = namespace(value=0) %}{% for message in messages %}{% if loop.first and message['role'] != 'system' %}<|im_start|>system
+You are a helpful assistant.<|im_end|>
+{% endif %}<|im_start|>{{ message['role'] }}
+{% if message['content'] is string %}{{ message['content'] }}<|im_end|>
+{% else %}{% for content in message['content'] %}{% if content['type'] == 'image' or 'image' in content or 'image_url' in content %}{% set image_count.value = image_count.value + 1 %}{% if add_vision_id %}Picture {{ image_count.value }}: {% endif %}<|vision_start|><|image_pad|><|vision_end|>{% elif content['type'] == 'video' or 'video' in content %}{% set video_count.value = video_count.value + 1 %}{% if add_vision_id %}Video {{ video_count.value }}: {% endif %}<|vision_start|><|video_pad|><|vision_end|>{% elif content['type'] == 'audio' or 'audio' in content %}{% set audio_count.value = audio_count.value + 1 %}{% if add_vision_id %}Audio {{ audio_count.value }}: {% endif %}<|vision_start|><|audio_pad|><|vision_end|>{% elif 'text' in content %}{{ content['text'] }}{% endif %}{% endfor %}<|im_end|>
+{% endif %}{% endfor %}{% if add_generation_prompt %}<|im_start|>assistant
+{% endif %}"""
+
+    
+    def __init__(self, tokenizer=None,
+                 image_processor=None,
+                 audio_processor=None,
+                 chat_template=None,
+                 image_token="<|image_pad|>",
+                 video_token="<|video_pad|>",
+                 audio_token="<|audio_pad|>",
+                 label_start_text="<|im_start|>assistant\n",
+                 label_end_text="<|im_end|>\n",
+                 **kwargs):
+        self.image_token = image_token if not hasattr(tokenizer, "image_token") else tokenizer.image_token
+        self.video_token = video_token if not hasattr(tokenizer, "video_token") else tokenizer.video_token
+        self.audio_token = audio_token if not hasattr(tokenizer, "audio_token") else tokenizer.audio_token
+        self.label_start_text = label_start_text
+        self.label_end_text = label_end_text
+        self.image_token_id = (
+            tokenizer.image_token_id
+            if getattr(tokenizer, "image_token_id", None)
+            else tokenizer.convert_tokens_to_ids(self.image_token)
+        )
+        self.video_token_id = (
+            tokenizer.video_token_id
+            if getattr(tokenizer, "video_token_id", None)
+            else tokenizer.convert_tokens_to_ids(self.video_token)
+        )
+        self.audio_token_id = (
+            tokenizer.audio_token_id
+            if getattr(tokenizer, "audio_token_id", None)
+            else tokenizer.convert_tokens_to_ids(self.audio_token)
+        )
+        if chat_template is None:
+            chat_template = self.chat_template
+        super().__init__(image_processor, audio_processor, tokenizer, 
+                         chat_template=chat_template)
+        
+    def __call__(self, 
+                 text: Union[str, List[str], None] = None,
+                 messages: Union[List[Dict], None] = None,
+                 images: Union[ImageInput, None] = None,
+                 videos: Union[VideoInput, None] = None,
+                 audio: Union[AudioInput, None] = None,
+                 do_train: bool = False,
+                 add_generation_prompt: bool = False,
+                 **kwargs: Unpack[QualityvProcessingKwargs]
+                 ):
+        '''
+        input
+            messages: list of dicts
+                example:
+                    [
+                        {"role": "user"
+                        "content": [
+                            {"type": "text", "text": "Hello, how are you?"},
+                            {"type": "image", "image":xxx)},
+                            {"type": "video", "video": xxx},
+                            ]
+                        },
+                        ...    
+                    ]
+        output:
+            input_ids
+            attention_mask
+            pixel_values,
+            pixel_values_videos
+            audio_values
+            labels, default None,
+        '''
+        input_ids = []
+        pixel_values = []
+        pixel_values_videos = []
+        audio_values = []
+        labels = None
+        
+        if not text and not messages:
+            raise ValueError("At least one of text or messages must be provided.")
+        if messages:
+            text = self.apply_chat_template(messages, add_generation_prompt=add_generation_prompt,
+                                            tokenize=False)
+        if isinstance(text, list):
+            text = text[0]
+        image_list = self.fill_modal_list(self.image_token, "image", messages, images, text)
+        image_list = self.process_str_in_modal_list(image_list, "image", **kwargs.get("images_kwargs", {}))
+        # replace image_token with num_images * num_image_token * image_token
+        if image_list and self.image_token in text:
+            tokens_per_image = kwargs.get("images_kwargs", {}).get("tokens_per_image", 197)
+            text = text.replace(self.image_token, tokens_per_image * self.image_token)
+            pixel_values = self.image_processor(images=image_list, return_tensors="pt")["pixel_values"]
+        
+        video_list = self.fill_modal_list(self.video_token, "video", messages, videos, text)
+        video_list = self.process_str_in_modal_list(video_list, "video", **kwargs.get("videos_kwargs", {}))
+        # replace video_token with num_videos * num_video_token * video_token
+        if video_list and self.video_token in text:
+            tokens_per_frame = kwargs.get("videos_kwargs", {}).get("tokens_per_frame", 197)
+            video_frame_list = []
+            for video, video_meta in video_list:
+                num_frames = video.shape[0]
+                replace_text = num_frames * tokens_per_frame * self.video_token
+                text = text.replace(self.video_token, replace_text, 1)
+                for frame in video:
+                    video_frame_list.append(frame)
+            pixel_values_videos = self.image_processor(images=video_frame_list, return_tensors="pt")["pixel_values"]
+        
+        audio_list = self.fill_modal_list(self.audio_token, "audio", messages, audio, text)
+        audio_list = self.process_str_in_modal_list(audio_list, "audio", **kwargs.get("audio_kwargs", {}))
+        # replace audio_token with num_audio_tokens * audio_token
+        if audio_list and self.audio_token in text:
+            audio_kwargs = kwargs.get("audio_kwargs", {})
+            sampling_rate = audio_kwargs.get("sampling_rate", 16000)
+            tokens_per_audio = audio_kwargs.get("tokens_per_audio", 1500)
+            for audio in audio_list:
+                replace_text = tokens_per_audio * self.audio_token
+                text = text.replace(self.audio_token, replace_text, 1)
+            audio_values = self.audio_processor(audio_list, return_tensors="pt", sampling_rate=sampling_rate)["input_features"]
+        
+        input_ids = self.tokenizer(text).input_ids
+        if do_train:
+            labels = self.get_labels(input_ids)
+            labels = torch.tensor(labels, dtype=torch.long)
+        input_ids = torch.tensor(input_ids, dtype=torch.long)
+        return {
+            "input_ids": input_ids,
+            "pixel_values": pixel_values if len(pixel_values) > 0 else None,
+            "pixel_values_videos": pixel_values_videos if len(pixel_values_videos) > 0 else None,
+            "audio_values": audio_values if len(audio_values) > 0 else None,
+            "labels": labels
+        }
+         
+    def fill_modal_list(self, modal_token: str, model_type: str, messages: List[Dict], modal_values: Union[AudioInput, VideoInput, ImageInput, None], text: str) -> List[Union[AudioInput, VideoInput, ImageInput]]:
+        modal_list = []
+        if modal_token in text:
+            if not modal_values and messages:
+                for msg in messages:
+                    if msg.get("role") == "user":
+                        for content in msg.get("content", []):
+                            if content.get('type') == model_type:
+                                modal_list.append(content.get(model_type))
+            elif modal_values:
+                if isinstance(modal_values, str):
+                    modal_list = [modal_values]
+                else:
+                    modal_list = modal_values
+        return modal_list
+    
+    def process_str_in_modal_list(self, modal_list: list, modal_type: str, **modal_kwargs: dict):
+        new_modal_list = []
+        if modal_list:
+            for modal_value in modal_list:
+                if isinstance(modal_value, str):
+                    new_modal_value = self.load_modal_str(modal_value, modal_type, **modal_kwargs)
+                    new_modal_list.append(new_modal_value)
+                else:
+                    new_modal_list.append(modal_value)
+        return new_modal_list
+    
+    def load_modal_str(self, model_path_or_url: str, modal_type: str, **modal_kwargs):
+        if modal_type == "image":
+            load_func = load_image_str
+        elif modal_type == "video":
+            load_func = load_video_str
+        elif modal_type == "audio":
+            load_func = load_audio_str
+        else:
+            raise ValueError(f"Invalid modal type: {modal_type}")
+        return load_func(model_path_or_url, **modal_kwargs)
+    
+    def get_labels(self, input_ids: List[int]) -> List[int]:
+        label_start_token_ids = self.tokenizer(self.label_start_text, add_special_tokens=False)["input_ids"]
+        label_end_token_ids = self.tokenizer(self.label_end_text, add_special_tokens=False)["input_ids"]
+
+        labels = [-100] * len(input_ids)
+
+        i = 0
+        while i < len(input_ids):
+            # Look for the assistant's response start marker.
+            if input_ids[i : i + len(label_start_token_ids)] == label_start_token_ids:
+                # The actual response begins after the start marker.
+                start_response = i + len(label_start_token_ids)
+                # Now, search for the end marker.
+                j = start_response
+                found_end = False
+                while j < len(input_ids):
+                    if input_ids[j : j + len(label_end_token_ids)] == label_end_token_ids:
+                        end_response = j + len(label_end_token_ids)  # Mark the end of the response (excluding the end marker)
+                        found_end = True
+                        break
+                    j += 1
+
+                if found_end:
+                    # Copy the tokens corresponding to the assistant's response into labels.
+                    labels[start_response:end_response] = input_ids[start_response:end_response]
+                    # Advance i beyond the end marker.
+                    i = end_response
+                    continue  # Continue scanning for the next assistant response.
+                else:
+                    # If no end marker is found, break out of the loop.
+                    break
+            else:
+                i += 1
+        pad_token_id = self.tokenizer.pad_token_id
+        if pad_token_id is not None:
+            for i in range(len(labels)):
+                if labels[i] == pad_token_id:
+                    labels[i] = -100
+        return labels
+    
+    def decode(self, *args, **kwargs):
+        return self.tokenizer.decode(*args, **kwargs)
+    
+    def batch_decode(self, *args, **kwargs):
+        return self.tokenizer.batch_decode(*args, **kwargs)
+    
+    
+