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

4
import asyncio
5
import json
6
import time
7
8
9
from collections.abc import AsyncGenerator, AsyncIterator
from collections.abc import Sequence as GenericSequence
from typing import Callable, Final, Optional, Union
10

11
import jinja2
12
import partial_json_parser
13
import regex as re
14
from fastapi import Request
15
from pydantic import TypeAdapter
16

17
from vllm.config import ModelConfig
18
from vllm.engine.protocol import EngineClient
19
from vllm.entrypoints.chat_utils import (ChatTemplateContentFormatOption,
20
21
                                         ConversationMessage,
                                         random_tool_call_id)
22
from vllm.entrypoints.logger import RequestLogger
23
from vllm.entrypoints.openai.protocol import (
24
25
    ChatCompletionLogProb, ChatCompletionLogProbs,
    ChatCompletionLogProbsContent, ChatCompletionNamedToolChoiceParam,
26
    ChatCompletionRequest, ChatCompletionResponse,
27
    ChatCompletionResponseChoice, ChatCompletionResponseStreamChoice,
28
    ChatCompletionStreamResponse, ChatMessage, DeltaFunctionCall, DeltaMessage,
29
30
    DeltaToolCall, ErrorResponse, FunctionCall, FunctionDefinition,
    PromptTokenUsageInfo, RequestResponseMetadata, ToolCall, UsageInfo)
31
32
from vllm.entrypoints.openai.serving_engine import (OpenAIServing,
                                                    clamp_prompt_logprobs)
33
from vllm.entrypoints.openai.serving_models import OpenAIServingModels
34
from vllm.entrypoints.openai.tool_parsers import ToolParser, ToolParserManager
35
36
from vllm.entrypoints.openai.tool_parsers.mistral_tool_parser import (
    MistralToolCall)
37
from vllm.entrypoints.utils import get_max_tokens
38
from vllm.logger import init_logger
39
from vllm.outputs import CompletionOutput, RequestOutput
40
from vllm.reasoning import ReasoningParser, ReasoningParserManager
41
from vllm.sampling_params import BeamSearchParams, SamplingParams
42
from vllm.sequence import Logprob
43
from vllm.transformers_utils.tokenizer import AnyTokenizer, MistralTokenizer
44
from vllm.transformers_utils.tokenizers import (maybe_serialize_tool_calls,
45
46
                                                truncate_tool_call_ids,
                                                validate_request_params)
47
48
49
50
51
52

logger = init_logger(__name__)


class OpenAIServingChat(OpenAIServing):

53
54
55
56
    def __init__(
        self,
        engine_client: EngineClient,
        model_config: ModelConfig,
57
        models: OpenAIServingModels,
58
59
60
61
62
63
        response_role: str,
        *,
        request_logger: Optional[RequestLogger],
        chat_template: Optional[str],
        chat_template_content_format: ChatTemplateContentFormatOption,
        return_tokens_as_token_ids: bool = False,
64
        reasoning_parser: str = "",
65
        enable_auto_tools: bool = False,
66
        exclude_tools_when_tool_choice_none: bool = False,
67
68
        tool_parser: Optional[str] = None,
        enable_prompt_tokens_details: bool = False,
69
        enable_force_include_usage: bool = False,
70
    ) -> None:
71
        super().__init__(engine_client=engine_client,
72
                         model_config=model_config,
73
                         models=models,
74
                         request_logger=request_logger,
75
76
                         return_tokens_as_token_ids=return_tokens_as_token_ids,
                         enable_force_include_usage=enable_force_include_usage)
77

78
        self.response_role = response_role
79
80
        self.chat_template = chat_template
        self.chat_template_content_format: Final = chat_template_content_format
81

82
83
84
85
86
87
88
89
        # set up tool use
        self.enable_auto_tools: bool = enable_auto_tools
        if self.enable_auto_tools:
            logger.info(
                "\"auto\" tool choice has been enabled please note that while"
                " the parallel_tool_calls client option is preset for "
                "compatibility reasons, it will be ignored.")

90
91
        self.reasoning_parser: Optional[Callable[[AnyTokenizer],
                                                 ReasoningParser]] = None
92
        if reasoning_parser:
93
94
95
96
            try:
                self.reasoning_parser = (
                    ReasoningParserManager.get_reasoning_parser(
                        reasoning_parser))
97
                assert self.reasoning_parser is not None
98
            except Exception as e:
99
100
                raise TypeError(
                    f"{reasoning_parser=} has not been registered") from e
101
102
        self.tool_parser: Optional[Callable[[AnyTokenizer], ToolParser]] = None
        if self.enable_auto_tools:
103
            try:
104
105
106
107
108
                if (tool_parser == "pythonic" and
                        model_config.model.startswith("meta-llama/Llama-3.2")):
                    logger.warning(
                        "Llama3.2 models may struggle to emit valid pythonic"
                        " tool calls")
109
110
111
                self.tool_parser = ToolParserManager.get_tool_parser(
                    tool_parser)
            except Exception as e:
112
                raise TypeError("Error: --enable-auto-tool-choice requires "
113
114
                                f"tool_parser:'{tool_parser}' which has not "
                                "been registered") from e
115
116
        self.exclude_tools_when_tool_choice_none = (
            exclude_tools_when_tool_choice_none)
117

118
        self.enable_prompt_tokens_details = enable_prompt_tokens_details
119
        self.enable_force_include_usage = enable_force_include_usage
120
121
122
        self.default_sampling_params = (
            self.model_config.get_diff_sampling_param())
        if self.default_sampling_params:
123
124
125
126
            source = self.model_config.generation_config
            source = "model" if source == "auto" else source
            logger.info("Using default chat sampling params from %s: %s",
                        source, self.default_sampling_params)
127

128
    async def create_chat_completion(
129
130
        self,
        request: ChatCompletionRequest,
131
132
133
        raw_request: Optional[Request] = None,
    ) -> Union[AsyncGenerator[str, None], ChatCompletionResponse,
               ErrorResponse]:
134
135
        """
        Chat Completion API similar to OpenAI's API.
136

137
138
        See https://platform.openai.com/docs/api-reference/chat/create
        for the API specification. This API mimics the OpenAI
139
        Chat Completion API.
140
141
142
        """
        error_check_ret = await self._check_model(request)
        if error_check_ret is not None:
143
            logger.error("Error with model %s", error_check_ret)
144
145
            return error_check_ret

146
147
148
149
150
151
        # If the engine is dead, raise the engine's DEAD_ERROR.
        # This is required for the streaming case, where we return a
        # success status before we actually start generating text :).
        if self.engine_client.errored:
            raise self.engine_client.dead_error

152
        try:
153
154
            lora_request = self._maybe_get_adapters(
                request, supports_default_mm_loras=True)
155

156
            model_name = self._get_model_name(request.model, lora_request)
157

158
            tokenizer = await self.engine_client.get_tokenizer(lora_request)
159

160
161
            tool_parser = self.tool_parser

162
            if isinstance(tokenizer, MistralTokenizer):
163
164
165
                # because of issues with pydantic we need to potentially
                # re-serialize the tool_calls field of the request
                # for more info: see comment in `maybe_serialize_tool_calls`
166
                maybe_serialize_tool_calls(request)
167
                truncate_tool_call_ids(request)
168
                validate_request_params(request)
169

170
171
172
173
174
175
176
177
178
            if (request.tool_choice == "auto" and
                    not (self.enable_auto_tools and tool_parser is not None)
                    and not isinstance(tokenizer, MistralTokenizer)):
                # for hf tokenizers, "auto" tools requires
                # --enable-auto-tool-choice and --tool-call-parser
                return self.create_error_response(
                    "\"auto\" tool choice requires "
                    "--enable-auto-tool-choice and --tool-call-parser to be set"
                )
179

180
181
182
            if (request.tools is None
                    or (request.tool_choice == "none"
                        and self.exclude_tools_when_tool_choice_none)):
183
184
185
                tool_dicts = None
            else:
                tool_dicts = [tool.model_dump() for tool in request.tools]
186

187
188
189
190
191
192
193
194
195
            (
                conversation,
                request_prompts,
                engine_prompts,
            ) = await self._preprocess_chat(
                request,
                tokenizer,
                request.messages,
                chat_template=request.chat_template or self.chat_template,
196
                chat_template_content_format=self.chat_template_content_format,
197
198
199
200
201
202
203
204
205
                add_generation_prompt=request.add_generation_prompt,
                continue_final_message=request.continue_final_message,
                tool_dicts=tool_dicts,
                documents=request.documents,
                chat_template_kwargs=request.chat_template_kwargs,
                tool_parser=tool_parser,
                truncate_prompt_tokens=request.truncate_prompt_tokens,
                add_special_tokens=request.add_special_tokens,
            )
206
207
        except (ValueError, TypeError, RuntimeError,
                jinja2.TemplateError) as e:
208
            logger.exception("Error in preprocessing prompt inputs")
209
            return self.create_error_response(f"{e} {e.__cause__}")
210

211
212
        request_id = "chatcmpl-" \
                     f"{self._base_request_id(raw_request, request.request_id)}"
213
214
215
216
217

        request_metadata = RequestResponseMetadata(request_id=request_id)
        if raw_request:
            raw_request.state.request_metadata = request_metadata

218
        # Schedule the request and get the result generator.
219
        generators: list[AsyncGenerator[RequestOutput, None]] = []
220
        try:
221
222
            for i, engine_prompt in enumerate(engine_prompts):
                sampling_params: Union[SamplingParams, BeamSearchParams]
223
224
225
226
227
228
229
230
231
232

                if self.default_sampling_params is None:
                    self.default_sampling_params = {}

                max_tokens = get_max_tokens(
                    max_model_len=self.max_model_len,
                    request=request,
                    input_length=len(engine_prompt["prompt_token_ids"]),
                    default_sampling_params=self.default_sampling_params)

233
234
                if request.use_beam_search:
                    sampling_params = request.to_beam_search_params(
235
                        max_tokens, self.default_sampling_params)
236
237
                else:
                    sampling_params = request.to_sampling_params(
238
                        max_tokens, self.model_config.logits_processor_pattern,
239
                        self.default_sampling_params)
240
241
242
243

                self._log_inputs(request_id,
                                 request_prompts[i],
                                 params=sampling_params,
244
                                 lora_request=lora_request)
245
246
247
248
249
250
251
252
253

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

                if isinstance(sampling_params, BeamSearchParams):
                    generator = self.engine_client.beam_search(
                        prompt=engine_prompt,
                        request_id=request_id,
                        params=sampling_params,
254
                        lora_request=lora_request,
255
256
257
258
259
260
261
262
263
264
265
266
                    )
                else:
                    generator = self.engine_client.generate(
                        engine_prompt,
                        sampling_params,
                        request_id,
                        lora_request=lora_request,
                        trace_headers=trace_headers,
                        priority=request.priority,
                    )

                generators.append(generator)
267
        except ValueError as e:
268
            # TODO: Use a vllm-specific Validation Error
269
270
            return self.create_error_response(str(e))

271
272
273
        assert len(generators) == 1
        result_generator, = generators

274
275
276
        # Streaming response
        if request.stream:
            return self.chat_completion_stream_generator(
277
278
279
280
281
282
283
284
                request,
                result_generator,
                request_id,
                model_name,
                conversation,
                tokenizer,
                request_metadata,
                enable_force_include_usage=self.enable_force_include_usage)
285

286
287
        try:
            return await self.chat_completion_full_generator(
288
289
                request, result_generator, request_id, model_name,
                conversation, tokenizer, request_metadata)
290
291
292
        except ValueError as e:
            # TODO: Use a vllm-specific Validation Error
            return self.create_error_response(str(e))
293
294
295
296

    def get_chat_request_role(self, request: ChatCompletionRequest) -> str:
        if request.add_generation_prompt:
            return self.response_role
