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

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

15
from vllm.inputs import DummyData, InputProcessingContext
16
from vllm.logger import init_logger
17
from vllm.transformers_utils.tokenizer import AnyTokenizer, MistralTokenizer
18
from vllm.utils import LRUCache, flatten_2d_lists, full_groupby
19

20
21
22
23
from .inputs import (MultiModalDataDict, MultiModalFieldConfig,
                     MultiModalFieldItem, MultiModalInputsV2, MultiModalKwargs,
                     PlaceholderRange)
from .parse import MultiModalDataItems, MultiModalDataParser
24

25
logger = init_logger(__name__)
26
27

_S = TypeVar("_S", str, list[int])
28
_PromptSeq = Union[str, list[int]]
29

30
31

@dataclass
32
33
class PromptReplacement:
    modality: str
34
    """The modality for which the replacement is made."""
35

36
37
    target: _PromptSeq
    """The text or token sequence to find and replace."""
38

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

45
46
    For convenience, you can pass in the replacement instead of a function
    if it does not depend on the input.
47
48
    """

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


58
59
60
61
62
63
64
65
66
67
68
69
70
71
def _encode(
    tokenizer: AnyTokenizer,
    text: str,
    *,
    add_special_tokens: bool = False,
) -> list[int]:
    """
    Backend-agnostic equivalent of HF's
    :code:`tokenizer.encode(text, add_special_tokens=...)`.
    """
    if isinstance(tokenizer, MistralTokenizer):
        return tokenizer.tokenizer.encode(text,
                                          bos=add_special_tokens,
                                          eos=add_special_tokens)
72

73
    return tokenizer.encode(text, add_special_tokens=add_special_tokens)
74
75


76
77
78
79
80
81
82
83
@lru_cache(maxsize=2048)
def _cached_encode(
    tokenizer: AnyTokenizer,
    text: str,
    *,
    add_special_tokens: bool = False,
) -> list[int]:
    return _encode(tokenizer, text, add_special_tokens=add_special_tokens)
84
85


86
87
88
89
90
91
92
93
94
95
96
def _decode(
    tokenizer: AnyTokenizer,
    token_ids: list[int],
    *,
    skip_special_tokens: bool = False,
) -> str:
    """
    Backend-agnostic equivalent of HF's
    :code:`tokenizer.decode(token_ids, skip_special_tokens=...)`.
    """
    return tokenizer.decode(token_ids, skip_special_tokens=skip_special_tokens)
97
98


99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
@lru_cache(maxsize=2048)
def _cached_decode(
    tokenizer: AnyTokenizer,
    token_ids: tuple[int, ...],
    *,
    skip_special_tokens: bool = False,
) -> str:
    return _decode(tokenizer,
                   list(token_ids),
                   skip_special_tokens=skip_special_tokens)


class _HasModalityAttr(Protocol):
    modality: str

114

115
class _HasModalityProp(Protocol):
116

117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
    @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:
132
133
    tokenizer: AnyTokenizer = field(repr=False)

134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
    _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
160
161
class _BoundPromptReplacement:
    tokenizer: AnyTokenizer = field(repr=False)
162
163
    modality: str

164
165
166
    _target: _PromptSeq
    _replacement: Union[Callable[[int], _PromptSeq],
                        _PromptSeq] = field(repr=False)
167

168
169
170
171
172
173
    def __post_init__(self) -> None:
        self._replacement_cache = dict[int, _BoundPromptSequence]()

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

175
176
177
178
179
        return _BoundPromptSequence(
            tokenizer=self.tokenizer,
            _text=target if isinstance(target, str) else None,
            _token_ids=target if isinstance(target, list) else None,
        )
180

181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
    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


204
205
206
class _TokenMatch(NamedTuple):
    start_idx: int
    end_idx: int
207
208


209
210
211
212
def iter_token_matches(
    token_ids: list[int],
    match_ids: list[int],
) -> Iterable[_TokenMatch]:
213
214
215
216
217
218
    """
    Yield each occurrence of :code:`match_ids` in :code:`token_ids`.

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

221
222
    if match_len == 0:
        return
223

224
225
    start_idx = 0
    while start_idx < prompt_len - match_len + 1:
226
        end_idx = start_idx + match_len
227

228
229
        if token_ids[start_idx:end_idx] == match_ids:
            yield _TokenMatch(start_idx=start_idx, end_idx=end_idx)
230
231
232
233
234

            # Exclude overlapping matches
            start_idx = end_idx
        else:
            start_idx += 1
235
236


