Process multimodal inputs including audio.
(
self,
images: list["ImageInput"],
videos: list["VideoInput"],
audios: list["AudioInput"],
processor: "MMProcessor",
imglens: Optional[list[int]] = None,
)
| 99 | |
| 100 | @override |
| 101 | def _get_mm_inputs( |
| 102 | self, |
| 103 | images: list["ImageInput"], |
| 104 | videos: list["VideoInput"], |
| 105 | audios: list["AudioInput"], |
| 106 | processor: "MMProcessor", |
| 107 | imglens: Optional[list[int]] = None, |
| 108 | ) -> dict[str, "torch.Tensor"]: |
| 109 | """Process multimodal inputs including audio.""" |
| 110 | |
| 111 | # Load audio files if they are paths |
| 112 | if len(audios) > 0 and (isinstance(audios[0], str) or isinstance(audios[0], list)): |
| 113 | audios = FunAudioChatPlugin.load_audio_batch( |
| 114 | audios, |
| 115 | sampling_rate=getattr(processor, "audio_sampling_rate", 16000), |
| 116 | audio_reader=self.audio_reader |
| 117 | ) |
| 118 | |
| 119 | mm_inputs = {} |
| 120 | |
| 121 | # Process audios |
| 122 | if len(audios) != 0: |
| 123 | feature_extractor = getattr(processor, "feature_extractor", None) |
| 124 | audios = self._regularize_audios( |
| 125 | audios, |
| 126 | sampling_rate=getattr(processor, "audio_sampling_rate", 16000), |
| 127 | )["audios"] |
| 128 | |
| 129 | with torch.inference_mode(): |
| 130 | mm_inputs.update( |
| 131 | feature_extractor( |
| 132 | audios, |
| 133 | sampling_rate=getattr(processor, "audio_sampling_rate", 16000), |
| 134 | return_attention_mask=True, |
| 135 | padding=True, |
| 136 | return_tensors="pt", |
| 137 | ) |
| 138 | ) |
| 139 | mm_inputs["feature_attention_mask"] = mm_inputs.pop("attention_mask") |
| 140 | |
| 141 | # Fix length mismatch between input_features and feature_attention_mask |
| 142 | input_features = mm_inputs["input_features"] |
| 143 | feature_attention_mask = mm_inputs["feature_attention_mask"] |
| 144 | |
| 145 | input_seq_len = input_features.shape[-1] # [batch, feature_dim, seq_len] |
| 146 | mask_seq_len = feature_attention_mask.shape[-1] # [batch, seq_len] |
| 147 | |
| 148 | if input_seq_len != mask_seq_len: |
| 149 | min_seq_len = min(input_seq_len, mask_seq_len) |
| 150 | input_features = input_features[..., :min_seq_len] |
| 151 | feature_attention_mask = feature_attention_mask[..., :min_seq_len] |
| 152 | mm_inputs["input_features"] = input_features |
| 153 | mm_inputs["feature_attention_mask"] = feature_attention_mask |
| 154 | |
| 155 | return mm_inputs |
| 156 | |
| 157 | @override |
| 158 | def process_messages( |
no test coverage detected