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

12
import torch
13
from fastapi import Request
14
from pydantic import BaseModel, ConfigDict, Field
15
from starlette.datastructures import Headers
16
17
from typing_extensions import TypeIs

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

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

78
from vllm.transformers_utils.tokenizers import CPM9GTokenizer
79
80
from vllm.utils import (AsyncMicrobatchTokenizer, is_list_of,
                        merge_async_iterators, random_uuid)
81
82
83

logger = init_logger(__name__)

84
85
86
87
88
89
90
91
92
CompletionLikeRequest = Union[
    CompletionRequest,
    DetokenizeRequest,
    EmbeddingCompletionRequest,
    RerankRequest,
    ClassificationRequest,
    ScoreRequest,
    TokenizeCompletionRequest,
]
93
94
95

ChatLikeRequest = Union[ChatCompletionRequest, EmbeddingChatRequest,
                        TokenizeChatRequest]
96
SpeechToTextRequest = Union[TranscriptionRequest, TranslationRequest]
97
98
99
100
101
102
103
AnyRequest = Union[
    CompletionLikeRequest,
    ChatLikeRequest,
    SpeechToTextRequest,
    ResponsesRequest,
    IOProcessorRequest,
]
104

105
106
107
108
109
110
111
112
113
114
115
AnyResponse = Union[
    CompletionResponse,
    ChatCompletionResponse,
    EmbeddingResponse,
    TranscriptionResponse,
    TokenizeResponse,
    PoolingResponse,
    ClassificationResponse,
    ScoreResponse,
]

116
117
118

class TextTokensPrompt(TypedDict):
    prompt: str
119
    prompt_token_ids: list[int]
120
121


122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
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)

138

139
140
141
142
143
RequestT = TypeVar("RequestT", bound=AnyRequest)


class RequestProcessingMixin(BaseModel):
    """
144
    Mixin for request processing,
145
146
    handling prompt preparation and engine input.
    """
147

148
    request_prompts: Optional[Sequence[RequestPrompt]] = []
149
    engine_prompts: Optional[list[EngineTokensPrompt]] = []
150
151
152
153
154
155

    model_config = ConfigDict(arbitrary_types_allowed=True)


class ResponseGenerationMixin(BaseModel):
    """
156
    Mixin for response generation,
157
158
    managing result generators and final batch results.
    """
159

160
161
162
163
164
165
166
167
    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)


168
169
170
171
172
173
class ServeContext(
        RequestProcessingMixin,
        ResponseGenerationMixin,
        BaseModel,
        Generic[RequestT],
):
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
205
206
    # 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

    # `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()
207
208


209
class OpenAIServing:
210
211
212
213
    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.”
    """
214

215
216
    def __init__(
        self,
217
        engine_client: EngineClient,
218
        model_config: ModelConfig,
219
        models: OpenAIServingModels,
220
221
        *,
        request_logger: Optional[RequestLogger],
222
        return_tokens_as_token_ids: bool = False,
223
        enable_force_include_usage: bool = False,
224
        log_error_stack: bool = False,
225
    ):
226
227
        super().__init__()

228
        self.engine_client = engine_client
229
        self.model_config = model_config
230
        self.max_model_len = model_config.max_model_len
231
        self.tokenizer_mode = model_config.tokenizer_mode
luopl's avatar
luopl committed
232

233
        if model_config.tokenizer_mode == "cpm":
luopl's avatar
luopl committed
234
            self.tokenizer = CPM9GTokenizer(model_config.model, trust_remote_code=True)
235

236
        self.models = models
237

238
        self.request_logger = request_logger
239
        self.return_tokens_as_token_ids = return_tokens_as_token_ids
240
        self.enable_force_include_usage = enable_force_include_usage
241

242
243
        self._tokenizer_executor = ThreadPoolExecutor(max_workers=1)

244
245
        self._async_tokenizer_pool: dict[AnyTokenizer,
                                         AsyncMicrobatchTokenizer] = {}
246
        self.log_error_stack = log_error_stack
247

248
249
250
251
252
253
254
255
256
257
    def _get_renderer(self, tokenizer: Optional[AnyTokenizer]) -> BaseRenderer:
        """
        Get a Renderer instance with the provided tokenizer.
        Uses shared async tokenizer pool for efficiency.
        """
        return CompletionRenderer(
            model_config=self.model_config,
            tokenizer=tokenizer,
            async_tokenizer_pool=self._async_tokenizer_pool)

258
259
260
261
262
263
264
265
266
267
268
269
270
    def _build_render_config(
        self,
        request: Any,
    ) -> RenderConfig:
        """
        Build and return a `RenderConfig` for an endpoint.

        Used by the renderer to control how prompts are prepared
        (e.g., tokenization and length handling). Endpoints should
        implement this with logic appropriate to their request type.
        """
        raise NotImplementedError

271
272
    def _get_async_tokenizer(self, tokenizer) -> AsyncMicrobatchTokenizer:
        """
