vllm.entrypoints.chat_utils

Classes:

• AsyncMultiModalContentParser –
• AudioURL –
• BaseMultiModalItemTracker –

  Tracks multi-modal items in a given request and ensures that the number

• ChatCompletionContentPartAudioEmbedsParam –
• ChatCompletionContentPartAudioParam –
• ChatCompletionContentPartImageEmbedsParam –
• ChatCompletionContentPartPromptEmbedsParam –
• ChatCompletionContentPartVideoEmbedsParam –
• ChatCompletionContentPartVideoParam –
• ChatTemplateResolutionError –

  Raised when chat template resolution fails.

• ConversationMessage –
• CustomChatCompletionContentPILImageParam –

  A simpler version of the param that only accepts a PIL image.

• CustomChatCompletionContentSimpleAudioParam –

  A simpler version of the param that only accepts a plain audio_url.

• CustomChatCompletionContentSimpleImageParam –

  A simpler version of the param that only accepts a plain image_url.

• CustomChatCompletionContentSimpleVideoParam –

  A simpler version of the param that only accepts a plain audio_url.

• CustomChatCompletionContentToolReferenceParam –

  A tool reference content param that only accepts a plain tool name.

• CustomChatCompletionMessageParam –

  Enables custom roles in the Chat Completion API.

• CustomThinkCompletionContentParam –

  A Think Completion Content Param that accepts a plain text and a boolean.

• MultiModalContentParser –
• PILImage –

  A PIL.Image.Image object.

• VideoURL –

Functions:

• get_tool_call_id_type –

  Return the tool-call ID type for a given model configuration.

• has_non_text_content –

  Whether any message in messages has a non-text content part.

• validate_chat_template –

  Raises if the provided chat template appears invalid.

Attributes:

• PROMPT_EMBEDS_PLACEHOLDER_TOKEN (Final[str]) –

  The special token used as a placeholder for each embedding

PROMPT_EMBEDS_PLACEHOLDER_TOKEN = '<prompt_embeds>' module-attribute

The special token used as a placeholder for each embedding position during chat template rendering.

Registered as an additional special token when --enable-prompt-embeds is set. See _ensure_prompt_embeds_placeholder_token in vllm/renderers/hf.py.

AsyncMultiModalContentParser

Bases: BaseMultiModalContentParser

Methods:

• parse_prompt_embeds –

  Schedule async prompt embeds decode and store the coroutine in the tracker.

Source code in vllm/entrypoints/chat_utils.py

class AsyncMultiModalContentParser(BaseMultiModalContentParser):
    def __init__(
        self,
        tracker: AsyncMultiModalItemTracker,
        mm_processor_kwargs: dict[str, Any] | None = None,
    ) -> None:
        super().__init__()

        self._tracker = tracker
        self._mm_processor_kwargs: dict[str, Any] | None = mm_processor_kwargs

    @cached_property
    def _connector(self) -> MediaConnector:
        # Connector setup may probe VLLM_MEDIA_CACHE. Defer it until a request
        # actually contains media so text-only parsing never blocks on that I/O.
        return MEDIA_CONNECTOR_REGISTRY.load(
            envs.VLLM_MEDIA_CONNECTOR,
            allowed_local_media_path=self._tracker.allowed_local_media_path,
            allowed_media_domains=self._tracker.allowed_media_domains,
        )

    @property
    def model_config(self) -> ModelConfig:
        return self._tracker.model_config

    async def _item_with_uuid_async(self, item: object, uuid: str | None):
        return item, uuid

    @override
    def parse_prompt_embeds(self, data: str) -> None:
        """Schedule async prompt embeds decode and store the coroutine in the tracker.

        Like the sync variant, emits a single sentinel `PROMPT_EMBEDS_PLACEHOLDER_TOKEN`
        per content part. Unlike the sync variant, the tensor decode is deferred to a
        thread-pool executor via `safe_load_prompt_embeds_async`.
        """
        if not self.model_config.enable_prompt_embeds:
            raise VLLMValidationError(
                _ENABLE_PROMPT_EMBEDS_ERROR, parameter="prompt_embeds"
            )

        self._tracker.add(
            "prompt_embeds", partial(self._load_prompt_embeds_async, data.encode())
        )
        self._add_placeholder("prompt_embeds", PROMPT_EMBEDS_PLACEHOLDER_TOKEN)

    async def _load_prompt_embeds_async(
        self, data_bytes: bytes
    ) -> tuple[torch.Tensor, None]:
        # Second tuple slot fills the tracker's generic `(item, uuid | None)`
        # contract. prompt_embeds has no UUID concept, so it's always `None`.
        tensor = await safe_load_prompt_embeds_async(self.model_config, data_bytes)
        return tensor, None

    async def _image_with_uuid_async(self, image_url: str | None, uuid: str | None):
        image = (
            await self._connector.fetch_image_async(
                image_url, self._tracker.get_media_io("image")
            )
            if image_url
            else None
        )
        return image, uuid

    def parse_image(self, image_url: str | None, uuid: str | None = None) -> None:
        placeholder = self._tracker.add(
            "image", partial(self._image_with_uuid_async, image_url, uuid)
        )
        self._add_placeholder("image", placeholder)

    def parse_image_embeds(
        self,
        image_embeds: MultiModalEmbedsPayload | None,
        uuid: str | None = None,
    ) -> None:
        mm_config = self.model_config.get_multimodal_config()
        if not mm_config.enable_mm_embeds:
            raise VLLMValidationError(
                "You must set `--enable-mm-embeds` to input `image_embeds`",
                parameter="image_embeds",
            )

        placeholder = self._tracker.add(
            "image_embeds",
            partial(self._image_embeds_with_uuid_async, image_embeds, uuid),
        )
        self._add_placeholder("image", placeholder)

    async def _image_embeds_with_uuid_async(
        self,
        image_embeds: MultiModalEmbedsPayload | None,
        uuid: str | None,
    ):
        if isinstance(image_embeds, dict):
            embeds = await _load_embeds_dict(
                image_embeds, self._connector.fetch_image_embedding_async
            )
        elif isinstance(image_embeds, str):
            embeds = await self._connector.fetch_image_embedding_async(image_embeds)
        else:
            embeds = None
        return embeds, uuid

    def parse_audio_embeds(
        self,
        audio_embeds: MultiModalEmbedsPayload | None,
        uuid: str | None = None,
    ) -> None:
        mm_config = self.model_config.get_multimodal_config()
        if not mm_config.enable_mm_embeds:
            raise VLLMValidationError(
                "You must set `--enable-mm-embeds` to input `audio_embeds`",
                parameter="audio_embeds",
            )

        placeholder = self._tracker.add(
            "audio_embeds",
            partial(self._audio_embeds_with_uuid_async, audio_embeds, uuid),
        )
        self._add_placeholder("audio", placeholder)

    async def _audio_embeds_with_uuid_async(
        self,
        audio_embeds: MultiModalEmbedsPayload | None,
        uuid: str | None,
    ):
        if isinstance(audio_embeds, dict):
            embeds = await _load_embeds_dict(
                audio_embeds, self._connector.fetch_audio_embedding_async
            )
        elif isinstance(audio_embeds, str):
            embeds = await self._connector.fetch_audio_embedding_async(audio_embeds)
        else:
            embeds = None
        return embeds, uuid

    def parse_image_pil(
        self,
        image_pil: Image.Image | None,
        uuid: str | None = None,
    ) -> None:
        placeholder = self._tracker.add(
            "image", partial(self._item_with_uuid_async, image_pil, uuid)
        )
        self._add_placeholder("image", placeholder)

    async def _audio_with_uuid_async(self, audio_url: str | None, uuid: str | None):
        audio = (
            await self._connector.fetch_audio_async(
                audio_url, self._tracker.get_media_io("audio")
            )
            if audio_url
            else None
        )
        return audio, uuid

    def parse_audio(self, audio_url: str | None, uuid: str | None = None) -> None:
        placeholder = self._tracker.add(
            "audio", partial(self._audio_with_uuid_async, audio_url, uuid)
        )
        self._add_placeholder("audio", placeholder)

    def parse_input_audio(
        self, input_audio: InputAudio | None, uuid: str | None = None
    ) -> None:
        if input_audio:
            audio_data = input_audio.get("data", "")
            audio_format = input_audio.get("format", "")
            if audio_data:
                audio_url = f"data:audio/{audio_format};base64,{audio_data}"
            else:
                # If a UUID is provided, audio data may be empty.
                audio_url = None
        else:
            audio_url = None

        return self.parse_audio(audio_url, uuid)

    async def _video_with_uuid_async(self, video_url: str | None, uuid: str | None):
        video = (
            await self._connector.fetch_video_async(
                video_url, self._tracker.get_media_io("video")
            )
            if video_url
            else None
        )
        return video, uuid

    async def _shared_with_uuid_async(
        self,
        fetch: _SharedFetch[Any],
        index: int,
        uuid: str | None,
    ):
        return (await fetch())[index], uuid

    def parse_video(self, video_url: str | None, uuid: str | None = None) -> None:
        video: _AsyncMultiModalItem = partial(
            self._video_with_uuid_async, video_url, uuid
        )
        audio: _AsyncMultiModalItem | None = None

        # Extract audio from video if use_audio_in_video is True
        if (
            video_url
            and self._mm_processor_kwargs
            and self._mm_processor_kwargs.get("use_audio_in_video", False)
        ):
            # The audio track lives in the video payload, so both entries await
            # one download instead of registering two fetches of the same URL.
            fetch = _SharedFetch(
                partial(
                    self._connector.fetch_video_and_audio_async,
                    video_url,
                    self._tracker.get_media_io("video"),
                    self._tracker.get_media_io("audio"),
                )
            )
            video = partial(self._shared_with_uuid_async, fetch, 0, uuid)
            audio = partial(self._shared_with_uuid_async, fetch, 1, uuid)

        placeholder = self._tracker.add("video", video)
        self._add_placeholder("video", placeholder)

        if audio is not None:
            audio_placeholder = self._tracker.add("audio", audio)
            self._add_placeholder("audio", audio_placeholder)

    def parse_video_embeds(
        self,
        video_embeds: MultiModalEmbedsPayload | None,
        uuid: str | None = None,
    ) -> None:
        mm_config = self.model_config.get_multimodal_config()
        if not mm_config.enable_mm_embeds:
            raise VLLMValidationError(
                "You must set `--enable-mm-embeds` to input `video_embeds`",
                parameter="video_embeds",
            )

        placeholder = self._tracker.add(
            "video_embeds",
            partial(self._video_embeds_with_uuid_async, video_embeds, uuid),
        )
        self._add_placeholder("video", placeholder)

    async def _video_embeds_with_uuid_async(
        self,
        video_embeds: MultiModalEmbedsPayload | None,
        uuid: str | None,
    ):
        if isinstance(video_embeds, dict):
            embeds = await _load_embeds_dict(
                video_embeds, self._connector.fetch_video_embedding_async
            )
        elif isinstance(video_embeds, str):
            embeds = await self._connector.fetch_video_embedding_async(video_embeds)
        else:
            embeds = None
        return embeds, uuid

