serving.py 68.5 KB
Newer Older
1
2
3
4
5
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project

import asyncio
import time
6
import uuid
7
from collections import deque
8
from collections.abc import AsyncGenerator, AsyncIterator, Callable, Sequence
9
from contextlib import AsyncExitStack
10
from copy import copy
11
from http import HTTPStatus
12
from typing import Final
13
14
15

import jinja2
from fastapi import Request
16
from openai.types.responses import (
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
    ResponseContentPartAddedEvent,
    ResponseContentPartDoneEvent,
    ResponseFunctionToolCall,
    ResponseOutputItem,
    ResponseOutputItemAddedEvent,
    ResponseOutputItemDoneEvent,
    ResponseOutputMessage,
    ResponseOutputText,
    ResponseReasoningItem,
    ResponseReasoningTextDeltaEvent,
    ResponseReasoningTextDoneEvent,
    ResponseStatus,
    ResponseTextDeltaEvent,
    ResponseTextDoneEvent,
    response_text_delta_event,
)
from openai.types.responses.response_output_text import Logprob, LogprobTopLogprob
34
from openai.types.responses.response_reasoning_item import (
35
36
    Content as ResponseReasoningTextContent,
)
37
from openai.types.responses.tool import Mcp, Tool
38
from openai_harmony import Message as OpenAIHarmonyMessage
39
from pydantic import TypeAdapter
40

41
from vllm import envs
42
from vllm.config.utils import replace
43
from vllm.engine.protocol import EngineClient
44
45
46
47
from vllm.entrypoints.chat_utils import (
    ChatCompletionMessageParam,
    ChatTemplateContentFormatOption,
)
48
from vllm.entrypoints.logger import RequestLogger
49
from vllm.entrypoints.mcp.tool_server import ToolServer
50
from vllm.entrypoints.openai.engine.protocol import (
51
52
53
54
    DeltaMessage,
    ErrorResponse,
    RequestResponseMetadata,
)
55
from vllm.entrypoints.openai.engine.serving import (
56
57
58
    GenerationError,
    OpenAIServing,
)
59
from vllm.entrypoints.openai.models.serving import OpenAIServingModels
60
61
62
63
64
65
66
67
68
69
70
71
from vllm.entrypoints.openai.parser.harmony_utils import (
    construct_harmony_previous_input_messages,
    get_developer_message,
    get_stop_tokens_for_assistant_actions,
    get_system_message,
    get_user_message,
    has_custom_tools,
    parse_output_message,
    parse_remaining_state,
    parse_response_input,
    render_for_completion,
)
72
73
74
75
76
77
78
from vllm.entrypoints.openai.responses.context import (
    ConversationContext,
    HarmonyContext,
    ParsableContext,
    SimpleContext,
    StreamingHarmonyContext,
)
79
80
81
82
83
84
85
86
87
88
89
90
from vllm.entrypoints.openai.responses.protocol import (
    InputTokensDetails,
    OutputTokensDetails,
    ResponseCompletedEvent,
    ResponseCreatedEvent,
    ResponseInProgressEvent,
    ResponseInputOutputMessage,
    ResponsesRequest,
    ResponsesResponse,
    ResponseUsage,
    StreamingResponsesResponse,
)
91
92
93
94
95
96
from vllm.entrypoints.openai.responses.streaming_events import (
    HarmonyStreamingState,
    emit_content_delta_events,
    emit_previous_item_done_events,
    emit_tool_action_events,
)
97
from vllm.entrypoints.openai.responses.utils import (
98
    construct_input_messages,
99
    construct_tool_dicts,
100
101
    extract_tool_types,
)
102
from vllm.entrypoints.utils import get_max_tokens
103
from vllm.exceptions import VLLMValidationError
104
from vllm.inputs.data import ProcessorInputs, token_inputs
105
from vllm.logger import init_logger
106
107
from vllm.logprobs import Logprob as SampleLogprob
from vllm.logprobs import SampleLogprobs
108
from vllm.outputs import CompletionOutput
109
from vllm.parser import ParserManager
110
from vllm.sampling_params import SamplingParams, StructuredOutputsParams
111
from vllm.tokenizers import TokenizerLike
112
113
114
115
116
from vllm.utils import random_uuid

logger = init_logger(__name__)


117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def _extract_allowed_tools_from_mcp_requests(
    tools: list[Tool],
) -> dict[str, list[str] | None]:
    """
    Extract allowed_tools mapping from MCP tool requests.

    Returns a dictionary mapping server_label to allowed_tools list.
    Handles both list format and McpAllowedToolsMcpToolFilter object format.

    Special handling:
    - If allowed_tools is None, returns None (allows all tools)
    - If allowed_tools contains "*", returns None (allows all tools)
    - Otherwise, returns the list of specific tool names

    This function can be reused for both harmony and non-harmony MCP calls.
    """
    allowed_tools_map: dict[str, list[str] | None] = {}
    for tool in tools:
        if not isinstance(tool, Mcp):
            continue

        # allowed_tools can be a list or an object with tool_names
        # Extract the actual list of tool names
        allowed_tools_val = None
        if tool.allowed_tools is not None:
            if isinstance(tool.allowed_tools, list):
                allowed_tools_val = tool.allowed_tools
            elif hasattr(tool.allowed_tools, "tool_names"):
                # It's an McpAllowedToolsMcpToolFilter object
                allowed_tools_val = tool.allowed_tools.tool_names

        # Normalize "*" to None (both mean "allow all tools")
        if allowed_tools_val is not None and "*" in allowed_tools_val:
            allowed_tools_val = None

        allowed_tools_map[tool.server_label] = allowed_tools_val
    return allowed_tools_map


156
157
158
159
160
161
class OpenAIServingResponses(OpenAIServing):
    def __init__(
        self,
        engine_client: EngineClient,
        models: OpenAIServingModels,
        *,
162
163
        request_logger: RequestLogger | None,
        chat_template: str | None,
164
165
166
167
        chat_template_content_format: ChatTemplateContentFormatOption,
        return_tokens_as_token_ids: bool = False,
        reasoning_parser: str = "",
        enable_auto_tools: bool = False,
168
169
        tool_parser: str | None = None,
        tool_server: ToolServer | None = None,
170
171
        enable_prompt_tokens_details: bool = False,
        enable_force_include_usage: bool = False,
172
        enable_log_outputs: bool = False,
173
        log_error_stack: bool = False,
174
175
176
177
178
179
    ) -> None:
        super().__init__(
            engine_client=engine_client,
            models=models,
            request_logger=request_logger,
            return_tokens_as_token_ids=return_tokens_as_token_ids,
180
            log_error_stack=log_error_stack,
181
182
183
184
        )

        self.chat_template = chat_template
        self.chat_template_content_format: Final = chat_template_content_format
185
        self.enable_log_outputs = enable_log_outputs
186

187
188
189
190
191
192
193
        # Set up the unified parser - either a unified parser or fall back to
        # separate parsers accessed through the parser interface
        self.parser = ParserManager.get_parser(
            tool_parser_name=tool_parser,
            reasoning_parser_name=reasoning_parser,
            enable_auto_tools=enable_auto_tools,
            model_name=self.model_config.model,
194
        )
195
196
        self.enable_prompt_tokens_details = enable_prompt_tokens_details
        self.enable_force_include_usage = enable_force_include_usage
197

198
        self.default_sampling_params = self.model_config.get_diff_sampling_param()
199
200
201
202
203
204
        mc = self.model_config
        self.override_max_tokens = (
            self.default_sampling_params.get("max_tokens")
            if mc.generation_config not in ("auto", "vllm")
            else getattr(mc, "override_generation_config", {}).get("max_new_tokens")
        )
205

