"vscode:/vscode.git/clone" did not exist on "34a8559be7473e3ba34e9b9e13827e60651e3b41"
serving_responses.py 84.1 KB
Newer Older
1
2
3
4
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project

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

import jinja2
from fastapi import Request
17
18
19
20
21
22
from openai.types.responses import (
    ResponseCodeInterpreterCallCodeDeltaEvent,
    ResponseCodeInterpreterCallCodeDoneEvent,
    ResponseCodeInterpreterCallCompletedEvent,
    ResponseCodeInterpreterCallInProgressEvent,
    ResponseCodeInterpreterCallInterpretingEvent,
23
24
25
    ResponseCodeInterpreterToolCallParam,
    ResponseContentPartAddedEvent,
    ResponseContentPartDoneEvent,
26
27
    ResponseFunctionCallArgumentsDeltaEvent,
    ResponseFunctionCallArgumentsDoneEvent,
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
    ResponseFunctionToolCall,
    ResponseFunctionWebSearch,
    ResponseOutputItem,
    ResponseOutputItemAddedEvent,
    ResponseOutputItemDoneEvent,
    ResponseOutputMessage,
    ResponseOutputText,
    ResponseReasoningItem,
    ResponseReasoningTextDeltaEvent,
    ResponseReasoningTextDoneEvent,
    ResponseStatus,
    ResponseTextDeltaEvent,
    ResponseTextDoneEvent,
    ResponseWebSearchCallCompletedEvent,
    ResponseWebSearchCallInProgressEvent,
    ResponseWebSearchCallSearchingEvent,
    response_function_web_search,
    response_text_delta_event,
)
from openai.types.responses.response_output_text import Logprob, LogprobTopLogprob
48
from openai.types.responses.response_reasoning_item import (
49
50
    Content as ResponseReasoningTextContent,
)
51
from openai_harmony import Message as OpenAIHarmonyMessage
52

53
from vllm import envs
54
from vllm.engine.protocol import EngineClient
55
56
57
58
59
60
61
62
63
64
from vllm.entrypoints.chat_utils import (
    ChatCompletionMessageParam,
    ChatTemplateContentFormatOption,
)
from vllm.entrypoints.context import (
    ConversationContext,
    HarmonyContext,
    SimpleContext,
    StreamingHarmonyContext,
)
65
from vllm.entrypoints.harmony_utils import (
66
    construct_harmony_previous_input_messages,
67
68
69
70
71
72
73
74
75
76
    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,
)
77
from vllm.entrypoints.logger import RequestLogger
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
from vllm.entrypoints.openai.protocol import (
    DeltaMessage,
    ErrorResponse,
    InputTokensDetails,
    OutputTokensDetails,
    RequestResponseMetadata,
    ResponseCompletedEvent,
    ResponseCreatedEvent,
    ResponseInProgressEvent,
    ResponseReasoningPartAddedEvent,
    ResponseReasoningPartDoneEvent,
    ResponsesRequest,
    ResponsesResponse,
    ResponseUsage,
    StreamingResponsesResponse,
)
94
95
from vllm.entrypoints.openai.serving_engine import OpenAIServing
from vllm.entrypoints.openai.serving_models import OpenAIServingModels
96
from vllm.entrypoints.tool_server import ToolServer
97
from vllm.inputs.data import TokensPrompt as EngineTokensPrompt
98
from vllm.logger import init_logger
99
100
from vllm.logprobs import Logprob as SampleLogprob
from vllm.logprobs import SampleLogprobs
101
from vllm.outputs import CompletionOutput
102
from vllm.sampling_params import SamplingParams, StructuredOutputsParams
103
104
105
106
107
108
109
110
111
112
113
114
from vllm.transformers_utils.tokenizer import AnyTokenizer
from vllm.utils import random_uuid

logger = init_logger(__name__)


class OpenAIServingResponses(OpenAIServing):
    def __init__(
        self,
        engine_client: EngineClient,
        models: OpenAIServingModels,
        *,
115
116
        request_logger: RequestLogger | None,
        chat_template: str | None,
117
118
119
120
        chat_template_content_format: ChatTemplateContentFormatOption,
        return_tokens_as_token_ids: bool = False,
        reasoning_parser: str = "",
        enable_auto_tools: bool = False,
121
122
        tool_parser: str | None = None,
        tool_server: ToolServer | None = None,
123
124
        enable_prompt_tokens_details: bool = False,
        enable_force_include_usage: bool = False,
125
        enable_log_outputs: bool = False,
126
        log_error_stack: bool = False,
127
128
129
130
131
132
    ) -> None:
        super().__init__(
            engine_client=engine_client,
            models=models,
            request_logger=request_logger,
            return_tokens_as_token_ids=return_tokens_as_token_ids,
133
            log_error_stack=log_error_stack,
134
135
136
137
        )

        self.chat_template = chat_template
        self.chat_template_content_format: Final = chat_template_content_format
138
        self.enable_log_outputs = enable_log_outputs
139

140
141
        self.reasoning_parser = self._get_reasoning_parser(
            reasoning_parser_name=reasoning_parser
142
        )
143
144
        self.enable_prompt_tokens_details = enable_prompt_tokens_details
        self.enable_force_include_usage = enable_force_include_usage
145
        self.default_sampling_params = self.model_config.get_diff_sampling_param()
146
147
148
        if self.default_sampling_params:
            source = self.model_config.generation_config
            source = "model" if source == "auto" else source
149
150
151
152
153
            logger.info(
                "Using default chat sampling params from %s: %s",
                source,
                self.default_sampling_params,
            )
154

155
156
157
158
159
        # 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.
160
        self.enable_store = envs.VLLM_ENABLE_RESPONSES_API_STORE
161
162
163
164
        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 "
165
166
                "the store."
            )
167

168
        self.use_harmony = self.model_config.hf_config.model_type == "gpt_oss"
169
        if self.use_harmony:
170
171
172
173
            logger.warning(
                "For gpt-oss, we ignore --enable-auto-tool-choice "
                "and always enable tool use."
            )
174
175
176
177
178
            # 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(
179
180
                get_stop_tokens_for_assistant_actions()
            )
181
182
183
184
185

        # set up tool use
        self.enable_auto_tools: bool = enable_auto_tools
        if self.enable_auto_tools:
            logger.info(
186
                '"auto" tool choice has been enabled please note that while'
187
                " the parallel_tool_calls client option is preset for "
188
189
                "compatibility reasons, it will be ignored."
            )
190

191
        # HACK(woosuk): This is a hack. We should use a better store.
192
193
        # FIXME: If enable_store=True, this may cause a memory leak since we
        # never remove responses from the store.
194
195
196
197
        self.response_store: dict[str, ResponsesResponse] = {}
        self.response_store_lock = asyncio.Lock()

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

202
203
204
        # 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.
205
206
207
        self.event_store: dict[
            str, tuple[deque[StreamingResponsesResponse], asyncio.Event]
        ] = {}
208

209
210
        self.background_tasks: dict[str, asyncio.Task] = {}

211
212
        self.tool_server = tool_server

213
    def _validate_generator_input(
214
        self, engine_prompt: EngineTokensPrompt
215
    ) -> ErrorResponse | None:
216
217
218
219
220
221
        """Add validations to the input to the generator here."""
        if self.max_model_len <= len(engine_prompt["prompt_token_ids"]):
            error_message = (
                "The engine prompt length"
                f" {len(engine_prompt['prompt_token_ids'])} "
                f"exceeds the max_model_len {self.max_model_len}. "
222
223
                "Please reduce prompt."
            )
