vllm.multimodal ¶

class MultiModalHasher:
    """Derives multi-modal cache keys.

    Every method here yields *framed* chunks: each chunk is preceded by its
    length and each container by its kind, so that the concatenation fed to the
    digest is uniquely decodable and distinct inputs cannot share a key.
    """

    @classmethod
    def serialize_item(cls, obj: object) -> Iterable[bytes | memoryview]:
        # Simple cases
        if isinstance(obj, (bytes, memoryview)):
            return _framed(obj)
        if isinstance(obj, str):
            return _framed(obj.encode("utf-8"))
        if isinstance(obj, (int, float)):
            return _framed(np.array(obj).tobytes())

        if isinstance(obj, Image.Image):
            image_id = _get_image_id_bytes(obj)
            if image_id is not None:
                return _framed(image_id)

            data = {"mode": obj.mode, "data": np.asarray(obj)}
            palette = obj.palette
            if palette is not None:
                data["palette"] = palette.palette
                if palette.rawmode is not None:
                    data["palette_rawmode"] = palette.rawmode

            return cls.iter_item_to_bytes("image", data)

        if isinstance(obj, MediaWithBytes) and isinstance(obj.media, Image.Image):
            image_id = _get_image_id_bytes(obj.media)
            if image_id is not None:
                return _framed(image_id)

            if obj.io_config:
                return cls.iter_item_to_bytes(
                    "image",
                    {"io_config": obj.io_config, "data": obj.original_bytes},
                )
            return cls.iter_item_to_bytes("image", obj.original_bytes)

        if isinstance(obj, MediaWithBytes) and isinstance(
            obj.media, (np.ndarray, torch.Tensor)
        ):
            frames = obj.media
            # Both np.ndarray and torch.Tensor expose .nbytes.
            if frames.nbytes < len(obj.original_bytes):
                return cls.iter_item_to_bytes("video", frames)
            return cls.iter_item_to_bytes("video", obj.original_bytes)

        if isinstance(obj, torch.Tensor):
            tensor_obj: torch.Tensor = obj.cpu()
            tensor_dtype = tensor_obj.dtype
            tensor_shape = tensor_obj.shape

            # NumPy does not support bfloat16.
            # Workaround: View the tensor as a contiguous 1D array of bytes
            if tensor_dtype == torch.bfloat16:
                tensor_obj = tensor_obj.contiguous()
                tensor_obj = tensor_obj.view((tensor_obj.numel(),)).view(torch.uint8)

                return cls.iter_item_to_bytes(
                    "tensor",
                    {
                        "original_dtype": str(tensor_dtype),
                        "original_shape": tuple(tensor_shape),
                        "data": tensor_obj.numpy(),
                    },
                )

            return cls.iter_item_to_bytes("tensor", tensor_obj.numpy())

        if isinstance(obj, np.ndarray):
            if obj.ndim == 0:
                arr_data = obj.item()
            elif obj.flags.c_contiguous:
                # Not valid for 0-D arrays
                arr_data = obj.view(np.uint8).data
            else:
                # If the array is non-contiguous, we need to copy it first
                arr_data = obj.tobytes()

            return cls.iter_item_to_bytes(
                "ndarray",
                {
                    "dtype": obj.dtype.str,
                    "shape": obj.shape,
                    "data": arr_data,
                },
            )

        logger.warning(
            "No serialization method found for %s. Falling back to pickle.", type(obj)
        )

        return _framed(pickle.dumps(obj))

    @classmethod
    def iter_item_to_bytes(
        cls,
        key: str,
        obj: object,
    ) -> Iterable[bytes | memoryview]:
        """Yield the digest input for a single ``key``/``obj`` pair."""
        yield from _framed(key.encode("utf-8"))
        yield from cls.iter_value_to_bytes(obj)

    @classmethod
    def iter_value_to_bytes(
        cls,
        obj: object,
    ) -> Iterable[bytes | memoryview]:
        """Yield the digest input for a value, tagged by its container kind.

        Containers carry their kind and length so that a nested structure can
        never serialize to the same bytes as a differently shaped one (a list
        and a mapping keyed by stringified indices, for example).
        """
        if obj is None:
            yield _TAG_NONE
        elif isinstance(obj, (list, tuple)):
            yield _TAG_SEQUENCE
            yield _encode_length(len(obj))
            for elem in obj:
                yield from cls.iter_value_to_bytes(elem)
        elif isinstance(obj, dict):
            yield _TAG_MAPPING
            yield _encode_length(len(obj))
            for k, v in obj.items():
                yield from _framed(str(k).encode("utf-8"))
                yield from cls.iter_value_to_bytes(v)
        else:
            yield _TAG_LEAF
            yield from cls.serialize_item(obj)

    @classmethod
    def hash_kwargs(
        cls,
        algorithm: MMHasherAlgorithm,
        /,
        **kwargs: object,
    ) -> str:
        hasher_factory = _get_hasher_factory(algorithm)
        hasher = hasher_factory()

        for k, v in sorted(kwargs.items(), key=lambda kv: kv[0]):
            for bytes_ in cls.iter_item_to_bytes(k, v):
                hasher.update(bytes_)

        return hasher.hexdigest()