parse_prompt_embeds(data)

Schedule async prompt embeds decode and store the coroutine in the tracker.

Like the sync variant, emits a single sentinel PROMPT_EMBEDS_PLACEHOLDER_TOKEN per content part. Unlike the sync variant, the tensor decode is deferred to a thread-pool executor via safe_load_prompt_embeds_async.

Source code in vllm/entrypoints/chat_utils.py

@override
def parse_prompt_embeds(self, data: str) -> None:
    """Schedule async prompt embeds decode and store the coroutine in the tracker.

    Like the sync variant, emits a single sentinel `PROMPT_EMBEDS_PLACEHOLDER_TOKEN`
    per content part. Unlike the sync variant, the tensor decode is deferred to a
    thread-pool executor via `safe_load_prompt_embeds_async`.
    """
    if not self.model_config.enable_prompt_embeds:
        raise VLLMValidationError(
            _ENABLE_PROMPT_EMBEDS_ERROR, parameter="prompt_embeds"
        )

    self._tracker.add(
        "prompt_embeds", partial(self._load_prompt_embeds_async, data.encode())
    )
    self._add_placeholder("prompt_embeds", PROMPT_EMBEDS_PLACEHOLDER_TOKEN)

AudioURL

Bases: TypedDict

Attributes:

• url (Required[str]) –

  Either a URL of the audio or a data URL with base64 encoded audio data.

Source code in vllm/entrypoints/chat_utils.py

class AudioURL(TypedDict, total=False):
    url: Required[str]
    """
    Either a URL of the audio or a data URL with base64 encoded audio data.
    """

url instance-attribute

Either a URL of the audio or a data URL with base64 encoded audio data.

BaseMultiModalItemTracker

Bases: ABC, Generic[_T]

Tracks multi-modal items in a given request and ensures that the number of multi-modal items in a given request does not exceed the configured maximum per prompt.

Methods:

• add –

  Add a multi-modal item to the current prompt and returns the

• get_media_io –

  The decoder the model's processor wants for modality.

Attributes:

• use_unified_vision_chunk_modality (bool) –

  Check if model uses unified vision_chunk modality for images/videos.

Source code in vllm/entrypoints/chat_utils.py

class BaseMultiModalItemTracker(ABC, Generic[_T]):
    """Tracks multi-modal items in a given request and ensures that the number
    of multi-modal items in a given request does not exceed the configured
    maximum per prompt.
    """

    def __init__(
        self,
        model_config: ModelConfig,
        media_io_kwargs: dict[str, dict[str, Any]] | None = None,
    ):
        super().__init__()

        self._model_config = model_config
        self._media_io_kwargs = media_io_kwargs

        self._items_by_modality = defaultdict[str, list[_T]](list)
        # Track original modality for each vision_chunk item (image or video)
        self._modality_order = defaultdict[str, list[str]](list)

    @cached_property
    def use_unified_vision_chunk_modality(self) -> bool:
        """Check if model uses unified vision_chunk modality for images/videos."""
        return getattr(self._model_config.hf_config, "use_unified_vision_chunk", False)

    @property
    def model_config(self) -> ModelConfig:
        return self._model_config

    @cached_property
    def model_cls(self) -> type[SupportsMultiModal]:
        from vllm.model_executor.model_loader import get_model_cls

        model_cls = get_model_cls(self.model_config)
        return cast(type[SupportsMultiModal], model_cls)

    @property
    def media_io_kwargs(self) -> dict[str, dict[str, Any]] | None:
        return self._media_io_kwargs or (
            self._model_config.multimodal_config.media_io_kwargs
            if self._model_config.multimodal_config
            else None
        )

    @property
    def allowed_local_media_path(self):
        return self._model_config.allowed_local_media_path

    @property
    def allowed_media_domains(self):
        return self._model_config.allowed_media_domains

    @property
    def mm_registry(self):
        return MULTIMODAL_REGISTRY

    @cached_property
    def mm_processor(self):
        return self.mm_registry.create_processor(self.model_config)

    @overload
    def get_media_io(self, modality: Literal["image"]) -> MediaIO[Image.Image]: ...

    @overload
    def get_media_io(
        self, modality: Literal["audio"]
    ) -> MediaIO[tuple["np.ndarray", int | float]]: ...

    @overload
    def get_media_io(self, modality: str) -> MediaIO[Any]: ...

    def get_media_io(self, modality: str) -> MediaIO[Any]:
        """The decoder the model's processor wants for `modality`.

        The connector only transports bytes; which decoder they are paired
        with -- and therefore the decode spec folded into the item's cache
        key -- is the model's decision, made by its processing info.
        """
        return self.mm_processor.info.get_media_io(modality, self.media_io_kwargs)

    def add(self, modality: ModalityStr, item: _T) -> str | None:
        """Add a multi-modal item to the current prompt and returns the
        placeholder string to use, if any.

        An optional uuid can be added which serves as a unique identifier of the
        media.

        Note:
            `prompt_embeds` bypass MM-processor validation because they are
            pre-computed embeddings that do not go through any HF processor, encoder,
            or model-specific placeholder logic. The corresponding placeholder string is
            managed by the parser via `_add_placeholder`, so we return None here.

        """
        add_info = self._validate_add(modality)
        if add_info is None:
            self._items_by_modality["prompt_embeds"].append(item)
            return None

        input_modality, original_modality, use_vision_chunk, num_items = add_info

        # Track original modality for vision_chunk items
        if use_vision_chunk:
            self._items_by_modality[input_modality].append(item)  # type: ignore
            self._modality_order["vision_chunk"].append(original_modality)
        else:
            self._items_by_modality[original_modality].append(item)

        return self.model_cls.get_placeholder_str(modality, num_items)

    def _validate_add(self, modality: ModalityStr) -> tuple[str, str, bool, int] | None:
        """Validate that one more item of the modality can be tracked."""
        if modality == "prompt_embeds":
            return None

        input_modality = modality.replace("_embeds", "")
        original_modality = modality
        use_vision_chunk = (
            self.use_unified_vision_chunk_modality
            and original_modality in ["video", "image"]
        )

        # If use_unified_vision_chunk_modality is enabled,
        # map image/video to vision_chunk
        if use_vision_chunk:
            # To avoid validation fail
            # because models with use_unified_vision_chunk_modality=True
            # will only accept vision_chunk modality.
            input_modality = "vision_chunk"
            num_items = len(self._items_by_modality[input_modality]) + 1
        else:
            num_items = len(self._items_by_modality[original_modality]) + 1

        mm_config = self.model_config.multimodal_config
        if (
            mm_config is not None
            and mm_config.enable_mm_embeds
            and mm_config.get_limit_per_prompt(input_modality) == 0
            and original_modality.endswith("_embeds")
        ):
            # Skip validation: embeddings bypass limit when enable_mm_embeds=True
            pass
        else:
            self.mm_processor.info.validate_num_items(input_modality, num_items)

        return input_modality, original_modality, use_vision_chunk, num_items

    @abstractmethod
    def create_parser(
        self, mm_processor_kwargs: dict[str, Any] | None = None
    ) -> "BaseMultiModalContentParser":
        raise NotImplementedError

