processing.py 39.3 KB
Newer Older
1
import pickle
2
3
import re
from abc import ABC, abstractmethod
4
from collections import defaultdict
5
from collections.abc import Callable, ItemsView, Iterable, Mapping, Sequence
6
from dataclasses import dataclass, field
7
from functools import lru_cache
8
from typing import Any, NamedTuple, Optional, Protocol, TypeVar, Union
9

10
import numpy as np
11
import torch
12
from blake3 import blake3
13
from PIL import Image
14
from transformers import BatchFeature, PretrainedConfig, ProcessorMixin
15

16
from vllm.inputs import DummyData, InputProcessingContext
17
from vllm.logger import init_logger
18
19
from vllm.transformers_utils.tokenizer import (AnyTokenizer, decode_tokens,
                                               encode_tokens)
20
from vllm.utils import LRUCache, flatten_2d_lists, full_groupby
21

22
from .inputs import (MultiModalDataDict, MultiModalFieldConfig,
23
24
                     MultiModalInputsV2, MultiModalKwargs,
                     MultiModalKwargsItem, PlaceholderRange)
25
from .parse import MultiModalDataItems, MultiModalDataParser
26
from .profiling import BaseProfilingInfo
27

28
logger = init_logger(__name__)
29
30

_S = TypeVar("_S", str, list[int])
31
_PromptSeq = Union[str, list[int]]
32

33
34

@dataclass
35
36
class PromptReplacement:
    modality: str
37
    """The modality for which the replacement is made."""
38

39
40
    target: _PromptSeq
    """The text or token sequence to find and replace."""
41

42
43
    replacement: Union[Callable[[int], _PromptSeq],
                       _PromptSeq] = field(repr=False)
44
    """
45
46
    Given the index of the processed item within :attr:`modality`, output the
    replacement text or token sequence.
47

48
49
    For convenience, you can pass in the replacement instead of a function
    if it does not depend on the input.
50
51
    """

52
    def bind(self, tokenizer: AnyTokenizer) -> "_BoundPromptReplacement":
53
        return _BoundPromptReplacement(
54
55
56
57
            tokenizer=tokenizer,
            modality=self.modality,
            _target=self.target,
            _replacement=self.replacement,
58
        )
59
60


61
62
63
64
65
66
67
@lru_cache(maxsize=2048)
def _cached_encode(
    tokenizer: AnyTokenizer,
    text: str,
    *,
    add_special_tokens: bool = False,
) -> list[int]:
68
69
70
    return encode_tokens(tokenizer,
                         text,
                         add_special_tokens=add_special_tokens)
71
72


73
74
75
76
77
78
79
@lru_cache(maxsize=2048)
def _cached_decode(
    tokenizer: AnyTokenizer,
    token_ids: tuple[int, ...],
    *,
    skip_special_tokens: bool = False,
) -> str:
80
81
82
    return decode_tokens(tokenizer,
                         list(token_ids),
                         skip_special_tokens=skip_special_tokens)
83
84
85
86
87


class _HasModalityAttr(Protocol):
    modality: str

88

89
class _HasModalityProp(Protocol):
90

91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
    @property
    def modality(self) -> str:
        ...


_M = TypeVar("_M", bound=Union[_HasModalityAttr, _HasModalityProp])


def full_groupby_modality(values: Iterable[_M]) -> ItemsView[str, list[_M]]:
    """Convenience function to apply :func:`full_groupby` based on modality."""
    return full_groupby(values, key=lambda x: x.modality)


@dataclass
class _BoundPromptSequence:
106
107
    tokenizer: AnyTokenizer = field(repr=False)

108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
    _text: Optional[str]
    _token_ids: Optional[list[int]]

    def __post_init__(self) -> None:
        if self._text is None and self._token_ids is None:
            raise ValueError("At least one of 'text' and 'token_ids' must be "
                             "specified")

    @property
    def text(self) -> str:
        if self._text is None:
            assert self._token_ids is not None
            self._text = _cached_decode(self.tokenizer, tuple(self._token_ids))

        return self._text

    @property
    def token_ids(self) -> list[int]:
        if self._token_ids is None:
            assert self._text is not None
            self._token_ids = _cached_encode(self.tokenizer, self._text)

        return self._token_ids


@dataclass
134
135
class _BoundPromptReplacement:
    tokenizer: AnyTokenizer = field(repr=False)
136
137
    modality: str

138
139
140
    _target: _PromptSeq
    _replacement: Union[Callable[[int], _PromptSeq],
                        _PromptSeq] = field(repr=False)
141

142
143
144
145
146
147
    def __post_init__(self) -> None:
        self._replacement_cache = dict[int, _BoundPromptSequence]()

    @property
    def target(self) -> _BoundPromptSequence:
        target = self._target
148

149
150
151
152
153
        return _BoundPromptSequence(
            tokenizer=self.tokenizer,
            _text=target if isinstance(target, str) else None,
            _token_ids=target if isinstance(target, list) else None,
        )
154

155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
    def get_replacement(self, item_idx: int) -> _BoundPromptSequence:
        replacement = self._replacement
        if callable(replacement):
            cache_key = item_idx
            if cache_key in self._replacement_cache:
                return self._replacement_cache[cache_key]

            replacement = replacement(item_idx)
        else:
            cache_key = None

        bound_replacement = _BoundPromptSequence(
            tokenizer=self.tokenizer,
            _text=replacement if isinstance(replacement, str) else None,
            _token_ids=replacement if isinstance(replacement, list) else None,
        )

        if cache_key is not None:
            self._replacement_cache[cache_key] = bound_replacement

        return bound_replacement


