serving_engine.py 31.8 KB
Newer Older
1
2
# SPDX-License-Identifier: Apache-2.0

3
import json
4
import sys
5
6
7
import time
from collections.abc import (AsyncGenerator, Iterable, Iterator, Mapping,
                             Sequence)
8
from concurrent.futures.thread import ThreadPoolExecutor
9
from http import HTTPStatus
10
from typing import (Annotated, Any, Callable, ClassVar, Generic, Optional,
11
                    TypeVar, Union)
12

13
from fastapi import Request
14
from pydantic import BaseModel, ConfigDict, Field
15
from starlette.datastructures import Headers
16

17
18
19
20
21
if sys.version_info >= (3, 12):
    from typing import TypedDict
else:
    from typing_extensions import TypedDict

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

logger = init_logger(__name__)

75
CompletionLikeRequest = Union[CompletionRequest, DetokenizeRequest,
76
                              EmbeddingCompletionRequest, RerankRequest,
77
78
                              ClassificationRequest, ScoreRequest,
                              TokenizeCompletionRequest]
79
80
81
82

ChatLikeRequest = Union[ChatCompletionRequest, EmbeddingChatRequest,
                        TokenizeChatRequest]

83
84
AnyRequest = Union[CompletionLikeRequest, ChatLikeRequest,
                   TranscriptionRequest]
85

86
87
88
89
90
91
92
93
94
95
96
AnyResponse = Union[
    CompletionResponse,
    ChatCompletionResponse,
    EmbeddingResponse,
    TranscriptionResponse,
    TokenizeResponse,
    PoolingResponse,
    ClassificationResponse,
    ScoreResponse,
]

97
98
99

class TextTokensPrompt(TypedDict):
    prompt: str
100
    prompt_token_ids: list[int]
101
102


103
RequestPrompt = Union[list[int], str, TextTokensPrompt]
104

105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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
160
161
162
163
164
165
166
167
168
169
170
171
RequestT = TypeVar("RequestT", bound=AnyRequest)


class RequestProcessingMixin(BaseModel):
    """
    Mixin for request processing, 
    handling prompt preparation and engine input.
    """
    request_prompts: Optional[Sequence[RequestPrompt]] = \
                            Field(default_factory=list)
    engine_prompts: Optional[list[TokensPrompt]] = \
                            Field(default_factory=list)

    model_config = ConfigDict(arbitrary_types_allowed=True)


class ResponseGenerationMixin(BaseModel):
    """
    Mixin for response generation, 
    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
    prompt_adapter_request: Optional[PromptAdapterRequest] = 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()

172

173
class OpenAIServing:
174
175
176
177
    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.”
    """
178

179
180
    def __init__(
        self,
181
        engine_client: EngineClient,
182
        model_config: ModelConfig,
183
        models: OpenAIServingModels,
184
185
        *,
        request_logger: Optional[RequestLogger],
186
        return_tokens_as_token_ids: bool = False,
187
    ):
188
189
        super().__init__()

190
        self.engine_client = engine_client
191
        self.model_config = model_config
192
193
        self.max_model_len = model_config.max_model_len

194
        self.models = models
195

196
        self.request_logger = request_logger
197
        self.return_tokens_as_token_ids = return_tokens_as_token_ids
198

199
200
201
202
203
204
205
206
        self._tokenizer_executor = ThreadPoolExecutor(max_workers=1)

        self._tokenize_prompt_input_async = make_async(
            self._tokenize_prompt_input, executor=self._tokenizer_executor)
        self._tokenize_prompt_input_or_inputs_async = make_async(
            self._tokenize_prompt_input_or_inputs,
            executor=self._tokenizer_executor)

207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
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
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
    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

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

            if not hasattr(ctx.request, "to_pooling_params"):
                return self.create_error_response(
                    "Request type does not support pooling parameters")

            pooling_params = ctx.request.to_pooling_params()

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

                self._log_inputs(
                    request_id_item,
                    ctx.request_prompts[i],
                    params=pooling_params,
                    lora_request=ctx.lora_request,
                    prompt_adapter_request=ctx.prompt_adapter_request)

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

368
369
370
371
372
373
374
375
376
    def create_error_response(
            self,
            message: str,
            err_type: str = "BadRequestError",
            status_code: HTTPStatus = HTTPStatus.BAD_REQUEST) -> ErrorResponse:
        return ErrorResponse(message=message,
                             type=err_type,
                             code=status_code.value)

377
378
379
380
381
382
383
384
385
386
387
388
389
    def create_streaming_error_response(
            self,
            message: str,
            err_type: str = "BadRequestError",
            status_code: HTTPStatus = HTTPStatus.BAD_REQUEST) -> str:
        json_str = json.dumps({
            "error":
            self.create_error_response(message=message,
                                       err_type=err_type,
                                       status_code=status_code).model_dump()
        })
        return json_str

