From 0cd6f767e36f78598c2a932478b59c2f78897ff3 Mon Sep 17 00:00:00 2001 From: yzong-rh Date: Sat, 4 Jul 2026 10:46:24 -0400 Subject: [PATCH] [Bugfix][Frontend][gpt-oss] Recover raw tail when Harmony parser ends non-terminal (#47379) --- tests/entrypoints/unit_tests/test_context.py | 35 ++++--- tests/parser/test_harmony.py | 104 +++++++++++++------ vllm/entrypoints/openai/responses/context.py | 12 +-- vllm/parser/harmony.py | 75 +++++++------ 4 files changed, 144 insertions(+), 82 deletions(-) diff --git a/tests/entrypoints/unit_tests/test_context.py b/tests/entrypoints/unit_tests/test_context.py index 0fa3661f8ff..1c1f6ed2359 100644 --- a/tests/entrypoints/unit_tests/test_context.py +++ b/tests/entrypoints/unit_tests/test_context.py @@ -74,7 +74,7 @@ class FakeHarmonyParser(HarmonyParser): self.reasoning_parser = None self.tool_parser = None self._chunk_results: list[ChunkResult] = [] - self._flush_results: list[Segment | None] = [] + self._flush_results: list[list[Segment]] = [] self.processed_chunks: list[list[int]] = [] def enqueue_chunk_result( @@ -89,7 +89,7 @@ class FakeHarmonyParser(HarmonyParser): ) ) - def enqueue_flush_result(self, segment: Segment | None) -> None: + def enqueue_flush_result(self, segment: list[Segment]) -> None: self._flush_results.append(segment) def process_chunk(self, token_ids) -> ChunkResult: @@ -98,10 +98,10 @@ class FakeHarmonyParser(HarmonyParser): return self._chunk_results.pop(0) return ChunkResult(segments=[], reasoning_token_count=0) - def flush(self) -> Segment | None: + def flush(self) -> list[Segment]: if self._flush_results: return self._flush_results.pop(0) - return None + return [] def make_harmony_context( @@ -598,13 +598,21 @@ async def test_streaming_message_synchronization(): content=[TextContent(text=response_text)], recipient=Role.USER, ) - flush_segment = Segment( - channel="commentary", - recipient=None, - delta="", - completed_message=message, - ) - parser.enqueue_flush_result(flush_segment) + flush_segments = [ + Segment( + channel="final", + recipient=None, + delta=response_text, + completed_message=None, + ), + Segment( + channel="final", + recipient=None, + delta="", + completed_message=message, + ), + ] + parser.enqueue_flush_result(flush_segments) # Create another output to trigger synchronization via flush() context.append_output( @@ -618,8 +626,9 @@ async def test_streaming_message_synchronization(): assert context.num_init_messages == 1 assert context._messages[2].content[0].text == response_text assert context.last_append_flush_status is True - assert len(context.last_append_segments) == 1 - assert context.last_append_segments[0].completed_message is message + assert len(context.last_append_segments) == 2 + assert context.last_append_segments[-2].delta == response_text + assert context.last_append_segments[-1].completed_message is message def test_turn_metrics_copy_and_reset(): diff --git a/tests/parser/test_harmony.py b/tests/parser/test_harmony.py index ba90252fe1c..3e9cff64aa1 100644 --- a/tests/parser/test_harmony.py +++ b/tests/parser/test_harmony.py @@ -7,7 +7,6 @@ from collections.abc import Sequence import pytest from openai_harmony import ( Conversation, - HarmonyError, Message, RenderConversationConfig, Role, @@ -51,6 +50,15 @@ def chat_request(): ) +@pytest.fixture +def malformed_msgs_str() -> list[str]: + return [ + "<|channel|>analysis<|message|>thinking<|end|>", + "<|start|>assistant<|channel|>commentary<|message|>thinking<|end|>", + '<|start|>assistant<|channel|>final {"answer": "hi"}<|return|>', + ] + + def encode_output(harmony_str: str) -> list[int]: return get_encoding().encode(harmony_str, allowed_special="all") @@ -131,13 +139,22 @@ def tool_call_entries(delta_message) -> list[tuple[int, str | None, str | None]] ] +def assert_parser_is_reset(harmony_parser: HarmonyParser): + assert harmony_parser._parser is None + assert harmony_parser._num_processed_messages == 0 + assert harmony_parser._current_message_tokens == [] + + class TestFlush: def test_flush(self, harmony_parser): harmony_parser.process_chunk( encode_output("<|channel|>analysis<|message|>Think") ) - flushed = harmony_parser.flush() + flushed_segments = harmony_parser.flush() + assert flushed_segments is not None + assert len(flushed_segments) == 1 + flushed = flushed_segments[0] assert flushed is not None assert flushed.channel == "analysis" @@ -145,15 +162,27 @@ class TestFlush: assert flushed.delta == "" assert flushed.completed_message is not None assert get_text(flushed.completed_message) == "Think" - assert harmony_parser._parser is None + assert_parser_is_reset(harmony_parser) - def test_flush_raises_and_resets_on_non_terminal_eos(self, harmony_parser): - harmony_parser.process_chunk(encode_output("<|channel|>analysis")) + def test_flush_recovers_invalid_output(self, harmony_parser, malformed_msgs_str): + for msg_str in malformed_msgs_str[:-1]: + chunk = harmony_parser.process_chunk(encode_output(msg_str)) + assert "".join(segment.delta for segment in chunk.segments) == "thinking" - with pytest.raises(HarmonyError): - harmony_parser.flush() + last_msg_str = malformed_msgs_str[-1] + harmony_parser.process_chunk(encode_output(last_msg_str)) + flushed_segments = harmony_parser.flush() + assert len(flushed_segments) == 2 + delta_segment = flushed_segments[0] + message_segment = flushed_segments[1] - assert harmony_parser._parser is None + assert delta_segment.channel == "final" + assert delta_segment.recipient is None + assert delta_segment.delta == last_msg_str + assert message_segment.channel == "final" + assert message_segment.recipient is None + assert get_text(message_segment.completed_message) == last_msg_str + assert_parser_is_reset(harmony_parser) class TestParse: @@ -364,7 +393,7 @@ class TestParse: assert reasoning is None assert content == "I'm in the middle of answering" assert tool_calls is None - assert harmony_parser._parser is None + assert_parser_is_reset(harmony_parser) def test_interrupted_reasoning_first_message(self, harmony_parser, chat_request): reasoning, content, tool_calls = harmony_parser.parse( @@ -378,7 +407,7 @@ class TestParse: assert reasoning == "I'm in the middle of thinking" assert content is None assert tool_calls is None - assert harmony_parser._parser is None + assert_parser_is_reset(harmony_parser) def test_truncated_output(self, harmony_parser, chat_request): reasoning, content, tool_calls = harmony_parser.parse( @@ -394,24 +423,23 @@ class TestParse: assert reasoning == "I'm thinking." assert content == "I'm in the middle of answering" assert tool_calls is None - assert harmony_parser._parser is None + assert_parser_is_reset(harmony_parser) - def test_malformed_final_recovers_raw_content(self, harmony_parser, chat_request): - raw_output = ( - "<|channel|>analysis<|message|>thinking<|end|>" - '<|start|>assistant<|channel|>final {"answer": "hi"}<|return|>' - ) + def test_malformed_msgs_recovers_raw_content( + self, harmony_parser, chat_request, malformed_msgs_str + ): + combined_output = "".join(malformed_msgs_str) reasoning, content, tool_calls = harmony_parser.parse( - raw_output, + "", chat_request, - model_output_token_ids=encode_output(raw_output), + model_output_token_ids=encode_output(combined_output), ) - assert content == raw_output - assert reasoning is None + assert reasoning == "thinking" + assert content == "thinking\n" + malformed_msgs_str[-1] assert tool_calls is None - assert harmony_parser._parser is None + assert_parser_is_reset(harmony_parser) @pytest.mark.parametrize( ("harmony_str", "expected_content"), @@ -489,7 +517,7 @@ class TestParseDelta: assert second_delta is not None assert second_delta.content == "Answer" assert second_delta.reasoning is None - assert parser._parser is None + assert_parser_is_reset(parser) def test_multi_token(self, gpt_oss_tokenizer, chat_request): parser = HarmonyParser(gpt_oss_tokenizer) @@ -506,25 +534,33 @@ class TestParseDelta: assert delta.reasoning is None assert not delta.tool_calls - def test_malformed_final_recovers_raw_content( - self, gpt_oss_tokenizer, chat_request + def test_malformed_msgs_recovers_raw_content( + self, gpt_oss_tokenizer, chat_request, malformed_msgs_str ): parser = HarmonyParser(gpt_oss_tokenizer) - delta = parser.parse_delta( - delta_text='final {"answer": "hi"}', - delta_token_ids=encode_output( - '<|channel|>final {"answer": "hi"}<|return|>' - ), + for msg_str in malformed_msgs_str[:-1]: + delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output(msg_str), + request=chat_request, + finished=False, + ) + assert delta.reasoning or delta.content == "thinking" + assert not delta.tool_calls + + last_delta = parser.parse_delta( + delta_text="", + delta_token_ids=encode_output(malformed_msgs_str[-1]), request=chat_request, finished=True, ) - assert delta is not None - assert delta.content == 'final {"answer": "hi"}' - assert delta.reasoning is None - assert not delta.tool_calls - assert parser._parser is None + assert last_delta is not None + assert last_delta.content == malformed_msgs_str[-1] + assert last_delta.reasoning is None + assert not last_delta.tool_calls + assert_parser_is_reset(parser) @pytest.mark.parametrize("tool_channel", ["commentary", "analysis"]) def test_tool_call_split_across_deltas( diff --git a/vllm/entrypoints/openai/responses/context.py b/vllm/entrypoints/openai/responses/context.py index d75e5d5a548..92aad86b211 100644 --- a/vllm/entrypoints/openai/responses/context.py +++ b/vllm/entrypoints/openai/responses/context.py @@ -18,7 +18,7 @@ from openai.types.responses.response_output_item import McpCall from openai.types.responses.response_output_message import ResponseOutputMessage from openai.types.responses.response_output_text import ResponseOutputText from openai.types.responses.tool import Mcp -from openai_harmony import Author, HarmonyError, Message, Role, TextContent +from openai_harmony import Author, Message, Role, TextContent from vllm import envs from vllm.entrypoints.chat_utils import ( @@ -616,7 +616,7 @@ class HarmonyContext(ConversationContext): self.num_tool_output_tokens = 0 self.last_append_segments: list[Segment] = [] - self.last_append_flush_status: bool | HarmonyError = False + self.last_append_flush_status: bool = False # Turn tracking - replaces multiple individual tracking variables self.current_turn_metrics = TurnMetrics() @@ -643,10 +643,10 @@ class HarmonyContext(ConversationContext): if output.finished: self.finish_reason = output.outputs[0].finish_reason - flushed = self.response_parser.flush() - if flushed is not None: - segments.append(flushed) - self.last_append_flush_status = flushed is not None + flushed_segments = self.response_parser.flush() + if flushed_segments: + segments.extend(flushed_segments) + self.last_append_flush_status = len(flushed_segments) > 0 self.all_turn_metrics.append(self.current_turn_metrics.copy()) self.current_turn_metrics.reset() diff --git a/vllm/parser/harmony.py b/vllm/parser/harmony.py index 80fd02e4cec..5043ca191f3 100644 --- a/vllm/parser/harmony.py +++ b/vllm/parser/harmony.py @@ -9,7 +9,7 @@ from dataclasses import dataclass from enum import Enum, auto from typing import TYPE_CHECKING, NamedTuple -from openai_harmony import HarmonyError +from openai_harmony import HarmonyError, Message, Role from vllm.entrypoints.chat_utils import make_tool_call_id from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest @@ -91,6 +91,9 @@ class HarmonyParser(DelegatingParser): self._next_tool_call_index = 0 self._num_processed_messages = 0 + # For error recovery + self._current_message_tokens: list[int] = [] + @property def _harmony_parser(self) -> StreamableParser: """Lazily initializes the Harmony parser.""" @@ -107,32 +110,48 @@ class HarmonyParser(DelegatingParser): self._num_processed_messages += 1 return msg - def flush(self) -> Segment | None: + def flush(self) -> list[Segment]: + segments: list[Segment] = [] try: self._harmony_parser.process_eos() msg = self._poll_completed_message() except HarmonyError: logger.warning( "Harmony parser ended in a non-terminal state; returning the " - "raw unparsed output. This usually indicates a malformed " - "assistant turn, e.g. a 'final' channel missing the " - "<|message|> delimiter." + "recovered raw output." ) - raise - finally: - # Reset to the initial assistant-parser state for the next turn. - self._parser = None - self._num_processed_messages = 0 + + final_channel = "final" + text = self.model_tokenizer.decode(self._current_message_tokens) + segments.append( + Segment( + channel=final_channel, + recipient=None, + delta=text, + completed_message=None, + ) + ) + msg = Message.from_role_and_content(Role.ASSISTANT, text).with_channel( + final_channel + ) + + # Reset to the initial assistant-parser state for the next turn. + self._parser = None + self._num_processed_messages = 0 + self._current_message_tokens.clear() if msg is None: - return None + return segments - return Segment( - channel=msg.channel, - recipient=msg.recipient, - delta="", - completed_message=msg, + segments.append( + Segment( + channel=msg.channel, + recipient=msg.recipient, + delta="", + completed_message=msg, + ) ) + return segments def parse( self, @@ -147,12 +166,9 @@ class HarmonyParser(DelegatingParser): Callers must decide whether to surface them. """ result = self.process_chunk(model_output_token_ids) - try: - flushed_segment = self.flush() - except HarmonyError: - return None, model_output, None - if flushed_segment is not None: - result.segments.append(flushed_segment) + flushed_segments = self.flush() + if flushed_segments: + result.segments.extend(flushed_segments) reasoning_parts: list[str] = [] content_parts: list[str] = [] @@ -209,13 +225,9 @@ class HarmonyParser(DelegatingParser): ) result = self.process_chunk(delta_token_ids) if finished: - try: - flushed_segment = self.flush() - except HarmonyError: - self._next_tool_call_index = 0 - return DeltaMessage(content=delta_text) - if flushed_segment is not None: - result.segments.append(flushed_segment) + flushed_segments = self.flush() + if flushed_segments: + result.segments.extend(flushed_segments) combined_content = "" combined_reasoning = "" tool_messages: list[DeltaToolCall] = [] @@ -298,6 +310,11 @@ class HarmonyParser(DelegatingParser): delta = self._harmony_parser.last_content_delta or "" completed_message = self._poll_completed_message() + if completed_message is not None: + self._current_message_tokens.clear() + else: + self._current_message_tokens.append(token_id) + if channel == "analysis" or ( channel == "commentary" and recipient is not None ):