224
225
226
227
228
229
230
            return self.create_error_response(
                err_type="invalid_request_error",
                message=error_message,
                status_code=HTTPStatus.BAD_REQUEST,
            )
        return None

231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
    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,
            )
        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,
            )
252
253
254
255
256
257
258
        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,
            )
259
260
        return None

261
262
263
    async def create_responses(
        self,
        request: ResponsesRequest,
264
265
266
267
268
269
        raw_request: Request | None = None,
    ) -> (
        AsyncGenerator[StreamingResponsesResponse, None]
        | ResponsesResponse
        | ErrorResponse
    ):
270
271
272
273
        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
274
275
276
        maybe_validation_error = self._validate_create_responses_input(request)
        if maybe_validation_error is not None:
            return maybe_validation_error
277
278
279
280
281
282
283

        # 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

284
        if request.store and not self.enable_store:
285
286
287
288
289
290
291
            # 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
292

293
294
295
296
297
298
299
300
301
302
303
        # 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:
304
            lora_request = self._maybe_get_adapters(request)
305
            model_name = self.models.model_name(lora_request)
306
            tokenizer = await self.engine_client.get_tokenizer()
307

308
309
            if self.use_harmony:
                messages, request_prompts, engine_prompts = (
310
311
                    self._make_request_with_harmony(request, prev_response)
                )
312
            else:
313
314
315
                messages, request_prompts, engine_prompts = await self._make_request(
                    request, prev_response, tokenizer
                )
316

317
318
319
320
321
322
323
        except (
            ValueError,
            TypeError,
            RuntimeError,
            jinja2.TemplateError,
            NotImplementedError,
        ) as e:
324
325
326
            logger.exception("Error in preprocessing prompt inputs")
            return self.create_error_response(f"{e} {e.__cause__}")

327
        request_metadata = RequestResponseMetadata(request_id=request.request_id)
328
329
330
331
        if raw_request:
            raw_request.state.request_metadata = request_metadata

        # Schedule the request and get the result generator.
332
        generators: list[AsyncGenerator[ConversationContext, None]] = []
333
334
335
336
337
338
339

        builtin_tool_list: list[str] = []
        if self.use_harmony and self.tool_server is not None:
            if self.tool_server.has_tool("browser"):
                builtin_tool_list.append("browser")
            if self.tool_server.has_tool("python"):
                builtin_tool_list.append("python")
340
341
            if self.tool_server.has_tool("container"):
                builtin_tool_list.append("container")
342

343
344
345
346
347
348
349
        if self.tool_server is not None:
            available_tools = builtin_tool_list
        else:
            assert len(builtin_tool_list) == 0
            available_tools = []
        try:
            for i, engine_prompt in enumerate(engine_prompts):
350
351
352
353
                maybe_error = self._validate_generator_input(engine_prompt)
                if maybe_error is not None:
                    return maybe_error

354
                default_max_tokens = self.max_model_len - len(
355
356
                    engine_prompt["prompt_token_ids"]
                )
357

358
                sampling_params = request.to_sampling_params(
359
360
                    default_max_tokens, self.default_sampling_params
                )
361

362
363
364
365
366
                trace_headers = (
                    None
                    if raw_request is None
                    else await self._get_trace_headers(raw_request.headers)
                )
367
368
369
370

                context: ConversationContext
                if self.use_harmony:
                    if request.stream:
371
                        context = StreamingHarmonyContext(messages, available_tools)
372
373
374
375
                    else:
                        context = HarmonyContext(messages, available_tools)
                else:
                    context = SimpleContext()
376
377
378
379
380
381
382
383
384
385
386
387
388

                if self.reasoning_parser is not None:
                    reasoning_parser = self.reasoning_parser(tokenizer)
                    if sampling_params.structured_outputs is None:
                        sampling_params.structured_outputs = StructuredOutputsParams()
                    struct_out = sampling_params.structured_outputs
                    if struct_out.all_non_structural_tag_constraints_none():
                        sampling_params.structured_outputs.structural_tag = (
                            reasoning_parser.prepare_structured_tag(
                                sampling_params.structured_outputs.structural_tag,
                                self.tool_server,
                            )
                        )
389
390
391
392
393
394
395
396
397
                generator = self._generate_with_builtin_tools(
                    request_id=request.request_id,
                    request_prompt=request_prompts[i],
                    engine_prompt=engine_prompt,
                    sampling_params=sampling_params,
                    context=context,
                    lora_request=lora_request,
                    priority=request.priority,
                    trace_headers=trace_headers,
398
                )
399
400
401
402
                generators.append(generator)
        except ValueError as e:
            # TODO: Use a vllm-specific Validation Error
            return self.create_error_response(str(e))
403

404
        assert len(generators) == 1
405
        (result_generator,) = generators
406
407
408
409

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

411
412
413
414
415
416
417
418
419
420
421
422
423
        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
424

425
            # Run the request in the background.
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
            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}",
                )
454

455
456
457
458
            # For cleanup.
            response_id = response.id
            self.background_tasks[response_id] = task
            task.add_done_callback(
459
460
                lambda _: self.background_tasks.pop(response_id, None)
            )
461
462

            if request.stream:
463
                return self.responses_background_stream_generator(request.request_id)
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
            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,
            )
        except Exception as e:
            return self.create_error_response(str(e))
489

490
491
492
    async def _make_request(
        self,
        request: ResponsesRequest,
493
        prev_response: ResponsesResponse | None,
494
495
        tokenizer: AnyTokenizer,
    ):
496
497
        if len(request.tools) > 0:
            raise NotImplementedError(
498
499
                "Tool use is not supported in Responses API without Harmony"
            )
500
501
502
503
504
505
506
507
508
509
510
511
512
513
        # Construct the input messages.
        messages = self._construct_input_messages(request, prev_response)
        _, request_prompts, engine_prompts = await self._preprocess_chat(
            request,
            tokenizer,
            messages,
            chat_template=self.chat_template,
            chat_template_content_format=self.chat_template_content_format,
        )
        return messages, request_prompts, engine_prompts

    def _make_request_with_harmony(
        self,
        request: ResponsesRequest,
514
        prev_response: ResponsesResponse | None,
515
516
517
    ):
        if request.tool_choice != "auto":
            raise NotImplementedError(
518
519
520
                "Only 'auto' tool_choice is supported in response API with Harmony"
            )
        messages = self._construct_input_messages_with_harmony(request, prev_response)
521
522
        prompt_token_ids = render_for_completion(messages)
        engine_prompt = EngineTokensPrompt(prompt_token_ids=prompt_token_ids)
523
524
525
526
527

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

528
529
        return messages, [prompt_token_ids], [engine_prompt]

530
531
532
533
534
535
    async def _initialize_tool_sessions(
        self,
        request: ResponsesRequest,
        context: ConversationContext,
        exit_stack: AsyncExitStack,
    ):
536
537
538
539
        # we should only initialize the tool session if the request needs tools
        if len(request.tools) == 0:
            return
        mcp_tools = {
540
            tool.server_label: tool for tool in request.tools if tool.type == "mcp"
541
        }
542
543
544
        await context.init_tool_sessions(
            self.tool_server, exit_stack, request.request_id, mcp_tools
        )
545

546
547
548
549
    async def responses_full_generator(
        self,
        request: ResponsesRequest,
        sampling_params: SamplingParams,
550
        result_generator: AsyncIterator[ConversationContext],
551
        context: ConversationContext,
552
553
554
        model_name: str,
        tokenizer: AnyTokenizer,
        request_metadata: RequestResponseMetadata,
555
556
        created_time: int | None = None,
    ) -> ErrorResponse | ResponsesResponse:
557
558
559
        if created_time is None:
            created_time = int(time.time())

560
561
        async with AsyncExitStack() as exit_stack:
            try:
562
                await self._initialize_tool_sessions(request, context, exit_stack)
563
564
565
566
567
568
569
                async for _ in result_generator:
                    pass
            except asyncio.CancelledError:
                return self.create_error_response("Client disconnected")
            except ValueError as e:
                # TODO: Use a vllm-specific Validation Error
                return self.create_error_response(str(e))
570

571
572
573
574
575
        # NOTE: Implementation of stauts is still WIP, but for now
        # we guarantee that if the status is not "completed", it is accurate.
        # "completed" is implemented as the "catch-all" for now.
        status: ResponseStatus = "completed"

576
577
        input_messages = None
        output_messages = None
578
579
580
        if self.use_harmony:
            assert isinstance(context, HarmonyContext)
            output = self._make_response_output_items_with_harmony(context)
581
            if request.enable_response_messages:
582
583
                input_messages = context.messages[: context.num_init_messages]
                output_messages = context.messages[context.num_init_messages :]
584
            num_tool_output_tokens = context.num_tool_output_tokens
585
586
587
588
589
590
591
            if len(output) > 0:
                if context.finish_reason == "length":
                    status = "incomplete"
                elif context.finish_reason == "abort":
                    status = "cancelled"
            else:
                status = "incomplete"
592
        else:
593
594
595
596
597
598
            assert isinstance(context, SimpleContext)
            final_res = context.last_output
            assert final_res is not None
            assert len(final_res.outputs) == 1
            final_output = final_res.outputs[0]

599
            output = self._make_response_output_items(request, final_output, tokenizer)
600

601
602
603
604
            # TODO: context for non-gptoss models doesn't use messages
            # so we can't get them out yet
            if request.enable_response_messages:
                raise NotImplementedError(
605
606
                    "enable_response_messages is currently only supported for gpt-oss"
                )
607
608
            # Calculate usage.
            assert final_res.prompt_token_ids is not None
609
610
            num_tool_output_tokens = 0

611
612
613
614
615
        assert isinstance(context, (SimpleContext, HarmonyContext))
        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
616
617
618
619

        usage = ResponseUsage(
            input_tokens=num_prompt_tokens,
            output_tokens=num_generated_tokens,
620
            total_tokens=num_prompt_tokens + num_generated_tokens,
621
622
623
624
625
626
627
628
629
            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
                ],
            ),
630
            output_tokens_details=OutputTokensDetails(
631
                reasoning_tokens=num_reasoning_tokens,
632
                tool_output_tokens=num_tool_output_tokens,
633
634
635
636
637
638
                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
                ],
639
            ),
