vllm.multimodal.parse

Classes:

• DictEmbeddingItems –

  Base class for data items that are expressed as a dictionary of tensors.

• EmbeddingItems –

  Base class for data items that are expressed as a batched embedding tensor,

• ModalityDataItems –

  Represents data items for a modality in

• MultiModalDataItems –

  A normalized MultiModalDataDict

• MultiModalDataParser –

  Parses MultiModalDataDict

• ProcessorBatchItems –

  Base class for data items that are arranged in a list.

• VideoProcessorItems –
• VisionChunkProcessorItems –

  Processor items for vision chunks (unified image and video chunks).

Functions:

• validate_embedding_ndim –

  Validate tensor ndim for multimodal embeddings.

Attributes:

• EmbeddingFieldRole (TypeAlias) –

  What a field of a pre-computed-embedding input carries.

• MultiModalUUIDItems (TypeAlias) –

  A normalized MultiModalUUIDDict

EmbeddingFieldRole = Literal['values', 'metadata'] module-attribute

What a field of a pre-computed-embedding input carries.

"values" is the embedding tensor itself; "metadata" is a key that sizes the prompt's placeholder range. The distinction is what lets one declaration serve both an ordinary request (everything present) and an EC consumer (embeddings arrive through the connector, metadata still in the request).

MultiModalUUIDItems = dict[str, Sequence[str | None]] module-attribute

A normalized MultiModalUUIDDict such that each entry corresponds to a list.

DictEmbeddingItems

Bases: ModalityDataItems[Mapping[str, Tensor], Mapping[str, Tensor]]

Base class for data items that are expressed as a dictionary of tensors.

Usually, the dictionary keys correspond to the outputs of HF processor.

Methods:

• __init__ –

  Args:

Source code in vllm/multimodal/parse.py

class DictEmbeddingItems(
    ModalityDataItems[Mapping[str, torch.Tensor], Mapping[str, torch.Tensor]]
):
    """Base class for data items that are expressed as a dictionary of tensors.

    Usually, the dictionary keys correspond to the outputs of HF processor.
    """

    def __init__(
        self,
        data: Mapping[str, torch.Tensor],
        modality: str,
        required_fields: Set[str],
        fields_factory: Callable[
            [Mapping[str, torch.Tensor]],
            Mapping[str, MultiModalFieldConfig],
        ],
        optional_fields: Set[str] = frozenset(),
    ) -> None:
        """Args:
        data: The dictionary of tensors for this modality.
        modality: The modality these items belong to.
        required_fields: Fields `data` must contain.
        fields_factory: Builds the field config from the data.
        optional_fields: Fields `data` may omit. Which fields these are is
            the caller's decision -- see
            `MultiModalDataParser.embedding_field_sets`, where a deployment
            that receives embeddings through an EC connector makes the
            embeddings optional. They still need a field config, since they
            are used whenever they *are* supplied.

        """
        from transformers.feature_extraction_utils import BatchFeature

        super().__init__(data, modality)

        # Nothing required would leave nothing to size the placeholder range
        # from, so the item would parse into zero entries and silently produce a
        # wrong prompt instead of failing here.
        if not required_fields:
            raise ValueError(
                f"Cannot parse {modality!r} embeddings: every declared field is "
                f"optional ({sorted(optional_fields)}), so nothing is left to "
                "size the placeholder range."
            )

        declared_fields = set(required_fields) | set(optional_fields)

        missing_required_data_keys = set(required_fields) - data.keys()
        if missing_required_data_keys:
            data_keys = set(data.keys())
            msg = (
                f"The data should contain the fields: {set(required_fields)}, "
                f"but only found the following keys: {data_keys}"
            )
            raise ValueError(msg)

        fields_config = fields_factory(data)
        # Check every declared field, not just the required ones: an optional
        # field still needs a config for when it is supplied.
        missing_required_fields = declared_fields - fields_config.keys()
        if missing_required_fields:
            fields = set(fields_config.keys())
            msg = f"{declared_fields=} should be a subset of {fields=}"
            raise ValueError(msg)

        self.fields_config = fields_config
        self.required_fields = set(required_fields)

        self._kwargs = MultiModalKwargsItems.from_hf_inputs(
            BatchFeature(dict(data)),
            fields_config,
        )

    def get_count(self) -> int:
        return len(self._kwargs[self.modality])

    def select(self, indices: Sequence[int]) -> Self:
        selected = self._copy_selected(indices)
        selected._kwargs = MultiModalKwargsItems(
            {self.modality: [self._kwargs[self.modality][index] for index in indices]}
        )
        selected.data = selected._kwargs.get_data()  # type: ignore[assignment]
        return selected

    def get(self, index: int) -> Mapping[str, torch.Tensor]:
        return self._kwargs[self.modality][index].get_data()

    def get_processor_data(self) -> Mapping[str, object]:
        return {}

    def get_passthrough_data(self) -> Mapping[str, object]:
        return self.data

__init__(data, modality, required_fields, fields_factory, optional_fields=frozenset())

Args: data: The dictionary of tensors for this modality. modality: The modality these items belong to. required_fields: Fields data must contain. fields_factory: Builds the field config from the data. optional_fields: Fields data may omit. Which fields these are is the caller's decision -- see MultiModalDataParser.embedding_field_sets, where a deployment that receives embeddings through an EC connector makes the embeddings optional. They still need a field config, since they are used whenever they are supplied.

Source code in vllm/multimodal/parse.py

def __init__(
    self,
    data: Mapping[str, torch.Tensor],
    modality: str,
    required_fields: Set[str],
    fields_factory: Callable[
        [Mapping[str, torch.Tensor]],
        Mapping[str, MultiModalFieldConfig],
    ],
    optional_fields: Set[str] = frozenset(),
) -> None:
    """Args:
    data: The dictionary of tensors for this modality.
    modality: The modality these items belong to.
    required_fields: Fields `data` must contain.
    fields_factory: Builds the field config from the data.
    optional_fields: Fields `data` may omit. Which fields these are is
        the caller's decision -- see
        `MultiModalDataParser.embedding_field_sets`, where a deployment
        that receives embeddings through an EC connector makes the
        embeddings optional. They still need a field config, since they
        are used whenever they *are* supplied.

    """
    from transformers.feature_extraction_utils import BatchFeature

    super().__init__(data, modality)

    # Nothing required would leave nothing to size the placeholder range
    # from, so the item would parse into zero entries and silently produce a
    # wrong prompt instead of failing here.
    if not required_fields:
        raise ValueError(
            f"Cannot parse {modality!r} embeddings: every declared field is "
            f"optional ({sorted(optional_fields)}), so nothing is left to "
            "size the placeholder range."
        )

    declared_fields = set(required_fields) | set(optional_fields)

    missing_required_data_keys = set(required_fields) - data.keys()
    if missing_required_data_keys:
        data_keys = set(data.keys())
        msg = (
            f"The data should contain the fields: {set(required_fields)}, "
            f"but only found the following keys: {data_keys}"
        )
        raise ValueError(msg)

    fields_config = fields_factory(data)
    # Check every declared field, not just the required ones: an optional
    # field still needs a config for when it is supplied.
    missing_required_fields = declared_fields - fields_config.keys()
    if missing_required_fields:
        fields = set(fields_config.keys())
        msg = f"{declared_fields=} should be a subset of {fields=}"
        raise ValueError(msg)

    self.fields_config = fields_config
    self.required_fields = set(required_fields)

    self._kwargs = MultiModalKwargsItems.from_hf_inputs(
        BatchFeature(dict(data)),
        fields_config,
    )

EmbeddingItems

Bases: ModalityDataItems[Tensor | list[Tensor], Tensor]

Base class for data items that are expressed as a batched embedding tensor, or a list of embedding tensors (one per item).

Source code in vllm/multimodal/parse.py

class EmbeddingItems(
    ModalityDataItems[torch.Tensor | list[torch.Tensor], torch.Tensor]
):
    """Base class for data items that are expressed as a batched embedding tensor,
    or a list of embedding tensors (one per item).
    """

    def __init__(
        self,
        data: torch.Tensor | list[torch.Tensor],
        modality: str,
        expected_hidden_size: int | None = None,
    ) -> None:
        super().__init__(data, modality)

        # Validate ndim first (before hidden_size which depends on correct ndim)
        self._validate_ndim()

        # Validate hidden dimension if expected size is provided
        if expected_hidden_size is not None:
            self._validate_hidden_size(expected_hidden_size)

    def _validate_ndim(self) -> None:
        """Validate that embedding tensors have correct ndim (2D or 3D)."""
        if isinstance(self.data, torch.Tensor):
            validate_embedding_ndim(self.data, self.modality)
        else:
            # List of tensors: each should be 2D (seq_len, hidden_size)
            for idx, tensor in enumerate(self.data):
                if tensor.ndim != 2:
                    raise ValueError(
                        f"{self.modality.capitalize()} embedding [{idx}] must be "
                        f"2D (seq_len, hidden_size), got {tensor.ndim}D tensor "
                        f"with shape {tuple(tensor.shape)}"
                    )

    def _validate_hidden_size(self, expected_hidden_size: int) -> None:
        """Validate that embedding hidden dimension matches expected size.

        This validates hidden dimensions to prevent vulnerabilities: Embeddings
        with correct ndim but wrong hidden dimension could bypass initial
        checks and cause crashes during model inference when dimensions don't match.
        """
        if isinstance(self.data, torch.Tensor):
            # Batched tensor: shape is (batch, seq_len, hidden_size)
            actual_hidden_size = self.data.shape[-1]
            if actual_hidden_size != expected_hidden_size:
                raise ValueError(
                    f"{self.modality.capitalize()} embedding hidden dimension "
                    f"mismatch: got {actual_hidden_size}, but model expects "
                    f"{expected_hidden_size}. Embedding shape: {tuple(self.data.shape)}"
                )
        else:
            # List of tensors: each has shape (seq_len, hidden_size)
            for idx, tensor in enumerate(self.data):
                actual_hidden_size = tensor.shape[-1]
                if actual_hidden_size != expected_hidden_size:
                    raise ValueError(
                        f"{self.modality.capitalize()} embedding [{idx}] hidden "
                        f"dimension mismatch: got {actual_hidden_size}, but model "
                        f"expects {expected_hidden_size}. "
                        f"Embedding shape: {tuple(tensor.shape)}"
                    )

    def select(self, indices: Sequence[int]) -> Self:
        selected = self._copy_selected(indices)
        if isinstance(self.data, torch.Tensor):
            selected.data = self.data[list(indices)]
        else:
            selected.data = [self.data[index] for index in indices]
        return selected

    def get_count(self) -> int:
        return len(self.data)

    def get(self, index: int) -> torch.Tensor:
        return self.data[index]

    def get_processor_data(self) -> Mapping[str, object]:
        return {}

    def get_passthrough_data(self) -> Mapping[str, object]:
        return {f"{self.modality}_embeds": self.data}

    def get_feature_size(self, item_idx: int) -> int:
        return len(self.get(item_idx))

_validate_hidden_size(expected_hidden_size)

Validate that embedding hidden dimension matches expected size.

This validates hidden dimensions to prevent vulnerabilities: Embeddings with correct ndim but wrong hidden dimension could bypass initial checks and cause crashes during model inference when dimensions don't match.

Source code in vllm/multimodal/parse.py