class MultiModalKwargsItems(UserDict[str, Sequence[_I]]):
    """A dictionary of processed multi-modal inputs by modality.

    For example, given a processor that processes
    images into `pixel_values` and `image_grid_thw`,
    and audios into `input_audio_features`,
    a prompt with 2 images and 1 audio will be processed
    into a `MultiModalKwargsItems` with the following structure:

    ```python
    MultiModalKwargsItems(
        {
            "image": [
                # For the first image
                MultiModalKwargsItem({"pixel_values": ..., "image_grid_thw": ...}),
                # For the second imgae
                MultiModalKwargsItem({"pixel_values": ..., "image_grid_thw": ...}),
            ],
            "audio": [
                # For the first audio
                MultiModalKwargsItem({"input_audio_features": ...}),
            ],
        }
    )
    ```

    Unlike HF processing which returns all items
    in a single dictionary with batched keyword arguments,
    we split up the items because some of them may already be cached.
    Also, items from multiple requests may be batched together to improve throughput,
    using the logic defined by the
    [`BaseMultiModalField`][vllm.multimodal.inputs.BaseMultiModalField]
    for each keyword argument.
    """

    @staticmethod
    def from_hf_inputs(
        hf_inputs: "BatchFeature",
        config_by_key: Mapping[str, MultiModalFieldConfig],
    ):
        # NOTE: This skips fields in `hf_inputs` that are not in `config_by_key`
        # We assume that those fields are not used in vLLM
        elems_by_key = dict[str, Sequence[MultiModalFieldElem]]()
        keys_by_modality = defaultdict[str, set[str]](set)
        for key, config in config_by_key.items():
            batch = hf_inputs.get(key)
            if batch is not None:
                elems = config.build_elems(key, batch)
                if len(elems) > 0:
                    elems_by_key[key] = elems
                    keys_by_modality[config.modality].add(key)

        items_by_modality = dict[str, list[MultiModalKwargsItem]]()
        for modality, keys in keys_by_modality.items():
            elems_in_modality = {k: elems_by_key[k] for k in keys}
            batch_sizes = {k: len(v) for k, v in elems_in_modality.items()}

            if len(set(batch_sizes.values())) > 1:
                raise ValueError(
                    f"Cannot merge different batch sizes for {modality=}! "
                    f"Found: {batch_sizes=}"
                )

            batch_size = next(iter(batch_sizes.values()))
            items_by_modality[modality] = [
                MultiModalKwargsItem({k: v[i] for k, v in elems_in_modality.items()})
                for i in range(batch_size)
            ]

        return MultiModalKwargsItems(items_by_modality)

    def __getitem__(self, modality: str) -> Sequence[_I]:
        if modality not in self:
            raise KeyError(
                f"Modality {modality!r} not found. "
                f"Available modalities: {set(self.keys())}"
            )

        return super().__getitem__(modality)  # type: ignore[return-value]

    def require_data(self) -> "MultiModalKwargsItems[MultiModalKwargsItem]":
        for modality, items in self.items():
            for i, item in enumerate(items):
                if item is None:
                    raise RuntimeError(f"Found empty mm_items[{modality}][{i}]")

        return self  # type: ignore[return-value]

    def get_data(
        self,
        *,
        device: torch.types.Device = None,
        pin_memory: bool = False,
    ) -> BatchedTensorInputs:
        """Construct a dictionary of keyword arguments to pass to the model."""
        from .utils import group_and_batch_mm_items

        items_by_modality = self.require_data()
        batches_by_modality = {
            modality: [
                data
                for _, data in group_and_batch_mm_items(
                    items,
                    device=device,
                    pin_memory=pin_memory,
                )
            ]
            for modality, items in items_by_modality.items()
            if len(items) > 0
        }

        out_data: BatchedTensorInputs = {}
        for _, batches in batches_by_modality.items():
            if len(batches) != 1:
                num_batches_by_modality = {
                    modality: len(batches)
                    for modality, batches in batches_by_modality.items()
                }

                raise RuntimeError(
                    f"Some modalities cannot be merged into a single batch "
                    f"({num_batches_by_modality=})"
                )

            out_data.update(batches[0])

        return out_data