297
        return request.messages[-1]["role"]
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
    @staticmethod
    def _bracket_level(s: str, opening='{', closing='}') -> int:
        """
        Calculate the current level of nested brackets in a given string.
        """
        level = 0
        for char in s:
            if char == opening:
                level += 1
            elif char == closing:
                level -= 1
        return level

    @staticmethod
    def _filter_delta_text(delta_text: str,
                           previous_text: str) -> tuple[str, bool]:
        # remove last '},' of the tool definition stemming from the
        # "name"/"parameters" outer object or closing ']' of the tool list
        # count occurrences of opening and closing curly braces and
        # once level 0 is reached stop outputting text
        # if 0 is reached while parsing the delta_text we know the current
        # tool will finish in this current iteration
        bracket_level = OpenAIServingChat._bracket_level(previous_text)
        updated_delta, passed_zero = "", False
        for c in delta_text:
            if c == '{':
                bracket_level += 1
                passed_zero = bracket_level == 0
            elif c == '}':
                bracket_level -= 1
                passed_zero = bracket_level == 0

            if bracket_level != 0:
                updated_delta += c
            else:
                # if a comma is reached at level 0 we can stop
                if c == ',':
                    break
        return updated_delta, passed_zero

    def extract_tool_call_required_streaming(
        self,
        previous_text: str,
342
        current_text: Optional[str],
343
344
345
        delta_text: str,
        function_name_returned: bool,
    ) -> tuple[Optional[DeltaMessage], bool]:
346
347
348
        if current_text is None or current_text == "":
            # if the current text is empty, we cannot parse it
            return None, function_name_returned
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
        try:
            obj = partial_json_parser.loads(current_text)
        except partial_json_parser.core.exceptions.MalformedJSON:
            logger.debug('not enough tokens to parse into JSON yet')
            obj = None

        # check if the current text is a valid array
        # containing a partial tool calling object
        # if not repeat
        if obj is None or not isinstance(obj, list) or not len(obj) > 0:
            function_name_returned = False
            delta_message = None
        else:
            _, finishes_previous_tool = OpenAIServingChat._filter_delta_text(
                delta_text, previous_text)
            # take the last tool call from the generated list
            current_tool_call = obj[-1]

            # once parameters have been generated the name is complete as well
            if not finishes_previous_tool and ("name" not in current_tool_call
                                               or "parameters"
                                               not in current_tool_call):
                function_name_returned = False
                delta_message = None
            else:
                if not function_name_returned:
                    # get partly generated arguments from the latest tool call
                    param_match = re.search(r'.*"parameters":\s*(.*)',
                                            current_text)
                    arguments = param_match.group(1) if param_match else ""
                    arguments, _ = OpenAIServingChat._filter_delta_text(
                        arguments, previous_text)

                    # if this iteration finishes a previous tool call but a
                    # new incomplete tool is already generated, take the
                    # previous from the list
                    if (finishes_previous_tool
                            and "parameters" not in current_tool_call):
                        current_tool_call = obj[-2]

                    function_name_returned = True
                    delta_message = DeltaMessage(tool_calls=[
391
392
393
394
                        DeltaToolCall(id=random_tool_call_id(),
                                      function=DeltaFunctionCall(
                                          name=current_tool_call["name"],
                                          arguments=arguments),
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
                                      index=len(obj) - 1,
                                      type="function")
                    ])

                else:
                    delta_text, _ = OpenAIServingChat._filter_delta_text(
                        delta_text, previous_text)

                    if delta_text != "":
                        delta_message = DeltaMessage(tool_calls=[
                            DeltaToolCall(
                                function=DeltaFunctionCall(
                                    # OpenAI API returns None
                                    # instead of name every time
                                    name=None,
                                    arguments=delta_text),
411
                                index=len(obj) - 1)
412
413
414
415
416
417
                        ])
                    else:
                        delta_message = None

        return delta_message, function_name_returned

418
    async def chat_completion_stream_generator(
419
420
421
422
        self,
        request: ChatCompletionRequest,
        result_generator: AsyncIterator[RequestOutput],
        request_id: str,
423
        model_name: str,
424
        conversation: list[ConversationMessage],
425
        tokenizer: AnyTokenizer,
426
        request_metadata: RequestResponseMetadata,
427
        enable_force_include_usage: bool,
428
    ) -> AsyncGenerator[str, None]:
429
        created_time = int(time.time())
430
        chunk_object_type: Final = "chat.completion.chunk"
431
        first_iteration = True
432
433

        # Send response for each token for each request.n (index)
434
435
436
        num_choices = 1 if request.n is None else request.n
        previous_num_tokens = [0] * num_choices
        finish_reason_sent = [False] * num_choices
437
        num_prompt_tokens = 0
438
        num_cached_tokens = None
439
440
441
442
443
444
445
446
447
448
449

        if isinstance(request.tool_choice, ChatCompletionNamedToolChoiceParam):
            tool_choice_function_name = request.tool_choice.function.name
        else:
            tool_choice_function_name = None

        # Determine whether tools are in use with "auto" tool choice
        tool_choice_auto = (
            not tool_choice_function_name
            and self._should_stream_with_auto_tool_parsing(request))

450
        all_previous_token_ids: Optional[list[list[int]]]
451
        function_name_returned = [False] * num_choices
452
453
454

        # Only one of these will be used, thus previous_texts and
        # all_previous_token_ids will not be used twice in the same iteration.
455
        if tool_choice_auto or self.reasoning_parser:
456
457
458
            # These are only required in "auto" tool choice case
            previous_texts = [""] * num_choices
            all_previous_token_ids = [[]] * num_choices
459
460
461
            # For reasoning parser and tool call all enabled
            added_content_delta_arr = [False] * num_choices
            reasoning_end_arr = [False] * num_choices
462
463
464
        elif request.tool_choice == "required":
            previous_texts = [""] * num_choices
            all_previous_token_ids = None
465
466
467
        else:
            previous_texts, all_previous_token_ids = None, None

468
        try:
469
            if self.reasoning_parser:
470
471
472
473
474
475
476
                reasoning_parser = self.reasoning_parser(tokenizer)
        except RuntimeError as e:
            logger.exception("Error in reasoning parser creation.")
            data = self.create_streaming_error_response(str(e))
            yield f"data: {data}\n\n"
            yield "data: [DONE]\n\n"
            return
477
478
479
        # Prepare the tool parser if it's needed
        try:
            if tool_choice_auto and self.tool_parser:
480
                tool_parsers: list[Optional[ToolParser]] = [
481
482
483
484
                    self.tool_parser(tokenizer)
                ] * num_choices
            else:
                tool_parsers = [None] * num_choices
485
        except Exception as e:
486
            logger.exception("Error in tool parser creation.")
487
488
489
490
491
            data = self.create_streaming_error_response(str(e))
            yield f"data: {data}\n\n"
            yield "data: [DONE]\n\n"
            return

492
493
        stream_options = request.stream_options
        if stream_options:
494
495
            include_usage = stream_options.include_usage \
                            or enable_force_include_usage
496
497
498
499
500
            include_continuous_usage = include_usage and \
                                       stream_options.continuous_usage_stats
        else:
            include_usage, include_continuous_usage = False, False

501
502
        try:
            async for res in result_generator:
503
504
                if res.prompt_token_ids is not None:
                    num_prompt_tokens = len(res.prompt_token_ids)
505
506
                    if res.encoder_prompt_token_ids is not None:
                        num_prompt_tokens += len(res.encoder_prompt_token_ids)
507

508
509
510
511
                # We need to do it here, because if there are exceptions in
                # the result_generator, it needs to be sent as the FIRST
                # response (by the try...catch).
                if first_iteration:
512
                    num_cached_tokens = res.num_cached_tokens
513
514
                    # Send first response for each request.n (index) with
                    # the role
515
                    role = self.get_chat_request_role(request)
516
517
518

                    # NOTE num_choices defaults to 1 so this usually executes
                    # once per request
519
                    for i in range(num_choices):
520
521
                        choice_data = ChatCompletionResponseStreamChoice(
                            index=i,
522
523
524
525
                            delta=DeltaMessage(
                                role=role,
                                content="",
                            ),
526
527
528
529
530
531
532
533
                            logprobs=None,
                            finish_reason=None)
                        chunk = ChatCompletionStreamResponse(
                            id=request_id,
                            object=chunk_object_type,
                            created=created_time,
                            choices=[choice_data],
                            model=model_name)
534

535
536
537
538
539
540
                        # if continuous usage stats are requested, add it
                        if include_continuous_usage:
                            chunk.usage = UsageInfo(
                                prompt_tokens=num_prompt_tokens,
                                completion_tokens=0,
                                total_tokens=num_prompt_tokens)
541

542
543
544
                        data = chunk.model_dump_json(exclude_unset=True)
                        yield f"data: {data}\n\n"

545
546
                    # Send response to echo the input portion of the
                    # last message
547
                    if request.echo:
548
                        last_msg_content: Union[str, list[dict[str, str]]] = ""
549
550
551
                        if conversation and "content" in conversation[
                                -1] and conversation[-1].get("role") == role:
                            last_msg_content = conversation[-1]["content"] or ""
552
553

                        if last_msg_content:
554
                            for i in range(num_choices):
555
556
557
558
559
                                choice_data = (
                                    ChatCompletionResponseStreamChoice(
                                        index=i,
                                        delta=DeltaMessage(
                                            content=last_msg_content),
560
                                        logprobs=None,
561
                                        finish_reason=None))
562
563
564
565
566
567
                                chunk = ChatCompletionStreamResponse(
                                    id=request_id,
                                    object=chunk_object_type,
                                    created=created_time,
                                    choices=[choice_data],
                                    model=model_name)
568
569
570
571
572
                                if include_continuous_usage:
                                    chunk.usage = UsageInfo(
                                        prompt_tokens=num_prompt_tokens,
                                        completion_tokens=0,
                                        total_tokens=num_prompt_tokens)
573

574
575
576
577
578
579
580
                                data = chunk.model_dump_json(
                                    exclude_unset=True)
                                yield f"data: {data}\n\n"
                    first_iteration = False

                for output in res.outputs:
                    i = output.index
581
                    tool_parser = tool_parsers[i]
582
583
584
585

                    if finish_reason_sent[i]:
                        continue

586
                    if request.logprobs and request.top_logprobs is not None:
587
                        assert output.logprobs is not None, (
588
                            "Did not output logprobs")
589
                        logprobs = self._create_chat_logprobs(
590
591
                            token_ids=output.token_ids,
                            top_logprobs=output.logprobs,
592
                            tokenizer=tokenizer,
593
                            num_output_top_logprobs=request.top_logprobs,
594
595
                            return_as_token_id=request.
                            return_tokens_as_token_ids,
596
597
598
599
                        )
                    else:
                        logprobs = None

600
                    delta_text = output.text
601
602
603
604
605
606

                    if not delta_text and not output.token_ids and \
                        not previous_num_tokens[i]:
                        # Chunked prefill case, don't return empty chunks
                        continue

607
                    delta_message: Optional[DeltaMessage]
608

609
                    # just update previous_texts and previous_token_ids
610
                    if tool_choice_auto or self.reasoning_parser:
611
612
613
614
615
                        assert previous_texts is not None
                        assert all_previous_token_ids is not None
                        previous_text = previous_texts[i]
                        previous_token_ids = all_previous_token_ids[i]
                        current_text = previous_text + delta_text
616
617
618
619
620
621
622

                        # avoid the None + list error.
                        if previous_token_ids:
                            current_token_ids = previous_token_ids + list(
                                output.token_ids)
                        else:
                            current_token_ids = list(output.token_ids)
623

624
625
                    # handle streaming deltas for tools with named tool_choice
                    if tool_choice_function_name:
626
                        if (self.reasoning_parser and not reasoning_end_arr[i]
627
628
629
630
631
632
633
634
635
636
637
638
639
                                and not reasoning_parser.is_reasoning_end(
                                    previous_token_ids)):
                            assert reasoning_parser is not None
                            delta_message = (
                                reasoning_parser.
                                extract_reasoning_content_streaming(
                                    previous_text,
                                    current_text,
                                    delta_text,
                                    previous_token_ids,
                                    current_token_ids,
                                    output.token_ids,
                                ))
640
641
642
643
644
                            # When encountering think end id in delta_token_ids
                            # or think end id in prompt_token_ids
                            # i.e {"enable_thinking": False},
                            # set reasoning status to end.
                            # Only keep 'content', remove 'reasoning_content'.
