vllm.renderers.chat_utils.item_tracker

Classes:

• BaseMultiModalItemTracker –

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

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

Attributes:

• use_unified_vision_chunk_modality (bool) –

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

Source code in vllm/renderers/chat_utils/item_tracker.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)
        self.decoded_budget = MediaDecodedBudget()
        self.download_budget = MediaRequestDownloadBudget()

    @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)

    @property
    def video_processor_name(self) -> str | None:
        return get_video_processor_cls_name(self.model_config)

    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/renderers/chat_utils/item_tracker.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/renderers/chat_utils/item_tracker.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)

_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/renderers/chat_utils/item_tracker.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