| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | import pytest |
| 5 | from openai.types.responses import ResponseFunctionWebSearch, ResponseStreamEvent |
| 6 | from openai_harmony import Message, Role |
| 7 | from pydantic import TypeAdapter |
| 8 | |
| 9 | from vllm.entrypoints.generate.base.protocol import ( |
| 10 | DeltaFunctionCall, |
| 11 | DeltaMessage, |
| 12 | DeltaToolCall, |
| 13 | ) |
| 14 | from vllm.entrypoints.openai.responses.streaming_events import ( |
| 15 | SimpleStreamingEventProcessor, |
| 16 | StreamingState, |
| 17 | _StateType, |
| 18 | emit_browser_tool_events, |
| 19 | emit_reasoning_delta_events, |
| 20 | emit_reasoning_done_events, |
| 21 | split_delta, |
| 22 | ) |
| 23 | |
| 24 | |
| 25 | def test_browser_find_uses_responses_action_type(): |
| 26 | """Both streaming output items must use the Responses find action type.""" |
| 27 | message = ( |
| 28 | Message.from_role_and_content( |
| 29 | Role.ASSISTANT, '{"pattern": "vLLM", "cursor": 42}' |
| 30 | ) |
| 31 | .with_channel("analysis") |
| 32 | .with_recipient("browser.find") |
| 33 | ) |
| 34 | |
| 35 | events = emit_browser_tool_events(message, StreamingState()) |
| 36 | |
| 37 | added, done = events[0], events[-1] |
| 38 | assert added.type == "response.output_item.added" |
| 39 | assert done.type == "response.output_item.done" |
| 40 | for event in (added, done): |
| 41 | assert isinstance(event.item, ResponseFunctionWebSearch) |
| 42 | assert event.item.action.type == "find_in_page" |
| 43 | assert event.item.action.pattern == "vLLM" |
| 44 | assert event.item.action.url == "cursor:42" |
| 45 | |
| 46 | |
| 47 | @pytest.mark.parametrize("harmony", [False, True], ids=["simple", "harmony"]) |
| 48 | def test_reasoning_content_parts_use_sdk_events(harmony): |
| 49 | text = "Let me think." |
| 50 | if harmony: |
| 51 | state = StreamingState() |
| 52 | events = emit_reasoning_delta_events(text, state) |
| 53 | events.extend(emit_reasoning_done_events(text, state)) |
| 54 | else: |
| 55 | processor = SimpleStreamingEventProcessor() |
| 56 | events = _run_through_processor(processor, DeltaMessage(reasoning=text)) |
| 57 | events.extend(processor.close_current()) |
| 58 | |
| 59 | assert [event.type for event in events] == [ |
| 60 | "response.output_item.added", |
| 61 | "response.content_part.added", |
| 62 | "response.reasoning_text.delta", |
| 63 | "response.reasoning_text.done", |
| 64 | "response.content_part.done", |
| 65 | "response.output_item.done", |
| 66 | ] |
| 67 | added, part_added, delta, text_done, part_done, done = events |
| 68 | assert part_added.part.type == part_done.part.type == "reasoning_text" |
| 69 | assert part_added.part.text == "" |
| 70 | assert delta.delta == text_done.text == part_done.part.text == text |
| 71 | assert done.item.content[0].text == text |
| 72 | assert done.item.id == added.item.id |
| 73 | for event in events[1:-1]: |
| 74 | assert event.item_id == added.item.id |
| 75 | assert event.content_index == part_added.content_index |
| 76 | validator = TypeAdapter(ResponseStreamEvent) |
| 77 | for event in events: |
| 78 | assert event.output_index == added.output_index |
| 79 | validator.validate_python(event.model_dump()) |
| 80 | |
| 81 | |
| 82 | def _make_tool_call( |
| 83 | index: int, name: str | None = None, arguments: str | None = None |
| 84 | ) -> DeltaToolCall: |
| 85 | fn = DeltaFunctionCall(name=name, arguments=arguments) |
| 86 | return DeltaToolCall(index=index, function=fn) |
| 87 | |
| 88 | |
| 89 | class TestSplitDelta: |
| 90 | def test_all_three_fields(self): |
| 91 | tc = _make_tool_call(0, name="f") |
| 92 | delta = DeltaMessage(reasoning="r", content="c", tool_calls=[tc]) |
| 93 | result = split_delta(delta) |
| 94 | |
| 95 | assert len(result) == 3 |
| 96 | assert result[0].reasoning == "r" and result[0].content is None |
| 97 | assert result[1].content == "c" and result[1].reasoning is None |
| 98 | assert len(result[2].tool_calls) == 1 and result[2].content is None |
| 99 | |
| 100 | def test_tool_calls_grouped_by_index(self): |
| 101 | tc0 = _make_tool_call(0, name="f1") |
| 102 | tc1 = _make_tool_call(1, name="f2") |
| 103 | tc0b = _make_tool_call(0, arguments='{"a":1}') |
| 104 | |
| 105 | # Different indices → split |
| 106 | result = split_delta(DeltaMessage(tool_calls=[tc0, tc1])) |
| 107 | assert len(result) == 2 |
| 108 | assert result[0].tool_calls == [tc0] |
| 109 | assert result[1].tool_calls == [tc1] |
| 110 | |
| 111 | # Same index → stays together |
| 112 | delta = DeltaMessage(tool_calls=[tc0, tc0b]) |
| 113 | result = split_delta(delta) |
| 114 | assert len(result) == 1 |
| 115 | assert result[0] is delta |
| 116 | |
| 117 | |
| 118 | def _run_through_processor( |
| 119 | processor: SimpleStreamingEventProcessor, |
| 120 | delta_message: DeltaMessage, |
| 121 | ) -> list: |
| 122 | """Simulate the streaming loop from serving.py for a single delta.""" |
| 123 | events = [] |
| 124 | for dm in split_delta(delta_message): |
| 125 | target_state, tool_call = processor.resolve_target_state(dm) |
| 126 | if target_state == _StateType.NONE: |
| 127 | continue |
| 128 | if processor.needs_transition(target_state, tool_call): |
| 129 | events.extend(processor.close_current()) |
| 130 | events.extend(processor.open(target_state, tool_call)) |
| 131 | events.extend(processor.emit_delta(dm, None)) |
| 132 | return events |
| 133 | |
| 134 | |
| 135 | class TestProcessorCompoundDeltas: |
| 136 | def test_all_three_states(self): |
| 137 | tc = _make_tool_call(0, name="f", arguments="{}") |
| 138 | delta = DeltaMessage(reasoning="r", content="c", tool_calls=[tc]) |
| 139 | |
| 140 | processor = SimpleStreamingEventProcessor() |
| 141 | events = _run_through_processor(processor, delta) |
| 142 | |
| 143 | types = [e.type for e in events] |
| 144 | r_idx = types.index("response.reasoning_text.delta") |
| 145 | c_idx = types.index("response.output_text.delta") |
| 146 | fc_idx = types.index("response.function_call_arguments.delta") |
| 147 | assert r_idx < c_idx < fc_idx |
| 148 | |
| 149 | def test_parallel_tool_calls(self): |
| 150 | tc0 = _make_tool_call(0, name="f1", arguments='{"a":1}') |
| 151 | tc1 = _make_tool_call(1, name="f2", arguments='{"b":2}') |
| 152 | delta = DeltaMessage(tool_calls=[tc0, tc1]) |
| 153 | |
| 154 | processor = SimpleStreamingEventProcessor() |
| 155 | events = _run_through_processor(processor, delta) |
| 156 | |
| 157 | added = [e for e in events if e.type == "response.output_item.added"] |
| 158 | deltas = [ |
| 159 | e for e in events if e.type == "response.function_call_arguments.delta" |
| 160 | ] |
| 161 | assert len(added) == 2 |
| 162 | assert len(deltas) == 2 |
| 163 | |
| 164 | def test_split_name_and_args_same_index(self): |
| 165 | """Regression: parsers like KimiK2 emit name and args as separate |
| 166 | DeltaToolCalls at the same index within one DeltaMessage.""" |
| 167 | tc_name = _make_tool_call(0, name="get_weather") |
| 168 | tc_args = _make_tool_call(0, arguments='{"city":"SF"}') |
| 169 | delta = DeltaMessage(tool_calls=[tc_name, tc_args]) |
| 170 | |
| 171 | processor = SimpleStreamingEventProcessor() |
| 172 | events = _run_through_processor(processor, delta) |
| 173 | |
| 174 | deltas = [ |
| 175 | e for e in events if e.type == "response.function_call_arguments.delta" |
| 176 | ] |
| 177 | assert len(deltas) == 1 |
| 178 | assert deltas[0].delta == '{"city":"SF"}' |
| 179 | |
| 180 | def test_reasoning_to_content_transition(self): |
| 181 | """Regression: the old special case in emit_delta handled this; |
| 182 | now split_delta handles it generically.""" |
| 183 | processor = SimpleStreamingEventProcessor() |
| 184 | _run_through_processor(processor, DeltaMessage(reasoning="think")) |
| 185 | assert processor.state.current_state == _StateType.REASONING |
| 186 | |
| 187 | events = _run_through_processor( |
| 188 | processor, DeltaMessage(reasoning="more", content="answer") |
| 189 | ) |
| 190 | types = [e.type for e in events] |
| 191 | assert "response.reasoning_text.delta" in types |
| 192 | assert "response.output_text.delta" in types |