178
179
180
class _TokenMatch(NamedTuple):
    start_idx: int
    end_idx: int
181
182


183
184
185
186
def iter_token_matches(
    token_ids: list[int],
    match_ids: list[int],
) -> Iterable[_TokenMatch]:
187
188
189
190
191
192
    """
    Yield each occurrence of :code:`match_ids` in :code:`token_ids`.

    Note that empty matches are ignored.
    """
    prompt_len = len(token_ids)
193
    match_len = len(match_ids)
194

195
196
    if match_len == 0:
        return
197

198
199
    start_idx = 0
    while start_idx < prompt_len - match_len + 1:
200
        end_idx = start_idx + match_len
201

202
203
        if token_ids[start_idx:end_idx] == match_ids:
            yield _TokenMatch(start_idx=start_idx, end_idx=end_idx)
204
205
206
207
208

            # Exclude overlapping matches
            start_idx = end_idx
        else:
            start_idx += 1
209
210


211
212
213
@dataclass(repr=False)
class _PromptReplacementMatch(ABC):
    prompt_repl: _BoundPromptReplacement
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234

    @property
    def modality(self) -> str:
        return self.prompt_repl.modality

    @property
    @abstractmethod
    def start_idx(self) -> int:
        raise NotImplementedError

    @property
    @abstractmethod
    def end_idx(self) -> int:
        raise NotImplementedError

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


@dataclass(repr=False)
235
class _PromptReplacementTokenMatch(_PromptReplacementMatch):
236
237
238
239
240
241
242
243
244
245
246
247
    match: _TokenMatch

    @property
    def start_idx(self) -> int:
        return self.match.start_idx

    @property
    def end_idx(self) -> int:
        return self.match.end_idx


@dataclass(repr=False)
248
class _PromptReplacementTextMatch(_PromptReplacementMatch):
249
250
251
252
253
254
255
256
257
258
    match: re.Match[str]

    @property
    def start_idx(self) -> int:
        return self.match.start()

    @property
    def end_idx(self) -> int:
        return self.match.end()

259

260
261
@dataclass
class _PlaceholderInfo:
262
    modality: str
263
    item_idx: int
264
    start_idx: int
265
    replacement: list[int]
266
267
268

    @property
    def length(self) -> int:
269
        return len(self.replacement)
270
271
272
273
274
275

    def to_range(self) -> PlaceholderRange:
        return PlaceholderRange(
            offset=self.start_idx,
            length=self.length,
        )
276
277
278
279


def find_token_matches(
    prompt: list[int],
280
281
    prompt_repls: Sequence[_BoundPromptReplacement],
) -> list[_PromptReplacementTokenMatch]:
282
283
284
285
286
287
288
289
290
291
    """Return each target of :code:`prompt_repls` found in :code:`prompt`."""
    return [
        _PromptReplacementTokenMatch(prompt_repl, match)
        for prompt_repl in prompt_repls
        for match in iter_token_matches(prompt, prompt_repl.target.token_ids)
    ]


def find_text_matches(
    prompt: str,
292
293
    prompt_repls: Sequence[_BoundPromptReplacement],
) -> list[_PromptReplacementTextMatch]:
294
295
296
297
298
299
300
301
302
    """Return each target of :code:`prompt_repls` found in :code:`prompt`."""
    return [
        _PromptReplacementTextMatch(prompt_repl, match)
        for prompt_repl in prompt_repls
        for match in re.finditer(re.escape(prompt_repl.target.text), prompt)
    ]


def _resolve_matches(
303
    prompt: _PromptSeq,
304
    mm_matches: Mapping[str, Sequence[_PromptReplacementMatch]],
305
) -> list[_PromptReplacementMatch]:
306
    """
307
    Resolve :code:`mm_matches` to ensure that there are no overlapping matches,
308
    and sort them such that earlier matches take priority over later ones.
309
    """
310
311
    matches = [m for matches in mm_matches.values() for m in matches]

312
313
    seen_matches: list[Optional[_PromptReplacementMatch]] = [None
                                                             ] * len(prompt)
314

315
    for match in matches:
316
317
318
319
320
        for idx in range(match.start_idx, match.end_idx):
            if seen_matches[idx] is not None:
                raise ValueError("Found overlapping matches "
                                 f"({seen_matches[idx]} and {match}) "
                                 f"at index={idx} of prompt={prompt}")
321

322
            seen_matches[idx] = match
323
324
325
326
327
328

    return sorted(matches, key=lambda x: x.start_idx)


def _replace_matches(
    prompt: _S,
329
    mm_matches: Mapping[str, Sequence[_PromptReplacementMatch]],
330
    mm_item_counts: Mapping[str, int],
331
) -> list[_S]:
332
    """Apply the replacements in :code:`mm_matches` to :code:`prompt`."""
333
334
    out_seqs = list[_S]()
    prev_end_idx = 0
335
    next_idx_by_modality = defaultdict[str, int](lambda: 0)
336

337
    for match in _resolve_matches(prompt, mm_matches):
