From e0871ad2259768add6dc43e2972bd364d0d13086 Mon Sep 17 00:00:00 2001 From: yzong-rh Date: Thu, 11 Jun 2026 21:09:47 -0400 Subject: [PATCH] [Refactor] Chat Completions Streaming Harmony Refactor and Bugfixes (#45104) Signed-off-by: Yifan Zong --- .../test_serving_chat_stream_harmony.py | 471 ------------------ tests/parser/test_harmony.py | 334 ++++++++++++- .../openai/chat_completion/serving.py | 73 +-- .../openai/chat_completion/stream_harmony.py | 167 ------- vllm/parser/harmony.py | 82 ++- 5 files changed, 394 insertions(+), 733 deletions(-) delete mode 100644 tests/entrypoints/openai/chat_completion/test_serving_chat_stream_harmony.py delete mode 100644 vllm/entrypoints/openai/chat_completion/stream_harmony.py diff --git a/tests/entrypoints/openai/chat_completion/test_serving_chat_stream_harmony.py b/tests/entrypoints/openai/chat_completion/test_serving_chat_stream_harmony.py deleted file mode 100644 index 1c058adaf0..0000000000 --- a/tests/entrypoints/openai/chat_completion/test_serving_chat_stream_harmony.py +++ /dev/null @@ -1,471 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project -""" -Unit tests for harmony streaming delta extraction. -""" - -from dataclasses import dataclass, field -from unittest.mock import patch - -import pytest - -from vllm.entrypoints.openai.chat_completion.stream_harmony import ( - TokenState, - extract_harmony_streaming_delta, -) - - -@dataclass -class MockMessage: - """Mock message object for testing.""" - - channel: str | None = None - recipient: str | None = None - - -@dataclass -class MockStreamableParser: - """Mock StreamableParser for testing without openai_harmony dependency.""" - - messages: list[MockMessage] = field(default_factory=list) - - -class TestExtractHarmonyStreamingDelta: - """Tests for extract_harmony_streaming_delta function.""" - - @pytest.mark.parametrize( - "delta_text,expected_content", - [ - ("Hello, world!", "Hello, world!"), - ("", ""), - ], - ) - def test_final_channel_returns_content_delta(self, delta_text, expected_content): - """Test that final channel returns a DeltaMessage with content.""" - parser = MockStreamableParser() - - # Updated to use TokenState list - token_states = [TokenState(channel="final", recipient=None, text=delta_text)] - - delta_message, tools_streamed = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient=None, - include_reasoning=False, - ) - - assert delta_message is not None - assert delta_message.content == expected_content - assert tools_streamed is False - - @pytest.mark.parametrize( - "include_reasoning,expected_has_message", - [ - (True, True), - (False, False), - ], - ) - def test_analysis_channel_reasoning(self, include_reasoning, expected_has_message): - """Test analysis channel respects include_reasoning flag.""" - parser = MockStreamableParser() - text = "Let me think..." - token_states = [TokenState(channel="analysis", recipient=None, text=text)] - - delta_message, tools_streamed = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient=None, - include_reasoning=include_reasoning, - ) - - if expected_has_message: - assert delta_message is not None - assert delta_message.reasoning == text - else: - assert delta_message is None - assert tools_streamed is False - - @pytest.mark.parametrize("channel", ["commentary", "analysis"]) - @patch("vllm.entrypoints.openai.chat_completion.stream_harmony.make_tool_call_id") - def test_new_tool_call(self, mock_make_tool_call_id, channel): - """Test new tool call creation when recipient changes.""" - mock_make_tool_call_id.return_value = "call_test123" - parser = MockStreamableParser() - - token_states = [ - TokenState(channel=channel, recipient="functions.get_weather", text="") - ] - - delta_message, tools_streamed = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient=None, - include_reasoning=False, - ) - - assert delta_message is not None - assert len(delta_message.tool_calls) == 1 - tool_call = delta_message.tool_calls[0] - assert tool_call.id == "call_test123" - assert tool_call.type == "function" - assert tool_call.function.name == "get_weather" - assert tool_call.function.arguments == "" - assert tool_call.index == 0 - assert tools_streamed is True - - @pytest.mark.parametrize("channel", ["commentary", "analysis"]) - def test_tool_call_argument_streaming(self, channel): - """Test streaming tool call arguments (same recipient).""" - parser = MockStreamableParser() - args_text = '{"location": "Paris"}' - - token_states = [ - TokenState( - channel=channel, - recipient="functions.get_weather", - text=args_text, - ) - ] - - delta_message, tools_streamed = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient="functions.get_weather", - include_reasoning=False, - ) - - assert delta_message is not None - tool_call = delta_message.tool_calls[0] - assert tool_call.id is None - assert tool_call.function.arguments == args_text - assert tool_call.index == 0 - assert tools_streamed is True - - @pytest.mark.parametrize("channel", ["commentary", "analysis"]) - def test_tool_call_empty_arguments_returns_none(self, channel): - """Test empty delta_text with same recipient returns None.""" - parser = MockStreamableParser() - - token_states = [ - TokenState(channel=channel, recipient="functions.get_weather", text="") - ] - - delta_message, tools_streamed = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient="functions.get_weather", - include_reasoning=False, - ) - - assert delta_message is None - assert tools_streamed is False - - def test_tool_call_index_from_previous_messages(self): - """Test tool call index accounts for previous function messages.""" - messages = [ - MockMessage(channel="analysis", recipient=None), # Not counted - MockMessage(channel="commentary", recipient="functions.tool1"), # Counted - MockMessage(channel="final", recipient=None), # Not counted - ] - parser = MockStreamableParser(messages=messages) - - token_states = [ - TokenState( - channel="commentary", - recipient="functions.tool2", - text="args", - ) - ] - - delta_message, _ = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient="functions.tool2", - include_reasoning=False, - ) - - assert delta_message.tool_calls[0].index == 1 - - def test_returns_preambles_as_content(self): - """Test that commentary with no recipient (preamble) is user content.""" - parser = MockStreamableParser() - delta_text = "some text" - - token_states = [ - TokenState(channel="commentary", recipient=None, text=delta_text) - ] - - delta_message, tools_streamed = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient=None, - include_reasoning=True, - ) - - assert delta_message.content == delta_text - assert tools_streamed is False - - @pytest.mark.parametrize("channel", ["commentary", "analysis"]) - @patch("vllm.entrypoints.openai.chat_completion.stream_harmony.make_tool_call_id") - def test_new_tool_call_dotted_function_name(self, mock_make_tool_call_id, channel): - mock_make_tool_call_id.return_value = "call_dotted123" - parser = MockStreamableParser() - - token_states = [TokenState(channel=channel, recipient="math.sum", text="")] - - delta_message, tools_streamed = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient=None, - include_reasoning=False, - ) - - assert delta_message is not None - assert len(delta_message.tool_calls) == 1 - tool_call = delta_message.tool_calls[0] - assert tool_call.id == "call_dotted123" - assert tool_call.type == "function" - assert tool_call.function.name == "math.sum" - assert tool_call.function.arguments == "" - assert tool_call.index == 0 - assert tools_streamed is True - - @pytest.mark.parametrize( - "channel,recipient", - [ - (None, None), - ("unknown_channel", None), - ("commentary", "browser.search"), - ("commentary", "assistant"), - ], - ) - def test_returns_none_for_invalid_inputs(self, channel, recipient): - """Test that invalid channel/recipient combinations return None.""" - parser = MockStreamableParser() - - token_states = [ - TokenState(channel=channel, recipient=recipient, text="some text") - ] - - delta_message, tools_streamed = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient=None, - include_reasoning=True, - ) - - assert delta_message is None - assert tools_streamed is False - - def test_consecutive_token_grouping(self): - """ - Test that consecutive tokens with the same channel/recipient - are merged into a single processing group. - """ - parser = MockStreamableParser() - token_states = [ - TokenState("final", None, "H"), - TokenState("final", None, "el"), - TokenState("final", None, "lo"), - TokenState("final", None, ","), - TokenState("final", None, " World"), - ] - - delta_message, _ = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient=None, - include_reasoning=False, - ) - - assert delta_message is not None - assert delta_message.content == "Hello, World" - - @patch("vllm.entrypoints.openai.chat_completion.stream_harmony.make_tool_call_id") - def test_complex_batch_permutation(self, mock_make_id): - """ - Test a complex permutation: Reasoning -> Tool Call -> Content. - This verifies that multiple distinct actions in one batch - are all captured in the single DeltaMessage. - """ - mock_make_id.return_value = "call_batch_test" - parser = MockStreamableParser() - - token_states = [ - # 1. Reasoning - TokenState("analysis", None, "Reasoning about query..."), - # 2. Tool Calling - TokenState("commentary", "functions.search", '{"query":'), - TokenState("commentary", "functions.search", ' "vllm"}'), - # 3. Final Content - TokenState("final", None, "."), - ] - - delta_message, tools_streamed = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient=None, - include_reasoning=True, - ) - - assert delta_message is not None - - assert delta_message.reasoning == "Reasoning about query..." - - # We expect 2 objects for 1 logical tool call: - # 1. The definition (id, name, type) - # 2. The arguments payload - assert len(delta_message.tool_calls) == 2 - - header = delta_message.tool_calls[0] - payload = delta_message.tool_calls[1] - - assert header.function.name == "search" - assert header.id == "call_batch_test" - assert header.index == 0 - - assert payload.index == 0 - assert payload.function.arguments == '{"query": "vllm"}' - - assert delta_message.content == "." - assert tools_streamed is True - - @patch("vllm.entrypoints.openai.chat_completion.stream_harmony.make_tool_call_id") - def test_tool_call_index_consistency_with_ongoing_call(self, mock_make_id): - """ - Test that an ongoing tool call continuation and subsequent new calls - maintain correct indexing when interleaved with content. - """ - mock_make_id.side_effect = ["id_b", "id_c"] - - messages = [ - MockMessage(channel="commentary", recipient="functions.previous_tool") - ] - parser = MockStreamableParser(messages=messages) - - token_states = [ - TokenState("commentary", "functions.tool_a", '{"key_a": "val_a"}'), - TokenState("final", None, "Thinking..."), - TokenState("commentary", "functions.tool_b", '{"key_b": "val_b"}'), - TokenState("final", None, " Thinking again..."), - TokenState("commentary", "functions.tool_c", '{"key_c": "val_c"}'), - ] - - delta_message, _ = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient="functions.tool_a", - include_reasoning=False, - ) - - assert delta_message is not None - - tool_a_deltas = [t for t in delta_message.tool_calls if t.index == 1] - assert len(tool_a_deltas) > 0 - assert tool_a_deltas[0].id is None - assert tool_a_deltas[0].function.arguments == '{"key_a": "val_a"}' - - tool_b_header = next(t for t in delta_message.tool_calls if t.id == "id_b") - assert tool_b_header.index == 2 - tool_b_args = next( - t for t in delta_message.tool_calls if t.index == 2 and t.id is None - ) - assert tool_b_args.function.arguments == '{"key_b": "val_b"}' - - tool_c_start = next(t for t in delta_message.tool_calls if t.id == "id_c") - assert tool_c_start.index == 3 - tool_c_args = next( - t for t in delta_message.tool_calls if t.index == 3 and t.id is None - ) - assert tool_c_args.function.arguments == '{"key_c": "val_c"}' - - assert delta_message.content == "Thinking... Thinking again..." - - -class TestToolCallsOnNonStandardChannels: - """Tool calls are detected by recipient, not channel. - - Models sometimes emit tool calls on unexpected channels (e.g. ``comment`` - instead of ``commentary``). These tests verify that the streaming delta - extraction is channel-agnostic for tool call detection. - """ - - @patch("vllm.entrypoints.openai.chat_completion.stream_harmony.make_tool_call_id") - def test_prefixed_tool_call_on_comment_channel(self, mock_make_tool_call_id): - mock_make_tool_call_id.return_value = "call_comment_chan" - parser = MockStreamableParser() - - token_states = [ - TokenState(channel="comment", recipient="functions.get_weather", text="") - ] - - delta_message, tools_streamed = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient=None, - include_reasoning=False, - ) - - assert delta_message is not None - assert len(delta_message.tool_calls) == 1 - assert delta_message.tool_calls[0].function.name == "get_weather" - assert tools_streamed is True - - @patch("vllm.entrypoints.openai.chat_completion.stream_harmony.make_tool_call_id") - def test_bare_tool_call_on_comment_channel(self, mock_make_tool_call_id): - mock_make_tool_call_id.return_value = "call_bare_comment" - parser = MockStreamableParser() - - token_states = [TokenState(channel="comment", recipient="get_weather", text="")] - - delta_message, tools_streamed = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient=None, - include_reasoning=False, - ) - - assert delta_message is not None - assert len(delta_message.tool_calls) == 1 - assert delta_message.tool_calls[0].function.name == "get_weather" - assert tools_streamed is True - - def test_tool_call_arguments_on_comment_channel(self): - parser = MockStreamableParser() - args_text = '{"location": "Paris"}' - - token_states = [ - TokenState( - channel="comment", recipient="functions.get_weather", text=args_text - ) - ] - - delta_message, tools_streamed = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient="functions.get_weather", - include_reasoning=False, - ) - - assert delta_message is not None - assert delta_message.tool_calls[0].function.arguments == args_text - assert tools_streamed is True - - def test_base_index_counts_tool_calls_on_comment_channel(self): - messages = [ - MockMessage(channel="comment", recipient="functions.tool1"), - ] - parser = MockStreamableParser(messages=messages) - - token_states = [ - TokenState(channel="commentary", recipient="functions.tool2", text="args") - ] - - delta_message, _ = extract_harmony_streaming_delta( - harmony_parser=parser, - token_states=token_states, - prev_recipient="functions.tool2", - include_reasoning=False, - ) - - assert delta_message.tool_calls[0].index == 1 diff --git a/tests/parser/test_harmony.py b/tests/parser/test_harmony.py index 98687b08ed..2740ccbca0 100644 --- a/tests/parser/test_harmony.py +++ b/tests/parser/test_harmony.py @@ -94,16 +94,36 @@ def get_text(msg: Message) -> str: return msg.content[0].text if msg.content else "" -def visible_segments(result) -> list[tuple[str | None, str | None, str]]: +def tool_call_tuples(tool_calls: list[FunctionCall] | None) -> list[tuple[str, str]]: + return [] if tool_calls is None else [(tc.name, tc.arguments) for tc in tool_calls] + + +def tool_call_headers(delta_message) -> list: + if delta_message is None or not delta_message.tool_calls: + return [] return [ - (segment.channel, segment.recipient, segment.delta) - for segment in result.segments - if not segment.is_boundary and segment.delta + tool_call + for tool_call in delta_message.tool_calls + if tool_call.function and tool_call.function.name ] -def tool_call_tuples(tool_calls: list[FunctionCall] | None) -> list[tuple[str, str]]: - return [] if tool_calls is None else [(tc.name, tc.arguments) for tc in tool_calls] +def tool_call_payloads(delta_message) -> list: + if delta_message is None or not delta_message.tool_calls: + return [] + return [ + tool_call + for tool_call in delta_message.tool_calls + if tool_call.function and tool_call.function.arguments + ] + + +def combined_tool_arguments(delta_message) -> dict[int, str]: + combined: dict[int, str] = {} + for tool_call in tool_call_payloads(delta_message): + combined.setdefault(tool_call.index, "") + combined[tool_call.index] += tool_call.function.arguments + return combined class TestParse: @@ -394,6 +414,276 @@ class TestParse: ] +class TestParseDelta: + def test_basic(self, gpt_oss_tokenizer, chat_request): + parser = HarmonyParser(gpt_oss_tokenizer) + + first_delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output("<|channel|>analysis<|message|>Thinking"), + request=chat_request, + finished=False, + ) + second_delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output( + "<|end|><|start|>assistant<|channel|>final<|message|>Answer" + ), + request=chat_request, + finished=False, + ) + + assert first_delta is not None + assert first_delta.reasoning == "Thinking" + assert first_delta.content is None + assert second_delta is not None + assert second_delta.content == "Answer" + assert second_delta.reasoning is None + + def test_multi_token(self, gpt_oss_tokenizer, chat_request): + parser = HarmonyParser(gpt_oss_tokenizer) + + delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output("<|channel|>final<|message|>Hello, world!"), + request=chat_request, + finished=False, + ) + + assert delta is not None + assert delta.content == "Hello, world!" + assert delta.reasoning is None + assert not delta.tool_calls + + @pytest.mark.parametrize("tool_channel", ["commentary", "analysis"]) + def test_tool_call_split_across_deltas( + self, gpt_oss_tokenizer, chat_request, tool_channel + ): + parser = HarmonyParser(gpt_oss_tokenizer) + + first_delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output( + "<|channel|>analysis<|message|>Thinking<|end|>" + f"<|start|>assistant to=functions.get_weather<|channel|>{tool_channel}" + '<|constrain|>json<|message|>{"location": ' + ), + request=chat_request, + finished=False, + ) + second_delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output('"Paris"}<|call|>'), + request=chat_request, + finished=False, + ) + + assert first_delta is not None + assert first_delta.reasoning == "Thinking" + assert first_delta.content is None + assert [tool.function.name for tool in tool_call_headers(first_delta)] == [ + "get_weather" + ] + assert combined_tool_arguments(first_delta) == {0: '{"location": '} + assert {tool.index for tool in first_delta.tool_calls} == {0} + + assert second_delta is not None + assert second_delta.reasoning is None + assert second_delta.content is None + assert not tool_call_headers(second_delta) + assert combined_tool_arguments(second_delta) == {0: '"Paris"}'} + assert {tool.index for tool in second_delta.tool_calls} == {0} + + def test_commentary_preamble_streaming(self, gpt_oss_tokenizer, chat_request): + parser = HarmonyParser(gpt_oss_tokenizer) + + delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output( + "<|channel|>commentary<|message|>I'll search for that" + ), + request=chat_request, + finished=False, + ) + + assert delta is not None + assert delta.content == "I'll search for that" + assert delta.reasoning is None + assert not delta.tool_calls + + def test_multiple_choices(self, gpt_oss_tokenizer, chat_request): + parser_a = HarmonyParser(gpt_oss_tokenizer) + parser_b = HarmonyParser(gpt_oss_tokenizer) + + delta_a = parser_a.parse_delta( + delta_text="", + delta_token_ids=encode_output( + "<|channel|>analysis<|message|>Check weather<|end|>" + "<|start|>assistant to=functions.get_weather<|channel|>commentary" + '<|constrain|>json<|message|>{"location": "Paris"}' + ), + request=chat_request, + finished=False, + ) + delta_b = parser_b.parse_delta( + delta_text="", + delta_token_ids=encode_output( + "<|channel|>analysis<|message|>Check time<|end|>" + "<|start|>assistant to=functions.get_time<|channel|>commentary" + '<|constrain|>json<|message|>{"timezone": "UTC"}' + ), + request=chat_request, + finished=False, + ) + + assert [tool.function.name for tool in tool_call_headers(delta_a)] == [ + "get_weather" + ] + assert [tool.function.name for tool in tool_call_headers(delta_b)] == [ + "get_time" + ] + assert {tool.index for tool in delta_a.tool_calls} == {0} + assert {tool.index for tool in delta_b.tool_calls} == {0} + + def test_dotted_function_name(self, gpt_oss_tokenizer, chat_request): + parser = HarmonyParser(gpt_oss_tokenizer) + + delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output( + "<|channel|>analysis<|message|>Compute this<|end|>" + "<|start|>assistant to=math.sum<|channel|>commentary" + '<|constrain|>json<|message|>{"a": 2, "b": 3}' + ), + request=chat_request, + finished=False, + ) + + assert delta is not None + assert [tool.function.name for tool in tool_call_headers(delta)] == ["math.sum"] + assert {tool.index for tool in delta.tool_calls} == {0} + + @pytest.mark.parametrize("recipient", ["assistant", "browser"]) + def test_builtin_recipient_skipped( + self, + gpt_oss_tokenizer, + chat_request, + recipient, + ): + parser = HarmonyParser(gpt_oss_tokenizer) + prompt = [Message.from_role_and_content(Role.USER, "Hello")] + response = [tool_call(recipient, "Ignore this", content_type=None)] + + delta = parser.parse_delta( + delta_text="", + delta_token_ids=get_model_output_tokens(prompt, response), + request=chat_request, + finished=False, + ) + + assert delta is None + + def test_cross_channel_with_tool(self, gpt_oss_tokenizer, chat_request): + parser = HarmonyParser(gpt_oss_tokenizer) + + delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output( + "<|channel|>analysis<|message|>Reasoning about query...<|end|>" + "<|start|>assistant to=functions.search<|channel|>commentary" + '<|constrain|>json<|message|>{"query": "vllm"}<|call|>' + "<|start|>assistant<|channel|>final<|message|>Done" + ), + request=chat_request, + finished=False, + ) + + assert delta is not None + assert delta.reasoning == "Reasoning about query..." + assert delta.content == "Done" + assert [tool.function.name for tool in tool_call_headers(delta)] == ["search"] + assert combined_tool_arguments(delta) == {0: '{"query": "vllm"}'} + + def test_tool_index_across_calls(self, gpt_oss_tokenizer, chat_request): + parser = HarmonyParser(gpt_oss_tokenizer) + + first_delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output( + "<|channel|>analysis<|message|>Thinking<|end|>" + "<|start|>assistant to=functions.get_weather<|channel|>commentary" + '<|constrain|>json<|message|>{"location": "Paris"}<|call|>' + ), + request=chat_request, + finished=False, + ) + second_delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output( + "<|start|>assistant to=functions.get_time<|channel|>commentary" + '<|constrain|>json<|message|>{"timezone": "UTC"}<|call|>' + ), + request=chat_request, + finished=False, + ) + + assert [tool.index for tool in tool_call_headers(first_delta)] == [0] + assert [tool.index for tool in tool_call_headers(second_delta)] == [1] + assert [tool.function.name for tool in tool_call_headers(second_delta)] == [ + "get_time" + ] + + def test_multi_tool_interleaved(self, gpt_oss_tokenizer, chat_request): + parser = HarmonyParser(gpt_oss_tokenizer) + + first_delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output( + "<|channel|>analysis<|message|>Plan<|end|>" + "<|start|>assistant to=functions.tool_a<|channel|>commentary" + '<|constrain|>json<|message|>{"a": 1}<|call|>' + "<|start|>assistant to=functions.tool_b<|channel|>commentary" + '<|constrain|>json<|message|>{"b": ' + ), + request=chat_request, + finished=False, + ) + second_delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output("2"), + request=chat_request, + finished=False, + ) + third_delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output( + "}<|call|><|start|>assistant<|channel|>final<|message|>Done<|end|>" + "<|start|>assistant to=functions.tool_c<|channel|>commentary" + '<|constrain|>json<|message|>{"c": 3}' + ), + request=chat_request, + finished=False, + ) + + assert [tool.index for tool in tool_call_headers(first_delta)] == [0, 1] + assert combined_tool_arguments(first_delta) == { + 0: '{"a": 1}', + 1: '{"b": ', + } + + assert second_delta is not None + assert [tool.index for tool in tool_call_payloads(second_delta)] == [1] + assert combined_tool_arguments(second_delta) == {1: "2"} + + assert third_delta is not None + assert third_delta.content == "Done" + assert combined_tool_arguments(third_delta) == { + 1: "}", + 2: '{"c": 3}', + } + assert [tool.index for tool in tool_call_headers(third_delta)] == [2] + + class TestProcessChunk: def test_empty(self, harmony_parser): result = harmony_parser.process_chunk([]) @@ -405,7 +695,9 @@ class TestProcessChunk: encode_output("<|channel|>final<|message|>Hello") ) - assert visible_segments(result) == [("final", None, "Hello")] + assert [ + (s.channel, s.recipient, s.delta) for s in result.segments if s.delta + ] == [("final", None, "Hello")] def test_cross_channel(self, harmony_parser): result = harmony_parser.process_chunk( @@ -415,24 +707,13 @@ class TestProcessChunk: ) ) - assert visible_segments(result) == [ + assert [ + (s.channel, s.recipient, s.delta) for s in result.segments if s.delta + ] == [ ("analysis", None, "Think"), ("final", None, "Answer"), ] - def test_boundary_detection(self, harmony_parser): - result = harmony_parser.process_chunk( - encode_output("<|channel|>final<|message|>Done<|end|>") - ) - - boundary_segments = [ - segment for segment in result.segments if segment.is_boundary - ] - assert len(boundary_segments) == 1 - assert boundary_segments[0].completed_message is not None - assert boundary_segments[0].completed_message.channel == "final" - assert get_text(boundary_segments[0].completed_message) == "Done" - def test_multi_boundary(self, harmony_parser): result = harmony_parser.process_chunk( encode_output( @@ -442,11 +723,14 @@ class TestProcessChunk: ) boundary_segments = [ - segment for segment in result.segments if segment.is_boundary + segment + for segment in result.segments + if segment.completed_message is not None ] assert [ - get_text(segment.completed_message) for segment in boundary_segments + (segment.completed_message.channel, get_text(segment.completed_message)) + for segment in boundary_segments ] == [ - "One", - "Two", + ("analysis", "One"), + ("final", "Two"), ] diff --git a/vllm/entrypoints/openai/chat_completion/serving.py b/vllm/entrypoints/openai/chat_completion/serving.py index 4924ceb8b5..52d18519ef 100644 --- a/vllm/entrypoints/openai/chat_completion/serving.py +++ b/vllm/entrypoints/openai/chat_completion/serving.py @@ -33,10 +33,6 @@ from vllm.entrypoints.openai.chat_completion.protocol import ( ChatCompletionStreamResponse, ChatMessage, ) -from vllm.entrypoints.openai.chat_completion.stream_harmony import ( - TokenState, - extract_harmony_streaming_delta, -) from vllm.entrypoints.openai.engine.protocol import ( DeltaMessage, ErrorResponse, @@ -52,9 +48,6 @@ from vllm.entrypoints.openai.engine.serving import ( clamp_prompt_logprobs, ) from vllm.entrypoints.openai.models.serving import OpenAIServingModels -from vllm.entrypoints.openai.parser.harmony_utils import ( - get_streamable_parser_for_assistant, -) from vllm.entrypoints.serve.utils.api_utils import get_max_tokens, should_include_usage from vllm.entrypoints.serve.utils.request_logger import RequestLogger from vllm.entrypoints.serve.utils.tool_calls_utils import ( @@ -155,7 +148,6 @@ class OpenAIServingChat(OpenAIServing): if mc.generation_config not in ("auto", "vllm") else getattr(mc, "override_generation_config", {}).get("max_new_tokens") ) - self.use_harmony = self.model_config.hf_config.model_type == "gpt_oss" self.tool_call_id_type = get_tool_call_id_type(self.model_config) # NOTE(woosuk): While OpenAI's chat completion API supports browsing @@ -408,11 +400,6 @@ class OpenAIServingChat(OpenAIServing): finish_reason_sent = [False] * num_choices num_prompt_tokens = 0 num_cached_tokens = None - if self.use_harmony: - harmony_parsers = [ - get_streamable_parser_for_assistant() for _ in range(num_choices) - ] - harmony_tools_streamed = [False] * num_choices tools_streamed = [False] * num_choices if isinstance(request.tool_choice, ChatCompletionNamedToolChoiceParam): @@ -443,6 +430,7 @@ class OpenAIServingChat(OpenAIServing): ] for p in parsers: if p is not None: + # NOTE: HarmonyParser ignores _stream_state (uses its own FSM). p._stream_state.tool_call_id_type = self.tool_call_id_type p._stream_state.history_tool_call_cnt = history_tool_call_cnt else: @@ -572,32 +560,7 @@ class OpenAIServingChat(OpenAIServing): else: logprobs = None - if self.use_harmony: - harmony_parser = harmony_parsers[i] - prev_recipient = harmony_parser.current_recipient - - # Track accumulated content per token with their state - token_states: list[TokenState] = [] - for token_id in output.token_ids: - harmony_parser.process(token_id) - token_delta = harmony_parser.last_content_delta or "" - token_states.append( - TokenState( - harmony_parser.current_channel, - harmony_parser.current_recipient, - token_delta, - ) - ) - delta_text = "".join(delta for _, _, delta in token_states) - cur_channel = harmony_parser.current_channel - - # handle the case where several tokens where generated at once - # including the final token, leading to a delta in the text - # but the current channel to be empty (start state) - if not cur_channel and delta_text: - cur_channel = "final" - else: - delta_text = output.text + delta_text = output.text if ( not delta_text @@ -609,17 +572,7 @@ class OpenAIServingChat(OpenAIServing): delta_message: DeltaMessage | None - if self.use_harmony: - delta_message, tools_streamed_flag = ( - extract_harmony_streaming_delta( - harmony_parser=harmony_parser, - token_states=token_states, - prev_recipient=prev_recipient, - include_reasoning=request.include_reasoning, - ) - ) - harmony_tools_streamed[i] |= tools_streamed_flag - elif parser is not None: + if parser is not None: delta_message = parser.parse_delta( delta_text=delta_text, delta_token_ids=as_list(output.token_ids), @@ -627,8 +580,20 @@ class OpenAIServingChat(OpenAIServing): prompt_token_ids=res.prompt_token_ids, finished=output.finish_reason is not None, ) - if delta_message and delta_message.tool_calls: - tools_streamed[i] = True + if delta_message is not None: + if delta_message.tool_calls: + tools_streamed[i] = True + + if ( + delta_message.reasoning + and not request.include_reasoning + ): + delta_message.reasoning = None + if not ( + delta_message.content or delta_message.tool_calls + ): + delta_message = None + # handle streaming just a content delta (no parsers) else: delta_message = DeltaMessage(content=delta_text) @@ -706,9 +671,7 @@ class OpenAIServingChat(OpenAIServing): # finish_reason is: # "tool_calls" for "auto" or "required" tool calls, # and "stop" for named tool calls. - if (tools_streamed[i] and not tool_choice_function_name) or ( - self.use_harmony and harmony_tools_streamed[i] - ): + if tools_streamed[i] and not tool_choice_function_name: finish_reason_ = "tool_calls" else: finish_reason_ = ( diff --git a/vllm/entrypoints/openai/chat_completion/stream_harmony.py b/vllm/entrypoints/openai/chat_completion/stream_harmony.py deleted file mode 100644 index 271f8e8c85..0000000000 --- a/vllm/entrypoints/openai/chat_completion/stream_harmony.py +++ /dev/null @@ -1,167 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project -""" -Harmony-specific streaming delta extraction for chat completions. - -This module handles the extraction of DeltaMessage objects from -harmony parser state during streaming chat completions. -""" - -from typing import NamedTuple - -from openai_harmony import StreamableParser - -from vllm.entrypoints.chat_utils import make_tool_call_id -from vllm.entrypoints.openai.engine.protocol import ( - DeltaFunctionCall, - DeltaMessage, - DeltaToolCall, -) -from vllm.entrypoints.openai.parser.harmony_utils import ( - extract_function_from_recipient, - is_function_recipient, -) - - -class TokenState(NamedTuple): - channel: str | None - recipient: str | None - text: str - - -def extract_harmony_streaming_delta( - harmony_parser: StreamableParser, - token_states: list[TokenState], - prev_recipient: str | None, - include_reasoning: bool, -) -> tuple[DeltaMessage | None, bool]: - """ - Extract a DeltaMessage from harmony parser state during streaming. - - Args: - harmony_parser: The StreamableParser instance tracking parse state - token_states: List of TokenState tuples for each token - prev_recipient: Previous recipient for detecting tool call transitions - include_reasoning: Whether to include reasoning content - - Returns: - A tuple of (DeltaMessage or None, tools_streamed_flag) - """ - - if not token_states: - return None, False - - tools_streamed = False - - # Group consecutive tokens with same channel/recipient - groups: list[TokenState] = [] - - current_channel = token_states[0].channel - current_recipient = token_states[0].recipient - current_text = token_states[0].text - - for i in range(1, len(token_states)): - state = token_states[i] - if state.channel == current_channel and state.recipient == current_recipient: - current_text += state.text - else: - groups.append(TokenState(current_channel, current_recipient, current_text)) - current_channel = state.channel - current_recipient = state.recipient - current_text = state.text - - groups.append(TokenState(current_channel, current_recipient, current_text)) - - # Process each group and create delta messages - delta_message = None - combined_content = "" - combined_reasoning = "" - tool_messages = [] - content_encountered = False - - # Calculate base_index once before the loop - # This counts completed tool calls in messages - base_index = 0 - for msg in harmony_parser.messages: - if msg.recipient and is_function_recipient(msg.recipient): - base_index += 1 - - # If there's an ongoing tool call from previous chunk, - # the next new tool call starts at base_index + 1 - if prev_recipient and is_function_recipient(prev_recipient): - next_tool_index = base_index + 1 - # Ongoing call is at base_index - ongoing_tool_index = base_index - else: - # No ongoing call, next new call is at base_index - next_tool_index = base_index - ongoing_tool_index = None - - for group in groups: - if group.channel == "final": - combined_content += group.text - content_encountered = True - elif group.recipient and is_function_recipient(group.recipient): - opened_new_call = False - if prev_recipient != group.recipient: - # New tool call - emit the opening message - tool_name = extract_function_from_recipient(group.recipient) - tool_messages.append( - DeltaToolCall( - id=make_tool_call_id(), - type="function", - function=DeltaFunctionCall( - name=tool_name, - arguments="", - ), - index=next_tool_index, - ) - ) - opened_new_call = True - prev_recipient = group.recipient - # Increment for subsequent new tool calls - next_tool_index += 1 - - if group.text: - # Stream arguments for the ongoing tool call - if opened_new_call: - # Just opened in this group - tool_call_index = next_tool_index - 1 - else: - # Continuing from previous chunk - # If ongoing_tool_index is None here, it means - # we're continuing a call but prev_recipient - # wasn't a function. Use base_index. - tool_call_index = ( - ongoing_tool_index - if ongoing_tool_index is not None - else base_index - ) - tool_messages.append( - DeltaToolCall( - index=tool_call_index, - function=DeltaFunctionCall(arguments=group.text), - ) - ) - elif group.channel == "commentary" and group.recipient is None: - # Tool call preambles meant to be shown to the user - combined_content += group.text - content_encountered = True - elif group.channel == "analysis" and include_reasoning: - combined_reasoning += group.text - - # Combine all non-empty fields into a single message - if content_encountered or combined_reasoning or tool_messages: - delta_kwargs: dict[str, str | list[DeltaToolCall]] = {} - if content_encountered: - delta_kwargs["content"] = combined_content - if combined_reasoning: - delta_kwargs["reasoning"] = combined_reasoning - if tool_messages: - delta_kwargs["tool_calls"] = tool_messages - tools_streamed = True - delta_message = DeltaMessage(**delta_kwargs) - else: - delta_message = None - - return delta_message, tools_streamed diff --git a/vllm/parser/harmony.py b/vllm/parser/harmony.py index c1eb7ea042..f19d3675da 100644 --- a/vllm/parser/harmony.py +++ b/vllm/parser/harmony.py @@ -9,9 +9,12 @@ from dataclasses import dataclass from enum import Enum, auto from typing import TYPE_CHECKING, NamedTuple +from vllm.entrypoints.chat_utils import make_tool_call_id from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest from vllm.entrypoints.openai.engine.protocol import ( + DeltaFunctionCall, DeltaMessage, + DeltaToolCall, FunctionCall, ) from vllm.entrypoints.openai.parser.harmony_utils import ( @@ -52,7 +55,6 @@ class Segment(NamedTuple): channel: str | None recipient: str | None delta: str - is_boundary: bool = False completed_message: Message | None = None @@ -81,10 +83,8 @@ class HarmonyParser(DelegatingParser): ) self._harmony_parser = get_streamable_parser_for_assistant() - - @property - def messages(self) -> list[Message]: - return self._harmony_parser.messages + self._next_tool_call_index = 0 + self._num_processed_messages = 0 @property def state(self) -> HarmonyStreamState: @@ -194,17 +194,69 @@ class HarmonyParser(DelegatingParser): *, finished: bool, ) -> DeltaMessage | None: - raise NotImplementedError( - "HarmonyParser streaming parsing is deferred. " - "Use the existing harmony streaming path." - ) + prev_recipient = self.current_recipient + result = self.process_chunk(delta_token_ids) + combined_content = "" + combined_reasoning = "" + tool_messages: list[DeltaToolCall] = [] + + for segment in result.segments: + if segment.completed_message is not None: + prev_recipient = None + continue + + segment_type = _SegmentType.from_channel_and_recipient( + segment.channel, segment.recipient + ) + match segment_type: + case _SegmentType.REASONING: + combined_reasoning += segment.delta + case _SegmentType.CONTENT: + combined_content += segment.delta + case _SegmentType.TOOL: + assert segment.recipient is not None + if prev_recipient != segment.recipient: + tool_name = extract_function_from_recipient(segment.recipient) + tool_messages.append( + DeltaToolCall( + # HarmonyParser does not use _stream_state; + # "random" tool_call_id_type is always used + id=make_tool_call_id(), + type="function", + function=DeltaFunctionCall( + name=tool_name, + arguments=segment.delta, + ), + index=self._next_tool_call_index, + ) + ) + self._next_tool_call_index += 1 + prev_recipient = segment.recipient + elif segment.delta: + tool_call_index = self._next_tool_call_index - 1 + tool_messages.append( + DeltaToolCall( + index=tool_call_index, + function=DeltaFunctionCall(arguments=segment.delta), + ) + ) + + if not combined_content and not combined_reasoning and not tool_messages: + return None + + delta_message = DeltaMessage() + if combined_content: + delta_message.content = combined_content + if combined_reasoning: + delta_message.reasoning = combined_reasoning + if tool_messages: + delta_message.tool_calls = tool_messages + return delta_message def process_chunk(self, token_ids: Sequence[int]) -> ChunkResult: if not token_ids: return ChunkResult(segments=[], reasoning_token_count=0) - from openai_harmony import StreamState - segments: list[Segment] = [] reasoning_token_count = 0 for token_id in token_ids: @@ -213,9 +265,10 @@ class HarmonyParser(DelegatingParser): recipient = self.current_recipient delta = self._harmony_parser.last_content_delta or "" completed_message = None - is_boundary = self.state == StreamState.EXPECT_START - if is_boundary and self.messages: - completed_message = self.messages[-1] + _messages = self._harmony_parser.messages + if len(_messages) > self._num_processed_messages: + completed_message = _messages[self._num_processed_messages] + self._num_processed_messages += 1 if channel == "analysis" or ( channel == "commentary" and recipient is not None @@ -227,7 +280,6 @@ class HarmonyParser(DelegatingParser): channel=channel, recipient=recipient, delta=delta, - is_boundary=is_boundary, completed_message=completed_message, ) )