use_unified_vision_chunk_modality cached property

Check if model uses unified vision_chunk modality for images/videos.

_validate_add(modality)

Validate that one more item of the modality can be tracked.

Source code in vllm/entrypoints/chat_utils.py

def _validate_add(self, modality: ModalityStr) -> tuple[str, str, bool, int] | None:
    """Validate that one more item of the modality can be tracked."""
    if modality == "prompt_embeds":
        return None

    input_modality = modality.replace("_embeds", "")
    original_modality = modality
    use_vision_chunk = (
        self.use_unified_vision_chunk_modality
        and original_modality in ["video", "image"]
    )

    # If use_unified_vision_chunk_modality is enabled,
    # map image/video to vision_chunk
    if use_vision_chunk:
        # To avoid validation fail
        # because models with use_unified_vision_chunk_modality=True
        # will only accept vision_chunk modality.
        input_modality = "vision_chunk"
        num_items = len(self._items_by_modality[input_modality]) + 1
    else:
        num_items = len(self._items_by_modality[original_modality]) + 1

    mm_config = self.model_config.multimodal_config
    if (
        mm_config is not None
        and mm_config.enable_mm_embeds
        and mm_config.get_limit_per_prompt(input_modality) == 0
        and original_modality.endswith("_embeds")
    ):
        # Skip validation: embeddings bypass limit when enable_mm_embeds=True
        pass
    else:
        self.mm_processor.info.validate_num_items(input_modality, num_items)

    return input_modality, original_modality, use_vision_chunk, num_items

add(modality, item)

Add a multi-modal item to the current prompt and returns the placeholder string to use, if any.

An optional uuid can be added which serves as a unique identifier of the media.

Note

prompt_embeds bypass MM-processor validation because they are pre-computed embeddings that do not go through any HF processor, encoder, or model-specific placeholder logic. The corresponding placeholder string is managed by the parser via _add_placeholder, so we return None here.

Source code in vllm/entrypoints/chat_utils.py

def add(self, modality: ModalityStr, item: _T) -> str | None:
    """Add a multi-modal item to the current prompt and returns the
    placeholder string to use, if any.

    An optional uuid can be added which serves as a unique identifier of the
    media.

    Note:
        `prompt_embeds` bypass MM-processor validation because they are
        pre-computed embeddings that do not go through any HF processor, encoder,
        or model-specific placeholder logic. The corresponding placeholder string is
        managed by the parser via `_add_placeholder`, so we return None here.

    """
    add_info = self._validate_add(modality)
    if add_info is None:
        self._items_by_modality["prompt_embeds"].append(item)
        return None

    input_modality, original_modality, use_vision_chunk, num_items = add_info

    # Track original modality for vision_chunk items
    if use_vision_chunk:
        self._items_by_modality[input_modality].append(item)  # type: ignore
        self._modality_order["vision_chunk"].append(original_modality)
    else:
        self._items_by_modality[original_modality].append(item)

    return self.model_cls.get_placeholder_str(modality, num_items)

get_media_io(modality)

get_media_io(
    modality: Literal["image"],
) -> MediaIO[Image.Image]

get_media_io(
    modality: Literal["audio"],
) -> MediaIO[tuple[np.ndarray, int | float]]

get_media_io(modality: str) -> MediaIO[Any]

The decoder the model's processor wants for modality.

The connector only transports bytes; which decoder they are paired with -- and therefore the decode spec folded into the item's cache key -- is the model's decision, made by its processing info.

Source code in vllm/entrypoints/chat_utils.py

def get_media_io(self, modality: str) -> MediaIO[Any]:
    """The decoder the model's processor wants for `modality`.

    The connector only transports bytes; which decoder they are paired
    with -- and therefore the decode spec folded into the item's cache
    key -- is the model's decision, made by its processing info.
    """
    return self.mm_processor.info.get_media_io(modality, self.media_io_kwargs)

ChatCompletionContentPartAudioEmbedsParam

Bases: TypedDict

Attributes:

• audio_embeds (MultiModalEmbedsPayload | None) –

  The audio embeddings. It can be either:

• type (Required[Literal['audio_embeds']]) –

  The type of the content part.

• uuid (str | None) –

  User-provided UUID of a media. User must guarantee that it is properly

Source code in vllm/entrypoints/chat_utils.py

class ChatCompletionContentPartAudioEmbedsParam(TypedDict, total=False):
    audio_embeds: MultiModalEmbedsPayload | None
    """
    The audio embeddings. It can be either:
    - A single base64 string representing a serialized torch tensor.
    - A dictionary of base64 tensors or numeric JSON metadata arrays.
    """
    type: Required[Literal["audio_embeds"]]
    """The type of the content part."""
    uuid: str | None
    """
    User-provided UUID of a media. User must guarantee that it is properly
    generated and unique for different medias.
    """

audio_embeds instance-attribute

The audio embeddings. It can be either: - A single base64 string representing a serialized torch tensor. - A dictionary of base64 tensors or numeric JSON metadata arrays.

type instance-attribute

The type of the content part.

uuid instance-attribute

User-provided UUID of a media. User must guarantee that it is properly generated and unique for different medias.

ChatCompletionContentPartAudioParam

Bases: TypedDict

Attributes:

• type (Required[Literal['audio_url']]) –

  The type of the content part.

Source code in vllm/entrypoints/chat_utils.py

class ChatCompletionContentPartAudioParam(TypedDict, total=False):
    audio_url: Required[AudioURL]

    type: Required[Literal["audio_url"]]
    """The type of the content part."""

type instance-attribute

The type of the content part.

ChatCompletionContentPartImageEmbedsParam

Bases: TypedDict

Attributes:

• image_embeds (MultiModalEmbedsPayload | None) –

  The image embeddings. It can be either:

• type (Required[Literal['image_embeds']]) –

  The type of the content part.

• uuid (str | None) –

  User-provided UUID of a media. User must guarantee that it is properly

Source code in vllm/entrypoints/chat_utils.py

class ChatCompletionContentPartImageEmbedsParam(TypedDict, total=False):
    image_embeds: MultiModalEmbedsPayload | None
    """
    The image embeddings. It can be either:
    - A single base64 string.
    - A dictionary of base64 tensors or numeric JSON metadata arrays.
    """
    type: Required[Literal["image_embeds"]]
    """The type of the content part."""
    uuid: str | None
    """
    User-provided UUID of a media. User must guarantee that it is properly
    generated and unique for different medias.
    """

image_embeds instance-attribute

The image embeddings. It can be either: - A single base64 string. - A dictionary of base64 tensors or numeric JSON metadata arrays.

type instance-attribute

The type of the content part.

uuid instance-attribute

User-provided UUID of a media. User must guarantee that it is properly generated and unique for different medias.

ChatCompletionContentPartPromptEmbedsParam

Bases: TypedDict

Attributes:

• data (Required[str]) –

  Base64-encoded bytes of a serialized torch.Tensor of shape

• type (Required[Literal['prompt_embeds']]) –

  The type of the content part.

Source code in vllm/entrypoints/chat_utils.py

class ChatCompletionContentPartPromptEmbedsParam(TypedDict, total=False):
    data: Required[str]
    """
    Base64-encoded bytes of a serialized `torch.Tensor` of shape
    `(num_tokens, hidden_size)`. The tensor's `dtype` and `hidden_size` must
    match the model's input embedding layer.
    """
    type: Required[Literal["prompt_embeds"]]
    """The type of the content part."""

data instance-attribute

Base64-encoded bytes of a serialized torch.Tensor of shape (num_tokens, hidden_size). The tensor's dtype and hidden_size must match the model's input embedding layer.

type instance-attribute

The type of the content part.

ChatCompletionContentPartVideoEmbedsParam

Bases: TypedDict

Attributes:

• type (Required[Literal['video_embeds']]) –

  The type of the content part.

• uuid (str | None) –

  User-provided UUID of a media. User must guarantee that it is properly

• video_embeds (MultiModalEmbedsPayload | None) –

  The video embeddings. It can be either:

Source code in vllm/entrypoints/chat_utils.py

class ChatCompletionContentPartVideoEmbedsParam(TypedDict, total=False):
    video_embeds: MultiModalEmbedsPayload | None
    """
    The video embeddings. It can be either:
    - A single base64 string representing a serialized torch tensor.
    - A dictionary of base64 tensors or numeric JSON metadata arrays.
    """
    type: Required[Literal["video_embeds"]]
    """The type of the content part."""
    uuid: str | None
    """
    User-provided UUID of a media. User must guarantee that it is properly
    generated and unique for different medias.
    """

