serving_engine.py 36.2 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
232
233
234
        self.tokenizer_mode = model_config.tokenizer_mode
        
        if model_config.tokenizer_mode == "cpm":
            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
383

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

384
385
                self._log_inputs(
                    request_id_item,
386
                    engine_prompt,
387
388
389
                    params=pooling_params,
                    lora_request=ctx.lora_request,
                )
390
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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

        input_text = prompt

        return self._validate_input(request, input_ids, input_text)

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

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

620
621
622
        if tokenizer is None:
            input_text = ""
        else:
623
624
            async_tokenizer = self._get_async_tokenizer(tokenizer) 
            input_text = await async_tokenizer.decode(input_ids) if self.tokenizer_mode != "cpm" else await self.tokenizer.decode_all(input_ids)
625

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

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

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

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

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

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

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

        return TextTokensPrompt(prompt=input_text, prompt_token_ids=input_ids)

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

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

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

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

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

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

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

        mm_data = await mm_data_future

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

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

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

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

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

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

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

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

863
    async def _generate_with_builtin_tools(
864
        self,
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
        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.
913
914
            sampling_params.max_tokens = self.max_model_len - len(
                prompt_token_ids)
915
916
917
            # OPTIMIZATION
            priority = orig_priority - 1

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

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

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

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

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

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

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

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

989
990
991

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