serving_engine.py 44.3 KB
Newer Older
1
# SPDX-License-Identifier: Apache-2.0
2
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
import asyncio
4
import io
5
import json
6
import sys
7
import time
8
import traceback
9
10
from collections.abc import AsyncGenerator, Iterable, Mapping, Sequence
from concurrent.futures import ThreadPoolExecutor
11
from http import HTTPStatus
12
from typing import (Annotated, Any, Callable, ClassVar, Generic, Optional,
13
                    TypeVar, Union, cast, overload)
14

15
import pybase64
16
import torch
17
from fastapi import Request
18
from pydantic import BaseModel, ConfigDict, Field
19
from starlette.datastructures import Headers
20
21
from typing_extensions import TypeIs

22
23
24
25
26
if sys.version_info >= (3, 12):
    from typing import TypedDict
else:
    from typing_extensions import TypedDict

27
import vllm.envs as envs
28
from vllm.config import ModelConfig
29
from vllm.engine.protocol import EngineClient
30
31
# yapf conflicts with isort for this block
# yapf: disable
32
from vllm.entrypoints.chat_utils import (ChatCompletionMessageParam,
33
                                         ChatTemplateContentFormatOption,
34
35
36
                                         ConversationMessage,
                                         apply_hf_chat_template,
                                         apply_mistral_chat_template,
37
38
                                         parse_chat_messages_futures,
                                         resolve_chat_template_content_format)
39
from vllm.entrypoints.context import ConversationContext
40
from vllm.entrypoints.logger import RequestLogger
41
from vllm.entrypoints.openai.protocol import (ChatCompletionRequest,
42
43
44
                                              ChatCompletionResponse,
                                              ClassificationRequest,
                                              ClassificationResponse,
45
                                              CompletionRequest,
46
                                              CompletionResponse,
47
                                              DetokenizeRequest,
48
49
                                              EmbeddingChatRequest,
                                              EmbeddingCompletionRequest,
50
                                              EmbeddingRequest,
51
52
53
54
                                              EmbeddingResponse, ErrorInfo,
                                              ErrorResponse, PoolingResponse,
                                              RerankRequest, ResponsesRequest,
                                              ScoreRequest, ScoreResponse,
55
                                              TokenizeChatRequest,
56
                                              TokenizeCompletionRequest,
57
58
                                              TokenizeResponse,
                                              TranscriptionRequest,
59
60
                                              TranscriptionResponse,
                                              TranslationRequest)
61
from vllm.entrypoints.openai.serving_models import OpenAIServingModels
62
from vllm.entrypoints.openai.tool_parsers import ToolParser
63
# yapf: enable
64
65
from vllm.inputs.data import EmbedsPrompt as EngineEmbedsPrompt
from vllm.inputs.data import TokensPrompt as EngineTokensPrompt
66
from vllm.inputs.parse import parse_and_batch_prompt
67
from vllm.logger import init_logger
68
from vllm.lora.request import LoRARequest
69
70
71
from vllm.multimodal import (  # noqa: F401 - Required to resolve Pydantic error in RequestProcessingMixin
    MultiModalDataDict)
from vllm.outputs import PoolingRequestOutput, RequestOutput
72
from vllm.pooling_params import PoolingParams
73
from vllm.sampling_params import BeamSearchParams, SamplingParams
74
from vllm.sequence import Logprob, PromptLogprobs
75
76
77
from vllm.tracing import (contains_trace_headers, extract_trace_headers,
                          log_tracing_disabled_warning)
from vllm.transformers_utils.tokenizer import AnyTokenizer, MistralTokenizer
78
79
from vllm.utils import (AsyncMicrobatchTokenizer, is_list_of,
                        merge_async_iterators, random_uuid)
80
81
82

logger = init_logger(__name__)

83
CompletionLikeRequest = Union[CompletionRequest, DetokenizeRequest,
84
                              EmbeddingCompletionRequest, RerankRequest,
85
86
                              ClassificationRequest, ScoreRequest,
                              TokenizeCompletionRequest]
87
88
89

ChatLikeRequest = Union[ChatCompletionRequest, EmbeddingChatRequest,
                        TokenizeChatRequest]
90
SpeechToTextRequest = Union[TranscriptionRequest, TranslationRequest]
91
92
AnyRequest = Union[CompletionLikeRequest, ChatLikeRequest, SpeechToTextRequest,
                   ResponsesRequest]
93

94
95
96
97
98
99
100
101
102
103
104
AnyResponse = Union[
    CompletionResponse,
    ChatCompletionResponse,
    EmbeddingResponse,
    TranscriptionResponse,
    TokenizeResponse,
    PoolingResponse,
    ClassificationResponse,
    ScoreResponse,
]

105
106
107

class TextTokensPrompt(TypedDict):
    prompt: str
108
    prompt_token_ids: list[int]
109
110


111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
class EmbedsPrompt(TypedDict):
    prompt_embeds: torch.Tensor


RequestPrompt = Union[list[int], str, TextTokensPrompt, EmbedsPrompt]


def is_text_tokens_prompt(prompt: RequestPrompt) -> TypeIs[TextTokensPrompt]:
    return (isinstance(prompt, dict) and "prompt_token_ids" in prompt
            and "prompt_embeds" not in prompt)


def is_embeds_prompt(prompt: RequestPrompt) -> TypeIs[EmbedsPrompt]:
    return (isinstance(prompt, dict) and "prompt_token_ids" not in prompt
            and "prompt_embeds" in prompt)