390
    async def _check_model(
391
392
        self,
        request: AnyRequest,
393
    ) -> Optional[ErrorResponse]:
394
395
396

        error_response = None

397
        if self._is_model_supported(request.model):
398
            return None
399
400
401
        if request.model in [
                lora.lora_name for lora in self.models.lora_requests
        ]:
402
            return None
403
404
405
406
407
408
409
        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 \
                load_result.code == HTTPStatus.BAD_REQUEST.value:
                error_response = load_result
410
411
        if request.model in [
                prompt_adapter.prompt_adapter_name
412
                for prompt_adapter in self.models.prompt_adapter_requests
413
414
        ]:
            return None
415
416

        return error_response or self.create_error_response(
417
418
419
420
            message=f"The model `{request.model}` does not exist.",
            err_type="NotFoundError",
            status_code=HTTPStatus.NOT_FOUND)

421
422
    def _maybe_get_adapters(
        self, request: AnyRequest
423
    ) -> Union[tuple[None, None], tuple[LoRARequest, None], tuple[
424
            None, PromptAdapterRequest]]:
425
        if self._is_model_supported(request.model):
426
            return None, None
427
        for lora in self.models.lora_requests:
428
            if request.model == lora.lora_name:
429
                return lora, None
430
        for prompt_adapter in self.models.prompt_adapter_requests:
431
            if request.model == prompt_adapter.prompt_adapter_name:
432
                return None, prompt_adapter
433
        # if _check_model has been called earlier, this will be unreachable
434
        raise ValueError(f"The model `{request.model}` does not exist.")
435

436
437
438
439
440
    def _normalize_prompt_text_to_input(
        self,
        request: AnyRequest,
        tokenizer: AnyTokenizer,
        prompt: str,
441
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=-1)]],
442
443
        add_special_tokens: bool,
    ) -> TextTokensPrompt:
444
445
446
447
448
        if (self.model_config.encoder_config is not None
                and self.model_config.encoder_config.get(
                    "do_lower_case", False)):
            prompt = prompt.lower()

449
450
        if truncate_prompt_tokens is None:
            encoded = tokenizer(prompt, add_special_tokens=add_special_tokens)
451
452
453
454
455
456
        elif truncate_prompt_tokens < 0:
            # Negative means we cap at the model's max length
            encoded = tokenizer(prompt,
                                add_special_tokens=add_special_tokens,
                                truncation=True,
                                max_length=self.max_model_len)
457
        else:
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
            encoded = tokenizer(prompt,
                                add_special_tokens=add_special_tokens,
                                truncation=True,
                                max_length=truncate_prompt_tokens)

        input_ids = encoded.input_ids

        input_text = prompt

        return self._validate_input(request, input_ids, input_text)

    def _normalize_prompt_tokens_to_input(
        self,
        request: AnyRequest,
        tokenizer: AnyTokenizer,
473
        prompt_ids: list[int],
474
475
476
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=1)]],
    ) -> TextTokensPrompt:
        if truncate_prompt_tokens is None:
477
            input_ids = prompt_ids
478
479
        elif truncate_prompt_tokens < 0:
            input_ids = prompt_ids[-self.max_model_len:]
480
481
482
483
        else:
            input_ids = prompt_ids[-truncate_prompt_tokens:]

        input_text = tokenizer.decode(input_ids)
484

485
486
487
488
489
        return self._validate_input(request, input_ids, input_text)

    def _validate_input(
        self,
        request: AnyRequest,
490
        input_ids: list[int],
491
492
        input_text: str,
    ) -> TextTokensPrompt:
493
494
        token_num = len(input_ids)

495
496
        # Note: EmbeddingRequest, ClassificationRequest,
        # and ScoreRequest doesn't have max_tokens
497
498
        if isinstance(request,
                      (EmbeddingChatRequest, EmbeddingCompletionRequest,
499
500
501
502
503
                       ScoreRequest, RerankRequest, ClassificationRequest)):
            operation = {
                ScoreRequest: "score",
                ClassificationRequest: "classification"
            }.get(type(request), "embedding generation")
504

505
506
507
508
            if token_num > self.max_model_len:
                raise ValueError(
                    f"This model's maximum context length is "
                    f"{self.max_model_len} tokens. However, you requested "
509
510
                    f"{token_num} tokens in the input for {operation}. "
                    f"Please reduce the length of the input.")
511
512
            return TextTokensPrompt(prompt=input_text,
                                    prompt_token_ids=input_ids)
513

514
515
        # Note: TokenizeRequest and DetokenizeRequest doesn't have max_tokens
        # and does not require model context length validation
516
517
518
519
        if isinstance(request, (TokenizeCompletionRequest, TokenizeChatRequest,
                                DetokenizeRequest)):
            return TextTokensPrompt(prompt=input_text,
                                    prompt_token_ids=input_ids)