type instance-attribute

The type of the content part.

uuid instance-attribute

User-provided UUID of a media. User must guarantee that it is properly generated and unique for different medias.

video_embeds instance-attribute

The video embeddings. It can be either: - A single base64 string representing a serialized torch tensor. - A dictionary of base64 tensors or numeric JSON metadata arrays.

ChatCompletionContentPartVideoParam

Bases: TypedDict

Attributes:

• type (Required[Literal['video_url']]) –

  The type of the content part.

Source code in vllm/entrypoints/chat_utils.py

class ChatCompletionContentPartVideoParam(TypedDict, total=False):
    video_url: Required[VideoURL]

    type: Required[Literal["video_url"]]
    """The type of the content part."""

type instance-attribute

The type of the content part.

ChatTemplateResolutionError

Bases: ValueError

Raised when chat template resolution fails.

This is a subclass of ValueError for backward compatibility with existing exception handlers.

Source code in vllm/entrypoints/chat_utils.py

class ChatTemplateResolutionError(ValueError):
    """Raised when chat template resolution fails.

    This is a subclass of ValueError for backward compatibility with
    existing exception handlers.
    """

ConversationMessage

Bases: TypedDict

Attributes:

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

  The contents of the message

• name (str | None) –

  The name of the function to call

• reasoning (str | None) –

  The reasoning content for interleaved thinking.

• reasoning_content (str | None) –

  Deprecated: The reasoning content for interleaved thinking.

• role (Required[str]) –

  The role of the message's author.

• task (str | None) –

  Model-specific task marker. Currently passed through for DeepSeek V4.

• tool_call_id (str | None) –

  Tool call that this message is responding to.

• tool_calls (list[ChatCompletionMessageToolCallParam] | None) –

  The tool calls generated by the model, such as function calls.

• tools (list[ChatCompletionFunctionToolParam] | None) –

  The tools for developer role.

Source code in vllm/entrypoints/chat_utils.py

class ConversationMessage(TypedDict, total=False):
    role: Required[str]
    """The role of the message's author."""

    content: str | None | list[dict[str, str]]
    """The contents of the message"""

    tool_call_id: str | None
    """Tool call that this message is responding to."""

    name: str | None
    """The name of the function to call"""

    tool_calls: list[ChatCompletionMessageToolCallParam] | None
    """The tool calls generated by the model, such as function calls."""

    reasoning: str | None
    """The reasoning content for interleaved thinking."""

    reasoning_content: str | None
    """Deprecated: The reasoning content for interleaved thinking."""

    tools: list[ChatCompletionFunctionToolParam] | None
    """The tools for developer role."""

    task: str | None
    """Model-specific task marker. Currently passed through for DeepSeek V4."""

content instance-attribute

The contents of the message

name instance-attribute

The name of the function to call

reasoning instance-attribute

The reasoning content for interleaved thinking.

reasoning_content instance-attribute

Deprecated: The reasoning content for interleaved thinking.

role instance-attribute

The role of the message's author.

task instance-attribute

Model-specific task marker. Currently passed through for DeepSeek V4.

tool_call_id instance-attribute

Tool call that this message is responding to.

tool_calls instance-attribute

The tool calls generated by the model, such as function calls.

tools instance-attribute

The tools for developer role.

CustomChatCompletionContentPILImageParam

Bases: TypedDict

A simpler version of the param that only accepts a PIL image.

Example: { "image_pil": ImageAsset('cherry_blossom').pil_image }

Attributes:

• uuid (str | None) –

  User-provided UUID of a media. User must guarantee that it is properly

Source code in vllm/entrypoints/chat_utils.py

class CustomChatCompletionContentPILImageParam(TypedDict, total=False):
    """A simpler version of the param that only accepts a PIL image.

    Example:
    {
        "image_pil": ImageAsset('cherry_blossom').pil_image
    }

    """

    image_pil: PILImage | None
    uuid: str | None
    """
    User-provided UUID of a media. User must guarantee that it is properly
    generated and unique for different medias.
    """

uuid instance-attribute

User-provided UUID of a media. User must guarantee that it is properly generated and unique for different medias.

CustomChatCompletionContentSimpleAudioParam

Bases: TypedDict

A simpler version of the param that only accepts a plain audio_url.

Example: { "audio_url": "https://example.com/audio.mp3" }

Source code in vllm/entrypoints/chat_utils.py

class CustomChatCompletionContentSimpleAudioParam(TypedDict, total=False):
    """A simpler version of the param that only accepts a plain audio_url.

    Example:
    {
        "audio_url": "https://example.com/audio.mp3"
    }

    """

    audio_url: str | None

CustomChatCompletionContentSimpleImageParam

Bases: TypedDict

A simpler version of the param that only accepts a plain image_url. This is supported by OpenAI API, although it is not documented.

Example: { "image_url": "https://example.com/image.jpg" }

Attributes:

• uuid (str | None) –

  User-provided UUID of a media. User must guarantee that it is properly

Source code in vllm/entrypoints/chat_utils.py

class CustomChatCompletionContentSimpleImageParam(TypedDict, total=False):
    """A simpler version of the param that only accepts a plain image_url.
    This is supported by OpenAI API, although it is not documented.

    Example:
    {
        "image_url": "https://example.com/image.jpg"
    }

    """

    image_url: str | None
    uuid: str | None
    """
    User-provided UUID of a media. User must guarantee that it is properly
    generated and unique for different medias.
    """

uuid instance-attribute

User-provided UUID of a media. User must guarantee that it is properly generated and unique for different medias.

CustomChatCompletionContentSimpleVideoParam

Bases: TypedDict

A simpler version of the param that only accepts a plain audio_url.

Example: { "video_url": "https://example.com/video.mp4" }

Attributes:

• uuid (str | None) –

  User-provided UUID of a media. User must guarantee that it is properly

Source code in vllm/entrypoints/chat_utils.py

class CustomChatCompletionContentSimpleVideoParam(TypedDict, total=False):
    """A simpler version of the param that only accepts a plain audio_url.

    Example:
    {
        "video_url": "https://example.com/video.mp4"
    }

    """

    video_url: str | None
    uuid: str | None
    """
    User-provided UUID of a media. User must guarantee that it is properly
    generated and unique for different medias.
    """

uuid instance-attribute

User-provided UUID of a media. User must guarantee that it is properly generated and unique for different medias.

CustomChatCompletionContentToolReferenceParam

Bases: TypedDict

A tool reference content param that only accepts a plain tool name.

Example: { "name": "get_weather", "type": "tool_reference" }

Attributes:

• name (str) –

  The name of the tool being referenced.

• type (Literal['tool_reference']) –

  The content type.

Source code in vllm/entrypoints/chat_utils.py

class CustomChatCompletionContentToolReferenceParam(TypedDict, total=False):
    """A tool reference content param that only accepts a plain tool name.

    Example:
    {
        "name": "get_weather",
        "type": "tool_reference"
    }

    """

    name: str
    """The name of the tool being referenced."""

    type: Literal["tool_reference"]
    """The content type."""

name instance-attribute

The name of the tool being referenced.

type instance-attribute

The content type.

CustomChatCompletionMessageParam

Bases: TypedDict

Enables custom roles in the Chat Completion API.

Attributes:

• content (str | list[ChatCompletionContentPartParam]) –

  The contents of the message.

• name (str) –

  An optional name for the participant.

• reasoning (str | None) –

  The reasoning content for interleaved thinking.

• role (Required[str]) –

  The role of the message's author.

• task (str | None) –

  Model-specific task marker. Currently passed through for DeepSeek V4.

• tool_call_id (str | None) –

  Tool call that this message is responding to.

• tool_calls (list[ChatCompletionMessageToolCallParam] | None) –

  The tool calls generated by the model, such as function calls.

• tools (list[ChatCompletionFunctionToolParam] | None) –

  The tools for developer role.

Source code in vllm/entrypoints/chat_utils.py

class CustomChatCompletionMessageParam(TypedDict, total=False):
    """Enables custom roles in the Chat Completion API."""

    role: Required[str]
    """The role of the message's author."""

    content: str | list[ChatCompletionContentPartParam]
    """The contents of the message."""

    name: str
    """An optional name for the participant.

    Provides the model information to differentiate between participants of the
    same role.
    """

    tool_call_id: str | None
    """Tool call that this message is responding to."""

    tool_calls: list[ChatCompletionMessageToolCallParam] | None
    """The tool calls generated by the model, such as function calls."""

    reasoning: str | None
    """The reasoning content for interleaved thinking."""

    tools: list[ChatCompletionFunctionToolParam] | None
    """The tools for developer role."""

    task: str | None
    """Model-specific task marker. Currently passed through for DeepSeek V4."""

content instance-attribute

The contents of the message.

name instance-attribute

An optional name for the participant.

Provides the model information to differentiate between participants of the same role.

reasoning instance-attribute