def _validate_hidden_size(self, expected_hidden_size: int) -> None:
    """Validate that embedding hidden dimension matches expected size.

    This validates hidden dimensions to prevent vulnerabilities: Embeddings
    with correct ndim but wrong hidden dimension could bypass initial
    checks and cause crashes during model inference when dimensions don't match.
    """
    if isinstance(self.data, torch.Tensor):
        # Batched tensor: shape is (batch, seq_len, hidden_size)
        actual_hidden_size = self.data.shape[-1]
        if actual_hidden_size != expected_hidden_size:
            raise ValueError(
                f"{self.modality.capitalize()} embedding hidden dimension "
                f"mismatch: got {actual_hidden_size}, but model expects "
                f"{expected_hidden_size}. Embedding shape: {tuple(self.data.shape)}"
            )
    else:
        # List of tensors: each has shape (seq_len, hidden_size)
        for idx, tensor in enumerate(self.data):
            actual_hidden_size = tensor.shape[-1]
            if actual_hidden_size != expected_hidden_size:
                raise ValueError(
                    f"{self.modality.capitalize()} embedding [{idx}] hidden "
                    f"dimension mismatch: got {actual_hidden_size}, but model "
                    f"expects {expected_hidden_size}. "
                    f"Embedding shape: {tuple(tensor.shape)}"
                )

_validate_ndim()

Validate that embedding tensors have correct ndim (2D or 3D).

Source code in vllm/multimodal/parse.py

def _validate_ndim(self) -> None:
    """Validate that embedding tensors have correct ndim (2D or 3D)."""
    if isinstance(self.data, torch.Tensor):
        validate_embedding_ndim(self.data, self.modality)
    else:
        # List of tensors: each should be 2D (seq_len, hidden_size)
        for idx, tensor in enumerate(self.data):
            if tensor.ndim != 2:
                raise ValueError(
                    f"{self.modality.capitalize()} embedding [{idx}] must be "
                    f"2D (seq_len, hidden_size), got {tensor.ndim}D tensor "
                    f"with shape {tuple(tensor.shape)}"
                )

ModalityDataItems

Bases: ABC, Generic[_T, _I]

Represents data items for a modality in MultiModalDataItems.

Methods:

• get –

  Get a data item by its index.

• get_all –

  Get all data items.

• get_count –

  Get the number of data items.

• get_passthrough_data –

  Get the data to pass directly to the model.

• get_processor_data –

  Get the data to pass to the HF processor.

• get_raw –

  Get a data item as stored, without decoding it.

• select –

  Select parsed items without repeating validation or normalization.

Source code in vllm/multimodal/parse.py

class ModalityDataItems(ABC, Generic[_T, _I]):
    """Represents data items for a modality in
    [`MultiModalDataItems`][vllm.multimodal.parse.MultiModalDataItems].
    """

    def __init__(self, data: _T, modality: str) -> None:
        super().__init__()

        self.data: _T = data
        self.modality = modality
        self._original_indices: list[int] | None = None

    def get_original_index(self, index: int) -> int:
        return (
            index if self._original_indices is None else self._original_indices[index]
        )

    @abstractmethod
    def select(self, indices: Sequence[int]) -> Self:
        """Select parsed items without repeating validation or normalization."""
        raise NotImplementedError

    def _copy_selected(self, indices: Sequence[int]) -> Self:
        selected = copy(self)
        selected._original_indices = [
            self.get_original_index(index) for index in indices
        ]
        return selected

    def __repr__(self) -> str:
        return f"{type(self).__name__}(modality={self.modality!r}, len={len(self)})"

    def __len__(self) -> int:
        return self.get_count()

    def __getitem__(self, index: int) -> _I:
        return self.get(index)

    if TYPE_CHECKING:
        # Auto-generated
        def __iter__(self) -> Iterator[_I]: ...

    @abstractmethod
    def get_count(self) -> int:
        """Get the number of data items."""
        raise NotImplementedError

    @abstractmethod
    def get(self, index: int) -> _I:
        """Get a data item by its index."""
        raise NotImplementedError

    def get_all(self) -> list[_I]:
        """Get all data items."""
        return [self.get(idx) for idx in range(self.get_count())]

    def get_raw(self, index: int) -> object:
        """Get a data item as stored, without decoding it.

        For fetched media this is the
        [`MediaRef`][vllm.multimodal.media.base.MediaRef] itself, whose
        precomputed `key` is the cache identity. Hashing, cache-miss
        selection and byte release all read items through here so that none
        of them pays for a decode.
        """
        return self.get(index)

    def get_all_raw(self) -> list[object]:
        return [self.get_raw(idx) for idx in range(self.get_count())]

    @abstractmethod
    def get_processor_data(self) -> Mapping[str, object]:
        """Get the data to pass to the HF processor."""
        raise NotImplementedError

    @abstractmethod
    def get_passthrough_data(self) -> Mapping[str, object]:
        """Get the data to pass directly to the model."""
        raise NotImplementedError

get(index) abstractmethod

Get a data item by its index.

Source code in vllm/multimodal/parse.py

@abstractmethod
def get(self, index: int) -> _I:
    """Get a data item by its index."""
    raise NotImplementedError

get_all()

Get all data items.

Source code in vllm/multimodal/parse.py

def get_all(self) -> list[_I]:
    """Get all data items."""
    return [self.get(idx) for idx in range(self.get_count())]

get_count() abstractmethod

Get the number of data items.

Source code in vllm/multimodal/parse.py

@abstractmethod
def get_count(self) -> int:
    """Get the number of data items."""
    raise NotImplementedError

get_passthrough_data() abstractmethod

Get the data to pass directly to the model.

Source code in vllm/multimodal/parse.py

@abstractmethod
def get_passthrough_data(self) -> Mapping[str, object]:
    """Get the data to pass directly to the model."""
    raise NotImplementedError

get_processor_data() abstractmethod

Get the data to pass to the HF processor.

Source code in vllm/multimodal/parse.py

@abstractmethod
def get_processor_data(self) -> Mapping[str, object]:
    """Get the data to pass to the HF processor."""
    raise NotImplementedError

get_raw(index)

Get a data item as stored, without decoding it.

For fetched media this is the MediaRef itself, whose precomputed key is the cache identity. Hashing, cache-miss selection and byte release all read items through here so that none of them pays for a decode.

Source code in vllm/multimodal/parse.py