640
641
642
643
        )
        response = ResponsesResponse.from_request(
            request,
            sampling_params,
644
645
            input_messages=input_messages,
            output_messages=output_messages,
646
647
648
            model_name=model_name,
            created_time=created_time,
            output=output,
649
            status=status,
650
651
652
653
654
655
656
            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.
657
                if stored_response is None or stored_response.status != "cancelled":
658
659
660
                    self.response_store[response.id] = response
        return response

661
662
663
664
665
666
    def _topk_logprobs(
        self,
        logprobs: dict[int, SampleLogprob],
        top_logprobs: int,
        tokenizer: AnyTokenizer,
    ) -> list[LogprobTopLogprob]:
667
668
669
670
671
        """Returns the top-k logprobs from the logprobs dictionary."""
        out = []
        for i, (token_id, _logprob) in enumerate(logprobs.items()):
            if i >= top_logprobs:
                break
672
673
674
675
676
            text = (
                _logprob.decoded_token
                if _logprob.decoded_token is not None
                else tokenizer.decode([token_id])
            )
677
678
679
680
681
            out.append(
                LogprobTopLogprob(
                    token=text,
                    logprob=max(_logprob.logprob, -9999.0),
                    bytes=list(text.encode("utf-8", errors="replace")),
682
683
                )
            )
684
685
686
        return out

    def _create_response_logprobs(
687
688
        self,
        token_ids: Sequence[int],
689
        logprobs: SampleLogprobs | None,
690
        tokenizer: AnyTokenizer,
691
        top_logprobs: int | None = None,
692
    ) -> list[Logprob]:
693
694
        assert logprobs is not None, "logprobs must be provided"
        assert len(token_ids) == len(logprobs), (
695
696
            "token_ids and logprobs.token_ids must have the same length"
        )
697
698
699
700
        out = []
        for i, token_id in enumerate(token_ids):
            logprob = logprobs[i]
            token_logprob = logprob[token_id]
701
702
703
704
705
            text = (
                token_logprob.decoded_token
                if token_logprob.decoded_token is not None
                else tokenizer.decode([token_id])
            )
706
707
708
709
710
            out.append(
                Logprob(
                    token=text,
                    logprob=max(token_logprob.logprob, -9999.0),
                    bytes=list(text.encode("utf-8", errors="replace")),
711
712
713
714
715
716
717
                    top_logprobs=(
                        self._topk_logprobs(
                            logprob, top_logprobs=top_logprobs, tokenizer=tokenizer
                        )
                        if top_logprobs
                        else []
                    ),
718
719
                )
            )
720
721
        return out

722
723
724
    def _create_stream_response_logprobs(
        self,
        token_ids: Sequence[int],
725
        logprobs: SampleLogprobs | None,
726
        tokenizer: AnyTokenizer,
727
        top_logprobs: int | None = None,
728
    ) -> list[response_text_delta_event.Logprob]:
729
730
731
732
733
734
        lgs = self._create_response_logprobs(
            token_ids=token_ids,
            logprobs=logprobs,
            tokenizer=tokenizer,
            top_logprobs=top_logprobs,
        )
735
736
737
738
739
740
        return [
            response_text_delta_event.Logprob(
                token=lg.token,
                logprob=lg.logprob,
                top_logprobs=[
                    response_text_delta_event.LogprobTopLogprob(
741
742
                        token=tl.token, logprob=tl.logprob
                    )
743
                    for tl in lg.top_logprobs
744
745
746
                ],
            )
            for lg in lgs
747
748
        ]

749
750
751
752
753
754
755
756
757
758
759
760
761
    def _make_response_output_items(
        self,
        request: ResponsesRequest,
        final_output: CompletionOutput,
        tokenizer: AnyTokenizer,
    ) -> list[ResponseOutputItem]:
        if self.reasoning_parser:
            try:
                reasoning_parser = self.reasoning_parser(tokenizer)
            except RuntimeError as e:
                logger.exception("Error in reasoning parser creation.")
                raise e

762
763
764
            reasoning_content, content = reasoning_parser.extract_reasoning_content(
                final_output.text, request=request
            )
765
766
767
768
        else:
            reasoning_content = None
            content = final_output.text

769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
        # Log complete response if output logging is enabled
        if self.enable_log_outputs and self.request_logger:
            output_text = ""
            if content:
                output_text = content
            elif reasoning_content:
                output_text = f"[reasoning: {reasoning_content}]"

            if output_text:
                self.request_logger.log_outputs(
                    request_id=request.request_id,
                    outputs=output_text,
                    output_token_ids=final_output.token_ids,
                    finish_reason=final_output.finish_reason,
                    is_streaming=False,
                    delta=False,
                )

787
788
789
790
791
792
793
        output = []
        if reasoning_content:
            reasoning_item = ResponseReasoningItem(
                id=f"rs_{random_uuid()}",
                summary=[],
                type="reasoning",
                content=[
794
795
796
                    ResponseReasoningTextContent(
                        text=reasoning_content, type="reasoning_text"
                    )
797
798
799
800
801
802
803
804
805
                ],
                status=None,  # NOTE: Only the last output item has status.
            )
            output.append(reasoning_item)
        if content:
            output_text = ResponseOutputText(
                text=content,
                annotations=[],  # TODO
                type="output_text",
806
807
808
809
810
811
812
813
814
815
                logprobs=(
                    self._create_response_logprobs(
                        token_ids=final_output.token_ids,
                        logprobs=final_output.logprobs,
                        tokenizer=tokenizer,
                        top_logprobs=request.top_logprobs,
                    )
                    if request.is_include_output_logprobs()
                    else None
                ),
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
            )
            message = ResponseOutputMessage(
                id=f"msg_{random_uuid()}",
                content=[output_text],
                role="assistant",
                status="completed",
                type="message",
            )
            output.append(message)
        return output

    def _make_response_output_items_with_harmony(
        self,
        context: HarmonyContext,
    ) -> list[ResponseOutputItem]:
831
        output_items: list[ResponseOutputItem] = []
832
833
834
835
836
837
838
839
840
        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

841
842
843
    def _construct_input_messages(
        self,
        request: ResponsesRequest,
844
        prev_response: ResponsesResponse | None = None,
845
846
847
    ) -> list[ChatCompletionMessageParam]:
        messages: list[ChatCompletionMessageParam] = []
        if request.instructions:
848
849
850
851
852
853
            messages.append(
                {
                    "role": "system",
                    "content": request.instructions,
                }
            )
854
855
856
857
858
859
860
861
862
863
864
865

        # Prepend the conversation history.
        if prev_response is not None:
            # Add the previous messages.
            prev_msg = self.msg_store[prev_response.id]
            messages.extend(prev_msg)

            # Add the previous output.
            for output_item in prev_response.output:
                # NOTE: We skip the reasoning output.
                if isinstance(output_item, ResponseOutputMessage):
                    for content in output_item.content:
866
867
868
869
870
871
                        messages.append(
                            {
                                "role": "assistant",
                                "content": content.text,
                            }
                        )
872
873

        # Append the new input.
874
        # Responses API supports simple text inputs without chat format.
875
876
877
878
879
880
        if isinstance(request.input, str):
            messages.append({"role": "user", "content": request.input})
        else:
            messages.extend(request.input)  # type: ignore
        return messages

881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
    def _construct_harmony_system_input_message(
        self, request: ResponsesRequest, with_custom_tools: bool, tool_types: list[str]
    ) -> OpenAIHarmonyMessage:
        reasoning_effort = request.reasoning.effort if request.reasoning else None
        enable_browser = (
            "web_search_preview" in tool_types
            and self.tool_server is not None
            and self.tool_server.has_tool("browser")
        )
        enable_code_interpreter = (
            "code_interpreter" in tool_types
            and self.tool_server is not None
            and self.tool_server.has_tool("python")
        )
        enable_container = (
            "container" in tool_types
            and self.tool_server is not None
            and self.tool_server.has_tool("container")
        )
        sys_msg = get_system_message(
            reasoning_effort=reasoning_effort,
            browser_description=(
                self.tool_server.get_tool_description("browser")
                if enable_browser and self.tool_server is not None
                else None
            ),
            python_description=(
                self.tool_server.get_tool_description("python")
                if enable_code_interpreter and self.tool_server is not None
                else None
            ),
            container_description=(
                self.tool_server.get_tool_description("container")
                if enable_container and self.tool_server is not None
                else None
            ),
            instructions=request.instructions,
            with_custom_tools=with_custom_tools,
        )
        return sys_msg

922
923
924
    def _construct_input_messages_with_harmony(
        self,
        request: ResponsesRequest,
925
        prev_response: ResponsesResponse | None,
926
927
928
929
930
    ) -> list[OpenAIHarmonyMessage]:
        messages: list[OpenAIHarmonyMessage] = []
        if prev_response is None:
            # New conversation.
            tool_types = [tool.type for tool in request.tools]
931
932
933
934
935
            # Allow the MCP Tool type to enable built in tools if the
            # server_label is allowlisted in
            # envs.GPT_OSS_SYSTEM_TOOL_MCP_LABELS
            if envs.GPT_OSS_SYSTEM_TOOL_MCP_LABELS:
                for tool in request.tools:
936
937
938
939
                    if (
                        tool.type == "mcp"
                        and tool.server_label in envs.GPT_OSS_SYSTEM_TOOL_MCP_LABELS
                    ):
940
                        tool_types.append(tool.server_label)
941
            with_custom_tools = has_custom_tools(tool_types)
942
943
944

            sys_msg = self._construct_harmony_system_input_message(
                request, with_custom_tools, tool_types
945
946
            )
            messages.append(sys_msg)
947
948
            if with_custom_tools:
                dev_msg = get_developer_message(
949
950
                    instructions=request.instructions, tools=request.tools
                )
951
                messages.append(dev_msg)
952
953
            messages += construct_harmony_previous_input_messages(request)

954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
        else:
            # Continue the previous conversation.
            # FIXME(woosuk): Currently, request params like reasoning and
            # instructions are ignored.
            prev_msgs = self.msg_store[prev_response.id]
            # Remove the previous chain-of-thoughts if there is a new "final"
            # message. Note that this also removes these messages from the
            # msg_store.
            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
973
974
                    recent_turn_msgs = prev_msgs[prev_final_msg_idx + 1 :]
                    del prev_msgs[prev_final_msg_idx + 1 :]
975
976
977
978
979
980
                    for msg in recent_turn_msgs:
                        assert isinstance(msg, OpenAIHarmonyMessage)
                        if msg.channel != "analysis":
                            prev_msgs.append(msg)
            messages.extend(prev_msgs)
        # Append the new input.
co63oc's avatar
co63oc committed
981
        # Responses API supports simple text inputs without chat format.
982
983
984
985
986
987
988
989
        if isinstance(request.input, str):
            messages.append(get_user_message(request.input))
        else:
            if prev_response is not None:
                prev_outputs = copy(prev_response.output)
            else:
                prev_outputs = []
            for response_msg in request.input:
990
                messages.append(parse_response_input(response_msg, prev_outputs))
991
                # User passes in a tool call request and its output. We need
992
993
994
                # 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.
995
996
997
998
999
                if (
                    isinstance(response_msg, dict)
                    and response_msg.get("type") == "function_call"
                ):
                    response_msg = ResponseFunctionToolCall.model_validate(response_msg)
1000
1001
1002
1003
                if isinstance(response_msg, ResponseFunctionToolCall):
                    prev_outputs.append(response_msg)
        return messages

1004
1005
1006
1007
1008
1009
    async def _run_background_request_stream(
        self,
        request: ResponsesRequest,
        *args,
        **kwargs,
    ):
1010
        event_deque: deque[StreamingResponsesResponse] = deque()
1011
1012
1013
1014
        new_event_signal = asyncio.Event()
        self.event_store[request.request_id] = (event_deque, new_event_signal)
        response = None
        try:
1015
            generator = self.responses_stream_generator(request, *args, **kwargs)
1016
1017
1018
1019
            async for event in generator:
                event_deque.append(event)
                new_event_signal.set()  # Signal new event available
        except Exception as e:
1020
            logger.exception("Background request failed for %s", request.request_id)
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
            response = self.create_error_response(str(e))
        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"

1034
1035
1036
1037
1038
1039
1040
    async def _run_background_request(
        self,
        request: ResponsesRequest,
        *args,
        **kwargs,
    ):
        try:
1041
            response = await self.responses_full_generator(request, *args, **kwargs)
1042
        except Exception as e:
1043
            logger.exception("Background request failed for %s", request.request_id)
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
            response = self.create_error_response(str(e))

        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"

1055
1056
1057
    async def responses_background_stream_generator(
        self,
        response_id: str,
1058
        starting_after: int | None = None,
1059
    ) -> AsyncGenerator[StreamingResponsesResponse, None]:
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
        if response_id not in self.event_store:
            raise ValueError(f"Unknown response_id: {response_id}")

        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
1074
                if getattr(event, "type", "unknown") == "response.completed":
1075
                    return
1076
1077
1078
1079
                current_index += 1

            await new_event_signal.wait()

1080
1081
1082
    async def retrieve_responses(
        self,
        response_id: str,
1083
1084
1085
1086
1087
1088
1089
        starting_after: int | None,
        stream: bool | None,
    ) -> (
        ErrorResponse
        | ResponsesResponse
        | AsyncGenerator[StreamingResponsesResponse, None]
    ):
1090
1091
1092
1093
1094
        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)
