interfaces.py 27.7 KB
Newer Older
1
# SPDX-License-Identifier: Apache-2.0
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3

4
from collections.abc import Iterable, Mapping, MutableSequence
5
from typing import (TYPE_CHECKING, ClassVar, Literal, Optional, Protocol,
6
                    Union, overload, runtime_checkable)
7

8
import numpy as np
9
import torch
10
from torch import Tensor
11
from transformers import PretrainedConfig
12
from transformers.models.whisper.tokenization_whisper import LANGUAGES
13
from typing_extensions import Self, TypeIs
14

15
from vllm.config import ModelConfig, SpeechToTextConfig
16
from vllm.inputs import TokensPrompt
17
from vllm.inputs.data import PromptType
18
from vllm.logger import init_logger
19
20
from vllm.model_executor.layers.quantization.base_config import (
    QuantizationConfig)
21
from vllm.utils import supports_kw
22

23
from .interfaces_base import is_pooling_model
24

25
if TYPE_CHECKING:
26
    from vllm.attention import AttentionMetadata
27
    from vllm.config import VllmConfig
28
    from vllm.model_executor.models.utils import WeightsMapper
29
30
    from vllm.sequence import IntermediateTensors

31
32
logger = init_logger(__name__)

33
34
35
36
37
38
39
40
MultiModalEmbeddings = Union[list[Tensor], Tensor, tuple[Tensor, ...]]
"""
The output embeddings must be one of the following formats:

- A list or tuple of 2D tensors, where each tensor corresponds to
    each input multimodal data item (e.g, image).
- A single 3D tensor, with the batch dimension grouping the 2D tensors.
"""
41

42
43

@runtime_checkable
44
class SupportsMultiModal(Protocol):
45
    """The interface required for all multi-modal models."""
46

47
    supports_multimodal: ClassVar[Literal[True]] = True
48
    """
49
    A flag that indicates this model supports multi-modal inputs.
50
51
52
53
54

    Note:
        There is no need to redefine this flag if this class is in the
        MRO of your model class.
    """
55

56
57
58
59
60
61
    supports_multimodal_raw_input_only: ClassVar[bool] = False
    """
    A flag that indicates this model supports multi-modal inputs and processes
    them in their raw form and not embeddings.
    """

62
63
64
65
66
67
    supports_encoder_tp_data: ClassVar[bool] = False
    """
    A flag that indicates whether this model supports
    `multimodal_config.mm_encoder_tp_mode="data"`.
    """

68
69
70
71
72
73
74
    @classmethod
    def get_placeholder_str(cls, modality: str, i: int) -> Optional[str]:
        """
        Get the placeholder text for the `i`th `modality` item in the prompt.
        """
        ...

75
76
    def get_multimodal_embeddings(self,
                                  **kwargs: object) -> MultiModalEmbeddings:
77
78
79
        """
        Returns multimodal embeddings generated from multimodal kwargs 
        to be merged with text embeddings.
80

81
        Note:
82
83
            The returned multimodal embeddings must be in the same order as
            the appearances of their corresponding multimodal data item in the
84
            input prompt.
85
86
87
        """
        ...

88
89
90
91
92
93
94
95
96
97
98
99
    def get_language_model(self) -> torch.nn.Module:
        """
        Returns the underlying language model used for text generation.

        This is typically the `torch.nn.Module` instance responsible for 
        processing the merged multimodal embeddings and producing hidden states

        Returns:
            torch.nn.Module: The core language model component.
        """
        ...

100
101
102
103
104
    # Only for models that support v0 chunked prefill
    # TODO(ywang96): Remove this overload once v0 is deprecated
    @overload
    def get_input_embeddings(
        self,
105
        input_ids: Tensor,
106
        multimodal_embeddings: Optional[MultiModalEmbeddings] = None,
107
        attn_metadata: Optional["AttentionMetadata"] = None,
108
    ) -> Tensor:
109
110
        ...

111
    # TODO: Remove this overload once v0 is deprecated
112
    @overload
