[Refactor] Chat Completions Streaming Harmony Refactor and Bugfixes (#45104)
Signed-off-by: Yifan Zong <yzong@redhat.com>
This commit is contained in:
yzong-rh
@@ -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
|
||||
+309
-25
@@ -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"),
|
||||
]
|
||||
|
||||
@@ -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_ = (
|
||||
|
||||
@@ -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
|
||||
+67
-15
@@ -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,
|
||||
)
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user