338
339
340
        modality = match.modality

        item_idx = next_idx_by_modality[modality]
341
        if item_idx >= mm_item_counts.get(modality, 0):
342
343
344
345
            continue

        start_idx = match.start_idx
        end_idx = match.end_idx
346

347
        repl_info = match.prompt_repl
348
349
350
351
352
353
354
355
        replacement = repl_info.get_replacement(item_idx)

        if isinstance(prompt, str):
            repl_seq = replacement.text
            out_seqs.append(prompt[prev_end_idx:start_idx] + repl_seq)
        else:
            repl_seq = replacement.token_ids
            out_seqs.append(prompt[prev_end_idx:start_idx] + repl_seq)
356
357
358
359
360
361
362
363
364
365
366

        prev_end_idx = end_idx
        next_idx_by_modality[modality] += 1

    out_seqs.append(prompt[prev_end_idx:])

    return out_seqs


def replace_token_matches(
    prompt: list[int],
367
    mm_matches: Mapping[str, Sequence[_PromptReplacementTokenMatch]],
368
    mm_item_counts: Mapping[str, int],
369
) -> list[int]:
370
371
    """Apply the replacements in :code:`mm_matches` to :code:`prompt`."""
    if not mm_matches:
372
373
        return prompt

374
    token_id_seqs = _replace_matches(prompt, mm_matches, mm_item_counts)
375
376

    return flatten_2d_lists(token_id_seqs)
377
378


379
380
def replace_text_matches(
    prompt: str,
381
    mm_matches: Mapping[str, Sequence[_PromptReplacementTextMatch]],
382
    mm_item_counts: Mapping[str, int],
383
) -> str:
384
385
    """Apply the replacements in :code:`mm_matches` to :code:`prompt`."""
    if not mm_matches:
386
        return prompt
387

388
    texts = _replace_matches(prompt, mm_matches, mm_item_counts)
389
390

    return "".join(texts)
391
392


393
394
395
396
def _iter_modality_placeholders(
    prompt: list[int],
    modality: str,
    modality_repls: Sequence[_BoundPromptReplacement],
397
    modal_item_count: int,
398
) -> Iterable[_PlaceholderInfo]:
399
    if modal_item_count == 0:
400
        return
401

402
    prompt_len = len(prompt)
403
    item_idx = 0
404
405
406
407
408
409

    start_idx = 0
    while start_idx < prompt_len:
        found = False

        for repl_info in modality_repls:
410
            replacement = repl_info.get_replacement(item_idx)
411
412
413
414
415
416
            repl_tokens = replacement.token_ids
            repl_len = len(repl_tokens)
            end_idx = start_idx + repl_len

            if repl_len == 0 or end_idx > prompt_len:
                continue
417

418
419
420
            if prompt[start_idx:end_idx] == repl_tokens:
                yield _PlaceholderInfo(
                    modality=modality,
421
                    item_idx=item_idx,
422
423
424
425
                    start_idx=start_idx,
                    replacement=repl_tokens,
                )

426
427
                item_idx += 1
                if item_idx >= modal_item_count:
428
429
430
431
432
433
434
435
436
                    return

                # Exclude overlapping matches
                start_idx = end_idx
                found = True
                break

        if not found:
            start_idx += 1
437
438


439
440
def _iter_placeholders(
    mm_prompt_repls: Mapping[str, Sequence[_BoundPromptReplacement]],
441
    prompt: list[int],
442
    mm_item_counts: Mapping[str, int],
443
) -> Iterable[_PlaceholderInfo]:
444
    """
445
446
    For each modality, yield each set of placeholder tokens found in
    :code:`prompt`.
447
448
449

    Note that empty matches are ignored.
    """
450
    for modality, modal_item_count in mm_item_counts.items():
451
        if modality in mm_prompt_repls:
452
453
454
            yield from _iter_modality_placeholders(
                prompt,
                modality,
455
                mm_prompt_repls[modality],
456
                modal_item_count,
457
458
            )

459

460
461
462
463
464
465
466
467
468
def find_mm_placeholders(
    mm_prompt_repls: Mapping[str, Sequence[_BoundPromptReplacement]],
    prompt: list[int],
    mm_item_counts: Mapping[str, int],
) -> Mapping[str, list[_PlaceholderInfo]]:
    it = _iter_placeholders(mm_prompt_repls, prompt, mm_item_counts)
    return dict(full_groupby_modality(it))


469
470
471
472
473
474
475
476
class ProcessingCache:

    def __init__(self, capacity: int) -> None:
        super().__init__()

        # DEBUG: Set to None to disable
        self.debug_cache_hit_ratio_steps: Optional[int] = None

477
        self._cache = LRUCache[str, MultiModalKwargsItem](capacity)
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494

    def _maybe_log_cache_stats(self) -> None:
        steps = self.debug_cache_hit_ratio_steps
        if not steps:
            return

        cache_stats = self._cache.stat()
        if cache_stats.total % steps == 0:
            logger.debug("ProcessingCache: hit_ratio = %.2f",
                         cache_stats.hit_ratio)

    def _serialize_item(self, obj: object) -> bytes:
        # Simple cases
        if isinstance(obj, str):
            return obj.encode("utf-8")
        if isinstance(obj, bytes):
            return obj