237
238
239
@dataclass(repr=False)
class _PromptReplacementMatch(ABC):
    prompt_repl: _BoundPromptReplacement
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260

    @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)
261
class _PromptReplacementTokenMatch(_PromptReplacementMatch):
262
263
264
265
266
267
268
269
270
271
272
273
    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)
274
class _PromptReplacementTextMatch(_PromptReplacementMatch):
275
276
277
278
279
280
281
282
283
284
    match: re.Match[str]

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

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

285
286
287
288

class _PlaceholderInfo(NamedTuple):
    modality: str
    start_idx: int
289
    replacement: list[int]
290
291
292

    @property
    def length(self) -> int:
293
        return len(self.replacement)
294
295
296
297
298
299

    def to_range(self) -> PlaceholderRange:
        return PlaceholderRange(
            offset=self.start_idx,
            length=self.length,
        )
300
301
302
303


def find_token_matches(
    prompt: list[int],
304
305
    prompt_repls: Sequence[_BoundPromptReplacement],
) -> list[_PromptReplacementTokenMatch]:
306
307
308
309
310
311
312
313
314
315
    """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,
316
317
    prompt_repls: Sequence[_BoundPromptReplacement],
) -> list[_PromptReplacementTextMatch]:
318
319
320
321
322
323
324
325
326
    """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(
327
328
329
    prompt: _PromptSeq,
    matches: Sequence[_PromptReplacementMatch],
) -> list[_PromptReplacementMatch]:
330
331
332
    """
    Resolve :code:`matches` to ensure that there are no overlapping matches,
    and sort them such that earlier matches take priority over later ones.
333
    """
334
335
    seen_matches: list[Optional[_PromptReplacementMatch]] = [None
                                                             ] * len(prompt)
336

337
    for match in matches:
338
339
340
341
342
        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}")
343

344
            seen_matches[idx] = match
345
346
347
348
349
350

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


def _replace_matches(
    prompt: _S,
351
    matches: Sequence[_PromptReplacementMatch],
352
    mm_item_counts: Mapping[str, int],
353
354
355
) -> list[_S]:
    out_seqs = list[_S]()
    prev_end_idx = 0
356
    next_idx_by_modality = {modality: 0 for modality in mm_item_counts}
357
358
359
360
361

    for match in _resolve_matches(prompt, matches):
        modality = match.modality

        item_idx = next_idx_by_modality[modality]
362
        if item_idx >= mm_item_counts[modality]:
363
364
365
366
            continue

        start_idx = match.start_idx
        end_idx = match.end_idx
367

368
        repl_info = match.prompt_repl
369
370
371
372
373
374
375
376
        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)
377
378
379
380
381
382
383
384
385
386
387

        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],
388
    matches: Sequence[_PromptReplacementTokenMatch],
389
    mm_item_counts: Mapping[str, int],
390
391
392
393
394
) -> list[int]:
    """Apply :code:`prompt_repls` to :code:`prompt`."""
    if not matches:
        return prompt

395
    token_id_seqs = _replace_matches(prompt, matches, mm_item_counts)
396
397

    return flatten_2d_lists(token_id_seqs)
398
399


400
401
def replace_text_matches(
    prompt: str,
402
    matches: Sequence[_PromptReplacementTextMatch],
403
    mm_item_counts: Mapping[str, int],
404
405
406
407
) -> str:
    """Apply :code:`prompt_repls` to :code:`prompt`."""
    if not matches:
        return prompt
408

409
    texts = _replace_matches(prompt, matches, mm_item_counts)
410
411

    return "".join(texts)
412
413


414
415
416
417
def _iter_modality_placeholders(
    prompt: list[int],
    modality: str,
    modality_repls: Sequence[_BoundPromptReplacement],
418
    modal_item_count: int,
419
) -> Iterable[_PlaceholderInfo]:
420
    if modal_item_count == 0:
421
        return
422

423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
    prompt_len = len(prompt)
    item_index = 0

    start_idx = 0
    while start_idx < prompt_len:
        found = False

        for repl_info in modality_repls:
            replacement = repl_info.get_replacement(item_index)
            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
438

439
440
441
442
443
444
445
446
            if prompt[start_idx:end_idx] == repl_tokens:
                yield _PlaceholderInfo(
                    modality=modality,
                    start_idx=start_idx,
                    replacement=repl_tokens,
                )

                item_index += 1
447
                if item_index >= modal_item_count:
448
449
450
451
452
453
454
455
456
                    return

                # Exclude overlapping matches
                start_idx = end_idx
                found = True
                break

        if not found:
            start_idx += 1
457
458
459