206
207
208
209
210
        # If False (default), the "store" option is (silently) ignored and the
        # response is not stored. If True, the response is stored in memory.
        # NOTE(woosuk): This may not be intuitive for users, as the default
        # behavior in OpenAI's Responses API is to store the response, but
        # vLLM's default behavior is not.
211
        self.enable_store = envs.VLLM_ENABLE_RESPONSES_API_STORE
212
213
214
215
        if self.enable_store:
            logger.warning_once(
                "`VLLM_ENABLE_RESPONSES_API_STORE` is enabled. This may "
                "cause a memory leak since we never remove responses from "
216
217
                "the store."
            )
218

219
        self.use_harmony = self.model_config.hf_config.model_type == "gpt_oss"
220
        if self.use_harmony:
221
222
223
224
            logger.warning(
                "For gpt-oss, we ignore --enable-auto-tool-choice "
                "and always enable tool use."
            )
225
226
227
228
229
            # OpenAI models have two EOS-like tokens: <|return|> and <|call|>.
            # We need to add them to the stop token ids.
            if "stop_token_ids" not in self.default_sampling_params:
                self.default_sampling_params["stop_token_ids"] = []
            self.default_sampling_params["stop_token_ids"].extend(
230
231
                get_stop_tokens_for_assistant_actions()
            )
232
233
234
235
236
237
238
239
240
241
242

        # Handle tool call ID type for Kimi K2 (supporting test mocking via overrides)
        hf_overrides = getattr(self.model_config, "hf_overrides", None)
        if self.model_config.hf_text_config.model_type == "kimi_k2" or (
            isinstance(hf_overrides, dict)
            and hf_overrides.get("model_type") == "kimi_k2"
        ):
            self.tool_call_id_type = "kimi_k2"
        else:
            self.tool_call_id_type = "random"

243
        self.enable_auto_tools = enable_auto_tools
244
        # HACK(woosuk): This is a hack. We should use a better store.
245
246
        # FIXME: If enable_store=True, this may cause a memory leak since we
        # never remove responses from the store.
247
248
249
250
        self.response_store: dict[str, ResponsesResponse] = {}
        self.response_store_lock = asyncio.Lock()

        # HACK(woosuk): This is a hack. We should use a better store.
251
252
        # FIXME: If enable_store=True, this may cause a memory leak since we
        # never remove messages from the store.
253
254
        self.msg_store: dict[str, list[ChatCompletionMessageParam]] = {}

255
256
257
        # HACK(wuhang): This is a hack. We should use a better store.
        # FIXME: If enable_store=True, this may cause a memory leak since we
        # never remove events from the store.
258
259
260
        self.event_store: dict[
            str, tuple[deque[StreamingResponsesResponse], asyncio.Event]
        ] = {}
261

262
263
        self.background_tasks: dict[str, asyncio.Task] = {}

264
265
        self.tool_server = tool_server

266
    def _validate_generator_input(
267
        self,
268
        engine_prompt: ProcessorInputs,
269
    ) -> ErrorResponse | None:
270
        """Add validations to the input to the generator here."""
271
        prompt_len = self._extract_prompt_len(engine_prompt)
272
273
274
        max_model_len = self.model_config.max_model_len

        if prompt_len >= max_model_len:
275
            error_message = (
276
                f"The engine prompt length {prompt_len} "
277
                f"exceeds the max_model_len {max_model_len}. "
278
279
                "Please reduce prompt."
            )
280
281
282
283
            return self.create_error_response(
                err_type="invalid_request_error",
                message=error_message,
                status_code=HTTPStatus.BAD_REQUEST,
284
                param="input",
285
            )
286

287
288
        return None

289
290
291
292
293
294
295
296
    def _validate_create_responses_input(
        self, request: ResponsesRequest
    ) -> ErrorResponse | None:
        if self.use_harmony and request.is_include_output_logprobs():
            return self.create_error_response(
                err_type="invalid_request_error",
                message="logprobs are not supported with gpt-oss models",
                status_code=HTTPStatus.BAD_REQUEST,
297
                param="logprobs",
298
299
300
301
302
303
304
305
306
307
308
309
            )
        if request.store and not self.enable_store and request.background:
            return self.create_error_response(
                err_type="invalid_request_error",
                message=(
                    "This vLLM engine does not support `store=True` and "
                    "therefore does not support the background mode. To "
                    "enable these features, set the environment variable "
                    "`VLLM_ENABLE_RESPONSES_API_STORE=1` when launching "
                    "the vLLM server."
                ),
                status_code=HTTPStatus.BAD_REQUEST,
310
                param="background",
311
            )
312
313
314
315
316
317
        if request.previous_input_messages and request.previous_response_id:
            return self.create_error_response(
                err_type="invalid_request_error",
                message="Only one of `previous_input_messages` and "
                "`previous_response_id` can be set.",
                status_code=HTTPStatus.BAD_REQUEST,
318
                param="previous_response_id",
319
            )
320
321
        return None

322
323
324
    async def create_responses(
        self,
        request: ResponsesRequest,
325
326
327
328
329
330
        raw_request: Request | None = None,
    ) -> (
        AsyncGenerator[StreamingResponsesResponse, None]
        | ResponsesResponse
        | ErrorResponse
    ):
331
332
333
334
        error_check_ret = await self._check_model(request)
        if error_check_ret is not None:
            logger.error("Error with model %s", error_check_ret)
            return error_check_ret
335
336
337
        maybe_validation_error = self._validate_create_responses_input(request)
        if maybe_validation_error is not None:
            return maybe_validation_error
338
339
340
341
342
343
344

        # 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

345
        if request.store and not self.enable_store:
346
347
348
349
350
351
352
            # Disable the store option.
            # NOTE(woosuk): Although returning an error is possible, we opted
            # to implicitly disable store and process the request anyway, as
            # we assume most users do not intend to actually store the response
            # (i.e., their request's `store=True` just because it's the default
            # value).
            request.store = False
353

354
355
356
357
358
359
360
361
362
363
364
        # Handle the previous response ID.
        prev_response_id = request.previous_response_id
        if prev_response_id is not None:
            async with self.response_store_lock:
                prev_response = self.response_store.get(prev_response_id)
            if prev_response is None:
                return self._make_not_found_error(prev_response_id)
        else:
            prev_response = None

        try:
365
            lora_request = self._maybe_get_adapters(request)
366
            model_name = self.models.model_name(lora_request)
367

368
            if self.use_harmony:
369
370
                messages, engine_prompts = self._make_request_with_harmony(
                    request, prev_response
371
                )
372
            else:
373
                messages, engine_prompts = await self._make_request(
374
                    request, prev_response
375
                )
376

377
378
379
380
381
382
383
        except (
            ValueError,
            TypeError,
            RuntimeError,
            jinja2.TemplateError,
            NotImplementedError,
        ) as e:
384
            logger.exception("Error in preprocessing prompt inputs")
385
            return self.create_error_response(e)
386

387
        request_metadata = RequestResponseMetadata(request_id=request.request_id)
388
389
390
391
        if raw_request:
            raw_request.state.request_metadata = request_metadata

        # Schedule the request and get the result generator.
392
        max_model_len = self.model_config.max_model_len
393
        generators: list[AsyncGenerator[ConversationContext, None]] = []
394

395
396
397
398
399
        # Only include builtin tools that the request actually asked for.
        # Without this filter, tools registered on the server (e.g. via
        # --tool-server demo) would be available for execution even when
        # the request didn't enable them.
        requested_tool_types = extract_tool_types(request.tools)
400
        builtin_tool_list: list[str] = []
401
        if self.tool_server is not None:
402
403
404
405
            if (
                self.tool_server.has_tool("browser")
                and "web_search_preview" in requested_tool_types
            ):