273
        Return (and cache) an `AsyncMicrobatchTokenizer` bound to the
274
275
276
277
278
279
280
        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
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
    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)

342
343
        if (truncate_prompt_tokens is not None
                and truncate_prompt_tokens > self.max_model_len):
344
345
346
347
            return self.create_error_response(
                "truncate_prompt_tokens value is "
                "greater than max_model_len."
                " Please, select a smaller truncation size.")
348
349
        return None

350
351
352
353
354
355
356
357
358
359
    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()

360
361
362
363
364
365
366
367
368
369
370
371
372
    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))

373
374
375
            pooling_params = self._create_pooling_params(ctx)
            if isinstance(pooling_params, ErrorResponse):
                return pooling_params
376
377
378
379
380
381
382

            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}"
luopl's avatar
luopl committed
383
384
                if pooling_params.qfeat is not None:
                    engine_prompt["qfeat"] = pooling_params.qfeat
385
386
                self._log_inputs(
                    request_id_item,
387
                    engine_prompt,
388
389
390
                    params=pooling_params,
                    lora_request=ctx.lora_request,
                )
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445

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

446
    def create_error_response(
447
448
449
450
451
        self,
        message: str,
        err_type: str = "BadRequestError",
        status_code: HTTPStatus = HTTPStatus.BAD_REQUEST,
    ) -> ErrorResponse:
452
453
454
455
456
457
        if self.log_error_stack:
            exc_type, _, _ = sys.exc_info()
            if exc_type is not None:
                traceback.print_exc()
            else:
                traceback.print_stack()
458
459
        return ErrorResponse(error=ErrorInfo(
            message=message, type=err_type, code=status_code.value))
460

461
    def create_streaming_error_response(
462
463
464
465
466
        self,
        message: str,
        err_type: str = "BadRequestError",
        status_code: HTTPStatus = HTTPStatus.BAD_REQUEST,
    ) -> str:
467
        json_str = json.dumps(
468
469
            self.create_error_response(message=message,
                                       err_type=err_type,
470
                                       status_code=status_code).model_dump())
471
472
        return json_str

473
    async def _check_model(
474
475
        self,
        request: AnyRequest,
476
    ) -> Optional[ErrorResponse]:
477
478
        error_response = None

479
        if self._is_model_supported(request.model):
480
            return None
481
        if request.model in self.models.lora_requests:
482
            return None
483
484
        if (envs.VLLM_ALLOW_RUNTIME_LORA_UPDATING and request.model and
            (load_result := await self.models.resolve_lora(request.model))):
485
486
            if isinstance(load_result, LoRARequest):
                return None
487
488
            if (isinstance(load_result, ErrorResponse) and
                    load_result.error.code == HTTPStatus.BAD_REQUEST.value):
489
490
491
                error_response = load_result

        return error_response or self.create_error_response(
492
493
            message=f"The model `{request.model}` does not exist.",
            err_type="NotFoundError",
494
495
            status_code=HTTPStatus.NOT_FOUND,
        )
496

497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
    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

521
    def _maybe_get_adapters(
522
523
524
        self,
        request: AnyRequest,
        supports_default_mm_loras: bool = False,
525
    ) -> Optional[LoRARequest]:
526
        if request.model in self.models.lora_requests:
527
            return self.models.lora_requests[request.model]
528
529
530
531
532
533

        # 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:
534
                return default_mm_lora
535

536
        if self._is_model_supported(request.model):
537
            return None
538

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

542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
    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

560
    async def _normalize_prompt_text_to_input(
561
562
563
        self,
        request: AnyRequest,
        prompt: str,
564
        tokenizer: AnyTokenizer,
565
566
        add_special_tokens: bool,
    ) -> TextTokensPrompt:
567
568
        async_tokenizer = self._get_async_tokenizer(tokenizer)

569
570
571
572
573
        if (self.model_config.encoder_config is not None
                and self.model_config.encoder_config.get(
                    "do_lower_case", False)):
            prompt = prompt.lower()

574
575
576
        truncate_prompt_tokens = getattr(request, "truncate_prompt_tokens",
                                         None)

577
        if truncate_prompt_tokens is None:
578
579
            encoded = await async_tokenizer(
                prompt, add_special_tokens=add_special_tokens)
580
581
        elif truncate_prompt_tokens < 0:
            # Negative means we cap at the model's max length
582
583
584
585
            encoded = await async_tokenizer(
                prompt,
                add_special_tokens=add_special_tokens,
                truncation=True,
586
587
                max_length=self.max_model_len,
            )
588
        else:
589
590
591
592
            encoded = await async_tokenizer(
                prompt,
                add_special_tokens=add_special_tokens,
                truncation=True,
593
594
                max_length=truncate_prompt_tokens,
            )
595

596
597
598
599
        if self.tokenizer_mode == "cpm":
            input_ids = [self.tokenizer.bos_id] + self.tokenizer.encode(prompt)
        else:
            input_ids = encoded.input_ids
600
601
602
603
604

        input_text = prompt

        return self._validate_input(request, input_ids, input_text)

605
    async def _normalize_prompt_tokens_to_input(
606
607
        self,
        request: AnyRequest,
608
        prompt_ids: list[int],
609
        tokenizer: Optional[AnyTokenizer],
610
    ) -> TextTokensPrompt:
611
612
        truncate_prompt_tokens = getattr(request, "truncate_prompt_tokens",
                                         None)
613

614
        if truncate_prompt_tokens is None:
615
            input_ids = prompt_ids
616
617
        elif truncate_prompt_tokens < 0:
            input_ids = prompt_ids[-self.max_model_len:]
618
619
620
        else:
            input_ids = prompt_ids[-truncate_prompt_tokens:]

621
622
623
        if tokenizer is None:
            input_text = ""
        else:
luopl's avatar
luopl committed
624
            async_tokenizer = self._get_async_tokenizer(tokenizer)
625
            input_text = await async_tokenizer.decode(input_ids) if self.tokenizer_mode != "cpm" else await self.tokenizer.decode_all(input_ids)
626

627
628
629
630
631
        return self._validate_input(request, input_ids, input_text)

    def _validate_input(
        self,
        request: AnyRequest,
632
        input_ids: list[int],
633
634
        input_text: str,
    ) -> TextTokensPrompt:
635
636
        token_num = len(input_ids)

637
638
        # Note: EmbeddingRequest, ClassificationRequest,
        # and ScoreRequest doesn't have max_tokens
639
640
641
642
643
644
645
646
647
648
        if isinstance(
                request,
            (
                EmbeddingChatRequest,
                EmbeddingCompletionRequest,
                ScoreRequest,
                RerankRequest,
                ClassificationRequest,
            ),
        ):
649
650
            # Note: input length can be up to the entire model context length
            # since these requests don't generate tokens.
651
            if token_num > self.max_model_len:
652
653
                operations: dict[type[AnyRequest], str] = {
                    ScoreRequest: "score",
654
                    ClassificationRequest: "classification",
655
656
657
                }
                operation = operations.get(type(request),
                                           "embedding generation")
658
659
660
                raise ValueError(
                    f"This model's maximum context length is "
                    f"{self.max_model_len} tokens. However, you requested "
661
662
                    f"{token_num} tokens in the input for {operation}. "
                    f"Please reduce the length of the input.")
663
664
            return TextTokensPrompt(prompt=input_text,
                                    prompt_token_ids=input_ids)
665

666
667
        # Note: TokenizeRequest and DetokenizeRequest doesn't have max_tokens
        # and does not require model context length validation
668
669
670
671
672
        if isinstance(
                request,
            (TokenizeCompletionRequest, TokenizeChatRequest,
             DetokenizeRequest),
        ):
673
674
            return TextTokensPrompt(prompt=input_text,
                                    prompt_token_ids=input_ids)
675

676
677
678
679
680
        # 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:
681
            max_tokens = getattr(request, "max_tokens", None)
682
683
684
685

        # Note: input length can be up to model context length - 1 for
        # completion-like requests.
        if token_num >= self.max_model_len:
686
            raise ValueError(
687
                f"This model's maximum context length is "
688
689
690
691
                f"{self.max_model_len} tokens. However, your request has "
                f"{token_num} input tokens. Please reduce the length of "
                "the input messages.")

692
693
        if (max_tokens is not None
                and token_num + max_tokens > self.max_model_len):
694
695
696
697
698
699
            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}).")
700
701
702

        return TextTokensPrompt(prompt=input_text, prompt_token_ids=input_ids)

703
    async def _tokenize_prompt_input_async(
704
705
706
        self,
        request: AnyRequest,
        tokenizer: AnyTokenizer,
707
        prompt_input: Union[str, list[int]],
708
709
710
        add_special_tokens: bool = True,
    ) -> TextTokensPrompt:
        """
711
        A simpler implementation that tokenizes a single prompt input.
712
        """
713
        async for result in self._tokenize_prompt_inputs_async(
714
715
                request,
                tokenizer,
716
            [prompt_input],
717
                add_special_tokens=add_special_tokens,
718
719
720
        ):
            return result
        raise ValueError("No results yielded from tokenization")
721

722
    async def _tokenize_prompt_inputs_async(
723
724
725
        self,
        request: AnyRequest,
        tokenizer: AnyTokenizer,
726
        prompt_inputs: Iterable[Union[str, list[int]]],
727
        add_special_tokens: bool = True,
728
    ) -> AsyncGenerator[TextTokensPrompt, None]:
729
        """
730
        A simpler implementation that tokenizes multiple prompt inputs.
731
        """
732
733
        for prompt in prompt_inputs:
            if isinstance(prompt, str):
734
                yield await self._normalize_prompt_text_to_input(
735
                    request,
736
737
                    prompt=prompt,
                    tokenizer=tokenizer,
738
739
740
                    add_special_tokens=add_special_tokens,
                )
            else:
741
                yield await self._normalize_prompt_tokens_to_input(
742
                    request,
743
744
                    prompt_ids=prompt,
                    tokenizer=tokenizer,
745
746
                )

747
748
    async def _preprocess_chat(
        self,
749
        request: Union[ChatLikeRequest, ResponsesRequest],
750
        tokenizer: AnyTokenizer,
751
        messages: list[ChatCompletionMessageParam],
752
753
        chat_template: Optional[str],
        chat_template_content_format: ChatTemplateContentFormatOption,
754
755
        add_generation_prompt: bool = True,
        continue_final_message: bool = False,
756
757
758
        tool_dicts: Optional[list[dict[str, Any]]] = None,
        documents: Optional[list[dict[str, str]]] = None,
        chat_template_kwargs: Optional[dict[str, Any]] = None,
759
760
        tool_parser: Optional[Callable[[AnyTokenizer], ToolParser]] = None,
        add_special_tokens: bool = False,
761
762
763
764
765
    ) -> tuple[
            list[ConversationMessage],
            Sequence[RequestPrompt],
            list[EngineTokensPrompt],
    ]:
766
767
        model_config = self.model_config

768
769
        resolved_content_format = resolve_chat_template_content_format(
            chat_template,
770
            tool_dicts,
771
772
            chat_template_content_format,
            tokenizer,
773
            model_config=model_config,
774
        )
775
        conversation, mm_data_future, mm_uuids = parse_chat_messages_futures(
776
            messages,
777
            model_config,
778
            tokenizer,
779
            content_format=resolved_content_format,
780
781
        )

782
        _chat_template_kwargs: dict[str, Any] = dict(
783
784
785
786
787
788
789
790
            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 {})

791
        request_prompt: Union[str, list[int]]
792
793
794
795

        if tokenizer is None:
            request_prompt = "placeholder"
        elif isinstance(tokenizer, MistralTokenizer):
796
797
798
            request_prompt = apply_mistral_chat_template(
                tokenizer,
                messages=messages,
799
                **_chat_template_kwargs,
800
801
802
            )
        else:
            request_prompt = apply_hf_chat_template(
803
                tokenizer=tokenizer,
804
                conversation=conversation,
805
                model_config=model_config,
806
                **_chat_template_kwargs,
807
808
809
810
            )

        mm_data = await mm_data_future