495
        if isinstance(obj, Image.Image):
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
            return obj.tobytes()

        # Convertible to NumPy arrays
        if isinstance(obj, torch.Tensor):
            obj = obj.numpy()
        if isinstance(obj, (int, float)):
            obj = np.array(obj)
        if isinstance(obj, np.ndarray):
            return obj.tobytes()

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

        return pickle.dumps(obj)

    def _item_to_bytes(
        self,
        key: str,
        obj: object,
    ) -> Iterable[tuple[bytes, bytes]]:
        # Recursive cases
        if isinstance(obj, (list, tuple)):
            for i, elem in enumerate(obj):
                yield from self._item_to_bytes(f"{key}.{i}", elem)
        elif isinstance(obj, dict):
            for k, v in obj.items():
                yield from self._item_to_bytes(f"{key}.{k}", v)
        else:
            key_bytes = self._serialize_item(key)
            value_bytes = self._serialize_item(obj)
            yield key_bytes, value_bytes

    def _hash_kwargs(self, **kwargs: object) -> str:
        hasher = blake3()

        for k, v in kwargs.items():
            for k_bytes, v_bytes in self._item_to_bytes(k, v):
                hasher.update(k_bytes)
                hasher.update(v_bytes)

        return hasher.hexdigest()

    def get(
        self,
        model_id: str,
        modality: str,
        input_item: object,
        input_kwargs: Mapping[str, object],
545
    ) -> Optional[MultiModalKwargsItem]:
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
        """
        Get a processed multi-modal item from the cache
        according to its dependencies, including:

        - The model ID
        - The modality of the item
        - The original data item passed to the HF processor
        - The configuration options of the HF processor
        """
        self._maybe_log_cache_stats()

        cache_key = self._hash_kwargs(model_id=model_id,
                                      **{modality: input_item},
                                      **input_kwargs)
        return self._cache.get(cache_key)

    def put(
        self,
        model_id: str,
        modality: str,
        input_item: object,
        input_kwargs: Mapping[str, object],
568
        output_kwargs: MultiModalKwargsItem,
569
570
571
572
573
574
575
576
577
    ) -> None:
        """
        Put a processed multi-modal item into the cache
        according to its dependencies (see :meth:`get`).
        """
        cache_key = self._hash_kwargs(model_id=model_id,
                                      **{modality: input_item},
                                      **input_kwargs)
        self._cache.put(cache_key, output_kwargs)
578
579


580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
class ProcessingMixin:
    """
    Contains helper functions to perform processing.

    Not to be confused with :class:`transformers.ProcessorMixin`.
    """
    ctx: InputProcessingContext

    def _get_tokenizer(self) -> AnyTokenizer:
        return self.ctx.tokenizer

    def _get_hf_config(self) -> PretrainedConfig:
        return self.ctx.get_hf_config()

    def _get_hf_processor(self, **kwargs: object) -> ProcessorMixin:
        """
        Subclasses can override this method to handle
        specific kwargs from model config or user inputs.
        """
        return self.ctx.get_hf_processor(**kwargs)


class BaseMultiModalProcessor(ProcessingMixin, ABC):
603
    """
604
    Abstract base class to process multi-modal inputs to be used in vLLM.
605
606

    Not to be confused with :class:`transformers.ProcessorMixin`.
607
608
    """

609
610
611
612
613
    def __init__(self,
                 ctx: InputProcessingContext,
                 *,
                 cache: Optional[ProcessingCache] = None,
                 enable_sanity_checks: bool = True) -> None:
614
615
616
        super().__init__()

        self.ctx = ctx
617
618
        self.cache = cache
        self.enable_sanity_checks = enable_sanity_checks
619

620
621
622
        self.data_parser = self._get_data_parser()
        self.profiling_info = self._get_profiling_info()

623
    def __call__(
624
        self,
625
626
        prompt: str,
        mm_data: MultiModalDataDict,
627
        hf_processor_mm_kwargs: Mapping[str, object],
628
    ) -> MultiModalInputsV2:
629
        return self.apply(prompt, mm_data, hf_processor_mm_kwargs)
630

631
632
    def _get_data_parser(self) -> MultiModalDataParser:
        """
633
        Construct a parser to preprocess multi-modal data items
634
635
636
637
638
639
640
        before passing them to :meth:`_get_hf_mm_data`.

        You can support additional modalities by creating a subclass
        of :class:`MultiModalDataParser` that has additional subparsers.
        """
        return MultiModalDataParser()

641
    def _get_profiling_info(self) -> BaseProfilingInfo:
642
        """
643
644
        Get the profiling information to find the worst-case memory usage of
        the model.
645
        """
646
        raise NotImplementedError
647

648
    def _to_mm_items(
649
650
651
        self,
        mm_data: MultiModalDataDict,
    ) -> MultiModalDataItems:
652
653
654
655
        """
        Normalize :class:`MultiModalDataDict` to :class:`MultiModalDataItems`
        before passing them to :meth:`_get_hf_mm_data`.
        """
656
        mm_items = self.data_parser.parse_mm_data(mm_data)
657
658
659
660
661
662
663
664
665
666
667

        mm_limits = self.ctx.get_mm_config().limit_per_prompt
        for modality, items in mm_items.items():
            limit = mm_limits.get(modality, 1)
            if len(items) > limit:
                raise ValueError(
                    f"You set {modality}={limit} (or defaulted to 1) in "
                    f"`--limit-mm-per-prompt`, but passed {len(items)} "
                    f"{modality} items in the same prompt.")

        return mm_items