113
114
    def get_input_embeddings(
        self,
115
        input_ids: Tensor,
116
        multimodal_embeddings: Optional[MultiModalEmbeddings] = None,
117
118
119
120
121
122
123
124
125
126
    ) -> Tensor:
        ...

    def get_input_embeddings(
        self,
        input_ids: Tensor,
        multimodal_embeddings: Optional[MultiModalEmbeddings] = None,
        # Only necessary so that the v0 overload is valid
        # TODO: Remove attn_metadata once v0 is deprecated
        attn_metadata: Optional["AttentionMetadata"] = None,
127
    ) -> Tensor:
128
129
130
131
132
133
134
        """
        Returns the input embeddings merged from the text embeddings from 
        input_ids and the multimodal embeddings generated from multimodal 
        kwargs.
        """
        ...

135
136

@overload
137
def supports_multimodal(
138
        model: type[object]) -> TypeIs[type[SupportsMultiModal]]:
139
140
141
142
    ...


@overload
143
def supports_multimodal(model: object) -> TypeIs[SupportsMultiModal]:
144
145
146
    ...


147
def supports_multimodal(
148
149
    model: Union[type[object], object],
) -> Union[TypeIs[type[SupportsMultiModal]], TypeIs[SupportsMultiModal]]:
150
    return getattr(model, "supports_multimodal", False)
151
152


153
def supports_multimodal_raw_input_only(
154
        model: Union[type[object], object]) -> bool:
155
    return getattr(model, "supports_multimodal_raw_input_only", False)
156

157

158
159
160
def supports_multimodal_encoder_tp_data(
        model: Union[type[object], object]) -> bool:
    return getattr(model, "supports_encoder_tp_data", False)
161
162


163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
@runtime_checkable
class SupportsScoreTemplate(Protocol):
    """The interface required for all models that support score template."""

    supports_score_template: ClassVar[Literal[True]] = True
    """
    A flag that indicates this model supports score template.

    Note:
        There is no need to redefine this flag if this class is in the
        MRO of your model class.
    """

    @classmethod
    def get_score_template(cls, query: str, document: str) -> Optional[str]:
        """
        Generate a full prompt by populating the score template with query and document content.
        """ # noqa: E501
        ...

    @classmethod
    def post_process_tokens(cls, prompt: TokensPrompt) -> None:
        """
        Perform architecture-specific manipulations on the input tokens.
        """
        ...


@overload
def supports_score_template(
        model: type[object]) -> TypeIs[type[SupportsScoreTemplate]]:
    ...


@overload
def supports_score_template(model: object) -> TypeIs[SupportsScoreTemplate]:
    ...


def supports_score_template(
    model: Union[type[object], object],
) -> Union[TypeIs[type[SupportsScoreTemplate]], TypeIs[SupportsScoreTemplate]]:
205
    return getattr(model, "supports_score_template", False)
206
207


208
209
210
211
@runtime_checkable
class SupportsLoRA(Protocol):
    """The interface required for all models that support LoRA."""

212
213
214
215
216
217
218
219
    supports_lora: ClassVar[Literal[True]] = True
    """
    A flag that indicates this model supports LoRA.

    Note:
        There is no need to redefine this flag if this class is in the
        MRO of your model class.
    """
220
221
    # The `embedding_module` and `embedding_padding_modules`
    # are empty by default.
222
223
224
    embedding_modules: ClassVar[dict[str, str]] = {}
    embedding_padding_modules: ClassVar[list[str]] = []
    packed_modules_mapping: ClassVar[dict[str, list[str]]] = {}
225
226
227
228
229
230
231
232


# We can't use runtime_checkable with ClassVar for issubclass checks
# so we need to treat the class as an instance and use isinstance instead
@runtime_checkable
class _SupportsLoRAType(Protocol):
    supports_lora: Literal[True]

233
234
235
    packed_modules_mapping: dict[str, list[str]]
    embedding_modules: dict[str, str]
    embedding_padding_modules: list[str]
236
237
238


@overload
239
def supports_lora(model: type[object]) -> TypeIs[type[SupportsLoRA]]:
240
241
242
243
    ...


@overload
244
def supports_lora(model: object) -> TypeIs[SupportsLoRA]:
245
246
247
248
    ...