811
812
813
814
815
816
817
        # 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:
818
819
820
821
            if not isinstance(request, ChatCompletionRequest):
                msg = "Tool usage is only supported for Chat Completions API"
                raise NotImplementedError(msg)

822
823
            request = tool_parser(tokenizer).adjust_request(  # type: ignore
                request=request)
824

825
826
        if tokenizer is None:
            assert isinstance(request_prompt, str), (
827
828
                "Prompt has to be a string",
                "when the tokenizer is not initialised",
829
830
831
832
            )
            prompt_inputs = TextTokensPrompt(prompt=request_prompt,
                                             prompt_token_ids=[1])
        elif isinstance(request_prompt, str):
833
            prompt_inputs = await self._tokenize_prompt_input_async(
834
835
836
837
838
839
840
841
842
843
844
                request,
                tokenizer,
                request_prompt,
                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),
845
846
                prompt_token_ids=request_prompt,
            )
847

848
        engine_prompt = EngineTokensPrompt(
849
850
851
            prompt_token_ids=prompt_inputs["prompt_token_ids"])
        if mm_data is not None:
            engine_prompt["multi_modal_data"] = mm_data
852
853
854
855

        if mm_uuids is not None:
            engine_prompt["multi_modal_uuids"] = mm_uuids

856
857
        if request.mm_processor_kwargs is not None:
            engine_prompt["mm_processor_kwargs"] = request.mm_processor_kwargs
858

859
860
861
        if hasattr(request, "cache_salt") and request.cache_salt is not None:
            engine_prompt["cache_salt"] = request.cache_salt

862
863
        return conversation, [request_prompt], [engine_prompt]

864
    async def _generate_with_builtin_tools(
865
        self,
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
        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.
914
915
            sampling_params.max_tokens = self.max_model_len - len(
                prompt_token_ids)
916
917
918
            # OPTIMIZATION
            priority = orig_priority - 1

919
920
921
    def _log_inputs(
        self,
        request_id: str,
922
        inputs: Union[RequestPrompt, PromptType],
923
924
        params: Optional[Union[SamplingParams, PoolingParams,
                               BeamSearchParams]],
925
926
927
928
        lora_request: Optional[LoRARequest],
    ) -> None:
        if self.request_logger is None:
            return
929
        prompt, prompt_token_ids, prompt_embeds = None, None, None
930
931
932
933
        if isinstance(inputs, str):
            prompt = inputs
        elif isinstance(inputs, list):
            prompt_token_ids = inputs
934
        else:
935
936
            prompt = getattr(inputs, 'prompt', None)
            prompt_token_ids = getattr(inputs, 'prompt_token_ids', None)
937
938
939
940
941

        self.request_logger.log_inputs(
            request_id,
            prompt,
            prompt_token_ids,
942
            prompt_embeds,
943
944
945
            params=params,
            lora_request=lora_request,
        )
946

947
948
949
950
951
952
953
954
955
956
957
958
959
960
    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

961
    @staticmethod
962
    def _base_request_id(raw_request: Optional[Request],
963
964
965
                         default: Optional[str] = None) -> Optional[str]:
        """Pulls the request id to use from a header, if provided"""
        default = default or random_uuid()
966
967
968
969
        if raw_request is None:
            return default

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

971
    @staticmethod
972
973
974
975
976
977
    def _get_decoded_token(
        logprob: Logprob,
        token_id: int,
        tokenizer: AnyTokenizer,
        return_as_token_id: bool = False,
    ) -> str:
978
979
980
        if return_as_token_id:
            return f"token_id:{token_id}"

981
982
        if logprob.decoded_token is not None:
            return logprob.decoded_token
983
        return tokenizer.decode(token_id)
984

985
    def _is_model_supported(self, model_name: Optional[str]) -> bool:
986
987
        if not model_name:
            return True
988
        return self.models.is_base_model(model_name)
989

990
991
992

def clamp_prompt_logprobs(
    prompt_logprobs: Union[PromptLogprobs,
993
                           None], ) -> Union[PromptLogprobs, None]:
994
995
996
997
998
999
1000
    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():
1001
            if logprob_values.logprob == float("-inf"):
1002
1003
                logprob_values.logprob = -9999.0
    return prompt_logprobs