127

128
129
130
131
132
RequestT = TypeVar("RequestT", bound=AnyRequest)


class RequestProcessingMixin(BaseModel):
    """
133
    Mixin for request processing,
134
135
    handling prompt preparation and engine input.
    """
136
    request_prompts: Optional[Sequence[RequestPrompt]] = []
137
    engine_prompts: Optional[Union[list[EngineTokensPrompt],
138
                                   list[EngineEmbedsPrompt]]] = []
139
140
141
142
143
144

    model_config = ConfigDict(arbitrary_types_allowed=True)


class ResponseGenerationMixin(BaseModel):
    """
145
    Mixin for response generation,
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
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
    managing result generators and final batch results.
    """
    result_generator: Optional[AsyncGenerator[tuple[int, Union[
        RequestOutput, PoolingRequestOutput]], None]] = None
    final_res_batch: list[Union[RequestOutput, PoolingRequestOutput]] = Field(
        default_factory=list)

    model_config = ConfigDict(arbitrary_types_allowed=True)


class ServeContext(RequestProcessingMixin, ResponseGenerationMixin, BaseModel,
                   Generic[RequestT]):
    # Shared across all requests
    request: RequestT
    raw_request: Optional[Request] = None
    model_name: str
    request_id: str
    created_time: int = Field(default_factory=lambda: int(time.time()))
    lora_request: Optional[LoRARequest] = None

    # Shared across most requests
    tokenizer: Optional[AnyTokenizer] = None
    truncate_prompt_tokens: Optional[Annotated[int, Field(ge=1)]] = None

    # `protected_namespaces` resolves Pydantic v2's warning
    # on conflict with protected namespace "model_"
    model_config = ConfigDict(
        protected_namespaces=(),
        arbitrary_types_allowed=True,
    )


ClassificationServeContext = ServeContext[ClassificationRequest]


class EmbeddingServeContext(ServeContext[EmbeddingRequest]):
    chat_template: Optional[str] = None
    chat_template_content_format: ChatTemplateContentFormatOption


# Used to resolve the Pydantic error related to
# forward reference of MultiModalDataDict in TokensPrompt
RequestProcessingMixin.model_rebuild()
ServeContext.model_rebuild()
ClassificationServeContext.model_rebuild()
EmbeddingServeContext.model_rebuild()

193

194
class OpenAIServing:
195
196
197
198
    request_id_prefix: ClassVar[str] = """
    A short string prepended to every request’s ID (e.g. "embd", "classify")
    so you can easily tell “this ID came from Embedding vs Classification.”
    """
199

200
201
    def __init__(
        self,
202
        engine_client: EngineClient,
203
        model_config: ModelConfig,
204
        models: OpenAIServingModels,
205
206
        *,
        request_logger: Optional[RequestLogger],
207
        return_tokens_as_token_ids: bool = False,
208
        enable_force_include_usage: bool = False,
209
        log_error_stack: bool = False,
210
    ):
211
212
        super().__init__()

213
        self.engine_client = engine_client
214
        self.model_config = model_config
215
216
        self.max_model_len = model_config.max_model_len

217
        self.models = models
218

219
        self.request_logger = request_logger
220
        self.return_tokens_as_token_ids = return_tokens_as_token_ids
221
        self.enable_force_include_usage = enable_force_include_usage
222

223
224
        self._tokenizer_executor = ThreadPoolExecutor(max_workers=1)

225
226
        self._async_tokenizer_pool: dict[AnyTokenizer,
                                         AsyncMicrobatchTokenizer] = {}
227
        self.log_error_stack = log_error_stack
228
229
230

    def _get_async_tokenizer(self, tokenizer) -> AsyncMicrobatchTokenizer:
        """
231
        Return (and cache) an `AsyncMicrobatchTokenizer` bound to the
232
233
234
235
236
237
238
        given tokenizer.
        """
        async_tokenizer = self._async_tokenizer_pool.get(tokenizer)
        if async_tokenizer is None:
            async_tokenizer = AsyncMicrobatchTokenizer(tokenizer)
            self._async_tokenizer_pool[tokenizer] = async_tokenizer
        return async_tokenizer
239

240
241
242
243
244
245
246
247
248
249
250
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
279
280
281
282
283
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
    async def _preprocess(
        self,
        ctx: ServeContext,
    ) -> Optional[ErrorResponse]:
        """
        Default preprocessing hook. Subclasses may override
        to prepare `ctx` (classification, embedding, etc.).
        """
        return None

    def _build_response(
        self,
        ctx: ServeContext,
    ) -> Union[AnyResponse, ErrorResponse]:
        """
        Default response builder. Subclass may override this method
        to return the appropriate response object.
        """
        return self.create_error_response("unimplemented endpoint")

    async def handle(
        self,
        ctx: ServeContext,
    ) -> Union[AnyResponse, ErrorResponse]:
        generation: AsyncGenerator[Union[AnyResponse, ErrorResponse], None]
        generation = self._pipeline(ctx)

        async for response in generation:
            return response

        return self.create_error_response("No response yielded from pipeline")

    async def _pipeline(
        self,
        ctx: ServeContext,
    ) -> AsyncGenerator[Union[AnyResponse, ErrorResponse], None]:
        """Execute the request processing pipeline yielding responses."""
        if error := await self._check_model(ctx.request):
            yield error
        if error := self._validate_request(ctx):
            yield error

        preprocess_ret = await self._preprocess(ctx)
        if isinstance(preprocess_ret, ErrorResponse):
            yield preprocess_ret

        generators_ret = await self._prepare_generators(ctx)
        if isinstance(generators_ret, ErrorResponse):
            yield generators_ret

        collect_ret = await self._collect_batch(ctx)
        if isinstance(collect_ret, ErrorResponse):
            yield collect_ret

        yield self._build_response(ctx)

    def _validate_request(self, ctx: ServeContext) -> Optional[ErrorResponse]:
        truncate_prompt_tokens = getattr(ctx.request, "truncate_prompt_tokens",
                                         None)

        if truncate_prompt_tokens is not None:
            if truncate_prompt_tokens <= self.max_model_len:
                ctx.truncate_prompt_tokens = truncate_prompt_tokens
            else:
                return self.create_error_response(
                    "truncate_prompt_tokens value is "
                    "greater than max_model_len."
                    " Please, select a smaller truncation size.")
        return None