645
                            if reasoning_parser.is_reasoning_end(
646
647
648
649
650
651
                                    list(output.token_ids)) or \
                                    (res.prompt_token_ids and
                                        reasoning_parser.is_reasoning_end(
                                            list(res.prompt_token_ids)
                                        )):
                                reasoning_end_arr[i] = True
652
653
654
655
656
657
658
659
                                if delta_message and delta_message.content:
                                    # This need to be added to next `delta_text`
                                    current_text = delta_message.content
                                    delta_message.content = None
                                else:
                                    current_text = ""
                        else:
                            # Just to add remaining `content`
660
                            if self.reasoning_parser:
661
662
663
                                delta_text = previous_text + delta_text
                                current_text = ""

664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
                            if function_name_returned[i]:
                                delta_tool_call = DeltaToolCall(
                                    function=DeltaFunctionCall(
                                        arguments=delta_text),
                                    index=i)
                            else:
                                delta_tool_call = DeltaToolCall(
                                    id=random_tool_call_id(),
                                    type="function",
                                    function=DeltaFunctionCall(
                                        name=tool_choice_function_name,
                                        arguments=delta_text),
                                    index=i)
                                function_name_returned[i] = True

679
                            delta_message = DeltaMessage(tool_calls=[
680
                                delta_tool_call,
681
682
                            ])

683
684
685
686
687
688
                    elif request.tool_choice == "required":
                        assert previous_texts is not None
                        previous_text = previous_texts[i]
                        current_text = previous_text + delta_text
                        fn_name_returned = function_name_returned[i]

689
690
691
692
693
694
695
696
                        if self.reasoning_parser:
                            _, content = \
                                reasoning_parser.extract_reasoning_content(
                                    current_text,
                                    request
                                )
                        else:
                            content = current_text
697
698
699
                        delta_message, function_name_returned[i] = (
                            self.extract_tool_call_required_streaming(
                                previous_text=previous_text,
700
                                current_text=content,
701
702
703
704
705
706
                                delta_text=delta_text,
                                function_name_returned=fn_name_returned))

                        # update the previous values for the next iteration
                        previous_texts[i] = current_text

707
708
                    # handle streaming deltas for tools with "auto" tool choice
                    # and reasoning parser
709
                    elif tool_choice_auto and self.reasoning_parser:
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
                        assert tool_parser is not None
                        assert reasoning_parser is not None
                        assert added_content_delta_arr is not None
                        assert reasoning_end_arr is not None
                        if not reasoning_end_arr[i]:
                            delta_message = (
                                reasoning_parser.
                                extract_reasoning_content_streaming(
                                    previous_text,
                                    current_text,
                                    delta_text,
                                    previous_token_ids,
                                    current_token_ids,
                                    output.token_ids,
                                ))
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
                            # When encountering think end id in prompt_token_ids
                            # i.e {"enable_thinking": False},
                            # set reasoning status to end.
                            # Remove the text and token ids related
                            # to 'reasoning_content'.
                            if res.prompt_token_ids and \
                                reasoning_parser.is_reasoning_end(
                                    list(res.prompt_token_ids)):
                                reasoning_end_arr[i] = True
                                current_token_ids = list(output.token_ids)
                                if delta_message and delta_message.content:
                                    current_text = delta_message.content
                                    delta_message.content = None
                                else:
                                    current_text = ""
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
                            # When encountering think end id in delta_token_ids,
                            # set reasoning status to end.
                            # Remove the text and token ids related
                            # to 'reasoning_content'.
                            if reasoning_parser.is_reasoning_end(
                                    list(output.token_ids)):
                                reasoning_end_arr[i] = True
                                current_token_ids =  \
                                    reasoning_parser.extract_content_ids(
                                        list(output.token_ids))
                                if delta_message and delta_message.content:
                                    current_text = delta_message.content
                                    delta_message.content = None
                                else:
                                    current_text = ""

                        # handle tool calls only after reasoning is done,
                        else:
                            delta_token_ids = list(output.token_ids)
                            # First time to tool call,
                            # add the remaining text and token ids
                            # to delta from previous
                            if not added_content_delta_arr[i]:
                                added_content_delta_arr[i] = True
                                previous_text = ""
                                previous_token_ids = []
                                delta_text = current_text
                                delta_token_ids = current_token_ids

                            delta_message = (
                                tool_parser.extract_tool_calls_streaming(
                                    previous_text=previous_text,
                                    current_text=current_text,
                                    delta_text=delta_text,
                                    previous_token_ids=previous_token_ids,
                                    current_token_ids=current_token_ids,
                                    delta_token_ids=delta_token_ids,
                                    request=request))
                    # when only tool calls
                    elif tool_choice_auto:
                        assert tool_parser is not None
781
782
                        delta_message = (
                            tool_parser.extract_tool_calls_streaming(
783
784
                                previous_text=previous_text,
                                current_text=current_text,
785
                                delta_text=delta_text,
786
787
                                previous_token_ids=previous_token_ids,
                                current_token_ids=current_token_ids,
788
789
                                delta_token_ids=output.token_ids,
                                request=request))
790
                    # when only reasoning
791
                    elif self.reasoning_parser:
792
793
794
795
796
797
798
799
800
                        delta_message = (reasoning_parser.
                                         extract_reasoning_content_streaming(
                                             previous_text,
                                             current_text,
                                             delta_text,
                                             previous_token_ids,
                                             current_token_ids,
                                             output.token_ids,
                                         ))
801
                    # handle streaming just a content delta
802
803
804
                    else:
                        delta_message = DeltaMessage(content=delta_text)

805
                    # update the previous values for the next iteration
806
                    if tool_choice_auto or self.reasoning_parser:
807
808
809
810
811
                        assert previous_texts is not None
                        assert all_previous_token_ids is not None
                        previous_texts[i] = current_text
                        all_previous_token_ids[i] = current_token_ids

812
                    # set the previous values for the next iteration
813
                    previous_num_tokens[i] += len(output.token_ids)
814
815
816
817
818
819
820
821

                    # if the message delta is None (e.g. because it was a
                    # "control token" for tool calls or the parser otherwise
                    # wasn't ready to send a token, then
                    #   get the next token without streaming a chunk
                    if delta_message is None:
                        continue

822
823
824
825
                    if output.finish_reason is None:
                        # Send token-by-token response for each request.n
                        choice_data = ChatCompletionResponseStreamChoice(
                            index=i,
826
                            delta=delta_message,
827
828
                            logprobs=logprobs,
                            finish_reason=None)
829
830

                    # if the model is finished generating
831
                    else:
832
833
834
835
                        # check to make sure we haven't "forgotten" to stream
                        #   any tokens that were generated but previously
                        #   matched by partial json parsing
                        # only happens if we are NOT using guided decoding
836
                        auto_tools_called = False
837
                        if tool_parser:
838
839
840
841
                            auto_tools_called = len(
                                tool_parser.prev_tool_call_arr) > 0
                            index = len(tool_parser.prev_tool_call_arr
                                        ) - 1 if auto_tools_called else 0
842
843
844
845
846
                        else:
                            index = 0

                        if self._should_check_for_unstreamed_tool_arg_tokens(
                                delta_message, output) and tool_parser:
847
848
849
850
851
852
853
854
855
856
                            latest_delta_len = 0
                            if ((isinstance(
                                    delta_message.tool_calls[0].function,
                                    DeltaFunctionCall)) and isinstance(
                                        delta_message.tool_calls[0].function.
                                        arguments, str)):
                                latest_delta_len = len(
                                    delta_message.tool_calls[0].function.
                                    arguments)

857
858
859
860
                            # get the expected call based on partial JSON
                            # parsing which "autocompletes" the JSON
                            expected_call = json.dumps(
                                tool_parser.prev_tool_call_arr[index].get(
861
862
                                    "arguments", {}),
                                ensure_ascii=False)
863

864
                            # get what we've streamed so far for arguments
865
866
867
                            # for the current tool
                            actual_call = tool_parser.streamed_args_for_tool[
                                index]
868
869
                            if (latest_delta_len > 0):
                                actual_call = actual_call[:-latest_delta_len]
870
871
872
873
874
875
876
877
878
879
880
881

                            # check to see if there's anything left to stream
                            remaining_call = expected_call.replace(
                                actual_call, "", 1)
                            # set that as a delta message
                            delta_message = DeltaMessage(tool_calls=[
                                DeltaToolCall(index=index,
                                              function=DeltaFunctionCall(
                                                  arguments=remaining_call).
                                              model_dump(exclude_none=True))
                            ])

882
883
884
                        # Send the finish response for each request.n only once
                        choice_data = ChatCompletionResponseStreamChoice(
                            index=i,
885
                            delta=delta_message,
886
                            logprobs=logprobs,
887
                            finish_reason=output.finish_reason
888
                            if not auto_tools_called else "tool_calls",
889
                            stop_reason=output.stop_reason)
890

891
                        finish_reason_sent[i] = True
892

893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
                    chunk = ChatCompletionStreamResponse(
                        id=request_id,
                        object=chunk_object_type,
                        created=created_time,
                        choices=[choice_data],
                        model=model_name)

                    # handle usage stats if requested & if continuous
                    if include_continuous_usage:
                        completion_tokens = previous_num_tokens[i]
                        chunk.usage = UsageInfo(
                            prompt_tokens=num_prompt_tokens,
                            completion_tokens=completion_tokens,
                            total_tokens=num_prompt_tokens + completion_tokens,
                        )

909
                    data = chunk.model_dump_json(exclude_unset=True)
910
911
                    yield f"data: {data}\n\n"

912
913
            # once the final token is handled, if stream_options.include_usage
            # is sent, send the usage
914
915
            if include_usage:
                completion_tokens = sum(previous_num_tokens)
916
917
918
919
920
921
922
                final_usage = UsageInfo(prompt_tokens=num_prompt_tokens,
                                        completion_tokens=completion_tokens,
                                        total_tokens=num_prompt_tokens +
                                        completion_tokens)
                if self.enable_prompt_tokens_details and num_cached_tokens:
                    final_usage.prompt_tokens_details = PromptTokenUsageInfo(
                        cached_tokens=num_cached_tokens)
923
924
925
926
927
928
929
930
931
932
933

                final_usage_chunk = ChatCompletionStreamResponse(
                    id=request_id,
                    object=chunk_object_type,
                    created=created_time,
                    choices=[],
                    model=model_name,
                    usage=final_usage)
                final_usage_data = (final_usage_chunk.model_dump_json(
                    exclude_unset=True, exclude_none=True))
                yield f"data: {final_usage_data}\n\n"
934

935
936
937
938
939
940
941
            # report to FastAPI middleware aggregate usage across all choices
            num_completion_tokens = sum(previous_num_tokens)
            request_metadata.final_usage_info = UsageInfo(
                prompt_tokens=num_prompt_tokens,
                completion_tokens=num_completion_tokens,
                total_tokens=num_prompt_tokens + num_completion_tokens)

942
        except Exception as e:
943
            # TODO: Use a vllm-specific Validation Error
944
            logger.exception("Error in chat completion stream generator.")
945
946
            data = self.create_streaming_error_response(str(e))
            yield f"data: {data}\n\n"
947
948
949
950
        # Send the final done message after all response.n are finished
        yield "data: [DONE]\n\n"

    async def chat_completion_full_generator(
951
952
953
954
        self,
        request: ChatCompletionRequest,
        result_generator: AsyncIterator[RequestOutput],
        request_id: str,
955
        model_name: str,
956
        conversation: list[ConversationMessage],
957
        tokenizer: AnyTokenizer,
958
        request_metadata: RequestResponseMetadata,
959
    ) -> Union[ErrorResponse, ChatCompletionResponse]:
960

961
        created_time = int(time.time())
962
        final_res: Optional[RequestOutput] = None
963

964
965
966
967
968
        try:
            async for res in result_generator:
                final_res = res
        except asyncio.CancelledError:
            return self.create_error_response("Client disconnected")