668

669
670
671
672
673
674
675
676
677
    @abstractmethod
    def _get_mm_fields_config(
        self,
        hf_inputs: BatchFeature,
        hf_processor_mm_kwargs: Mapping[str, object],
    ) -> Mapping[str, MultiModalFieldConfig]:
        """Given the HF-processed data, output the metadata of each field."""
        raise NotImplementedError

678
679
    @abstractmethod
    def _get_prompt_replacements(
680
        self,
681
        mm_items: MultiModalDataItems,
682
683
        hf_processor_mm_kwargs: Mapping[str, object],
        out_mm_kwargs: MultiModalKwargs,
684
685
686
687
688
    ) -> list[PromptReplacement]:
        """
        Given the original multi-modal items for this modality
        and HF-processed data, output the replacements to perform.

689
690
691
692
693
694
695
696
        Notes:
            - You should not assume that HF processor always performs prompt
              replacement: in :meth:`_apply_hf_processor_missing`, this method
              is called on text-only and multimodal-only inputs separately,
              instead of passing them in the same call.
            - The replacement information returned by this method is also used
              to determine the placeholder token positions for each multi-modal
              item.
697
698
        """
        raise NotImplementedError
699

700
    def _find_mm_placeholders(
701
        self,
702
        mm_prompt_repls: Mapping[str, Sequence[_BoundPromptReplacement]],
703
        new_token_ids: list[int],
704
        mm_item_counts: Mapping[str, int],
705
706
707
    ) -> Mapping[str, list[_PlaceholderInfo]]:
        return find_mm_placeholders(mm_prompt_repls, new_token_ids,
                                    mm_item_counts)
708

709
    def _get_hf_mm_data(
710
        self,
711
712
        mm_items: MultiModalDataItems,
    ) -> tuple[dict[str, Any], dict[str, Any]]:
713
714
        processor_data = dict[str, Any]()
        passthrough_data = dict[str, Any]()
715

716
717
718
        for items in mm_items.values():
            processor_data.update(items.get_processor_data())
            passthrough_data.update(items.get_passthrough_data())
719

720
721
        return processor_data, passthrough_data

722
723
724
    def _call_hf_processor(
        self,
        prompt: str,
725
726
727
728
        # Not to be confused with `mm_data` in `self.apply`.
        # This refers to the data to be passed to HF processor.
        mm_data: Mapping[str, object],
        mm_kwargs: Mapping[str, object],
729
    ) -> BatchFeature:
730
731
732
733
        """
        Call the HF processor on the prompt text and
        associated multi-modal data.
        """
734
        return self.ctx.call_hf_processor(
735
736
737
            self._get_hf_processor(**mm_kwargs),
            dict(text=prompt, **mm_data),
            mm_kwargs,
738
739
        )

740
741
    def _apply_hf_processor(
        self,
742
        prompt_text: str,
743
        mm_items: MultiModalDataItems,
744
745
746
        hf_processor_mm_kwargs: Mapping[str, object],
    ) -> tuple[list[int], MultiModalKwargs]:
        """
747
748
        Wrapper of :meth:`_call_hf_processor` that applies
        additional pre-processing and post-processing.
749
750
751
752
753
754
755
756
757
        """
        processor_data, passthrough_data = self._get_hf_mm_data(mm_items)

        processed_data = self._call_hf_processor(
            prompt=prompt_text,
            mm_data=processor_data,
            mm_kwargs=hf_processor_mm_kwargs,
        )
        processed_data.update(passthrough_data)
758

759
        prompt_ids, = processed_data.pop("input_ids").tolist()
760

761
762
763
        mm_kwargs = MultiModalKwargs.from_hf_inputs(
            processed_data,
            self._get_mm_fields_config(processed_data, hf_processor_mm_kwargs),
764
        )
765

766
767
768
769
770
771
772
773
774
775
776
777
        return prompt_ids, mm_kwargs

    def _apply_hf_processor_missing(
        self,
        prompt_text: str,
        mm_missing_data_items: MultiModalDataItems,
        hf_processor_mm_kwargs: Mapping[str, object],
    ):
        """
        Apply the HF processor on the full prompt text, but only on the
        multi-modal data that are missing from the cache.

778
779
780
781
782
        Note:
            We pass prompt text and multi-modal data into the HF processor
            in separate calls to avoid HF prompt replacement being done for
            cached items; instead, we rely on our own prompt replacement logic
            (:meth:`_get_prompt_replacements`) for the full text.
783
        """
784
        mm_missing_counts = mm_missing_data_items.get_all_counts()
785
786
787
788
789
790
791
792
793

        prompt_ids, _ = self._apply_hf_processor(
            prompt_text=prompt_text,
            mm_items=MultiModalDataItems({}),
            hf_processor_mm_kwargs={},
        )

        # Some HF processors (e.g. Qwen2-VL) expect corresponding
        # multi-modal tokens to be in the prompt text