def supports_lora(
249
250
    model: Union[type[object], object],
) -> Union[TypeIs[type[SupportsLoRA]], TypeIs[SupportsLoRA]]:
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
    result = _supports_lora(model)

    if not result:
        lora_attrs = (
            "packed_modules_mapping",
            "embedding_modules",
            "embedding_padding_modules",
        )
        missing_attrs = tuple(attr for attr in lora_attrs
                              if not hasattr(model, attr))

        if getattr(model, "supports_lora", False):
            if missing_attrs:
                logger.warning(
                    "The model (%s) sets `supports_lora=True`, "
                    "but is missing LoRA-specific attributes: %s",
                    model,
                    missing_attrs,
                )
        else:
            if not missing_attrs:
                logger.warning(
                    "The model (%s) contains all LoRA-specific attributes, "
                    "but does not set `supports_lora=True`.", model)

    return result


279
def _supports_lora(model: Union[type[object], object]) -> bool:
280
281
282
283
    if isinstance(model, type):
        return isinstance(model, _SupportsLoRAType)

    return isinstance(model, SupportsLoRA)
284
285


286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
@runtime_checkable
class SupportsPP(Protocol):
    """The interface required for all models that support pipeline parallel."""

    supports_pp: ClassVar[Literal[True]] = True
    """
    A flag that indicates this model supports pipeline parallel.

    Note:
        There is no need to redefine this flag if this class is in the
        MRO of your model class.
    """

    def make_empty_intermediate_tensors(
        self,
        batch_size: int,
        dtype: torch.dtype,
        device: torch.device,
    ) -> "IntermediateTensors":
        """Called when PP rank > 0 for profiling purposes."""
        ...

    def forward(
        self,
310
        *,
311
        intermediate_tensors: Optional["IntermediateTensors"],
312
    ) -> Union[Tensor, "IntermediateTensors"]:
313
        """
314
315
        Accept [`IntermediateTensors`][vllm.sequence.IntermediateTensors] when
        PP rank > 0.
316

317
318
        Return [`IntermediateTensors`][vllm.sequence.IntermediateTensors] only
        for the last PP rank.
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
        """
        ...


# We can't use runtime_checkable with ClassVar for issubclass checks
# so we need to treat the class as an instance and use isinstance instead
@runtime_checkable
class _SupportsPPType(Protocol):
    supports_pp: Literal[True]

    def make_empty_intermediate_tensors(
        self,
        batch_size: int,
        dtype: torch.dtype,
        device: torch.device,
    ) -> "IntermediateTensors":
        ...

    def forward(
        self,
339
        *,
340
        intermediate_tensors: Optional["IntermediateTensors"],
341
    ) -> Union[Tensor, "IntermediateTensors"]:
342
343
344
345
        ...


@overload
346
def supports_pp(model: type[object]) -> TypeIs[type[SupportsPP]]:
347
348
349
350
351
352
353
354
355
    ...


@overload
def supports_pp(model: object) -> TypeIs[SupportsPP]:
    ...


def supports_pp(
356
357
    model: Union[type[object], object],
) -> Union[bool, TypeIs[type[SupportsPP]], TypeIs[SupportsPP]]:
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
    supports_attributes = _supports_pp_attributes(model)
    supports_inspect = _supports_pp_inspect(model)

    if supports_attributes and not supports_inspect:
        logger.warning(
            "The model (%s) sets `supports_pp=True`, but does not accept "
            "`intermediate_tensors` in its `forward` method", model)

    if not supports_attributes:
        pp_attrs = ("make_empty_intermediate_tensors", )
        missing_attrs = tuple(attr for attr in pp_attrs
                              if not hasattr(model, attr))

        if getattr(model, "supports_pp", False):
            if missing_attrs:
                logger.warning(
                    "The model (%s) sets `supports_pp=True`, "
                    "but is missing PP-specific attributes: %s",
                    model,
                    missing_attrs,
                )
        else:
            if not missing_attrs:
                logger.warning(
                    "The model (%s) contains all PP-specific attributes, "
                    "but does not set `supports_pp=True`.", model)

    return supports_attributes and supports_inspect


388
def _supports_pp_attributes(model: Union[type[object], object]) -> bool:
389
390
391
392
393
394
    if isinstance(model, type):
        return isinstance(model, _SupportsPPType)

    return isinstance(model, SupportsPP)


