# SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project import json from collections.abc import Sequence import pytest from openai_harmony import ( Conversation, Message, RenderConversationConfig, Role, ) from transformers import AutoTokenizer from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest from vllm.entrypoints.openai.engine.protocol import FunctionCall from vllm.entrypoints.openai.parser.harmony_utils import ( get_encoding, ) from vllm.parser.harmony import HarmonyParser from vllm.parser.parser_manager import ParserManager REASONING_MODEL_NAME = "openai/gpt-oss-20b" @pytest.fixture(scope="module") def gpt_oss_tokenizer(): return AutoTokenizer.from_pretrained(REASONING_MODEL_NAME) @pytest.fixture def harmony_parser(gpt_oss_tokenizer): parser_cls = ParserManager.get_parser( tool_parser_name="openai", reasoning_parser_name="openai_gptoss", enable_auto_tools=True, model_name=REASONING_MODEL_NAME, is_harmony=True, ) assert parser_cls is HarmonyParser return parser_cls(gpt_oss_tokenizer) @pytest.fixture def chat_request(): return ChatCompletionRequest( model="openai/gpt-oss-20b", messages=[{"role": "user", "content": "Hello"}], ) @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") def assistant(content: str, channel: str) -> Message: return Message.from_role_and_content(Role.ASSISTANT, content).with_channel(channel) def tool_call( recipient: str, content: str, channel: str = "commentary", content_type: str | None = "json", ) -> Message: message = assistant(content, channel).with_recipient(recipient) return message if content_type is None else message.with_content_type(content_type) def get_model_output_tokens( prompt_messages: Sequence[Message], response_messages: Sequence[Message], ) -> list[int]: enc = get_encoding() # Keep analysis messages when synthesizing model-output-only token sequences # for parser tests; the default render path drops them after a later final turn. config = RenderConversationConfig(auto_drop_analysis=False) prompt_ids = enc.render_conversation_for_completion( Conversation.from_messages(list(prompt_messages)), Role.ASSISTANT, config=config, ) full_ids = enc.render_conversation( Conversation.from_messages([*prompt_messages, *response_messages]), config=config, ) assert full_ids[: len(prompt_ids)] == prompt_ids return full_ids[len(prompt_ids) :] def get_text(msg: Message) -> str: return msg.content[0].text if msg.content else "" 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 [ tool_call for tool_call in delta_message.tool_calls if tool_call.function and tool_call.function.name ] 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 tool_call_entries(delta_message) -> list[tuple[int, str | None, str | None]]: if delta_message is None or not delta_message.tool_calls: return [] return [ ( tool_call.index, tool_call.function.name if tool_call.function else None, tool_call.function.arguments if tool_call.function else None, ) for tool_call in delta_message.tool_calls ] 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_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" assert flushed.recipient is None assert flushed.delta == "" assert flushed.completed_message is not None assert get_text(flushed.completed_message) == "Think" assert_parser_is_reset(harmony_parser) 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" 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 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: # Rendered conversation outputs. def test_reasoning_only(self, harmony_parser, chat_request): prompt = [Message.from_role_and_content(Role.USER, "Why?")] response = [assistant("This is reasoning", "analysis")] reasoning, content, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=get_model_output_tokens(prompt, response), ) assert reasoning == "This is reasoning" assert content is None assert tool_calls is None def test_content_only(self, harmony_parser, chat_request): prompt = [Message.from_role_and_content(Role.USER, "Hello")] response = [assistant("This is a test", "final")] reasoning, content, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=get_model_output_tokens(prompt, response), ) assert reasoning is None assert content == "This is a test" assert tool_calls is None def test_reasoning_and_content(self, harmony_parser, chat_request): prompt = [Message.from_role_and_content(Role.USER, "What is 2+2?")] response = [ assistant("I should think first.", "analysis"), assistant("The answer is 4.", "final"), ] reasoning, content, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=get_model_output_tokens(prompt, response), ) assert reasoning == "I should think first." assert content == "The answer is 4." assert tool_calls is None @pytest.mark.parametrize( "tool_args", [ '{"location": "Tokyo"}', '{\n"location": "Tokyo"\n}', ], ) @pytest.mark.parametrize("tool_channel", ["commentary", "analysis"]) def test_single_tool_call( self, harmony_parser, chat_request, tool_args, tool_channel ): prompt = [ Message.from_role_and_content(Role.USER, "What is the weather in Tokyo?") ] response = [tool_call("functions.get_current_weather", tool_args, tool_channel)] reasoning, content, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=get_model_output_tokens(prompt, response), ) assert reasoning is None assert content is None assert tool_call_tuples(tool_calls) == [ ("get_current_weather", json.dumps({"location": "Tokyo"})) ] def test_multiple_tool_calls_varied_formats(self, harmony_parser, chat_request): prompt = [ Message.from_role_and_content( Role.USER, "What is the weather in Tokyo based on where I'm at?" ) ] response = [ tool_call("functions.get_current_weather", '{"location": "Tokyo"}'), tool_call("functions.get_user_location", '{"location": "Tokyo"}'), tool_call( "functions.no_content_type", '{"location": "Tokyo"}', content_type=None, ), tool_call("functions.not_json_no_content_type", "foo", content_type=None), tool_call("functions.empty_args", "{}"), tool_call("functions.no_args", ""), ] _, content, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=get_model_output_tokens(prompt, response), ) assert content is None assert tool_call_tuples(tool_calls) == [ ("get_current_weather", json.dumps({"location": "Tokyo"})), ("get_user_location", json.dumps({"location": "Tokyo"})), ("no_content_type", json.dumps({"location": "Tokyo"})), ("not_json_no_content_type", "foo"), ("empty_args", json.dumps({})), ("no_args", ""), ] def test_tool_call_bare_recipient(self, harmony_parser, chat_request): prompt = [Message.from_role_and_content(Role.USER, "Weather?")] response = [tool_call("get_current_weather", '{"location": "Tokyo"}')] _, _, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=get_model_output_tokens(prompt, response), ) assert tool_call_tuples(tool_calls) == [ ("get_current_weather", json.dumps({"location": "Tokyo"})) ] def test_multiple_tool_calls_bare_recipients(self, harmony_parser, chat_request): prompt = [Message.from_role_and_content(Role.USER, "Use both tools.")] response = [ tool_call("get_current_weather", '{"location": "Tokyo"}'), tool_call("get_user_location", "{}"), ] _, _, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=get_model_output_tokens(prompt, response), ) assert tool_call_tuples(tool_calls) == [ ("get_current_weather", json.dumps({"location": "Tokyo"})), ("get_user_location", json.dumps({})), ] def test_assistant_recipient_not_tool(self, harmony_parser, chat_request): prompt = [Message.from_role_and_content(Role.USER, "Hello")] response = [ tool_call("assistant", "Some tool response", content_type=None), assistant("Here is the answer", "final"), ] reasoning, content, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=get_model_output_tokens(prompt, response), ) assert reasoning is None assert content == "Here is the answer" assert tool_calls is None def test_tool_call_dotted_name(self, harmony_parser, chat_request): prompt = [Message.from_role_and_content(Role.USER, "Compute 2+3")] response = [tool_call("math.sum", '{"a": 2, "b": 3}')] _, _, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=get_model_output_tokens(prompt, response), ) assert tool_call_tuples(tool_calls) == [ ("math.sum", json.dumps({"a": 2, "b": 3})) ] def test_tool_calls_with_final_content(self, harmony_parser, chat_request): prompt = [Message.from_role_and_content(Role.USER, "What is the weather?")] response = [ assistant("User asked about the weather.", "analysis"), tool_call("functions.get_current_weather", '{"location": "Tokyo"}'), assistant("This tool call will get the weather.", "final"), ] reasoning, content, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=get_model_output_tokens(prompt, response), ) assert reasoning == "User asked about the weather." assert content == "This tool call will get the weather." assert tool_call_tuples(tool_calls) == [ ("get_current_weather", json.dumps({"location": "Tokyo"})) ] # Raw/truncated Harmony output streams. def test_interrupted_first_message(self, harmony_parser, chat_request): reasoning, content, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=encode_output( "<|channel|>final<|message|>I'm in the middle of answering" ), ) assert reasoning is None assert content == "I'm in the middle of answering" assert tool_calls 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( "", chat_request, model_output_token_ids=encode_output( "<|channel|>analysis<|message|>I'm in the middle of thinking" ), ) assert reasoning == "I'm in the middle of thinking" assert content is None assert tool_calls is None assert_parser_is_reset(harmony_parser) def test_truncated_output(self, harmony_parser, chat_request): reasoning, content, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=encode_output( "<|channel|>analysis<|message|>I'm thinking.<|end|>" "<|start|>assistant<|channel|>final<|message|>" "I'm in the middle of answering" ), ) assert reasoning == "I'm thinking." assert content == "I'm in the middle of answering" assert tool_calls is None assert_parser_is_reset(harmony_parser) 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( "", chat_request, model_output_token_ids=encode_output(combined_output), ) assert reasoning == "thinking" assert content == "thinking\n" + malformed_msgs_str[-1] assert tool_calls is None assert_parser_is_reset(harmony_parser) @pytest.mark.parametrize( ("harmony_str", "expected_content"), [ ( "<|channel|>commentary<|message|>I'll search for that", "I'll search for that", ), ( "<|channel|>commentary<|message|>Let me look that up.<|end|>" "<|start|>assistant<|channel|>final<|message|>The answer is 42.<|end|>", "Let me look that up.\nThe answer is 42.", ), ], ) def test_commentary_preambles( self, harmony_parser, chat_request, harmony_str, expected_content, ): reasoning, content, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=encode_output(harmony_str), ) assert reasoning is None assert content == expected_content assert tool_calls is None def test_commentary_with_recipient_excluded(self, harmony_parser, chat_request): reasoning, content, tool_calls = harmony_parser.parse( "", chat_request, model_output_token_ids=encode_output( "<|channel|>commentary" "<|message|>Let me check the weather.<|end|>" "<|start|>assistant to=functions.get_weather" "<|channel|>commentary" '<|message|>{"location": "SF"}<|end|>' ), ) assert reasoning is None assert content == "Let me check the weather." assert tool_call_tuples(tool_calls) == [ ("get_weather", json.dumps({"location": "SF"})) ] 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=True, ) 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 assert_parser_is_reset(parser) 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 def test_malformed_msgs_recovers_raw_content( self, gpt_oss_tokenizer, chat_request, malformed_msgs_str ): parser = HarmonyParser(gpt_oss_tokenizer) 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 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( 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_call_entries(first_delta) == [ (0, "get_weather", '{"location": '), ] assert second_delta is not None assert second_delta.reasoning is None assert second_delta.content is None assert tool_call_entries(second_delta) == [(0, None, '"Paris"}')] 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_call_entries(delta) == [(0, "search", '{"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_call_entries(first_delta) == [ (0, "tool_a", '{"a": 1}'), (1, "tool_b", '{"b": '), ] assert [tool.index for tool in tool_call_headers(first_delta)] == [0, 1] assert second_delta is not None assert tool_call_entries(second_delta) == [(1, None, "2")] assert [tool.index for tool in tool_call_payloads(second_delta)] == [1] assert third_delta is not None assert third_delta.content == "Done" assert tool_call_entries(third_delta) == [ (1, None, "}"), (2, "tool_c", '{"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([]) assert result.segments == [] assert result.reasoning_token_count == 0 def test_single_channel(self, harmony_parser): result = harmony_parser.process_chunk( encode_output("<|channel|>final<|message|>Hello") ) assert [ (s.channel, s.recipient, s.delta) for s in result.segments if s.delta ] == [("final", None, "Hello")] def test_constrained_output_segment_recipient_normalized(self, harmony_parser): result = harmony_parser.process_chunk( encode_output( '<|channel|>final <|constrain|>json<|message|>{"result":true}<|end|>' ) ) content_segments = [segment for segment in result.segments if segment.delta] assert all(segment.channel == "final" for segment in content_segments) assert all(segment.recipient is None for segment in content_segments) assert ( "".join(segment.delta for segment in content_segments) == '{"result":true}' ) completed_messages = [ segment.completed_message for segment in result.segments if segment.completed_message is not None ] assert len(completed_messages) == 1 assert completed_messages[0].recipient is None def test_cross_channel(self, harmony_parser): result = harmony_parser.process_chunk( encode_output( "<|channel|>analysis<|message|>Think<|end|>" "<|start|>assistant<|channel|>final<|message|>Answer" ) ) assert [ (s.channel, s.recipient, s.delta) for s in result.segments if s.delta ] == [ ("analysis", None, "Think"), ("final", None, "Answer"), ] def test_multi_boundary(self, harmony_parser): result = harmony_parser.process_chunk( encode_output( "<|channel|>analysis<|message|>One<|end|>" "<|start|>assistant<|channel|>final<|message|>Two<|end|>" ) ) boundary_segments = [ segment for segment in result.segments if segment.completed_message is not None ] assert [ (segment.completed_message.channel, get_text(segment.completed_message)) for segment in boundary_segments ] == [ ("analysis", "One"), ("final", "Two"), ]