A high-throughput and memory-efficient inference and serving engine for LLMs
Mirrored from https://github.com/tduncan-lum/Nvidia-Optimization-Stack---VLLM.git
Synced Nov 2, 17:01 UTC
[Bugfix][Frontend] streaming tool-call serializer drops first args chunk when name and args share a DeltaMessage (#42683)
Signed-off-by: ignaciosica <mignacio.sica@gmail.com> Signed-off-by: sfeng33 <4florafeng@gmail.com> Co-authored-by: sfeng33 <4florafeng@gmail.com>
Ignacio Sica committed
May 28, 2026 at 02:20 UTC
7909f82a450405fe61280ca747d31a8e9978e1fe
4 files changed
+241
-39
tests/entrypoints/openai/responses/test_serving_responses.py
+63
@@ -1124,3 +1124,66 @@ class TestAutoToolStreaming:
1124
if event.type == "response.function_call_arguments.delta"
1125
]
1126
assert "".join(argument_deltas) == '{"location":"Berlin"}'
1127
+
1128
+ @pytest.mark.skip_global_cleanup
1129
+ @pytest.mark.asyncio
1130
+ async def test_compound_content_and_tool_name_args_same_delta(self, monkeypatch):
1131
+ monkeypatch.setattr(envs, "VLLM_USE_EXPERIMENTAL_PARSER_CONTEXT", False)
1132
+
1133
+ tool_args = '{"location":"Berlin"}'
1134
+
1135
+ delta_sequence = [
1136
+ DeltaMessage(
1137
+ content="Let me check.",
1138
+ tool_calls=[
1139
+ DeltaToolCall(
1140
+ id="call_weather",
1141
+ type="function",
1142
+ index=0,
1143
+ function=DeltaFunctionCall(name="get_weather"),
1144
+ ),
1145
+ DeltaToolCall(
1146
+ index=0,
1147
+ function=DeltaFunctionCall(arguments=tool_args),
1148
+ ),
1149
+ ],
1150
+ )
1151
+ ]
1152
+
1153
+ events = await self._collect_events(delta_sequence)
1154
+
1155
+ text_deltas = [
1156
+ event.delta
1157
+ for event in events
1158
+ if event.type == "response.output_text.delta"
1159
+ ]
1160
+ assert text_deltas == ["Let me check."]
1161
+
1162
+ argument_deltas = [
1163
+ event.delta
1164
+ for event in events
1165
+ if event.type == "response.function_call_arguments.delta"
1166
+ ]
1167
+ assert argument_deltas == [tool_args]
1168
+
1169
+ types = [event.type for event in events]
1170
+ assert types.index("response.output_text.delta") < types.index(
1171
+ "response.function_call_arguments.delta"
1172
+ )
1173
+
1174
+ argument_done = [
1175
+ event
1176
+ for event in events
1177
+ if event.type == "response.function_call_arguments.done"
1178
+ ]
1179
+ assert [event.arguments for event in argument_done] == [tool_args]
1180
+
1181
+ function_done = [
1182
+ event
1183
+ for event in events
1184
+ if event.type == "response.output_item.done"
1185
+ and getattr(event.item, "type", None) == "function_call"
1186
+ ]
1187
+ assert len(function_done) == 1
1188
+ assert function_done[0].item.name == "get_weather"
1189
+ assert function_done[0].item.arguments == tool_args
tests/entrypoints/openai/responses/test_streaming_events.py
new
+126
@@ -0,0 +1,126 @@
1
+# SPDX-License-Identifier: Apache-2.0
2
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
3
+
4
+from vllm.entrypoints.openai.engine.protocol import (
5
+ DeltaFunctionCall,
6
+ DeltaMessage,
7
+ DeltaToolCall,
8
+)
9
+from vllm.entrypoints.openai.responses.streaming_events import (
10
+ SimpleStreamingEventProcessor,
11
+ _StateType,
12
+ split_delta,
13
+)
14
+
15
+
16
+def _make_tool_call(
17
+ index: int, name: str | None = None, arguments: str | None = None
18
+) -> DeltaToolCall:
19
+ fn = DeltaFunctionCall(name=name, arguments=arguments)
20
+ return DeltaToolCall(index=index, function=fn)
21
+
22
+
23
+class TestSplitDelta:
24
+ def test_all_three_fields(self):
25
+ tc = _make_tool_call(0, name="f")
26
+ delta = DeltaMessage(reasoning="r", content="c", tool_calls=[tc])
27
+ result = split_delta(delta)
28
+
29
+ assert len(result) == 3
30
+ assert result[0].reasoning == "r" and result[0].content is None
31
+ assert result[1].content == "c" and result[1].reasoning is None
32
+ assert len(result[2].tool_calls) == 1 and result[2].content is None
33
+
34
+ def test_tool_calls_grouped_by_index(self):
35
+ tc0 = _make_tool_call(0, name="f1")
36
+ tc1 = _make_tool_call(1, name="f2")
37
+ tc0b = _make_tool_call(0, arguments='{"a":1}')
38
+
39
+ # Different indices → split
40
+ result = split_delta(DeltaMessage(tool_calls=[tc0, tc1]))
41
+ assert len(result) == 2
42
+ assert result[0].tool_calls == [tc0]
43
+ assert result[1].tool_calls == [tc1]
44
+
45
+ # Same index → stays together
46
+ delta = DeltaMessage(tool_calls=[tc0, tc0b])
47
+ result = split_delta(delta)
48
+ assert len(result) == 1
49
+ assert result[0] is delta
50
+
51
+
52
+def _run_through_processor(
53
+ processor: SimpleStreamingEventProcessor,
54
+ delta_message: DeltaMessage,
55
+) -> list:
56
+ """Simulate the streaming loop from serving.py for a single delta."""
57
+ events = []
58
+ for dm in split_delta(delta_message):
59
+ target_state, tool_call = processor.resolve_target_state(dm)
60
+ if target_state == _StateType.NONE:
61
+ continue
62
+ if processor.needs_transition(target_state, tool_call):
63
+ events.extend(processor.close_current())
64
+ events.extend(processor.open(target_state, tool_call))
65
+ events.extend(processor.emit_delta(dm, None))
66
+ return events
67
+
68
+
69
+class TestProcessorCompoundDeltas:
70
+ def test_all_three_states(self):
71
+ tc = _make_tool_call(0, name="f", arguments="{}")
72
+ delta = DeltaMessage(reasoning="r", content="c", tool_calls=[tc])
73
+
74
+ processor = SimpleStreamingEventProcessor()
75
+ events = _run_through_processor(processor, delta)
76
+
77
+ types = [e.type for e in events]
78
+ r_idx = types.index("response.reasoning_text.delta")
79
+ c_idx = types.index("response.output_text.delta")
80
+ fc_idx = types.index("response.function_call_arguments.delta")
81
+ assert r_idx < c_idx < fc_idx
82
+
83
+ def test_parallel_tool_calls(self):
84
+ tc0 = _make_tool_call(0, name="f1", arguments='{"a":1}')
85
+ tc1 = _make_tool_call(1, name="f2", arguments='{"b":2}')
86
+ delta = DeltaMessage(tool_calls=[tc0, tc1])
87
+
88
+ processor = SimpleStreamingEventProcessor()
89
+ events = _run_through_processor(processor, delta)
90
+
91
+ added = [e for e in events if e.type == "response.output_item.added"]
92
+ deltas = [
93
+ e for e in events if e.type == "response.function_call_arguments.delta"
94
+ ]
95
+ assert len(added) == 2
96
+ assert len(deltas) == 2
97
+
98
+ def test_split_name_and_args_same_index(self):
99
+ """Regression: parsers like KimiK2 emit name and args as separate
100
+ DeltaToolCalls at the same index within one DeltaMessage."""
101
+ tc_name = _make_tool_call(0, name="get_weather")
102
+ tc_args = _make_tool_call(0, arguments='{"city":"SF"}')
103
+ delta = DeltaMessage(tool_calls=[tc_name, tc_args])
104
+
105
+ processor = SimpleStreamingEventProcessor()
106
+ events = _run_through_processor(processor, delta)
107
+
108
+ deltas = [
109
+ e for e in events if e.type == "response.function_call_arguments.delta"
110
+ ]
111
+ assert len(deltas) == 1
112
+ assert deltas[0].delta == '{"city":"SF"}'
113
+
114
+ def test_reasoning_to_content_transition(self):
115
+ """Regression: the old special case in emit_delta handled this;
116
+ now split_delta handles it generically."""
117
+ processor = SimpleStreamingEventProcessor()
118
+ _run_through_processor(processor, DeltaMessage(reasoning="think"))
119
+ assert processor.state.current_state == _StateType.REASONING
120
+
121
+ events = _run_through_processor(
122
+ processor, DeltaMessage(reasoning="more", content="answer")
123
+ )
124
+ types = [e.type for e in events]
125
+ assert "response.reasoning_text.delta" in types
126
+ assert "response.output_text.delta" in types
vllm/entrypoints/openai/responses/serving.py
+12
-10
@@ -85,6 +85,7 @@ from vllm.entrypoints.openai.responses.streaming_events import (
85
emit_content_delta_events,
86
emit_previous_item_done_events,
87
emit_tool_action_events,
88
+ split_delta,
89
)
90
from vllm.entrypoints.openai.responses.utils import (
91
construct_input_messages,
@@ -1414,18 +1415,19 @@ class OpenAIServingResponses(OpenAIServing):
1415
if not delta_message:
1416
continue
1417
1417
- target_state, tool_call = processor.resolve_target_state(delta_message)
1418
- if target_state == _StateType.NONE:
1419
- continue
1418
+ for dm in split_delta(delta_message):
1419
+ target_state, tool_call = processor.resolve_target_state(dm)
1420
+ if target_state == _StateType.NONE:
1421
+ continue
1422
1421
- if processor.needs_transition(target_state, tool_call):
1422
- for event in processor.close_current():
1423
- yield _increment_sequence_number_and_return(event)
1424
- for event in processor.open(target_state, tool_call):
1425
- yield _increment_sequence_number_and_return(event)
1423
+ if processor.needs_transition(target_state, tool_call):
1424
+ for event in processor.close_current():
1425
+ yield _increment_sequence_number_and_return(event)
1426
+ for event in processor.open(target_state, tool_call):
1427
+ yield _increment_sequence_number_and_return(event)
1428
1427
- for event in processor.emit_delta(delta_message, output, _get_logprobs):
1428
- yield _increment_sequence_number_and_return(event)
1429
+ for event in processor.emit_delta(dm, output, _get_logprobs):
1430
+ yield _increment_sequence_number_and_return(event)
1431
1432
for event in processor.close_current():
1433
yield _increment_sequence_number_and_return(event)
vllm/entrypoints/openai/responses/streaming_events.py
+40
-29
@@ -61,7 +61,7 @@ from openai.types.responses.response_reasoning_item import (
61
from openai_harmony import Message as HarmonyMessage
62
63
from vllm.entrypoints.mcp.tool_server import ToolServer
64
-from vllm.entrypoints.openai.engine.protocol import DeltaMessage
64
+from vllm.entrypoints.openai.engine.protocol import DeltaMessage, DeltaToolCall
65
from vllm.entrypoints.openai.parser.harmony_utils import (
66
extract_function_from_recipient,
67
is_function_recipient,
@@ -1118,6 +1118,38 @@ class _StateHandlers(NamedTuple):
1118
done_fn: Callable[..., list[StreamingResponsesResponse]]
1119
1120
1121
+def split_delta(delta: DeltaMessage) -> list[DeltaMessage]:
1122
+ """Decompose a DeltaMessage with multiple fields into atomic deltas.
1123
+
1124
+ The Responses API emits typed SSE events (one type per event), so a
1125
+ compound DeltaMessage must be split before entering the state machine.
1126
+ Order: reasoning -> content -> tool_calls (grouped by index).
1127
+ """
1128
+ has_reasoning = delta.reasoning is not None
1129
+ has_content = delta.content is not None
1130
+ has_tools = bool(delta.tool_calls)
1131
+ parts = int(has_reasoning) + int(has_content) + int(has_tools)
1132
+
1133
+ if parts <= 1 and (
1134
+ not has_tools
1135
+ or len({tc.index for tc in delta.tool_calls if tc.index is not None}) <= 1
1136
+ ):
1137
+ return [delta]
1138
+
1139
+ deltas: list[DeltaMessage] = []
1140
+ if has_reasoning:
1141
+ deltas.append(DeltaMessage(reasoning=delta.reasoning))
1142
+ if has_content:
1143
+ deltas.append(DeltaMessage(content=delta.content))
1144
+ if has_tools:
1145
+ groups: dict[int | None, list[DeltaToolCall]] = {}
1146
+ for tc in delta.tool_calls:
1147
+ groups.setdefault(tc.index, []).append(tc)
1148
+ for tcs in groups.values():
1149
+ deltas.append(DeltaMessage(tool_calls=tcs))
1150
+ return deltas or [delta]
1151
+
1152
+
1153
class SimpleStreamingEventProcessor:
1154
"""
1155
State-machine processor for the simple (non-Harmony) streaming path.
@@ -1223,38 +1255,17 @@ class SimpleStreamingEventProcessor:
1255
]
1256
| None = None,
1257
) -> list[StreamingResponsesResponse]:
1226
- """
1227
- Emit incremental events for the current state from the delta.
1228
-
1229
- Special case: when already in REASONING and the same delta also
1230
- carries content, we emit the reasoning delta, close reasoning,
1231
- open content, and then emit the content delta.
1232
- """
1258
+ """Emit incremental events for the current state from the delta."""
1259
handlers = self._STATE_HANDLERS[self.state.current_state]
1234
- events: list[StreamingResponsesResponse] = []
1235
-
1236
- # Special case: reasoning -> content inside a single delta.
1237
- if (
1238
- self.state.current_state == _StateType.REASONING
1239
- and delta_message.reasoning is not None
1240
- and delta_message.content is not None
1241
- ):
1242
- events.extend(handlers.delta_fn(self.state, delta_message.reasoning))
1243
- events.extend(self.close_current())
1244
- events.extend(self.open(_StateType.CONTENT))
1245
- content_handlers = self._STATE_HANDLERS[_StateType.CONTENT]
1246
- logprobs = get_logprobs(output) if get_logprobs else []
1247
- events.extend(
1248
- content_handlers.delta_fn(self.state, delta_message.content, logprobs)
1249
- )
1250
- return events
1260
1261
if self.state.current_state == _StateType.TOOL_CALL:
1262
assert delta_message.tool_calls is not None
1254
- tool_call_function = delta_message.tool_calls[0].function
1255
- assert tool_call_function is not None
1256
- if tool_call_function.arguments:
1257
- return handlers.delta_fn(self.state, tool_call_function.arguments)
1263
+ combined_args = ""
1264
+ for tc in delta_message.tool_calls:
1265
+ if tc.function is not None and tc.function.arguments:
1266
+ combined_args += tc.function.arguments
1267
+ if combined_args:
1268
+ return handlers.delta_fn(self.state, combined_args)
1269
return []
1270
elif self.state.current_state == _StateType.REASONING:
1271
assert delta_message.reasoning is not None