1095
1096
1097
1098
1099
1100

        if stream:
            return self.responses_background_stream_generator(
                response_id,
                starting_after,
            )
1101
1102
1103
1104
1105
        return response

    async def cancel_responses(
        self,
        response_id: str,
1106
    ) -> ErrorResponse | ResponsesResponse:
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
        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.",
                )

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

        # Abort the request.
1123
        if task := self.background_tasks.get(response_id):
1124
1125
1126
1127
            task.cancel()
            try:
                await task
            except asyncio.CancelledError:
1128
                logger.exception("Background task for %s was cancelled", response_id)
1129
1130
1131
1132
1133
1134
1135
1136
        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,
        )
1137
1138
1139
1140

    def _make_store_not_supported_error(self) -> ErrorResponse:
        return self.create_error_response(
            err_type="invalid_request_error",
1141
1142
1143
1144
1145
1146
            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."
            ),
1147
1148
            status_code=HTTPStatus.BAD_REQUEST,
        )
1149

1150
    async def _process_simple_streaming_events(
1151
1152
1153
        self,
        request: ResponsesRequest,
        sampling_params: SamplingParams,
1154
        result_generator: AsyncIterator[ConversationContext | None],
1155
1156
1157
1158
        context: ConversationContext,
        model_name: str,
        tokenizer: AnyTokenizer,
        request_metadata: RequestResponseMetadata,
1159
        created_time: int,
1160
        _increment_sequence_number_and_return: Callable[
1161
1162
            [StreamingResponsesResponse], StreamingResponsesResponse
        ],
1163
    ) -> AsyncGenerator[StreamingResponsesResponse, None]:
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
        current_content_index = 0
        current_output_index = 0
        current_item_id = ""
        reasoning_parser = None
        if self.reasoning_parser:
            reasoning_parser = self.reasoning_parser(tokenizer)
        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]
                if reasoning_parser:
1181
                    delta_message = (
1182
                        reasoning_parser.extract_reasoning_content_streaming(
1183
1184
1185
1186
1187
1188
1189
                            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,
                        )
1190
1191
                    )
                else:
1192
1193
1194
                    delta_message = DeltaMessage(
                        content=output.text,
                    )
1195
1196
1197
1198
1199
1200
1201
                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())
                    if delta_message.reasoning_content:
1202
                        yield _increment_sequence_number_and_return(
1203
1204
1205
1206
                            ResponseOutputItemAddedEvent(
                                type="response.output_item.added",
                                sequence_number=-1,
                                output_index=current_output_index,
1207
                                item=ResponseReasoningItem(
1208
1209
1210
1211
1212
                                    type="reasoning",
                                    id=current_item_id,
                                    summary=[],
                                    status="in_progress",
                                ),
1213
1214
                            )
                        )
1215
                    else:
1216
                        yield _increment_sequence_number_and_return(
1217
1218
1219
1220
                            ResponseOutputItemAddedEvent(
                                type="response.output_item.added",
                                sequence_number=-1,
                                output_index=current_output_index,
1221
                                item=ResponseOutputMessage(
1222
1223
1224
1225
1226
1227
                                    id=current_item_id,
                                    type="message",
                                    role="assistant",
                                    content=[],
                                    status="in_progress",
                                ),
1228
1229
                            )
                        )
1230
                    yield _increment_sequence_number_and_return(
1231
                        ResponseContentPartAddedEvent(
1232
1233
1234
1235
1236
                            type="response.content_part.added",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item_id=current_item_id,
                            content_index=current_content_index,
1237
                            part=ResponseOutputText(
1238
1239
1240
1241
1242
                                type="output_text",
                                text="",
                                annotations=[],
                                logprobs=[],
                            ),
1243
1244
                        )
                    )
1245
1246
1247
1248
1249
1250
                    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
1251
1252
1253
1254
1255
                if (
                    previous_delta_messages
                    and previous_delta_messages[-1].reasoning_content is not None
                    and delta_message.content is not None
                ):
1256
1257
                    # from reasoning to normal content, send done
                    # event for reasoning
1258
1259
1260
1261
1262
                    reason_content = "".join(
                        pm.reasoning_content
                        for pm in previous_delta_messages
                        if pm.reasoning_content is not None
                    )
1263
                    yield _increment_sequence_number_and_return(
1264
1265
1266
1267
1268
1269
1270
                        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,
1271
1272
                        )
                    )
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
                    current_content_index = 0
                    reasoning_item = ResponseReasoningItem(
                        type="reasoning",
                        content=[
                            ResponseReasoningTextContent(
                                text=reason_content,
                                type="reasoning_text",
                            ),
                        ],
                        status="completed",
                        id=current_item_id,
                        summary=[],
                    )
1286
                    yield _increment_sequence_number_and_return(
1287
1288
1289
1290
1291
                        ResponseOutputItemDoneEvent(
                            type="response.output_item.done",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item=reasoning_item,
1292
1293
                        )
                    )
1294
                    yield _increment_sequence_number_and_return(
1295
                        ResponseOutputItemAddedEvent(
1296
1297
1298
                            type="response.output_item.added",
                            sequence_number=-1,
                            output_index=current_output_index,
1299
                            item=ResponseOutputMessage(
1300
1301
1302
1303
1304
1305
                                id=current_item_id,
                                type="message",
                                role="assistant",
                                content=[],
                                status="in_progress",
                            ),
1306
1307
                        )
                    )
1308
1309
                    current_output_index += 1
                    current_item_id = str(uuid.uuid4())
1310
                    yield _increment_sequence_number_and_return(
1311
                        ResponseContentPartAddedEvent(
1312
1313
1314
1315
1316
                            type="response.content_part.added",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item_id=current_item_id,
                            content_index=current_content_index,
1317
                            part=ResponseOutputText(
1318
1319
1320
1321
1322
                                type="output_text",
                                text="",
                                annotations=[],
                                logprobs=[],
                            ),
1323
1324
                        )
                    )
1325
1326
1327
                    current_content_index += 1
                    # reset previous delta messages
                    previous_delta_messages = []
1328

1329
                if delta_message.reasoning_content is not None:
1330
                    yield _increment_sequence_number_and_return(
1331
1332
1333
1334
1335
1336
1337
                        ResponseReasoningTextDeltaEvent(
                            type="response.reasoning_text.delta",
                            sequence_number=-1,
                            content_index=current_content_index,
                            output_index=current_output_index,
                            item_id=current_item_id,
                            delta=delta_message.reasoning_content,
1338
1339
                        )
                    )
1340
                elif delta_message.content is not None:
1341
                    yield _increment_sequence_number_and_return(
1342
                        ResponseTextDeltaEvent(
1343
1344
1345
1346
1347
1348
                            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,
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
                            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 []
                            ),
1359
1360
                        )
                    )
1361
1362
1363
1364
1365
                current_content_index += 1

                previous_delta_messages.append(delta_message)
        if previous_delta_messages:
            if previous_delta_messages[-1].reasoning_content is not None:
1366
1367
1368
1369
1370
                reason_content = "".join(
                    pm.reasoning_content
                    for pm in previous_delta_messages
                    if pm.reasoning_content is not None
                )
1371
                yield _increment_sequence_number_and_return(
1372
1373
1374
1375
1376
1377
1378
                    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,
1379
1380
                    )
                )
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
                current_content_index += 1
                reasoning_item = ResponseReasoningItem(
                    type="reasoning",
                    content=[
                        ResponseReasoningTextContent(
                            text=reason_content,
                            type="reasoning_text",
                        ),
                    ],
                    status="completed",
                    id=current_item_id,
                    summary=[],
                )
1394
                yield _increment_sequence_number_and_return(
1395
1396
1397
1398
1399
                    ResponseOutputItemDoneEvent(
                        type="response.output_item.done",
                        sequence_number=-1,
                        output_index=current_output_index,
                        item=reasoning_item,
1400
1401
                    )
                )
1402
            elif previous_delta_messages[-1].content is not None:
1403
1404
1405
1406
1407
                final_content = "".join(
                    pm.content
                    for pm in previous_delta_messages
                    if pm.content is not None
                )
1408
                yield _increment_sequence_number_and_return(
1409
                    ResponseTextDoneEvent(
1410
1411
1412
1413
1414
1415
1416
                        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,
1417
1418
                    )
                )
1419
1420
1421
1422
1423
1424
                current_content_index += 1
                part = ResponseOutputText(
                    text=final_content,
                    type="output_text",
                    annotations=[],
                )
1425
                yield _increment_sequence_number_and_return(
1426
                    ResponseContentPartDoneEvent(
1427
1428
1429
1430
1431
1432
                        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,
1433
1434
                    )
                )
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
                current_content_index += 1
                item = ResponseOutputMessage(
                    type="message",
                    role="assistant",
                    content=[
                        part,
                    ],
                    status="completed",
                    id=current_item_id,
                    summary=[],
                )
1446
                yield _increment_sequence_number_and_return(
1447
1448
1449
1450
1451
                    ResponseOutputItemDoneEvent(
                        type="response.output_item.done",
                        sequence_number=-1,
                        output_index=current_output_index,
                        item=item,
1452
1453
                    )
                )
1454
1455
1456
1457
1458

    async def _process_harmony_streaming_events(
        self,
        request: ResponsesRequest,
        sampling_params: SamplingParams,
1459
        result_generator: AsyncIterator[ConversationContext | None],
1460
1461
1462
1463
1464
        context: ConversationContext,
        model_name: str,
        tokenizer: AnyTokenizer,
        request_metadata: RequestResponseMetadata,
        created_time: int,
1465
        _increment_sequence_number_and_return: Callable[
1466
1467
            [StreamingResponsesResponse], StreamingResponsesResponse
        ],
1468
    ) -> AsyncGenerator[StreamingResponsesResponse, None]:
1469
        current_content_index = -1
1470
        current_output_index = 0
1471
        current_item_id: str = ""
1472
        sent_output_item_added = False
1473
        is_first_function_call_delta = False
1474
1475
1476
1477
1478
1479
        async for ctx in result_generator:
            assert isinstance(ctx, StreamingHarmonyContext)

            if ctx.is_expecting_start():
                current_output_index += 1
                sent_output_item_added = False
1480
                is_first_function_call_delta = False
1481
1482
1483
                if len(ctx.parser.messages) > 0:
                    previous_item = ctx.parser.messages[-1]
                    if previous_item.recipient is not None:
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
                        # Deal with tool call
                        if previous_item.recipient.startswith("functions."):
                            function_name = previous_item.recipient[len("functions.") :]
                            yield _increment_sequence_number_and_return(
                                ResponseFunctionCallArgumentsDoneEvent(
                                    type="response.function_call_arguments.done",
                                    arguments=previous_item.content[0].text,
                                    name=function_name,
                                    item_id=current_item_id,
                                    output_index=current_output_index,
                                    sequence_number=-1,
                                )
                            )
                            function_call_item = ResponseFunctionToolCall(
                                type="function_call",
                                arguments=previous_item.content[0].text,
                                name=function_name,
                                item_id=current_item_id,
                                output_index=current_output_index,
                                sequence_number=-1,
                                call_id=f"fc_{random_uuid()}",
                                status="completed",
                            )
                            yield _increment_sequence_number_and_return(
                                ResponseOutputItemDoneEvent(
                                    type="response.output_item.done",
                                    sequence_number=-1,
                                    output_index=current_output_index,
                                    item=function_call_item,
                                )
                            )
1515
                    elif previous_item.channel == "analysis":
1516
1517
1518
1519
                        content = ResponseReasoningTextContent(
                            text=previous_item.content[0].text,
                            type="reasoning_text",
                        )
1520
1521
                        reasoning_item = ResponseReasoningItem(
                            type="reasoning",
1522
                            content=[content],
1523
                            status="completed",
1524
1525
                            id=current_item_id,
                            summary=[],
1526
                        )
1527
                        yield _increment_sequence_number_and_return(
1528
1529
1530
1531
1532
1533
1534
                            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=previous_item.content[0].text,
1535
1536
                            )
                        )
1537
1538
1539
1540
1541
1542
1543
1544
                        yield _increment_sequence_number_and_return(
                            ResponseReasoningPartDoneEvent(
                                type="response.reasoning_part.done",
                                sequence_number=-1,
                                item_id=current_item_id,
                                output_index=current_output_index,
                                content_index=current_content_index,
                                part=content,
1545
1546
                            )
                        )
1547
                        yield _increment_sequence_number_and_return(
1548
1549
1550
1551
1552
                            ResponseOutputItemDoneEvent(
                                type="response.output_item.done",
                                sequence_number=-1,
                                output_index=current_output_index,
                                item=reasoning_item,
1553
1554
                            )
                        )
1555
1556
1557
1558
1559
1560
                    elif previous_item.channel == "final":
                        text_content = ResponseOutputText(
                            type="output_text",
                            text=previous_item.content[0].text,
                            annotations=[],
                        )
1561
                        yield _increment_sequence_number_and_return(
1562
                            ResponseTextDoneEvent(
1563
1564
1565
1566
1567
1568
1569
                                type="response.output_text.done",
                                sequence_number=-1,
                                output_index=current_output_index,
                                content_index=current_content_index,
                                text=previous_item.content[0].text,
                                logprobs=[],
                                item_id=current_item_id,
1570
1571
                            )
                        )
1572
                        yield _increment_sequence_number_and_return(
1573
1574
1575
1576
1577
1578
1579
                            ResponseContentPartDoneEvent(
                                type="response.content_part.done",
                                sequence_number=-1,
                                item_id=current_item_id,
                                output_index=current_output_index,
                                content_index=current_content_index,
                                part=text_content,
1580
1581
                            )
                        )