969
970
971
        except ValueError as e:
            # TODO: Use a vllm-specific Validation Error
            return self.create_error_response(str(e))
972

973
974
        assert final_res is not None

975
        choices: list[ChatCompletionResponseChoice] = []
976

977
978
        role = self.get_chat_request_role(request)
        for output in final_res.outputs:
979
            token_ids = output.token_ids
980
            out_logprobs = output.logprobs
981

982
983
            if request.logprobs and request.top_logprobs is not None:
                assert out_logprobs is not None, "Did not output logprobs"
984
                logprobs = self._create_chat_logprobs(
985
                    token_ids=token_ids,
986
                    top_logprobs=out_logprobs,
987
                    num_output_top_logprobs=request.top_logprobs,
988
                    tokenizer=tokenizer,
989
                    return_as_token_id=request.return_tokens_as_token_ids,
990
991
992
                )
            else:
                logprobs = None
993
            auto_tools_called = False
994

995
            if self.reasoning_parser:
996
997
998
999
1000
                try:
                    reasoning_parser = self.reasoning_parser(tokenizer)
                except RuntimeError as e:
                    logger.exception("Error in reasoning parser creation.")
                    return self.create_error_response(str(e))
1001
1002
                # If the reasoning parser is enabled,
                # tool calls are extracted exclusively from the content.
1003
1004
1005
                reasoning_content, content = (
                    reasoning_parser.extract_reasoning_content(
                        output.text, request=request))
1006
1007
1008
            else:
                reasoning_content = None
                content = output.text
1009

1010
1011
            # if auto tools are not enabled, and a named tool choice using
            #   outlines is not being used
1012
1013
1014
1015
            if (not self.enable_auto_tools or not self.tool_parser) and \
                (not isinstance(request.tool_choice,
                                ChatCompletionNamedToolChoiceParam
                                ) and request.tool_choice != "required"):
1016
1017
1018
                message = ChatMessage(role=role,
                                      reasoning_content=reasoning_content,
                                      content=content)
1019
1020
1021

            # if the request uses tools and specified a tool choice
            elif request.tool_choice and type(
1022
                    request.tool_choice) is ChatCompletionNamedToolChoiceParam:
1023

1024
1025
                tool_call_class = MistralToolCall if isinstance(
                    tokenizer, MistralTokenizer) else ToolCall
1026
1027
                message = ChatMessage(
                    role=role,
1028
                    reasoning_content=reasoning_content,
1029
1030
                    content="",
                    tool_calls=[
1031
                        tool_call_class(function=FunctionCall(
1032
                            name=request.tool_choice.function.name,
1033
                            arguments=content))
1034
                    ])
1035

1036
1037
1038
1039
1040
1041
            elif request.tool_choice and request.tool_choice == "required":
                tool_call_class = MistralToolCall if isinstance(
                    tokenizer, MistralTokenizer) else ToolCall

                # the fields of FunctionDefinition are a superset of the
                # tool call outputs and can be used for parsing
1042
                assert content is not None
1043
                tool_calls = TypeAdapter(
1044
                    list[FunctionDefinition]).validate_json(content)
1045
1046
1047
                message = ChatMessage(
                    role=role,
                    content="",
1048
                    reasoning_content=reasoning_content,
1049
1050
1051
                    tool_calls=[
                        tool_call_class(function=FunctionCall(
                            name=tool_call.name,
1052
1053
                            arguments=json.dumps(tool_call.parameters,
                                                 ensure_ascii=False)))
1054
1055
1056
                        for tool_call in tool_calls
                    ])

1057
1058
            # if the request doesn't use tool choice
            # OR specifies to not use a tool
1059
            elif not request.tool_choice or request.tool_choice == "none":
1060

1061
1062
1063
                message = ChatMessage(role=role,
                                      reasoning_content=reasoning_content,
                                      content=content)
1064
1065
1066
1067
1068
1069
1070

            # handle when there are tools and tool choice is auto
            elif request.tools and (
                    request.tool_choice == "auto"
                    or request.tool_choice is None) and self.enable_auto_tools \
                    and self.tool_parser:

1071
1072
1073
                try:
                    tool_parser = self.tool_parser(tokenizer)
                except RuntimeError as e:
1074
                    logger.exception("Error in tool parser creation.")
1075
1076
                    return self.create_error_response(str(e))

1077
                tool_call_info = tool_parser.extract_tool_calls(
1078
                    content if content is not None else "", request=request)
1079
1080
1081
1082
                # In the OpenAI API the finish_reason is "tools_called"
                # if the tool choice is auto and the model produced a tool
                # call. The same is not true for named function calls
                auto_tools_called = tool_call_info.tools_called
1083
1084
                if tool_call_info.tools_called:
                    message = ChatMessage(role=role,
1085
                                          reasoning_content=reasoning_content,
1086
1087
1088
1089
1090
1091
                                          content=tool_call_info.content,
                                          tool_calls=tool_call_info.tool_calls)

                else:
                    # FOR NOW make it a chat message; we will have to detect
                    # the type to make it later.
1092
1093
1094
1095
1096
1097
1098
1099
                    ret_content = content

                    # try to use content return from tool parser first,
                    # tool parser may do some modify for the content.
                    if (tool_call_info.content
                            and len(tool_call_info.content) > 0):
                        ret_content = tool_call_info.content

1100
1101
                    message = ChatMessage(role=role,
                                          reasoning_content=reasoning_content,
1102
                                          content=ret_content)
1103
1104
1105
1106
1107
1108
1109

            # undetermined case that is still important to handle
            else:
                logger.error(
                    "Error in chat_completion_full_generator - cannot determine"
                    " if tools should be extracted. Returning a standard chat "
                    "completion.")
1110
1111
1112
                message = ChatMessage(role=role,
                                      reasoning_content=reasoning_content,
                                      content=content)
1113

1114
1115
            choice_data = ChatCompletionResponseChoice(
                index=output.index,
1116
                message=message,
1117
                logprobs=logprobs,
1118
                finish_reason="tool_calls" if auto_tools_called else
1119
                output.finish_reason if output.finish_reason else "stop",
1120
                stop_reason=output.stop_reason)
1121
1122
            choices.append(choice_data)