794
        dummy_inputs = self.profiling_info.get_dummy_processor_inputs(
795
796
797
            self.ctx.model_config.max_model_len,
            mm_missing_counts,
        )
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819

        _, mm_missing_kwargs = self._apply_hf_processor(
            prompt_text=dummy_inputs.prompt_text,
            mm_items=mm_missing_data_items,
            hf_processor_mm_kwargs=hf_processor_mm_kwargs,
        )

        return prompt_ids, mm_missing_kwargs

    def _cached_apply_hf_processor(
        self,
        prompt_text: str,
        mm_data_items: MultiModalDataItems,
        hf_processor_mm_kwargs: Mapping[str, object],
    ) -> tuple[list[int], MultiModalKwargs]:
        """
        Apply the HF processor on the full prompt text,
        caching the results and reusing cached results.
        """
        cache = self.cache
        model_id = self.ctx.model_config.model

820
821
        _, passthrough_data = self._get_hf_mm_data(mm_data_items)
        if cache is None or passthrough_data:
822
823
824
825
826
827
            return self._apply_hf_processor(
                prompt_text=prompt_text,
                mm_items=mm_data_items,
                hf_processor_mm_kwargs=hf_processor_mm_kwargs,
            )

828
        mm_maybe_cached_kw_items = {
829
830
831
832
833
834
835
836
            modality: [
                cache.get(model_id, modality, item, hf_processor_mm_kwargs)
                for item in items
            ]
            for modality, items in mm_data_items.items()
        }

        mm_missing_idxs = {
837
838
839
            modality:
            [idx for idx, item in enumerate(kw_items) if item is None]
            for modality, kw_items in mm_maybe_cached_kw_items.items()
840
841
842
843
844
        }
        mm_missing_data = {
            modality: [mm_data_items[modality][idx] for idx in idxs]
            for modality, idxs in mm_missing_idxs.items()
        }
845
        mm_missing_data_items = self._to_mm_items(mm_missing_data)
846
847
848
849
850
851
852
853
854
855
856
857

        prompt_ids, mm_missing_kwargs = self._apply_hf_processor_missing(
            prompt_text=prompt_text,
            mm_missing_data_items=mm_missing_data_items,
            hf_processor_mm_kwargs=hf_processor_mm_kwargs,
        )

        mm_missing_next_idx = {
            modality: 0
            for modality in mm_missing_data_items
        }

858
859
860
861
862
        merged_kw_items = list[MultiModalKwargsItem]()
        for modality, kw_items in mm_maybe_cached_kw_items.items():
            for idx, kw_item in enumerate(kw_items):
                if kw_item is None:
                    kw_item = mm_missing_kwargs.get_item(
863
864
865
866
867
868
869
870
871
                        modality,
                        mm_missing_next_idx[modality],
                    )

                    cache.put(
                        model_id,
                        modality,
                        mm_data_items[modality][idx],
                        hf_processor_mm_kwargs,
872
                        kw_item,
873
874
875
876
                    )

                    mm_missing_next_idx[modality] += 1

877
                merged_kw_items.append(kw_item)
878
879

        if self.enable_sanity_checks:
880
            mm_missing_counts = mm_missing_data_items.get_all_counts()
881
882
883
884
885
886
            assert all(
                item_count == mm_missing_counts[modality]
                for modality, item_count in mm_missing_next_idx.items()), dict(
                    mm_missing_next_idx=mm_missing_next_idx,
                    mm_missing_counts=mm_missing_counts)

887
        mm_kwargs = MultiModalKwargs.from_items(merged_kw_items)
888
889

        return prompt_ids, mm_kwargs
890

891
    def _bind_and_group_repls(
892
        self,
893
        prompt_repls: list[PromptReplacement],
894
    ) -> dict[str, list[_BoundPromptReplacement]]:
895
        tokenizer = self._get_tokenizer()
896

897
898
        it = (prompt_repl.bind(tokenizer) for prompt_repl in prompt_repls)
        return dict(full_groupby_modality(it))
899

900
901
902
903
    def _always_apply_prompt_replacements(self) -> bool:
        """
        A flag which can be overridden so that
        :meth:`_apply_prompt_replacements` is always called even if we
904
905
        detect that HF has performed processing via
        :meth:`_find_placeholders_by_modality`.
906

907
908
        This is useful in cases where :meth:`_find_placeholders_by_modality`
        cannot be reliably used to detect whether HF has performed processing.
909
910
911
        """
        return False

912
913
914
    def _apply_prompt_replacements(
        self,
        token_ids: list[int],
915
        mm_prompt_repls: Mapping[str, Sequence[_BoundPromptReplacement]],
916
        mm_item_counts: Mapping[str, int],
917
    ) -> tuple[list[int], str, Mapping[str, list[_PlaceholderInfo]]]:
918
        tokenizer = self._get_tokenizer()
919

920
921
922
923
        mm_token_matches = {
            modality: find_token_matches(token_ids, prompt_repls)
            for modality, prompt_repls in mm_prompt_repls.items()
        }
924
925
        mm_match_counts = {
            modality: len(matches)
926
            for modality, matches in mm_token_matches.items()
927
        }