class MultiModalRegistry:
    """A registry that dispatches data processing according to the model."""

    def register_processor(
        self,
        processor: MultiModalProcessorFactory[_I],
        *,
        info: ProcessingInfoFactory[_I],
        dummy_inputs: DummyInputsBuilderFactory[_I],
    ):
        """Register a multi-modal processor to a model class. The processor
        is constructed lazily, hence a factory method should be passed.

        When the model receives multi-modal data, the provided function is
        invoked to transform the data into a dictionary of model inputs.
        """

        def wrapper(model_cls: N) -> N:
            if "_processor_factory" in model_cls.__dict__:
                logger.warning(
                    "Model class %s already has a multi-modal processor "
                    "registered to %s. It is overwritten by the new one.",
                    model_cls,
                    self,
                )

            model_cls._processor_factory = _ProcessorFactories(
                info=info,
                dummy_inputs=dummy_inputs,
                processor=processor,
            )

            return model_cls

        return wrapper

    def _get_model_cls(self, model_config: "ModelConfig") -> "SupportsMultiModal":
        # Avoid circular import
        from vllm.model_executor.model_loader import get_model_architecture

        model_cls, _ = get_model_architecture(model_config)
        if not hasattr(model_cls, "_processor_factory"):
            raise ValueError(
                f"Model class {model_cls.__name__} has no registered "
                "multimodal processor"
            )
        return cast("SupportsMultiModal", model_cls)

    def _create_processing_ctx(
        self,
        model_config: "ModelConfig",
        tokenizer: TokenizerLike | None = None,
    ) -> InputProcessingContext:
        if tokenizer is None:
            tokenizer = cached_tokenizer_from_config(model_config)

        return InputProcessingContext(model_config, tokenizer)

    def _create_processing_info(
        self,
        model_config: "ModelConfig",
        tokenizer: TokenizerLike | None = None,
    ) -> BaseProcessingInfo:
        model_cls = self._get_model_cls(model_config)
        factories = model_cls._processor_factory
        ctx = self._create_processing_ctx(model_config, tokenizer)
        return factories.info(ctx)

    def get_processing_info(self, model_config: "ModelConfig") -> BaseProcessingInfo:
        return self._create_processing_info(model_config, tokenizer=None)

    def create_processor(
        self,
        model_config: "ModelConfig",
        *,
        tokenizer: TokenizerLike | None = None,
    ) -> BaseMultiModalProcessor[BaseProcessingInfo]:
        """Create a multi-modal processor for a specific model and tokenizer."""
        if not model_config.is_multimodal_model:
            model_name = model_config.served_model_name or model_config.model
            raise ValueError(f"{model_name} is not a multimodal model")

        model_cls = self._get_model_cls(model_config)
        factories = model_cls._processor_factory

        ctx = self._create_processing_ctx(model_config, tokenizer)

        return factories.build_processor(ctx)