The reasoning content for interleaved thinking.

role instance-attribute

The role of the message's author.

task instance-attribute

Model-specific task marker. Currently passed through for DeepSeek V4.

tool_call_id instance-attribute

Tool call that this message is responding to.

tool_calls instance-attribute

The tool calls generated by the model, such as function calls.

tools instance-attribute

The tools for developer role.

CustomThinkCompletionContentParam

Bases: TypedDict

A Think Completion Content Param that accepts a plain text and a boolean.

Example: { "thinking": "I am thinking about the answer", "closed": True, "type": "thinking" }

Attributes:

• closed (bool) –

  Whether the thinking is closed.

• thinking (Required[str]) –

  The thinking content.

• type (Required[Literal['thinking']]) –

  The thinking type.

Source code in vllm/entrypoints/chat_utils.py

class CustomThinkCompletionContentParam(TypedDict, total=False):
    """A Think Completion Content Param that accepts a plain text and a boolean.

    Example:
    {
        "thinking": "I am thinking about the answer",
        "closed": True,
        "type": "thinking"
    }

    """

    thinking: Required[str]
    """The thinking content."""

    closed: bool
    """Whether the thinking is closed."""

    type: Required[Literal["thinking"]]
    """The thinking type."""

closed instance-attribute

Whether the thinking is closed.

thinking instance-attribute

The thinking content.

type instance-attribute

The thinking type.

MultiModalContentParser

Bases: BaseMultiModalContentParser

Methods:

• parse_prompt_embeds –

  Decode a base64 prompt embeds tensor and store it in the tracker.

Source code in vllm/entrypoints/chat_utils.py

class MultiModalContentParser(BaseMultiModalContentParser):
    def __init__(
        self,
        tracker: MultiModalItemTracker,
        mm_processor_kwargs: dict[str, Any] | None = None,
    ) -> None:
        super().__init__()

        self._tracker = tracker
        self._mm_processor_kwargs = mm_processor_kwargs

    @cached_property
    def _connector(self) -> MediaConnector:
        # Connector setup may probe VLLM_MEDIA_CACHE. Defer it until a request
        # actually contains media so text-only parsing never blocks on that I/O.
        return MEDIA_CONNECTOR_REGISTRY.load(
            envs.VLLM_MEDIA_CONNECTOR,
            allowed_local_media_path=self._tracker.allowed_local_media_path,
            allowed_media_domains=self._tracker.allowed_media_domains,
        )

    @property
    def model_config(self) -> ModelConfig:
        return self._tracker.model_config

    @override
    def parse_prompt_embeds(self, data: str) -> None:
        """Decode a base64 prompt embeds tensor and store it in the tracker.

        Emits a single `PROMPT_EMBEDS_PLACEHOLDER_TOKEN` sentinel per
        content part. The renderer later expands each sentinel to a span of
        `tensor.shape[0]` placeholder tokens after tokenization.
        """
        if not self.model_config.enable_prompt_embeds:
            raise VLLMValidationError(
                _ENABLE_PROMPT_EMBEDS_ERROR, parameter="prompt_embeds"
            )

        tensor = safe_load_prompt_embeds(self.model_config, data.encode())
        self._tracker.add("prompt_embeds", (tensor, None))
        self._add_placeholder("prompt_embeds", PROMPT_EMBEDS_PLACEHOLDER_TOKEN)

    def parse_image(self, image_url: str | None, uuid: str | None = None) -> None:
        image = (
            self._connector.fetch_image(image_url, self._tracker.get_media_io("image"))
            if image_url
            else None
        )

        placeholder = self._tracker.add("image", (image, uuid))
        self._add_placeholder("image", placeholder)

    def parse_image_embeds(
        self,
        image_embeds: MultiModalEmbedsPayload | None,
        uuid: str | None = None,
    ) -> None:
        mm_config = self.model_config.get_multimodal_config()
        if not mm_config.enable_mm_embeds:
            raise VLLMValidationError(
                "You must set `--enable-mm-embeds` to input `image_embeds`",
                parameter="image_embeds",
            )

        if isinstance(image_embeds, dict):
            embeds = {
                k: self._connector.fetch_image_embedding(v) if isinstance(v, str) else v
                for k, v in image_embeds.items()
            }
            placeholder = self._tracker.add("image_embeds", (embeds, uuid))

        if isinstance(image_embeds, str):
            embedding = self._connector.fetch_image_embedding(image_embeds)
            placeholder = self._tracker.add("image_embeds", (embedding, uuid))

        if image_embeds is None:
            placeholder = self._tracker.add("image_embeds", (None, uuid))

        self._add_placeholder("image", placeholder)

    def parse_audio_embeds(
        self,
        audio_embeds: MultiModalEmbedsPayload | None,
        uuid: str | None = None,
    ) -> None:
        mm_config = self.model_config.get_multimodal_config()
        if not mm_config.enable_mm_embeds:
            raise VLLMValidationError(
                "You must set `--enable-mm-embeds` to input `audio_embeds`",
                parameter="audio_embeds",
            )

        if isinstance(audio_embeds, dict):
            embeds = {
                k: self._connector.fetch_audio_embedding(v) if isinstance(v, str) else v
                for k, v in audio_embeds.items()
            }
            placeholder = self._tracker.add("audio_embeds", (embeds, uuid))
        elif isinstance(audio_embeds, str):
            embedding = self._connector.fetch_audio_embedding(audio_embeds)
            placeholder = self._tracker.add("audio_embeds", (embedding, uuid))
        else:
            placeholder = self._tracker.add("audio_embeds", (None, uuid))

        self._add_placeholder("audio", placeholder)

    def parse_image_pil(
        self, image_pil: Image.Image | None, uuid: str | None = None
    ) -> None:
        placeholder = self._tracker.add("image", (image_pil, uuid))
        self._add_placeholder("image", placeholder)

    def parse_audio(self, audio_url: str | None, uuid: str | None = None) -> None:
        audio = (
            self._connector.fetch_audio(audio_url, self._tracker.get_media_io("audio"))
            if audio_url
            else None
        )

        placeholder = self._tracker.add("audio", (audio, uuid))
        self._add_placeholder("audio", placeholder)

    def parse_input_audio(
        self, input_audio: InputAudio | None, uuid: str | None = None
    ) -> None:
        if input_audio:
            audio_data = input_audio.get("data", "")
            audio_format = input_audio.get("format", "")
            if audio_data:
                audio_url = f"data:audio/{audio_format};base64,{audio_data}"
            else:
                # If a UUID is provided, audio data may be empty.
                audio_url = None
        else:
            audio_url = None

        return self.parse_audio(audio_url, uuid)

    def parse_video(self, video_url: str | None, uuid: str | None = None) -> None:
        video = None
        audio = None
        if video_url:
            video_io = self._tracker.get_media_io("video")
            if self._mm_processor_kwargs and self._mm_processor_kwargs.get(
                "use_audio_in_video", False
            ):
                # The audio track lives in the video payload, so one download
                # feeds both decoders instead of fetching the URL twice.
                video, audio = self._connector.fetch_video_and_audio(
                    video_url, video_io, self._tracker.get_media_io("audio")
                )
            else:
                video = self._connector.fetch_video(video_url, video_io)

        placeholder = self._tracker.add("video", (video, uuid))
        self._add_placeholder("video", placeholder)

        # Extract audio from video if use_audio_in_video is True
        if audio is not None:
            audio_placeholder = self._tracker.add("audio", (audio, uuid))
            self._add_placeholder("audio", audio_placeholder)

    def parse_video_embeds(
        self,
        video_embeds: MultiModalEmbedsPayload | None,
        uuid: str | None = None,
    ) -> None:
        mm_config = self.model_config.get_multimodal_config()
        if not mm_config.enable_mm_embeds:
            raise VLLMValidationError(
                "You must set `--enable-mm-embeds` to input `video_embeds`",
                parameter="video_embeds",
            )

        if isinstance(video_embeds, dict):
            embeds = {
                k: self._connector.fetch_video_embedding(v) if isinstance(v, str) else v
                for k, v in video_embeds.items()
            }
            placeholder = self._tracker.add("video_embeds", (embeds, uuid))
        elif isinstance(video_embeds, str):
            embedding = self._connector.fetch_video_embedding(video_embeds)
            placeholder = self._tracker.add("video_embeds", (embedding, uuid))
        else:
            placeholder = self._tracker.add("video_embeds", (None, uuid))

        self._add_placeholder("video", placeholder)

parse_prompt_embeds(data)

Decode a base64 prompt embeds tensor and store it in the tracker.

Emits a single PROMPT_EMBEDS_PLACEHOLDER_TOKEN sentinel per content part. The renderer later expands each sentinel to a span of tensor.shape[0] placeholder tokens after tokenization.

Source code in vllm/entrypoints/chat_utils.py

