diff --git a/tests/entrypoints/openai/responses/test_serving_responses.py b/tests/entrypoints/openai/responses/test_serving_responses.py index 591ba5c42a..acab4cb6e2 100644 --- a/tests/entrypoints/openai/responses/test_serving_responses.py +++ b/tests/entrypoints/openai/responses/test_serving_responses.py @@ -67,7 +67,7 @@ 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.parser.harmony import ChunkResult, HarmonyParser, Segment from vllm.renderers import TokenizeParams from vllm.renderers.online_renderer import ( OnlineRenderer, @@ -2129,6 +2129,87 @@ async def test_stream_completed_response_reuses_streamed_items(monkeypatch): assert function_call.call_id == "chatcmpl-tool-parser-id" +def _harmony_msg(channel: str, text: str, recipient: str | None = None): + msg = OpenAIHarmonyMessage.from_role_and_content(Role.ASSISTANT, text) + msg = msg.with_channel(channel) + return msg.with_recipient(recipient) if recipient else msg + + +def _harmony_segments(msg, *, streamed: bool = True) -> list[Segment]: + """Segments the parser yields for one message: a content delta (absent + for a zero-delta message), then the completed message.""" + delta = [Segment(msg.channel, msg.recipient, msg.content[0].text)] + completed = Segment(msg.channel, msg.recipient, "", completed_message=msg) + return (delta if streamed else []) + [completed] + + +async def _harmony_stream_events(segment_chunks: list[list[Segment]]): + serving = _make_serving_instance() + serving.use_harmony = True + parser = MagicMock(spec=HarmonyParser) + parser.process_chunk.side_effect = [ + ChunkResult(segments=segments, reasoning_token_count=0) + for segments in segment_chunks + ] + context = HarmonyContext([], [], frozenset({"get_weather"}), parser) + + async def result_generator(): + for token_id in range(len(segment_chunks)): + context.append_output(_make_request_output("", [token_id])) + yield context + + return [ + event + async for event in serving.responses_stream_generator( + request=ResponsesRequest(input="hi", 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"), + ) + ] + + +@pytest.mark.asyncio +async def test_harmony_stream_completed_response_reuses_streamed_ids(): + """response.completed must carry the id and call_id each item was streamed + with, not ones minted when the harmony messages are converted again.""" + messages = [ + _harmony_msg("analysis", "think"), + _harmony_msg("commentary", '{"city": "Paris"}', "functions.get_weather"), + _harmony_msg("final", "Sunny"), + ] + + events = await _harmony_stream_events([_harmony_segments(msg) for msg in messages]) + + streamed = [e.item for e in events if e.type == "response.output_item.done"] + output = events[-1].response.output + assert [item.type for item in output] == ["reasoning", "function_call", "message"] + assert [item.id for item in output] == [item.id for item in streamed] + assert output[1].call_id == streamed[1].call_id + assert output[2].content[0].text == "Sunny" + + +@pytest.mark.asyncio +async def test_harmony_stream_unstreamed_item_keeps_own_id(): + """An item with no done event (zero-delta) must not take the streamed id + of a later item of the same type.""" + unstreamed = _harmony_msg("analysis", "skipped") + reasoning = _harmony_msg("analysis", "think") + + events = await _harmony_stream_events( + [_harmony_segments(unstreamed, streamed=False), _harmony_segments(reasoning)] + ) + + (streamed,) = [e.item for e in events if e.type == "response.output_item.done"] + first, second = events[-1].response.output + assert first.content[0].text == "skipped" + assert first.id != streamed.id + assert second.id == streamed.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 e71af75a0e..a68df6dd4d 100644 --- a/vllm/entrypoints/openai/responses/context.py +++ b/vllm/entrypoints/openai/responses/context.py @@ -656,6 +656,11 @@ class HarmonyContext(ConversationContext): self.last_append_segments: list[Segment] = [] self.last_append_flush_status: bool = False + # Each completed message with the items sent for it in + # output_item.done events, so the final response can reuse their ids. + self.streamed_items_by_message: list[ + tuple[Message, list[ResponseOutputItem]] + ] = [] # Turn tracking - replaces multiple individual tracking variables self.current_turn_metrics = TurnMetrics() diff --git a/vllm/entrypoints/openai/responses/serving.py b/vllm/entrypoints/openai/responses/serving.py index bd876e4916..3a417ca7a5 100644 --- a/vllm/entrypoints/openai/responses/serving.py +++ b/vllm/entrypoints/openai/responses/serving.py @@ -12,6 +12,7 @@ from typing import Any, Final, cast from fastapi import Request from openai.types.responses import ( ResponseOutputItem, + ResponseOutputItemDoneEvent, ResponseOutputMessage, ResponseOutputText, ResponseStatus, @@ -68,6 +69,7 @@ from vllm.entrypoints.openai.responses.utils import ( build_response_output_items, extract_function_tool_names, extract_tool_types, + reuse_streamed_item_ids, ) from vllm.entrypoints.serve.engine.protocol import ErrorResponse from vllm.entrypoints.serve.utils.api_utils import get_max_tokens @@ -781,17 +783,21 @@ class OpenAIServingResponses(GenerateBaseServing): assert isinstance(context, HarmonyContext) output = [] harmony_msgs = context.messages[context.num_init_messages :] - if harmony_msgs: - fn_names = context.function_tool_names - for msg in harmony_msgs[:-1]: - output.extend(harmony_to_response_output(msg, fn_names)) - output.extend( - harmony_to_response_output( - harmony_msgs[-1], - fn_names, - incomplete=context.last_append_flush_status, - ) + # Streamed messages are a subsequence of harmony_msgs, in order. + streamed = iter(context.streamed_items_by_message) + next_streamed = next(streamed, None) + for i, msg in enumerate(harmony_msgs): + items = harmony_to_response_output( + msg, + context.function_tool_names, + incomplete=( + i == len(harmony_msgs) - 1 and context.last_append_flush_status + ), ) + if next_streamed is not None and next_streamed[0] is msg: + reuse_streamed_item_ids(items, next_streamed[1]) + next_streamed = next(streamed, None) + output.extend(items) if request.enable_response_messages: input_messages = context.messages[: context.num_init_messages] @@ -1291,16 +1297,24 @@ class OpenAIServingResponses(GenerateBaseServing): yield _increment_sequence_number_and_return(event) elif completed_message := segment.completed_message: + done_items: list[ResponseOutputItem] = [] # TODO: Fix browser emitted as MCP calls for event in emit_previous_item_done_events( completed_message, state, ctx.function_tool_names ): + if isinstance(event, ResponseOutputItemDoneEvent): + done_items.append(event.item) yield _increment_sequence_number_and_return(event) for event in emit_tool_action_events( completed_message, state, self.tool_server ): + if isinstance(event, ResponseOutputItemDoneEvent): + done_items.append(event.item) yield _increment_sequence_number_and_return(event) + ctx.streamed_items_by_message.append( + (completed_message, done_items) + ) state.reset_for_new_item() async def responses_stream_generator( diff --git a/vllm/entrypoints/openai/responses/streaming_events.py b/vllm/entrypoints/openai/responses/streaming_events.py index b4731dc687..2f5b151a1b 100644 --- a/vllm/entrypoints/openai/responses/streaming_events.py +++ b/vllm/entrypoints/openai/responses/streaming_events.py @@ -202,7 +202,7 @@ def emit_reasoning_delta_events( events: list[StreamingResponsesResponse] = [] if not state.sent_output_item_added: state.sent_output_item_added = True - state.current_item_id = f"msg_{random_uuid()}" + state.current_item_id = f"rs_{random_uuid()}" events.append( ResponseOutputItemAddedEvent( type="response.output_item.added", diff --git a/vllm/entrypoints/openai/responses/utils.py b/vllm/entrypoints/openai/responses/utils.py index 3caff4b056..6c8507840f 100644 --- a/vllm/entrypoints/openai/responses/utils.py +++ b/vllm/entrypoints/openai/responses/utils.py @@ -112,6 +112,28 @@ def build_response_output_items( return outputs +def reuse_streamed_item_ids( + items: list[ResponseOutputItem], + streamed_items: list[ResponseOutputItem], +) -> None: + """Give rebuilt output items the ids already streamed for the same message. + + Items are paired by type in order; an item without a streamed counterpart + keeps its own id. + """ + remaining = list(streamed_items) + for item in items: + for i, streamed in enumerate(remaining): + if streamed.type != item.type: + continue + item.id = streamed.id + if isinstance(item, ResponseFunctionToolCall): + assert isinstance(streamed, ResponseFunctionToolCall) + item.call_id = streamed.call_id + del remaining[i] + break + + def should_continue_final_message( request_input: str | list[ResponseInputOutputItem], ) -> bool: