[Bugfix][Responses API] Build streamed final response from streamed items (#59307)
Signed-off-by: sfeng33 <4florafeng@gmail.com>
Flora Feng committed
Sep 29, 2026 at 20:16 UTC
7ca31ef5993dc4426d912f32b8eb3cd70fb7a078
5 files changed
+149
-73
tests/entrypoints/openai/responses/conftest.py
+5
@@ -126,6 +126,11 @@ def _validate_event_pairing(events: list, pairs_of_event_types: dict[str, str])
126
etype = event.type
127
if etype in end_events:
128
expected_start = pairs_of_event_types[etype]
129
+ # Deltas are optional, e.g. a function call with no arguments.
130
+ if expected_start.endswith(".delta") and (
131
+ not stack or stack[-1] != expected_start
132
+ ):
133
+ continue
134
assert stack and stack[-1] == expected_start, (
135
f"Stack mismatch for {etype}: "
136
f"expected {expected_start}, "
tests/entrypoints/openai/responses/test_serving_responses.py
+58
@@ -64,6 +64,7 @@ from vllm.entrypoints.openai.responses.streaming_events import (
64
)
65
from vllm.entrypoints.serve.engine.protocol import ErrorResponse
66
from vllm.inputs import tokens_input
67
+from vllm.logprobs import Logprob as SampleLogprob
68
from vllm.outputs import CompletionOutput, RequestOutput
69
from vllm.parser.harmony import Segment
70
from vllm.renderers import TokenizeParams
@@ -1982,6 +1983,63 @@ class TestAutoToolStreaming:
1983
assert function_done[0].item.arguments == tool_args
1984
1985
1986
+@pytest.mark.skip_global_cleanup
1987
+@pytest.mark.asyncio
1988
+async def test_stream_completed_response_reuses_streamed_items(monkeypatch):
1989
+ """response.completed must carry the streamed items (same ids,
1990
+ call_id, logprobs) rather than a reparse of the full output."""
1991
+ monkeypatch.setattr(envs, "VLLM_USE_EXPERIMENTAL_PARSER_CONTEXT", False)
1992
+ serving = _make_serving_instance_with_reasoning()
1993
+ call = DeltaToolCall(
1994
+ id="chatcmpl-tool-parser-id",
1995
+ index=0,
1996
+ function=DeltaFunctionCall(name="get_weather", arguments="{}"),
1997
+ )
1998
+ response_parser = _mock_parser_with_reasoning(
1999
+ serving,
2000
+ [
2001
+ DeltaMessage(reasoning="think"),
2002
+ DeltaMessage(content="Hi"),
2003
+ DeltaMessage(tool_calls=[call]),
2004
+ ],
2005
+ )
2006
+ response_parser.count_reasoning_tokens = MagicMock(return_value=1)
2007
+ context = SimpleContext(response_parser=response_parser)
2008
+
2009
+ async def result_generator():
2010
+ for token_id, text in [(10, "think"), (20, "Hi"), (30, "call")]:
2011
+ output = _make_request_output(text, [token_id])
2012
+ output.outputs[0].logprobs = [
2013
+ {token_id: SampleLogprob(logprob=-0.5, decoded_token=text)}
2014
+ ]
2015
+ context.append_output(output)
2016
+ yield context
2017
+
2018
+ events = [
2019
+ event
2020
+ async for event in serving.responses_stream_generator(
2021
+ request=ResponsesRequest(
2022
+ input="hi",
2023
+ include=["message.output_text.logprobs"],
2024
+ stream=True,
2025
+ store=False,
2026
+ ),
2027
+ sampling_params=SamplingParams(max_tokens=16),
2028
+ result_generator=result_generator(),
2029
+ context=context,
2030
+ model_name="test-model",
2031
+ tokenizer=MagicMock(),
2032
+ request_metadata=RequestResponseMetadata(request_id="req"),
2033
+ )
2034
+ ]
2035
+
2036
+ streamed = [e.item for e in events if e.type == "response.output_item.done"]
2037
+ assert events[-1].response.output == streamed
2038
+ _, message, function_call = streamed
2039
+ assert [lp.token for lp in message.content[0].logprobs] == ["Hi"]
2040
+ assert function_call.call_id == "chatcmpl-tool-parser-id"
2041
+
2042
+
2043
@pytest.mark.asyncio
2044
@pytest.mark.parametrize(("parallel_tool_calls", "num_calls"), [(False, 1), (None, 3)])
2045
async def test_streaming_parallel_tool_calls(
vllm/entrypoints/openai/responses/context.py
+3
@@ -194,6 +194,9 @@ class SimpleContext(ConversationContext):
194
self._accumulated_text: str = ""
195
self._accumulated_token_ids: list[int] = []
196
self._accumulated_logprobs: list = []
197
+ # Items already sent as output_item.done events; the final
198
+ # streaming response reuses them instead of reparsing.
199
+ self.streamed_output_items: list[ResponseOutputItem] | None = None
200
201
self.num_prompt_tokens = 0
202
self.num_output_tokens = 0
vllm/entrypoints/openai/responses/serving.py
+28
-45
@@ -15,7 +15,6 @@ from openai.types.responses import (
15
ResponseOutputMessage,
16
ResponseOutputText,
17
ResponseStatus,
18
- response_text_delta_event,
18
)
19
from openai.types.responses.response_output_text import Logprob, LogprobTopLogprob
20
from pydantic import TypeAdapter
@@ -813,14 +812,19 @@ class OpenAIServingResponses(GenerateBaseServing):
812
if final_output.finish_reason == "length":
813
status = "incomplete"
814
816
- # TODO: Build final response items from the accumulated streaming
817
- # parser results instead of reparsing the complete output.
818
- output = self._make_response_output_items(
819
- request,
820
- final_output,
821
- tokenizer,
822
- parser=context.response_parser,
815
+ streamed_items = context.streamed_output_items
816
+ self._log_final_output(
817
+ request, final_output, is_streaming=streamed_items is not None
818
)
819
+ if streamed_items is not None:
820
+ output = streamed_items
821
+ else:
822
+ output = self._make_response_output_items(
823
+ request,
824
+ final_output,
825
+ tokenizer,
826
+ parser=context.response_parser,
827
+ )
828
829
if request.enable_response_messages:
830
input_messages = context.input_messages
@@ -962,51 +966,29 @@ class OpenAIServingResponses(GenerateBaseServing):
966
)
967
return out
968
965
- def _create_stream_response_logprobs(
966
- self,
967
- token_ids: Sequence[int],
968
- logprobs: SampleLogprobs | None,
969
- tokenizer: TokenizerLike,
970
- top_logprobs: int | None = None,
971
- ) -> list[response_text_delta_event.Logprob]:
972
- lgs = self._create_response_logprobs(
973
- token_ids=token_ids,
974
- logprobs=logprobs,
975
- tokenizer=tokenizer,
976
- top_logprobs=top_logprobs,
977
- )
978
- return [
979
- response_text_delta_event.Logprob(
980
- token=lg.token,
981
- logprob=lg.logprob,
982
- top_logprobs=[
983
- response_text_delta_event.LogprobTopLogprob(
984
- token=tl.token, logprob=tl.logprob
985
- )
986
- for tl in lg.top_logprobs
987
- ],
988
- )
989
- for lg in lgs
990
- ]
991
-
992
- def _make_response_output_items(
969
+ def _log_final_output(
970
self,
971
request: ResponsesRequest,
972
final_output: CompletionOutput,
996
- tokenizer: TokenizerLike,
997
- parser: Parser | None = None,
998
- ) -> list[ResponseOutputItem]:
999
- # Log complete response if output logging is enabled
973
+ is_streaming: bool,
974
+ ) -> None:
975
if self.enable_log_outputs and self.request_logger:
976
self.request_logger.log_outputs(
977
request_id=request.request_id,
978
outputs=final_output.text,
979
output_token_ids=final_output.token_ids,
980
finish_reason=final_output.finish_reason,
1006
- is_streaming=False,
981
+ is_streaming=is_streaming,
982
delta=False,
983
)
984
985
+ def _make_response_output_items(
986
+ self,
987
+ request: ResponsesRequest,
988
+ final_output: CompletionOutput,
989
+ tokenizer: TokenizerLike,
990
+ parser: Parser | None = None,
991
+ ) -> list[ResponseOutputItem]:
992
# Compute logprobs if requested
993
logprobs = None
994
if request.is_include_output_logprobs() and final_output.logprobs:
@@ -1197,14 +1179,12 @@ class OpenAIServingResponses(GenerateBaseServing):
1179
1180
hide_stream_metadata = not request.include_reasoning and self.parser is not None
1181
1200
- def _get_logprobs(
1201
- output: CompletionOutput,
1202
- ) -> list[response_text_delta_event.Logprob]:
1182
+ def _get_logprobs(output: CompletionOutput) -> list[Logprob]:
1183
if not request.is_include_output_logprobs():
1184
return []
1185
if hide_stream_metadata:
1186
return []
1207
- return self._create_stream_response_logprobs(
1187
+ return self._create_response_logprobs(
1188
token_ids=output.token_ids,
1189
logprobs=output.logprobs,
1190
tokenizer=tokenizer,
@@ -1253,6 +1233,9 @@ class OpenAIServingResponses(GenerateBaseServing):
1233
for event in processor.close_current():
1234
yield _increment_sequence_number_and_return(event)
1235
1236
+ assert isinstance(context, SimpleContext)
1237
+ context.streamed_output_items = processor.output_items
1238
+
1239
async def _process_harmony_streaming_events(
1240
self,
1241
request: ResponsesRequest,
vllm/entrypoints/openai/responses/streaming_events.py
+55
-28
@@ -38,6 +38,7 @@ from openai.types.responses import (
38
ResponseMcpCallArgumentsDoneEvent,
39
ResponseMcpCallCompletedEvent,
40
ResponseMcpCallInProgressEvent,
41
+ ResponseOutputItem,
42
ResponseOutputItemAddedEvent,
43
ResponseOutputItemDoneEvent,
44
ResponseOutputMessage,
@@ -51,6 +52,7 @@ from openai.types.responses import (
52
ResponseWebSearchCallInProgressEvent,
53
ResponseWebSearchCallSearchingEvent,
54
response_function_web_search,
55
+ response_output_text,
56
response_text_delta_event,
57
)
58
from openai.types.responses.response_output_item import McpCall
@@ -60,6 +62,7 @@ from openai.types.responses.response_reasoning_item import (
62
from openai.types.responses.tool import Tool
63
from openai_harmony import Message as HarmonyMessage
64
65
+from vllm.entrypoints.chat_utils import make_tool_call_id
66
from vllm.entrypoints.generate.base.protocol import DeltaMessage, DeltaToolCall
67
from vllm.entrypoints.mcp.tool_server import ToolServer
68
from vllm.entrypoints.openai.parser.harmony_utils import (
@@ -816,11 +819,13 @@ class SimpleStreamingState:
819
current_item_id: str = ""
820
content_index: int = 0
821
accumulated_text: str = ""
822
+ accumulated_logprobs: list[response_output_text.Logprob] = field(
823
+ default_factory=list
824
+ )
825
tool_call_id: str = ""
826
tool_call_name: str = ""
827
tool_call_namespace: str | None = None
828
tool_call_index: int | None = None
823
- has_emitted_tool_call_delta: bool = False
829
current_state: _StateType = field(default_factory=lambda: _StateType.NONE)
830
831
@@ -828,9 +833,10 @@ def emit_simple_content_open(
833
state: SimpleStreamingState,
834
) -> list[StreamingResponsesResponse]:
835
state.current_state = _StateType.CONTENT
831
- state.current_item_id = random_uuid()
836
+ state.current_item_id = f"msg_{random_uuid()}"
837
state.content_index = 0
838
state.accumulated_text = ""
839
+ state.accumulated_logprobs = []
840
return [
841
ResponseOutputItemAddedEvent(
842
type="response.output_item.added",
@@ -860,12 +866,32 @@ def emit_simple_content_open(
866
]
867
868
869
+def _to_stream_logprobs(
870
+ logprobs: list[response_output_text.Logprob],
871
+) -> list[response_text_delta_event.Logprob]:
872
+ return [
873
+ response_text_delta_event.Logprob(
874
+ token=lp.token,
875
+ logprob=lp.logprob,
876
+ top_logprobs=[
877
+ response_text_delta_event.LogprobTopLogprob(
878
+ token=tl.token, logprob=tl.logprob
879
+ )
880
+ for tl in lp.top_logprobs
881
+ ],
882
+ )
883
+ for lp in logprobs
884
+ ]
885
+
886
+
887
def emit_simple_content_delta(
888
state: SimpleStreamingState,
889
delta: str,
866
- logprobs: list[response_text_delta_event.Logprob] | None = None,
890
+ logprobs: list[response_output_text.Logprob] | None = None,
891
) -> list[StreamingResponsesResponse]:
892
state.accumulated_text += delta
893
+ if logprobs:
894
+ state.accumulated_logprobs.extend(logprobs)
895
return [
896
ResponseTextDeltaEvent(
897
type="response.output_text.delta",
@@ -874,7 +900,7 @@ def emit_simple_content_delta(
900
output_index=state.output_index,
901
item_id=state.current_item_id,
902
delta=delta,
877
- logprobs=logprobs or [],
903
+ logprobs=_to_stream_logprobs(logprobs or []),
904
)
905
]
906
@@ -886,6 +912,7 @@ def emit_simple_content_done(
912
type="output_text",
913
text=state.accumulated_text,
914
annotations=[],
915
+ logprobs=state.accumulated_logprobs or None,
916
)
917
events: list[StreamingResponsesResponse] = [
918
ResponseTextDoneEvent(
@@ -915,7 +942,6 @@ def emit_simple_content_done(
942
role="assistant",
943
content=[part] if state.accumulated_text else [],
944
status="completed",
918
- summary=[],
945
),
946
),
947
]
@@ -928,7 +954,7 @@ def emit_simple_reasoning_open(
954
state: SimpleStreamingState,
955
) -> list[StreamingResponsesResponse]:
956
state.current_state = _StateType.REASONING
931
- state.current_item_id = random_uuid()
957
+ state.current_item_id = f"rs_{random_uuid()}"
958
state.content_index = 0
959
state.accumulated_text = ""
960
return [
@@ -1021,15 +1047,15 @@ def emit_simple_tool_call_open(
1047
name: str,
1048
index: int | None,
1049
namespace: str | None = None,
1050
+ call_id: str | None = None,
1051
) -> list[StreamingResponsesResponse]:
1052
state.current_state = _StateType.TOOL_CALL
1026
- state.current_item_id = random_uuid()
1027
- state.tool_call_id = f"call_{random_uuid()}"
1053
+ state.current_item_id = f"fc_{random_uuid()}"
1054
+ state.tool_call_id = call_id or make_tool_call_id()
1055
state.tool_call_name = name
1056
state.tool_call_namespace = namespace
1057
state.tool_call_index = index
1058
state.accumulated_text = ""
1032
- state.has_emitted_tool_call_delta = False
1059
return [
1060
ResponseOutputItemAddedEvent(
1061
type="response.output_item.added",
@@ -1053,7 +1079,6 @@ def emit_simple_tool_call_delta(
1079
delta: str,
1080
) -> list[StreamingResponsesResponse]:
1081
state.accumulated_text += delta
1056
- state.has_emitted_tool_call_delta = True
1082
return [
1083
ResponseFunctionCallArgumentsDeltaEvent(
1084
type="response.function_call_arguments.delta",
@@ -1068,19 +1093,15 @@ def emit_simple_tool_call_delta(
1093
def emit_simple_tool_call_done(
1094
state: SimpleStreamingState,
1095
) -> list[StreamingResponsesResponse]:
1071
- events: list[StreamingResponsesResponse] = []
1072
- if state.has_emitted_tool_call_delta:
1073
- events.append(
1074
- ResponseFunctionCallArgumentsDoneEvent(
1075
- type="response.function_call_arguments.done",
1076
- sequence_number=-1,
1077
- output_index=state.output_index,
1078
- item_id=state.current_item_id,
1079
- arguments=state.accumulated_text,
1080
- name=state.tool_call_name,
1081
- )
1082
- )
1083
- events.append(
1096
+ events: list[StreamingResponsesResponse] = [
1097
+ ResponseFunctionCallArgumentsDoneEvent(
1098
+ type="response.function_call_arguments.done",
1099
+ sequence_number=-1,
1100
+ output_index=state.output_index,
1101
+ item_id=state.current_item_id,
1102
+ arguments=state.accumulated_text,
1103
+ name=state.tool_call_name,
1104
+ ),
1105
ResponseOutputItemDoneEvent(
1106
type="response.output_item.done",
1107
sequence_number=-1,
@@ -1095,7 +1116,7 @@ def emit_simple_tool_call_done(
1116
call_id=state.tool_call_id,
1117
),
1118
),
1098
- )
1119
+ ]
1120
state.output_index += 1
1121
state.tool_call_namespace = None
1122
state.current_state = _StateType.NONE
@@ -1181,6 +1202,7 @@ class SimpleStreamingEventProcessor:
1202
) -> None:
1203
self.state = state or SimpleStreamingState()
1204
self.tool_call_name_map = build_responses_tool_call_name_map(tools)
1205
+ self.output_items: list[ResponseOutputItem] = []
1206
1207
def resolve_target_state(
1208
self, delta_message: DeltaMessage
@@ -1226,7 +1248,13 @@ class SimpleStreamingEventProcessor:
1248
handlers = self._STATE_HANDLERS.get(self.state.current_state)
1249
if handlers is None:
1250
return []
1229
- return handlers.done_fn(self.state)
1251
+ events = handlers.done_fn(self.state)
1252
+ self.output_items.extend(
1253
+ event.item
1254
+ for event in events
1255
+ if isinstance(event, ResponseOutputItemDoneEvent)
1256
+ )
1257
+ return events
1258
1259
def open(
1260
self, target_state: _StateType, tool_call: Any = None
@@ -1244,6 +1272,7 @@ class SimpleStreamingEventProcessor:
1272
call_name.name,
1273
tool_call.index,
1274
call_name.namespace,
1275
+ tool_call.id,
1276
)
1277
return handlers.open_fn(self.state)
1278
@@ -1251,9 +1280,7 @@ class SimpleStreamingEventProcessor:
1280
self,
1281
delta_message: DeltaMessage,
1282
output: CompletionOutput,
1254
- get_logprobs: Callable[
1255
- [CompletionOutput], list[response_text_delta_event.Logprob]
1256
- ]
1283
+ get_logprobs: Callable[[CompletionOutput], list[response_output_text.Logprob]]
1284
| None = None,
1285
) -> list[StreamingResponsesResponse]:
1286
"""Emit incremental events for the current state from the delta."""