310
311
312
313
314
315
316
317
318
319
    def _create_pooling_params(
        self,
        ctx: ServeContext,
    ) -> Union[PoolingParams, ErrorResponse]:
        if not hasattr(ctx.request, "to_pooling_params"):
            return self.create_error_response(
                "Request type does not support pooling parameters")

        return ctx.request.to_pooling_params()

320
321
322
323
324
325
326
327
328
329
330
331
332
    async def _prepare_generators(
        self,
        ctx: ServeContext,
    ) -> Optional[ErrorResponse]:
        """Schedule the request and get the result generator."""
        generators: list[AsyncGenerator[Union[RequestOutput,
                                              PoolingRequestOutput],
                                        None]] = []

        try:
            trace_headers = (None if ctx.raw_request is None else await
                             self._get_trace_headers(ctx.raw_request.headers))

333
334
335
            pooling_params = self._create_pooling_params(ctx)
            if isinstance(pooling_params, ErrorResponse):
                return pooling_params
336
337
338
339
340
341
342
343
344
345
346
347

            if ctx.engine_prompts is None:
                return self.create_error_response(
                    "Engine prompts not available")

            for i, engine_prompt in enumerate(ctx.engine_prompts):
                request_id_item = f"{ctx.request_id}-{i}"

                if ctx.request_prompts is None:
                    return self.create_error_response(
                        "Request prompts not available")

348
349
350
351
                self._log_inputs(request_id_item,
                                 ctx.request_prompts[i],
                                 params=pooling_params,
                                 lora_request=ctx.lora_request)
352

353
354
355
356
357
358
                # Mypy has an existing bug related to inferring the variance of
                # TypedDicts with `builtins.enumerate`:
                # https://github.com/python/mypy/issues/8586#issuecomment-2867698435
                engine_prompt = cast(
                    Union[EngineTokensPrompt, EngineEmbedsPrompt],
                    engine_prompt)
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
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
                generator = self.engine_client.encode(
                    engine_prompt,
                    pooling_params,
                    request_id_item,
                    lora_request=ctx.lora_request,
                    trace_headers=trace_headers,
                    priority=getattr(ctx.request, "priority", 0),
                )

                generators.append(generator)

            ctx.result_generator = merge_async_iterators(*generators)

            return None

        except Exception as e:
            # TODO: Use a vllm-specific Validation Error
            return self.create_error_response(str(e))

    async def _collect_batch(
        self,
        ctx: ServeContext,
    ) -> Optional[ErrorResponse]:
        """Collect batch results from the result generator."""
        try:
            if ctx.engine_prompts is None:
                return self.create_error_response(
                    "Engine prompts not available")

            num_prompts = len(ctx.engine_prompts)
            final_res_batch: list[Optional[Union[RequestOutput,
                                                 PoolingRequestOutput]]]
            final_res_batch = [None] * num_prompts

            if ctx.result_generator is None:
                return self.create_error_response(
                    "Result generator not available")

            async for i, res in ctx.result_generator:
                final_res_batch[i] = res

            if None in final_res_batch:
                return self.create_error_response(
                    "Failed to generate results for all prompts")

            ctx.final_res_batch = [
                res for res in final_res_batch if res is not None
            ]

            return None

        except Exception as e:
            return self.create_error_response(str(e))

413
414
415
416
417
    def create_error_response(
            self,
            message: str,
            err_type: str = "BadRequestError",
            status_code: HTTPStatus = HTTPStatus.BAD_REQUEST) -> ErrorResponse:
418
419
420
421
422
423
        if self.log_error_stack:
            exc_type, _, _ = sys.exc_info()
            if exc_type is not None:
                traceback.print_exc()
            else:
                traceback.print_stack()
424
425
        return ErrorResponse(error=ErrorInfo(
            message=message, type=err_type, code=status_code.value))
426

427
428
429
430
431
    def create_streaming_error_response(
            self,
            message: str,
            err_type: str = "BadRequestError",
            status_code: HTTPStatus = HTTPStatus.BAD_REQUEST) -> str:
432
        json_str = json.dumps(
433
434
            self.create_error_response(message=message,
                                       err_type=err_type,
435
                                       status_code=status_code).model_dump())
436
437
        return json_str

438
    async def _check_model(
439
440
        self,
        request: AnyRequest,
441
    ) -> Optional[ErrorResponse]:
442
443
444

        error_response = None

445
        if self._is_model_supported(request.model):
446
            return None
447
        if request.model in self.models.lora_requests:
448
            return None
449
450
451
452
453
        if envs.VLLM_ALLOW_RUNTIME_LORA_UPDATING and request.model and (
                load_result := await self.models.resolve_lora(request.model)):
            if isinstance(load_result, LoRARequest):
                return None
            if isinstance(load_result, ErrorResponse) and \
454
                load_result.error.code == HTTPStatus.BAD_REQUEST.value:
455
456
457
                error_response = load_result

        return error_response or self.create_error_response(
458
459
460
461
            message=f"The model `{request.model}` does not exist.",
            err_type="NotFoundError",
            status_code=HTTPStatus.NOT_FOUND)

462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
    def _get_active_default_mm_loras(
            self, request: AnyRequest) -> Optional[LoRARequest]:
        """Determine if there are any active default multimodal loras."""
        # TODO: Currently this is only enabled for chat completions
        # to be better aligned with only being enabled for .generate
        # when run offline. It would be nice to support additional
        # tasks types in the future.
        message_types = self._get_message_types(request)
        default_mm_loras = set()

        for lora in self.models.lora_requests.values():
            # Best effort match for default multimodal lora adapters;
            # There is probably a better way to do this, but currently
            # this matches against the set of 'types' in any content lists
            # up until '_', e.g., to match audio_url -> audio
            if lora.lora_name in message_types:
                default_mm_loras.add(lora)

        # Currently only support default modality specific loras if
        # we have exactly one lora matched on the request.
        if len(default_mm_loras) == 1:
            return default_mm_loras.pop()
        return None

486
    def _maybe_get_adapters(
487
488
489
        self,
        request: AnyRequest,
        supports_default_mm_loras: bool = False,
490
    ) -> Optional[LoRARequest]:
491

492
        if request.model in self.models.lora_requests:
493
            return self.models.lora_requests[request.model]
494
495
496
497
498
499

        # Currently only support default modality specific loras
        # if we have exactly one lora matched on the request.
        if supports_default_mm_loras:
            default_mm_lora = self._get_active_default_mm_loras(request)
            if default_mm_lora is not None:
500
                return default_mm_lora
501
502

        if self._is_model_supported(request.model):
503
            return None
504

505
        # if _check_model has been called earlier, this will be unreachable
506
        raise ValueError(f"The model `{request.model}` does not exist.")
507

508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
    def _get_message_types(self, request: AnyRequest) -> set[str]:
        """Retrieve the set of types from message content dicts up
        until `_`; we use this to match potential multimodal data
        with default per modality loras.
        """
        message_types: set[str] = set()

        if not hasattr(request, "messages"):
            return message_types

        for message in request.messages:
            if (isinstance(message, dict) and "content" in message
                    and isinstance(message["content"], list)):
                for content_dict in message["content"]:
                    if "type" in content_dict:
                        message_types.add(content_dict["type"].split("_")[0])
        return message_types

526
    async def _normalize_prompt_text_to_input(
527
528
529
        self,
        request: AnyRequest,
        prompt: str,
530
        tokenizer: AnyTokenizer,
531
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=-1)]],
532
533
        add_special_tokens: bool,
    ) -> TextTokensPrompt:
534
535
        async_tokenizer = self._get_async_tokenizer(tokenizer)

536
537
538
539
540
        if (self.model_config.encoder_config is not None
                and self.model_config.encoder_config.get(
                    "do_lower_case", False)):
            prompt = prompt.lower()

541
        if truncate_prompt_tokens is None:
542
543
            encoded = await async_tokenizer(
                prompt, add_special_tokens=add_special_tokens)
544
545
        elif truncate_prompt_tokens < 0:
            # Negative means we cap at the model's max length
546
547
548
549
550
            encoded = await async_tokenizer(
                prompt,
                add_special_tokens=add_special_tokens,
                truncation=True,
                max_length=self.max_model_len)
551
        else:
552
553
554
555
556
            encoded = await async_tokenizer(
                prompt,
                add_special_tokens=add_special_tokens,
                truncation=True,
                max_length=truncate_prompt_tokens)
557
558
559
560
561
562

        input_ids = encoded.input_ids
        input_text = prompt

        return self._validate_input(request, input_ids, input_text)

563
    async def _normalize_prompt_tokens_to_input(
564
565
        self,
        request: AnyRequest,
566
        prompt_ids: list[int],
567
        tokenizer: Optional[AnyTokenizer],
568
569
570
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=1)]],
    ) -> TextTokensPrompt:
        if truncate_prompt_tokens is None:
571
            input_ids = prompt_ids
572
573
        elif truncate_prompt_tokens < 0:
            input_ids = prompt_ids[-self.max_model_len:]
574
575
576
        else:
            input_ids = prompt_ids[-truncate_prompt_tokens:]

577
578
579
580
581
        if tokenizer is None:
            input_text = ""
        else:
            async_tokenizer = self._get_async_tokenizer(tokenizer)
            input_text = await async_tokenizer.decode(input_ids)
582

583
584
585
586
587
        return self._validate_input(request, input_ids, input_text)

    def _validate_input(
        self,
        request: AnyRequest,
588
        input_ids: list[int],
589
590
        input_text: str,
    ) -> TextTokensPrompt:
591
592
        token_num = len(input_ids)

593
594
        # Note: EmbeddingRequest, ClassificationRequest,
        # and ScoreRequest doesn't have max_tokens
595
596
        if isinstance(request,
                      (EmbeddingChatRequest, EmbeddingCompletionRequest,
597
                       ScoreRequest, RerankRequest, ClassificationRequest)):
598

599
600
            # Note: input length can be up to the entire model context length
            # since these requests don't generate tokens.
601
            if token_num > self.max_model_len:
602
603
604
605
606
607
                operations: dict[type[AnyRequest], str] = {
                    ScoreRequest: "score",
                    ClassificationRequest: "classification"
                }
                operation = operations.get(type(request),
                                           "embedding generation")
608
609
610
                raise ValueError(
                    f"This model's maximum context length is "
                    f"{self.max_model_len} tokens. However, you requested "
611
612
                    f"{token_num} tokens in the input for {operation}. "
                    f"Please reduce the length of the input.")
613
614
            return TextTokensPrompt(prompt=input_text,
                                    prompt_token_ids=input_ids)
615

616
617
        # Note: TokenizeRequest and DetokenizeRequest doesn't have max_tokens
        # and does not require model context length validation
618
619
620
621
        if isinstance(request, (TokenizeCompletionRequest, TokenizeChatRequest,
                                DetokenizeRequest)):
            return TextTokensPrompt(prompt=input_text,
                                    prompt_token_ids=input_ids)
622

623
624
625
626
627
        # chat completion endpoint supports max_completion_tokens
        if isinstance(request, ChatCompletionRequest):
            # TODO(#9845): remove max_tokens when field dropped from OpenAI API
            max_tokens = request.max_completion_tokens or request.max_tokens
        else:
628
            max_tokens = getattr(request, "max_tokens", None)
629
630
631
632

        # Note: input length can be up to model context length - 1 for
        # completion-like requests.
        if token_num >= self.max_model_len:
633
            raise ValueError(
634
                f"This model's maximum context length is "
635
636
637
638
639
640
641
642
643
644
645
646
                f"{self.max_model_len} tokens. However, your request has "
                f"{token_num} input tokens. Please reduce the length of "
                "the input messages.")

        if max_tokens is not None and \
            token_num + max_tokens > self.max_model_len:
            raise ValueError(
                "'max_tokens' or 'max_completion_tokens' is too large: "
                f"{max_tokens}. This model's maximum context length is "
                f"{self.max_model_len} tokens and your request has "
                f"{token_num} input tokens ({max_tokens} > {self.max_model_len}"
                f" - {token_num}).")
647
648
649

        return TextTokensPrompt(prompt=input_text, prompt_token_ids=input_ids)

650
    async def _tokenize_prompt_input_async(
651
652
653
        self,
        request: AnyRequest,
        tokenizer: AnyTokenizer,
654
        prompt_input: Union[str, list[int]],
655
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=-1)]] = None,
656
657
658
        add_special_tokens: bool = True,
    ) -> TextTokensPrompt:
        """
659
660
        A simpler implementation of
        [`_tokenize_prompt_input_or_inputs`][vllm.entrypoints.openai.serving_engine.OpenAIServing._tokenize_prompt_input_or_inputs]
661
662
        that assumes single input.
        """
663
        async for result in self._tokenize_prompt_inputs_async(
664
665
                request,
                tokenizer,
666
            [prompt_input],
667
668
                truncate_prompt_tokens=truncate_prompt_tokens,
                add_special_tokens=add_special_tokens,
669
670
671
        ):
            return result
        raise ValueError("No results yielded from tokenization")
672

673
    async def _tokenize_prompt_inputs_async(
674
675
676
        self,
        request: AnyRequest,
        tokenizer: AnyTokenizer,
677
        prompt_inputs: Iterable[Union[str, list[int]]],
678
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=-1)]] = None,
679
        add_special_tokens: bool = True,
680
    ) -> AsyncGenerator[TextTokensPrompt, None]:
681
        """
682
683
        A simpler implementation of
        [`_tokenize_prompt_input_or_inputs`][vllm.entrypoints.openai.serving_engine.OpenAIServing._tokenize_prompt_input_or_inputs]
684
685
        that assumes multiple inputs.
        """
686
687
        for prompt in prompt_inputs:
            if isinstance(prompt, str):
688
                yield await self._normalize_prompt_text_to_input(
689
                    request,
690
691
                    prompt=prompt,
                    tokenizer=tokenizer,
692
693
694
695
                    truncate_prompt_tokens=truncate_prompt_tokens,
                    add_special_tokens=add_special_tokens,
                )
            else:
696
                yield await self._normalize_prompt_tokens_to_input(
697
                    request,
698
699
                    prompt_ids=prompt,
                    tokenizer=tokenizer,
700
701
702
                    truncate_prompt_tokens=truncate_prompt_tokens,
                )

703
    async def _tokenize_prompt_input_or_inputs_async(
704
705
        self,
        request: AnyRequest,
706
        tokenizer: Optional[AnyTokenizer],
707
708
        input_or_inputs: Optional[Union[str, list[str], list[int],
                                        list[list[int]]]],
709
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=-1)]] = None,
710
        add_special_tokens: bool = True,
711
    ) -> tuple[list[TextTokensPrompt], list[EmbedsPrompt]]:
712
713
714
715
716
717
718
        """
        Tokenize/detokenize depending on the input format.

        According to `OpenAI API <https://platform.openai.com/docs/api-reference/embeddings/create>`_
        , each input can be a string or array of tokens. Note that each request
        can pass one or more inputs.
        """