395
def _supports_pp_inspect(model: Union[type[object], object]) -> bool:
396
397
398
399
    model_forward = getattr(model, "forward", None)
    if not callable(model_forward):
        return False

400
    return supports_kw(model_forward, "intermediate_tensors")
401
402


403
404
405
406
407
408
409
410
@runtime_checkable
class HasInnerState(Protocol):
    """The interface required for all models that has inner state."""

    has_inner_state: ClassVar[Literal[True]] = True
    """
        A flag that indicates this model has inner state.
        Models that has inner state usually need access to the scheduler_config
411
        for max_num_seqs, etc. True for e.g. both Mamba and Jamba.
412
413
414
415
    """


@overload
416
def has_inner_state(model: object) -> TypeIs[HasInnerState]:
417
418
419
420
    ...


@overload
421
def has_inner_state(model: type[object]) -> TypeIs[type[HasInnerState]]:
422
423
424
425
    ...


def has_inner_state(
426
427
    model: Union[type[object], object]
) -> Union[TypeIs[type[HasInnerState]], TypeIs[HasInnerState]]:
428
    return getattr(model, "has_inner_state", False)
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449


@runtime_checkable
class IsAttentionFree(Protocol):
    """The interface required for all models like Mamba that lack attention,
    but do have state whose size is constant wrt the number of tokens."""

    is_attention_free: ClassVar[Literal[True]] = True
    """
        A flag that indicates this model has no attention.
        Used for block manager and attention backend selection.
        True for Mamba but not Jamba.
    """


@overload
def is_attention_free(model: object) -> TypeIs[IsAttentionFree]:
    ...


@overload
450
def is_attention_free(model: type[object]) -> TypeIs[type[IsAttentionFree]]:
451
452
453
454
    ...


def is_attention_free(
455
456
    model: Union[type[object], object]
) -> Union[TypeIs[type[IsAttentionFree]], TypeIs[IsAttentionFree]]:
457
    return getattr(model, "is_attention_free", False)
458
459


460
461
462
463
464
465
466
467
468
469
470
471
@runtime_checkable
class IsHybrid(Protocol):
    """The interface required for all models like Jamba that have both
    attention and mamba blocks, indicates that 
    hf_config has 'layers_block_type'"""

    is_hybrid: ClassVar[Literal[True]] = True
    """
        A flag that indicates this model has both mamba and attention blocks
        , also indicates that the model's hf_config has 
        'layers_block_type' """

472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
    @classmethod
    def get_mamba_state_shape_from_config(
        cls,
        vllm_config: "VllmConfig",
        use_v1: bool = True,
    ) -> tuple[tuple[int, int], tuple[int, int, int]]:
        """Calculate shapes for Mamba's convolutional and state caches.

        Args:
            vllm_config: vLLM config
            use_v1: Get shapes for V1 (or V0)

        Returns:
            Tuple containing:
            - conv_state_shape: Shape for convolutional state cache
            - temporal_state_shape: Shape for state space model cache
        """
        ...

491
492
493
494
495
496
497

@overload
def is_hybrid(model: object) -> TypeIs[IsHybrid]:
    ...


@overload
498
def is_hybrid(model: type[object]) -> TypeIs[type[IsHybrid]]:
499
500
501
502
    ...


