Procesadores multimodales

Un procesador combina un tokenizador con uno o varios procesadores de modalidades, como un procesador de imágenes, un procesador de video o un extractor de características. Expone un único método __call__ que enruta cada entrada al componente adecuado y fusiona las salidas en un único diccionario.

Algunos modelos multimodales intercalan texto con imágenes, vídeos o audio. Para estos modelos, ProcessorMixin puede reemplazar los tokens de marcador de posición como <image>, <video> y <audio> con el patrón de tokens esperado por el modelo.

Agregar un nuevo procesador

Defina una clase de procesador creando src/transformers/models/<model>/processing_<my_model_name>.py y subclase ProcessorMixin. Asegúrese de definir un objeto TypedDict con valores predeterminados y asignarlo como cls.valid_processor_kwargs

from ...processing_utils import ProcessorMixin, ProcessingKwargs, Unpack

class MyModelProcessorKwargs(ProcessingKwargs, total=False):
    images_kwargs: MyModelImageProcessorKwargs
    _defaults = {
        "text_kwargs": {"padding": True},
        "images_kwargs": {"do_convert_rgb": True},
    }

class MyModelProcessor(ProcessorMixin):
    valid_processor_kwargs = MyModelProcessorKwargs

    def __init__(self, image_processor, tokenizer, chat_template=None, **kwargs):
        self.image_token = tokenizer.image_token
        self.image_token_id = tokenizer.image_token_id
        super().__init__(
            image_processor=image_processor,
            tokenizer=tokenizer,
            chat_template=chat_template,
            **kwargs,
        )

Implementa replace_<modality>_token si es necesario. Recibe el diccionario de salida completo del subprocesador y el índice de la entrada actual, y devuelve la cadena de reemplazo expandida para esa entrada. La cadena de reemplazo es lo que el modelo espera en la secuencia de entrada.

Si el modelo no utiliza repetición de marcadores en absoluto (no se define image_token), no es necesario sobrescribir este método. Deja self.image_token sin establecer y la clase base omitirá la sustitución por completo.

def replace_image_token(self, image_inputs: dict, image_idx: int) -> str:
    num_crops = image_inputs["num_crops"][image_idx]
    return f"{self.boi_token}{self.image_token * self.num_image_tokens * num_crops}{self.eoi_token}"

Opcionalmente, sobrescriba los métodos prepare_inputs_layout y validate_inputs. Si el modelo requiere una estructura de entrada específica antes de que comience el procesamiento, como reordenar las imágenes en una lista anidada, o una validación específica del modelo además de las comprobaciones comunes.

def prepare_inputs_layout(self, images=None, text=None, videos=None, audio=None, **kwargs):
    # Call `super()` to apply common preparation steps first 
    images, text, videos, audio = super().prepare_inputs_layout(images, text, videos, audio)
    if images is not None:
        images = make_nested_list_of_images(images)
    return images, text, videos, audio

def validate_inputs(self, images=None, text=None, videos=None, audio=None, **kwargs):
    super().validate_inputs(images=images, text=text, **kwargs)
    if text is not None and images is not None:
        n_tokens = [s.count(self.image_token) for s in text]
        n_images = [len(img_list) for img_list in images]
        if n_tokens != n_images:
            raise ValueError(
                f"Number of {self.image_token} tokens in text {n_tokens} does not match "
                f"number of images {n_images}."
            )

Consulte Gemma4Processor y Qwen2VLProcessor como referencia.

Pruebas

Todos los procesadores multimodales deben tener una clase de prueba que herede de ProcessorTesterMixin. Este mixin proporciona una suite estándar que cubre la tokenización, el procesamiento de imágenes, el batching y la codificación de ida y vuelta.

# tests/models/my_model_name/test_processor_<my_model_name>.py

from transformers.testing_utils import require_vision
from transformers.utils import is_vision_available
from ...test_processing_common import ProcessorTesterMixin

if is_vision_available():
    from transformers import MyModelProcessor

@require_vision
class MyModelProcessorTest(ProcessorTesterMixin, unittest.TestCase):
    processor_class = MyModelProcessor

    def get_processor(self):
        return MyModelProcessor.from_pretrained("hf-internal-testing/my-model-test")

Update on GitHub

Transformers

Procesadores multimodales

Agregar un nuevo procesador

Pruebas