406
                builtin_tool_list.append("browser")
407
408
409
410
            if (
                self.tool_server.has_tool("python")
                and "code_interpreter" in requested_tool_types
            ):
411
                builtin_tool_list.append("python")
412
413
414
415
            if (
                self.tool_server.has_tool("container")
                and "container" in requested_tool_types
            ):
416
                builtin_tool_list.append("container")
417

418
419
420
421
422
423
        if self.tool_server is not None:
            available_tools = builtin_tool_list
        else:
            assert len(builtin_tool_list) == 0
            available_tools = []
        try:
424
            tokenizer = self.renderer.get_tokenizer()
425

426
            for engine_prompt in engine_prompts:
427
428
429
430
                maybe_error = self._validate_generator_input(engine_prompt)
                if maybe_error is not None:
                    return maybe_error

431
                default_max_tokens = get_max_tokens(
432
                    max_model_len,
433
                    request.max_output_tokens,
434
                    self._extract_prompt_len(engine_prompt),
435
                    self.default_sampling_params,
436
                    self.override_max_tokens,
437
                )
438

439
                sampling_params = request.to_sampling_params(
440
441
                    default_max_tokens, self.default_sampling_params
                )
442

443
444
445
446
447
                trace_headers = (
                    None
                    if raw_request is None
                    else await self._get_trace_headers(raw_request.headers)
                )
448
449
450
451

                context: ConversationContext
                if self.use_harmony:
                    if request.stream:
452
                        context = StreamingHarmonyContext(messages, available_tools)
453
454
455
                    else:
                        context = HarmonyContext(messages, available_tools)
                else:
456
                    if envs.VLLM_USE_EXPERIMENTAL_PARSER_CONTEXT:
457
                        # This is a feature in development for parsing
458
459
460
                        # tokens during generation instead of at the end
                        context = ParsableContext(
                            response_messages=messages,
461
                            tokenizer=tokenizer,
462
463
464
                            reasoning_parser_cls=self.parser.reasoning_parser_cls
                            if self.parser
                            else None,
465
                            request=request,
466
467
468
                            tool_parser_cls=self.parser.tool_parser_cls
                            if self.parser
                            else None,
469
470
471
                            available_tools=available_tools,
                            chat_template=self.chat_template,
                            chat_template_content_format=self.chat_template_content_format,
472
473
474
                        )
                    else:
                        context = SimpleContext()
475

476
477
                if self.parser and self.parser.reasoning_parser_cls is not None:
                    reasoning_parser = self.parser.reasoning_parser_cls(tokenizer)
478
479
480
481
482
483
484
485
486
487
488
489
                    if (
                        isinstance(
                            struct_out := sampling_params.structured_outputs,
                            StructuredOutputsParams,
                        )
                        and struct_out.all_non_structural_tag_constraints_none()
                    ):
                        sampling_params.structured_outputs = replace(
                            struct_out,
                            structural_tag=reasoning_parser.prepare_structured_tag(
                                struct_out.structural_tag, self.tool_server
                            ),
490
                        )
491
492
493
494
495
496
497
498
                generator = self._generate_with_builtin_tools(
                    request_id=request.request_id,
                    engine_prompt=engine_prompt,
                    sampling_params=sampling_params,
                    context=context,
                    lora_request=lora_request,
                    priority=request.priority,
                    trace_headers=trace_headers,
499
                )
500
501
                generators.append(generator)
        except ValueError as e:
502
            return self.create_error_response(e)
503

504
        assert len(generators) == 1
505
        (result_generator,) = generators
506
507
508
509

        # Store the input messages.
        if request.store:
            self.msg_store[request.request_id] = messages
510

511
512
513
514
515
516
517
518
519
520
521
522
523
        if request.background:
            created_time = int(time.time())
            response = ResponsesResponse.from_request(
                request,
                sampling_params,
                model_name=model_name,
                created_time=created_time,
                output=[],
                status="queued",
                usage=None,
            )
            async with self.response_store_lock:
                self.response_store[response.id] = response
524

525
            # Run the request in the background.
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
            if request.stream:
                task = asyncio.create_task(
                    self._run_background_request_stream(
                        request,
                        sampling_params,
                        result_generator,
                        context,
                        model_name,
                        tokenizer,
                        request_metadata,
                        created_time,
                    ),
                    name=f"create_{request.request_id}",
                )
            else:
                task = asyncio.create_task(
                    self._run_background_request(
                        request,
                        sampling_params,
                        result_generator,
                        context,
                        model_name,
                        tokenizer,
                        request_metadata,
                        created_time,
                    ),
                    name=f"create_{response.id}",
                )
554

555
556
557
558
            # For cleanup.
            response_id = response.id
            self.background_tasks[response_id] = task
            task.add_done_callback(
559
560
                lambda _: self.background_tasks.pop(response_id, None)
            )
561
562

            if request.stream:
563
                return self.responses_background_stream_generator(request.request_id)
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
            return response

        if request.stream:
            return self.responses_stream_generator(
                request,
                sampling_params,
                result_generator,
                context,
                model_name,
                tokenizer,
                request_metadata,
            )

        try:
            return await self.responses_full_generator(
                request,
                sampling_params,
                result_generator,
                context,
                model_name,
                tokenizer,
                request_metadata,
            )
587
588
        except GenerationError as e:
            return self._convert_generation_error_to_response(e)
589
        except Exception as e:
590
            return self.create_error_response(e)
591

592
593
594
    async def _make_request(
        self,
        request: ResponsesRequest,
595
        prev_response: ResponsesResponse | None,
596
    ):
597
        tool_dicts = construct_tool_dicts(request.tools, request.tool_choice)
598
        # Construct the input messages.
599
600
601
602
603
604
        messages = construct_input_messages(
            request_instructions=request.instructions,
            request_input=request.input,
            prev_msg=self.msg_store.get(prev_response.id) if prev_response else None,
            prev_response_output=prev_response.output if prev_response else None,
        )
605

606
        _, engine_prompts = await self._preprocess_chat(
607
608
            request,
            messages,
609
610
611
            default_template=self.chat_template,
            default_template_content_format=self.chat_template_content_format,
            default_template_kwargs=None,
612
            tool_dicts=tool_dicts,
613
            tool_parser=self.parser.tool_parser_cls if self.parser else None,
614
        )
615
        return messages, engine_prompts
616
617
618
619

    def _make_request_with_harmony(
        self,
        request: ResponsesRequest,
620
        prev_response: ResponsesResponse | None,
621
622
623
    ):
        if request.tool_choice != "auto":
            raise NotImplementedError(
624
625
                "Only 'auto' tool_choice is supported in response API with Harmony"
            )
626

627
        messages = self._construct_input_messages_with_harmony(request, prev_response)
628
        prompt_token_ids = render_for_completion(messages)
629
        engine_prompt = token_inputs(prompt_token_ids)
630
631
632
633
634

        # Add cache_salt if provided in the request
        if request.cache_salt is not None:
            engine_prompt["cache_salt"] = request.cache_salt

635
        return messages, [engine_prompt]
636

637
638
639
640
641
642
    async def _initialize_tool_sessions(
        self,
        request: ResponsesRequest,
        context: ConversationContext,
        exit_stack: AsyncExitStack,
    ):
643
644
645
646
        # we should only initialize the tool session if the request needs tools
        if len(request.tools) == 0:
            return
        mcp_tools = {
647
            tool.server_label: tool for tool in request.tools if tool.type == "mcp"
648
        }
649
650
651
        await context.init_tool_sessions(
            self.tool_server, exit_stack, request.request_id, mcp_tools
        )
652

