[Frontend][Gpt-oss] Use `process_eos()` to flush Harmony Parser outputs. (#46437)
Signed-off-by: Yifan Zong <yzong@redhat.com>
yzong-rh committed
Jun 26, 2026 at 17:18 UTC
77f8796d164ae938072f78561aa72da14990419c
2 files changed
+91
-61
tests/parser/test_harmony.py
+30
-1
@@ -131,6 +131,31 @@ def tool_call_entries(delta_message) -> list[tuple[int, str | None, str | None]]
131
]
132
133
134
+class TestFlush:
135
+ def test_flush(self, harmony_parser):
136
+ harmony_parser.process_chunk(
137
+ encode_output("<|channel|>analysis<|message|>Think")
138
+ )
139
+
140
+ flushed = harmony_parser.flush()
141
+
142
+ assert flushed is not None
143
+ assert flushed.channel == "analysis"
144
+ assert flushed.recipient is None
145
+ assert flushed.delta == ""
146
+ assert flushed.completed_message is not None
147
+ assert get_text(flushed.completed_message) == "Think"
148
+ assert harmony_parser._parser is None
149
+
150
+ def test_flush_resets_after_eos_error(self, harmony_parser):
151
+ harmony_parser.process_chunk(encode_output("<|channel|>analysis"))
152
+
153
+ flushed = harmony_parser.flush()
154
+
155
+ assert flushed is None
156
+ assert harmony_parser._parser is None
157
+
158
+
159
class TestParse:
160
# Rendered conversation outputs.
161
@@ -339,6 +364,7 @@ class TestParse:
364
assert reasoning is None
365
assert content == "I'm in the middle of answering"
366
assert tool_calls is None
367
+ assert harmony_parser._parser is None
368
369
def test_interrupted_reasoning_first_message(self, harmony_parser, chat_request):
370
reasoning, content, tool_calls = harmony_parser.parse(
@@ -352,6 +378,7 @@ class TestParse:
378
assert reasoning == "I'm in the middle of thinking"
379
assert content is None
380
assert tool_calls is None
381
+ assert harmony_parser._parser is None
382
383
def test_truncated_output(self, harmony_parser, chat_request):
384
reasoning, content, tool_calls = harmony_parser.parse(
@@ -367,6 +394,7 @@ class TestParse:
394
assert reasoning == "I'm thinking."
395
assert content == "I'm in the middle of answering"
396
assert tool_calls is None
397
+ assert harmony_parser._parser is None
398
399
@pytest.mark.parametrize(
400
("harmony_str", "expected_content"),
@@ -435,7 +463,7 @@ class TestParseDelta:
463
"<|end|><|start|>assistant<|channel|>final<|message|>Answer"
464
),
465
request=chat_request,
438
- finished=False,
466
+ finished=True,
467
)
468
469
assert first_delta is not None
@@ -444,6 +472,7 @@ class TestParseDelta:
472
assert second_delta is not None
473
assert second_delta.content == "Answer"
474
assert second_delta.reasoning is None
475
+ assert parser._parser is None
476
477
def test_multi_token(self, gpt_oss_tokenizer, chat_request):
478
parser = HarmonyParser(gpt_oss_tokenizer)
vllm/parser/harmony.py
+61
-60
@@ -3,12 +3,15 @@
3
4
from __future__ import annotations
5
6
+import contextlib
7
import json
8
from collections.abc import Sequence
9
from dataclasses import dataclass
10
from enum import Enum, auto
11
from typing import TYPE_CHECKING, NamedTuple
12
13
+from openai_harmony import HarmonyError
14
+
15
from vllm.entrypoints.chat_utils import make_tool_call_id
16
from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest
17
from vllm.entrypoints.openai.engine.protocol import (
@@ -28,8 +31,7 @@ from vllm.reasoning.gptoss_reasoning_parser import GptOssReasoningParser
31
from vllm.tool_parsers.gptoss_tool_parser import GptOssToolParser
32
33
if TYPE_CHECKING:
31
- from openai_harmony import Message, Role
32
- from openai_harmony import StreamState as HarmonyStreamState
34
+ from openai_harmony import Message, StreamableParser
35
36
37
class _SegmentType(Enum):
@@ -82,33 +84,46 @@ class HarmonyParser(DelegatingParser):
84
f"got {self.tool_parser.__class__.__name__}."
85
)
86
85
- self._harmony_parser = get_streamable_parser_for_assistant()
87
+ self._parser: StreamableParser | None = None
88
self._next_tool_call_index = 0
89
self._num_processed_messages = 0
90
91
@property
90
- def state(self) -> HarmonyStreamState:
91
- return self._harmony_parser.state
92
+ def _harmony_parser(self) -> StreamableParser:
93
+ """Lazily initializes the Harmony parser."""
94
+ if self._parser is None:
95
+ self._parser = get_streamable_parser_for_assistant()
96
+ return self._parser
97
+
98
+ def _poll_completed_message(self) -> Message | None:
99
+ messages = self._harmony_parser.messages
100
+ if len(messages) <= self._num_processed_messages:
101
+ return None
102
+ msg = messages[self._num_processed_messages]
103
+ self._num_processed_messages += 1
104
+ return msg
105
93
- @property
94
- def current_role(self) -> Role | None:
95
- return self._harmony_parser.current_role
106
+ def flush(self) -> Segment | None:
107
+ msg = None
108
+ with contextlib.suppress(HarmonyError):
109
+ self._harmony_parser.process_eos()
110
+ # TODO: Consider reraising
111
97
- @property
98
- def current_channel(self) -> str | None:
99
- return self._harmony_parser.current_channel
112
+ msg = self._poll_completed_message()
113
101
- @property
102
- def current_recipient(self) -> str | None:
103
- return self._harmony_parser.current_recipient
114
+ # Reset to the initial assistant-parser state for the next turn.
115
+ self._parser = None
116
+ self._num_processed_messages = 0
117
105
- @property
106
- def current_content(self) -> str:
107
- return self._harmony_parser.current_content
118
+ if msg is None:
119
+ return None
120
109
- @property
110
- def current_content_type(self) -> str | None:
111
- return self._harmony_parser.current_content_type
121
+ return Segment(
122
+ channel=msg.channel,
123
+ recipient=msg.recipient,
124
+ delta="",
125
+ completed_message=msg,
126
+ )
127
128
def parse(
129
self,
@@ -123,24 +138,32 @@ class HarmonyParser(DelegatingParser):
138
Callers must decide whether to surface them.
139
"""
140
result = self.process_chunk(model_output_token_ids)
141
+ flushed_segment = self.flush()
142
+ if flushed_segment is not None:
143
+ result.segments.append(flushed_segment)
144
145
reasoning_parts: list[str] = []
146
content_parts: list[str] = []
147
tool_calls: list[FunctionCall] = []
148
131
- def _append_parsed_message(
132
- channel: str | None,
133
- recipient: str | None,
134
- text: str,
135
- content_type: str | None = None,
136
- ) -> None:
137
- segment_type = _SegmentType.from_channel_and_recipient(channel, recipient)
149
+ for segment in result.segments:
150
+ msg = segment.completed_message
151
+ if msg is None:
152
+ continue
153
+ if msg.author.role != "assistant" or not msg.content:
154
+ continue
155
+ text = msg.content[0].text
156
+ segment_type = _SegmentType.from_channel_and_recipient(
157
+ msg.channel, msg.recipient
158
+ )
159
match segment_type:
160
case _SegmentType.REASONING if self.reasoning_parser and text:
161
reasoning_parts.append(text)
162
case _SegmentType.CONTENT if text:
163
content_parts.append(text)
164
case _SegmentType.TOOL if self.tool_parser:
165
+ recipient = msg.recipient
166
+ content_type = msg.content_type
167
assert recipient is not None
168
if content_type is not None and "json" not in content_type:
169
arguments = text
@@ -156,31 +179,6 @@ class HarmonyParser(DelegatingParser):
179
)
180
)
181
159
- for segment in result.segments:
160
- msg = segment.completed_message
161
- if msg is None:
162
- continue
163
- if msg.author.role != "assistant" or not msg.content:
164
- continue
165
- _append_parsed_message(
166
- channel=msg.channel,
167
- recipient=msg.recipient,
168
- text=msg.content[0].text,
169
- content_type=msg.content_type,
170
- )
171
-
172
- if (
173
- self.current_channel is not None
174
- or self.current_recipient is not None
175
- or self.current_content
176
- ):
177
- _append_parsed_message(
178
- channel=self.current_channel,
179
- recipient=self.current_recipient,
180
- text=self.current_content,
181
- content_type=self.current_content_type,
182
- )
183
-
182
reasoning = "\n".join(reasoning_parts) or None
183
content = "\n".join(content_parts) or None
184
return reasoning, content, tool_calls or None
@@ -194,8 +192,12 @@ class HarmonyParser(DelegatingParser):
192
*,
193
finished: bool,
194
) -> DeltaMessage | None:
197
- prev_recipient = self.current_recipient
195
+ prev_recipient = self._harmony_parser.current_recipient
196
result = self.process_chunk(delta_token_ids)
197
+ if finished:
198
+ flushed_segment = self.flush()
199
+ if flushed_segment is not None:
200
+ result.segments.append(flushed_segment)
201
combined_content = ""
202
combined_reasoning = ""
203
tool_messages: list[DeltaToolCall] = []
@@ -248,6 +250,9 @@ class HarmonyParser(DelegatingParser):
250
)
251
)
252
253
+ if finished:
254
+ self._next_tool_call_index = 0
255
+
256
if not combined_content and not combined_reasoning and not tool_messages:
257
return None
258
@@ -268,14 +273,10 @@ class HarmonyParser(DelegatingParser):
273
reasoning_token_count = 0
274
for token_id in token_ids:
275
self._harmony_parser.process(token_id)
271
- channel = self.current_channel
272
- recipient = self.current_recipient
276
+ channel = self._harmony_parser.current_channel
277
+ recipient = self._harmony_parser.current_recipient
278
delta = self._harmony_parser.last_content_delta or ""
274
- completed_message = None
275
- _messages = self._harmony_parser.messages
276
- if len(_messages) > self._num_processed_messages:
277
- completed_message = _messages[self._num_processed_messages]
278
- self._num_processed_messages += 1
279
+ completed_message = self._poll_completed_message()
280
281
if channel == "analysis" or (
282
channel == "commentary" and recipient is not None