719
720
721
722
723
724
725
726
727
728
729
730
731
732
        inputs_embeds = list[EmbedsPrompt]()
        inputs_text = list[TextTokensPrompt]()

        if (isinstance(request, CompletionRequest)
                and request.prompt_embeds is not None):
            inputs_embeds.extend(
                self._load_prompt_embeds(request.prompt_embeds,
                                         truncate_prompt_tokens))

        # Empty prompts are okay as long as there are prompt embeddings
        if input_or_inputs is None or (inputs_embeds
                                       and input_or_inputs == ""):
            return [], inputs_embeds

733
734
        # Although our type checking is based on mypy,
        # VSCode Pyright extension should still work properly
735
        # "is False" is required for Pyright to perform type narrowing
736
        # See: https://github.com/microsoft/pyright/issues/7672
737
738
739
740
741
742
743
744

        # Parse and batch the input prompts
        batch_inputs = parse_and_batch_prompt(input_or_inputs)

        # Process each input in the batch concurrently
        tasks = []
        for prompt_input in batch_inputs:
            if prompt_input["is_tokens"] is False:
745
746
                assert tokenizer is not None, \
                    "Tokenizer is required for text prompts"
747
748
749
                task = self._normalize_prompt_text_to_input(
                    request,
                    prompt_input["content"],
750
                    tokenizer=tokenizer,
751
752
753
754
755
756
                    truncate_prompt_tokens=truncate_prompt_tokens,
                    add_special_tokens=add_special_tokens)
            else:
                task = self._normalize_prompt_tokens_to_input(
                    request,
                    prompt_input["content"],
757
                    tokenizer=tokenizer,
758
759
760
761
762
763
                    truncate_prompt_tokens=truncate_prompt_tokens)
            tasks.append(task)

        # Wait for all tokenization tasks to complete
        results = await asyncio.gather(*tasks)
        inputs_text.extend(results)
764
765

        return inputs_text, inputs_embeds
766

767
    @overload
768
    async def _preprocess_completion(
769
        self,
770
771
772
        request: Union[DetokenizeRequest, EmbeddingCompletionRequest,
                       RerankRequest, ClassificationRequest, ScoreRequest,
                       TokenizeCompletionRequest],
773
        tokenizer: Optional[AnyTokenizer],
774
        input_or_inputs: Union[str, list[str], list[int], list[list[int]]],
775
776
777
778
779
780
781
782
783
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=-1)]] = ...,
        add_special_tokens: bool = ...,
    ) -> tuple[list[TextTokensPrompt], list[EngineTokensPrompt]]:
        ...

    @overload
    async def _preprocess_completion(
        self,
        request: CompletionRequest,
784
        tokenizer: Optional[AnyTokenizer],
785
786
787
788
789
790
791
792
793
794
795
        input_or_inputs: Optional[Union[str, list[str], list[int],
                                        list[list[int]]]],
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=-1)]] = ...,
        add_special_tokens: bool = ...,
    ) -> tuple[list[Union[TextTokensPrompt, EmbedsPrompt]], list[Union[
            EngineTokensPrompt, EngineEmbedsPrompt]]]:
        ...

    async def _preprocess_completion(
        self,
        request: CompletionLikeRequest,
796
        tokenizer: Optional[AnyTokenizer],
797
798
        input_or_inputs: Optional[Union[str, list[str], list[int],
                                        list[list[int]]]],
799
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=-1)]] = None,
800
        add_special_tokens: bool = True,
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
    ) -> tuple[Union[list[TextTokensPrompt], list[Union[
            TextTokensPrompt, EmbedsPrompt]]], Union[
                list[EngineTokensPrompt], list[Union[EngineTokensPrompt,
                                                     EngineEmbedsPrompt]]]]:
        if not isinstance(request,
                          CompletionRequest) and input_or_inputs is None:
            raise ValueError(
                "Prompt embeds with non-completion requests is not"
                " currently supported.")

        (request_prompts_text, request_prompts_embeds
         ) = await self._tokenize_prompt_input_or_inputs_async(
             request,
             tokenizer,
             input_or_inputs,
             truncate_prompt_tokens=truncate_prompt_tokens,
             add_special_tokens=add_special_tokens,
         )

        engine_prompts_text = [
            EngineTokensPrompt(
                prompt_token_ids=request_prompt_text["prompt_token_ids"])
            for request_prompt_text in request_prompts_text
        ]
825
826
827
828
829
830
        cache_salt = request.cache_salt if (
            hasattr(request, "cache_salt")
            and request.cache_salt is not None) else None
        if cache_salt:
            for prompt_text in engine_prompts_text:
                prompt_text["cache_salt"] = cache_salt
831

832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
        # This check is equivalent to simply checking if
        # `request_prompts_embeds` is empty, but it's difficult to propagate
        # overloads to the private helper functions to enable this check.
        # This overload is needed because only TextPrompts are allowed for
        # non-completion requests and if we don't add the overload here,
        # everywhere this function is used outside of serving_completion will
        # need logic asserting that only text prompts are in the request.
        if not isinstance(request,
                          CompletionRequest) and input_or_inputs is not None:
            return request_prompts_text, engine_prompts_text

        engine_prompts_embeds = [
            EngineEmbedsPrompt(
                prompt_embeds=request_prompt_embeds["prompt_embeds"])
            for request_prompt_embeds in request_prompts_embeds
847
        ]
848
849
850
        if cache_salt:
            for prompt_embed in engine_prompts_embeds:
                prompt_embed["cache_salt"] = cache_salt