653
654
655
656
    async def responses_full_generator(
        self,
        request: ResponsesRequest,
        sampling_params: SamplingParams,
657
        result_generator: AsyncIterator[ConversationContext],
658
        context: ConversationContext,
659
        model_name: str,
660
        tokenizer: TokenizerLike,
661
        request_metadata: RequestResponseMetadata,
662
663
        created_time: int | None = None,
    ) -> ErrorResponse | ResponsesResponse:
664
665
666
        if created_time is None:
            created_time = int(time.time())

667
668
        async with AsyncExitStack() as exit_stack:
            try:
669
                await self._initialize_tool_sessions(request, context, exit_stack)
670
671
672
673
674
                async for _ in result_generator:
                    pass
            except asyncio.CancelledError:
                return self.create_error_response("Client disconnected")
            except ValueError as e:
675
                return self.create_error_response(e)
676

677
        # NOTE: Implementation of status is still WIP, but for now
678
679
680
681
        # we guarantee that if the status is not "completed", it is accurate.
        # "completed" is implemented as the "catch-all" for now.
        status: ResponseStatus = "completed"

682
683
        input_messages: ResponseInputOutputMessage | None = None
        output_messages: ResponseInputOutputMessage | None = None
684
685
686
        if self.use_harmony:
            assert isinstance(context, HarmonyContext)
            output = self._make_response_output_items_with_harmony(context)
687
            if request.enable_response_messages:
688
689
                input_messages = context.messages[: context.num_init_messages]
                output_messages = context.messages[context.num_init_messages :]
690
            num_tool_output_tokens = context.num_tool_output_tokens
691
692
693
694
695
            if len(output) > 0:
                if context.finish_reason == "length":
                    status = "incomplete"
                elif context.finish_reason == "abort":
                    status = "cancelled"
696
697
                else:
                    self._raise_if_error(context.finish_reason, request.request_id)
698
699
            else:
                status = "incomplete"
700
        elif isinstance(context, ParsableContext):
701
            output = context.parser.make_response_output_items_from_parsable_context()
702
703

            if request.enable_response_messages:
704
705
                input_messages = context.input_messages
                output_messages = context.output_messages
706
707
708
709

            # TODO: Calculate usage.
            # assert final_res.prompt_token_ids is not None
            num_tool_output_tokens = 0
710
711
712
713

            # Check finish reason from the parser
            if context.parser.finish_reason == "length":
                status = "incomplete"
714
        else:
715
            assert isinstance(context, SimpleContext)
716
717
            # Use final_output which has accumulated text/token_ids/logprobs
            final_res = context.final_output
718
719
720
721
            assert final_res is not None
            assert len(final_res.outputs) == 1
            final_output = final_res.outputs[0]

722
723
724
            # finish_reason='error' indicates retryable internal error
            self._raise_if_error(final_output.finish_reason, request.request_id)

725
726
727
728
            # Check if generation was stopped due to max_tokens
            if final_output.finish_reason == "length":
                status = "incomplete"

729
            output = self._make_response_output_items(request, final_output, tokenizer)
730

731
            if request.enable_response_messages:
732
733
734
                input_messages = context.input_messages
                output_messages = context.output_messages

735
736
            # Calculate usage.
            assert final_res.prompt_token_ids is not None
737
738
            num_tool_output_tokens = 0

739
        assert isinstance(context, (SimpleContext, HarmonyContext, ParsableContext))
740
741
742
743
        num_prompt_tokens = context.num_prompt_tokens
        num_generated_tokens = context.num_output_tokens
        num_cached_tokens = context.num_cached_tokens
        num_reasoning_tokens = context.num_reasoning_tokens
744
745
746
747
748
749
750
751
752
753
754
755
756
        # For text-based reasoning parsers (e.g., <think>...</think>),
        # HarmonyContext already counts reasoning tokens via channels.
        # For Simple/Parsable contexts, derive reasoning_tokens from
        # accumulated output token IDs using the parser if not already set.
        if (
            num_reasoning_tokens == 0
            and self.parser is not None
            and self.parser.reasoning_parser_cls is not None
            and isinstance(context, (SimpleContext, ParsableContext))
        ):
            reasoning_parser = self.parser.reasoning_parser_cls(tokenizer)
            accumulated = getattr(context, "_accumulated_token_ids", []) or []
            num_reasoning_tokens = reasoning_parser.count_reasoning_tokens(accumulated)
757
758
759
760

        usage = ResponseUsage(
            input_tokens=num_prompt_tokens,
            output_tokens=num_generated_tokens,
761
            total_tokens=num_prompt_tokens + num_generated_tokens,
762
763
764
765
766
767
768
769
770
            input_tokens_details=InputTokensDetails(
                cached_tokens=num_cached_tokens,
                input_tokens_per_turn=[
                    turn.input_tokens for turn in context.all_turn_metrics
                ],
                cached_tokens_per_turn=[
                    turn.cached_input_tokens for turn in context.all_turn_metrics
                ],
            ),
771
            output_tokens_details=OutputTokensDetails(
772
                reasoning_tokens=num_reasoning_tokens,
773
                tool_output_tokens=num_tool_output_tokens,
774
775
776
777
778
779
                output_tokens_per_turn=[
                    turn.output_tokens for turn in context.all_turn_metrics
                ],
                tool_output_tokens_per_turn=[
                    turn.tool_output_tokens for turn in context.all_turn_metrics
                ],
780
            ),
781
782
783
784
        )
        response = ResponsesResponse.from_request(
            request,
            sampling_params,
785
786
            input_messages=input_messages,
            output_messages=output_messages,
787
788
789
            model_name=model_name,
            created_time=created_time,
            output=output,
790
            status=status,
791
792
793
794
795
796
797
            usage=usage,
        )

        if request.store:
            async with self.response_store_lock:
                stored_response = self.response_store.get(response.id)
                # If the response is already cancelled, don't update it.
798
                if stored_response is None or stored_response.status != "cancelled":
799
800
801
                    self.response_store[response.id] = response
        return response

802
803
804
805
    def _topk_logprobs(
        self,
        logprobs: dict[int, SampleLogprob],
        top_logprobs: int,
806
        tokenizer: TokenizerLike,
807
    ) -> list[LogprobTopLogprob]:
808
809
810
811
812
        """Returns the top-k logprobs from the logprobs dictionary."""
        out = []
        for i, (token_id, _logprob) in enumerate(logprobs.items()):
            if i >= top_logprobs:
                break
813
814
815
816
817
            text = self._get_decoded_token(
                logprob=_logprob,
                token_id=token_id,
                tokenizer=tokenizer,
                return_as_token_id=self.return_tokens_as_token_ids,
818
            )
819
820
821
822
823
            out.append(
                LogprobTopLogprob(
                    token=text,
                    logprob=max(_logprob.logprob, -9999.0),
                    bytes=list(text.encode("utf-8", errors="replace")),
824
825
                )
            )
826
827
828
        return out

    def _create_response_logprobs(
829
830
        self,
        token_ids: Sequence[int],
831
        logprobs: SampleLogprobs | None,
832
        tokenizer: TokenizerLike,
833
        top_logprobs: int | None = None,
834
    ) -> list[Logprob]:
835
836
        assert logprobs is not None, "logprobs must be provided"
        assert len(token_ids) == len(logprobs), (
837
838
            "token_ids and logprobs.token_ids must have the same length"
        )
839
840
841
842
        out = []
        for i, token_id in enumerate(token_ids):
            logprob = logprobs[i]
            token_logprob = logprob[token_id]
843
844
845
846
847
            text = self._get_decoded_token(
                logprob=token_logprob,
                token_id=token_id,
                tokenizer=tokenizer,
                return_as_token_id=self.return_tokens_as_token_ids,
848
            )