def get_raw(self, index: int) -> object:
    """Get a data item as stored, without decoding it.

    For fetched media this is the
    [`MediaRef`][vllm.multimodal.media.base.MediaRef] itself, whose
    precomputed `key` is the cache identity. Hashing, cache-miss
    selection and byte release all read items through here so that none
    of them pays for a decode.
    """
    return self.get(index)

select(indices) abstractmethod

Select parsed items without repeating validation or normalization.

Source code in vllm/multimodal/parse.py

@abstractmethod
def select(self, indices: Sequence[int]) -> Self:
    """Select parsed items without repeating validation or normalization."""
    raise NotImplementedError

MultiModalDataItems

Bases: UserDict[str, ModalityDataItems[Any, Any]]

A normalized MultiModalDataDict such that each entry corresponds to a list.

Methods:

• get_all_counts –

  Get the number of items belonging to each modality.

• get_count –

  Get the number of data items belonging to a modality.

• get_items –

  Get the data items belonging to a modality,

• select –

  Construct a new MultiModalDataItems instance containing only the

Source code in vllm/multimodal/parse.py

class MultiModalDataItems(UserDict[str, ModalityDataItems[Any, Any]]):
    """A normalized [`MultiModalDataDict`][vllm.inputs.MultiModalDataDict]
    such that each entry corresponds to a list.
    """

    def select(self, modalities: Set[str]):
        """Construct a new `MultiModalDataItems` instance containing only the
        selected modalities.
        """
        return MultiModalDataItems(
            {modality: self[modality] for modality in modalities}
        )

    def get_count(self, modality: str, *, strict: bool = True) -> int:
        """Get the number of data items belonging to a modality.

        If `strict=False`, return `0` instead of raising [`KeyError`][]
        even if the modality is not found.
        """
        if modality not in self:
            if strict:
                available_modalities = set(self.keys())
                raise KeyError(
                    f"Modality {modality!r} not found. "
                    f"Available modalities: {available_modalities}"
                )

            return 0

        return self[modality].get_count()

    def get_all_counts(self) -> Mapping[str, int]:
        """Get the number of items belonging to each modality."""
        return {m: items.get_count() for m, items in self.items()}

    def get_items(
        self,
        modality: str,
        typ: type[_D] | tuple[type[_D], ...],
    ) -> _D:
        """Get the data items belonging to a modality,
        requiring that they belong to a certain type.
        """
        if modality not in self:
            available_modalities = set(self.keys())
            raise KeyError(
                f"Modality {modality!r} not found. "
                f"Available modalities: {available_modalities}"
            )

        items = self[modality]
        if not isinstance(items, typ):
            raise TypeError(
                f"Invalid type of data items for {modality=}. "
                f"Expected type: {typ}, but "
                f"found type: {type(items)}"
            )

        return items  # type: ignore[return-value]

get_all_counts()

Get the number of items belonging to each modality.

Source code in vllm/multimodal/parse.py

def get_all_counts(self) -> Mapping[str, int]:
    """Get the number of items belonging to each modality."""
    return {m: items.get_count() for m, items in self.items()}

get_count(modality, *, strict=True)

Get the number of data items belonging to a modality.

If strict=False, return 0 instead of raising KeyError even if the modality is not found.

Source code in vllm/multimodal/parse.py

def get_count(self, modality: str, *, strict: bool = True) -> int:
    """Get the number of data items belonging to a modality.

    If `strict=False`, return `0` instead of raising [`KeyError`][]
    even if the modality is not found.
    """
    if modality not in self:
        if strict:
            available_modalities = set(self.keys())
            raise KeyError(
                f"Modality {modality!r} not found. "
                f"Available modalities: {available_modalities}"
            )

        return 0

    return self[modality].get_count()

get_items(modality, typ)

Get the data items belonging to a modality, requiring that they belong to a certain type.

Source code in vllm/multimodal/parse.py

def get_items(
    self,
    modality: str,
    typ: type[_D] | tuple[type[_D], ...],
) -> _D:
    """Get the data items belonging to a modality,
    requiring that they belong to a certain type.
    """
    if modality not in self:
        available_modalities = set(self.keys())
        raise KeyError(
            f"Modality {modality!r} not found. "
            f"Available modalities: {available_modalities}"
        )

    items = self[modality]
    if not isinstance(items, typ):
        raise TypeError(
            f"Invalid type of data items for {modality=}. "
            f"Expected type: {typ}, but "
            f"found type: {type(items)}"
        )

    return items  # type: ignore[return-value]

select(modalities)

Construct a new MultiModalDataItems instance containing only the selected modalities.

Source code in vllm/multimodal/parse.py

def select(self, modalities: Set[str]):
    """Construct a new `MultiModalDataItems` instance containing only the
    selected modalities.
    """
    return MultiModalDataItems(
        {modality: self[modality] for modality in modalities}
    )

MultiModalDataParser

Parses MultiModalDataDict into MultiModalDataItems.

Parameters:

• target_sr

  (float, default: None ) –

  Enables automatic resampling of audio items to the model's expected sampling rate.

• audio_resample_method

  (str, default: 'torchaudio' ) –

  Backend used for the resampling above. Defaults to torchaudio; models with specific needs may override (e.g. phi4mm uses scipy).

• target_channels

  (int, default: None ) –

  Target number of audio channels. If provided, normalizes audio to this many channels (e.g., 1 for mono). If None, audio channels are passed through unchanged.

• expected_hidden_size

  (int, default: None ) –

  Expected hidden dimension for embedding inputs. If provided, validates that user-supplied embeddings have the correct hidden size to prevent crashes during model inference.

• allow_missing_mm_embeddings

  (bool, default: False ) –

  Whether pre-computed embedding tensors may be absent from the request on a disaggregated consumer. Derived by BaseProcessingInfo.allow_missing_mm_embeddings.

Methods:

• embedding_field_sets –

  modality's (required, optional) fields for this deployment.

• placeholder_metadata_fields –

  The keys that size modality's placeholder range.

Attributes:

• embedding_fields (Mapping[str, Mapping[str, EmbeddingFieldRole]]) –

  Per-modality field roles for pre-computed-embedding inputs.

Source code in vllm/multimodal/parse.py

class MultiModalDataParser:
    """Parses [`MultiModalDataDict`][vllm.inputs.MultiModalDataDict]
    into [`MultiModalDataItems`][vllm.multimodal.parse.MultiModalDataItems].

    Args:
        target_sr (float, optional): Enables automatic resampling of audio
            items to the model's expected sampling rate.
        audio_resample_method (str): Backend used for the resampling above.
            Defaults to torchaudio; models with specific needs may override
            (e.g. phi4mm uses scipy).
        target_channels (int, optional): Target number of audio channels.
            If provided, normalizes audio to this many channels (e.g., 1 for mono).
            If None, audio channels are passed through unchanged.
        expected_hidden_size (int, optional): Expected hidden dimension for
            embedding inputs. If provided, validates that user-supplied
            embeddings have the correct hidden size to prevent crashes
            during model inference.
        allow_missing_mm_embeddings (bool): Whether pre-computed embedding
            tensors may be absent from the request on a disaggregated consumer.
            Derived by `BaseProcessingInfo.allow_missing_mm_embeddings`.

    """

    embedding_fields: Mapping[str, Mapping[str, EmbeddingFieldRole]] = {}
    """Per-modality field roles for pre-computed-embedding inputs.

    Declared here rather than inside `_parse_*_data` so an EC producer can read
    it too: on a producer the request carries real media, so the branch that
    builds `DictEmbeddingItems` never runs, yet the producer still has to know
    which processed keys to publish. One declaration, both sides, no drift.

    A modality absent from this mapping can only be sent whole in the request.
    """

    @classmethod
    def placeholder_metadata_fields(cls, modality: str) -> set[str]:
        """The keys that size `modality`'s placeholder range.

        What an EC producer publishes alongside the embedding, and what a
        consumer keeps requiring once the embedding itself is gone.
        """
        return {
            field
            for field, role in cls.embedding_fields.get(modality, {}).items()
            if role == "metadata"
        }

    def embedding_field_sets(self, modality: str) -> tuple[set[str], set[str]]:
        """`modality`'s (required, optional) fields for this deployment.

        Resolves the static roles in `embedding_fields` against where the
        embeddings actually come from: on an EC consumer they arrive through the
        connector, so the request may omit them; anywhere else a request that
        claims to carry pre-computed embeddings has to actually carry them.
        """
        metadata = self.placeholder_metadata_fields(modality)
        values = set(self.embedding_fields.get(modality, {})) - metadata
        if self.allow_missing_mm_embeddings:
            return metadata, values
        return metadata | values, set()

    def __init__(
        self,
        *,
        target_sr: float | None = None,
        target_channels: int | None = None,
        audio_resample_method: Literal["pyav", "scipy", "soxr", "torchaudio"] = (
            "torchaudio"
        ),
        video_needs_metadata: bool = False,
        expected_hidden_size: int | None = None,
        allow_missing_mm_embeddings: bool = False,
    ) -> None:
        super().__init__()

        self.allow_missing_mm_embeddings = allow_missing_mm_embeddings

        self.audio_resampler = AudioResampler(
            target_sr=target_sr,
            method=audio_resample_method,
        )
        self.target_channels = target_channels
        self.video_needs_metadata = video_needs_metadata
        self.expected_hidden_size = expected_hidden_size

    @classmethod
    def is_embeddings(
        cls, data: object
    ) -> TypeGuard[torch.Tensor | list[torch.Tensor]]:
        if isinstance(data, torch.Tensor):
            return data.ndim == 3
        if is_list_of(data, torch.Tensor) and len(data) > 0:
            return data[0].ndim == 2  # type: ignore[index]

        return False

    def _get_audio_with_sr(
        self,
        audio: AudioItem,
    ) -> tuple[np.ndarray, float | None]:
        if isinstance(audio, tuple):
            return audio
        if isinstance(audio, list):
            return np.array(audio), None
        if isinstance(audio, np.ndarray):
            return audio, None
        if isinstance(audio, torch.Tensor):
            return audio.numpy(), None
        if isinstance(audio, MediaRef):
            # Only reached by transforms stacked via MediaRef.map(), whose
            # input is the decoded media by construction.
            return audio.decode()

        assert_never(audio)

    def _get_video_with_metadata(
        self,
        video: VideoItem,
    ) -> tuple[HfVideoItem | MediaRef[DecodedVideo], dict[str, Any] | None]:
        if isinstance(video, MediaRef):
            # Keep metadata in the raw ref so cache miss reparsing preserves it.
            def unpack(decoded: DecodedVideo) -> DecodedVideo:
                if isinstance(decoded, tuple):
                    frames, metadata = decoded
                else:
                    frames, metadata = decoded, None
                if self.video_needs_metadata and metadata is None:
                    raise ValueError(
                        "Video metadata is required but not found in mm input. "
                        "Please check your video input in `multi_modal_data`"
                    )
                return (frames, metadata) if metadata is not None else frames

            return video.map(unpack), None
        if isinstance(video, tuple):
            return video
        if isinstance(video, list):
            return np.array(video), None
        if isinstance(video, (np.ndarray, torch.Tensor)):
            # Tensors pass through untouched: HF video processors accept them
            # directly, and device tensors (e.g. NVDEC-decoded frames) must
            # stay on-device to avoid a D2H round-trip.
            return video, None

        assert_never(video)

    def _resample_normalize_audio(self, loaded: AudioItem) -> NormalizedAudio:
        audio, orig_sr = self._get_audio_with_sr(loaded)
        if orig_sr is None:
            new_audio = audio
        else:
            new_audio = self.audio_resampler.resample(audio, orig_sr=orig_sr)

        # Apply channel normalization if target_channels is set
        if self.target_channels is not None:
            spec = AudioSpec(target_channels=self.target_channels)
            new_audio = normalize_audio(new_audio, spec)

        return new_audio

    def _parse_audio_embedding_data(
        self, data: dict[str, torch.Tensor]
    ) -> DictEmbeddingItems:
        counts = data.get("audio_num_tokens")
        if (
            not isinstance(counts, torch.Tensor)
            or counts.dtype not in (torch.int32, torch.int64)
            or counts.ndim not in (1, 2)
            or (counts.ndim == 2 and counts.shape[1] != 1)
            or not torch.all(counts > 0)
        ):
            raise ValueError(
                "audio_num_tokens must contain one positive integer per audio"
            )

        counts = counts.flatten()
        if "audio_embeds" in data:
            embeds = data["audio_embeds"]
            if len(embeds) != len(counts) or any(
                len(embedding) != count for embedding, count in zip(embeds, counts)
            ):
                raise ValueError("audio_num_tokens does not match audio_embeds")

        required, optional = self.embedding_field_sets("audio")
        return DictEmbeddingItems(
            {**data, "audio_num_tokens": counts},
            modality="audio",
            required_fields=required,
            optional_fields=optional,
            fields_factory=lambda _: {
                "audio_embeds": MultiModalFieldConfig.batched("audio"),
                "audio_num_tokens": MultiModalFieldConfig.batched(
                    "audio", keep_on_cpu=True
                ),
            },
        )

    def _parse_audio_data(
        self,
        data: dict[str, torch.Tensor] | ModalityData[AudioItem],
    ) -> ModalityDataItems[Any, Any] | None:
        if data is None:
            return None

        if self.is_embeddings(data):
            return AudioEmbeddingItems(data, self.expected_hidden_size)

        if isinstance(data, dict) and "audio_num_tokens" in self.embedding_fields.get(
            "audio", {}
        ):
            return self._parse_audio_embedding_data(data)

        data_items: list[AudioItem | None]
        if (
            (is_list_of(data, float) and len(data) > 0)
            or (isinstance(data, (np.ndarray, torch.Tensor)) and data.ndim == 1)
            or isinstance(data, (tuple, MediaRef))
        ):
            data_items = [data]
        elif isinstance(data, (np.ndarray, torch.Tensor)):
            data_items = [elem for elem in data]
        else:
            data_items = data  # type: ignore[assignment]

        new_audios = list[NormalizedAudio | MediaRef[NormalizedAudio] | None]()
        for data_item in data_items:
            # Requests can omit audio samples when reusing a cached UUID.
            if data_item is None:
                new_audios.append(None)
                continue

            if isinstance(data_item, MediaRef):
                # Defer decoding; resampling and channel normalization are
                # stacked onto the decode so they run in the parallel decode
                # phase. Their settings extend the spec, so a change to
                # either cannot reuse a stale cache entry.
                new_audios.append(
                    data_item.map(
                        self._resample_normalize_audio,
                        target_sr=self.audio_resampler.target_sr,
                        resample_method=self.audio_resampler.method,
                        target_channels=self.target_channels,
                    )
                )
                continue

            new_audios.append(self._resample_normalize_audio(data_item))

        return AudioProcessorItems(new_audios)

    def _parse_image_data(
        self,
        data: ModalityData[ImageItem],
    ) -> ModalityDataItems[Any, Any] | None:
        if data is None:
            return None

        if self.is_embeddings(data):
            return ImageEmbeddingItems(data, self.expected_hidden_size)

        if isinstance(data, (PILImage.Image, MediaRef)) or (
            isinstance(data, (np.ndarray, torch.Tensor)) and data.ndim == 3
        ):
            data_items = [data]
        elif isinstance(data, (np.ndarray, torch.Tensor)):
            data_items = [elem for elem in data]
        else:
            data_items = data

        data_items = [
            convert_image_mode(normalize_image(item), "RGB")
            if isinstance(item, PILImage.Image)
            else item
            for item in data_items
        ]

        return ImageProcessorItems(data_items)

    def _parse_video_data(
        self,
        data: ModalityData[VideoItem],
    ) -> ModalityDataItems[Any, Any] | None:
        if data is None:
            return None

        if self.is_embeddings(data):
            return VideoEmbeddingItems(data, self.expected_hidden_size)

        data_items: list[VideoItem]
        if (
            (is_list_of(data, PILImage.Image) and len(data) > 0)
            or (
                is_list_of(data, (np.ndarray, torch.Tensor), check="all")
                and len(data) > 0
                and all(item.ndim == 3 for item in data)
            )
            or (isinstance(data, (np.ndarray, torch.Tensor)) and data.ndim == 4)
        ):
            data_items = [data]
        elif isinstance(data, (np.ndarray, torch.Tensor)):
            data_items = [elem for elem in data]
        elif isinstance(data, MediaRef) or (isinstance(data, tuple) and len(data) == 2):
            data_items = [data]
        else:
            data_items = data  # type: ignore[assignment]

        new_videos = list[
            HfVideoItem
            | MediaRef[DecodedVideo]
            | tuple[HfVideoItem, dict[str, Any]]
            | None
        ]()
        metadata_lst: list[dict[str, Any] | None] = []
        for data_item in data_items:
            # Allow None video items, valid requests can contain empty URLs
            # if they use multi-modal uuids.
            if data_item is None:
                new_videos.append(None)
                metadata_lst.append(None)
                continue

            video, metadata = self._get_video_with_metadata(data_item)
            if (
                self.video_needs_metadata
                and metadata is None
                and not isinstance(video, MediaRef)
            ):
                raise ValueError(
                    "Video metadata is required but not found in mm input. "
                    "Please check your video input in `multi_modal_data`"
                )
            if metadata is not None:
                new_videos.append((video, metadata))
            else:
                # For refs the metadata check is deferred to decode time
                # (see _get_video_with_metadata).
                new_videos.append(video)
            metadata_lst.append(metadata)

        return VideoProcessorItems(
            new_videos,
            metadata=metadata_lst,
            video_needs_metadata=self.video_needs_metadata,
        )

    def _parse_vision_chunk_data(
        self,
        data: ModalityData[Any],
    ) -> ModalityDataItems[Any, Any] | None:
        """Parse vision chunk data (unified image and video chunks)."""
        if data is None:
            return None

        if self.is_embeddings(data):
            raise ValueError("Do not support embedding data for vision_chunk right now")

        if isinstance(data, dict):
            data = [data]

        return VisionChunkProcessorItems(data)

    def _get_subparsers(self) -> Mapping[str, ModalityDataParser]:
        return {
            "audio": self._parse_audio_data,
            "image": self._parse_image_data,
            "video": self._parse_video_data,
            "vision_chunk": self._parse_vision_chunk_data,
        }

    def parse_mm_data(self, mm_data: MultiModalDataDict) -> MultiModalDataItems:
        subparsers = self._get_subparsers()

        mm_items = MultiModalDataItems()
        for k, v in mm_data.items():
            if k not in subparsers:
                raise ValueError(f"Unsupported modality: {k}")

            # ignore empty embedding data
            if (parsed_data := subparsers[k](v)) is not None:
                mm_items[k] = parsed_data

        return mm_items

embedding_fields = {} class-attribute instance-attribute

Per-modality field roles for pre-computed-embedding inputs.

Declared here rather than inside _parse_*_data so an EC producer can read it too: on a producer the request carries real media, so the branch that builds DictEmbeddingItems never runs, yet the producer still has to know which processed keys to publish. One declaration, both sides, no drift.

A modality absent from this mapping can only be sent whole in the request.

_parse_vision_chunk_data(data)

Parse vision chunk data (unified image and video chunks).

Source code in vllm/multimodal/parse.py

def _parse_vision_chunk_data(
    self,
    data: ModalityData[Any],
) -> ModalityDataItems[Any, Any] | None:
    """Parse vision chunk data (unified image and video chunks)."""
    if data is None:
        return None

    if self.is_embeddings(data):
        raise ValueError("Do not support embedding data for vision_chunk right now")

    if isinstance(data, dict):
        data = [data]

    return VisionChunkProcessorItems(data)

embedding_field_sets(modality)

modality's (required, optional) fields for this deployment.

Resolves the static roles in embedding_fields against where the embeddings actually come from: on an EC consumer they arrive through the connector, so the request may omit them; anywhere else a request that claims to carry pre-computed embeddings has to actually carry them.

Source code in vllm/multimodal/parse.py

def embedding_field_sets(self, modality: str) -> tuple[set[str], set[str]]:
    """`modality`'s (required, optional) fields for this deployment.

    Resolves the static roles in `embedding_fields` against where the
    embeddings actually come from: on an EC consumer they arrive through the
    connector, so the request may omit them; anywhere else a request that
    claims to carry pre-computed embeddings has to actually carry them.
    """
    metadata = self.placeholder_metadata_fields(modality)
    values = set(self.embedding_fields.get(modality, {})) - metadata
    if self.allow_missing_mm_embeddings:
        return metadata, values
    return metadata | values, set()