def is_hybrid(
503
504
    model: Union[type[object], object]
) -> Union[TypeIs[type[IsHybrid]], TypeIs[IsHybrid]]:
505
    return getattr(model, "is_hybrid", False)
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
@runtime_checkable
class MixtureOfExperts(Protocol):
    """
    Check if the model is a mixture of experts (MoE) model.
    """

    expert_weights: MutableSequence[Iterable[Tensor]]
    """
    Expert weights saved in this rank.

    The first dimension is the layer, and the second dimension is different
    parameters in the layer, e.g. up/down projection weights.
    """

    num_moe_layers: int
    """Number of MoE layers in this model."""

    num_expert_groups: int
    """Number of expert groups in this model."""

    num_logical_experts: int
    """Number of logical experts in this model."""

    num_physical_experts: int
    """Number of physical experts in this model."""

    num_local_physical_experts: int
    """Number of local physical experts in this model."""

    num_routed_experts: int
    """Number of routed experts in this model."""

    num_shared_experts: int
    """Number of shared experts in this model."""

    num_redundant_experts: int
    """Number of redundant experts in this model."""

    def set_eplb_state(
        self,
        expert_load_view: Tensor,
        logical_to_physical_map: Tensor,
        logical_replica_count: Tensor,
    ) -> None:
        """
        Register the EPLB state in the MoE model.
        
        Since these are views of the actual EPLB state, any changes made by
        the EPLB algorithm are automatically reflected in the model's behavior
        without requiring additional method calls to set new states.

        You should also collect model's `expert_weights` here instead of in
        the weight loader, since after initial weight loading, further
        processing like quantization may be applied to the weights.

        Args:
            expert_load_view: A view of the expert load metrics tensor.
            logical_to_physical_map: Mapping from logical to physical experts.
            logical_replica_count: Count of replicas for each logical expert.
        """
        ...

570
571
572
573
574
575
576
    def update_physical_experts_metadata(
        self,
        num_physical_experts: int,
        num_local_physical_experts: int,
    ) -> None:
        ...

577
578
579
580
581

def is_mixture_of_experts(model: object) -> TypeIs[MixtureOfExperts]:
    return isinstance(model, MixtureOfExperts)


582
583
584
585
586
587
588
589
590
591
592
@runtime_checkable
class HasNoOps(Protocol):
    has_noops: ClassVar[Literal[True]] = True


@overload
def has_noops(model: object) -> TypeIs[HasNoOps]:
    ...


@overload
593
def has_noops(model: type[object]) -> TypeIs[type[HasNoOps]]:
594
595
596
597
    ...


def has_noops(
598
599
    model: Union[type[object], object]
) -> Union[TypeIs[type[HasNoOps]], TypeIs[HasNoOps]]:
600
    return getattr(model, "has_noops", False)
601
602


603
604
605
606
607
608
609
610
611
@runtime_checkable
class SupportsCrossEncoding(Protocol):
    """The interface required for all models that support cross encoding."""

    supports_cross_encoding: ClassVar[Literal[True]] = True


@overload
def supports_cross_encoding(
612
        model: type[object]) -> TypeIs[type[SupportsCrossEncoding]]:
613
614
615
616
617
618
619
620
621
    ...


@overload
def supports_cross_encoding(model: object) -> TypeIs[SupportsCrossEncoding]:
    ...


def _supports_cross_encoding(
622
623
    model: Union[type[object], object],
) -> Union[TypeIs[type[SupportsCrossEncoding]], TypeIs[SupportsCrossEncoding]]:
624
    return getattr(model, "supports_cross_encoding", False)
625
626
627


def supports_cross_encoding(
628
629
    model: Union[type[object], object],
) -> Union[TypeIs[type[SupportsCrossEncoding]], TypeIs[SupportsCrossEncoding]]:
630
    return is_pooling_model(model) and _supports_cross_encoding(model)
631
632


633
634
635
class SupportsQuant:
    """The interface required for all models that support quantization."""

636
637
    hf_to_vllm_mapper: ClassVar[Optional["WeightsMapper"]] = None
    packed_modules_mapping: ClassVar[Optional[dict[str, list[str]]]] = None
638
639
    quant_config: Optional[QuantizationConfig] = None

640
    def __new__(cls, *args, **kwargs) -> Self:
641
        instance = super().__new__(cls)
642
643

        # find config passed in arguments
644
645
        quant_config = cls._find_quant_config(*args, **kwargs)
        if quant_config is not None:
646
647

            # attach config to model for general use
648
            instance.quant_config = quant_config
649
650

            # apply model mappings to config for proper config-model matching
651
652
653
            if (hf_to_vllm_mapper := instance.hf_to_vllm_mapper) is not None:
                instance.quant_config.apply_vllm_mapper(hf_to_vllm_mapper)
            if instance.packed_modules_mapping is not None:
654
655
656
                instance.quant_config.packed_modules_mapping.update(
                    instance.packed_modules_mapping)

657
658
659
660
        return instance

    @staticmethod
    def _find_quant_config(*args, **kwargs) -> Optional[QuantizationConfig]:
661
        """Find quant config passed through model constructor args"""
662
663
664
665
666
667
668
669
670
671
672
673
674
        from vllm.config import VllmConfig  # avoid circular import

        args_values = list(args) + list(kwargs.values())
        for arg in args_values:
            if isinstance(arg, VllmConfig):
                return arg.quant_config

            if isinstance(arg, QuantizationConfig):
                return arg

        return None


675
676
677
@runtime_checkable
class SupportsTranscription(Protocol):
    """The interface required for all models that support transcription."""
678
679
    # Mapping from ISO639_1 language codes: language names
    supported_languages: ClassVar[Mapping[str, str]]
680
681
682

    supports_transcription: ClassVar[Literal[True]] = True

683
684
685
686
687
688
    supports_transcription_only: ClassVar[bool] = False
    """
    Transcription models can opt out of text generation by setting this to
    `True`.
    """

689
690
691
692
693
694
695
696
697
698
699
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        # language codes in supported_languages
        # that don't exist in the full language map
        invalid = set(cls.supported_languages) - set(LANGUAGES.keys())
        if invalid:
            raise ValueError(
                f"{cls.__name__}.supported_languages contains invalid "
                f"language codes: {sorted(invalid)}\n. "
                f"Valid choices are: {sorted(LANGUAGES.keys())}")

700
    @classmethod
701
    def get_generation_prompt(cls, audio: np.ndarray,
Patrick von Platen's avatar
Patrick von Platen committed
702
                              stt_config: SpeechToTextConfig,
703
                              model_config: ModelConfig,
704
705
706
707
                              language: Optional[str],
                              task_type: Literal["transcribe", "translate"],
                              request_prompt: str,
                              to_language: Optional[str]) -> PromptType:
708
709
710
        """Get the prompt for the ASR model.
        The model has control over the construction, as long as it
        returns a valid PromptType."""
711
712
713
        ...

    @classmethod
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
    def get_other_languages(cls) -> Mapping[str, str]:
        # other possible language codes from the whisper map
        return {
            k: v
            for k, v in LANGUAGES.items() if k not in cls.supported_languages
        }

    @classmethod
    def validate_language(cls, language: Optional[str]) -> Optional[str]:
        """
        Ensure the language specified in the transcription request 
        is a valid ISO 639-1 language code. If the request language is 
        valid, but not natively supported by the model, trigger a 
        warning (but not an exception).
        """
        if language is None or language in cls.supported_languages:
            return language
        elif language in cls.get_other_languages():
            logger.warning(
                "Language %r is not natively supported by %s; "
                "results may be less accurate. Supported languages: %r",
                language,
                cls.__name__,
                list(cls.supported_languages.keys()),
            )
            return language
        else:
            raise ValueError(
                f"Unsupported language: {language!r}.  Must be one of "
                f"{list(cls.supported_languages.keys())}.")
744

745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
    @classmethod
    def get_speech_to_text_config(
            cls, model_config: ModelConfig,
            task_type: Literal["transcribe",
                               "translate"]) -> SpeechToTextConfig:
        """Get the speech to text config for the ASR model."""
        ...

    @classmethod
    def get_num_audio_tokens(cls, audio_duration_s: float,
                             stt_config: SpeechToTextConfig,
                             model_config: ModelConfig) -> Optional[int]:
        """
        Map from audio duration to number of audio tokens produced by the ASR 
        model, without running a forward pass.
        This is used for estimating the amount of processing for this audio.
        """
        return None

764
765
766

@overload
def supports_transcription(
767
        model: type[object]) -> TypeIs[type[SupportsTranscription]]:
768
769
770
771
772
773
774
775
776
    ...


@overload
def supports_transcription(model: object) -> TypeIs[SupportsTranscription]:
    ...


def supports_transcription(
777
778
    model: Union[type[object], object],
) -> Union[TypeIs[type[SupportsTranscription]], TypeIs[SupportsTranscription]]:
779
    return getattr(model, "supports_transcription", False)
780
781
782
783
784
785
786
787
788
789


@runtime_checkable
class SupportsV0Only(Protocol):
    """Models with this interface are not compatible with V1 vLLM."""

    supports_v0_only: ClassVar[Literal[True]] = True