def iter_placeholders(
460
    prompt_repls: Sequence[_BoundPromptReplacement],
461
    prompt: list[int],
462
    mm_item_counts: Mapping[str, int],
463
) -> Iterable[_PlaceholderInfo]:
464
465
466
467
468
469
470
    """
    Yield each set of placeholder tokens found in :code:`prompt`.

    Note that empty matches are ignored.
    """
    repls_by_modality = dict(full_groupby_modality(prompt_repls))

471
    for modality, modal_item_count in mm_item_counts.items():
472
473
474
475
476
        if modality in repls_by_modality:
            yield from _iter_modality_placeholders(
                prompt,
                modality,
                repls_by_modality[modality],
477
                modal_item_count,
478
479
            )

480

481
482
483
@dataclass
class ProcessorInputs:
    """Keyword arguments to :meth:`BaseMultiModalProcessor`."""
484
485
    prompt_text: str
    mm_data: MultiModalDataDict
486
487
488
489
490
491
492
493
494
495
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
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
    hf_processor_mm_kwargs: Mapping[str, object] = field(default_factory=dict)


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

        self._cache = LRUCache[str, Mapping[str,
                                            MultiModalFieldItem]](capacity)

    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
        if isinstance(obj, Image):
            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],
    ) -> Optional[Mapping[str, MultiModalFieldItem]]:
        """
        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],
        output_kwargs: Mapping[str, MultiModalFieldItem],
    ) -> 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)
599
600


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

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

        self.ctx = ctx
614
615
        self.cache = cache
        self.enable_sanity_checks = enable_sanity_checks
616

617
    def __call__(
618
        self,
619
620
        prompt: str,
        mm_data: MultiModalDataDict,
621
        hf_processor_mm_kwargs: Mapping[str, object],
622
    ) -> MultiModalInputsV2:
623
        return self.apply(prompt, mm_data, hf_processor_mm_kwargs)
624

625
626
627
628
629
630
631
632
633
634
    def _get_data_parser(self) -> MultiModalDataParser:
        """
        Construct a data parser to preprocess multi-modal data items
        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()

635
636
637
638
639
    def _get_hf_processor(self) -> ProcessorMixin:
        """
        Subclasses can add keyword arguments to this method to accept
        additional kwargs from model config or user inputs.
        """
640
641
642
643
644
        return self.ctx.get_hf_processor()

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

645
    def _to_mm_items(
646
647
648
        self,
        mm_data: MultiModalDataDict,
    ) -> MultiModalDataItems:
649
650
651
652
653
654
        """
        Normalize :class:`MultiModalDataDict` to :class:`MultiModalDataItems`
        before passing them to :meth:`_get_hf_mm_data`.
        """
        parser = self._get_data_parser()
        return parser.parse_mm_data(mm_data)
655

656
657
658
659
660
661
662
663
664
    @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

665
666
    @abstractmethod
    def _get_prompt_replacements(
667
        self,
668
        mm_items: MultiModalDataItems,
669
670
        hf_processor_mm_kwargs: Mapping[str, object],
        out_mm_kwargs: MultiModalKwargs,
671
672
673
674
675
676
677
678
679
680
681
    ) -> list[PromptReplacement]:
        """
        Given the original multi-modal items for this modality
        and HF-processed data, output the replacements to perform.

        Note:
            Even when the HF processor already performs replacement for us,
            we still use this replacement information to determine
            the placeholder token positions for each multi-modal item.
        """
        raise NotImplementedError
682

683
684
    def _find_placeholders(
        self,
685
        all_prompt_repls: Sequence[_BoundPromptReplacement],
686
        new_token_ids: list[int],
687
        mm_item_counts: Mapping[str, int],
688
689
    ) -> list[_PlaceholderInfo]:
        return list(
690
            iter_placeholders(all_prompt_repls, new_token_ids, mm_item_counts))
691

692
    def _get_hf_mm_data(
693
        self,
694
695
        mm_items: MultiModalDataItems,
    ) -> tuple[dict[str, Any], dict[str, Any]]:
696
697
        processor_data = dict[str, Any]()
        passthrough_data = dict[str, Any]()
698

699
700
701
        for items in mm_items.values():
            processor_data.update(items.get_processor_data())
            passthrough_data.update(items.get_passthrough_data())
702

703
704
        return processor_data, passthrough_data

705
706
707
    def _call_hf_processor(
        self,
        prompt: str,
708
709
710
711
        # 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],
712
713
    ) -> BatchFeature:
        return self.ctx.call_hf_processor(
714
715
716
            self._get_hf_processor(**mm_kwargs),
            dict(text=prompt, **mm_data),
            mm_kwargs,
717
718
        )