928
929
930
931
932
933
934
935
936
937
938
939

        # If the search text does not represent a special token,
        # it may have different token IDs in the prompt, because
        # the tokens may go across the boundaries of the search text.
        # ----
        # e.g. when searching for "foo" in "food", if "food" itself makes
        # up a token, then the token ID of "foo" will not appear at all
        # ----
        # Since it is inefficient to search for all possible tokenizations
        # of the search text in the prompt, we instead perform string
        # replacement on the decoded token IDs, then encode them back.
        if all(
940
941
            mm_match_counts.get(modality, 0) >= item_count
            for modality, item_count in mm_item_counts.items()
942
943
944
        ):  # yapf: disable
            token_ids = replace_token_matches(
                token_ids,
945
                mm_token_matches,
946
                mm_item_counts,
947
948
            )

949
950
951
952
953
            text = decode_tokens(tokenizer, token_ids)
            matched_repls = {
                modality: [match.prompt_repl for match in token_matches]
                for modality, token_matches in mm_token_matches.items()
            }
954
        else:
955
            text = decode_tokens(tokenizer, token_ids)
956

957
958
959
960
            mm_text_matches = {
                modality: find_text_matches(text, prompt_repls)
                for modality, prompt_repls in mm_prompt_repls.items()
            }
961
962
            text = replace_text_matches(
                text,
963
                mm_text_matches,
964
                mm_item_counts,
965
966
            )

967
968
969
            token_ids = encode_tokens(tokenizer,
                                      text,
                                      add_special_tokens=False)
970
971
972
973
974
975
976
977
978
979
            matched_repls = {
                modality: [match.prompt_repl for match in token_matches]
                for modality, token_matches in mm_text_matches.items()
            }

        placeholders = self._find_mm_placeholders(
            matched_repls,
            token_ids,
            mm_item_counts,
        )
980
981

        return token_ids, text, placeholders
982

983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
    def _validate_mm_kwargs(
        self,
        mm_kwargs: MultiModalKwargs,
        mm_item_counts: Mapping[str, int],
    ) -> None:
        for modality, item_count in mm_item_counts.items():
            if modality in mm_kwargs.modalities:
                items = mm_kwargs.get_items(modality)
            else:
                items = []

            if len(items) != item_count:
                raise RuntimeError(
                    f"Expected there to be {item_count} {modality} items in "
                    f"keyword arguments corresponding to {item_count} "
                    f"{modality} data items, but only found {len(items)}! "
                    "There is likely a problem with your "
                    "implementation of merged multi-modal processor for this "
                    "model (usually arising from an inconsistency between "
                    "`_call_hf_processor` and `_get_mm_fields_config`).")

    def _validate_mm_placeholders(
        self,
        mm_placeholders: Mapping[str, list[_PlaceholderInfo]],
        mm_item_counts: Mapping[str, int],
        *,
        allow_missing: bool = False,
    ) -> Mapping[str, int]:
        missing_repl_counts = dict[str, int]()

        for modality, item_count in mm_item_counts.items():
            placeholders = mm_placeholders.get(modality, [])

            if len(placeholders) != item_count and not allow_missing:
                raise RuntimeError(
                    f"Expected there to be {item_count} prompt replacements "
                    f"corresponding to {item_count} {modality} items, but only "
                    f"found {len(placeholders)} prompt replacements! Either "
                    "the prompt text has missing/incorrect tokens for "
                    "multi-modal inputs, or there is a problem with your "
                    "implementation of merged multi-modal processor for this "
                    "model (usually arising from an inconsistency between "
                    "`_call_hf_processor` and `_get_prompt_replacements`).")

            missing_repl_counts[modality] = item_count - len(placeholders)

        return missing_repl_counts

1031
1032
1033
1034
    def apply(
        self,
        prompt_text: str,
        mm_data: MultiModalDataDict,
1035
        hf_processor_mm_kwargs: Mapping[str, object],
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
    ) -> MultiModalInputsV2:
        """
        Process multi-modal inputs to be used in vLLM.

        The main steps are:

        1. Apply HF Processor on prompt text and multi-modal data together,
           outputting token IDs and processed tensors.
        2. Find and replace sequences in the token IDs with placeholder tokens.
           The number of placeholder tokens equals the feature size of the
           multi-modal data outputted by the multi-modal encoder.
        3. Extract information about the placeholder tokens from the
           processed token IDs.
        """
1050
        mm_items = self._to_mm_items(mm_data)
1051

1052
1053
1054
1055
1056
        prompt_ids, mm_kwargs = self._cached_apply_hf_processor(
            prompt_text,
            mm_items,
            hf_processor_mm_kwargs,
        )
1057

1058
1059
1060
1061
1062
        unbound_prompt_repls = self._get_prompt_replacements(
            mm_items,
            hf_processor_mm_kwargs,
            mm_kwargs,
        )
1063
        mm_prompt_repls = self._bind_and_group_repls(unbound_prompt_repls)
1064

1065
        mm_item_counts = mm_items.get_all_counts()
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
        self._validate_mm_kwargs(mm_kwargs, mm_item_counts)

        hf_mm_placeholders = self._find_mm_placeholders(
            mm_prompt_repls,
            prompt_ids,
            mm_item_counts,
        )

        if self._always_apply_prompt_replacements():
            mm_missing_repl_counts = mm_item_counts
            mm_missing_repls = dict(mm_prompt_repls)
        else:
            mm_missing_repl_counts = self._validate_mm_placeholders(
                hf_mm_placeholders,
                mm_item_counts,
                allow_missing=True,
            )

            mm_missing_repls = dict[str, list[_BoundPromptReplacement]]()
            for modality, missing_repl_count in mm_missing_repl_counts.items():
                if missing_repl_count == 0:
                    mm_missing_repls[modality] = []
                elif missing_repl_count == mm_item_counts.get(modality, 0):
                    mm_missing_repls[modality] = mm_prompt_repls[modality]
                else:
                    raise ValueError("Partial prompt replacement within "
                                     f"{modality=} is not supported")