@overload
790
def supports_v0_only(model: type[object]) -> TypeIs[type[SupportsV0Only]]:
791
792
793
794
795
796
797
798
799
    ...


@overload
def supports_v0_only(model: object) -> TypeIs[SupportsV0Only]:
    ...


def supports_v0_only(
800
801
    model: Union[type[object], object],
) -> Union[TypeIs[type[SupportsV0Only]], TypeIs[SupportsV0Only]]:
802
    return getattr(model, "supports_v0_only", False)
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826


@runtime_checkable
class SupportsEagle3(Protocol):
    """The interface required for models that support 
    EAGLE3 speculative decoding."""

    supports_eagle3: ClassVar[Literal[True]] = True
    """
    A flag that indicates this model supports EAGLE3 
    speculative decoding.

    Note:
        There is no need to redefine this flag if this class is in the
        MRO of your model class.
    """

    def set_aux_hidden_state_layers(self, layers: tuple[int, ...]) -> None:
        """
        Set which layers should output auxiliary
        hidden states for EAGLE3.
        
        Args:
            layers: Tuple of layer indices that should output auxiliary
827
                hidden states.
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
        """
        ...

    def get_eagle3_aux_hidden_state_layers(self) -> tuple[int, ...]:
        """
        Get the layer indices that should output auxiliary hidden states
        for EAGLE3.
        
        Returns:
            Tuple of layer indices for auxiliary hidden state outputs.
        """
        ...


@overload
def supports_eagle3(model: type[object]) -> TypeIs[type[SupportsEagle3]]:
    ...


@overload
def supports_eagle3(model: object) -> TypeIs[SupportsEagle3]:
    ...


def supports_eagle3(
    model: Union[type[object], object],
) -> Union[TypeIs[type[SupportsEagle3]], TypeIs[SupportsEagle3]]:
    return isinstance(model, SupportsEagle3)
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922


@runtime_checkable
class SupportsMRoPE(Protocol):
    """The interface required for all models that support M-RoPE."""

    supports_mrope: ClassVar[Literal[True]] = True
    """
    A flag that indicates this model supports M-RoPE.
    
    Note:
        There is no need to redefine this flag if this class is in the
        MRO of your model class.
    """

    def get_mrope_input_positions(
        self,
        input_tokens: list[int],
        hf_config: PretrainedConfig,
        image_grid_thw: Optional[Union[list[list[int]], torch.Tensor]],
        video_grid_thw: Optional[Union[list[list[int]], torch.Tensor]],
        second_per_grid_ts: Optional[list[float]] = None,
        context_len: int = 0,
        seq_len: Optional[int] = None,
        audio_feature_lengths: Optional[torch.Tensor] = None,
        use_audio_in_video: bool = False,
    ) -> tuple[torch.Tensor, int]:
        """
        Get M-RoPE input positions and delta value for this specific model.
        
        This method should be implemented by each model that supports M-RoPE
        to provide model-specific logic for computing input positions.
        
        Args:
            input_tokens: List of input token IDs
            hf_config: HuggingFace model configuration
            image_grid_thw: Image grid dimensions (t, h, w)
            video_grid_thw: Video grid dimensions (t, h, w)
            second_per_grid_ts: Seconds per grid timestep for videos
            context_len: Context length
            seq_len: Sequence length
            audio_feature_lengths: Audio feature lengths for multimodal models
            use_audio_in_video: Whether to use audio in video for interleaving
            
        Returns:
            Tuple of (llm_positions, mrope_position_delta)
            - llm_positions: Tensor of shape [3, num_tokens]
                with T/H/W positions
            - mrope_position_delta: Delta for position calculations
        """
        ...


@overload
def supports_mrope(model: type[object]) -> TypeIs[type[SupportsMRoPE]]:
    ...


@overload
def supports_mrope(model: object) -> TypeIs[SupportsMRoPE]:
    ...


def supports_mrope(
    model: Union[type[object], object],
) -> Union[TypeIs[type[SupportsMRoPE]], TypeIs[SupportsMRoPE]]:
    return isinstance(model, SupportsMRoPE)