719
720
    def _apply_hf_processor(
        self,
721
        prompt_text: str,
722
        mm_items: MultiModalDataItems,
723
724
725
726
727
728
729
730
731
732
733
734
735
        hf_processor_mm_kwargs: Mapping[str, object],
    ) -> tuple[list[int], MultiModalKwargs]:
        """
        Apply the HF processor on the full prompt text and multi-modal data.
        """
        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)
736

737
        prompt_ids, = processed_data.pop("input_ids").tolist()
738

739
740
741
742
        mm_kwargs = MultiModalKwargs.from_hf_inputs(
            processed_data,
            self._get_mm_fields_config(processed_data, hf_processor_mm_kwargs),
            enable_sanity_checks=self.enable_sanity_checks,
743
        )
744

745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
        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.

        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
        for the full text.
        """
762
        mm_missing_counts = mm_missing_data_items.get_all_counts()
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794

        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
        dummy_inputs = self._get_dummy_mm_inputs(mm_missing_counts)

        _, 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

795
796
        _, passthrough_data = self._get_hf_mm_data(mm_data_items)
        if cache is None or passthrough_data:
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
            return self._apply_hf_processor(
                prompt_text=prompt_text,
                mm_items=mm_data_items,
                hf_processor_mm_kwargs=hf_processor_mm_kwargs,
            )

        mm_maybe_cached_field_items = {
            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 = {
            modality: [idx for idx, out in enumerate(fields) if out is None]
            for modality, fields in mm_maybe_cached_field_items.items()
        }
        mm_missing_data = {
            modality: [mm_data_items[modality][idx] for idx in idxs]
            for modality, idxs in mm_missing_idxs.items()
        }
819
        mm_missing_data_items = self._to_mm_items(mm_missing_data)
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858

        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
        }

        mm_merged_field_items = dict[str, list[Mapping[str,
                                                       MultiModalFieldItem]]]()
        for modality, modal_items_lst in mm_maybe_cached_field_items.items():
            merged_modal_items_lst = list[Mapping[str, MultiModalFieldItem]]()

            for idx, modal_items in enumerate(modal_items_lst):
                if modal_items is None:
                    modal_items = mm_missing_kwargs.get_items_by_modality(
                        modality,
                        mm_missing_next_idx[modality],
                    )

                    cache.put(
                        model_id,
                        modality,
                        mm_data_items[modality][idx],
                        hf_processor_mm_kwargs,
                        modal_items,
                    )

                    mm_missing_next_idx[modality] += 1

                merged_modal_items_lst.append(modal_items)

            mm_merged_field_items[modality] = merged_modal_items_lst

        if self.enable_sanity_checks:
859
            mm_missing_counts = mm_missing_data_items.get_all_counts()
860
861
862
863
864
865
866
867
868
869
870
871
            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)

        mm_kwargs = MultiModalKwargs.from_items_by_modality(
            mm_merged_field_items,
            enable_sanity_checks=self.enable_sanity_checks,
        )

        if self.enable_sanity_checks:
872
            mm_item_counts = mm_data_items.get_all_counts()
873
874
875
876
877
878
879
880
881
882

            for modality, item_count in mm_item_counts.items():
                for item_idx in range(item_count):
                    try:
                        mm_kwargs.get_items_by_modality(modality, item_idx)
                    except Exception as e:
                        # Make it easy to set a breakpoint in the debugger
                        raise e

        return prompt_ids, mm_kwargs
883

884
885
    def _bind_prompt_replacements(
        self,
886
887
        prompt_repls: list[PromptReplacement],
    ) -> list[_BoundPromptReplacement]:
888
        tokenizer = self._get_tokenizer()
889

890
        return [prompt_repl.bind(tokenizer) for prompt_repl in prompt_repls]
891

892
893
894
    def _apply_prompt_replacements(
        self,
        token_ids: list[int],
895
        prompt_repls: Sequence[_BoundPromptReplacement],
896
        mm_item_counts: Mapping[str, int],
897
    ) -> tuple[list[int], str, list[_PlaceholderInfo]]:
898
        tokenizer = self._get_tokenizer()
899

900
        token_matches = find_token_matches(token_ids, prompt_repls)
901
902
903
904
        mm_match_counts = {
            modality: len(matches)
            for modality, matches in full_groupby_modality(token_matches)
        }
905
906
907
908
909
910
911
912
913
914
915
916

        # 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(
917
918
            mm_match_counts.get(modality, 0) >= item_count
            for modality, item_count in mm_item_counts.items()
919
920
921
922
        ):  # yapf: disable
            token_ids = replace_token_matches(
                token_ids,
                token_matches,
923
                mm_item_counts,
924
925
926
927
928
929
930
931
932
933
934
            )

            text = _decode(tokenizer, token_ids)
            matched_repls = [match.prompt_repl for match in token_matches]
        else:
            text = _decode(tokenizer, token_ids)

            text_matches = find_text_matches(text, prompt_repls)
            text = replace_text_matches(
                text,
                text_matches,
935
                mm_item_counts,
936
937
938
939
940
            )

            token_ids = _encode(tokenizer, text)
            matched_repls = [match.prompt_repl for match in text_matches]

941
        placeholders = self._find_placeholders(matched_repls, token_ids,
942
                                               mm_item_counts)
943
944

        return token_ids, text, placeholders
945

946
947
948
949
    def apply(
        self,
        prompt_text: str,
        mm_data: MultiModalDataDict,
950
        hf_processor_mm_kwargs: Mapping[str, object],
951
952
953
954
955
956
957
958
959
960
961
962
963
964
    ) -> 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.
        """