849
850
851
852
853
            out.append(
                Logprob(
                    token=text,
                    logprob=max(token_logprob.logprob, -9999.0),
                    bytes=list(text.encode("utf-8", errors="replace")),
854
855
856
857
858
859
860
                    top_logprobs=(
                        self._topk_logprobs(
                            logprob, top_logprobs=top_logprobs, tokenizer=tokenizer
                        )
                        if top_logprobs
                        else []
                    ),
861
862
                )
            )
863
864
        return out

865
866
867
    def _create_stream_response_logprobs(
        self,
        token_ids: Sequence[int],
868
        logprobs: SampleLogprobs | None,
869
        tokenizer: TokenizerLike,
870
        top_logprobs: int | None = None,
871
    ) -> list[response_text_delta_event.Logprob]:
872
873
874
875
876
877
        lgs = self._create_response_logprobs(
            token_ids=token_ids,
            logprobs=logprobs,
            tokenizer=tokenizer,
            top_logprobs=top_logprobs,
        )
878
879
880
881
882
883
        return [
            response_text_delta_event.Logprob(
                token=lg.token,
                logprob=lg.logprob,
                top_logprobs=[
                    response_text_delta_event.LogprobTopLogprob(
884
885
                        token=tl.token, logprob=tl.logprob
                    )
886
                    for tl in lg.top_logprobs
887
888
889
                ],
            )
            for lg in lgs
890
891
        ]

892
893
894
895
    def _make_response_output_items(
        self,
        request: ResponsesRequest,
        final_output: CompletionOutput,
896
        tokenizer: TokenizerLike,
897
    ) -> list[ResponseOutputItem]:
898
899
        # Log complete response if output logging is enabled
        if self.enable_log_outputs and self.request_logger:
900
901
902
903
904
905
906
907
            self.request_logger.log_outputs(
                request_id=request.request_id,
                outputs=final_output.text,
                output_token_ids=final_output.token_ids,
                finish_reason=final_output.finish_reason,
                is_streaming=False,
                delta=False,
            )
908

909
910
911
912
913
914
915
916
        # Compute logprobs if requested
        logprobs = None
        if request.is_include_output_logprobs() and final_output.logprobs:
            logprobs = self._create_response_logprobs(
                token_ids=final_output.token_ids,
                logprobs=final_output.logprobs,
                tokenizer=tokenizer,
                top_logprobs=request.top_logprobs,
917
            )
918

919
920
921
922
923
924
925
926
927
928
929
930
931
932
        # Use parser to extract and create response output items
        if self.parser:
            parser = self.parser(tokenizer)
            return parser.extract_response_outputs(
                model_output=final_output.text,
                request=request,
                enable_auto_tools=self.enable_auto_tools,
                tool_call_id_type=self.tool_call_id_type,
                logprobs=logprobs,
            )

        # Fallback when no parser is configured
        return [
            ResponseOutputMessage(
933
                id=f"msg_{random_uuid()}",
934
935
936
937
938
939
940
941
942
943
                content=[
                    ResponseOutputText(
                        text=final_output.text,
                        annotations=[],
                        type="output_text",
                        logprobs=logprobs,
                    )
                ]
                if final_output.text
                else [],
944
945
946
947
                role="assistant",
                status="completed",
                type="message",
            )
948
        ]
949
950
951
952
953

    def _make_response_output_items_with_harmony(
        self,
        context: HarmonyContext,
    ) -> list[ResponseOutputItem]:
954
        output_items: list[ResponseOutputItem] = []
955
956
957
958
959
960
961
962
963
        num_init_messages = context.num_init_messages
        for msg in context.messages[num_init_messages:]:
            output_items.extend(parse_output_message(msg))
        # Handle the generation stopped in the middle (if any).
        last_items = parse_remaining_state(context.parser)
        if last_items:
            output_items.extend(last_items)
        return output_items

964
965
966
    def _extract_system_message_from_request(
        self, request: ResponsesRequest
    ) -> str | None:
967
968
969
970
971
972
973
        system_msg = None
        if not isinstance(request.input, str):
            for response_msg in request.input:
                if (
                    isinstance(response_msg, dict)
                    and response_msg.get("role") == "system"
                ):
974
975
976
977
978
979
980
981
982
983
984
                    content = response_msg.get("content")
                    if isinstance(content, str):
                        system_msg = content
                    elif isinstance(content, list):
                        for param in content:
                            if (
                                isinstance(param, dict)
                                and param.get("type") == "input_text"
                            ):
                                system_msg = param.get("text")
                                break
985
986
987
                    break
        return system_msg

988
    def _construct_harmony_system_input_message(
989
        self, request: ResponsesRequest, with_custom_tools: bool, tool_types: set[str]
990
    ) -> OpenAIHarmonyMessage:
991
992
        model_identity = self._extract_system_message_from_request(request)

993
        reasoning_effort = request.reasoning.effort if request.reasoning else None
994
995
996
997
998
999
1000
1001
1002
1003
1004

        # Extract allowed_tools from MCP tool requests
        allowed_tools_map = _extract_allowed_tools_from_mcp_requests(request.tools)

        # Get filtered tool descriptions first.
        # If get_tool_description returns None (due to filtering), the tool is disabled.
        browser_description = (
            self.tool_server.get_tool_description(
                "browser", allowed_tools_map.get("web_search_preview")
            )
            if "web_search_preview" in tool_types
1005
1006
            and self.tool_server is not None
            and self.tool_server.has_tool("browser")
1007
            else None
1008
        )
1009
1010
1011
1012
1013
        python_description = (
            self.tool_server.get_tool_description(
                "python", allowed_tools_map.get("code_interpreter")
            )
            if "code_interpreter" in tool_types
1014
1015
            and self.tool_server is not None
            and self.tool_server.has_tool("python")
1016
            else None
1017
        )
1018
1019
1020
1021
1022
        container_description = (
            self.tool_server.get_tool_description(
                "container", allowed_tools_map.get("container")
            )
            if "container" in tool_types
1023
1024
            and self.tool_server is not None
            and self.tool_server.has_tool("container")
1025
            else None
1026
        )
1027

1028
        sys_msg = get_system_message(
1029
            model_identity=model_identity,
1030
            reasoning_effort=reasoning_effort,
1031
1032
1033
            browser_description=browser_description,
            python_description=python_description,
            container_description=container_description,
1034
1035
1036
1037
1038
            instructions=request.instructions,
            with_custom_tools=with_custom_tools,
        )
        return sys_msg

1039
1040
1041
    def _construct_input_messages_with_harmony(
        self,
        request: ResponsesRequest,
1042
        prev_response: ResponsesResponse | None,
1043
1044
1045
1046
    ) -> list[OpenAIHarmonyMessage]:
        messages: list[OpenAIHarmonyMessage] = []
        if prev_response is None:
            # New conversation.
1047
            tool_types = extract_tool_types(request.tools)
1048
            with_custom_tools = has_custom_tools(tool_types)
1049
1050
1051

            sys_msg = self._construct_harmony_system_input_message(
                request, with_custom_tools, tool_types
1052
1053
            )
            messages.append(sys_msg)
1054
1055
            if with_custom_tools:
                dev_msg = get_developer_message(
1056
1057
                    instructions=request.instructions, tools=request.tools
                )
1058
                messages.append(dev_msg)
1059
1060
            messages += construct_harmony_previous_input_messages(request)

1061
1062
1063
1064
1065
        else:
            # Continue the previous conversation.
            # FIXME(woosuk): Currently, request params like reasoning and
            # instructions are ignored.
            prev_msgs = self.msg_store[prev_response.id]