@override
def parse_prompt_embeds(self, data: str) -> None:
    """Decode a base64 prompt embeds tensor and store it in the tracker.

    Emits a single `PROMPT_EMBEDS_PLACEHOLDER_TOKEN` sentinel per
    content part. The renderer later expands each sentinel to a span of
    `tensor.shape[0]` placeholder tokens after tokenization.
    """
    if not self.model_config.enable_prompt_embeds:
        raise VLLMValidationError(
            _ENABLE_PROMPT_EMBEDS_ERROR, parameter="prompt_embeds"
        )

    tensor = safe_load_prompt_embeds(self.model_config, data.encode())
    self._tracker.add("prompt_embeds", (tensor, None))
    self._add_placeholder("prompt_embeds", PROMPT_EMBEDS_PLACEHOLDER_TOKEN)

PILImage

Bases: BaseModel

A PIL.Image.Image object.

Source code in vllm/entrypoints/chat_utils.py

class PILImage(BaseModel):
    """A PIL.Image.Image object."""

    image_pil: Image.Image
    model_config = ConfigDict(arbitrary_types_allowed=True)

VideoURL

Bases: TypedDict

Attributes:

• url (Required[str]) –

  Either a URL of the video or a data URL with base64 encoded video data.

Source code in vllm/entrypoints/chat_utils.py

class VideoURL(TypedDict, total=False):
    url: Required[str]
    """
    Either a URL of the video or a data URL with base64 encoded video data.
    """

url instance-attribute

Either a URL of the video or a data URL with base64 encoded video data.

_SharedFetch

Bases: Generic[_T]

One fetch awaited by several tracker entries.

use_audio_in_video registers a video item and an audio item for the same URL, and both have to resolve from a single download. The task is created on first await, so a request that fails before its items are resolved does not leave an unawaited task behind.

Source code in vllm/entrypoints/chat_utils.py

class _SharedFetch(Generic[_T]):
    """One fetch awaited by several tracker entries.

    `use_audio_in_video` registers a video item and an audio item for the same
    URL, and both have to resolve from a single download. The task is created
    on first await, so a request that fails before its items are resolved does
    not leave an unawaited task behind.
    """

    def __init__(self, factory: Callable[[], Awaitable[_T]]) -> None:
        super().__init__()

        self._factory = factory
        self._task: asyncio.Future[_T] | None = None

    async def __call__(self) -> _T:
        if self._task is None:
            self._task = asyncio.ensure_future(self._factory())

        return await self._task

_decode_media_ref(data, modality, index)

Return the materialized media, wrapping decode failures.

Decode failures are wrapped as VLLMUnprocessableEntityError so corrupt media surfaces as a client error (422) instead of a server error.

Unlike the processor's _decode_error, this can name a request field: only image and video are remapped to vision_chunk, and only a URL-fetched item is a MediaRef, so {modality}_url is where it came from.

Source code in vllm/entrypoints/chat_utils.py

def _decode_media_ref(data: object, modality: str, index: int) -> object:
    """Return the materialized media, wrapping decode failures.

    Decode failures are wrapped as VLLMUnprocessableEntityError so corrupt
    media surfaces as a client error (422) instead of a server error.

    Unlike the processor's `_decode_error`, this can name a request field:
    only image and video are remapped to `vision_chunk`, and only a
    URL-fetched item is a `MediaRef`, so `{modality}_url` is where it came
    from.
    """
    if not isinstance(data, MediaRef):
        return data
    try:
        return data.decode()
    except Exception as error:
        raise _vision_chunk_decode_error(modality, index, error) from error

_get_full_multimodal_text_prompt(placeholder_storage, texts, interleave_strings, multimodal_content_part_separator='\n')

Combine multimodal prompts for a multimodal language model.

Source code in vllm/entrypoints/chat_utils.py

def _get_full_multimodal_text_prompt(
    placeholder_storage: dict[str, list],
    texts: list[str],
    interleave_strings: bool,
    multimodal_content_part_separator: str = "\n",
) -> str:
    """Combine multimodal prompts for a multimodal language model."""
    # flatten storage to make it looks like
    # {
    #   "<|image|>": 2,
    #   "<|audio|>": 1
    # }
    placeholder_counts = Counter(
        [v for elem in placeholder_storage.values() for v in elem]
    )

    if interleave_strings:
        text_prompt = _get_interleaved_text_prompt(placeholder_storage, texts)
    else:
        text_prompt = "\n".join(texts)

    # Pass interleaved text further in case the user used image placeholders
    # himself, but forgot to disable the 'interleave_strings' flag

    # Look through the text prompt to check for missing placeholders
    missing_placeholders: list[str] = []
    for placeholder in placeholder_counts:
        # For any existing placeholder in the text prompt, we leave it as is
        placeholder_counts[placeholder] -= text_prompt.count(placeholder)

        if placeholder_counts[placeholder] < 0:
            logger.error(
                "Placeholder count is negative! "
                "Ensure that the 'interleave_strings' flag is disabled "
                "(current value: %s) "
                "when manually placing image placeholders.",
                interleave_strings,
            )
            logger.debug("Input prompt: %s", text_prompt)
            raise VLLMValidationError(
                f"Found more '{placeholder}' placeholders in input prompt than "
                "actual multimodal data items.",
                parameter="messages",
            )

        missing_placeholders.extend([placeholder] * placeholder_counts[placeholder])

    # NOTE: Default behaviour: we always add missing placeholders
    # at the front of the prompt, if interleave_strings=False
    if text_prompt:
        return multimodal_content_part_separator.join(
            missing_placeholders + [text_prompt]
        )
    else:
        return multimodal_content_part_separator.join(missing_placeholders)

_indexed_vision_chunks(vision_chunk_items, vision_chunks_modality_order)

Pair each vision_chunk item with its index within its own modality.

Numbering per modality rather than per chunk matches the index the mm processor reports for the same item, so a decode failure names the same position whichever layer caught it.

Source code in vllm/entrypoints/chat_utils.py

def _indexed_vision_chunks(
    vision_chunk_items: list[tuple[object, str | None]],
    vision_chunks_modality_order: list[str],
) -> Iterator[tuple[int, str, object, str | None]]:
    """Pair each vision_chunk item with its index within its own modality.

    Numbering per modality rather than per chunk matches the index the mm
    processor reports for the same item, so a decode failure names the same
    position whichever layer caught it.
    """
    seen: dict[str, int] = {}
    for inner_modality, (data, uuid) in zip(
        vision_chunks_modality_order, vision_chunk_items
    ):
        index = seen.get(inner_modality, 0)
        seen[inner_modality] = index + 1
        yield index, inner_modality, data, uuid

_parse_chat_message_content_mm_part(part)

Parses a given multi-modal content part based on its type.

Parameters:

• part

  (ChatCompletionContentPartParam) –

  A dict containing the content part, with a potential 'type' field.

Returns:

• str –

  A tuple (part_type, content) where:

• _ContentPart –
  • part_type: Type of the part (e.g., 'text', 'image_url').
• tuple[str, _ContentPart] –
  • content: Parsed content (e.g., text, image URL).

Raises:

• ValueError –

  If the 'type' field is missing and no direct URL is found.

Source code in vllm/entrypoints/chat_utils.py