851

852
853
        request_prompts = request_prompts_embeds + request_prompts_text
        engine_prompts = engine_prompts_embeds + engine_prompts_text
854
855
856
857
        return request_prompts, engine_prompts

    async def _preprocess_chat(
        self,
858
        request: Union[ChatLikeRequest, ResponsesRequest],
859
        tokenizer: AnyTokenizer,
860
        messages: list[ChatCompletionMessageParam],
861
862
        chat_template: Optional[str],
        chat_template_content_format: ChatTemplateContentFormatOption,
863
864
        add_generation_prompt: bool = True,
        continue_final_message: bool = False,
865
866
867
        tool_dicts: Optional[list[dict[str, Any]]] = None,
        documents: Optional[list[dict[str, str]]] = None,
        chat_template_kwargs: Optional[dict[str, Any]] = None,
868
869
870
        tool_parser: Optional[Callable[[AnyTokenizer], ToolParser]] = None,
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=1)]] = None,
        add_special_tokens: bool = False,
871
    ) -> tuple[list[ConversationMessage], Sequence[RequestPrompt],
872
               list[EngineTokensPrompt]]:
873
874
        model_config = self.model_config

875
876
        resolved_content_format = resolve_chat_template_content_format(
            chat_template,
877
            tool_dicts,
878
879
            chat_template_content_format,
            tokenizer,
880
            model_config=model_config,
881
        )
882
883
        conversation, mm_data_future = parse_chat_messages_futures(
            messages,
884
            model_config,
885
            tokenizer,
886
            content_format=resolved_content_format,
887
888
        )

889
        _chat_template_kwargs: dict[str, Any] = dict(
890
891
892
893
894
895
896
897
            chat_template=chat_template,
            add_generation_prompt=add_generation_prompt,
            continue_final_message=continue_final_message,
            tools=tool_dicts,
            documents=documents,
        )
        _chat_template_kwargs.update(chat_template_kwargs or {})

898
        request_prompt: Union[str, list[int]]
899
900
901
902

        if tokenizer is None:
            request_prompt = "placeholder"
        elif isinstance(tokenizer, MistralTokenizer):
903
904
905
            request_prompt = apply_mistral_chat_template(
                tokenizer,
                messages=messages,
906
                **_chat_template_kwargs,
907
908
909
            )
        else:
            request_prompt = apply_hf_chat_template(
910
                tokenizer=tokenizer,
911
                conversation=conversation,
912
                model_config=model_config,
913
                **_chat_template_kwargs,
914
915
916
917
            )

        mm_data = await mm_data_future

918
919
920
921
922
923
924
        # tool parsing is done only if a tool_parser has been set and if
        # tool_choice is not "none" (if tool_choice is "none" but a tool_parser
        # is set, we want to prevent parsing a tool_call hallucinated by the LLM
        should_parse_tools = tool_parser is not None and (hasattr(
            request, "tool_choice") and request.tool_choice != "none")

        if should_parse_tools:
925
926
927
928
            if not isinstance(request, ChatCompletionRequest):
                msg = "Tool usage is only supported for Chat Completions API"
                raise NotImplementedError(msg)

929
930
            request = tool_parser(tokenizer).adjust_request(  # type: ignore
                request=request)
931

932
933
934
935
936
937
938
939
        if tokenizer is None:
            assert isinstance(request_prompt, str), (
                "Prompt has to be a string", \
                "when the tokenizer is not initialised"
            )
            prompt_inputs = TextTokensPrompt(prompt=request_prompt,
                                             prompt_token_ids=[1])
        elif isinstance(request_prompt, str):
940
            prompt_inputs = await self._tokenize_prompt_input_async(
941
942
943
944
945
946
947
948
949
950
951
952
953
954
                request,
                tokenizer,
                request_prompt,
                truncate_prompt_tokens=truncate_prompt_tokens,
                add_special_tokens=add_special_tokens,
            )
        else:
            # For MistralTokenizer
            assert is_list_of(request_prompt, int), (
                "Prompt has to be either a string or a list of token ids")
            prompt_inputs = TextTokensPrompt(
                prompt=tokenizer.decode(request_prompt),
                prompt_token_ids=request_prompt)

955
        engine_prompt = EngineTokensPrompt(
956
957
958
            prompt_token_ids=prompt_inputs["prompt_token_ids"])
        if mm_data is not None:
            engine_prompt["multi_modal_data"] = mm_data
959
960
        if request.mm_processor_kwargs is not None:
            engine_prompt["mm_processor_kwargs"] = request.mm_processor_kwargs
961

962
963
964
        if hasattr(request, "cache_salt") and request.cache_salt is not None:
            engine_prompt["cache_salt"] = request.cache_salt

965
966
        return conversation, [request_prompt], [engine_prompt]

967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
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
    async def _generate_with_builtin_tools(
        self,
        request_id: str,
        request_prompt: RequestPrompt,
        engine_prompt: EngineTokensPrompt,
        sampling_params: SamplingParams,
        context: ConversationContext,
        lora_request: Optional[LoRARequest] = None,
        priority: int = 0,
        **kwargs,
    ):
        orig_priority = priority
        while True:
            self._log_inputs(
                request_id,
                request_prompt,
                params=sampling_params,
                lora_request=lora_request,
            )
            generator = self.engine_client.generate(
                engine_prompt,
                sampling_params,
                request_id,
                lora_request=lora_request,
                priority=priority,
                **kwargs,
            )
            async for res in generator:
                context.append_output(res)
                # NOTE(woosuk): The stop condition is handled by the engine.
                yield context

            if not context.need_builtin_tool_call():
                # The model did not ask for a tool call, so we're done.
                break

            # Call the tool and update the context with the result.
            tool_output = await context.call_tool()
            context.append_output(tool_output)

            # TODO: uncomment this and enable tool output streaming
            # yield context

            # Create inputs for the next turn.
            # Render the next prompt token ids.
            prompt_token_ids = context.render_for_completion()
            engine_prompt = EngineTokensPrompt(
                prompt_token_ids=prompt_token_ids)
            request_prompt = prompt_token_ids
            # Update the sampling params.
            sampling_params.max_tokens = (self.max_model_len -
                                          len(prompt_token_ids))
            # OPTIMIZATION
            priority = orig_priority - 1

1022
    @staticmethod
1023
1024
1025
1026
1027
1028
    def _load_prompt_embeds(
        prompt_embeds: Optional[Union[bytes, list[bytes]]],
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=1)]] = None
    ) -> list[EmbedsPrompt]:

        def _load_and_validate_embed(embed: bytes) -> EmbedsPrompt:
1029
1030
            tensor = torch.load(io.BytesIO(
                pybase64.b64decode(embed, validate=True)),
1031
1032
                                weights_only=True,
                                map_location=torch.device("cpu"))
1033
1034
1035
1036
1037
            assert isinstance(tensor, torch.Tensor) and tensor.dtype in (
                torch.float32,
                torch.bfloat16,
                torch.float16,
            )
1038
            tensor = tensor.to_dense()
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
            if tensor.dim() > 2:
                tensor = tensor.squeeze(0)
                assert tensor.dim() == 2
            if truncate_prompt_tokens is not None:
                tensor = tensor[-truncate_prompt_tokens:]
            return {"prompt_embeds": tensor}

        if prompt_embeds:
            if isinstance(prompt_embeds, list):
                return [
                    _load_and_validate_embed(embed) for embed in prompt_embeds
                ]
            else:
                return [_load_and_validate_embed(prompt_embeds)]
        else:
            return []

1056
1057
1058
    def _log_inputs(
        self,
        request_id: str,
1059
        inputs: RequestPrompt,
1060
1061
        params: Optional[Union[SamplingParams, PoolingParams,
                               BeamSearchParams]],
1062
1063
1064
1065
        lora_request: Optional[LoRARequest],
    ) -> None:
        if self.request_logger is None:
            return
1066
        prompt, prompt_token_ids, prompt_embeds = None, None, None
1067
1068
1069
1070
        if isinstance(inputs, str):
            prompt = inputs
        elif isinstance(inputs, list):
            prompt_token_ids = inputs
1071
1072
        elif 'prompt_embeds' in inputs:
            prompt_embeds = inputs.get("prompt_embeds")
1073
        else:
1074
1075
1076
1077
1078
1079
1080
            prompt = inputs["prompt"]
            prompt_token_ids = inputs["prompt_token_ids"]

        self.request_logger.log_inputs(
            request_id,
            prompt,
            prompt_token_ids,
1081
            prompt_embeds,
1082
1083
1084
            params=params,
            lora_request=lora_request,
        )
1085

1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
    async def _get_trace_headers(
        self,
        headers: Headers,
    ) -> Optional[Mapping[str, str]]:
        is_tracing_enabled = await self.engine_client.is_tracing_enabled()

        if is_tracing_enabled:
            return extract_trace_headers(headers)

        if contains_trace_headers(headers):
            log_tracing_disabled_warning()

        return None

1100
    @staticmethod
1101
    def _base_request_id(raw_request: Optional[Request],
1102
1103
1104
                         default: Optional[str] = None) -> Optional[str]:
        """Pulls the request id to use from a header, if provided"""
        default = default or random_uuid()
1105
1106
1107
1108
        if raw_request is None:
            return default

        return raw_request.headers.get("X-Request-Id", default)
1109

1110
    @staticmethod
1111
1112
1113
1114
1115
1116
1117
    def _get_decoded_token(logprob: Logprob,
                           token_id: int,
                           tokenizer: AnyTokenizer,
                           return_as_token_id: bool = False) -> str:
        if return_as_token_id:
            return f"token_id:{token_id}"

1118
1119
        if logprob.decoded_token is not None:
            return logprob.decoded_token
1120
        return tokenizer.decode(token_id)
1121

1122
    def _is_model_supported(self, model_name: Optional[str]) -> bool:
1123
1124
        if not model_name:
            return True
1125
        return self.models.is_base_model(model_name)
1126
1127
1128
1129
1130
1131

    def _get_model_name(self,
                        model_name: Optional[str] = None,
                        lora_request: Optional[LoRARequest] = None) -> str:
        if lora_request:
            return lora_request.lora_name
1132
        if not model_name:
1133
1134
            return self.models.base_model_paths[0].name
        return model_name
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149


def clamp_prompt_logprobs(
    prompt_logprobs: Union[PromptLogprobs,
                           None]) -> Union[PromptLogprobs, None]:
    if prompt_logprobs is None:
        return prompt_logprobs

    for logprob_dict in prompt_logprobs:
        if logprob_dict is None:
            continue
        for logprob_values in logprob_dict.values():
            if logprob_values.logprob == float('-inf'):
                logprob_values.logprob = -9999.0
    return prompt_logprobs