1123
        if request.echo:
1124
            last_msg_content: Union[str, list[dict[str, str]]] = ""
1125
1126
            if conversation and "content" in conversation[-1] and conversation[
                    -1].get("role") == role:
1127
                last_msg_content = conversation[-1]["content"] or ""
1128
1129
1130
            if isinstance(last_msg_content, list):
                last_msg_content = "\n".join(msg['text']
                                             for msg in last_msg_content)
1131
1132

            for choice in choices:
1133
1134
                full_message = last_msg_content + (choice.message.content
                                                   or "")
1135
1136
                choice.message.content = full_message

1137
        assert final_res.prompt_token_ids is not None
1138
        num_prompt_tokens = len(final_res.prompt_token_ids)
1139
1140
        if final_res.encoder_prompt_token_ids is not None:
            num_prompt_tokens += len(final_res.encoder_prompt_token_ids)
1141
1142
        num_generated_tokens = sum(
            len(output.token_ids) for output in final_res.outputs)
1143
1144
1145
1146
1147
1148
1149
        usage = UsageInfo(prompt_tokens=num_prompt_tokens,
                          completion_tokens=num_generated_tokens,
                          total_tokens=num_prompt_tokens +
                          num_generated_tokens)
        if self.enable_prompt_tokens_details and final_res.num_cached_tokens:
            usage.prompt_tokens_details = PromptTokenUsageInfo(
                cached_tokens=final_res.num_cached_tokens)
1150
1151
1152

        request_metadata.final_usage_info = usage

1153
1154
1155
1156
1157
1158
        response = ChatCompletionResponse(
            id=request_id,
            created=created_time,
            model=model_name,
            choices=choices,
            usage=usage,
1159
            prompt_logprobs=clamp_prompt_logprobs(final_res.prompt_logprobs),
Robert Shaw's avatar
Robert Shaw committed
1160
            kv_transfer_params=final_res.kv_transfer_params,
1161
1162
        )

1163
        return response
1164
1165

    def _get_top_logprobs(
1166
            self, logprobs: dict[int, Logprob], top_logprobs: Optional[int],
1167
1168
            tokenizer: AnyTokenizer,
            should_return_as_token_id: bool) -> list[ChatCompletionLogProb]:
1169
        return [
1170
1171
1172
1173
            ChatCompletionLogProb(token=(token := self._get_decoded_token(
                p[1],
                p[0],
                tokenizer,
1174
                return_as_token_id=should_return_as_token_id)),
1175
1176
1177
                                  logprob=max(p[1].logprob, -9999.0),
                                  bytes=list(
                                      token.encode("utf-8", errors="replace")))
1178
1179
1180
1181
1182
1183
1184
            for i, p in enumerate(logprobs.items())
            if top_logprobs and i < top_logprobs
        ]

    def _create_chat_logprobs(
        self,
        token_ids: GenericSequence[int],
1185
        top_logprobs: GenericSequence[Optional[dict[int, Logprob]]],
1186
        tokenizer: AnyTokenizer,
1187
        num_output_top_logprobs: Optional[int] = None,
1188
        return_as_token_id: Optional[bool] = None,
1189
1190
    ) -> ChatCompletionLogProbs:
        """Create OpenAI-style logprobs."""
1191
        logprobs_content: list[ChatCompletionLogProbsContent] = []
1192

1193
1194
        should_return_as_token_id = return_as_token_id if \
            return_as_token_id is not None else self.return_tokens_as_token_ids
1195
1196
        for i, token_id in enumerate(token_ids):
            step_top_logprobs = top_logprobs[i]
1197
1198
            if step_top_logprobs is None or step_top_logprobs.get(
                    token_id) is None:
1199
                token = tokenizer.decode(token_id)
1200
                if should_return_as_token_id:
1201
                    token = f"token_id:{token_id}"
1202

1203
1204
                logprobs_content.append(
                    ChatCompletionLogProbsContent(
1205
                        token=token,
1206
1207
                        bytes=list(token.encode("utf-8", errors="replace")),
                    ))
1208
            else:
1209
1210
1211
                step_token = step_top_logprobs[token_id]
                step_decoded = step_token.decoded_token

1212
1213
                logprobs_content.append(
                    ChatCompletionLogProbsContent(
1214
                        token=self._get_decoded_token(
1215
1216
1217
                            step_token,
                            token_id,
                            tokenizer,
1218
                            should_return_as_token_id,
1219
1220
1221
1222
                        ),
                        logprob=max(step_token.logprob, -9999.0),
                        bytes=None if step_decoded is None else list(
                            step_decoded.encode("utf-8", errors="replace")),
1223
                        top_logprobs=self._get_top_logprobs(
1224
1225
                            step_top_logprobs, num_output_top_logprobs,
                            tokenizer, should_return_as_token_id),
1226
                    ))
1227
1228

        return ChatCompletionLogProbs(content=logprobs_content)
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257

    def _should_stream_with_auto_tool_parsing(self,
                                              request: ChatCompletionRequest):
        """
        Utility function to check if streamed tokens should go through the tool
        call parser that was configured.

        We only want to do this IF user-provided tools are set, a tool parser
        is configured, "auto" tool choice is enabled, and the request's tool
        choice field indicates that "auto" tool choice should be used.
        """
        return (request.tools and self.tool_parser and self.enable_auto_tools
                and request.tool_choice in ['auto', None])

    def _should_check_for_unstreamed_tool_arg_tokens(
        self,
        delta_message: Optional[DeltaMessage],
        output: CompletionOutput,
    ) -> bool:
        """
        Check to see if we should check for unstreamed tool arguments tokens.
        This is only applicable when auto tool parsing is enabled, the delta
        is a tool call with arguments.
        """

        # yapf: disable
        return bool(
            # if there is a delta message that includes tool calls which
            # include a function that has arguments
1258
1259
            output.finish_reason is not None
            and self.enable_auto_tools and self.tool_parser and delta_message
1260
1261
1262
1263
            and delta_message.tool_calls and delta_message.tool_calls[0]
            and delta_message.tool_calls[0].function
            and delta_message.tool_calls[0].function.arguments is not None
        )