1582
                        yield _increment_sequence_number_and_return(
1583
                            ResponseOutputItemDoneEvent(
1584
1585
1586
1587
1588
1589
1590
1591
1592
1593
                                type="response.output_item.done",
                                sequence_number=-1,
                                output_index=current_output_index,
                                item=ResponseOutputMessage(
                                    id=current_item_id,
                                    type="message",
                                    role="assistant",
                                    content=[text_content],
                                    status="completed",
                                ),
1594
1595
                            )
                        )
1596

1597
            # stream the output of a harmony message
1598
            if ctx.parser.last_content_delta:
1599
1600
1601
1602
                if (
                    ctx.parser.current_channel == "final"
                    and ctx.parser.current_recipient is None
                ):
1603
1604
                    if not sent_output_item_added:
                        sent_output_item_added = True
1605
                        current_item_id = f"msg_{random_uuid()}"
1606
                        yield _increment_sequence_number_and_return(
1607
1608
1609
1610
                            ResponseOutputItemAddedEvent(
                                type="response.output_item.added",
                                sequence_number=-1,
                                output_index=current_output_index,
1611
                                item=ResponseOutputMessage(
1612
1613
1614
1615
1616
1617
                                    id=current_item_id,
                                    type="message",
                                    role="assistant",
                                    content=[],
                                    status="in_progress",
                                ),
1618
1619
                            )
                        )
1620
                        current_content_index += 1
1621
                        yield _increment_sequence_number_and_return(
1622
1623
1624
1625
1626
1627
                            ResponseContentPartAddedEvent(
                                type="response.content_part.added",
                                sequence_number=-1,
                                output_index=current_output_index,
                                item_id=current_item_id,
                                content_index=current_content_index,
1628
                                part=ResponseOutputText(
1629
1630
1631
1632
1633
                                    type="output_text",
                                    text="",
                                    annotations=[],
                                    logprobs=[],
                                ),
1634
1635
                            )
                        )
1636
                    yield _increment_sequence_number_and_return(
1637
                        ResponseTextDeltaEvent(
1638
1639
1640
1641
1642
1643
1644
1645
                            type="response.output_text.delta",
                            sequence_number=-1,
                            content_index=current_content_index,
                            output_index=current_output_index,
                            item_id=current_item_id,
                            delta=ctx.parser.last_content_delta,
                            # TODO, use logprobs from ctx.last_request_output
                            logprobs=[],
1646
1647
1648
1649
1650
1651
                        )
                    )
                elif (
                    ctx.parser.current_channel == "analysis"
                    and ctx.parser.current_recipient is None
                ):
1652
1653
                    if not sent_output_item_added:
                        sent_output_item_added = True
1654
                        current_item_id = f"msg_{random_uuid()}"
1655
                        yield _increment_sequence_number_and_return(
1656
1657
1658
1659
                            ResponseOutputItemAddedEvent(
                                type="response.output_item.added",
                                sequence_number=-1,
                                output_index=current_output_index,
1660
                                item=ResponseReasoningItem(
1661
1662
1663
1664
1665
                                    type="reasoning",
                                    id=current_item_id,
                                    summary=[],
                                    status="in_progress",
                                ),
1666
1667
                            )
                        )
1668
                        current_content_index += 1
1669
                        yield _increment_sequence_number_and_return(
1670
1671
                            ResponseReasoningPartAddedEvent(
                                type="response.reasoning_part.added",
1672
1673
1674
1675
                                sequence_number=-1,
                                output_index=current_output_index,
                                item_id=current_item_id,
                                content_index=current_content_index,
1676
                                part=ResponseReasoningTextContent(
1677
                                    text="",
1678
                                    type="reasoning_text",
1679
                                ),
1680
1681
                            )
                        )
1682
                    yield _increment_sequence_number_and_return(
1683
1684
1685
1686
1687
1688
1689
                        ResponseReasoningTextDeltaEvent(
                            type="response.reasoning_text.delta",
                            item_id=current_item_id,
                            output_index=current_output_index,
                            content_index=current_content_index,
                            delta=ctx.parser.last_content_delta,
                            sequence_number=-1,
1690
1691
                        )
                    )
1692
1693
1694
                # built-in tools will be triggered on the analysis channel
                # However, occasionally built-in tools will
                # still be output to commentary.
1695
1696
1697
1698
                elif (
                    ctx.parser.current_channel == "commentary"
                    or ctx.parser.current_channel == "analysis"
                ) and ctx.parser.current_recipient == "python":
1699
1700
                    if not sent_output_item_added:
                        sent_output_item_added = True
1701
                        current_item_id = f"tool_{random_uuid()}"
1702
                        yield _increment_sequence_number_and_return(
1703
1704
1705
1706
                            ResponseOutputItemAddedEvent(
                                type="response.output_item.added",
                                sequence_number=-1,
                                output_index=current_output_index,
1707
                                item=ResponseCodeInterpreterToolCallParam(
1708
1709
1710
1711
1712
1713
1714
                                    type="code_interpreter_call",
                                    id=current_item_id,
                                    code=None,
                                    container_id="auto",
                                    outputs=None,
                                    status="in_progress",
                                ),
1715
1716
                            )
                        )
1717
                        yield _increment_sequence_number_and_return(
1718
                            ResponseCodeInterpreterCallInProgressEvent(
1719
                                type="response.code_interpreter_call.in_progress",
1720
1721
1722
                                sequence_number=-1,
                                output_index=current_output_index,
                                item_id=current_item_id,
1723
1724
                            )
                        )
1725
                    yield _increment_sequence_number_and_return(
1726
1727
1728
1729
1730
1731
                        ResponseCodeInterpreterCallCodeDeltaEvent(
                            type="response.code_interpreter_call_code.delta",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item_id=current_item_id,
                            delta=ctx.parser.last_content_delta,
1732
1733
                        )
                    )
1734
1735

            # stream tool call outputs
1736
1737
            if ctx.is_assistant_action_turn() and len(ctx.parser.messages) > 0:
                previous_item = ctx.parser.messages[-1]
1738
1739
1740
1741
1742
1743
1744
                if (
                    self.tool_server is not None
                    and self.tool_server.has_tool("browser")
                    and previous_item.recipient is not None
                    and previous_item.recipient.startswith("browser.")
                ):
                    function_name = previous_item.recipient[len("browser.") :]
1745
1746
1747
                    action = None
                    parsed_args = json.loads(previous_item.content[0].text)
                    if function_name == "search":
1748
                        action = response_function_web_search.ActionSearch(
1749
1750
                            type="search",
                            query=parsed_args["query"],
1751
                        )
1752
                    elif function_name == "open":
1753
1754
1755
1756
1757
                        action = response_function_web_search.ActionOpenPage(
                            type="open_page",
                            # TODO: translate to url
                            url=f"cursor:{parsed_args.get('cursor', '')}",
                        )
1758
                    elif function_name == "find":
1759
1760
1761
1762
1763
1764
                        action = response_function_web_search.ActionFind(
                            type="find",
                            pattern=parsed_args["pattern"],
                            # TODO: translate to url
                            url=f"cursor:{parsed_args.get('cursor', '')}",
                        )
1765
                    else:
1766
                        raise ValueError(f"Unknown function name: {function_name}")
1767

1768
                    current_item_id = f"tool_{random_uuid()}"