def _parse_chat_message_content_mm_part(
    part: ChatCompletionContentPartParam,
) -> tuple[str, _ContentPart]:
    """Parses a given multi-modal content part based on its type.

    Args:
        part: A dict containing the content part, with a potential 'type' field.

    Returns:
        A tuple (part_type, content) where:
        - part_type: Type of the part (e.g., 'text', 'image_url').
        - content: Parsed content (e.g., text, image URL).

    Raises:
        ValueError: If the 'type' field is missing and no direct URL is found.

    """
    assert isinstance(
        part, dict
    )  # This is needed to avoid mypy errors: part.get() from str
    part_type = part.get("type", None)
    uuid = part.get("uuid", None)

    if isinstance(part_type, str) and part_type in MM_PARSER_MAP and uuid is None:  # noqa: E501
        content = MM_PARSER_MAP[part_type](part)

        # Special case for 'image_url.detail'
        # We only support 'auto', which is the default
        if part_type == "image_url" and part.get("detail", "auto") != "auto":
            logger.warning(
                "'image_url.detail' is currently not supported and will be ignored."
            )

        return part_type, content

    # Handle missing 'type' but provided direct URL fields.
    # 'type' is required field by pydantic
    if part_type is None or uuid is not None:
        if "image_url" in part:
            image_params = cast(CustomChatCompletionContentSimpleImageParam, part)
            image_url = image_params.get("image_url", None)
            if isinstance(image_url, dict):
                # Can potentially happen if user provides a uuid
                # with url as a dict of {"url": url}
                image_url = image_url.get("url", None)
            return "image_url", image_url
        if "image_pil" in part:
            # "image_pil" could be None if UUID is provided.
            image_params = cast(  # type: ignore
                CustomChatCompletionContentPILImageParam, part
            )
            image_pil = image_params.get("image_pil", None)
            return "image_pil", image_pil
        if "image_embeds" in part:
            # "image_embeds" could be None if UUID is provided.
            image_params = cast(  # type: ignore
                ChatCompletionContentPartImageEmbedsParam, part
            )
            image_embeds = image_params.get("image_embeds", None)
            return "image_embeds", image_embeds
        if "audio_embeds" in part:
            # "audio_embeds" could be None if UUID is provided.
            audio_params = cast(  # type: ignore[assignment]
                ChatCompletionContentPartAudioEmbedsParam, part
            )
            audio_embeds = audio_params.get("audio_embeds", None)
            return "audio_embeds", audio_embeds
        if "video_embeds" in part:
            # "video_embeds" could be None if UUID is provided.
            video_embeds_params = cast(  # type: ignore[assignment]
                ChatCompletionContentPartVideoEmbedsParam, part
            )
            video_embeds = video_embeds_params.get("video_embeds", None)
            return "video_embeds", video_embeds
        if "prompt_embeds" in part:
            prompt_embeds_params = cast(  # type: ignore[assignment]
                ChatCompletionContentPartPromptEmbedsParam, part
            )
            return "prompt_embeds", prompt_embeds_params.get("data", None)
        if "audio_url" in part:
            audio_params = cast(  # type: ignore[assignment]
                CustomChatCompletionContentSimpleAudioParam, part
            )
            audio_url = audio_params.get("audio_url", None)
            if isinstance(audio_url, dict):
                # Can potentially happen if user provides a uuid
                # with url as a dict of {"url": url}
                audio_url = audio_url.get("url", None)
            return "audio_url", audio_url
        if "input_audio" in part:
            input_audio_params = _InputAudioParser(part).get("input_audio", None)
            return "input_audio", input_audio_params
        if "video_url" in part:
            video_params = cast(CustomChatCompletionContentSimpleVideoParam, part)
            video_url = video_params.get("video_url", None)
            if isinstance(video_url, dict):
                # Can potentially happen if user provides a uuid
                # with url as a dict of {"url": url}
                video_url = video_url.get("url", None)
            return "video_url", video_url
        if "tool_reference" in part:
            tool_reference_params = cast(
                CustomChatCompletionContentToolReferenceParam, part
            )
            tool_reference = tool_reference_params.get("name", None)
            return "tool_reference", tool_reference
        # Raise an error if no 'type' or direct URL is found.
        raise VLLMValidationError(
            "Missing 'type' field in multimodal part.", parameter="type"
        )

    if not isinstance(part_type, str):
        raise VLLMValidationError(
            "Invalid 'type' field in multimodal part.", parameter="type"
        )
    return part_type, "unknown part_type content"

_parse_chat_message_content_part(part, mm_parser, *, wrap_dicts, interleave_strings)

Parses a single part of a conversation. If wrap_dicts is True, structured dictionary pieces for texts and images will be wrapped in dictionaries, i.e., {"type": "text", "text", ...} and {"type": "image"}, respectively. Otherwise multimodal data will be handled by mm_parser, and texts will be returned as strings to be joined with multimodal placeholders.

Source code in vllm/entrypoints/chat_utils.py

def _parse_chat_message_content_part(
    part: ChatCompletionContentPartParam,
    mm_parser: BaseMultiModalContentParser,
    *,
    wrap_dicts: bool,
    interleave_strings: bool,
) -> _ContentPart | None:
    """Parses a single part of a conversation. If wrap_dicts is True,
    structured dictionary pieces for texts and images will be
    wrapped in dictionaries, i.e., {"type": "text", "text", ...} and
    {"type": "image"}, respectively. Otherwise multimodal data will be
    handled by mm_parser, and texts will be returned as strings to be joined
    with multimodal placeholders.
    """
    if isinstance(part, str):  # Handle plain text parts
        _reject_reserved_placeholder_in_text(part, mm_parser.model_config)
        if wrap_dicts:
            return {"type": "text", "text": part}
        return part
    # Handle structured dictionary parts
    part_type, content = _parse_chat_message_content_mm_part(part)
    # if part_type is text/refusal/image_url/audio_url/video_url/input_audio but
    # content is None, log a warning and skip
    if part_type in PART_TYPES_TO_SKIP_NONE_CONTENT and content is None:
        logger.warning(
            "Skipping multimodal part '%s' (type: '%s') "
            "with empty / unparsable content.",
            part,
            part_type,
        )
        return None

    if part_type in TEXT_PART_TYPES:
        str_content = cast(str, content)
        _reject_reserved_placeholder_in_text(str_content, mm_parser.model_config)
        if wrap_dicts:
            result: dict[str, Any] = {"type": "text", "text": str_content}
            result.update(_collect_extra_fields(cast(dict[str, Any], part)))
            return result
        else:
            return str_content

    # For media items, if a user has provided one, use it. Otherwise, insert
    # a placeholder empty uuid.
    uuid = part.get("uuid", None)
    if uuid is not None:
        uuid = str(uuid)

    modality = None
    if part_type == "image_pil":
        image_content = cast(Image.Image, content) if content is not None else None
        mm_parser.parse_image_pil(image_content, uuid)
        modality = "image"
    elif part_type in ("image_url", "input_image"):
        str_content = cast(str, content)
        mm_parser.parse_image(str_content, uuid)
        modality = "image"
    elif part_type == "image_embeds":
        content = (
            cast(MultiModalEmbedsPayload, content) if content is not None else None
        )
        mm_parser.parse_image_embeds(content, uuid)
        modality = "image"
    elif part_type == "audio_embeds":
        content = (
            cast(MultiModalEmbedsPayload, content) if content is not None else None
        )
        mm_parser.parse_audio_embeds(content, uuid)
        modality = "audio"
    elif part_type == "video_embeds":
        content = (
            cast(MultiModalEmbedsPayload, content) if content is not None else None
        )
        mm_parser.parse_video_embeds(content, uuid)
        modality = "video"
    elif part_type == "prompt_embeds":
        if not content:
            raise VLLMValidationError(
                _PROMPT_EMBEDS_MISSING_DATA_ERROR, parameter="prompt_embeds"
            )
        mm_parser.parse_prompt_embeds(cast(str, content))
        modality = "prompt_embeds"
    elif part_type == "audio_url":
        str_content = cast(str, content)
        mm_parser.parse_audio(str_content, uuid)
        modality = "audio"
    elif part_type == "input_audio":
        dict_content = cast(InputAudio, content)
        mm_parser.parse_input_audio(dict_content, uuid)
        modality = "audio"
    elif part_type == "video_url":
        str_content = cast(str, content)
        mm_parser.parse_video(str_content, uuid)
        modality = "video"
    elif part_type == "tool_reference":
        # Tool references are not multimodal data — they reference deferred
        # tools and are passed through as-is for the chat template to expand.
        if wrap_dicts:
            return {"type": "tool_reference", "name": cast(str, content)}
        return cast(str, content)
    else:
        supported = sorted(MM_PARSER_MAP.keys() | set(PART_TYPES_TO_SKIP_NONE_CONTENT))
        raise VLLMValidationError(
            f"Unsupported chat content part type: {part_type!r}. "
            f"Supported types: {', '.join(supported)}.",
            parameter="type",
            value=part_type,
        )

    if wrap_dicts:
        if modality == "prompt_embeds":
            # Chat templates don't know about the "prompt_embeds" modality,
            # emit the single sentinel token as text so the template renders
            # it inline. The renderer later expands it to N tokens post-tokenize.
            return {"type": "text", "text": PROMPT_EMBEDS_PLACEHOLDER_TOKEN}
        result = {"type": modality}
        result.update(_collect_extra_fields(cast(dict[str, Any], part)))
        return result
    if modality == "prompt_embeds":
        # Emit the renderer token inline regardless of `interleave_strings`,
        # prompt_embeds are spliced at the token offset so position matters.
        # Falling back to front-padding via `missing_placeholders` would
        # reorder them relative to surrounding text.
        return PROMPT_EMBEDS_PLACEHOLDER_TOKEN
    return MODALITY_PLACEHOLDERS_MAP[modality] if interleave_strings else None

_predecode_vision_chunk_items(vision_chunk_items, vision_chunks_modality_order) async

Decode vision_chunk media refs concurrently on the media thread pool.

Waits for every submitted decode to finish, then wraps the first failure with _vision_chunk_decode_error.

Source code in vllm/entrypoints/chat_utils.py

async def _predecode_vision_chunk_items(
    vision_chunk_items: list[tuple[object, str | None]],
    vision_chunks_modality_order: list[str],
) -> None:
    """Decode vision_chunk media refs concurrently on the media thread pool.

    Waits for every submitted decode to finish, then wraps the first failure
    with `_vision_chunk_decode_error`.
    """
    lazy_items = [
        (index, inner_modality, data)
        for index, inner_modality, data, _uuid in _indexed_vision_chunks(
            vision_chunk_items, vision_chunks_modality_order
        )
        if isinstance(data, MediaRef) and not data.is_decoded
    ]
    if not lazy_items:
        return

    jobs: list[MediaDecodeJob] = []
    try:
        submit_media_decodes(
            ((modality, index, data) for index, modality, data in lazy_items),
            global_thread_pool,
            jobs,
        )
    except BaseException:
        # Preserve submission errors after draining already submitted work.
        with suppress(BaseException):
            await collect_media_decodes_async(jobs, _vision_chunk_decode_error)
        raise
    await collect_media_decodes_async(jobs, _vision_chunk_decode_error)

