serving_chat.py 56.9 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.logger import init_logger
38
from vllm.outputs import CompletionOutput, RequestOutput
39
from vllm.reasoning import ReasoningParser, ReasoningParserManager
40
from vllm.sampling_params import BeamSearchParams, SamplingParams
41
from vllm.sequence import Logprob
42
from vllm.transformers_utils.tokenizer import AnyTokenizer, MistralTokenizer
43
from vllm.transformers_utils.tokenizers import (maybe_serialize_tool_calls,
44
45
                                                truncate_tool_call_ids,
                                                validate_request_params)
46
47
48
49
50
51

logger = init_logger(__name__)


class OpenAIServingChat(OpenAIServing):

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

76
        self.response_role = response_role
77
78
        self.chat_template = chat_template
        self.chat_template_content_format: Final = chat_template_content_format
79

80
81
82
83
84
85
86
87
        # 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.")

88
89
        self.reasoning_parser: Optional[Callable[[AnyTokenizer],
                                                 ReasoningParser]] = None
90
        if reasoning_parser:
91
92
93
94
            try:
                self.reasoning_parser = (
                    ReasoningParserManager.get_reasoning_parser(
                        reasoning_parser))
95
                assert self.reasoning_parser is not None
96
            except Exception as e:
97
98
                raise TypeError(
                    f"{reasoning_parser=} has not been registered") from e
99
100
        self.tool_parser: Optional[Callable[[AnyTokenizer], ToolParser]] = None
        if self.enable_auto_tools:
101
            try:
102
103
104
105
106
                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")
107
108
109
                self.tool_parser = ToolParserManager.get_tool_parser(
                    tool_parser)
            except Exception as e:
110
                raise TypeError("Error: --enable-auto-tool-choice requires "
111
112
                                f"tool_parser:'{tool_parser}' which has not "
                                "been registered") from e
113

114
        self.enable_prompt_tokens_details = enable_prompt_tokens_details
115
        self.enable_force_include_usage = enable_force_include_usage
116
117
118
        self.default_sampling_params = (
            self.model_config.get_diff_sampling_param())
        if self.default_sampling_params:
119
120
121
122
            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)
123

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

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

142
143
144
145
146
147
        # 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

148
        try:
149
150
151
152
153
            (
                lora_request,
                prompt_adapter_request,
            ) = self._maybe_get_adapters(request)

154
            model_name = self._get_model_name(request.model, lora_request)
155

156
            tokenizer = await self.engine_client.get_tokenizer(lora_request)
157

158
159
            tool_parser = self.tool_parser

160
            if isinstance(tokenizer, MistralTokenizer):
161
162
163
                # 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`
164
                maybe_serialize_tool_calls(request)
165
                truncate_tool_call_ids(request)
166
                validate_request_params(request)
167

168
169
170
171
172
173
174
175
176
            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"
                )
177

178
179
180
181
            tool_dicts = None if request.tools is None else [
                tool.model_dump() for tool in request.tools
            ]

182
183
184
185
186
187
188
189
190
            (
                conversation,
                request_prompts,
                engine_prompts,
            ) = await self._preprocess_chat(
                request,
                tokenizer,
                request.messages,
                chat_template=request.chat_template or self.chat_template,
191
                chat_template_content_format=self.chat_template_content_format,
192
193
194
195
196
197
198
199
200
                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,
            )
201
202
        except (ValueError, TypeError, RuntimeError,
                jinja2.TemplateError) as e:
203
            logger.exception("Error in preprocessing prompt inputs")
204
            return self.create_error_response(f"{e} {e.__cause__}")
205

206
207
        request_id = "chatcmpl-" \
                     f"{self._base_request_id(raw_request, request.request_id)}"
208
209
210
211
212

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

213
        # Schedule the request and get the result generator.
214
        generators: list[AsyncGenerator[RequestOutput, None]] = []
215
        try:
216
217
218
219
220
221
            for i, engine_prompt in enumerate(engine_prompts):
                sampling_params: Union[SamplingParams, BeamSearchParams]
                default_max_tokens = self.max_model_len - len(
                    engine_prompt["prompt_token_ids"])
                if request.use_beam_search:
                    sampling_params = request.to_beam_search_params(
222
                        default_max_tokens, self.default_sampling_params)
223
224
                else:
                    sampling_params = request.to_sampling_params(
225
                        default_max_tokens,
226
                        self.model_config.logits_processor_pattern,
227
                        self.default_sampling_params)
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242

                self._log_inputs(request_id,
                                 request_prompts[i],
                                 params=sampling_params,
                                 lora_request=lora_request,
                                 prompt_adapter_request=prompt_adapter_request)

                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,
243
                        lora_request=lora_request,
244
245
246
247
248
249
250
251
252
253
254
255
256
                    )
                else:
                    generator = self.engine_client.generate(
                        engine_prompt,
                        sampling_params,
                        request_id,
                        lora_request=lora_request,
                        trace_headers=trace_headers,
                        prompt_adapter_request=prompt_adapter_request,
                        priority=request.priority,
                    )

                generators.append(generator)
257
        except ValueError as e:
258
            # TODO: Use a vllm-specific Validation Error
259
260
            return self.create_error_response(str(e))

261
262
263
        assert len(generators) == 1
        result_generator, = generators

264
265
266
        # Streaming response
        if request.stream:
            return self.chat_completion_stream_generator(
267
268
269
270
271
272
273
274
                request,
                result_generator,
                request_id,
                model_name,
                conversation,
                tokenizer,
                request_metadata,
                enable_force_include_usage=self.enable_force_include_usage)
275

276
277
        try:
            return await self.chat_completion_full_generator(
278
279
                request, result_generator, request_id, model_name,
                conversation, tokenizer, request_metadata)
280
281
282
        except ValueError as e:
            # TODO: Use a vllm-specific Validation Error
            return self.create_error_response(str(e))
283
284
285
286

    def get_chat_request_role(self, request: ChatCompletionRequest) -> str:
        if request.add_generation_prompt:
            return self.response_role
287
        return request.messages[-1]["role"]
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
    @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,
332
        current_text: Optional[str],
333
334
335
        delta_text: str,
        function_name_returned: bool,
    ) -> tuple[Optional[DeltaMessage], bool]:
336
337
338
        if current_text is None or current_text == "":
            # if the current text is empty, we cannot parse it
            return None, function_name_returned
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
        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=[
381
382
383
384
                        DeltaToolCall(id=random_tool_call_id(),
                                      function=DeltaFunctionCall(
                                          name=current_tool_call["name"],
                                          arguments=arguments),
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
                                      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),
401
                                index=len(obj) - 1)
402
403
404
405
406
407
                        ])
                    else:
                        delta_message = None

        return delta_message, function_name_returned

408
    async def chat_completion_stream_generator(
409
410
411
412
        self,
        request: ChatCompletionRequest,
        result_generator: AsyncIterator[RequestOutput],
        request_id: str,
413
        model_name: str,
414
        conversation: list[ConversationMessage],
415
        tokenizer: AnyTokenizer,
416
        request_metadata: RequestResponseMetadata,
417
        enable_force_include_usage: bool,
418
    ) -> AsyncGenerator[str, None]:
419
        created_time = int(time.time())
420
        chunk_object_type: Final = "chat.completion.chunk"
421
        first_iteration = True
422
423

        # Send response for each token for each request.n (index)
424
425
426
        num_choices = 1 if request.n is None else request.n
        previous_num_tokens = [0] * num_choices
        finish_reason_sent = [False] * num_choices
427
        num_prompt_tokens = 0
428
        num_cached_tokens = None
429
430
431
432
433
434
435
436
437
438
439

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

440
        all_previous_token_ids: Optional[list[list[int]]]
441
        function_name_returned = [False] * num_choices
442
443
444

        # Only one of these will be used, thus previous_texts and
        # all_previous_token_ids will not be used twice in the same iteration.