placeholder_metadata_fields(modality) classmethod

The keys that size modality's placeholder range.

What an EC producer publishes alongside the embedding, and what a consumer keeps requiring once the embedding itself is gone.

Source code in vllm/multimodal/parse.py

@classmethod
def placeholder_metadata_fields(cls, modality: str) -> set[str]:
    """The keys that size `modality`'s placeholder range.

    What an EC producer publishes alongside the embedding, and what a
    consumer keeps requiring once the embedding itself is gone.
    """
    return {
        field
        for field, role in cls.embedding_fields.get(modality, {}).items()
        if role == "metadata"
    }

ProcessorBatchItems

Bases: ModalityDataItems[Sequence[_T | MediaRef[_T]], _T]

Base class for data items that are arranged in a list.

Methods:

• get –

  Get the decoded data item.

• map_raw –

  Replace raw items in a copy, preserving their request indices.

Source code in vllm/multimodal/parse.py

class ProcessorBatchItems(ModalityDataItems[Sequence[_T | MediaRef[_T]], _T]):
    """Base class for data items that are arranged in a list."""

    def select(self, indices: Sequence[int]) -> Self:
        selected = self._copy_selected(indices)
        selected.data = [self.data[index] for index in indices]
        return selected

    def map_raw(
        self, transform: Callable[[_T | MediaRef[_T]], _T | MediaRef[_T]]
    ) -> Self:
        """Replace raw items in a copy, preserving their request indices."""
        mapped = copy(self)
        mapped.data = [transform(item) for item in self.data]
        return mapped

    def get_count(self) -> int:
        return len(self.data)

    def get(self, index: int) -> _T:
        """Get the decoded data item.

        A `MediaRef` decodes here; by the time items are consumed the
        processor's batch decode has already run, so this is a cache read.
        """
        item = self.data[index]
        return item.decode() if isinstance(item, MediaRef) else item

    def get_raw(self, index: int) -> _T | MediaRef[_T]:
        return self.data[index]

    def get_processor_data(self) -> Mapping[str, object]:
        return {f"{self.modality}s": self.get_all()}

    def get_passthrough_data(self) -> Mapping[str, object]:
        return {}

get(index)

Get the decoded data item.

A MediaRef decodes here; by the time items are consumed the processor's batch decode has already run, so this is a cache read.

Source code in vllm/multimodal/parse.py

def get(self, index: int) -> _T:
    """Get the decoded data item.

    A `MediaRef` decodes here; by the time items are consumed the
    processor's batch decode has already run, so this is a cache read.
    """
    item = self.data[index]
    return item.decode() if isinstance(item, MediaRef) else item

map_raw(transform)

Replace raw items in a copy, preserving their request indices.

Source code in vllm/multimodal/parse.py

def map_raw(
    self, transform: Callable[[_T | MediaRef[_T]], _T | MediaRef[_T]]
) -> Self:
    """Replace raw items in a copy, preserving their request indices."""
    mapped = copy(self)
    mapped.data = [transform(item) for item in self.data]
    return mapped

VideoProcessorItems

Bases: ProcessorBatchItems[DecodedVideo | None]

Methods:

• get_frames –

  Return frames regardless of the HF processor's metadata setting.

• get_metadata –

  Resolve metadata for one video without decoding its siblings.