1093

1094
1095
1096
        # If HF processor already inserts placeholder tokens,
        # there is no need for us to insert them
        if all(len(repls) == 0 for repls in mm_missing_repls.items()):
1097
            tokenizer = self._get_tokenizer()
1098
1099
            prompt_text = decode_tokens(tokenizer, prompt_ids)
            mm_placeholders = hf_mm_placeholders
1100
1101
1102
1103
        else:
            (
                prompt_ids,
                prompt_text,
1104
                missing_mm_placeholders,
1105
1106
            ) = self._apply_prompt_replacements(
                prompt_ids,
1107
1108
                mm_missing_repls,
                mm_missing_repl_counts,
1109
1110
            )

1111
1112
1113
1114
1115
1116
1117
1118
            mm_placeholders = {**hf_mm_placeholders, **missing_mm_placeholders}

        self._validate_mm_placeholders(mm_placeholders, mm_item_counts)

        mm_placeholder_ranges = {
            modality: [item.to_range() for item in placeholders]
            for modality, placeholders in mm_placeholders.items()
        }
1119
1120
1121

        return MultiModalInputsV2(
            type="multimodal",
1122
1123
            prompt=prompt_text,
            prompt_token_ids=prompt_ids,
1124
            mm_kwargs=mm_kwargs,
1125
            mm_placeholders=mm_placeholder_ranges,
1126
        )
1127

1128
1129
1130
1131
1132
    def _get_dummy_mm_inputs(
        self,
        seq_len: int,
        mm_counts: Mapping[str, int],
    ) -> MultiModalInputsV2:
1133
1134
1135
        profiling = self.profiling_info
        processor_inputs = profiling.get_dummy_processor_inputs(
            seq_len, mm_counts)
1136
1137
1138
1139
1140
1141
1142

        return self.apply(
            prompt_text=processor_inputs.prompt_text,
            mm_data=processor_inputs.mm_data,
            hf_processor_mm_kwargs=processor_inputs.hf_processor_mm_kwargs,
        )

1143
    def get_dummy_data(self, seq_len: int) -> DummyData:
1144
1145
1146
        # Avoid circular import
        from vllm.sequence import SequenceData

1147
1148
1149
        profiling = self.profiling_info
        mm_counts = profiling.get_mm_limits()
        mm_max_tokens_per_item = profiling.get_mm_max_tokens_per_item(seq_len)
1150
1151
1152
1153
1154
1155
1156
        if mm_counts.keys() != mm_max_tokens_per_item.keys():
            raise AssertionError(
                "The keys returned by `get_supported_mm_limits`"
                f"({set(mm_counts.keys())}) should be the same as those "
                "returned by `get_mm_max_tokens_per_item` "
                f"({set(mm_max_tokens_per_item.keys())})")

1157
        mm_inputs = self._get_dummy_mm_inputs(seq_len, mm_counts)
1158
1159
1160
        prompt_token_ids = mm_inputs["prompt_token_ids"]
        placeholders_by_modality = mm_inputs["mm_placeholders"]

1161
1162
1163
1164
1165
        total_placeholders_by_modality = {
            modality: sum(item["length"] for item in placeholders)
            for modality, placeholders in placeholders_by_modality.items()
        }
        expected_placeholders_by_modality = {
1166
            modality: mm_max_tokens_per_item[modality] * mm_counts[modality]
1167
1168
1169
1170
1171
1172
1173
1174
            for modality in placeholders_by_modality
        }
        if total_placeholders_by_modality != expected_placeholders_by_modality:
            raise AssertionError(
                f"The processed dummy data has a total of "
                f"{total_placeholders_by_modality} placeholder tokens, which "
                f"is not the expected {expected_placeholders_by_modality} "
                "tokens.")
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186

        total_len = len(prompt_token_ids)
        if total_len > seq_len:
            logger.warning(
                "The context length (%d) of the model is too short "
                "to hold the multi-modal embeddings in the worst case "
                "(%d tokens in total, out of which %s are reserved for "
                "multi-modal embeddings). This may cause certain multi-modal "
                "inputs to fail during inference, even when the input text is "
                "short. To avoid this, you should increase `max_model_len`, "
                "reduce `max_num_seqs`, and/or reduce `mm_counts`.", seq_len,
                total_len, total_placeholders_by_modality)
1187

1188
1189
1190
1191
1192
1193
            return DummyData(
                seq_data=SequenceData.from_prompt_token_counts((0, seq_len)),
                multi_modal_data=None,
                multi_modal_placeholders=None,
            )

1194
1195
1196
1197
        prompt_token_ids.extend([0] * (seq_len - len(prompt_token_ids)))

        return DummyData(
            seq_data=SequenceData.from_seqs(prompt_token_ids),
1198
1199
            multi_modal_data=mm_inputs["mm_kwargs"],
            multi_modal_placeholders=placeholders_by_modality,
1200
        )