| 654 | |
| 655 | |
| 656 | def default_multimodal_input_loader( |
| 657 | *, |
| 658 | tokenizer: Optional[Union[TransformersTokenizer, TokenizerBase]], |
| 659 | model_dir: str, |
| 660 | model_type: str, |
| 661 | modality: str, |
| 662 | prompts: List[str], |
| 663 | media: Optional[Union[List[str], List[List[str]]]] = None, |
| 664 | image_data_format: str = "pt", |
| 665 | num_frames: int = 8, |
| 666 | mm_embeddings: Optional[Union[List[torch.Tensor], |
| 667 | List[List[torch.Tensor]]]] = None, |
| 668 | device: str = "cpu") -> List[dict[str, Union[str, torch.Tensor]]]: |
| 669 | |
| 670 | def convert_to_conversation_message( |
| 671 | prompt: str, |
| 672 | media: Union[Any, List[Any]], |
| 673 | modality: str, |
| 674 | is_embedding: bool = False, |
| 675 | ) -> ConversationMessage: |
| 676 | if isinstance(media, str): |
| 677 | media = [media] |
| 678 | if modality in ["image", "multiple_image"]: |
| 679 | if is_embedding: |
| 680 | _load = lambda mm: mm |
| 681 | |
| 682 | # each mm_embedding corresponds to each image placeholder |
| 683 | if not isinstance(media, list): |
| 684 | media = [media] |
| 685 | else: |
| 686 | _load = lambda mm: load_image( |
| 687 | mm, format=image_data_format, device=device) |
| 688 | |
| 689 | mm_data = [ |
| 690 | MultimodalData(modality=modality, |
| 691 | data=_load(mm), |
| 692 | is_embedding=is_embedding) for mm in media |
| 693 | ] |
| 694 | elif modality == "video": |
| 695 | if is_embedding: |
| 696 | raise ValueError( |
| 697 | "External embedding is not supported for video modality yet." |
| 698 | ) |
| 699 | mm_data = [ |
| 700 | MultimodalData( |
| 701 | modality=modality, |
| 702 | data=load_video(i, |
| 703 | num_frames, |
| 704 | format=image_data_format, |
| 705 | device=device), |
| 706 | is_embedding=False, |
| 707 | ) for i in media |
| 708 | ] |
| 709 | elif modality == "audio": |
| 710 | if is_embedding: |
| 711 | raise ValueError( |
| 712 | "External embedding is not supported for audio modality yet." |
| 713 | ) |