965
        mm_items = self._to_mm_items(mm_data)
966

967
968
969
970
971
        prompt_ids, mm_kwargs = self._cached_apply_hf_processor(
            prompt_text,
            mm_items,
            hf_processor_mm_kwargs,
        )
972

973
974
975
976
977
978
        unbound_prompt_repls = self._get_prompt_replacements(
            mm_items,
            hf_processor_mm_kwargs,
            mm_kwargs,
        )
        prompt_repls = self._bind_prompt_replacements(unbound_prompt_repls)
979

980
981
        # If HF processor already inserts placeholder tokens,
        # there is no need for us to insert them
982
        mm_item_counts = mm_items.get_all_counts()
983
984
        all_placeholders = self._find_placeholders(prompt_repls, prompt_ids,
                                                   mm_item_counts)
985

986
        if all_placeholders:
987
            tokenizer = self._get_tokenizer()
988
989
990
991
992
993
994
995
            prompt_text = _decode(tokenizer, prompt_ids)
        else:
            (
                prompt_ids,
                prompt_text,
                all_placeholders,
            ) = self._apply_prompt_replacements(
                prompt_ids,
996
                prompt_repls,
997
                mm_item_counts,
998
999
1000
1001
1002
1003
            )

        mm_placeholders = {
            modality: [item.to_range() for item in items]
            for modality, items in full_groupby_modality(all_placeholders)
        }
1004
1005
1006

        return MultiModalInputsV2(
            type="multimodal",
1007
1008
            prompt=prompt_text,
            prompt_token_ids=prompt_ids,
1009
1010
1011
            mm_kwargs=mm_kwargs,
            mm_placeholders=mm_placeholders,
        )
1012
1013

    @abstractmethod
1014
    def _get_dummy_mm_inputs(
1015
1016
        self,
        mm_counts: Mapping[str, int],
1017
    ) -> ProcessorInputs:
1018
        """
1019
1020
        Build the multi-modal portion of the input which, after processing,
        results in `mm_max_tokens` in :meth:`get_dummy_data`.
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
        """
        raise NotImplementedError

    def get_dummy_data(
        self,
        seq_len: int,
        mm_counts: Mapping[str, int],
        mm_max_tokens: Mapping[str, int],
    ) -> DummyData:
        # Avoid circular import
        from vllm.sequence import SequenceData

1033
        processor_inputs = self._get_dummy_mm_inputs(mm_counts)
1034
1035
1036
1037
1038
        mm_inputs = self.apply(
            prompt_text=processor_inputs.prompt_text,
            mm_data=processor_inputs.mm_data,
            hf_processor_mm_kwargs=processor_inputs.hf_processor_mm_kwargs,
        )
1039
1040
1041
1042

        prompt_token_ids = mm_inputs["prompt_token_ids"]
        placeholders_by_modality = mm_inputs["mm_placeholders"]

1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
        total_placeholders_by_modality = {
            modality: sum(item["length"] for item in placeholders)
            for modality, placeholders in placeholders_by_modality.items()
        }
        expected_placeholders_by_modality = {
            modality: mm_max_tokens[modality]
            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.")
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068

        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)
1069
1070
1071
1072
1073

        prompt_token_ids.extend([0] * (seq_len - len(prompt_token_ids)))

        return DummyData(
            seq_data=SequenceData.from_seqs(prompt_token_ids),
1074
1075
            multi_modal_data=mm_inputs["mm_kwargs"],
            multi_modal_placeholders=placeholders_by_modality,
1076
        )