Attributes:

• metadata (dict[str, Any] | list[dict[str, Any] | None] | None) –

  Resolve fetched video metadata when a consumer requests it.

Source code in vllm/multimodal/parse.py

class VideoProcessorItems(ProcessorBatchItems[DecodedVideo | None]):
    def __init__(
        self,
        data: Sequence[DecodedVideo | MediaRef[DecodedVideo] | None],
        metadata: dict[str, Any] | list[dict[str, Any] | None] | None = None,
        *,
        video_needs_metadata: bool = True,
    ) -> None:
        super().__init__(data, "video")

        self._metadata = metadata
        self.video_needs_metadata = video_needs_metadata

    def select(self, indices: Sequence[int]) -> Self:
        selected = super().select(indices)
        if isinstance(self._metadata, list):
            selected._metadata = [self._metadata[index] for index in indices]
        return selected

    def get(self, index: int) -> DecodedVideo | None:
        video = super().get(index)
        if isinstance(video, tuple):
            frames, metadata = video
            if isinstance(self._metadata, list) and self._metadata[index] is None:
                self._metadata[index] = metadata
            return video if self.video_needs_metadata else frames
        return video

    def get_frames(self, index: int) -> HfVideoItem | None:
        """Return frames regardless of the HF processor's metadata setting."""
        video = super().get(index)
        return video[0] if isinstance(video, tuple) else video

    def get_metadata(self, index: int) -> dict[str, Any] | None:
        """Resolve metadata for one video without decoding its siblings."""
        if isinstance(self._metadata, dict):
            return self._metadata
        if isinstance(self._metadata, list) and self._metadata[index] is not None:
            return self._metadata[index]
        video = super().get(index)
        metadata = video[1] if isinstance(video, tuple) else None
        if isinstance(self._metadata, list):
            self._metadata[index] = metadata
        return metadata

    @property
    def metadata(self) -> dict[str, Any] | list[dict[str, Any] | None] | None:
        """Resolve fetched video metadata when a consumer requests it."""
        if isinstance(self._metadata, list):
            for index in range(self.get_count()):
                self.get_metadata(index)
        return self._metadata

    def get_num_frames(self, item_idx: int) -> int:
        video = self.get_frames(item_idx)
        if video is None:
            raise ValueError(f"Cannot get length of cached video at {item_idx}")

        return len(video)

    def get_frame_size(self, item_idx: int) -> ImageSize:
        video = self.get_frames(item_idx)
        if video is None:
            raise ValueError(f"Cannot get size of cached video at {item_idx}")
        if len(video) == 0:
            raise ValueError(f"Cannot get size of empty video at {item_idx}")

        image = video[0]

        if isinstance(image, PILImage.Image):
            return ImageSize(*image.size)
        if isinstance(image, (np.ndarray, torch.Tensor)):
            if image.ndim == 3 and image.shape[-1] in (1, 3, 4):
                # HWC format (e.g. from np.array(PIL.Image) via
                # _get_video_with_metadata).  PIL images are always
                # channels-last.
                h, w = image.shape[0], image.shape[1]
            else:
                # CHW format (standard PyTorch / numpy convention).
                _, h, w = image.shape
            return ImageSize(w, h)

        assert_never(image)

metadata property

Resolve fetched video metadata when a consumer requests it.

get_frames(index)

Return frames regardless of the HF processor's metadata setting.

Source code in vllm/multimodal/parse.py

def get_frames(self, index: int) -> HfVideoItem | None:
    """Return frames regardless of the HF processor's metadata setting."""
    video = super().get(index)
    return video[0] if isinstance(video, tuple) else video

get_metadata(index)

Resolve metadata for one video without decoding its siblings.

Source code in vllm/multimodal/parse.py

def get_metadata(self, index: int) -> dict[str, Any] | None:
    """Resolve metadata for one video without decoding its siblings."""
    if isinstance(self._metadata, dict):
        return self._metadata
    if isinstance(self._metadata, list) and self._metadata[index] is not None:
        return self._metadata[index]
    video = super().get(index)
    metadata = video[1] if isinstance(video, tuple) else None
    if isinstance(self._metadata, list):
        self._metadata[index] = metadata
    return metadata

VisionChunkProcessorItems

Bases: ProcessorBatchItems[Any]

Processor items for vision chunks (unified image and video chunks).

Source code in vllm/multimodal/parse.py

class VisionChunkProcessorItems(ProcessorBatchItems[Any]):
    """Processor items for vision chunks (unified image and video chunks)."""

    def __init__(self, data: Sequence[Any]) -> None:
        super().__init__(data, "vision_chunk")

validate_embedding_ndim(tensor, modality, index=None)

Validate tensor ndim for multimodal embeddings.

Single embeddings should be 2D (seq_len, hidden_size). Batched embeddings should be 3D (batch, seq_len, hidden_size).

Parameters:

• tensor

  (Tensor) –

  The tensor to validate.

• modality

  (str) –

  The modality name for error messages (e.g., "image", "audio").

• index

  (int | None, default: None ) –

  Optional index for list items, included in error messages.

Source code in vllm/multimodal/parse.py

def validate_embedding_ndim(
    tensor: torch.Tensor,
    modality: str,
    index: int | None = None,
) -> None:
    """Validate tensor ndim for multimodal embeddings.

    Single embeddings should be 2D (seq_len, hidden_size).
    Batched embeddings should be 3D (batch, seq_len, hidden_size).

    Args:
        tensor: The tensor to validate.
        modality: The modality name for error messages (e.g., "image", "audio").
        index: Optional index for list items, included in error messages.

    """
    if tensor.ndim < 2 or tensor.ndim > 3:
        idx_str = f" [{index}]" if index is not None else ""
        raise ValueError(
            f"{modality.capitalize()} embedding{idx_str} must be 2D "
            f"(seq_len, hidden_size) or 3D (batch, seq_len, hidden_size), "
            f"got {tensor.ndim}D tensor with shape {tuple(tensor.shape)}"
        )