diff --git a/src/transformers/models/minicpm_o_2_6/feature_extractor_minicpm_o_2_6.py b/src/transformers/models/minicpm_o_2_6/feature_extractor_minicpm_o_2_6.py
index c77702731096..897945af4535 100644
--- a/src/transformers/models/minicpm_o_2_6/feature_extractor_minicpm_o_2_6.py
+++ b/src/transformers/models/minicpm_o_2_6/feature_extractor_minicpm_o_2_6.py
@@ -36,6 +36,7 @@ def __call__(
chunk_length: Optional[int] = 1,
**kwargs,
):
+ # in batch inference, it may be [[]]
if isinstance(audios, np.ndarray):
audios_list = [[audios]]
elif isinstance(audios[0], np.ndarray):
diff --git a/src/transformers/models/minicpm_o_2_6/image_processing_minicpm.py b/src/transformers/models/minicpm_o_2_6/image_processing_minicpm.py
index 9a2e94cf84aa..05880e5e4443 100755
--- a/src/transformers/models/minicpm_o_2_6/image_processing_minicpm.py
+++ b/src/transformers/models/minicpm_o_2_6/image_processing_minicpm.py
@@ -259,7 +259,13 @@ def preprocess(
return_tensors: Optional[Union[str, TensorType]] = None,
**kwargs,
) -> MiniCPMOBatchFeature:
- images_list = make_nested_list_of_images(images)
+ # in batch inference, it may be [[]], so we can't use `make_nested_list_of_images`
+ if isinstance(images, Image.Image):
+ images_list = [[images]]
+ elif isinstance(images[0], Image.Image):
+ images_list = [images]
+ else:
+ images_list = images
to_tensor = transforms.ToTensor()
normalize_transform = transforms.Normalize(
@@ -308,7 +314,9 @@ def preprocess(
(slice_image.shape[1] // self.patch_size, slice_image.shape[2] // self.patch_size))
)
- tgt_sizes = np.vstack(tgt_sizes)
+ # in batch inference, it may be []
+ if tgt_sizes:
+ tgt_sizes = np.vstack(tgt_sizes)
new_images_list.append(new_images)
image_sizes_list.append(image_sizes)
diff --git a/src/transformers/models/minicpm_o_2_6/modeling_minicpm_o_2_6.py b/src/transformers/models/minicpm_o_2_6/modeling_minicpm_o_2_6.py
index 0d136c880812..bcb47d7adbac 100644
--- a/src/transformers/models/minicpm_o_2_6/modeling_minicpm_o_2_6.py
+++ b/src/transformers/models/minicpm_o_2_6/modeling_minicpm_o_2_6.py
@@ -45,6 +45,7 @@
from transformers.utils import ModelOutput, add_start_docstrings, add_start_docstrings_to_model_forward, is_flash_attn_2_available, logging, replace_return_docstrings, can_return_tuple, auto_docstring, LossKwargs
from transformers.cache_utils import Cache, DynamicCache, EncoderDecoderCache, StaticCache
from transformers.generation import GenerationMixin
+from transformers.generation.utils import GenerateOutput
from transformers.generation.logits_process import LogitsProcessor, TopKLogitsWarper, TopPLogitsWarper
from transformers.masking_utils import create_causal_mask, create_sliding_window_causal_mask
from transformers.models.siglip.configuration_siglip import SiglipVisionConfig
@@ -78,6 +79,7 @@
@dataclass
class OmniOutput(ModelOutput):
text: Optional[Union[str, List[str], Iterator]] = None
+ outputs: (GenerateOutput | torch.LongTensor) = None
spk_embeds: Optional[torch.FloatTensor] = None
audio_wav: Optional[np.ndarray] = None
sampling_rate: Optional[int] = None
@@ -549,7 +551,7 @@ def __init__(self, config):
# feature_extractor = MiniCPM_o_2_6FeatureExtractor.from_pretrained(config._name_or_path)
# self.processor = MiniCPM_o_2_6Processor(image_processor=image_processor, feature_extractor=feature_extractor, tokenizer=tokenizer)
- self.terminators = ["<|im_end|>", "<|endoftext|>"]
+ # self.terminators = ["<|im_end|>", "<|endoftext|>"]
self.force_no_stop = False
@@ -1019,54 +1021,6 @@ def get_omni_embedding(self, data, input_embeddings, chunk_length=-1, stream_inp
return input_embeddings
- # def forward(self, data = None, **kwargs):
- # r"""
- # Perform a forward pass with multimodal inputs including vision and optional audio features.
-
- # Args:
- # data (`Dict`):
- # A dictionary containing multimodal inputs required to compute embeddings:
-
- # - `pixel_values` (`torch.FloatTensor` of shape `(B, C, H, W)`): Image tensors used for visual embedding.
- # - `tgt_sizes` (`torch.LongTensor` of shape `(num_images, 2)`): Target spatial size (H, W) after patch embedding.
- # - `image_bound` (`torch.LongTensor` of shape `(num_images, 2)`): Token index boundaries in the input sequence for each image.
- # - `input_ids` (`torch.LongTensor`): Tokenized input ids including placeholders for multimodal tokens.
- # - `position_ids` (`torch.LongTensor`): Positional indices, will be cast to `torch.int64` if needed.
- # - `audio_values` (`torch.FloatTensor`, *optional*): Optional raw audio input, used if `config.init_audio` is True.
-
- # **kwargs:
- # Additional standard generation/model arguments passed to the underlying language model, e.g.:
- # - `attention_mask`
- # - `past_key_values`
- # - `use_cache`
- # - `output_attentions`
- # - `output_hidden_states`
- # - `return_dict`
-
- # Returns:
- # Model outputs from the underlying language model, depending on `return_dict`.
- # """
- # output_attentions = kwargs.pop('output_attentions') if kwargs.get('output_attentions', None) is not None else self.config.output_attentions
- # output_hidden_states = (
- # kwargs.pop('output_hidden_states') if kwargs.get('output_hidden_states', None) is not None else self.config.output_hidden_states
- # )
- # outputs = self.language_model(
- # output_attentions=output_attentions,
- # output_hidden_states=output_hidden_states,
- # **kwargs
- # )
- # hidden_states = outputs.last_hidden_state
- # # Only compute necessary logits, and do not upcast them to float if we are not computing the loss
- # slice_indices = slice(0, None)
- # logits = self.lm_head(hidden_states[:, slice_indices, :])
-
- # return CausalLMOutputWithPast(
- # loss=None,
- # logits=logits,
- # past_key_values=outputs.past_key_values,
- # hidden_states=outputs.hidden_states,
- # attentions=outputs.attentions,
- # )
def forward(
self,
input_ids: torch.LongTensor = None,
@@ -1140,26 +1094,12 @@ def forward(
attentions=outputs.attentions,
)
- def _decode(self, inputs_embeds, tokenizer, attention_mask, **kwargs):
- terminators = [tokenizer.convert_tokens_to_ids(i) for i in self.terminators]
- outputs = super().generate(
- inputs_embeds=inputs_embeds,
- pad_token_id=0,
- eos_token_id=terminators,
- attention_mask=attention_mask,
- output_hidden_states=True,
- return_dict_in_generate=True,
- **kwargs,
- )
- return outputs
-
def _decode_stream(self, inputs_embeds, tokenizer, **kwargs):
- terminators = [tokenizer.convert_tokens_to_ids(i) for i in self.terminators]
streamer = TextIteratorStreamer(tokenizer=tokenizer)
generation_kwargs = {
"inputs_embeds": inputs_embeds,
"pad_token_id": 0,
- "eos_token_id": terminators,
+ "eos_token_id": tokenizer.terminator_ids,
"streamer": streamer,
}
generation_kwargs.update(kwargs)
@@ -1180,13 +1120,44 @@ def generate(
audio_bounds=None,
spk_bounds=None,
attention_mask=None,
- tokenizer=None,
+ processor=None,
vision_hidden_states=None,
stream=False,
+ generate_audio=False,
+ return_spk_embed=False,
+ return_dict=False,
+ batched=False,
+ use_tts_template=False,
+ sampling=False,
+ max_new_tokens=2048,
+ min_new_tokens=0,
+ force_no_stop=False,
**kwargs,
):
assert input_ids is not None
assert len(input_ids) == len(pixel_values)
+ assert sampling or not stream, "if use stream mode, make sure sampling=True"
+
+ kwargs.pop("image_sizes")
+ if sampling:
+ generation_config = {
+ "top_p": 0.8,
+ "top_k": 100,
+ "temperature": 0.7,
+ "do_sample": True,
+ "repetition_penalty": 1.05,
+ }
+ else:
+ generation_config = {
+ "num_beams": 3,
+ "repetition_penalty": 1.2,
+ }
+ generation_config["max_new_tokens"] = max_new_tokens
+ generation_config["min_new_tokens"] = min_new_tokens
+ generation_config.update((k, kwargs[k]) for k in generation_config.keys() & kwargs.keys())
+
+ if generate_audio or return_spk_embed:
+ return_dict = True
model_inputs = {
"input_ids": input_ids,
@@ -1212,213 +1183,22 @@ def generate(
)
if stream:
- result = self._decode_stream(model_inputs["inputs_embeds"], tokenizer, **kwargs)
- # if stream return TextIteratorStreamer and output is empty
- outputs = {}
+ result = self._decode_stream(model_inputs["inputs_embeds"], processor.tokenizer, **generation_config)
else:
- terminators = [tokenizer.convert_tokens_to_ids(i) for i in self.terminators]
outputs = super().generate(
inputs_embeds=model_inputs["inputs_embeds"],
pad_token_id=0,
- eos_token_id=terminators,
+ eos_token_id=processor.tokenizer.terminator_ids,
attention_mask=attention_mask,
output_hidden_states=True,
return_dict_in_generate=True,
- **kwargs,
- )
- result = self.processor.decode_text(outputs.sequences, tokenizer, self.terminators)
-
- return result, outputs
-
- def chat(
- self,
- image=None,
- msgs=None,
- tokenizer=None,
- processor=None,
- vision_hidden_states=None,
- max_new_tokens=2048,
- min_new_tokens=0,
- sampling=True,
- max_inp_length=32768,
- stream=False,
- chunk_input=True,
- omni_input=False,
- max_slice_nums=None,
- use_image_id=None,
- use_tts_template=False,
- generate_audio=False,
- return_spk_embed=False,
- return_dict=False,
- output_audio_path=None,
- **kwargs,
- ):
- """
- Unified chat function
-
- Args:
- image: use for batch_size=1 vqa, It is not recommended to continue to use this parameter
- msgs: the input chat msgs, support text: (string) / image: (PIL.Image) / audio (numpy.ndarray)
- tokenizer: tokenizer for llm
- processor: if None, use the default processor
- max_new_tokens: the maximum length of the generation
- min_new_tokens: the minimum length of the generation
- sampling: whether to use sampling decoding or beam search decoding
- max_inp_length: the maximum length of input
- stream: whether to return generator, only used when tts is not required
- chunk_input: whether to split audio into 1s chunks
- omni_input: determine whether it is omni mode
- max_slice_nums: control the maximum number of image slices
- use_image_id: for video understanding or omni understanding, use_image_id should be False
- use_tts_template: if the msgs contain audio, use_tts_template should be True
- generate_audio: whether to generate audio output, only used when return_dict=True
- return_spk_embed: whether to return spk embedding, only used when return_dict=True
- return_dict: whether to return dict
- output_audio_path: audio save path when generate_audio
- **kwargs:
- """
- if isinstance(msgs[0], list):
- batched = True
- else:
- batched = False
-
- if generate_audio or return_spk_embed:
- return_dict = True
-
- msgs_list = msgs
- images_list = image
-
- if batched is False:
- images_list, msgs_list = [images_list], [msgs_list]
- else:
- assert images_list is None, "Please integrate image to msgs when using batch inference."
- images_list = [None] * len(msgs_list)
- assert len(images_list) == len(msgs_list), "The batch dim of images_list and msgs_list should be the same."
-
- if processor is None:
- processor = self.processor
-
- assert (
- self.config.query_num == processor.image_processor.image_feature_size
- ), "These two values should be the same. Check `config.json` and `preprocessor_config.json`."
- assert (
- self.config.patch_size == processor.image_processor.patch_size
- ), "These two values should be the same. Check `config.json` and `preprocessor_config.json`."
- assert (
- self.config.use_image_id == processor.image_processor.use_image_id
- ), "These two values should be the same. Check `config.json` and `preprocessor_config.json`."
- assert (
- self.config.slice_config.max_slice_nums == processor.image_processor.max_slice_nums
- ), "These two values should be the same. Check `config.json` and `preprocessor_config.json`."
- assert (
- self.config.slice_mode == processor.image_processor.slice_mode
- ), "These two values should be the same. Check `config.json` and `preprocessor_config.json`."
-
- prompts_lists = []
- input_images_list = []
- input_audios_list = []
- audio_parts_list = []
-
- for image, msgs in zip(images_list, msgs_list):
- if isinstance(msgs, str):
- msgs = json.loads(msgs)
- copy_msgs = deepcopy(msgs)
-
- assert len(msgs) > 0, "msgs is empty"
- assert sampling or not stream, "if use stream mode, make sure sampling=True"
-
- if image is not None and isinstance(copy_msgs[0]["content"], str):
- copy_msgs[0]["content"] = [image, copy_msgs[0]["content"]]
-
- images = []
- audios = []
- audio_parts = []
- for i, msg in enumerate(copy_msgs):
- role = msg["role"]
- content = msg["content"]
- assert role in ["system", "user", "assistant"]
- if i == 0:
- assert role in ["user", "system"], "The role of first msg should be user"
- if isinstance(content, str):
- content = [content]
- cur_msgs = []
- for c in content:
- if isinstance(c, Image.Image):
- images.append(c)
- cur_msgs.append("(./)")
- elif isinstance(c, np.ndarray): # audio
- audios.append(c)
- audio_parts.append(i)
- cur_msgs.append("()")
- use_tts_template = True
- elif isinstance(c, str):
- cur_msgs.append(c)
- if omni_input:
- msg["content"] = "".join(cur_msgs)
- else:
- msg["content"] = "\n".join(cur_msgs)
-
- prompts_lists.append(
- processor.tokenizer.apply_chat_template(
- copy_msgs,
- tokenize=False,
- add_generation_prompt=True,
- chat_template=self.processor.default_tts_chat_template if use_tts_template else None,
+ **generation_config,
)
- )
- if images:
- input_images_list.append(images)
- if audios:
- input_audios_list.append(audios)
- audio_parts_list.append(audio_parts)
-
- inputs = processor(
- prompts_lists,
- input_images_list,
- input_audios_list,
- audio_parts=audio_parts_list,
- max_slice_nums=max_slice_nums,
- use_image_id=use_image_id,
- chunk_input=chunk_input,
- return_tensors="pt",
- max_length=max_inp_length,
- ).to(self.device)
-
- if sampling:
- generation_config = {
- "top_p": 0.8,
- "top_k": 100,
- "temperature": 0.7,
- "do_sample": True,
- "repetition_penalty": 1.05,
- }
- else:
- generation_config = {
- "num_beams": 3,
- "repetition_penalty": 1.2,
- }
-
- if min_new_tokens > 0:
- generation_config["min_new_tokens"] = min_new_tokens
-
- generation_config.update((k, kwargs[k]) for k in generation_config.keys() & kwargs.keys())
-
- inputs.pop("image_sizes")
- with torch.inference_mode():
- res, outputs = self.generate(
- **inputs,
- tokenizer=tokenizer,
- max_new_tokens=max_new_tokens,
- vision_hidden_states=vision_hidden_states,
- stream=stream,
- **generation_config,
- )
if stream:
-
def stream_gen():
- for text in res:
- for term in self.terminators:
+ for text in result:
+ for term in processor.tokenizer.terminators:
text = text.replace(term, "")
yield text
@@ -1429,35 +1209,27 @@ def stream_gen():
else:
spk_embeds = wav_numpy = sr = None
-
- if batched:
- answer = res
- else:
- answer = res[0]
-
- if use_tts_template and generate_audio:
- mel_spec = self._generate_mel_spec(inputs, outputs, answer, tts_config={'top_p': 0.7, 'top_k': 20, 'repetition_penalty': 1.0})
- wav_numpy, sr = self.decode_mel_to_audio(mel_spec, output_audio_path)
+
+ if not batched and use_tts_template and generate_audio:
+ result = processor.decode_text(outputs.sequences, processor.tokenizer)
+ mel_spec = self._generate_mel_spec(model_inputs, outputs, result[0], tts_config={'top_p': 0.7, 'top_k': 20, 'repetition_penalty': 1.0}, force_no_stop=force_no_stop)
+ wav_numpy, sr = self.decode_mel_to_audio(mel_spec, kwargs.get('output_audio_path', None))
if return_spk_embed:
- spk_embeds = self._get_last_spk_embeds(inputs, outputs)
-
- if isinstance(answer, list):
- answer = [i.replace(tokenizer.tts_end, "") for i in answer]
- else:
- answer = answer.replace(tokenizer.tts_end, "")
+ spk_embeds = self._get_last_spk_embeds(model_inputs, outputs)
if return_dict:
- return OmniOutput(text=answer, spk_embeds=spk_embeds, audio_wav=wav_numpy, sampling_rate=sr)
+ return OmniOutput(outputs=outputs, spk_embeds=spk_embeds, audio_wav=wav_numpy, sampling_rate=sr)
else:
- return answer
+ return outputs
+
@torch.inference_mode()
def streaming_prefill(
self,
session_id,
msgs,
- tokenizer,
+ processor,
omni_input=True,
max_slice_nums=None,
ls_temperature=1.0,
@@ -1520,15 +1292,15 @@ def streaming_prefill(
self.reset_session()
self.session_id = session_id
- prompt = tokenizer.apply_chat_template(
- copy_msgs, tokenize=False, add_generation_prompt=False, chat_template=self.processor.default_tts_chat_template
+ prompt = processor.tokenizer.apply_chat_template(
+ copy_msgs, tokenize=False, add_generation_prompt=False, chat_template=processor.default_tts_chat_template
)
add_special_tokens = True # add bos
else:
prompt = copy_msgs[0]["content"]
add_special_tokens = False
- model_inputs = self.processor(
+ model_inputs = processor(
[prompt],
[images] if images else None,
[audios] if audios else None,
@@ -1661,18 +1433,20 @@ def prepare_inputs_for_generation(
def streaming_generate(
self,
session_id,
- tokenizer,
+ processor,
max_new_tokens=512,
min_new_tokens=0,
sampling=True,
generate_audio=True,
enable_regenerate=False,
+ force_no_stop=False,
**kwargs,
):
"""
Streaming video/audio input and output audio stream
Args:
"""
+ tokenizer = processor.tokenizer
if sampling:
generation_config = {
"top_p": 0.8,
@@ -1696,7 +1470,6 @@ def streaming_generate(
self.llm_generate_completed = False
self.audio_past_key_values = None # apm kv cache
- terminators = [tokenizer.convert_tokens_to_ids(i) for i in self.terminators]
generate_prompt = "<|im_end|>\n<|im_start|>assistant\n<|spk_bos|><|spk|><|spk_eos|><|tts_bos|>"
input_ids = tokenizer(generate_prompt, return_tensors="pt", add_special_tokens=False)["input_ids"].cuda()
@@ -1710,11 +1483,12 @@ def streaming_generate(
attention_mask = torch.ones((1, cache_length + input_ids.shape[1]), dtype=torch.bool, device=self.device)
generation_config["max_new_tokens"] = max_new_tokens
- streamer = self.llm_generate_chunk(input_ids, attention_mask, tokenizer, terminators, generation_config)
+ streamer = self.llm_generate_chunk(input_ids, attention_mask, tokenizer, tokenizer.terminator_ids, generation_config)
if generate_audio:
result = self._generate_mel_spec_audio_streaming(
- spk_bounds, streamer, output_chunk_size=25, enable_regenerate=enable_regenerate, tts_config={"top_p": 0.7, "top_k": 20, "repetition_penalty": 1.0}
+ spk_bounds, streamer, output_chunk_size=25, enable_regenerate=enable_regenerate,
+ tts_config={"top_p": 0.7, "top_k": 20, "repetition_penalty": 1.0}, force_no_stop=force_no_stop
)
return result
else:
@@ -1761,7 +1535,7 @@ def check_uncompleted_token(ids):
end = check_uncompleted_token(cur_ids[0])
left_ids = cur_ids[:, end:]
cur_ids = cur_ids[:, :end]
- text = self.processor.decode_text(cur_ids, tokenizer, self.terminators)[0] if end > 0 else ""
+ text = self.processor.decode_text(cur_ids, tokenizer)[0] if end > 0 else ""
self.llm_past_key_values = outputs.past_key_values
input_ids = outputs.sequences[:, -1:]
@@ -1826,7 +1600,7 @@ def _get_last_spk_embeds(self, inputs, outputs):
spk_embeds = last_hidden_states[spk_bound[0] : spk_bound[1]]
return spk_embeds
- def _generate_mel_spec(self, inputs, outputs, text, output_chunk_size=25, tts_max_new_tokens=2048, tts_config: dict = {"top_p": 0.7, "top_k": 20, "repetition_penalty": 1.0}):
+ def _generate_mel_spec(self, inputs, outputs, text, force_no_stop=False, output_chunk_size=25, tts_max_new_tokens=2048, tts_config: dict = {"top_p": 0.7, "top_k": 20, "repetition_penalty": 1.0}):
spk_embeds = self._get_last_spk_embeds(inputs, outputs)
text = text.split("<|tts_bos|>")[-1]
@@ -1913,7 +1687,7 @@ def _generate_mel_spec(self, inputs, outputs, text, output_chunk_size=25, tts_ma
past_key_values=past_key_values,
streaming_tts_text_mask=streaming_tts_text_mask,
max_new_token=output_chunk_size,
- force_no_stop=self.force_no_stop,
+ force_no_stop=force_no_stop,
temperature=torch.tensor([0.1, 0.3, 0.1, 0.3], dtype=torch.float, device=self.tts.device),
eos_token=torch.tensor([625], dtype=torch.long, device=self.tts.device),
logits_warpers=logits_warpers,
@@ -1935,7 +1709,7 @@ def _generate_mel_spec(self, inputs, outputs, text, output_chunk_size=25, tts_ma
past_key_values=past_key_values,
streaming_tts_text_mask=streaming_tts_text_mask,
max_new_token=output_chunk_size,
- force_no_stop=self.force_no_stop,
+ force_no_stop=force_no_stop,
temperature=torch.tensor([0.1, 0.3, 0.1, 0.3], dtype=torch.float, device=self.tts.device),
eos_token=torch.tensor([625], dtype=torch.long, device=self.tts.device),
logits_warpers=logits_warpers,
@@ -1996,6 +1770,7 @@ def _generate_mel_spec_audio_streaming(
prev_seg_text_left="",
prev_seg_audio_ids=None,
enable_regenerate=False,
+ force_no_stop=False,
tts_config: dict = {"top_p": 0.7, "top_k": 20, "repetition_penalty": 1.0},
):
# get spk_embedding
@@ -2147,7 +1922,7 @@ def _generate_mel_spec_audio_streaming(
past_key_values=past_key_values,
streaming_tts_text_mask=streaming_tts_text_mask,
max_new_token=output_chunk_size,
- force_no_stop=self.force_no_stop,
+ force_no_stop=force_no_stop,
temperature=torch.tensor([0.1, 0.3, 0.1, 0.3], dtype=torch.float, device=self.tts.device),
eos_token=torch.tensor([625], dtype=torch.long, device=self.tts.device),
logits_warpers=logits_warpers,
@@ -2188,7 +1963,7 @@ def _generate_mel_spec_audio_streaming(
past_key_values=past_kv,
streaming_tts_text_mask=streaming_tts_text_mask,
max_new_token=N,
- force_no_stop=self.force_no_stop,
+ force_no_stop=force_no_stop,
temperature=torch.tensor([0.1, 0.3, 0.1, 0.3], dtype=torch.float, device=self.tts.device),
eos_token=torch.tensor([625], dtype=torch.long, device=self.tts.device),
logits_warpers=logits_warpers,
@@ -2259,7 +2034,7 @@ def _generate_mel_spec_audio_streaming(
past_key_values=past_key_values,
streaming_tts_text_mask=streaming_tts_text_mask,
max_new_token=output_chunk_size,
- force_no_stop=self.force_no_stop,
+ force_no_stop=force_no_stop,
# temperature=torch.tensor([0.1] * self.tts.num_vq, dtype=torch.float, device=self.tts.device),
temperature=torch.tensor([0.1, 0.3, 0.1, 0.3], dtype=torch.float, device=self.tts.device),
eos_token=torch.tensor([625], dtype=torch.long, device=self.tts.device),
@@ -2299,7 +2074,7 @@ def _generate_mel_spec_audio_streaming(
past_key_values=past_kv,
streaming_tts_text_mask=streaming_tts_text_mask,
max_new_token=N,
- force_no_stop=self.force_no_stop,
+ force_no_stop=force_no_stop,
temperature=torch.tensor([0.1, 0.3, 0.1, 0.3], dtype=torch.float, device=self.tts.device),
eos_token=torch.tensor([625], dtype=torch.long, device=self.tts.device),
logits_warpers=logits_warpers,
@@ -2363,7 +2138,8 @@ def _generate_mel_spec_audio_streaming(
text_left,
prev_seg_audio_ids,
enable_regenerate=enable_regenerate,
- tts_config={"top_p": 0.7, "top_k": 20, "repetition_penalty": 1.0}
+ tts_config={"top_p": 0.7, "top_k": 20, "repetition_penalty": 1.0},
+ force_no_stop=force_no_stop,
)
for res in result:
yield res
@@ -2853,6 +2629,7 @@ def __init__(
nn.GELU(),
)
+ # These hardcodings match our model and cannot be modified.
self.encoder = DVAEDecoder(
idim=512,
odim=1024,
@@ -2879,20 +2656,21 @@ def __init__(
)
@torch.inference_mode()
- def forward(self, inp: torch.Tensor, mode: Literal["encode", "decode"] = "decode") -> torch.Tensor:
- if mode == "encode" and hasattr(self, "encoder") and self.vq_layer is not None:
- mel = inp.clone()
- x: torch.Tensor = self.downsample_conv(
- torch.div(mel, self.coef.view(100, 1).expand(mel.shape), out=mel),
- ).unsqueeze_(0)
- del mel
- x = self.encoder(x)
- ind = self.vq_layer(x)
- del x
- return ind
+ def encode(self, mel: torch.Tensor) -> torch.Tensor:
+ """Encode mel spectrogram to VQ indices."""
+ mel = mel.clone()
+ x = self.downsample_conv(torch.div(mel, self.coef.view(100, 1).expand(mel.shape), out=mel))
+ x = x.unsqueeze(0)
+ x = self.encoder(x)
+ ind = self.vq_layer(x)
+ del x
+ return ind
+ def decode(self, inp: torch.Tensor) -> torch.Tensor:
+ """Decode VQ inputs (either indices or embeddings) back to spectrogram."""
vq_feats = self.vq_layer._embed(inp)
+ # reshape embeddings
vq_feats = (
vq_feats.view(
(vq_feats.size(0), 2, vq_feats.size(1) // 2, vq_feats.size(2)),
@@ -2901,16 +2679,19 @@ def forward(self, inp: torch.Tensor, mode: Literal["encode", "decode"] = "decode
.flatten(2)
)
- dec_out = self.out_conv(
- self.decoder(
- x=vq_feats,
- ),
- )
-
+ dec_out = self.decoder(x=vq_feats)
+ dec_out = self.out_conv(dec_out)
del vq_feats
-
+ # re-apply coef
return torch.mul(dec_out, self.coef, out=dec_out)
+ def forward(self, inp: torch.Tensor) -> torch.Tensor:
+ """Standard forward = encode + decode → full reconstruction."""
+ indices = self.encode(inp)
+ recon = self.decode(indices)
+ return recon
+
+
def apply_spk_emb(
input_ids: torch.Tensor = None,
spk_emb: torch.Tensor = None,
@@ -3129,7 +2910,7 @@ def _prepare_4d_causal_attention_mask_with_cache_position(
class PatchLlamaModel(LlamaModel):
"""
This is a patch for LlamaModel to support our audio
- Mainly modifies by changing create_causal_mask to _update_causal_mask
+ Mainly modifies by changing `create_causal_mask` to `_update_causal_mask`
"""
@can_return_tuple
@@ -3180,8 +2961,8 @@ def forward(
if position_ids is None:
position_ids = cache_position.unsqueeze(0)
- # ! in transformers>=4.53.1, this is `create_causal_mask`, but it will be wrong in our case
- # so copy `_update_causal_mask` from LlamaModel which transformers=4.44.2
+ # in transformers>=4.53.1, this is `create_causal_mask`, but it will be wrong in our case
+ # so copy `_update_causal_mask` from `LlamaModel` which transformers=4.44.2
causal_mask = self._update_causal_mask(
attention_mask, inputs_embeds, cache_position, past_key_values, output_attentions
)
@@ -3888,7 +3669,7 @@ def decode_to_mel_specs(
batch_result[i].narrow(1, 0, src.size(0)).copy_(src.permute(1, 0))
del src
- mel_specs = decoder(batch_result)
+ mel_specs = decoder.decode(batch_result)
del batch_result
return mel_specs
diff --git a/src/transformers/models/minicpm_o_2_6/processing_minicpm_o_2_6.py b/src/transformers/models/minicpm_o_2_6/processing_minicpm_o_2_6.py
index 96c12ef2fb95..7ed6bc4322b3 100644
--- a/src/transformers/models/minicpm_o_2_6/processing_minicpm_o_2_6.py
+++ b/src/transformers/models/minicpm_o_2_6/processing_minicpm_o_2_6.py
@@ -25,7 +25,10 @@
import numpy as np
import torch
import torchaudio
+import json
+from copy import deepcopy
+from PIL import Image
from transformers.image_utils import ImageInput
from transformers.processing_utils import ProcessorMixin, ProcessingKwargs, Unpack, ImagesKwargs, AudioKwargs
from transformers.tokenization_utils_base import PreTokenizedInput, TextInput
@@ -155,15 +158,11 @@ class MiniCPM_o_2_6Processor(ProcessorMixin):
image_processor_class = "AutoImageProcessor"
feature_extractor_class = "MiniCPM_o_2_6FeatureExtractor"
- def __init__(self, tokenizer=None, image_processor=None, feature_extractor=None):
- super().__init__(tokenizer, image_processor, feature_extractor)
+ def __init__(self, tokenizer=None, image_processor=None, feature_extractor=None, chat_template=None):
+ super().__init__(tokenizer, image_processor,
+ feature_extractor, chat_template=chat_template)
self.version = image_processor.version
self.default_tts_chat_template = "{% for message in messages %}{{'<|im_start|>' + message['role'] + '\n' + message['content'] + '<|im_end|>' + '\n'}}{% endfor %}{% if add_generation_prompt %}{{ '<|im_start|>assistant\n<|spk_bos|><|spk|><|spk_eos|><|tts_bos|>' }}{% endif %}"
- self.image_tag = "(./)"
- self.image_pattern = "\(./\)"
- self.audio_tag = "()"
- self.audio_pattern = "\(\)"
- self.split_pattern = f"({self.image_pattern}|{self.audio_pattern})"
def __call__(
self,
@@ -212,40 +211,130 @@ def __call__(
return MiniCPMOBatchFeature(data={**model_inputs})
- # Copied from transformers.models.clip.processing_clip.CLIPProcessor.batch_decode with CLIP->Llama
- def batch_decode(self, *args, **kwargs):
- """
- This method forwards all its arguments to LlamaTokenizerFast's [`~PreTrainedTokenizer.batch_decode`]. Please
- refer to the docstring of this method for more information.
+ def apply_chat_template(
+ self,
+ msgs,
+ chunk_input=True,
+ max_slice_nums=None,
+ max_length=32768,
+ omni_input=False,
+ use_image_id=None,
+ use_tts_template=False,
+ **kwargs,
+ ):
"""
- output_ids = args[0]
- result_text = []
- for result in output_ids:
- result = result[result != 0]
- result_text.append(self.tokenizer.decode(
- result, *args[1:], **kwargs).strip())
- return result_text
- # return self.tokenizer.batch_decode(*args, **kwargs)
+ Unified chat function
- # Copied from transformers.models.clip.processing_clip.CLIPProcessor.decode with CLIP->Llama
- def decode(self, *args, **kwargs):
- """
- This method forwards all its arguments to LlamaTokenizerFast's [`~PreTrainedTokenizer.decode`]. Please refer to
- the docstring of this method for more information.
+ Args:
+ msgs: the input chat msgs, support text: (string) / image: (PIL.Image) / audio (numpy.ndarray)
+ chunk_input: whether to split audio into 1s chunks
+ max_length: the maximum length of input
+ max_slice_nums: control the maximum number of image slices
+ omni_input: determine whether it is omni mode
+ use_image_id: for video understanding or omni understanding, use_image_id should be False
+ use_tts_template: if the msgs contain audio, use_tts_template should be True
"""
- result = args[0]
- result = result[result != 0]
- return self.tokenizer.decode(result, *args[1:], **kwargs).strip()
+ if isinstance(msgs[0], list):
+ batched = True
+ else:
+ batched = False
+
+ msgs_list = msgs
+ if not batched:
+ msgs_list = [msgs_list]
+
+ prompts_lists = []
+ input_images_list = []
+ input_audios_list = []
+ audio_parts_list = []
+
+ for msgs in msgs_list:
+ if isinstance(msgs, str):
+ msgs = json.loads(msgs)
+ copy_msgs = deepcopy(msgs)
+
+ assert len(msgs) > 0, "msgs is empty"
+
+ images = []
+ audios = []
+ audio_parts = []
+ for i, msg in enumerate(copy_msgs):
+ role = msg["role"]
+ content = msg["content"]
+ assert role in ["system", "user", "assistant"]
+ if i == 0:
+ assert role in ["user", "system"], "The role of first msg should be user"
+ if isinstance(content, str):
+ content = [content]
+ cur_msgs = []
+ for c in content:
+ if isinstance(c, Image.Image):
+ images.append(c)
+ cur_msgs.append("(./)")
+ elif isinstance(c, np.ndarray): # audio
+ audios.append(c)
+ audio_parts.append(i)
+ cur_msgs.append("()")
+ use_tts_template = True
+ elif isinstance(c, str):
+ cur_msgs.append(c)
+ if omni_input:
+ msg["content"] = "".join(cur_msgs)
+ else:
+ msg["content"] = "\n".join(cur_msgs)
+
+ prompts_lists.append(
+ self.tokenizer.apply_chat_template(
+ copy_msgs,
+ tokenize=False,
+ add_generation_prompt=True,
+ chat_template=self.default_tts_chat_template if use_tts_template else None,
+ )
+ )
+ input_images_list.append(images)
+ input_audios_list.append(audios)
+ audio_parts_list.append(audio_parts)
+
+ inputs = self.__call__(
+ prompts_lists,
+ input_images_list,
+ input_audios_list,
+ audio_parts=audio_parts_list,
+ max_slice_nums=max_slice_nums,
+ use_image_id=use_image_id,
+ chunk_input=chunk_input,
+ return_tensors="pt",
+ max_length=max_length,
+ )
+ return inputs
+
+ def decode(self, outputs, batched=False):
+ result = self.decode_text(
+ outputs.sequences, self.tokenizer)
+ if not batched:
+ result = result[0]
+ if isinstance(result, list):
+ result = [i.replace(self.tokenizer.tts_end, "") for i in result]
+ else:
+ result = result.replace(self.tokenizer.tts_end, "")
+ return result
- def decode_text(self, result_ids, tokenizer, terminators):
- terminators = [tokenizer.convert_tokens_to_ids(i) for i in terminators]
+ def decode_text(self, result_ids, tokenizer):
result_text = []
for result in result_ids:
result = result[result != 0]
- if result[0] == tokenizer.bos_id:
- result = result[1:]
- if result[-1] in terminators:
- result = result[:-1]
+ start, end = 0, len(result)
+ for i, tok in enumerate(result):
+ if tok == tokenizer.bos_id:
+ start = i+1
+ else:
+ break
+ for i in range(len(result)-1, -1, -1):
+ if result[i] in tokenizer.terminator_ids:
+ end = i
+ else:
+ break
+ result = result[start:end]
result_text.append(tokenizer.decode(result))
return result_text
@@ -421,10 +510,10 @@ def _convert_omni_to_inputs(
spk_bounds_list = []
for index, text in enumerate(texts):
- text_chunks = re.split(self.split_pattern, text)
+ text_chunks = re.split(self.tokenizer.split_pattern, text)
- image_tags = re.findall(self.image_pattern, text)
- audio_tags = re.findall(self.audio_pattern, text)
+ image_tags = re.findall(self.tokenizer.image_pattern, text)
+ audio_tags = re.findall(self.tokenizer.audio_pattern, text)
if image_tags:
assert images is not None
@@ -436,13 +525,13 @@ def _convert_omni_to_inputs(
image_id = 0
audio_id = 0
for i, chunk in enumerate(text_chunks):
- if chunk == self.image_tag:
+ if chunk == self.tokenizer.image_tag:
image_placeholder = self.image_processor.get_slice_image_placeholder(
self.tokenizer, image_sizes[index][image_id], image_id, max_slice_nums, use_image_id
)
image_id += 1
text_chunks[i] = image_placeholder
- elif chunk == self.audio_tag:
+ elif chunk == self.tokenizer.audio_tag:
audio_placeholder = audio_phs[index][audio_id]
audio_id += 1
text_chunks[i] = audio_placeholder
diff --git a/src/transformers/models/minicpm_o_2_6/tokenization_minicpm_o_2_6_fast.py b/src/transformers/models/minicpm_o_2_6/tokenization_minicpm_o_2_6_fast.py
index 2ec227488fb0..8d943508c40e 100644
--- a/src/transformers/models/minicpm_o_2_6/tokenization_minicpm_o_2_6_fast.py
+++ b/src/transformers/models/minicpm_o_2_6/tokenization_minicpm_o_2_6_fast.py
@@ -15,6 +15,7 @@
from transformers import Qwen2TokenizerFast
+
class MiniCPM_o_2_6TokenizerFast(Qwen2TokenizerFast):
def __init__(self, **kwargs):
super().__init__(**kwargs)
@@ -31,6 +32,8 @@ def __init__(self, **kwargs):
self.slice_end = ""
self.im_id_start = ""
self.im_id_end = ""
+ self.image_tag = f"({self.im_start}./{self.im_end})"
+ self.image_pattern = "\(./\)"
# audio
self.audio_start = "<|audio_start|>"
@@ -40,6 +43,12 @@ def __init__(self, **kwargs):
self.tts_start = "<|tts_bos|>"
self.tts_end = "<|tts_eos|>"
self.unk_token = ""
+ self.audio_tag = "()"
+ self.audio_pattern = "\(\)"
+
+ self.split_pattern = f"({self.image_pattern}|{self.audio_pattern})"
+
+ self.terminator_tokens = ["<|im_end|>", "<|endoftext|>", self.tts_end]
@property
def eos_id(self):
@@ -53,6 +62,10 @@ def bos_id(self):
def unk_id(self):
return self.unk_token_id
+ @property
+ def terminators(self):
+ return self.terminator_tokens
+
@property
def im_start_id(self):
return self.convert_tokens_to_ids(self.im_start)
@@ -101,6 +114,10 @@ def tts_start_id(self):
def tts_end_id(self):
return self.convert_tokens_to_ids(self.tts_end)
+ @property
+ def terminator_ids(self):
+ return [self.convert_tokens_to_ids(t) for t in self.terminator_tokens]
+
@staticmethod
def escape(text: str) -> str:
return text
diff --git a/tests/models/minicpm_o_2_6/test.py b/tests/models/minicpm_o_2_6/test.py
deleted file mode 100644
index f4fb3f622156..000000000000
--- a/tests/models/minicpm_o_2_6/test.py
+++ /dev/null
@@ -1,47 +0,0 @@
-from transformers import AutoModelForCausalLM, AutoModel, AutoTokenizer
-import torch
-
-model_path = "xxx"
-
-model = AutoModel.from_pretrained(
- model_path,
- trust_remote_code=True,
- attn_implementation='sdpa',
- torch_dtype=torch.bfloat16,
- init_vision=True,
- init_audio=True,
- init_tts=True
-)
-
-
-tokenizer = AutoTokenizer.from_pretrained(
- model_path,
- trust_remote_code=True,
- local_files_only=True,
- )
-
-# device = torch.device("mps")
-# model.to(device)
-
-from PIL import Image
-
-image = Image.open('xxx.jpeg').convert('RGB')
-question = 'What is in the image?'
-msgs = [{'role': 'user', 'content': [image, question]}]
-res = model.chat(
- image=None,
- msgs=msgs,
- tokenizer=tokenizer
-)
-print(res)
-
-res = model.chat(
- msgs=msgs,
- tokenizer=tokenizer,
- sampling=True,
- stream=True
-)
-generated_text = ""
-for new_text in res:
- generated_text += new_text
- print(new_text, flush=True, end='')
\ No newline at end of file
diff --git a/tests/models/minicpm_o_2_6/test_modeling_minicpm_o_2_6.py b/tests/models/minicpm_o_2_6/test_modeling_minicpm_o_2_6.py
index a3a4e36da4d5..2fed53bd8d07 100644
--- a/tests/models/minicpm_o_2_6/test_modeling_minicpm_o_2_6.py
+++ b/tests/models/minicpm_o_2_6/test_modeling_minicpm_o_2_6.py
@@ -21,14 +21,14 @@
import numpy as np
import librosa
import soundfile as sf
-from moviepy.editor import VideoFileClip
+from moviepy import VideoFileClip
from PIL import Image
import requests
from io import BytesIO
from transformers import (
AutoModel,
- AutoTokenizer,
+ AutoProcessor,
is_torch_available,
)
from transformers.testing_utils import (
@@ -38,12 +38,12 @@
require_vision,
require_soundfile,
require_sentencepiece,
- require_tokenizers,
)
if is_torch_available():
import torch
+
@require_torch
class MiniCPM_o_2_6ModelIngestionTest(unittest.TestCase):
"""Test for MiniCPM_o_2_6Model."""
@@ -55,7 +55,7 @@ def setUp(self):
self.video_path = os.path.join(self.assets_dir, "Skiing.mp4")
self.ref_audio_path = os.path.join(self.assets_dir, "demo.wav")
-
+
if not os.path.exists(self.video_path):
video_url = "https://huggingface.co/openbmb/MiniCPM-o-2_6/resolve/main/assets/Skiing.mp4"
response = requests.get(video_url, stream=True)
@@ -72,18 +72,15 @@ def setUp(self):
for chunk in response.iter_content(chunk_size=8192):
f.write(chunk)
+ self.model_path = "openbmb/MiniCPM-o-2_6"
self.model = AutoModel.from_pretrained(
- "openbmb/MiniCPM-o-2_6",
- trust_remote_code=True,
+ self.model_path,
attn_implementation='sdpa',
- torch_dtype=torch.bfloat16,
- init_vision=True,
- init_audio=True,
- init_tts=True
+ torch_dtype=torch.bfloat16
)
self.model = self.model.eval().to(torch_device)
- self.tokenizer = AutoTokenizer.from_pretrained('openbmb/MiniCPM-o-2_6', trust_remote_code=True)
self.model.init_tts()
+ self.processor = AutoProcessor.from_pretrained(self.model_path)
def tearDown(self):
"""clean up test environment"""
@@ -94,16 +91,10 @@ def tearDown(self):
if os.path.exists(self.assets_dir):
os.rmdir(self.assets_dir)
- @slow
- def test_minicpm_o_2_6_model_base(self):
- """test base model loading"""
- base_model = AutoModel.from_pretrained("openbmb/MiniCPM-o-2_6")
- self.assertIsNotNone(base_model)
-
def _get_video_chunk_content(self, video_path, flatten=True):
"""process video content, extract frames and audio"""
video = VideoFileClip(video_path)
-
+
with tempfile.NamedTemporaryFile(suffix=".wav", delete=True) as temp_audio_file:
video.audio.write_audiofile(
temp_audio_file.name,
@@ -111,21 +102,22 @@ def _get_video_chunk_content(self, video_path, flatten=True):
fps=16000,
logger=None
)
- audio_np, sr = librosa.load(temp_audio_file.name, sr=16000, mono=True)
-
+ audio_np, sr = librosa.load(
+ temp_audio_file.name, sr=16000, mono=True)
+
num_units = math.ceil(video.duration)
contents = []
-
+
for i in range(num_units):
frame = video.get_frame(i+1)
image = Image.fromarray(frame.astype(np.uint8))
audio_segment = audio_np[sr*i:sr*(i+1)]
-
+
if flatten:
contents.extend(["", image, audio_segment])
else:
contents.append(["", image, audio_segment])
-
+
video.close()
return contents
@@ -133,21 +125,25 @@ def _get_video_chunk_content(self, video_path, flatten=True):
@require_vision
@require_soundfile
@require_sentencepiece
- @require_tokenizers
- def test_omni_chat(self):
+ def test_omni_generate(self):
try:
- ref_audio, _ = librosa.load(self.ref_audio_path, sr=16000, mono=True)
- sys_msg = self.model.get_sys_prompt(ref_audio=ref_audio, mode='omni', language='en')
-
+ ref_audio, _ = librosa.load(
+ self.ref_audio_path, sr=16000, mono=True)
+ sys_msg = self.processor.get_sys_prompt(
+ ref_audio=ref_audio, mode='omni', language='en')
+
contents = self._get_video_chunk_content(self.video_path)
msg = {"role": "user", "content": contents}
msgs = [sys_msg, msg]
+ inputs = self.processor.apply_chat_template(
+ msgs=msgs).to(self.model.device)
- output_audio_path = tempfile.NamedTemporaryFile(suffix=".wav", delete=False).name
+ output_audio_path = tempfile.NamedTemporaryFile(
+ suffix=".wav", delete=False).name
- res = self.model.chat(
- msgs=msgs,
- tokenizer=self.tokenizer,
+ res = self.model.generate(
+ **inputs,
+ processor=self.processor,
sampling=True,
temperature=0.5,
max_new_tokens=128,
@@ -159,16 +155,17 @@ def test_omni_chat(self):
use_image_id=False,
return_dict=True
)
+ res = self.processor.decode(res.outputs)
self.assertIsNotNone(res, "Chat response should not be empty")
- self.assertIn("text", res, "Chat response should contain text")
- self.assertTrue(len(res["text"]) > 0, "Chat response text should not be empty")
-
- if res.get("audio_path"):
- self.assertTrue(sf.info(output_audio_path).frames > 0, "Generated audio file should not be empty")
-
+ self.assertTrue(len(res) > 0,
+ "Chat response text should not be empty")
+ self.assertTrue(sf.info(output_audio_path).frames >
+ 0, "Generated audio file should not be empty")
+
except FileNotFoundError:
- self.skipTest(f"资源文件未找到: {self.video_path} 或 {self.ref_audio_path}")
+ self.skipTest(
+ f"资源文件未找到: {self.video_path} 或 {self.ref_audio_path}")
finally:
if 'output_audio_path' in locals() and os.path.exists(output_audio_path):
os.remove(output_audio_path)
@@ -177,44 +174,45 @@ def test_omni_chat(self):
@require_vision
@require_soundfile
@require_sentencepiece
- @require_tokenizers
def test_streaming_inference(self):
try:
self.model.reset_session()
-
- ref_audio, _ = librosa.load(self.ref_audio_path, sr=16000, mono=True)
- sys_msg = self.model.get_sys_prompt(ref_audio=ref_audio, mode='omni', language='en')
-
- contents = self._get_video_chunk_content(self.video_path, flatten=False)
+
+ ref_audio, _ = librosa.load(
+ self.ref_audio_path, sr=16000, mono=True)
+ sys_msg = self.processor.get_sys_prompt(
+ ref_audio=ref_audio, mode='omni', language='en')
+
+ contents = self._get_video_chunk_content(
+ self.video_path, flatten=False)
session_id = 'test_session'
generate_audio = True
- res = self.model.streaming_prefill(
+ self.model.streaming_prefill(
session_id=session_id,
msgs=[sys_msg],
- tokenizer=self.tokenizer
+ processor=self.processor,
)
- self.assertIsNotNone(res, "System prompt prefilling should not return empty")
for content in contents:
msgs = [{"role": "user", "content": content}]
- res = self.model.streaming_prefill(
+ self.model.streaming_prefill(
session_id=session_id,
msgs=msgs,
- tokenizer=self.tokenizer
+ processor=self.processor,
)
- self.assertIsNotNone(res, "Content prefilling should not return empty")
res = self.model.streaming_generate(
session_id=session_id,
- tokenizer=self.tokenizer,
+ processor=self.processor,
temperature=0.5,
generate_audio=generate_audio
)
audios = []
text = ""
- output_audio_path = tempfile.NamedTemporaryFile(suffix=".wav", delete=False).name
+ output_audio_path = tempfile.NamedTemporaryFile(
+ suffix=".wav", delete=False).name
if generate_audio:
for r in res:
@@ -226,17 +224,22 @@ def test_streaming_inference(self):
text += txt
res_audio = np.concatenate(audios)
- sf.write(output_audio_path, res_audio, samplerate=sampling_rate)
-
- self.assertTrue(len(text) > 0, "Generated text should not be empty")
- self.assertTrue(sf.info(output_audio_path).frames > 0, "Generated audio file should not be empty")
+ sf.write(output_audio_path, res_audio,
+ samplerate=sampling_rate)
+
+ self.assertTrue(
+ len(text) > 0, "Generated text should not be empty")
+ self.assertTrue(sf.info(output_audio_path).frames >
+ 0, "Generated audio file should not be empty")
else:
for r in res:
text += r['text']
- self.assertTrue(len(text) > 0, "Generated text should not be empty")
+ self.assertTrue(
+ len(text) > 0, "Generated text should not be empty")
except FileNotFoundError:
- self.skipTest(f"Resource file not found: {self.video_path} or {self.ref_audio_path}")
+ self.skipTest(
+ f"Resource file not found: {self.video_path} or {self.ref_audio_path}")
finally:
if 'output_audio_path' in locals() and os.path.exists(output_audio_path):
os.remove(output_audio_path)
@@ -244,14 +247,12 @@ def test_streaming_inference(self):
@slow
@require_soundfile
@require_sentencepiece
- @require_tokenizers
def test_audio_mimick(self):
try:
self.model.init_tts()
- self.model.tts.float()
mimick_prompt = "Please repeat each user's speech, including voice style and speech content."
-
+
audio_urls = [
'https://huggingface.co/openbmb/MiniCPM-o-2_6/resolve/main/assets/input_examples/Trump_WEF_2018_10s.mp3',
'https://huggingface.co/openbmb/MiniCPM-o-2_6/resolve/main/assets/input_examples/cxk_original.wav',
@@ -262,24 +263,29 @@ def test_audio_mimick(self):
for audio_url in audio_urls:
try:
-
+
response = requests.get(audio_url, stream=True)
response.raise_for_status()
-
+
with tempfile.NamedTemporaryFile(suffix=os.path.splitext(audio_url)[1], delete=False) as temp_file:
for chunk in response.iter_content(chunk_size=8192):
temp_file.write(chunk)
temp_file_path = temp_file.name
- audio_input, _ = librosa.load(temp_file_path, sr=16000, mono=True)
-
- msgs = [{'role': 'user', 'content': [mimick_prompt, audio_input]}]
-
- output_audio_path = tempfile.NamedTemporaryFile(suffix=".wav", delete=False).name
-
- res = self.model.chat(
- msgs=msgs,
- tokenizer=self.tokenizer,
+ audio_input, _ = librosa.load(
+ temp_file_path, sr=16000, mono=True)
+
+ msgs = [{'role': 'user', 'content': [
+ mimick_prompt, audio_input]}]
+ inputs = self.processor.apply_chat_template(
+ msgs=msgs).to(self.model.device)
+
+ output_audio_path = tempfile.NamedTemporaryFile(
+ suffix=".wav", delete=False).name
+
+ res = self.model.generate(
+ **inputs,
+ processor=self.processor,
sampling=True,
max_new_tokens=128,
use_tts_template=True,
@@ -287,10 +293,14 @@ def test_audio_mimick(self):
generate_audio=True,
output_audio_path=output_audio_path
)
+ res = self.processor.decode(res.outputs)
- self.assertIsNotNone(res, "Mimic response should not be empty")
- self.assertTrue(os.path.exists(output_audio_path), "Output audio file should exist")
- self.assertTrue(sf.info(output_audio_path).frames > 0, "Generated audio file should not be empty")
+ self.assertIsNotNone(
+ res, "Mimic response should not be empty")
+ self.assertTrue(os.path.exists(
+ output_audio_path), "Output audio file should exist")
+ self.assertTrue(sf.info(output_audio_path).frames >
+ 0, "Generated audio file should not be empty")
except requests.exceptions.RequestException as e:
self.skipTest(f"Failed to download audio file: {str(e)}")
@@ -308,44 +318,47 @@ def test_audio_mimick(self):
@slow
@require_vision
@require_sentencepiece
- @require_tokenizers
def test_single_image_inference(self):
try:
- image_url = "https://www.ilankelman.org/stopsigns/australia.jpg"
+ image_url = "https://bkimg.cdn.bcebos.com/pic/d043ad4bd11373f082022267f9585cfbfbedaa64aeb6"
response = requests.get(image_url, stream=True)
response.raise_for_status()
-
+
image = Image.open(BytesIO(response.content)).convert('RGB')
question = 'What is in the image?'
-
+
msgs = [{'role': 'user', 'content': [image, question]}]
-
- res = self.model.chat(
- image=None,
- msgs=msgs,
- tokenizer=self.tokenizer
+ inputs = self.processor.apply_chat_template(
+ msgs=msgs).to(self.model.device)
+
+ res = self.model.generate(
+ **inputs,
+ processor=self.processor,
)
-
- self.assertIsNotNone(res, "Normal inference response should not be empty")
- self.assertTrue(len(res) > 0, "Normal inference response text should not be empty")
-
- res = self.model.chat(
- msgs=msgs,
- tokenizer=self.tokenizer,
+ res = self.processor.decode(res)
+
+ self.assertIsNotNone(
+ res, "Normal inference response should not be empty")
+ self.assertTrue(
+ len(res) > 0, "Normal inference response text should not be empty")
+
+ res = self.model.generate(
+ **inputs,
+ processor=self.processor,
sampling=True,
stream=True
)
-
+
generated_text = ""
for new_text in res:
generated_text += new_text
- self.assertIsNotNone(new_text, "Each part of streaming reasoning should not be empty")
-
- self.assertTrue(len(generated_text) > 0, "Text should not be empty")
-
+ self.assertIsNotNone(
+ new_text, "Each part of streaming reasoning should not be empty")
+
+ self.assertTrue(len(generated_text) > 0,
+ "Text should not be empty")
+
except requests.exceptions.RequestException as e:
self.skipTest(f"Failed to download image: {str(e)}")
except Exception as e:
self.fail(f"Single image inference test failed: {str(e)}")
-
-
\ No newline at end of file
diff --git a/tests/models/minicpm_o_2_6/test_processing_minicpm_o_2_6.py b/tests/models/minicpm_o_2_6/test_processing_minicpm_o_2_6.py
index 36cd6ca33aa9..624e660f30ab 100644
--- a/tests/models/minicpm_o_2_6/test_processing_minicpm_o_2_6.py
+++ b/tests/models/minicpm_o_2_6/test_processing_minicpm_o_2_6.py
@@ -13,23 +13,23 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import inspect
import shutil
import tempfile
import unittest
import numpy as np
-import pytest
+from PIL import Image
import librosa
import requests
+import pytest
+import soundfile as sf
+import io
from transformers import (
AutoProcessor,
MiniCPM_o_2_6Processor,
- AutoTokenizer,
- WhisperFeatureExtractor,
)
-from transformers.testing_utils import require_av, require_librosa, require_torch, require_torchaudio, require_vision
+from transformers.testing_utils import require_librosa, require_torch, require_torchaudio, require_vision
from transformers.utils import is_torch_available, is_vision_available
from ...test_processing_common import ProcessorTesterMixin
@@ -45,14 +45,29 @@
@require_vision
@require_torch
@require_torchaudio
-class MiniCPM_o_2_6ProcessorTest(ProcessorTesterMixin, unittest.TestCase):
+class MiniCPM_o_2_6ProcessorTest(unittest.TestCase):
processor_class = MiniCPM_o_2_6Processor
- def setUp(self):
- self.tmpdirname = tempfile.mkdtemp()
- processor_kwargs = self.prepare_processor_dict()
- processor = MiniCPM_o_2_6Processor.from_pretrained("openbmb/MiniCPM-O-2.6", **processor_kwargs)
- processor.save_pretrained(self.tmpdirname)
+ # 添加这个属性，告知测试框架处理器有自己的chat template实现
+ has_chat_template = True
+
+ # 输入名称定义
+ images_input_name = "pixel_values"
+ audio_input_name = "audio_features"
+
+ @classmethod
+ def setUpClass(cls):
+ cls.tmpdirname = tempfile.mkdtemp()
+ processor_kwargs = cls.prepare_processor_dict()
+ processor = AutoProcessor.from_pretrained(
+ "openbmb/MiniCPM-o-2_6", **processor_kwargs)
+ processor.save_pretrained(cls.tmpdirname)
+
+ @staticmethod
+ def prepare_processor_dict():
+ return {
+ "chat_template": "{% for message in messages %}{% if message['role'] == 'user' %}<|im_start|>user\n{{ message['content'] }}<|im_end|>\n{% elif message['role'] == 'assistant' %}<|im_start|>assistant\n{{ message['content'] }}<|im_end|>\n{% endif %}{% endfor %}"
+ }
def get_tokenizer(self, **kwargs):
return AutoProcessor.from_pretrained(self.tmpdirname, **kwargs).tokenizer
@@ -63,342 +78,225 @@ def get_image_processor(self, **kwargs):
def get_feature_extractor(self, **kwargs):
return AutoProcessor.from_pretrained(self.tmpdirname, **kwargs).feature_extractor
- @staticmethod
- def prepare_processor_dict():
- return {
- "chat_template": "{% for message in messages %}{% if message['role'] == 'user' %}<|im_start|>user\n{{ message['content'] }}<|im_end|>\n{% elif message['role'] == 'assistant' %}<|im_start|>assistant\n{{ message['content'] }}<|im_end|>\n{% endif %}{% endfor %}"
- }
+ def get_processor(self, **kwargs):
+ return MiniCPM_o_2_6Processor.from_pretrained(self.tmpdirname, **kwargs)
+
+ @classmethod
+ def tearDownClass(cls):
+ shutil.rmtree(cls.tmpdirname, ignore_errors=True)
def tearDown(self):
- shutil.rmtree(self.tmpdirname, ignore_errors=True)
+ # 防止复写父类的tearDown方法
+ pass
- def prepare_image_inputs(self):
+ def prepare_image_inputs(self, batch_size=1):
"""Preparing the image input for testing"""
- image_url = "https://www.ilankelman.org/stopsigns/australia.jpg"
+ image_url = "https://bkimg.cdn.bcebos.com/pic/d043ad4bd11373f082022267f9585cfbfbedaa64aeb6"
image = Image.open(requests.get(image_url, stream=True).raw)
- return image
+ if batch_size == 1:
+ return image
+ return [[image]] * batch_size
- def prepare_audio_inputs(self):
+ def prepare_audio_inputs(self, batch_size=1):
"""Preparing the audio input for testing"""
audio_url = "https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen2-Audio/audio/f2641_0_throatclearing.wav"
- try:
- audio, _ = librosa.load(audio_url, sr=16000)
+ response = requests.get(audio_url)
+ audio, _ = librosa.load(io.BytesIO(
+ response.content), sr=16000, mono=True)
+ if batch_size == 1:
return audio
- except:
- # 如果下载失败，使用随机生成的音频
- return np.random.rand(160000) * 2 - 1 # 模拟1秒的音频
-
- def test_save_load_pretrained_default(self):
- """Test saving and loading pretrained models"""
- image_processor = self.get_image_processor()
- tokenizer = self.get_tokenizer()
- feature_extractor = self.get_feature_extractor()
- processor = self.processor_class(
- tokenizer=tokenizer,
- feature_extractor=feature_extractor,
- image_processor=image_processor,
- )
-
- processor.save_pretrained(self.tmpdirname)
- processor = MiniCPM_o_2_6Processor.from_pretrained(self.tmpdirname)
-
- self.assertEqual(processor.tokenizer.get_vocab(), tokenizer.get_vocab())
- self.assertEqual(processor.image_processor.to_json_string(), image_processor.to_json_string())
- self.assertEqual(processor.feature_extractor.to_json_string(), feature_extractor.to_json_string())
-
- def test_image_processor(self):
- """Testing image processing functions"""
- image_processor = self.get_image_processor()
- tokenizer = self.get_tokenizer()
- feature_extractor = self.get_feature_extractor()
- processor = self.processor_class(
- tokenizer=tokenizer,
- feature_extractor=feature_extractor,
- image_processor=image_processor,
- )
+ return [[audio]] * batch_size
+
+ def prepare_text_inputs(self, batch_size=1, modality="text"):
+ """准备文本输入"""
+ if modality == "text":
+ text = "Hello, how are you?"
+ elif modality == "image":
+ text = "What is in this image: (./)?"
+ elif modality == "audio":
+ text = "What was said in this audio: ()?"
+ elif modality == "mixed":
+ text = "Describe this image (./) and this audio ()."
+
+ if batch_size == 1:
+ return text
+ return [text] * batch_size
+
+ def test_processor_with_multi_modal_inputs(self):
+ """Test the processor with multiple modality inputs"""
+ processor = self.get_processor()
- image_input = self.prepare_image_inputs()
- input_image_proc = image_processor(image_input, return_tensors="np")
- input_processor = processor(images=image_input, text="dummy", return_tensors="np")
-
- for key in input_image_proc.keys():
- self.assertAlmostEqual(input_image_proc[key].sum(), input_processor[key].sum(), delta=1e-2)
-
- def test_processor(self):
- """Testing overall processing functions"""
- image_processor = self.get_image_processor()
- tokenizer = self.get_tokenizer()
- feature_extractor = self.get_feature_extractor()
- processor = self.processor_class(
- tokenizer=tokenizer,
- feature_extractor=feature_extractor,
- image_processor=image_processor,
- )
+ input_text = self.prepare_text_inputs(modality="mixed")
+ image = self.prepare_image_inputs()
+ audio = self.prepare_audio_inputs()
- input_str = "Test text"
- image_input = self.prepare_image_inputs()
- audio_input = self.prepare_audio_inputs()
- inputs = processor(text=input_str, images=image_input, audios=audio_input)
-
- keys = list(inputs.keys())
- self.assertListEqual(
- keys,
- [
- "input_ids",
- "attention_mask",
- "pixel_values",
- "image_sizes",
- "image_bound",
- "tgt_sizes",
- "audio_features",
- "audio_feature_lens",
- "audio_bounds",
- "spk_bounds",
- ],
- )
+ # Test with all modalities
+ inputs = processor(text=[input_text], images=[image],
+ audios=[audio], return_tensors="pt")
- # no input
- with pytest.raises(ValueError):
- processor()
-
- # no text input
- with pytest.raises(ValueError):
- processor(images=image_input)
-
- def test_model_input_names(self):
- """Testing model input names"""
- image_processor = self.get_image_processor()
- tokenizer = self.get_tokenizer()
- feature_extractor = self.get_feature_extractor()
- processor = self.processor_class(
- tokenizer=tokenizer,
- feature_extractor=feature_extractor,
- image_processor=image_processor,
- )
+ # 检查输入中包含所有模态相关的键
+ self.assertIn("input_ids", inputs)
+ self.assertIn("attention_mask", inputs)
+ self.assertIn("pixel_values", inputs)
+ self.assertIn("image_bound", inputs)
+ self.assertIn("audio_features", inputs)
+ self.assertIn("audio_bounds", inputs)
- input_str = "Test text"
- image_input = self.prepare_image_inputs()
- audio_input = self.prepare_audio_inputs()
+ # 检查处理器能处理批量输入
+ batch_size = 2
+ input_texts = self.prepare_text_inputs(
+ batch_size=batch_size, modality="mixed")
+ images = self.prepare_image_inputs(batch_size=batch_size)
+ audios = self.prepare_audio_inputs(batch_size=batch_size)
+
+ batch_inputs = processor(
+ text=input_texts, images=images, audios=audios, return_tensors="pt")
+ self.assertEqual(batch_inputs["input_ids"].shape[0], batch_size)
+ self.assertEqual(batch_inputs["attention_mask"].shape[0], batch_size)
+
+ @require_torch
+ def test_apply_chat_template_text(self):
+ """Test applying chat template with text only"""
+ processor = self.get_processor()
- inputs = processor(text=input_str, images=image_input, audios=audio_input)
- self.assertListEqual(sorted(inputs.keys()), sorted(processor.model_input_names))
+ # 准备聊天消息
+ messages = [
+ {"role": "system", "content": "You are a helpful assistant."},
+ {"role": "user", "content": "Hello, how are you?"},
+ {"role": "assistant", "content": "I'm doing well, thank you!"}
+ ]
- def test_apply_chat_template(self):
- """Testing chat template application"""
+ # 测试不带标记化的应用
+ formatted_prompt = processor.tokenizer.apply_chat_template(
+ messages, tokenize=False)
+ self.assertIsInstance(formatted_prompt, str)
+ self.assertTrue("<|im_start|>" in formatted_prompt)
+ self.assertTrue("<|im_end|>" in formatted_prompt)
+
+ # 测试带标记化的应用
+ tokenized_prompt = processor.tokenizer.apply_chat_template(
+ messages, tokenize=True, return_tensors="pt")
+ self.assertIsInstance(tokenized_prompt, torch.Tensor)
+
+ @require_torch
+ @require_vision
+ def test_apply_chat_template_with_image(self):
+ """Test applying chat template with image"""
processor = self.get_processor()
- if processor.chat_template is None:
- self.skipTest("Processor does not have a chat template")
+ image = self.prepare_image_inputs()
+ # 准备包含图像的消息
messages = [
- [
- {
- "role": "user",
- "content": "Please describe this image.",
- },
- {
- "role": "assistant",
- "content": "This is an image.",
- },
- ]
+ {"role": "user", "content": [image, "What is in this image?"]}
]
- formatted_prompt = processor.apply_chat_template(messages, add_generation_prompt=True, tokenize=False)
- self.assertEqual(len(formatted_prompt), 1)
+ # 测试聊天模板应用
+ inputs = processor.apply_chat_template(
+ msgs=messages, omni_input=True, return_tensors="pt")
- formatted_prompt_tokenized = processor.apply_chat_template(
- messages, add_generation_prompt=True, tokenize=True
- )
- expected_output = processor.tokenizer(formatted_prompt, return_tensors=None).input_ids
- self.assertListEqual(expected_output, formatted_prompt_tokenized)
+ # 检查输出中有图像相关的数据
+ self.assertIn("input_ids", inputs)
+ self.assertIn("attention_mask", inputs)
+ self.assertIn("pixel_values", inputs)
+ self.assertIn("image_bound", inputs)
- def test_audio_processing(self):
- """Testing audio processing functions"""
+ @require_torch
+ @require_librosa
+ def test_apply_chat_template_with_audio(self):
+ """Test applying chat template with audio"""
processor = self.get_processor()
- audio_input = self.prepare_audio_inputs()
-
- # test audio feature extraction
- audio_features = processor.feature_extractor(
- audio_input,
- sampling_rate=16000,
- return_attention_mask=True,
- padding="max_length",
- return_tensors="pt",
- )
-
- self.assertIn("input_features", audio_features)
- self.assertIn("attention_mask", audio_features)
-
- # test audio and text combination processing
- inputs = processor(
- text="Describe this audio",
- audios=audio_input,
- return_tensors="pt",
+ audio = self.prepare_audio_inputs()
+
+ # 准备包含音频的消息
+ messages = [
+ {"role": "user", "content": [
+ audio, "What was said in this audio?"]}
+ ]
+
+ # 测试聊天模板应用
+ inputs = processor.apply_chat_template(
+ msgs=messages, omni_input=True, use_tts_template=True, return_tensors="pt"
)
-
+
+ # 检查输出中有音频相关的数据
+ self.assertIn("input_ids", inputs)
+ self.assertIn("attention_mask", inputs)
self.assertIn("audio_features", inputs)
- self.assertIn("audio_feature_lens", inputs)
self.assertIn("audio_bounds", inputs)
- def test_number_to_text_converter(self):
- """Testing number to text conversion functions"""
- processor = self.get_processor()
- converter = processor.NumberToTextConverter()
-
- # test chinese number conversion
- chinese_text = converter.replace_numbers_with_text("我有2个苹果", language="chinese")
- self.assertEqual(chinese_text, "我有两个苹果")
-
- # test english number conversion
- english_text = converter.replace_numbers_with_text("I have 23 books", language="english")
- self.assertEqual(english_text, "I have two three books")
-
- # test automatic language detection
- mixed_text = converter.replace_numbers_with_text("我有3个苹果和4个梨")
- self.assertIn("三", mixed_text)
- self.assertIn("四", mixed_text)
-
- def test_voice_checker(self):
- """Testing voice checking functions"""
+ @require_torch
+ @require_vision
+ @require_librosa
+ def test_apply_chat_template_with_mixed_media(self):
+ """Test applying chat template with both image and audio"""
processor = self.get_processor()
- checker = processor.VoiceChecker()
-
- # test silent detection
- silent_audio = np.zeros(16000) # 1 second of silence
- mel_spec = np.random.rand(100, 100) # simulate mel spectrogram
- is_bad = checker.is_bad(silent_audio, mel_spec)
- self.assertTrue(is_bad)
-
- # test normal audio
- normal_audio = np.random.rand(16000) * 2 - 1 # 1 second of normal audio
- is_bad = checker.is_bad(normal_audio, mel_spec)
- self.assertFalse(is_bad)
-
- # test reset function
- checker.reset()
- self.assertIsNone(checker.previous_mel)
- self.assertEqual(checker.consecutive_zeros, 0)
- self.assertEqual(checker.consecutive_low_distance, 0)
-
- def test_image_processor_basic(self):
- """Testing basic image processor functions"""
- image_processor = self.get_image_processor()
-
- # test image preprocessing
image = self.prepare_image_inputs()
- processed = image_processor.preprocess(image)
-
- self.assertIn("pixel_values", processed)
- self.assertIn("image_sizes", processed)
- self.assertIn("tgt_sizes", processed)
-
- # test image slicing function
- sliced_images = image_processor.get_sliced_images(image)
- self.assertIsInstance(sliced_images, list)
- self.assertTrue(len(sliced_images) > 0)
-
- # test image placeholder generation
- placeholder = image_processor.get_slice_image_placeholder(image.size)
- self.assertIsInstance(placeholder, str)
- self.assertTrue(len(placeholder) > 0)
-
- def test_image_processor_edge_cases(self):
- """Testing image processor edge cases"""
- image_processor = self.get_image_processor()
-
- # test empty image list
- with self.assertRaises(ValueError):
- image_processor.preprocess([])
-
- # test invalid image
- invalid_image = np.zeros((100, 100))
- with self.assertRaises(ValueError):
- image_processor.preprocess(invalid_image)
-
- # test large image
- large_image = Image.new('RGB', (2000, 2000))
- processed = image_processor.preprocess(large_image)
- self.assertIn("pixel_values", processed)
-
- def test_chat_tts_processor(self):
- """Testing ChatTTS processor functions"""
- processor = self.get_processor()
- tts_processor = processor.ChatTTSProcessor(processor.tokenizer)
-
- # prepare test data
- text_list = ["Test text 1", "Test text 2"]
- audio_list = [np.random.rand(16000) for _ in range(2)]
-
- # test processing function
- outputs = tts_processor(text_list, audio_list)
-
- self.assertIn("tts_input_ids_varlen", outputs)
- self.assertIn("tts_input_features_varlen", outputs)
- self.assertEqual(len(outputs["tts_input_ids_varlen"]), len(text_list))
- self.assertEqual(len(outputs["tts_input_features_varlen"]), len(audio_list))
-
- def test_mel_spectrogram_features(self):
- """Testing mel spectrogram feature extraction functions"""
- processor = self.get_processor()
- mel_processor = processor.MelSpectrogramFeatures()
-
- # prepare test audio
- audio = torch.randn(1, 16000) # 1 second of audio
-
- # test feature extraction
- features = mel_processor(audio)
-
- self.assertIsInstance(features, torch.Tensor)
- self.assertEqual(features.dim(), 2) # should be 2D tensor
- self.assertEqual(features.shape[0], 100) # mel filter number
-
- def test_batch_processing(self):
- """Testing batch processing functions"""
- processor = self.get_processor()
-
- # prepare batch processing data
- texts = ["Text 1", "Text 2", "Text 3"]
- images = [self.prepare_image_inputs() for _ in range(3)]
- audios = [np.random.rand(16000) for _ in range(3)]
-
- # test batch processing
- batch_outputs = processor(
- text=texts,
- images=images,
- audios=audios,
- return_tensors="pt"
+ audio = self.prepare_audio_inputs()
+
+ # 准备包含多种媒体的消息
+ messages = [
+ {"role": "user", "content": [
+ image, audio, "Describe this image and audio."]}
+ ]
+
+ # 测试聊天模板应用
+ inputs = processor.apply_chat_template(
+ msgs=messages, omni_input=True, use_tts_template=True, return_tensors="pt"
)
-
- # verify outputs
- self.assertIn("input_ids", batch_outputs)
- self.assertIn("attention_mask", batch_outputs)
- self.assertIn("pixel_values", batch_outputs)
- self.assertIn("audio_features", batch_outputs)
-
- # verify batch size
- self.assertEqual(batch_outputs["input_ids"].shape[0], len(texts))
- self.assertEqual(len(batch_outputs["pixel_values"]), len(images))
- self.assertEqual(len(batch_outputs["audio_features"]), len(audios))
-
- def test_error_handling(self):
- """Testing error handling functions"""
+
+ # 检查输出中有所有模态相关的数据
+ self.assertIn("input_ids", inputs)
+ self.assertIn("attention_mask", inputs)
+ self.assertIn("pixel_values", inputs)
+ self.assertIn("image_bound", inputs)
+ self.assertIn("audio_features", inputs)
+ self.assertIn("audio_bounds", inputs)
+
+ def test_get_sys_prompt(self):
+ """Test the system prompt generation function"""
+ processor = self.get_processor()
+ audio = self.prepare_audio_inputs()
+
+ # 测试默认系统提示
+ sys_prompt = processor.get_sys_prompt()
+ self.assertIn("role", sys_prompt)
+ self.assertIn("content", sys_prompt)
+
+ # 测试 omni 模式
+ sys_prompt = processor.get_sys_prompt(mode="omni", language="zh")
+ self.assertIn("你是一个AI助手", sys_prompt["content"][0])
+
+ # 测试英文提示
+ sys_prompt = processor.get_sys_prompt(mode="omni", language="en")
+ self.assertIn("You are a helpful assistant", sys_prompt["content"][0])
+
+ # 测试语音克隆提示
+ sys_prompt = processor.get_sys_prompt(
+ ref_audio=audio, mode="voice_cloning", language="en")
+ self.assertIn("Clone the voice", sys_prompt["content"][0])
+
+ # 测试语音助手模式
+ sys_prompt = processor.get_sys_prompt(
+ ref_audio=audio, mode="audio_assistant", language="zh")
+ self.assertIn("模仿输入音频中的声音特征", sys_prompt["content"][0])
+
+ def test_decode(self):
+ """Test the decoding functionality"""
processor = self.get_processor()
-
- # test invalid input
- with self.assertRaises(ValueError):
- processor(text=None, images=None, audios=None)
-
- # test mismatch input length
- with self.assertRaises(AssertionError):
- processor(
- text=["Text 1", "Text 2"],
- images=[self.prepare_image_inputs()],
- audios=[np.random.rand(16000)]
- )
-
- # test invalid audio sampling rate
- with self.assertRaises(ValueError):
- processor(
- text="Test text",
- audios=np.random.rand(16000),
- sampling_rate=0
- )
+
+ # 创建模拟输出
+ class MockOutput:
+ def __init__(self, sequences):
+ self.sequences = sequences
+
+ # 测试单个输出
+ mock_output = MockOutput(torch.tensor([[1, 2, 3, 4, 0, 0]]))
+ result = processor.decode(mock_output)
+ self.assertIsInstance(result, str)
+
+ # 测试批量输出
+ mock_output = MockOutput(torch.tensor(
+ [[1, 2, 3, 4, 0, 0], [5, 6, 7, 8, 0, 0]]))
+ result = processor.decode(mock_output, batched=True)
+ self.assertIsInstance(result, list)
+ self.assertEqual(len(result), 2)