diff --git a/tests/entrypoints/openai/responses/conftest.py b/tests/entrypoints/openai/responses/conftest.py index 5bba59781f..7ee4629df3 100644 --- a/tests/entrypoints/openai/responses/conftest.py +++ b/tests/entrypoints/openai/responses/conftest.py @@ -126,6 +126,11 @@ def _validate_event_pairing(events: list, pairs_of_event_types: dict[str, str]) etype = event.type if etype in end_events: expected_start = pairs_of_event_types[etype] + # Deltas are optional, e.g. a function call with no arguments. + if expected_start.endswith(".delta") and ( + not stack or stack[-1] != expected_start + ): + continue assert stack and stack[-1] == expected_start, ( f"Stack mismatch for {etype}: " f"expected {expected_start}, " diff --git a/tests/entrypoints/openai/responses/test_serving_responses.py b/tests/entrypoints/openai/responses/test_serving_responses.py index c06e2b5915..33e5eaa323 100644 --- a/tests/entrypoints/openai/responses/test_serving_responses.py +++ b/tests/entrypoints/openai/responses/test_serving_responses.py @@ -64,6 +64,7 @@ from vllm.entrypoints.openai.responses.streaming_events import ( ) from vllm.entrypoints.serve.engine.protocol import ErrorResponse from vllm.inputs import tokens_input +from vllm.logprobs import Logprob as SampleLogprob from vllm.outputs import CompletionOutput, RequestOutput from vllm.parser.harmony import Segment from vllm.renderers import TokenizeParams @@ -1982,6 +1983,63 @@ class TestAutoToolStreaming: assert function_done[0].item.arguments == tool_args +@pytest.mark.skip_global_cleanup +@pytest.mark.asyncio +async def test_stream_completed_response_reuses_streamed_items(monkeypatch): + """response.completed must carry the streamed items (same ids, + call_id, logprobs) rather than a reparse of the full output.""" + monkeypatch.setattr(envs, "VLLM_USE_EXPERIMENTAL_PARSER_CONTEXT", False) + serving = _make_serving_instance_with_reasoning() + call = DeltaToolCall( + id="chatcmpl-tool-parser-id", + index=0, + function=DeltaFunctionCall(name="get_weather", arguments="{}"), + ) + response_parser = _mock_parser_with_reasoning( + serving, + [ + DeltaMessage(reasoning="think"), + DeltaMessage(content="Hi"), + DeltaMessage(tool_calls=[call]), + ], + ) + response_parser.count_reasoning_tokens = MagicMock(return_value=1) + context = SimpleContext(response_parser=response_parser) + + async def result_generator(): + for token_id, text in [(10, "think"), (20, "Hi"), (30, "call")]: + output = _make_request_output(text, [token_id]) + output.outputs[0].logprobs = [ + {token_id: SampleLogprob(logprob=-0.5, decoded_token=text)} + ] + context.append_output(output) + yield context + + events = [ + event + async for event in serving.responses_stream_generator( + request=ResponsesRequest( + input="hi", + include=["message.output_text.logprobs"], + stream=True, + store=False, + ), + sampling_params=SamplingParams(max_tokens=16), + result_generator=result_generator(), + context=context, + model_name="test-model", + tokenizer=MagicMock(), + request_metadata=RequestResponseMetadata(request_id="req"), + ) + ] + + streamed = [e.item for e in events if e.type == "response.output_item.done"] + assert events[-1].response.output == streamed + _, message, function_call = streamed + assert [lp.token for lp in message.content[0].logprobs] == ["Hi"] + assert function_call.call_id == "chatcmpl-tool-parser-id" + + @pytest.mark.asyncio @pytest.mark.parametrize(("parallel_tool_calls", "num_calls"), [(False, 1), (None, 3)]) async def test_streaming_parallel_tool_calls( diff --git a/vllm/entrypoints/openai/responses/context.py b/vllm/entrypoints/openai/responses/context.py index 1a218f1692..e71af75a0e 100644 --- a/vllm/entrypoints/openai/responses/context.py +++ b/vllm/entrypoints/openai/responses/context.py @@ -194,6 +194,9 @@ class SimpleContext(ConversationContext): self._accumulated_text: str = "" self._accumulated_token_ids: list[int] = [] self._accumulated_logprobs: list = [] + # Items already sent as output_item.done events; the final + # streaming response reuses them instead of reparsing. + self.streamed_output_items: list[ResponseOutputItem] | None = None self.num_prompt_tokens = 0 self.num_output_tokens = 0 diff --git a/vllm/entrypoints/openai/responses/serving.py b/vllm/entrypoints/openai/responses/serving.py index 4facd2fb3b..34f6fb44a2 100644 --- a/vllm/entrypoints/openai/responses/serving.py +++ b/vllm/entrypoints/openai/responses/serving.py @@ -15,7 +15,6 @@ from openai.types.responses import ( ResponseOutputMessage, ResponseOutputText, ResponseStatus, - response_text_delta_event, ) from openai.types.responses.response_output_text import Logprob, LogprobTopLogprob from pydantic import TypeAdapter @@ -813,14 +812,19 @@ class OpenAIServingResponses(GenerateBaseServing): if final_output.finish_reason == "length": status = "incomplete" - # TODO: Build final response items from the accumulated streaming - # parser results instead of reparsing the complete output. - output = self._make_response_output_items( - request, - final_output, - tokenizer, - parser=context.response_parser, + streamed_items = context.streamed_output_items + self._log_final_output( + request, final_output, is_streaming=streamed_items is not None ) + if streamed_items is not None: + output = streamed_items + else: + output = self._make_response_output_items( + request, + final_output, + tokenizer, + parser=context.response_parser, + ) if request.enable_response_messages: input_messages = context.input_messages @@ -962,32 +966,21 @@ class OpenAIServingResponses(GenerateBaseServing): ) return out - def _create_stream_response_logprobs( + def _log_final_output( self, - token_ids: Sequence[int], - logprobs: SampleLogprobs | None, - tokenizer: TokenizerLike, - top_logprobs: int | None = None, - ) -> list[response_text_delta_event.Logprob]: - lgs = self._create_response_logprobs( - token_ids=token_ids, - logprobs=logprobs, - tokenizer=tokenizer, - top_logprobs=top_logprobs, - ) - return [ - response_text_delta_event.Logprob( - token=lg.token, - logprob=lg.logprob, - top_logprobs=[ - response_text_delta_event.LogprobTopLogprob( - token=tl.token, logprob=tl.logprob - ) - for tl in lg.top_logprobs - ], + request: ResponsesRequest, + final_output: CompletionOutput, + is_streaming: bool, + ) -> None: + if self.enable_log_outputs and self.request_logger: + self.request_logger.log_outputs( + request_id=request.request_id, + outputs=final_output.text, + output_token_ids=final_output.token_ids, + finish_reason=final_output.finish_reason, + is_streaming=is_streaming, + delta=False, ) - for lg in lgs - ] def _make_response_output_items( self, @@ -996,17 +989,6 @@ class OpenAIServingResponses(GenerateBaseServing): tokenizer: TokenizerLike, parser: Parser | None = None, ) -> list[ResponseOutputItem]: - # Log complete response if output logging is enabled - if self.enable_log_outputs and self.request_logger: - self.request_logger.log_outputs( - request_id=request.request_id, - outputs=final_output.text, - output_token_ids=final_output.token_ids, - finish_reason=final_output.finish_reason, - is_streaming=False, - delta=False, - ) - # Compute logprobs if requested logprobs = None if request.is_include_output_logprobs() and final_output.logprobs: @@ -1197,14 +1179,12 @@ class OpenAIServingResponses(GenerateBaseServing): hide_stream_metadata = not request.include_reasoning and self.parser is not None - def _get_logprobs( - output: CompletionOutput, - ) -> list[response_text_delta_event.Logprob]: + def _get_logprobs(output: CompletionOutput) -> list[Logprob]: if not request.is_include_output_logprobs(): return [] if hide_stream_metadata: return [] - return self._create_stream_response_logprobs( + return self._create_response_logprobs( token_ids=output.token_ids, logprobs=output.logprobs, tokenizer=tokenizer, @@ -1253,6 +1233,9 @@ class OpenAIServingResponses(GenerateBaseServing): for event in processor.close_current(): yield _increment_sequence_number_and_return(event) + assert isinstance(context, SimpleContext) + context.streamed_output_items = processor.output_items + async def _process_harmony_streaming_events( self, request: ResponsesRequest, diff --git a/vllm/entrypoints/openai/responses/streaming_events.py b/vllm/entrypoints/openai/responses/streaming_events.py index 4651bb7205..ae9fc2b25e 100644 --- a/vllm/entrypoints/openai/responses/streaming_events.py +++ b/vllm/entrypoints/openai/responses/streaming_events.py @@ -38,6 +38,7 @@ from openai.types.responses import ( ResponseMcpCallArgumentsDoneEvent, ResponseMcpCallCompletedEvent, ResponseMcpCallInProgressEvent, + ResponseOutputItem, ResponseOutputItemAddedEvent, ResponseOutputItemDoneEvent, ResponseOutputMessage, @@ -51,6 +52,7 @@ from openai.types.responses import ( ResponseWebSearchCallInProgressEvent, ResponseWebSearchCallSearchingEvent, response_function_web_search, + response_output_text, response_text_delta_event, ) from openai.types.responses.response_output_item import McpCall @@ -60,6 +62,7 @@ from openai.types.responses.response_reasoning_item import ( from openai.types.responses.tool import Tool from openai_harmony import Message as HarmonyMessage +from vllm.entrypoints.chat_utils import make_tool_call_id from vllm.entrypoints.generate.base.protocol import DeltaMessage, DeltaToolCall from vllm.entrypoints.mcp.tool_server import ToolServer from vllm.entrypoints.openai.parser.harmony_utils import ( @@ -816,11 +819,13 @@ class SimpleStreamingState: current_item_id: str = "" content_index: int = 0 accumulated_text: str = "" + accumulated_logprobs: list[response_output_text.Logprob] = field( + default_factory=list + ) tool_call_id: str = "" tool_call_name: str = "" tool_call_namespace: str | None = None tool_call_index: int | None = None - has_emitted_tool_call_delta: bool = False current_state: _StateType = field(default_factory=lambda: _StateType.NONE) @@ -828,9 +833,10 @@ def emit_simple_content_open( state: SimpleStreamingState, ) -> list[StreamingResponsesResponse]: state.current_state = _StateType.CONTENT - state.current_item_id = random_uuid() + state.current_item_id = f"msg_{random_uuid()}" state.content_index = 0 state.accumulated_text = "" + state.accumulated_logprobs = [] return [ ResponseOutputItemAddedEvent( type="response.output_item.added", @@ -860,12 +866,32 @@ def emit_simple_content_open( ] +def _to_stream_logprobs( + logprobs: list[response_output_text.Logprob], +) -> list[response_text_delta_event.Logprob]: + return [ + response_text_delta_event.Logprob( + token=lp.token, + logprob=lp.logprob, + top_logprobs=[ + response_text_delta_event.LogprobTopLogprob( + token=tl.token, logprob=tl.logprob + ) + for tl in lp.top_logprobs + ], + ) + for lp in logprobs + ] + + def emit_simple_content_delta( state: SimpleStreamingState, delta: str, - logprobs: list[response_text_delta_event.Logprob] | None = None, + logprobs: list[response_output_text.Logprob] | None = None, ) -> list[StreamingResponsesResponse]: state.accumulated_text += delta + if logprobs: + state.accumulated_logprobs.extend(logprobs) return [ ResponseTextDeltaEvent( type="response.output_text.delta", @@ -874,7 +900,7 @@ def emit_simple_content_delta( output_index=state.output_index, item_id=state.current_item_id, delta=delta, - logprobs=logprobs or [], + logprobs=_to_stream_logprobs(logprobs or []), ) ] @@ -886,6 +912,7 @@ def emit_simple_content_done( type="output_text", text=state.accumulated_text, annotations=[], + logprobs=state.accumulated_logprobs or None, ) events: list[StreamingResponsesResponse] = [ ResponseTextDoneEvent( @@ -915,7 +942,6 @@ def emit_simple_content_done( role="assistant", content=[part] if state.accumulated_text else [], status="completed", - summary=[], ), ), ] @@ -928,7 +954,7 @@ def emit_simple_reasoning_open( state: SimpleStreamingState, ) -> list[StreamingResponsesResponse]: state.current_state = _StateType.REASONING - state.current_item_id = random_uuid() + state.current_item_id = f"rs_{random_uuid()}" state.content_index = 0 state.accumulated_text = "" return [ @@ -1021,15 +1047,15 @@ def emit_simple_tool_call_open( name: str, index: int | None, namespace: str | None = None, + call_id: str | None = None, ) -> list[StreamingResponsesResponse]: state.current_state = _StateType.TOOL_CALL - state.current_item_id = random_uuid() - state.tool_call_id = f"call_{random_uuid()}" + state.current_item_id = f"fc_{random_uuid()}" + state.tool_call_id = call_id or make_tool_call_id() state.tool_call_name = name state.tool_call_namespace = namespace state.tool_call_index = index state.accumulated_text = "" - state.has_emitted_tool_call_delta = False return [ ResponseOutputItemAddedEvent( type="response.output_item.added", @@ -1053,7 +1079,6 @@ def emit_simple_tool_call_delta( delta: str, ) -> list[StreamingResponsesResponse]: state.accumulated_text += delta - state.has_emitted_tool_call_delta = True return [ ResponseFunctionCallArgumentsDeltaEvent( type="response.function_call_arguments.delta", @@ -1068,19 +1093,15 @@ def emit_simple_tool_call_delta( def emit_simple_tool_call_done( state: SimpleStreamingState, ) -> list[StreamingResponsesResponse]: - events: list[StreamingResponsesResponse] = [] - if state.has_emitted_tool_call_delta: - events.append( - ResponseFunctionCallArgumentsDoneEvent( - type="response.function_call_arguments.done", - sequence_number=-1, - output_index=state.output_index, - item_id=state.current_item_id, - arguments=state.accumulated_text, - name=state.tool_call_name, - ) - ) - events.append( + events: list[StreamingResponsesResponse] = [ + ResponseFunctionCallArgumentsDoneEvent( + type="response.function_call_arguments.done", + sequence_number=-1, + output_index=state.output_index, + item_id=state.current_item_id, + arguments=state.accumulated_text, + name=state.tool_call_name, + ), ResponseOutputItemDoneEvent( type="response.output_item.done", sequence_number=-1, @@ -1095,7 +1116,7 @@ def emit_simple_tool_call_done( call_id=state.tool_call_id, ), ), - ) + ] state.output_index += 1 state.tool_call_namespace = None state.current_state = _StateType.NONE @@ -1181,6 +1202,7 @@ class SimpleStreamingEventProcessor: ) -> None: self.state = state or SimpleStreamingState() self.tool_call_name_map = build_responses_tool_call_name_map(tools) + self.output_items: list[ResponseOutputItem] = [] def resolve_target_state( self, delta_message: DeltaMessage @@ -1226,7 +1248,13 @@ class SimpleStreamingEventProcessor: handlers = self._STATE_HANDLERS.get(self.state.current_state) if handlers is None: return [] - return handlers.done_fn(self.state) + events = handlers.done_fn(self.state) + self.output_items.extend( + event.item + for event in events + if isinstance(event, ResponseOutputItemDoneEvent) + ) + return events def open( self, target_state: _StateType, tool_call: Any = None @@ -1244,6 +1272,7 @@ class SimpleStreamingEventProcessor: call_name.name, tool_call.index, call_name.namespace, + tool_call.id, ) return handlers.open_fn(self.state) @@ -1251,9 +1280,7 @@ class SimpleStreamingEventProcessor: self, delta_message: DeltaMessage, output: CompletionOutput, - get_logprobs: Callable[ - [CompletionOutput], list[response_text_delta_event.Logprob] - ] + get_logprobs: Callable[[CompletionOutput], list[response_output_text.Logprob]] | None = None, ) -> list[StreamingResponsesResponse]: """Emit incremental events for the current state from the delta."""