520

521
522
523
524
525
        # 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:
526
            max_tokens = getattr(request, "max_tokens", None)
527
        if max_tokens is None:
528
529
530
531
532
            if token_num >= self.max_model_len:
                raise ValueError(
                    f"This model's maximum context length is "
                    f"{self.max_model_len} tokens. However, you requested "
                    f"{token_num} tokens in the messages, "
533
                    f"Please reduce the length of the messages.")
534
        elif token_num + max_tokens > self.max_model_len:
535
            raise ValueError(
536
537
                f"This model's maximum context length is "
                f"{self.max_model_len} tokens. However, you requested "
538
                f"{max_tokens + token_num} tokens "
539
                f"({token_num} in the messages, "
540
                f"{max_tokens} in the completion). "
541
542
543
544
545
546
547
548
                f"Please reduce the length of the messages or completion.")

        return TextTokensPrompt(prompt=input_text, prompt_token_ids=input_ids)

    def _tokenize_prompt_input(
        self,
        request: AnyRequest,
        tokenizer: AnyTokenizer,
549
        prompt_input: Union[str, list[int]],
550
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=-1)]] = None,
551
552
553
        add_special_tokens: bool = True,
    ) -> TextTokensPrompt:
        """
554
        A simpler implementation of {meth}`_tokenize_prompt_input_or_inputs`
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
        that assumes single input.
        """
        return next(
            self._tokenize_prompt_inputs(
                request,
                tokenizer,
                [prompt_input],
                truncate_prompt_tokens=truncate_prompt_tokens,
                add_special_tokens=add_special_tokens,
            ))

    def _tokenize_prompt_inputs(
        self,
        request: AnyRequest,
        tokenizer: AnyTokenizer,
570
        prompt_inputs: Iterable[Union[str, list[int]]],
571
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=-1)]] = None,
572
573
574
        add_special_tokens: bool = True,
    ) -> Iterator[TextTokensPrompt]:
        """
575
        A simpler implementation of {meth}`_tokenize_prompt_input_or_inputs`
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
        that assumes multiple inputs.
        """
        for text in prompt_inputs:
            if isinstance(text, str):
                yield self._normalize_prompt_text_to_input(
                    request,
                    tokenizer,
                    prompt=text,
                    truncate_prompt_tokens=truncate_prompt_tokens,
                    add_special_tokens=add_special_tokens,
                )
            else:
                yield self._normalize_prompt_tokens_to_input(
                    request,
                    tokenizer,
                    prompt_ids=text,
                    truncate_prompt_tokens=truncate_prompt_tokens,
                )

    def _tokenize_prompt_input_or_inputs(
        self,
        request: AnyRequest,
        tokenizer: AnyTokenizer,
599
        input_or_inputs: Union[str, list[str], list[int], list[list[int]]],
600
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=-1)]] = None,
601
        add_special_tokens: bool = True,
602
    ) -> list[TextTokensPrompt]:
603
604
605
606
607
608
609
        """
        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.
        """
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
        # Although our type checking is based on mypy,
        # VSCode Pyright extension should still work properly
        # "is True" is required for Pyright to perform type narrowing
        # See: https://github.com/microsoft/pyright/issues/7672
        return [
            self._normalize_prompt_text_to_input(
                request,
                tokenizer,
                prompt=prompt_input["content"],
                truncate_prompt_tokens=truncate_prompt_tokens,
                add_special_tokens=add_special_tokens)
            if prompt_input["is_tokens"] is False else
            self._normalize_prompt_tokens_to_input(
                request,
                tokenizer,
                prompt_ids=prompt_input["content"],
                truncate_prompt_tokens=truncate_prompt_tokens)
            for prompt_input in parse_and_batch_prompt(input_or_inputs)
        ]
629

630
    async def _preprocess_completion(
631
632
633
        self,
        request: CompletionLikeRequest,
        tokenizer: AnyTokenizer,
634
        input_or_inputs: Union[str, list[str], list[int], list[list[int]]],
635
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=-1)]] = None,
636
        add_special_tokens: bool = True,
637
    ) -> tuple[list[TextTokensPrompt], list[TokensPrompt]]:
638
639
640
641
642
643
644
        request_prompts = 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,
        )