1066
1067
1068
1069
1070
1071
1072
1073
1074

            # FIXME(woosuk): The slice-delete-reappend cycle below is
            # currently a no-op --- it removes messages then puts them all
            # back unfiltered. It may be intentionally deferred (see FIXME
            # above) or redundant if the Harmony encoder already strips
            # analysis messages at render time. If analysis messages need
            # to be dropped here, add a channel != "analysis" filter when
            # re-appending, similar to auto_drop_analysis_messages in
            # harmony_utils.py.
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
            if len(prev_msgs) > 0:
                last_msg = prev_msgs[-1]
                assert isinstance(last_msg, OpenAIHarmonyMessage)
                if last_msg.channel == "final":
                    prev_final_msg_idx = -1
                    for i in range(len(prev_msgs) - 2, -1, -1):
                        prev_msg_i = prev_msgs[i]
                        assert isinstance(prev_msg_i, OpenAIHarmonyMessage)
                        if prev_msg_i.channel == "final":
                            prev_final_msg_idx = i
                            break
1086
1087
                    recent_turn_msgs = prev_msgs[prev_final_msg_idx + 1 :]
                    del prev_msgs[prev_final_msg_idx + 1 :]
1088
1089
                    for msg in recent_turn_msgs:
                        assert isinstance(msg, OpenAIHarmonyMessage)
1090
                        prev_msgs.append(msg)
1091
1092
            messages.extend(prev_msgs)
        # Append the new input.
co63oc's avatar
co63oc committed
1093
        # Responses API supports simple text inputs without chat format.
1094
        if isinstance(request.input, str):
1095
1096
1097
1098
1099
            # Skip empty string input when previous_input_messages supplies
            # the full conversation history --- an empty trailing user message
            # confuses the model into thinking nothing was sent.
            if request.input or not request.previous_input_messages:
                messages.append(get_user_message(request.input))
1100
1101
1102
1103
1104
1105
        else:
            if prev_response is not None:
                prev_outputs = copy(prev_response.output)
            else:
                prev_outputs = []
            for response_msg in request.input:
1106
1107
1108
1109
                new_msg = parse_response_input(response_msg, prev_outputs)
                if new_msg.author.role != "system":
                    messages.append(new_msg)

1110
                # User passes in a tool call request and its output. We need
1111
1112
1113
1114
1115
1116
1117
                # to add the tool call request to prev_outputs so that the
                # parse_response_input can find the tool call request when
                # parsing the tool call output.
                if isinstance(response_msg, ResponseFunctionToolCall):
                    prev_outputs.append(response_msg)
        return messages

1118
1119
1120
1121
1122
1123
    async def _run_background_request_stream(
        self,
        request: ResponsesRequest,
        *args,
        **kwargs,
    ):
1124
        event_deque: deque[StreamingResponsesResponse] = deque()
1125
1126
1127
1128
        new_event_signal = asyncio.Event()
        self.event_store[request.request_id] = (event_deque, new_event_signal)
        response = None
        try:
1129
            generator = self.responses_stream_generator(request, *args, **kwargs)
1130
1131
1132
            async for event in generator:
                event_deque.append(event)
                new_event_signal.set()  # Signal new event available
1133
1134
        except GenerationError as e:
            response = self._convert_generation_error_to_response(e)
1135
        except Exception as e:
1136
            logger.exception("Background request failed for %s", request.request_id)
1137
            response = self.create_error_response(e)
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
        finally:
            new_event_signal.set()

        if response is not None and isinstance(response, ErrorResponse):
            # If the request has failed, update the status to "failed".
            response_id = request.request_id
            async with self.response_store_lock:
                stored_response = self.response_store.get(response_id)
                assert stored_response is not None
                if stored_response.status not in ("completed", "cancelled"):
                    stored_response.status = "failed"

1150
1151
1152
1153
1154
1155
1156
    async def _run_background_request(
        self,
        request: ResponsesRequest,
        *args,
        **kwargs,
    ):
        try:
1157
            response = await self.responses_full_generator(request, *args, **kwargs)
1158
1159
        except GenerationError as e:
            response = self._convert_generation_error_to_response(e)
1160
        except Exception as e:
1161
            logger.exception("Background request failed for %s", request.request_id)
1162
            response = self.create_error_response(e)
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172

        if isinstance(response, ErrorResponse):
            # If the request has failed, update the status to "failed".
            response_id = request.request_id
            async with self.response_store_lock:
                stored_response = self.response_store.get(response_id)
                assert stored_response is not None
                if stored_response.status not in ("completed", "cancelled"):
                    stored_response.status = "failed"

1173
1174
1175
    async def responses_background_stream_generator(
        self,
        response_id: str,
1176
        starting_after: int | None = None,
1177
    ) -> AsyncGenerator[StreamingResponsesResponse, None]:
1178
        if response_id not in self.event_store:
1179
1180
1181
1182
1183
            raise VLLMValidationError(
                f"Unknown response_id: {response_id}",
                parameter="response_id",
                value=response_id,
            )
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195

        event_deque, new_event_signal = self.event_store[response_id]
        start_index = 0 if starting_after is None else starting_after + 1
        current_index = start_index

        while True:
            new_event_signal.clear()

            # Yield existing events from start_index
            while current_index < len(event_deque):
                event = event_deque[current_index]
                yield event
1196
                if getattr(event, "type", "unknown") == "response.completed":
1197
                    return
1198
1199
1200
1201
                current_index += 1

            await new_event_signal.wait()

1202
1203
1204
    async def retrieve_responses(
        self,
        response_id: str,
1205
1206
1207
1208
1209
1210
1211
        starting_after: int | None,
        stream: bool | None,
    ) -> (
        ErrorResponse
        | ResponsesResponse
        | AsyncGenerator[StreamingResponsesResponse, None]
    ):
1212
1213
1214
1215
1216
        async with self.response_store_lock:
            response = self.response_store.get(response_id)

        if response is None:
            return self._make_not_found_error(response_id)
1217
1218
1219
1220
1221
1222

        if stream:
            return self.responses_background_stream_generator(
                response_id,
                starting_after,
            )
1223
1224
1225
1226
1227
        return response

    async def cancel_responses(
        self,
        response_id: str,
1228
    ) -> ErrorResponse | ResponsesResponse:
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
        async with self.response_store_lock:
            response = self.response_store.get(response_id)
            if response is None:
                return self._make_not_found_error(response_id)

            prev_status = response.status
            if prev_status not in ("queued", "in_progress"):
                return self.create_error_response(
                    err_type="invalid_request_error",
                    message="Cannot cancel a synchronous response.",
1239
                    param="response_id",
1240
1241
1242
1243
1244
1245
                )

            # Update the status to "cancelled".
            response.status = "cancelled"

        # Abort the request.
1246
        if task := self.background_tasks.get(response_id):
1247
1248
1249
1250
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
1251
                logger.exception("Background task for %s was cancelled", response_id)
1252
1253
1254
1255
1256
1257
1258
        return response

    def _make_not_found_error(self, response_id: str) -> ErrorResponse:
        return self.create_error_response(
            err_type="invalid_request_error",
            message=f"Response with id '{response_id}' not found.",
            status_code=HTTPStatus.NOT_FOUND,
1259
            param="response_id",
1260
        )
1261
1262
1263
1264

    def _make_store_not_supported_error(self) -> ErrorResponse:
        return self.create_error_response(
            err_type="invalid_request_error",
1265
1266
1267
1268
1269
1270
            message=(
                "`store=True` (default) is not supported. Please set "
                "`store=False` in Responses API or set "
                "`VLLM_ENABLE_RESPONSES_API_STORE=1` in the env var when "
                "starting the vLLM server."
            ),
1271
            status_code=HTTPStatus.BAD_REQUEST,
1272
            param="store",
1273
        )
1274