445
        if tool_choice_auto or self.reasoning_parser:
446
447
448
            # These are only required in "auto" tool choice case
            previous_texts = [""] * num_choices
            all_previous_token_ids = [[]] * num_choices
449
450
451
            # For reasoning parser and tool call all enabled
            added_content_delta_arr = [False] * num_choices
            reasoning_end_arr = [False] * num_choices
452
453
454
        elif request.tool_choice == "required":
            previous_texts = [""] * num_choices
            all_previous_token_ids = None
455
456
457
        else:
            previous_texts, all_previous_token_ids = None, None

458
        try:
459
            if self.reasoning_parser:
460
461
462
463
464
465
466
                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
467
468
469
        # Prepare the tool parser if it's needed
        try:
            if tool_choice_auto and self.tool_parser:
470
                tool_parsers: list[Optional[ToolParser]] = [
471
472
473
474
                    self.tool_parser(tokenizer)
                ] * num_choices
            else:
                tool_parsers = [None] * num_choices
475
        except Exception as e:
476
            logger.exception("Error in tool parser creation.")
477
478
479
480
481
            data = self.create_streaming_error_response(str(e))
            yield f"data: {data}\n\n"
            yield "data: [DONE]\n\n"
            return

482
483
        stream_options = request.stream_options
        if stream_options:
484
485
            include_usage = stream_options.include_usage \
                            or enable_force_include_usage
486
487
488
489
490
            include_continuous_usage = include_usage and \
                                       stream_options.continuous_usage_stats
        else:
            include_usage, include_continuous_usage = False, False

491
492
        try:
            async for res in result_generator:
493
494
                if res.prompt_token_ids is not None:
                    num_prompt_tokens = len(res.prompt_token_ids)
495
496
                    if res.encoder_prompt_token_ids is not None:
                        num_prompt_tokens += len(res.encoder_prompt_token_ids)
497

498
499
500
501
                # 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:
502
                    num_cached_tokens = res.num_cached_tokens
503
504
                    # Send first response for each request.n (index) with
                    # the role
505
                    role = self.get_chat_request_role(request)
506
507
508

                    # NOTE num_choices defaults to 1 so this usually executes
                    # once per request
509
                    for i in range(num_choices):
510
511
                        choice_data = ChatCompletionResponseStreamChoice(
                            index=i,
512
513
514
515
                            delta=DeltaMessage(
                                role=role,
                                content="",
                            ),
516
517
518
519
520
521
522
523
                            logprobs=None,
                            finish_reason=None)
                        chunk = ChatCompletionStreamResponse(
                            id=request_id,
                            object=chunk_object_type,
                            created=created_time,
                            choices=[choice_data],
                            model=model_name)
524

525
526
527
528
529
530
                        # 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)
531

532
533
534
                        data = chunk.model_dump_json(exclude_unset=True)
                        yield f"data: {data}\n\n"

535
536
                    # Send response to echo the input portion of the
                    # last message
537
                    if request.echo:
538
                        last_msg_content: Union[str, list[dict[str, str]]] = ""
539
540
541
                        if conversation and "content" in conversation[
                                -1] and conversation[-1].get("role") == role:
                            last_msg_content = conversation[-1]["content"] or ""
542
543

                        if last_msg_content:
544
                            for i in range(num_choices):
545
546
547
548
549
                                choice_data = (
                                    ChatCompletionResponseStreamChoice(
                                        index=i,
                                        delta=DeltaMessage(
                                            content=last_msg_content),
550
                                        logprobs=None,
551
                                        finish_reason=None))
552
553
554
555
556
557
                                chunk = ChatCompletionStreamResponse(
                                    id=request_id,
                                    object=chunk_object_type,
                                    created=created_time,
                                    choices=[choice_data],
                                    model=model_name)
558
559
560
561
562
                                if include_continuous_usage:
                                    chunk.usage = UsageInfo(
                                        prompt_tokens=num_prompt_tokens,
                                        completion_tokens=0,
                                        total_tokens=num_prompt_tokens)