645
646
647
648
649
650
651
652
653
654
655
656

        engine_prompts = [
            TokensPrompt(prompt_token_ids=request_prompt["prompt_token_ids"])
            for request_prompt in request_prompts
        ]

        return request_prompts, engine_prompts

    async def _preprocess_chat(
        self,
        request: ChatLikeRequest,
        tokenizer: AnyTokenizer,
657
        messages: list[ChatCompletionMessageParam],
658
659
        chat_template: Optional[str],
        chat_template_content_format: ChatTemplateContentFormatOption,
660
661
        add_generation_prompt: bool = True,
        continue_final_message: bool = False,
662
663
664
        tool_dicts: Optional[list[dict[str, Any]]] = None,
        documents: Optional[list[dict[str, str]]] = None,
        chat_template_kwargs: Optional[dict[str, Any]] = None,
665
666
667
        tool_parser: Optional[Callable[[AnyTokenizer], ToolParser]] = None,
        truncate_prompt_tokens: Optional[Annotated[int, Field(ge=1)]] = None,
        add_special_tokens: bool = False,
668
669
    ) -> tuple[list[ConversationMessage], Sequence[RequestPrompt],
               list[TokensPrompt]]:
670
671
        model_config = self.model_config

672
        resolved_content_format = resolve_chat_template_content_format(
673
            model_config,
674
            chat_template,
675
            tool_dicts,
676
677
678
            chat_template_content_format,
            tokenizer,
        )
679
680
        conversation, mm_data_future = parse_chat_messages_futures(
            messages,
681
            model_config,
682
            tokenizer,
683
            content_format=resolved_content_format,
684
685
        )

686
        _chat_template_kwargs: dict[str, Any] = dict(
687
688
689
690
691
692
693
694
            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 {})

695
        request_prompt: Union[str, list[int]]
696
        if isinstance(tokenizer, MistralTokenizer):
697
698
699
            request_prompt = apply_mistral_chat_template(
                tokenizer,
                messages=messages,
700
                **_chat_template_kwargs,
701
702
703
            )
        else:
            request_prompt = apply_hf_chat_template(
704
                model_config,
705
706
                tokenizer,
                conversation=conversation,
707
                **_chat_template_kwargs,
708
709
710
711
            )

        mm_data = await mm_data_future

712
713
714
715
716
717
718
        # 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:
719
720
721
722
            if not isinstance(request, ChatCompletionRequest):
                msg = "Tool usage is only supported for Chat Completions API"
                raise NotImplementedError(msg)

723
724
            request = tool_parser(tokenizer).adjust_request(  # type: ignore
                request=request)
725
726

        if isinstance(request_prompt, str):
727
            prompt_inputs = await self._tokenize_prompt_input_async(
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
                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)

        engine_prompt = TokensPrompt(
            prompt_token_ids=prompt_inputs["prompt_token_ids"])
        if mm_data is not None:
            engine_prompt["multi_modal_data"] = mm_data
746
747
        if request.mm_processor_kwargs is not None:
            engine_prompt["mm_processor_kwargs"] = request.mm_processor_kwargs
748

749
750
751
        if hasattr(request, "cache_salt") and request.cache_salt is not None:
            engine_prompt["cache_salt"] = request.cache_salt

752
753
        return conversation, [request_prompt], [engine_prompt]

754
755
756
    def _log_inputs(
        self,
        request_id: str,
757
        inputs: RequestPrompt,
758
759
        params: Optional[Union[SamplingParams, PoolingParams,
                               BeamSearchParams]],
760
761
762
763
764
765
766
767
768
769
770
771
        lora_request: Optional[LoRARequest],
        prompt_adapter_request: Optional[PromptAdapterRequest],
    ) -> None:
        if self.request_logger is None:
            return

        if isinstance(inputs, str):
            prompt = inputs
            prompt_token_ids = None
        elif isinstance(inputs, list):
            prompt = None
            prompt_token_ids = inputs
772
        else:
773
774
775
776
777
778
779
780
781
782
783
            prompt = inputs["prompt"]
            prompt_token_ids = inputs["prompt_token_ids"]

        self.request_logger.log_inputs(
            request_id,
            prompt,
            prompt_token_ids,
            params=params,
            lora_request=lora_request,
            prompt_adapter_request=prompt_adapter_request,
        )
784

785
786
787
788
789
790
791
792
793
794
795
796
797
798
    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

799
    @staticmethod
800
    def _base_request_id(raw_request: Optional[Request],
801
802
803
                         default: Optional[str] = None) -> Optional[str]:
        """Pulls the request id to use from a header, if provided"""
        default = default or random_uuid()
804
805
806
807
        if raw_request is None:
            return default

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

809
    @staticmethod
810
811
812
813
814
815
816
    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}"

817
818
        if logprob.decoded_token is not None:
            return logprob.decoded_token
819
        return tokenizer.decode(token_id)
820

821
    def _is_model_supported(self, model_name: Optional[str]) -> bool:
822
823
        if not model_name:
            return True
824
        return self.models.is_base_model(model_name)
825
826
827
828
829
830

    def _get_model_name(self,
                        model_name: Optional[str] = None,
                        lora_request: Optional[LoRARequest] = None) -> str:
        if lora_request:
            return lora_request.lora_name
831
        if not model_name:
832
833
            return self.models.base_model_paths[0].name
        return model_name
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848


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