1275
    async def _process_simple_streaming_events(
1276
1277
1278
        self,
        request: ResponsesRequest,
        sampling_params: SamplingParams,
1279
        result_generator: AsyncIterator[ConversationContext | None],
1280
1281
        context: ConversationContext,
        model_name: str,
1282
        tokenizer: TokenizerLike,
1283
        request_metadata: RequestResponseMetadata,
1284
        created_time: int,
1285
        _increment_sequence_number_and_return: Callable[
1286
1287
            [StreamingResponsesResponse], StreamingResponsesResponse
        ],
1288
    ) -> AsyncGenerator[StreamingResponsesResponse, None]:
1289
1290
1291
1292
        current_content_index = 0
        current_output_index = 0
        current_item_id = ""
        reasoning_parser = None
1293
1294
        if self.parser and self.parser.reasoning_parser_cls:
            reasoning_parser = self.parser.reasoning_parser_cls(tokenizer)
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
        previous_text = ""
        previous_token_ids: list[int] = []
        first_delta_sent = False
        previous_delta_messages: list[DeltaMessage] = []
        async for ctx in result_generator:
            assert isinstance(ctx, SimpleContext)
            if ctx.last_output is None:
                continue
            if ctx.last_output.outputs:
                output = ctx.last_output.outputs[0]
1305
1306
                # finish_reason='error' indicates a retryable error
                self._raise_if_error(output.finish_reason, request.request_id)
1307
                if reasoning_parser:
1308
1309
1310
1311
1312
1313
1314
                    delta_message = reasoning_parser.extract_reasoning_streaming(
                        previous_text=previous_text,
                        current_text=previous_text + output.text,
                        delta_text=output.text,
                        previous_token_ids=previous_token_ids,
                        current_token_ids=previous_token_ids + output.token_ids,
                        delta_token_ids=output.token_ids,
1315
1316
                    )
                else:
1317
1318
1319
                    delta_message = DeltaMessage(
                        content=output.text,
                    )
1320
1321
1322
1323
1324
1325
                previous_text += output.text
                previous_token_ids += output.token_ids
                if not delta_message:
                    continue
                if not first_delta_sent:
                    current_item_id = str(uuid.uuid4())
1326
                    if delta_message.reasoning:
1327
                        yield _increment_sequence_number_and_return(
1328
1329
1330
1331
                            ResponseOutputItemAddedEvent(
                                type="response.output_item.added",
                                sequence_number=-1,
                                output_index=current_output_index,
1332
                                item=ResponseReasoningItem(
1333
1334
1335
1336
1337
                                    type="reasoning",
                                    id=current_item_id,
                                    summary=[],
                                    status="in_progress",
                                ),
1338
1339
                            )
                        )
1340
                    else:
1341
                        yield _increment_sequence_number_and_return(
1342
1343
1344
1345
                            ResponseOutputItemAddedEvent(
                                type="response.output_item.added",
                                sequence_number=-1,
                                output_index=current_output_index,
1346
                                item=ResponseOutputMessage(
1347
1348
1349
1350
1351
1352
                                    id=current_item_id,
                                    type="message",
                                    role="assistant",
                                    content=[],
                                    status="in_progress",
                                ),
1353
1354
                            )
                        )
1355
                    yield _increment_sequence_number_and_return(
1356
                        ResponseContentPartAddedEvent(
1357
1358
1359
1360
1361
                            type="response.content_part.added",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item_id=current_item_id,
                            content_index=current_content_index,
1362
                            part=ResponseOutputText(
1363
1364
1365
1366
1367
                                type="output_text",
                                text="",
                                annotations=[],
                                logprobs=[],
                            ),
1368
1369
                        )
                    )
1370
1371
1372
1373
1374
1375
                    current_content_index += 1
                    first_delta_sent = True
                # todo(kebe7jun) tool call support

                # check delta message and previous delta message are
                # same as content or reasoning content
1376
1377
                if (
                    previous_delta_messages
1378
                    and previous_delta_messages[-1].reasoning is not None
1379
1380
                    and delta_message.content is not None
                ):
1381
1382
                    # from reasoning to normal content, send done
                    # event for reasoning
1383
                    reason_content = "".join(
1384
                        pm.reasoning
1385
                        for pm in previous_delta_messages
1386
                        if pm.reasoning is not None
1387
                    )
1388
                    yield _increment_sequence_number_and_return(
1389
1390
1391
1392
1393
1394
1395
                        ResponseReasoningTextDoneEvent(
                            type="response.reasoning_text.done",
                            item_id=current_item_id,
                            sequence_number=-1,
                            output_index=current_output_index,
                            content_index=current_content_index,
                            text=reason_content,
1396
1397
                        )
                    )
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
                    current_content_index = 0
                    reasoning_item = ResponseReasoningItem(
                        type="reasoning",
                        content=[
                            ResponseReasoningTextContent(
                                text=reason_content,
                                type="reasoning_text",
                            ),
                        ],
                        status="completed",
                        id=current_item_id,
                        summary=[],
                    )
1411
                    yield _increment_sequence_number_and_return(
1412
1413
1414
1415
1416
                        ResponseOutputItemDoneEvent(
                            type="response.output_item.done",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item=reasoning_item,
1417
1418
                        )
                    )
1419
                    yield _increment_sequence_number_and_return(
1420
                        ResponseOutputItemAddedEvent(
1421
1422
1423
                            type="response.output_item.added",
                            sequence_number=-1,
                            output_index=current_output_index,
1424
                            item=ResponseOutputMessage(
1425
1426
1427
1428
1429
1430
                                id=current_item_id,
                                type="message",
                                role="assistant",
                                content=[],
                                status="in_progress",
                            ),
1431
1432
                        )
                    )
1433
1434
                    current_output_index += 1
                    current_item_id = str(uuid.uuid4())
1435
                    yield _increment_sequence_number_and_return(
1436
                        ResponseContentPartAddedEvent(
1437
1438
1439
1440
1441
                            type="response.content_part.added",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item_id=current_item_id,
                            content_index=current_content_index,
1442
                            part=ResponseOutputText(
1443
1444
1445
1446
1447
                                type="output_text",
                                text="",
                                annotations=[],
                                logprobs=[],
                            ),
1448
1449
                        )
                    )
1450
1451
1452
                    current_content_index += 1
                    # reset previous delta messages
                    previous_delta_messages = []
1453

1454
                if delta_message.reasoning is not None:
1455
                    yield _increment_sequence_number_and_return(
1456
1457
1458
1459
1460
1461
                        ResponseReasoningTextDeltaEvent(
                            type="response.reasoning_text.delta",
                            sequence_number=-1,
                            content_index=current_content_index,
                            output_index=current_output_index,
                            item_id=current_item_id,
1462
                            delta=delta_message.reasoning,
1463
1464
                        )
                    )
1465
                elif delta_message.content is not None:
1466
                    yield _increment_sequence_number_and_return(
1467
                        ResponseTextDeltaEvent(
1468
1469
1470
1471
1472
1473
                            type="response.output_text.delta",
                            sequence_number=-1,
                            content_index=current_content_index,
                            output_index=current_output_index,
                            item_id=current_item_id,
                            delta=delta_message.content,
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
                            logprobs=(
                                self._create_stream_response_logprobs(
                                    token_ids=output.token_ids,
                                    logprobs=output.logprobs,
                                    tokenizer=tokenizer,
                                    top_logprobs=request.top_logprobs,
                                )
                                if request.is_include_output_logprobs()
                                else []
                            ),
1484
1485
                        )
                    )
1486
1487
1488
1489
                current_content_index += 1

                previous_delta_messages.append(delta_message)
        if previous_delta_messages:
1490
            if previous_delta_messages[-1].reasoning is not None:
1491
                reason_content = "".join(
1492
                    pm.reasoning
1493
                    for pm in previous_delta_messages
1494
                    if pm.reasoning is not None
1495
                )
1496
                yield _increment_sequence_number_and_return(
1497
1498
1499
1500
1501
1502
1503
                    ResponseReasoningTextDoneEvent(
                        type="response.reasoning_text.done",
                        item_id=current_item_id,
                        sequence_number=-1,
                        output_index=current_output_index,
                        content_index=current_content_index,
                        text=reason_content,
1504
1505
                    )
                )
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
                current_content_index += 1
                reasoning_item = ResponseReasoningItem(
                    type="reasoning",
                    content=[
                        ResponseReasoningTextContent(
                            text=reason_content,
                            type="reasoning_text",
                        ),
                    ],
                    status="completed",
                    id=current_item_id,
                    summary=[],
                )
1519
                yield _increment_sequence_number_and_return(
1520
1521
1522
1523
1524
                    ResponseOutputItemDoneEvent(
                        type="response.output_item.done",
                        sequence_number=-1,
                        output_index=current_output_index,
                        item=reasoning_item,
1525
1526
                    )
                )
1527
            elif previous_delta_messages[-1].content is not None:
1528
1529
1530
1531
1532
                final_content = "".join(
                    pm.content
                    for pm in previous_delta_messages
                    if pm.content is not None
                )
1533
                yield _increment_sequence_number_and_return(
1534
                    ResponseTextDoneEvent(
1535
1536
1537
1538
1539
1540
1541
                        type="response.output_text.done",
                        sequence_number=-1,
                        output_index=current_output_index,
                        content_index=current_content_index,
                        text=final_content,
                        logprobs=[],
                        item_id=current_item_id,
1542
1543
                    )
                )
1544
1545
1546
1547
1548
1549
                current_content_index += 1
                part = ResponseOutputText(
                    text=final_content,
                    type="output_text",
                    annotations=[],
                )
1550
                yield _increment_sequence_number_and_return(
1551
                    ResponseContentPartDoneEvent(
1552
1553
1554
1555
1556
1557
                        type="response.content_part.done",
                        sequence_number=-1,
                        item_id=current_item_id,
                        output_index=current_output_index,
                        content_index=current_content_index,
                        part=part,
1558
1559
                    )
                )
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
                current_content_index += 1
                item = ResponseOutputMessage(
                    type="message",
                    role="assistant",
                    content=[
                        part,
                    ],
                    status="completed",
                    id=current_item_id,
                    summary=[],
                )
1571
                yield _increment_sequence_number_and_return(
1572
1573
1574
1575
1576
                    ResponseOutputItemDoneEvent(
                        type="response.output_item.done",
                        sequence_number=-1,
                        output_index=current_output_index,
                        item=item,
1577
1578
                    )
                )
1579
1580
1581
1582
1583

    async def _process_harmony_streaming_events(
        self,
        request: ResponsesRequest,
        sampling_params: SamplingParams,
1584
        result_generator: AsyncIterator[ConversationContext | None],
1585
1586
        context: ConversationContext,
        model_name: str,
1587
        tokenizer: TokenizerLike,
1588
1589
        request_metadata: RequestResponseMetadata,
        created_time: int,
1590
        _increment_sequence_number_and_return: Callable[
1591
1592
            [StreamingResponsesResponse], StreamingResponsesResponse
        ],
1593
    ) -> AsyncGenerator[StreamingResponsesResponse, None]:
1594
1595
        state = HarmonyStreamingState()

1596
1597
1598
        async for ctx in result_generator:
            assert isinstance(ctx, StreamingHarmonyContext)

1599
1600
1601
            # finish_reason='error' indicates a retryable error
            self._raise_if_error(ctx.finish_reason, request.request_id)

1602
1603
1604
            if ctx.is_expecting_start():
                if len(ctx.parser.messages) > 0:
                    previous_item = ctx.parser.messages[-1]
1605
                    for event in emit_previous_item_done_events(previous_item, state):
1606
1607
1608
1609
                        yield _increment_sequence_number_and_return(event)
                state.reset_for_new_item()

            # Stream the output of a harmony message
1610
            for event in emit_content_delta_events(ctx, state):
1611
1612
1613
                yield _increment_sequence_number_and_return(event)

            # Stream tool call outputs
1614
            for event in emit_tool_action_events(ctx, state, self.tool_server):
1615
                yield _increment_sequence_number_and_return(event)
1616

1617
1618
1619
1620
    async def responses_stream_generator(
        self,
        request: ResponsesRequest,
        sampling_params: SamplingParams,
1621
        result_generator: AsyncIterator[ConversationContext | None],
1622
1623
        context: ConversationContext,
        model_name: str,
1624
        tokenizer: TokenizerLike,
1625
        request_metadata: RequestResponseMetadata,
1626
        created_time: int | None = None,
1627
    ) -> AsyncGenerator[StreamingResponsesResponse, None]:
1628
1629
1630
1631
1632
        # TODO:
        # 1. Handle disconnect

        created_time = created_time or int(time.time())

1633
1634
        sequence_number = 0

1635
        def _increment_sequence_number_and_return(
1636
            event: StreamingResponsesResponse,
1637
        ) -> StreamingResponsesResponse:
1638
1639
            nonlocal sequence_number
            # Set sequence_number if the event has this attribute
1640
            if hasattr(event, "sequence_number"):
1641
1642
                event.sequence_number = sequence_number
            sequence_number += 1
1643
            return event
1644

1645
        async with AsyncExitStack() as exit_stack:
1646
            if self.use_harmony:
1647
1648
                # TODO: in streaming, we noticed this bug:
                # https://github.com/vllm-project/vllm/issues/25697
1649
                await self._initialize_tool_sessions(request, context, exit_stack)
1650
1651
1652
                processer = self._process_harmony_streaming_events
            else:
                processer = self._process_simple_streaming_events
1653
            # TODO Hanchen make sampling params to include the structural tag
1654
1655
1656
1657
1658
1659
1660
1661
1662
1663

            initial_response = ResponsesResponse.from_request(
                request,
                sampling_params,
                model_name=model_name,
                created_time=created_time,
                output=[],
                status="in_progress",
                usage=None,
            ).model_dump()
1664
            yield _increment_sequence_number_and_return(
1665
1666
1667
1668
                ResponseCreatedEvent(
                    type="response.created",
                    sequence_number=-1,
                    response=initial_response,
1669
1670
                )
            )
1671
            yield _increment_sequence_number_and_return(
1672
1673
1674
1675
                ResponseInProgressEvent(
                    type="response.in_progress",
                    sequence_number=-1,
                    response=initial_response,
1676
1677
                )
            )
1678

1679
1680
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
            try:
                async for event_data in processer(
                    request,
                    sampling_params,
                    result_generator,
                    context,
                    model_name,
                    tokenizer,
                    request_metadata,
                    created_time,
                    _increment_sequence_number_and_return,
                ):
                    yield event_data
            except GenerationError as e:
                error_json = self._convert_generation_error_to_streaming_response(e)
                yield _increment_sequence_number_and_return(
                    TypeAdapter(StreamingResponsesResponse).validate_json(error_json)
                )
                return
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714

            async def empty_async_generator():
                # A hack to trick Python to think this is a generator but
                # in fact it immediately returns.
                if False:
                    yield

            final_response = await self.responses_full_generator(
                request,
                sampling_params,
                empty_async_generator(),
                context,
                model_name,
                tokenizer,
                request_metadata,
                created_time=created_time,
            )
1715
            yield _increment_sequence_number_and_return(
1716
                ResponseCompletedEvent(
1717
1718
                    type="response.completed",
                    sequence_number=-1,
1719
                    response=final_response,
1720
1721
                )
            )