563

564
565
566
567
568
569
570
                                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
571
                    tool_parser = tool_parsers[i]
572
573
574
575

                    if finish_reason_sent[i]:
                        continue

576
                    if request.logprobs and request.top_logprobs is not None:
577
                        assert output.logprobs is not None, (
578
                            "Did not output logprobs")
579
                        logprobs = self._create_chat_logprobs(
580
581
                            token_ids=output.token_ids,
                            top_logprobs=output.logprobs,
582
                            tokenizer=tokenizer,
583
                            num_output_top_logprobs=request.top_logprobs,
584
585
                            return_as_token_id=request.
                            return_tokens_as_token_ids,
586
587
588
589
                        )
                    else:
                        logprobs = None

590
                    delta_text = output.text
591
592
593
594
595
596

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

597
                    delta_message: Optional[DeltaMessage]
598

599
                    # just update previous_texts and previous_token_ids
600
                    if tool_choice_auto or self.reasoning_parser:
601
602
603
604
605
606
607
608
                        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
                        current_token_ids = previous_token_ids + list(
                            output.token_ids)

609
610
                    # handle streaming deltas for tools with named tool_choice
                    if tool_choice_function_name:
611
                        if (self.reasoning_parser
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
                                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,
                                ))
                            # When encountering think end id in delta_token_ids,
                            # process the `content`. Only keep 'content',
                            # remove 'reasoning_content'
                            if reasoning_parser.is_reasoning_end(
                                    list(output.token_ids)):
                                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`
638
                            if self.reasoning_parser:
639
640
641
                                delta_text = previous_text + delta_text
                                current_text = ""

642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
                            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

657
                            delta_message = DeltaMessage(tool_calls=[
658
                                delta_tool_call,
659
660
                            ])

661
662
663
664
665
666
                    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]

667
668
669
670
671
672
673
674
                        if self.reasoning_parser:
                            _, content = \
                                reasoning_parser.extract_reasoning_content(
                                    current_text,
                                    request
                                )
                        else:
                            content = current_text
675
676
677
                        delta_message, function_name_returned[i] = (
                            self.extract_tool_call_required_streaming(
                                previous_text=previous_text,
678
                                current_text=content,
679
680
681
682
683
684
                                delta_text=delta_text,
                                function_name_returned=fn_name_returned))

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

685
686
                    # handle streaming deltas for tools with "auto" tool choice
                    # and reasoning parser
687
                    elif tool_choice_auto and self.reasoning_parser:
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
                        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,
                                ))
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
                            # 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 = ""
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
                            # 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
759
760
                        delta_message = (
                            tool_parser.extract_tool_calls_streaming(
761
762
                                previous_text=previous_text,
                                current_text=current_text,
763
                                delta_text=delta_text,
764
765
                                previous_token_ids=previous_token_ids,
                                current_token_ids=current_token_ids,
766
767
                                delta_token_ids=output.token_ids,
                                request=request))
768
                    # when only reasoning
769
                    elif self.reasoning_parser:
770
771
772
773
774
775
776
777
778
                        delta_message = (reasoning_parser.
                                         extract_reasoning_content_streaming(
                                             previous_text,
                                             current_text,
                                             delta_text,
                                             previous_token_ids,
                                             current_token_ids,
                                             output.token_ids,
                                         ))
779
                    # handle streaming just a content delta
780
781
782
                    else:
                        delta_message = DeltaMessage(content=delta_text)

783
                    # update the previous values for the next iteration
784
                    if tool_choice_auto or self.reasoning_parser:
785
786
787
788
789
                        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

790
                    # set the previous values for the next iteration
791
                    previous_num_tokens[i] += len(output.token_ids)
792
793
794
795
796
797
798
799

                    # 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

800
801
802
803
                    if output.finish_reason is None:
                        # Send token-by-token response for each request.n
                        choice_data = ChatCompletionResponseStreamChoice(
                            index=i,
804
                            delta=delta_message,
805
806
                            logprobs=logprobs,
                            finish_reason=None)
807
808

                    # if the model is finished generating
809
                    else:
810
811
812
813
                        # 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
814
                        auto_tools_called = False
815
                        if tool_parser:
816
817
818
819
                            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
820
821
822
823
824
                        else:
                            index = 0

                        if self._should_check_for_unstreamed_tool_arg_tokens(
                                delta_message, output) and tool_parser:
825
826
827
828
829
830
831
832
833
834
                            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)

835
836
837
838
                            # 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(
839
840
                                    "arguments", {}),
                                ensure_ascii=False)
841

842
                            # get what we've streamed so far for arguments
843
844
845
                            # for the current tool
                            actual_call = tool_parser.streamed_args_for_tool[
                                index]
846
847
                            if (latest_delta_len > 0):
                                actual_call = actual_call[:-latest_delta_len]
848
849
850
851
852
853
854
855
856
857
858
859

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

860
861
862
                        # Send the finish response for each request.n only once
                        choice_data = ChatCompletionResponseStreamChoice(
                            index=i,
863
                            delta=delta_message,
864
                            logprobs=logprobs,
865
                            finish_reason=output.finish_reason
866
                            if not auto_tools_called else "tool_calls",
867
                            stop_reason=output.stop_reason)
868

869
                        finish_reason_sent[i] = True
870

871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
                    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,
                        )

887
                    data = chunk.model_dump_json(exclude_unset=True)
888
889
                    yield f"data: {data}\n\n"

890
891
            # once the final token is handled, if stream_options.include_usage
            # is sent, send the usage
892
893
            if include_usage:
                completion_tokens = sum(previous_num_tokens)
894
895
896
897
898
899
900
                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)
901
902
903
904
905
906
907
908
909
910
911

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

913
914
915
916
917
918
919
            # 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)

920
        except Exception as e:
921
            # TODO: Use a vllm-specific Validation Error
922
            logger.exception("Error in chat completion stream generator.")
923
924
            data = self.create_streaming_error_response(str(e))
            yield f"data: {data}\n\n"
925
926
927
928
        # Send the final done message after all response.n are finished
        yield "data: [DONE]\n\n"

    async def chat_completion_full_generator(
929
930
931
932
        self,
        request: ChatCompletionRequest,
        result_generator: AsyncIterator[RequestOutput],
        request_id: str,
933
        model_name: str,
934
        conversation: list[ConversationMessage],
935
        tokenizer: AnyTokenizer,
936
        request_metadata: RequestResponseMetadata,
937
    ) -> Union[ErrorResponse, ChatCompletionResponse]:
938

939
        created_time = int(time.time())
940
        final_res: Optional[RequestOutput] = None
941

942
943
944
945
946
        try:
            async for res in result_generator:
                final_res = res
        except asyncio.CancelledError:
            return self.create_error_response("Client disconnected")
947
948
949
        except ValueError as e:
            # TODO: Use a vllm-specific Validation Error
            return self.create_error_response(str(e))
950

951
952
        assert final_res is not None

953
        choices: list[ChatCompletionResponseChoice] = []
954

955
956
        role = self.get_chat_request_role(request)
        for output in final_res.outputs:
957
            token_ids = output.token_ids
958
            out_logprobs = output.logprobs
959

960
961
            if request.logprobs and request.top_logprobs is not None:
                assert out_logprobs is not None, "Did not output logprobs"
962
                logprobs = self._create_chat_logprobs(
963
                    token_ids=token_ids,
964
                    top_logprobs=out_logprobs,
965
                    num_output_top_logprobs=request.top_logprobs,
966
                    tokenizer=tokenizer,
967
                    return_as_token_id=request.return_tokens_as_token_ids,
968
969
970
                )
            else:
                logprobs = None
971
            auto_tools_called = False
972

973
            if self.reasoning_parser:
974
975
976
977
978
                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))
979
980
                # If the reasoning parser is enabled,
                # tool calls are extracted exclusively from the content.
981
982
983
                reasoning_content, content = (
                    reasoning_parser.extract_reasoning_content(
                        output.text, request=request))
984
985
986
            else:
                reasoning_content = None
                content = output.text
987

988
989
            # if auto tools are not enabled, and a named tool choice using
            #   outlines is not being used
990
991
992
993
            if (not self.enable_auto_tools or not self.tool_parser) and \
                (not isinstance(request.tool_choice,
                                ChatCompletionNamedToolChoiceParam
                                ) and request.tool_choice != "required"):
994
995
996
                message = ChatMessage(role=role,
                                      reasoning_content=reasoning_content,
                                      content=content)
997
998
999

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

1002
1003
                tool_call_class = MistralToolCall if isinstance(
                    tokenizer, MistralTokenizer) else ToolCall
1004
1005
                message = ChatMessage(
                    role=role,
1006
                    reasoning_content=reasoning_content,
1007
1008
                    content="",
                    tool_calls=[
1009
                        tool_call_class(function=FunctionCall(
1010
                            name=request.tool_choice.function.name,
1011
                            arguments=content))
1012
                    ])
1013

1014
1015
1016
1017
1018
1019
            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
1020
                assert content is not None
1021
                tool_calls = TypeAdapter(
1022
                    list[FunctionDefinition]).validate_json(content)
1023
1024
1025
1026
1027
1028
                message = ChatMessage(
                    role=role,
                    content="",
                    tool_calls=[
                        tool_call_class(function=FunctionCall(
                            name=tool_call.name,
1029
1030
                            arguments=json.dumps(tool_call.parameters,
                                                 ensure_ascii=False)))
1031
1032
1033
                        for tool_call in tool_calls
                    ])

1034
1035
            # if the request doesn't use tool choice
            # OR specifies to not use a tool
1036
            elif not request.tool_choice or request.tool_choice == "none":
1037

1038
1039
1040
                message = ChatMessage(role=role,
                                      reasoning_content=reasoning_content,
                                      content=content)
1041
1042
1043
1044
1045
1046
1047

            # 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:

1048
1049
1050
                try:
                    tool_parser = self.tool_parser(tokenizer)
                except RuntimeError as e:
1051
                    logger.exception("Error in tool parser creation.")
1052
1053
                    return self.create_error_response(str(e))

1054
                tool_call_info = tool_parser.extract_tool_calls(
1055
                    content if content is not None else "", request=request)
1056
1057
1058
1059
                # 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
1060
1061
                if tool_call_info.tools_called:
                    message = ChatMessage(role=role,
1062
                                          reasoning_content=reasoning_content,
1063
1064
1065
1066
1067
1068
                                          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.
1069
1070
1071
                    message = ChatMessage(role=role,
                                          reasoning_content=reasoning_content,
                                          content=content)
1072
1073
1074
1075
1076
1077
1078

            # 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.")
1079
1080
1081
                message = ChatMessage(role=role,
                                      reasoning_content=reasoning_content,
                                      content=content)
1082

1083
1084
            choice_data = ChatCompletionResponseChoice(
                index=output.index,
1085
                message=message,
1086
                logprobs=logprobs,
1087
                finish_reason="tool_calls" if auto_tools_called else
1088
                output.finish_reason if output.finish_reason else "stop",
1089
                stop_reason=output.stop_reason)
1090
1091
            choices.append(choice_data)

1092
        if request.echo:
1093
            last_msg_content: Union[str, list[dict[str, str]]] = ""
1094
1095
            if conversation and "content" in conversation[-1] and conversation[
                    -1].get("role") == role:
1096
                last_msg_content = conversation[-1]["content"] or ""
1097
1098
1099
            if isinstance(last_msg_content, list):
                last_msg_content = "\n".join(msg['text']
                                             for msg in last_msg_content)
1100
1101

            for choice in choices:
1102
1103
                full_message = last_msg_content + (choice.message.content
                                                   or "")
1104
1105
                choice.message.content = full_message

1106
        assert final_res.prompt_token_ids is not None
1107
        num_prompt_tokens = len(final_res.prompt_token_ids)
1108
1109
        if final_res.encoder_prompt_token_ids is not None:
            num_prompt_tokens += len(final_res.encoder_prompt_token_ids)
1110
1111
        num_generated_tokens = sum(
            len(output.token_ids) for output in final_res.outputs)
1112
1113
1114
1115
1116
1117
1118
        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)
1119
1120
1121

        request_metadata.final_usage_info = usage

1122
1123
1124
1125
1126
1127
        response = ChatCompletionResponse(
            id=request_id,
            created=created_time,
            model=model_name,
            choices=choices,
            usage=usage,
1128
            prompt_logprobs=clamp_prompt_logprobs(final_res.prompt_logprobs),
Robert Shaw's avatar
Robert Shaw committed
1129
            kv_transfer_params=final_res.kv_transfer_params,
1130
1131
        )

1132
        return response
1133
1134

    def _get_top_logprobs(
1135
            self, logprobs: dict[int, Logprob], top_logprobs: Optional[int],
1136
1137
            tokenizer: AnyTokenizer,
            should_return_as_token_id: bool) -> list[ChatCompletionLogProb]:
1138
        return [
1139
1140
1141
1142
            ChatCompletionLogProb(token=(token := self._get_decoded_token(
                p[1],
                p[0],
                tokenizer,
1143
                return_as_token_id=should_return_as_token_id)),
1144
1145
1146
                                  logprob=max(p[1].logprob, -9999.0),
                                  bytes=list(
                                      token.encode("utf-8", errors="replace")))
1147
1148
1149
1150
1151
1152
1153
            for i, p in enumerate(logprobs.items())
            if top_logprobs and i < top_logprobs
        ]

    def _create_chat_logprobs(
        self,
        token_ids: GenericSequence[int],
1154
        top_logprobs: GenericSequence[Optional[dict[int, Logprob]]],
1155
        tokenizer: AnyTokenizer,
1156
        num_output_top_logprobs: Optional[int] = None,
1157
        return_as_token_id: Optional[bool] = None,
1158
1159
    ) -> ChatCompletionLogProbs:
        """Create OpenAI-style logprobs."""
1160
        logprobs_content: list[ChatCompletionLogProbsContent] = []
1161

1162
1163
        should_return_as_token_id = return_as_token_id if \
            return_as_token_id is not None else self.return_tokens_as_token_ids
1164
1165
        for i, token_id in enumerate(token_ids):
            step_top_logprobs = top_logprobs[i]
1166
1167
            if step_top_logprobs is None or step_top_logprobs.get(
                    token_id) is None:
1168
                token = tokenizer.decode(token_id)
1169
                if should_return_as_token_id:
1170
                    token = f"token_id:{token_id}"
1171

1172
1173
                logprobs_content.append(
                    ChatCompletionLogProbsContent(
1174
                        token=token,
1175
1176
                        bytes=list(token.encode("utf-8", errors="replace")),
                    ))
1177
            else:
1178
1179
1180
                step_token = step_top_logprobs[token_id]
                step_decoded = step_token.decoded_token

1181
1182
                logprobs_content.append(
                    ChatCompletionLogProbsContent(
1183
                        token=self._get_decoded_token(
1184
1185
1186
                            step_token,
                            token_id,
                            tokenizer,
1187
                            should_return_as_token_id,
1188
1189
1190
1191
                        ),
                        logprob=max(step_token.logprob, -9999.0),
                        bytes=None if step_decoded is None else list(
                            step_decoded.encode("utf-8", errors="replace")),
1192
                        top_logprobs=self._get_top_logprobs(
1193
1194
                            step_top_logprobs, num_output_top_logprobs,
                            tokenizer, should_return_as_token_id),
1195
                    ))
1196
1197

        return ChatCompletionLogProbs(content=logprobs_content)
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226

    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
1227
1228
            output.finish_reason is not None
            and self.enable_auto_tools and self.tool_parser and delta_message
1229
1230
1231
1232
            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
        )