_reject_reserved_placeholder_in_text(text, model_config)

Reject user-supplied text parts that contains the reserved prompt_embeds placeholder sentinel.

When the server accepts prompt_embeds, the placeholder token is registered as a single unsplittable special token on the tokenizer. Any user text that happens to contain the literal sequence would tokenize to the same ID and be mistaken for a splice point by the renderer, letting a caller move or inject splice positions via plain text content.

Source code in vllm/entrypoints/chat_utils.py

def _reject_reserved_placeholder_in_text(text: str, model_config: ModelConfig) -> None:
    """Reject user-supplied text parts that contains the reserved `prompt_embeds`
    placeholder sentinel.

    When the server accepts `prompt_embeds`, the placeholder token is
    registered as a single unsplittable special token on the tokenizer. Any
    user text that happens to contain the literal sequence would tokenize to
    the same ID and be mistaken for a splice point by the renderer, letting a
    caller move or inject splice positions via plain text content.
    """
    if model_config.enable_prompt_embeds and PROMPT_EMBEDS_PLACEHOLDER_TOKEN in text:
        raise VLLMValidationError(
            _RESERVED_PLACEHOLDER_IN_TEXT_ERROR.format(
                token=PROMPT_EMBEDS_PLACEHOLDER_TOKEN
            ),
            parameter="messages",
        )

_resolve_items(items_by_modality, mm_processor, modality_order)

Materialize the tracker's per-modality items into mm_data / mm_uuids.

Note

mm_processor is None for text-only models (no registered HF processor) whose only modality is prompt_embeds. Every other modality requires a processor, enforced by the guard below.

Source code in vllm/entrypoints/chat_utils.py

def _resolve_items(
    items_by_modality: dict[str, list[tuple[object, str | None]]],
    mm_processor: BaseMultiModalProcessor | None,
    modality_order: dict[str, list[str]],
) -> tuple[MultiModalDataDict, MultiModalUUIDDict]:
    """Materialize the tracker's per-modality items into `mm_data` / `mm_uuids`.

    Note:
        `mm_processor` is `None` for text-only models (no registered HF
        processor) whose only modality is `prompt_embeds`. Every other
        modality requires a processor, enforced by the guard below.

    """
    if "image" in items_by_modality and "image_embeds" in items_by_modality:
        raise VLLMValidationError(
            "Mixing raw image and embedding inputs is not allowed",
            parameter="image_embeds",
        )
    if "audio" in items_by_modality and "audio_embeds" in items_by_modality:
        raise VLLMValidationError(
            "Mixing raw audio and embedding inputs is not allowed",
            parameter="audio_embeds",
        )
    if "video" in items_by_modality and "video_embeds" in items_by_modality:
        raise VLLMValidationError(
            "Mixing raw video and embedding inputs is not allowed",
            parameter="video_embeds",
        )
    # `prompt_embeds` bypasses HF MM processors. Every other modality requires one.
    processor_modalities = items_by_modality.keys() - {"prompt_embeds"}
    if processor_modalities and mm_processor is None:
        raise RuntimeError(
            _REQUIRE_MM_PROCESSOR_ERROR.format(modality=processor_modalities)
        )

    mm_data = {}
    mm_uuids = {}
    if "image_embeds" in items_by_modality:
        assert mm_processor is not None
        mm_data["image"] = _get_embeds_data(
            "image",
            [data for data, uuid in items_by_modality["image_embeds"]],
            mm_processor,
        )
        mm_uuids["image"] = [uuid for data, uuid in items_by_modality["image_embeds"]]
    if "image" in items_by_modality:
        mm_data["image"] = [data for data, uuid in items_by_modality["image"]]
        mm_uuids["image"] = [uuid for data, uuid in items_by_modality["image"]]
    if "audio_embeds" in items_by_modality:
        assert mm_processor is not None
        mm_data["audio"] = _get_embeds_data(
            "audio",
            [data for data, uuid in items_by_modality["audio_embeds"]],
            mm_processor,
        )
        mm_uuids["audio"] = [uuid for data, uuid in items_by_modality["audio_embeds"]]
    if "audio" in items_by_modality:
        mm_data["audio"] = [data for data, uuid in items_by_modality["audio"]]
        mm_uuids["audio"] = [uuid for data, uuid in items_by_modality["audio"]]
    if "video" in items_by_modality:
        mm_data["video"] = [data for data, uuid in items_by_modality["video"]]
        mm_uuids["video"] = [uuid for data, uuid in items_by_modality["video"]]
    if "video_embeds" in items_by_modality:
        assert mm_processor is not None
        mm_data["video"] = _get_embeds_data(
            "video",
            [data for data, uuid in items_by_modality["video_embeds"]],
            mm_processor,
        )
        mm_uuids["video"] = [uuid for data, uuid in items_by_modality["video_embeds"]]
    if "vision_chunk" in items_by_modality:
        assert mm_processor is not None
        # Process vision_chunk items - extract from (data, modality) tuples
        # and convert to VisionChunk types with proper UUID handling
        processed_chunks, vision_chunk_uuids = _resolve_vision_chunk_items(
            items_by_modality["vision_chunk"],
            mm_processor,
            modality_order.get("vision_chunk", []),
        )
        mm_data["vision_chunk"] = processed_chunks
        mm_uuids["vision_chunk"] = vision_chunk_uuids
    if "prompt_embeds" in items_by_modality:
        mm_data["prompt_embeds"] = [
            data for data, _uuid in items_by_modality["prompt_embeds"]
        ]

    return mm_data, mm_uuids

get_tool_call_id_type(model_config)

Return the tool-call ID type for a given model configuration.

Source code in vllm/entrypoints/chat_utils.py

def get_tool_call_id_type(model_config: ModelConfig) -> str:
    """Return the tool-call ID type for a given model configuration."""
    hf_overrides = getattr(model_config, "hf_overrides", None)
    hf_config = getattr(model_config, "hf_config", None)
    hf_text_config = getattr(model_config, "hf_text_config", None)
    model_types = (
        getattr(hf_config, "model_type", None),
        getattr(hf_text_config, "model_type", None),
    )
    if any(model_type in _KIMI_MODEL_TYPES for model_type in model_types) or (
        isinstance(hf_overrides, dict)
        and hf_overrides.get("model_type") in _KIMI_MODEL_TYPES
    ):
        return "kimi_k2"
    return "random"

has_non_text_content(messages)

Whether any message in messages has a non-text content part.

Only list content is inspected, so validated chat content that is a one-shot iterator is never consumed.

Source code in vllm/entrypoints/chat_utils.py

def has_non_text_content(messages: Any) -> bool:
    """Whether any message in ``messages`` has a non-text content part.

    Only list content is inspected, so validated chat content that is a
    one-shot iterator is never consumed.
    """
    if not isinstance(messages, list):
        return False
    for msg in messages:
        content = msg.get("content") if isinstance(msg, dict) else None
        if not isinstance(content, list):
            continue
        for part in content:
            if isinstance(part, dict) and (
                any(key in part for key in _MEDIA_CONTENT_PART_KEYS)
                or not isinstance(part_type := part.get("type", "text"), str)
                or part_type not in _TEXT_CONTENT_PART_TYPES
            ):
                return True
    return False

validate_chat_template(chat_template)

Raises if the provided chat template appears invalid.

Source code in vllm/entrypoints/chat_utils.py

def validate_chat_template(chat_template: Path | str | None):
    """Raises if the provided chat template appears invalid."""
    if chat_template is None:
        return

    elif isinstance(chat_template, Path) and not chat_template.exists():
        raise FileNotFoundError("the supplied chat template path doesn't exist")

    elif isinstance(chat_template, str):
        JINJA_CHARS = "{}\n"
        if (
            not any(c in chat_template for c in JINJA_CHARS)
            and not Path(chat_template).exists()
        ):
            # Try to find the template in the built-in templates directory
            from vllm.transformers_utils.chat_templates.registry import (
                CHAT_TEMPLATES_DIR,
            )

            builtin_template_path = CHAT_TEMPLATES_DIR / chat_template
            if not builtin_template_path.exists():
                raise VLLMValidationError(
                    f"The supplied chat template string ({chat_template}) "
                    f"appears path-like, but doesn't exist! "
                    f"Tried: {chat_template} and {builtin_template_path}",
                    parameter="chat_template",
                )

    else:
        raise VLLMValidationError(
            f"{type(chat_template)} is not a valid chat template type",
            parameter="chat_template",
        )