1769
                    yield _increment_sequence_number_and_return(
1770
                        ResponseOutputItemAddedEvent(
1771
1772
1773
                            type="response.output_item.added",
                            sequence_number=-1,
                            output_index=current_output_index,
1774
                            item=response_function_web_search.ResponseFunctionWebSearch(
1775
1776
1777
1778
1779
1780
                                # TODO: generate a unique id for web search call
                                type="web_search_call",
                                id=current_item_id,
                                action=action,
                                status="in_progress",
                            ),
1781
1782
                        )
                    )
1783
                    yield _increment_sequence_number_and_return(
1784
1785
1786
1787
1788
                        ResponseWebSearchCallInProgressEvent(
                            type="response.web_search_call.in_progress",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item_id=current_item_id,
1789
1790
                        )
                    )
1791
                    yield _increment_sequence_number_and_return(
1792
1793
1794
1795
1796
                        ResponseWebSearchCallSearchingEvent(
                            type="response.web_search_call.searching",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item_id=current_item_id,
1797
1798
                        )
                    )
1799
1800

                    # enqueue
1801
                    yield _increment_sequence_number_and_return(
1802
1803
1804
1805
1806
                        ResponseWebSearchCallCompletedEvent(
                            type="response.web_search_call.completed",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item_id=current_item_id,
1807
1808
                        )
                    )
1809
                    yield _increment_sequence_number_and_return(
1810
                        ResponseOutputItemDoneEvent(
1811
1812
1813
                            type="response.output_item.done",
                            sequence_number=-1,
                            output_index=current_output_index,
1814
                            item=ResponseFunctionWebSearch(
1815
1816
1817
1818
1819
                                type="web_search_call",
                                id=current_item_id,
                                action=action,
                                status="completed",
                            ),
1820
1821
                        )
                    )
1822

1823
1824
1825
1826
1827
1828
                if (
                    self.tool_server is not None
                    and self.tool_server.has_tool("python")
                    and previous_item.recipient is not None
                    and previous_item.recipient.startswith("python")
                ):
1829
                    yield _increment_sequence_number_and_return(
1830
1831
1832
1833
1834
                        ResponseCodeInterpreterCallCodeDoneEvent(
                            type="response.code_interpreter_call_code.done",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item_id=current_item_id,
1835
                            code=previous_item.content[0].text,
1836
1837
                        )
                    )
1838
                    yield _increment_sequence_number_and_return(
1839
1840
1841
1842
1843
                        ResponseCodeInterpreterCallInterpretingEvent(
                            type="response.code_interpreter_call.interpreting",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item_id=current_item_id,
1844
1845
                        )
                    )
1846
                    yield _increment_sequence_number_and_return(
1847
1848
1849
1850
1851
                        ResponseCodeInterpreterCallCompletedEvent(
                            type="response.code_interpreter_call.completed",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item_id=current_item_id,
1852
1853
                        )
                    )
1854
                    yield _increment_sequence_number_and_return(
1855
                        ResponseOutputItemDoneEvent(
1856
1857
1858
                            type="response.output_item.done",
                            sequence_number=-1,
                            output_index=current_output_index,
1859
                            item=ResponseCodeInterpreterToolCallParam(
1860
1861
1862
1863
1864
1865
1866
1867
                                type="code_interpreter_call",
                                id=current_item_id,
                                code=previous_item.content[0].text,
                                container_id="auto",
                                # TODO: add outputs here
                                outputs=[],
                                status="completed",
                            ),
1868
1869
                        )
                    )
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
1892
1893
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
1906
            # developer tools will be triggered on the commentary channel
            # and recipient starts with "functions.TOOL_NAME"
            if (
                ctx.parser.current_channel == "commentary"
                and ctx.parser.current_recipient
                and ctx.parser.current_recipient.startswith("functions.")
            ):
                if is_first_function_call_delta is False:
                    is_first_function_call_delta = True
                    fc_name = ctx.parser.current_recipient[len("functions.") :]
                    tool_call_item = ResponseFunctionToolCall(
                        name=fc_name,
                        type="function_call",
                        id=current_item_id,
                        call_id=f"call_{random_uuid()}",
                        arguments="",
                        status="in_progress",
                    )
                    current_item_id = f"fc_{random_uuid()}"
                    yield _increment_sequence_number_and_return(
                        ResponseOutputItemAddedEvent(
                            type="response.output_item.added",
                            sequence_number=-1,
                            output_index=current_output_index,
                            item=tool_call_item,
                        )
                    )
                else:
                    yield _increment_sequence_number_and_return(
                        ResponseFunctionCallArgumentsDeltaEvent(
                            item_id=current_item_id,
                            delta=ctx.parser.last_content_delta,
                            output_index=current_output_index,
                            sequence_number=-1,
                            type="response.function_call_arguments.delta",
                        )
                    )
1907

1908
1909
1910
1911
    async def responses_stream_generator(
        self,
        request: ResponsesRequest,
        sampling_params: SamplingParams,
1912
        result_generator: AsyncIterator[ConversationContext | None],
1913
1914
1915
1916
        context: ConversationContext,
        model_name: str,
        tokenizer: AnyTokenizer,
        request_metadata: RequestResponseMetadata,
1917
        created_time: int | None = None,
1918
    ) -> AsyncGenerator[StreamingResponsesResponse, None]:
1919
1920
1921
1922
1923
        # TODO:
        # 1. Handle disconnect

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

1924
1925
        sequence_number = 0

1926
        def _increment_sequence_number_and_return(
1927
            event: StreamingResponsesResponse,
1928
        ) -> StreamingResponsesResponse:
1929
1930
            nonlocal sequence_number
            # Set sequence_number if the event has this attribute
1931
            if hasattr(event, "sequence_number"):
1932
1933
                event.sequence_number = sequence_number
            sequence_number += 1
1934
            return event
1935

1936
        async with AsyncExitStack() as exit_stack:
1937
1938
            processer = None
            if self.use_harmony:
1939
1940
                # TODO: in streaming, we noticed this bug:
                # https://github.com/vllm-project/vllm/issues/25697
1941
                await self._initialize_tool_sessions(request, context, exit_stack)
1942
1943
1944
                processer = self._process_harmony_streaming_events
            else:
                processer = self._process_simple_streaming_events
1945
            # TODO Hanchen make sampling params to include the structural tag
1946
1947
1948
1949
1950
1951
1952
1953
1954
1955

            initial_response = ResponsesResponse.from_request(
                request,
                sampling_params,
                model_name=model_name,
                created_time=created_time,
                output=[],
                status="in_progress",
                usage=None,
            ).model_dump()
1956
            yield _increment_sequence_number_and_return(
1957
1958
1959
1960
                ResponseCreatedEvent(
                    type="response.created",
                    sequence_number=-1,
                    response=initial_response,
1961
1962
                )
            )
1963
            yield _increment_sequence_number_and_return(
1964
1965
1966
1967
                ResponseInProgressEvent(
                    type="response.in_progress",
                    sequence_number=-1,
                    response=initial_response,
1968
1969
                )
            )
1970

1971
            async for event_data in processer(
1972
1973
1974
1975
1976
1977
1978
1979
1980
1981
                request,
                sampling_params,
                result_generator,
                context,
                model_name,
                tokenizer,
                request_metadata,
                created_time,
                _increment_sequence_number_and_return,
            ):
1982
                yield event_data
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999

            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,
            )
2000
            yield _increment_sequence_number_and_return(
2001
                ResponseCompletedEvent(
2002
2003
                    type="response.completed",
                    sequence_number=-1,
2004
                    response=final_response,
2005
2006
                )
            )