[Frontend]Responses API supports Tool/Function calling with streaming with required (#40700)
Signed-off-by: chaunceyjiang <chaunceyjiang@gmail.com>
This commit is contained in:
Chauncey
@@ -321,12 +321,100 @@ async def test_function_calling_with_streaming_expected_arguments(
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("model_name", [MODEL_NAME])
|
||||
@pytest.mark.parametrize(
|
||||
"tool_choice",
|
||||
["auto", "required"],
|
||||
)
|
||||
async def test_function_calling_with_streaming_types(
|
||||
client: openai.AsyncOpenAI, model_name: str
|
||||
client: openai.AsyncOpenAI, model_name: str, tool_choice
|
||||
):
|
||||
# this links the "done" type with the "start" type
|
||||
# so every "done" type should have a corresponding "start" type
|
||||
# and every open block should be closed by the end of the stream
|
||||
#
|
||||
# stream of events for a response with function call could look like this:
|
||||
# option1: reasoning -> content(option) -> function_call
|
||||
# response.created
|
||||
# -> response.in_progress
|
||||
# -> response.output_item.added
|
||||
# -> response.reasoning_part.added
|
||||
# -> response.reasoning_text.delta
|
||||
# ....
|
||||
# -> response.reasoning_text.delta
|
||||
# -> response.reasoning_text.done
|
||||
# -> response.reasoning_part.done
|
||||
# -> response.output_item.done
|
||||
# -> response.output_item.added
|
||||
# -> response.content_part.added
|
||||
# -> response.output_text.delta
|
||||
# ...
|
||||
# -> response.output_text.delta
|
||||
# -> response.output_text.done
|
||||
# -> response.content_part.done
|
||||
# -> response.output_item.done
|
||||
# -> response.output_item.added
|
||||
# -> response.function_call_arguments.delta
|
||||
# ...
|
||||
# -> response.function_call_arguments.delta
|
||||
# -> response.function_call_arguments.done
|
||||
# -> response.output_item.done
|
||||
# -> response.completed
|
||||
#
|
||||
#
|
||||
# option2: reasoning -> content
|
||||
# response.created
|
||||
# -> response.in_progress
|
||||
# -> response.output_item.added
|
||||
# -> response.reasoning_part.added
|
||||
# -> response.reasoning_text.delta
|
||||
# ....
|
||||
# -> response.reasoning_text.delta
|
||||
# -> response.reasoning_text.done
|
||||
# -> response.reasoning_part.done
|
||||
# -> response.output_item.done
|
||||
# -> response.output_item.added
|
||||
# -> response.content_part.added
|
||||
# -> response.output_text.delta
|
||||
# ..
|
||||
# -> response.output_text.delta
|
||||
# -> response.output_text.done
|
||||
# -> response.content_part.done
|
||||
# -> response.output_item.done
|
||||
# -> response.completed
|
||||
#
|
||||
# option3: content
|
||||
#
|
||||
# response.created
|
||||
# -> response.in_progress
|
||||
# -> response.output_item.added
|
||||
# -> response.content_part.added
|
||||
# -> response.output_text.delta
|
||||
# ...
|
||||
# -> response.output_text.delta
|
||||
# -> response.output_text.done
|
||||
# -> response.content_part.done
|
||||
# -> response.output_item.done
|
||||
# -> response.completed
|
||||
#
|
||||
# option4: content -> function_call
|
||||
# response.created
|
||||
# -> response.in_progress
|
||||
# -> response.output_item.added
|
||||
# -> response.content_part.added
|
||||
# -> response.output_text.delta
|
||||
# ...
|
||||
# -> response.output_text.delta
|
||||
# -> response.output_text.done
|
||||
# -> response.content_part.done
|
||||
# -> response.output_item.done
|
||||
# -> response.output_item.added
|
||||
# -> response.function_call_arguments.delta
|
||||
# ...
|
||||
# -> response.function_call_arguments.delta
|
||||
# -> response.function_call_arguments.done
|
||||
# -> response.output_item.done
|
||||
# -> response.completed
|
||||
|
||||
pairs_of_event_types = {
|
||||
"response.completed": "response.created",
|
||||
"response.output_item.done": "response.output_item.added",
|
||||
@@ -347,6 +435,7 @@ async def test_function_calling_with_streaming_types(
|
||||
model=model_name,
|
||||
input=input_list,
|
||||
tools=tools,
|
||||
tool_choice=tool_choice,
|
||||
stream=True,
|
||||
)
|
||||
|
||||
@@ -367,3 +456,69 @@ async def test_function_calling_with_streaming_types(
|
||||
assert stack_of_event_types[-1] == pairs_of_event_types[event.type]
|
||||
stack_of_event_types.pop()
|
||||
assert len(stack_of_event_types) == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("model_name", [MODEL_NAME])
|
||||
@pytest.mark.parametrize(
|
||||
"tool_choice",
|
||||
["required", "auto"],
|
||||
)
|
||||
async def test_function_calling_with_streaming_forced_tool_choice(
|
||||
client: openai.AsyncOpenAI, model_name: str, tool_choice: str
|
||||
):
|
||||
tools = [
|
||||
{
|
||||
"type": "function",
|
||||
"name": "get_weather",
|
||||
"description": "Get current temperature for provided location in celsius.",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"location": {"type": "string"},
|
||||
},
|
||||
"required": ["location"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
"strict": True,
|
||||
}
|
||||
]
|
||||
|
||||
stream_response = await client.responses.create(
|
||||
model=model_name,
|
||||
input="Call the get_weather function for Berlin and do not answer directly.",
|
||||
tools=tools,
|
||||
tool_choice=tool_choice,
|
||||
stream=True,
|
||||
)
|
||||
|
||||
tool_call_item = None
|
||||
completed_event = None
|
||||
text_deltas = []
|
||||
async for event in stream_response:
|
||||
if (
|
||||
event.type == "response.output_item.added"
|
||||
and event.item.type == "function_call"
|
||||
):
|
||||
tool_call_item = event.item
|
||||
elif event.type == "response.output_text.delta":
|
||||
text_deltas.append(event.delta)
|
||||
elif event.type == "response.function_call_arguments.delta" and tool_call_item:
|
||||
tool_call_item.arguments += event.delta
|
||||
elif (
|
||||
event.type == "response.output_item.done"
|
||||
and event.item.type == "function_call"
|
||||
):
|
||||
completed_event = event
|
||||
|
||||
assert tool_call_item is not None
|
||||
assert tool_call_item.type == "function_call"
|
||||
assert tool_call_item.name == "get_weather"
|
||||
assert completed_event is not None
|
||||
assert tool_call_item.arguments == completed_event.item.arguments
|
||||
assert tool_call_item.name == completed_event.item.name
|
||||
args = json.loads(tool_call_item.arguments)
|
||||
assert "location" in args
|
||||
assert args["location"] is not None
|
||||
# Forced tool choice should not leak tool-call JSON via output_text delta.
|
||||
assert "".join(text_deltas).strip() == ""
|
||||
|
||||
@@ -9,10 +9,7 @@ from collections.abc import Sequence as GenericSequence
|
||||
from http import HTTPStatus
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
|
||||
import partial_json_parser
|
||||
import regex as re
|
||||
from fastapi import Request
|
||||
from partial_json_parser.core.options import Allow
|
||||
|
||||
from vllm.engine.protocol import EngineClient
|
||||
from vllm.entrypoints.chat_utils import (
|
||||
@@ -73,7 +70,7 @@ from vllm.reasoning import ReasoningParser
|
||||
from vllm.renderers import ChatParams
|
||||
from vllm.sampling_params import BeamSearchParams, SamplingParams
|
||||
from vllm.tokenizers import TokenizerLike
|
||||
from vllm.tool_parsers.utils import partial_json_loads
|
||||
from vllm.tool_parsers.streaming import extract_required_tool_call_streaming
|
||||
from vllm.utils.collection_utils import as_list
|
||||
from vllm.utils.mistral import is_mistral_tokenizer, is_mistral_tool_parser
|
||||
|
||||
@@ -383,45 +380,6 @@ class OpenAIServingChat(OpenAIServing):
|
||||
return self.response_role
|
||||
return request.messages[-1]["role"]
|
||||
|
||||
@staticmethod
|
||||
def _bracket_level(s: str, opening="{", closing="}") -> int:
|
||||
"""
|
||||
Calculate the current level of nested brackets in a given string.
|
||||
"""
|
||||
level = 0
|
||||
for char in s:
|
||||
if char == opening:
|
||||
level += 1
|
||||
elif char == closing:
|
||||
level -= 1
|
||||
return level
|
||||
|
||||
@staticmethod
|
||||
def _filter_delta_text(delta_text: str, previous_text: str) -> tuple[str, bool]:
|
||||
# remove last '},' of the tool definition stemming from the
|
||||
# "name"/"parameters" outer object or closing ']' of the tool list
|
||||
# count occurrences of opening and closing curly braces and
|
||||
# once level 0 is reached stop outputting text
|
||||
# if 0 is reached while parsing the delta_text we know the current
|
||||
# tool will finish in this current iteration
|
||||
bracket_level = OpenAIServingChat._bracket_level(previous_text)
|
||||
updated_delta, passed_zero = "", False
|
||||
for c in delta_text:
|
||||
if c == "{":
|
||||
bracket_level += 1
|
||||
passed_zero = bracket_level == 0
|
||||
elif c == "}":
|
||||
bracket_level -= 1
|
||||
passed_zero = bracket_level == 0
|
||||
|
||||
if bracket_level != 0:
|
||||
updated_delta += c
|
||||
else:
|
||||
# if a comma is reached at level 0 we can stop
|
||||
if c == ",":
|
||||
break
|
||||
return updated_delta, passed_zero
|
||||
|
||||
def extract_tool_call_required_streaming(
|
||||
self,
|
||||
previous_text: str,
|
||||
@@ -430,97 +388,14 @@ class OpenAIServingChat(OpenAIServing):
|
||||
function_name_returned: bool,
|
||||
tool_call_idx: int | None = None,
|
||||
) -> tuple[DeltaMessage | None, bool]:
|
||||
if current_text is None or current_text == "":
|
||||
# if the current text is empty, we cannot parse it
|
||||
return None, function_name_returned
|
||||
try:
|
||||
flags = Allow.ALL
|
||||
obj, _ = partial_json_loads(current_text, flags)
|
||||
except (
|
||||
partial_json_parser.core.exceptions.MalformedJSON,
|
||||
json.JSONDecodeError,
|
||||
):
|
||||
logger.debug("not enough tokens to parse into JSON yet")
|
||||
obj = None
|
||||
|
||||
# check if the current text is a valid array
|
||||
# containing a partial tool calling object
|
||||
# if not repeat
|
||||
if obj is None or not isinstance(obj, list) or not len(obj) > 0:
|
||||
function_name_returned = False
|
||||
delta_message = None
|
||||
else:
|
||||
_, finishes_previous_tool = OpenAIServingChat._filter_delta_text(
|
||||
delta_text, previous_text
|
||||
)
|
||||
# take the last tool call from the generated list
|
||||
current_tool_call = obj[-1]
|
||||
|
||||
# once parameters have been generated the name is complete as well
|
||||
if not finishes_previous_tool and (
|
||||
"name" not in current_tool_call or "parameters" not in current_tool_call
|
||||
):
|
||||
function_name_returned = False
|
||||
delta_message = None
|
||||
else:
|
||||
if not function_name_returned:
|
||||
# get partly generated arguments from the latest tool call
|
||||
param_match = re.search(
|
||||
r'.*"parameters":\s*(.*)', current_text, re.DOTALL
|
||||
)
|
||||
arguments = param_match.group(1) if param_match else ""
|
||||
arguments, _ = OpenAIServingChat._filter_delta_text(
|
||||
arguments, previous_text
|
||||
)
|
||||
|
||||
# if this iteration finishes a previous tool call but a
|
||||
# new incomplete tool is already generated, take the
|
||||
# previous from the list
|
||||
if finishes_previous_tool and "parameters" not in current_tool_call:
|
||||
current_tool_call = obj[-2]
|
||||
|
||||
function_name_returned = True
|
||||
tool_call_id = make_tool_call_id(
|
||||
id_type=self.tool_call_id_type,
|
||||
func_name=current_tool_call["name"],
|
||||
idx=tool_call_idx,
|
||||
)
|
||||
delta_message = DeltaMessage(
|
||||
tool_calls=[
|
||||
DeltaToolCall(
|
||||
id=tool_call_id,
|
||||
function=DeltaFunctionCall(
|
||||
name=current_tool_call["name"], arguments=arguments
|
||||
),
|
||||
index=len(obj) - 1,
|
||||
type="function",
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
else:
|
||||
delta_text, _ = OpenAIServingChat._filter_delta_text(
|
||||
delta_text, previous_text
|
||||
)
|
||||
|
||||
if delta_text != "":
|
||||
delta_message = DeltaMessage(
|
||||
tool_calls=[
|
||||
DeltaToolCall(
|
||||
function=DeltaFunctionCall(
|
||||
# OpenAI API returns None
|
||||
# instead of name every time
|
||||
name=None,
|
||||
arguments=delta_text,
|
||||
),
|
||||
index=len(obj) - 1,
|
||||
)
|
||||
]
|
||||
)
|
||||
else:
|
||||
delta_message = None
|
||||
|
||||
return delta_message, function_name_returned
|
||||
return extract_required_tool_call_streaming(
|
||||
previous_text=previous_text,
|
||||
current_text=current_text,
|
||||
delta_text=delta_text,
|
||||
function_name_returned=function_name_returned,
|
||||
tool_call_idx=tool_call_idx,
|
||||
tool_call_id_type=self.tool_call_id_type,
|
||||
)
|
||||
|
||||
async def chat_completion_stream_generator(
|
||||
self,
|
||||
|
||||
@@ -3,7 +3,6 @@
|
||||
|
||||
import asyncio
|
||||
import time
|
||||
import uuid
|
||||
from collections import deque
|
||||
from collections.abc import AsyncGenerator, AsyncIterator, Callable, Mapping, Sequence
|
||||
from contextlib import AsyncExitStack
|
||||
@@ -13,29 +12,14 @@ from typing import Any, Final
|
||||
|
||||
from fastapi import Request
|
||||
from openai.types.responses import (
|
||||
ResponseContentPartAddedEvent,
|
||||
ResponseContentPartDoneEvent,
|
||||
ResponseFunctionCallArgumentsDeltaEvent,
|
||||
ResponseFunctionCallArgumentsDoneEvent,
|
||||
ResponseFunctionToolCall,
|
||||
ResponseFunctionToolCallItem,
|
||||
ResponseOutputItem,
|
||||
ResponseOutputItemAddedEvent,
|
||||
ResponseOutputItemDoneEvent,
|
||||
ResponseOutputMessage,
|
||||
ResponseOutputText,
|
||||
ResponseReasoningItem,
|
||||
ResponseReasoningTextDeltaEvent,
|
||||
ResponseReasoningTextDoneEvent,
|
||||
ResponseStatus,
|
||||
ResponseTextDeltaEvent,
|
||||
ResponseTextDoneEvent,
|
||||
response_text_delta_event,
|
||||
)
|
||||
from openai.types.responses.response_output_text import Logprob, LogprobTopLogprob
|
||||
from openai.types.responses.response_reasoning_item import (
|
||||
Content as ResponseReasoningTextContent,
|
||||
)
|
||||
from openai.types.responses.tool import Mcp, Tool
|
||||
from openai_harmony import Message as OpenAIHarmonyMessage
|
||||
from pydantic import TypeAdapter
|
||||
@@ -89,15 +73,15 @@ from vllm.entrypoints.openai.responses.protocol import (
|
||||
ResponseInProgressEvent,
|
||||
ResponseInputOutputItem,
|
||||
ResponseInputOutputMessage,
|
||||
ResponseReasoningPartAddedEvent,
|
||||
ResponseReasoningPartDoneEvent,
|
||||
ResponsesRequest,
|
||||
ResponsesResponse,
|
||||
ResponseUsage,
|
||||
StreamingResponsesResponse,
|
||||
)
|
||||
from vllm.entrypoints.openai.responses.streaming_events import (
|
||||
SimpleStreamingEventProcessor,
|
||||
StreamingState,
|
||||
_StateType,
|
||||
emit_content_delta_events,
|
||||
emit_previous_item_done_events,
|
||||
emit_tool_action_events,
|
||||
@@ -1353,571 +1337,59 @@ class OpenAIServingResponses(OpenAIServing):
|
||||
[StreamingResponsesResponse], StreamingResponsesResponse
|
||||
],
|
||||
) -> AsyncGenerator[StreamingResponsesResponse, None]:
|
||||
current_content_index = 0
|
||||
current_output_index = 0
|
||||
current_item_id = ""
|
||||
current_tool_call_index: int | None = None
|
||||
processor = SimpleStreamingEventProcessor()
|
||||
parser = self.parser(tokenizer, request.tools) if self.parser else None
|
||||
first_delta_sent = False
|
||||
previous_delta_messages: list[DeltaMessage] = []
|
||||
|
||||
def _get_logprobs(
|
||||
output: CompletionOutput,
|
||||
) -> list[response_text_delta_event.Logprob]:
|
||||
if not request.is_include_output_logprobs():
|
||||
return []
|
||||
return self._create_stream_response_logprobs(
|
||||
token_ids=output.token_ids,
|
||||
logprobs=output.logprobs,
|
||||
tokenizer=tokenizer,
|
||||
top_logprobs=request.top_logprobs,
|
||||
)
|
||||
|
||||
async for ctx in result_generator:
|
||||
assert isinstance(ctx, SimpleContext)
|
||||
if ctx.last_output is None:
|
||||
if ctx.last_output is None or not ctx.last_output.outputs:
|
||||
continue
|
||||
if ctx.last_output.outputs:
|
||||
output = ctx.last_output.outputs[0]
|
||||
# finish_reason='error' indicates a retryable error
|
||||
self._raise_if_error(output.finish_reason, request.request_id)
|
||||
delta_text = output.text
|
||||
delta_token_ids = as_list(output.token_ids)
|
||||
|
||||
if parser:
|
||||
delta_message = parser.parse_delta(
|
||||
delta_text=delta_text,
|
||||
delta_token_ids=delta_token_ids,
|
||||
request=request,
|
||||
prompt_token_ids=ctx.last_output.prompt_token_ids,
|
||||
)
|
||||
else:
|
||||
delta_message = DeltaMessage(
|
||||
content=output.text,
|
||||
)
|
||||
if not delta_message:
|
||||
continue
|
||||
tool_call_item_started = False
|
||||
if not first_delta_sent:
|
||||
current_item_id = random_uuid()
|
||||
if delta_message.tool_calls:
|
||||
current_tool_call_id = f"call_{random_uuid()}"
|
||||
assert len(delta_message.tool_calls) == 1, (
|
||||
"Multiple tool calls in one delta is not supported"
|
||||
)
|
||||
assert delta_message.tool_calls[0].function is not None, (
|
||||
"Tool call without function is not supported"
|
||||
)
|
||||
assert delta_message.tool_calls[0].function.name is not None, (
|
||||
"Tool call without function name is not supported"
|
||||
)
|
||||
current_tool_call_name = delta_message.tool_calls[
|
||||
0
|
||||
].function.name
|
||||
current_tool_call_index = delta_message.tool_calls[0].index
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseOutputItemAddedEvent(
|
||||
type="response.output_item.added",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item=ResponseFunctionToolCallItem(
|
||||
type="function_call",
|
||||
id=current_item_id,
|
||||
call_id=current_tool_call_id,
|
||||
name=current_tool_call_name,
|
||||
arguments="",
|
||||
status="in_progress",
|
||||
),
|
||||
)
|
||||
)
|
||||
tool_call_item_started = True
|
||||
elif delta_message.reasoning:
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseOutputItemAddedEvent(
|
||||
type="response.output_item.added",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item=ResponseReasoningItem(
|
||||
type="reasoning",
|
||||
id=current_item_id,
|
||||
summary=[],
|
||||
status="in_progress",
|
||||
),
|
||||
)
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseReasoningPartAddedEvent(
|
||||
type="response.reasoning_part.added",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item_id=current_item_id,
|
||||
content_index=current_content_index,
|
||||
part=ResponseReasoningTextContent(
|
||||
text="",
|
||||
type="reasoning_text",
|
||||
),
|
||||
)
|
||||
)
|
||||
elif not delta_message.tool_calls:
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseOutputItemAddedEvent(
|
||||
type="response.output_item.added",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item=ResponseOutputMessage(
|
||||
id=current_item_id,
|
||||
type="message",
|
||||
role="assistant",
|
||||
content=[],
|
||||
status="in_progress",
|
||||
),
|
||||
)
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseContentPartAddedEvent(
|
||||
type="response.content_part.added",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item_id=current_item_id,
|
||||
content_index=current_content_index,
|
||||
part=ResponseOutputText(
|
||||
type="output_text",
|
||||
text="",
|
||||
annotations=[],
|
||||
logprobs=[],
|
||||
),
|
||||
)
|
||||
)
|
||||
first_delta_sent = True
|
||||
output = ctx.last_output.outputs[0]
|
||||
self._raise_if_error(output.finish_reason, request.request_id)
|
||||
delta_text = output.text
|
||||
delta_token_ids = as_list(output.token_ids)
|
||||
|
||||
# check delta message and previous delta message are
|
||||
# same as content or reasoning content
|
||||
if (
|
||||
previous_delta_messages
|
||||
and previous_delta_messages[-1].reasoning is not None
|
||||
and delta_message.content is not None
|
||||
):
|
||||
# from reasoning to normal content, send done
|
||||
# event for reasoning
|
||||
reason_content = "".join(
|
||||
pm.reasoning
|
||||
for pm in previous_delta_messages
|
||||
if pm.reasoning is not None
|
||||
)
|
||||
if parser:
|
||||
delta_message = parser.parse_delta(
|
||||
delta_text=delta_text,
|
||||
delta_token_ids=delta_token_ids,
|
||||
request=request,
|
||||
prompt_token_ids=ctx.last_output.prompt_token_ids,
|
||||
)
|
||||
else:
|
||||
delta_message = DeltaMessage(content=output.text)
|
||||
|
||||
# delta message could have both reasoning and
|
||||
# content. Include current delta's reasoning in the
|
||||
# finalization since it may carry the tail end of
|
||||
# reasoning text (e.g. when reasoning end and
|
||||
# content start arrive in the same delta).
|
||||
if delta_message.reasoning is not None:
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseReasoningTextDeltaEvent(
|
||||
type="response.reasoning_text.delta",
|
||||
sequence_number=-1,
|
||||
content_index=current_content_index,
|
||||
output_index=current_output_index,
|
||||
item_id=current_item_id,
|
||||
delta=delta_message.reasoning,
|
||||
)
|
||||
)
|
||||
reason_content += delta_message.reasoning
|
||||
delta_message = DeltaMessage(content=delta_message.content)
|
||||
if not delta_message:
|
||||
continue
|
||||
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseReasoningTextDoneEvent(
|
||||
type="response.reasoning_text.done",
|
||||
item_id=current_item_id,
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
content_index=current_content_index,
|
||||
text=reason_content,
|
||||
)
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseReasoningPartDoneEvent(
|
||||
type="response.reasoning_part.done",
|
||||
sequence_number=-1,
|
||||
item_id=current_item_id,
|
||||
output_index=current_output_index,
|
||||
content_index=current_content_index,
|
||||
part=ResponseReasoningTextContent(
|
||||
text=reason_content,
|
||||
type="reasoning_text",
|
||||
),
|
||||
)
|
||||
)
|
||||
current_content_index = 0
|
||||
reasoning_item = ResponseReasoningItem(
|
||||
type="reasoning",
|
||||
content=[
|
||||
ResponseReasoningTextContent(
|
||||
text=reason_content,
|
||||
type="reasoning_text",
|
||||
),
|
||||
],
|
||||
status="completed",
|
||||
id=current_item_id,
|
||||
summary=[],
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseOutputItemDoneEvent(
|
||||
type="response.output_item.done",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item=reasoning_item,
|
||||
)
|
||||
)
|
||||
current_output_index += 1
|
||||
current_item_id = str(uuid.uuid4())
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseOutputItemAddedEvent(
|
||||
type="response.output_item.added",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item=ResponseOutputMessage(
|
||||
id=current_item_id,
|
||||
type="message",
|
||||
role="assistant",
|
||||
content=[],
|
||||
status="in_progress",
|
||||
),
|
||||
)
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseContentPartAddedEvent(
|
||||
type="response.content_part.added",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item_id=current_item_id,
|
||||
content_index=current_content_index,
|
||||
part=ResponseOutputText(
|
||||
type="output_text",
|
||||
text="",
|
||||
annotations=[],
|
||||
logprobs=[],
|
||||
),
|
||||
)
|
||||
)
|
||||
# reset previous delta messages
|
||||
previous_delta_messages = []
|
||||
if delta_message.tool_calls and delta_message.tool_calls[0].function:
|
||||
tool_call = delta_message.tool_calls[0]
|
||||
tool_call_function = tool_call.function
|
||||
if (
|
||||
current_tool_call_index is not None
|
||||
and tool_call.index is not None
|
||||
and tool_call.index != current_tool_call_index
|
||||
and tool_call_function is not None
|
||||
and tool_call_function.name is not None
|
||||
):
|
||||
# From one tool call to another, finalize the previous
|
||||
# function-call item before opening the next one.
|
||||
parts = []
|
||||
for pm in previous_delta_messages:
|
||||
if pm.tool_calls:
|
||||
previous_tool_call = pm.tool_calls[0]
|
||||
if previous_tool_call.function is not None:
|
||||
parts.append(
|
||||
previous_tool_call.function.arguments or ""
|
||||
)
|
||||
target_state, tool_call = processor.resolve_target_state(delta_message)
|
||||
if target_state == _StateType.NONE:
|
||||
continue
|
||||
|
||||
tool_call_arguments = "".join(parts)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseFunctionCallArgumentsDoneEvent(
|
||||
type="response.function_call_arguments.done",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item_id=current_item_id,
|
||||
arguments=tool_call_arguments,
|
||||
name=current_tool_call_name,
|
||||
)
|
||||
)
|
||||
function_call_item = ResponseFunctionToolCall(
|
||||
type="function_call",
|
||||
name=current_tool_call_name,
|
||||
arguments=tool_call_arguments,
|
||||
status="completed",
|
||||
id=current_item_id,
|
||||
call_id=current_tool_call_id,
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseOutputItemDoneEvent(
|
||||
type="response.output_item.done",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item=function_call_item,
|
||||
)
|
||||
)
|
||||
# Reset previous delta messages so the next tool call
|
||||
# does not reuse arguments from the completed item.
|
||||
previous_delta_messages = []
|
||||
current_output_index += 1
|
||||
current_item_id = random_uuid()
|
||||
current_tool_call_name = tool_call_function.name
|
||||
current_tool_call_id = f"call_{random_uuid()}"
|
||||
current_tool_call_index = tool_call.index
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseOutputItemAddedEvent(
|
||||
type="response.output_item.added",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item=ResponseFunctionToolCallItem(
|
||||
type="function_call",
|
||||
id=current_item_id,
|
||||
call_id=current_tool_call_id,
|
||||
name=current_tool_call_name,
|
||||
arguments="",
|
||||
status="in_progress",
|
||||
),
|
||||
)
|
||||
)
|
||||
current_content_index = 0
|
||||
tool_call_item_started = True
|
||||
if processor.needs_transition(target_state, tool_call):
|
||||
for event in processor.close_current():
|
||||
yield _increment_sequence_number_and_return(event)
|
||||
for event in processor.open(target_state, tool_call):
|
||||
yield _increment_sequence_number_and_return(event)
|
||||
|
||||
if delta_message.tool_calls[0].function.arguments:
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseFunctionCallArgumentsDeltaEvent(
|
||||
type="response.function_call_arguments.delta",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item_id=current_item_id,
|
||||
delta=delta_message.tool_calls[0].function.arguments,
|
||||
)
|
||||
)
|
||||
# tool call initiated with no arguments
|
||||
elif (
|
||||
delta_message.tool_calls[0].function.name
|
||||
and not tool_call_item_started
|
||||
):
|
||||
# send done with current content part
|
||||
# and add new function call item
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseTextDoneEvent(
|
||||
type="response.output_text.done",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
content_index=current_content_index,
|
||||
text="",
|
||||
logprobs=[],
|
||||
item_id=current_item_id,
|
||||
)
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseContentPartDoneEvent(
|
||||
type="response.content_part.done",
|
||||
sequence_number=-1,
|
||||
item_id=current_item_id,
|
||||
output_index=current_output_index,
|
||||
content_index=current_content_index,
|
||||
part=ResponseOutputText(
|
||||
type="output_text",
|
||||
text="",
|
||||
annotations=[],
|
||||
logprobs=[],
|
||||
),
|
||||
)
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseOutputItemDoneEvent(
|
||||
type="response.output_item.done",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item=ResponseOutputMessage(
|
||||
id=current_item_id,
|
||||
type="message",
|
||||
role="assistant",
|
||||
content=[],
|
||||
status="completed",
|
||||
),
|
||||
)
|
||||
)
|
||||
current_output_index += 1
|
||||
current_item_id = random_uuid()
|
||||
current_tool_call_name = delta_message.tool_calls[
|
||||
0
|
||||
].function.name
|
||||
current_tool_call_id = f"call_{random_uuid()}"
|
||||
current_tool_call_index = delta_message.tool_calls[0].index
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseOutputItemAddedEvent(
|
||||
type="response.output_item.added",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item=ResponseFunctionToolCallItem(
|
||||
type="function_call",
|
||||
id=current_item_id,
|
||||
call_id=current_tool_call_id,
|
||||
name=current_tool_call_name,
|
||||
arguments="",
|
||||
status="in_progress",
|
||||
),
|
||||
)
|
||||
)
|
||||
# skip content part for tool call
|
||||
current_content_index = 1
|
||||
continue
|
||||
elif delta_message.reasoning is not None:
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseReasoningTextDeltaEvent(
|
||||
type="response.reasoning_text.delta",
|
||||
sequence_number=-1,
|
||||
content_index=current_content_index,
|
||||
output_index=current_output_index,
|
||||
item_id=current_item_id,
|
||||
delta=delta_message.reasoning,
|
||||
)
|
||||
)
|
||||
elif delta_message.content:
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseTextDeltaEvent(
|
||||
type="response.output_text.delta",
|
||||
sequence_number=-1,
|
||||
content_index=current_content_index,
|
||||
output_index=current_output_index,
|
||||
item_id=current_item_id,
|
||||
delta=delta_message.content,
|
||||
logprobs=(
|
||||
self._create_stream_response_logprobs(
|
||||
token_ids=output.token_ids,
|
||||
logprobs=output.logprobs,
|
||||
tokenizer=tokenizer,
|
||||
top_logprobs=request.top_logprobs,
|
||||
)
|
||||
if request.is_include_output_logprobs()
|
||||
else []
|
||||
),
|
||||
)
|
||||
)
|
||||
for event in processor.emit_delta(delta_message, output, _get_logprobs):
|
||||
yield _increment_sequence_number_and_return(event)
|
||||
|
||||
previous_delta_messages.append(delta_message)
|
||||
|
||||
if previous_delta_messages:
|
||||
parts = []
|
||||
for pm in previous_delta_messages:
|
||||
if pm.tool_calls:
|
||||
assert len(pm.tool_calls) == 1, (
|
||||
"Multiple tool calls in one delta is not supported"
|
||||
)
|
||||
assert pm.tool_calls[0].function is not None, (
|
||||
"Tool call without function is not supported"
|
||||
)
|
||||
parts.append(pm.tool_calls[0].function.arguments or "")
|
||||
|
||||
tool_call_arguments = "".join(parts)
|
||||
if tool_call_arguments:
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseFunctionCallArgumentsDoneEvent(
|
||||
type="response.function_call_arguments.done",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item_id=current_item_id,
|
||||
arguments=tool_call_arguments,
|
||||
name=current_tool_call_name,
|
||||
)
|
||||
)
|
||||
current_content_index = 0
|
||||
function_call_item = ResponseFunctionToolCall(
|
||||
type="function_call",
|
||||
name=current_tool_call_name,
|
||||
arguments=tool_call_arguments,
|
||||
status="completed",
|
||||
id=current_item_id,
|
||||
call_id=current_tool_call_id,
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseOutputItemDoneEvent(
|
||||
type="response.output_item.done",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item=function_call_item,
|
||||
)
|
||||
)
|
||||
|
||||
elif previous_delta_messages[-1].reasoning is not None:
|
||||
reason_content = "".join(
|
||||
pm.reasoning
|
||||
for pm in previous_delta_messages
|
||||
if pm.reasoning is not None
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseReasoningTextDoneEvent(
|
||||
type="response.reasoning_text.done",
|
||||
item_id=current_item_id,
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
content_index=current_content_index,
|
||||
text=reason_content,
|
||||
)
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseReasoningPartDoneEvent(
|
||||
type="response.reasoning_part.done",
|
||||
sequence_number=-1,
|
||||
item_id=current_item_id,
|
||||
output_index=current_output_index,
|
||||
content_index=current_content_index,
|
||||
part=ResponseReasoningTextContent(
|
||||
text=reason_content,
|
||||
type="reasoning_text",
|
||||
),
|
||||
)
|
||||
)
|
||||
reasoning_item = ResponseReasoningItem(
|
||||
type="reasoning",
|
||||
content=[
|
||||
ResponseReasoningTextContent(
|
||||
text=reason_content,
|
||||
type="reasoning_text",
|
||||
),
|
||||
],
|
||||
status="completed",
|
||||
id=current_item_id,
|
||||
summary=[],
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseOutputItemDoneEvent(
|
||||
type="response.output_item.done",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item=reasoning_item,
|
||||
)
|
||||
)
|
||||
elif previous_delta_messages[-1].content:
|
||||
final_content = "".join(
|
||||
pm.content for pm in previous_delta_messages if pm.content
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseTextDoneEvent(
|
||||
type="response.output_text.done",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
content_index=current_content_index,
|
||||
text=final_content,
|
||||
logprobs=[],
|
||||
item_id=current_item_id,
|
||||
)
|
||||
)
|
||||
part = ResponseOutputText(
|
||||
text=final_content,
|
||||
type="output_text",
|
||||
annotations=[],
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseContentPartDoneEvent(
|
||||
type="response.content_part.done",
|
||||
sequence_number=-1,
|
||||
item_id=current_item_id,
|
||||
output_index=current_output_index,
|
||||
content_index=current_content_index,
|
||||
part=part,
|
||||
)
|
||||
)
|
||||
item = ResponseOutputMessage(
|
||||
type="message",
|
||||
role="assistant",
|
||||
content=[
|
||||
part,
|
||||
],
|
||||
status="completed",
|
||||
id=current_item_id,
|
||||
summary=[],
|
||||
)
|
||||
yield _increment_sequence_number_and_return(
|
||||
ResponseOutputItemDoneEvent(
|
||||
type="response.output_item.done",
|
||||
sequence_number=-1,
|
||||
output_index=current_output_index,
|
||||
item=item,
|
||||
)
|
||||
)
|
||||
for event in processor.close_current():
|
||||
yield _increment_sequence_number_and_return(event)
|
||||
|
||||
async def _process_harmony_streaming_events(
|
||||
self,
|
||||
|
||||
@@ -16,8 +16,10 @@ The file is organized as:
|
||||
"""
|
||||
|
||||
import json
|
||||
from dataclasses import dataclass
|
||||
from typing import Final
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass, field
|
||||
from enum import Enum, auto
|
||||
from typing import Any, ClassVar, Final, NamedTuple
|
||||
|
||||
from openai.types.responses import (
|
||||
ResponseCodeInterpreterCallCodeDeltaEvent,
|
||||
@@ -31,6 +33,7 @@ from openai.types.responses import (
|
||||
ResponseFunctionCallArgumentsDeltaEvent,
|
||||
ResponseFunctionCallArgumentsDoneEvent,
|
||||
ResponseFunctionToolCall,
|
||||
ResponseFunctionToolCallItem,
|
||||
ResponseFunctionWebSearch,
|
||||
ResponseMcpCallArgumentsDeltaEvent,
|
||||
ResponseMcpCallArgumentsDoneEvent,
|
||||
@@ -49,6 +52,7 @@ from openai.types.responses import (
|
||||
ResponseWebSearchCallInProgressEvent,
|
||||
ResponseWebSearchCallSearchingEvent,
|
||||
response_function_web_search,
|
||||
response_text_delta_event,
|
||||
)
|
||||
from openai.types.responses.response_output_item import McpCall
|
||||
from openai.types.responses.response_reasoning_item import (
|
||||
@@ -57,12 +61,14 @@ from openai.types.responses.response_reasoning_item import (
|
||||
from openai_harmony import Message as HarmonyMessage
|
||||
|
||||
from vllm.entrypoints.mcp.tool_server import ToolServer
|
||||
from vllm.entrypoints.openai.engine.protocol import DeltaMessage
|
||||
from vllm.entrypoints.openai.responses.context import StreamingHarmonyContext
|
||||
from vllm.entrypoints.openai.responses.protocol import (
|
||||
ResponseReasoningPartAddedEvent,
|
||||
ResponseReasoningPartDoneEvent,
|
||||
StreamingResponsesResponse,
|
||||
)
|
||||
from vllm.outputs import CompletionOutput
|
||||
from vllm.utils import random_uuid
|
||||
|
||||
TOOL_NAME_TO_MCP_SERVER_LABEL: Final[dict[str, str]] = {
|
||||
@@ -796,3 +802,457 @@ def emit_tool_action_events(
|
||||
)
|
||||
|
||||
return events
|
||||
|
||||
|
||||
# =====================================================================
|
||||
# Simple streaming helpers
|
||||
# =====================================================================
|
||||
|
||||
|
||||
class _StateType(Enum):
|
||||
NONE = auto()
|
||||
CONTENT = auto()
|
||||
REASONING = auto()
|
||||
TOOL_CALL = auto()
|
||||
|
||||
|
||||
@dataclass
|
||||
class SimpleStreamingState:
|
||||
output_index: int = 0
|
||||
current_item_id: str = ""
|
||||
content_index: int = 0
|
||||
accumulated_text: str = ""
|
||||
tool_call_id: str = ""
|
||||
tool_call_name: str = ""
|
||||
tool_call_index: int | None = None
|
||||
has_emitted_tool_call_delta: bool = False
|
||||
current_state: _StateType = field(default_factory=lambda: _StateType.NONE)
|
||||
|
||||
|
||||
def emit_simple_content_open(
|
||||
state: SimpleStreamingState,
|
||||
) -> list[StreamingResponsesResponse]:
|
||||
state.current_state = _StateType.CONTENT
|
||||
state.current_item_id = random_uuid()
|
||||
state.content_index = 0
|
||||
state.accumulated_text = ""
|
||||
return [
|
||||
ResponseOutputItemAddedEvent(
|
||||
type="response.output_item.added",
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
item=ResponseOutputMessage(
|
||||
id=state.current_item_id,
|
||||
type="message",
|
||||
role="assistant",
|
||||
content=[],
|
||||
status="in_progress",
|
||||
),
|
||||
),
|
||||
ResponseContentPartAddedEvent(
|
||||
type="response.content_part.added",
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
item_id=state.current_item_id,
|
||||
content_index=state.content_index,
|
||||
part=ResponseOutputText(
|
||||
type="output_text",
|
||||
text="",
|
||||
annotations=[],
|
||||
logprobs=[],
|
||||
),
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def emit_simple_content_delta(
|
||||
state: SimpleStreamingState,
|
||||
delta: str,
|
||||
logprobs: list[response_text_delta_event.Logprob] | None = None,
|
||||
) -> list[StreamingResponsesResponse]:
|
||||
state.accumulated_text += delta
|
||||
return [
|
||||
ResponseTextDeltaEvent(
|
||||
type="response.output_text.delta",
|
||||
sequence_number=-1,
|
||||
content_index=state.content_index,
|
||||
output_index=state.output_index,
|
||||
item_id=state.current_item_id,
|
||||
delta=delta,
|
||||
logprobs=logprobs or [],
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def emit_simple_content_done(
|
||||
state: SimpleStreamingState,
|
||||
) -> list[StreamingResponsesResponse]:
|
||||
part = ResponseOutputText(
|
||||
type="output_text",
|
||||
text=state.accumulated_text,
|
||||
annotations=[],
|
||||
)
|
||||
events: list[StreamingResponsesResponse] = [
|
||||
ResponseTextDoneEvent(
|
||||
type="response.output_text.done",
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
content_index=state.content_index,
|
||||
text=state.accumulated_text,
|
||||
logprobs=[],
|
||||
item_id=state.current_item_id,
|
||||
),
|
||||
ResponseContentPartDoneEvent(
|
||||
type="response.content_part.done",
|
||||
sequence_number=-1,
|
||||
item_id=state.current_item_id,
|
||||
output_index=state.output_index,
|
||||
content_index=state.content_index,
|
||||
part=part,
|
||||
),
|
||||
ResponseOutputItemDoneEvent(
|
||||
type="response.output_item.done",
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
item=ResponseOutputMessage(
|
||||
id=state.current_item_id,
|
||||
type="message",
|
||||
role="assistant",
|
||||
content=[part] if state.accumulated_text else [],
|
||||
status="completed",
|
||||
summary=[],
|
||||
),
|
||||
),
|
||||
]
|
||||
state.output_index += 1
|
||||
state.current_state = _StateType.NONE
|
||||
return events
|
||||
|
||||
|
||||
def emit_simple_reasoning_open(
|
||||
state: SimpleStreamingState,
|
||||
) -> list[StreamingResponsesResponse]:
|
||||
state.current_state = _StateType.REASONING
|
||||
state.current_item_id = random_uuid()
|
||||
state.content_index = 0
|
||||
state.accumulated_text = ""
|
||||
return [
|
||||
ResponseOutputItemAddedEvent(
|
||||
type="response.output_item.added",
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
item=ResponseReasoningItem(
|
||||
type="reasoning",
|
||||
id=state.current_item_id,
|
||||
summary=[],
|
||||
status="in_progress",
|
||||
),
|
||||
),
|
||||
ResponseReasoningPartAddedEvent(
|
||||
type="response.reasoning_part.added",
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
item_id=state.current_item_id,
|
||||
content_index=state.content_index,
|
||||
part=ResponseReasoningTextContent(
|
||||
text="",
|
||||
type="reasoning_text",
|
||||
),
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def emit_simple_reasoning_delta(
|
||||
state: SimpleStreamingState,
|
||||
delta: str,
|
||||
) -> list[StreamingResponsesResponse]:
|
||||
state.accumulated_text += delta
|
||||
return [
|
||||
ResponseReasoningTextDeltaEvent(
|
||||
type="response.reasoning_text.delta",
|
||||
item_id=state.current_item_id,
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
content_index=state.content_index,
|
||||
delta=delta,
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def emit_simple_reasoning_done(
|
||||
state: SimpleStreamingState,
|
||||
) -> list[StreamingResponsesResponse]:
|
||||
part = ResponseReasoningTextContent(
|
||||
text=state.accumulated_text,
|
||||
type="reasoning_text",
|
||||
)
|
||||
events: list[StreamingResponsesResponse] = [
|
||||
ResponseReasoningTextDoneEvent(
|
||||
type="response.reasoning_text.done",
|
||||
item_id=state.current_item_id,
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
content_index=state.content_index,
|
||||
text=state.accumulated_text,
|
||||
),
|
||||
ResponseReasoningPartDoneEvent(
|
||||
type="response.reasoning_part.done",
|
||||
sequence_number=-1,
|
||||
item_id=state.current_item_id,
|
||||
output_index=state.output_index,
|
||||
content_index=state.content_index,
|
||||
part=part,
|
||||
),
|
||||
ResponseOutputItemDoneEvent(
|
||||
type="response.output_item.done",
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
item=ResponseReasoningItem(
|
||||
type="reasoning",
|
||||
content=[part],
|
||||
status="completed",
|
||||
id=state.current_item_id,
|
||||
summary=[],
|
||||
),
|
||||
),
|
||||
]
|
||||
state.output_index += 1
|
||||
state.current_state = _StateType.NONE
|
||||
return events
|
||||
|
||||
|
||||
def emit_simple_tool_call_open(
|
||||
state: SimpleStreamingState,
|
||||
name: str,
|
||||
index: int | None,
|
||||
) -> list[StreamingResponsesResponse]:
|
||||
state.current_state = _StateType.TOOL_CALL
|
||||
state.current_item_id = random_uuid()
|
||||
state.tool_call_id = f"call_{random_uuid()}"
|
||||
state.tool_call_name = name
|
||||
state.tool_call_index = index
|
||||
state.accumulated_text = ""
|
||||
state.has_emitted_tool_call_delta = False
|
||||
return [
|
||||
ResponseOutputItemAddedEvent(
|
||||
type="response.output_item.added",
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
item=ResponseFunctionToolCallItem(
|
||||
type="function_call",
|
||||
id=state.current_item_id,
|
||||
call_id=state.tool_call_id,
|
||||
name=name,
|
||||
arguments="",
|
||||
status="in_progress",
|
||||
),
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
def emit_simple_tool_call_delta(
|
||||
state: SimpleStreamingState,
|
||||
delta: str,
|
||||
) -> list[StreamingResponsesResponse]:
|
||||
state.accumulated_text += delta
|
||||
state.has_emitted_tool_call_delta = True
|
||||
return [
|
||||
ResponseFunctionCallArgumentsDeltaEvent(
|
||||
type="response.function_call_arguments.delta",
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
item_id=state.current_item_id,
|
||||
delta=delta,
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def emit_simple_tool_call_done(
|
||||
state: SimpleStreamingState,
|
||||
) -> list[StreamingResponsesResponse]:
|
||||
events: list[StreamingResponsesResponse] = []
|
||||
if state.has_emitted_tool_call_delta:
|
||||
events.append(
|
||||
ResponseFunctionCallArgumentsDoneEvent(
|
||||
type="response.function_call_arguments.done",
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
item_id=state.current_item_id,
|
||||
arguments=state.accumulated_text,
|
||||
name=state.tool_call_name,
|
||||
)
|
||||
)
|
||||
events.append(
|
||||
ResponseOutputItemDoneEvent(
|
||||
type="response.output_item.done",
|
||||
sequence_number=-1,
|
||||
output_index=state.output_index,
|
||||
item=ResponseFunctionToolCall(
|
||||
type="function_call",
|
||||
name=state.tool_call_name,
|
||||
arguments=state.accumulated_text,
|
||||
status="completed",
|
||||
id=state.current_item_id,
|
||||
call_id=state.tool_call_id,
|
||||
),
|
||||
),
|
||||
)
|
||||
state.output_index += 1
|
||||
state.current_state = _StateType.NONE
|
||||
return events
|
||||
|
||||
|
||||
class _StateHandlers(NamedTuple):
|
||||
"""Tuple for each state: open(start), delta(chunk), done(finish)."""
|
||||
|
||||
open_fn: Callable[..., list[StreamingResponsesResponse]]
|
||||
delta_fn: Callable[..., list[StreamingResponsesResponse]]
|
||||
done_fn: Callable[..., list[StreamingResponsesResponse]]
|
||||
|
||||
|
||||
class SimpleStreamingEventProcessor:
|
||||
"""
|
||||
State-machine processor for the simple (non-Harmony) streaming path.
|
||||
|
||||
Core flow:
|
||||
1. Resolve the target state from the delta_message
|
||||
(CONTENT / REASONING / TOOL_CALL).
|
||||
2. If the target state differs from the current one,
|
||||
close_current() then open() the new state.
|
||||
3. emit_delta() produces the incremental events for the state.
|
||||
|
||||
State lifecycle:
|
||||
open() -> repeated emit_delta() -> close_current()
|
||||
"""
|
||||
|
||||
_STATE_HANDLERS: ClassVar[dict[_StateType, _StateHandlers]] = {
|
||||
_StateType.CONTENT: _StateHandlers(
|
||||
emit_simple_content_open,
|
||||
emit_simple_content_delta,
|
||||
emit_simple_content_done,
|
||||
),
|
||||
_StateType.REASONING: _StateHandlers(
|
||||
emit_simple_reasoning_open,
|
||||
emit_simple_reasoning_delta,
|
||||
emit_simple_reasoning_done,
|
||||
),
|
||||
_StateType.TOOL_CALL: _StateHandlers(
|
||||
emit_simple_tool_call_open,
|
||||
emit_simple_tool_call_delta,
|
||||
emit_simple_tool_call_done,
|
||||
),
|
||||
}
|
||||
|
||||
def __init__(self, state: SimpleStreamingState | None = None) -> None:
|
||||
self.state = state or SimpleStreamingState()
|
||||
|
||||
def resolve_target_state(
|
||||
self, delta_message: DeltaMessage
|
||||
) -> tuple[_StateType, Any]:
|
||||
"""
|
||||
Decide which state the next delta belongs to.
|
||||
|
||||
Priority: TOOL_CALL > REASONING > CONTENT, fallback to NONE.
|
||||
For TOOL_CALL the first tool_call object is also returned so
|
||||
callers can detect a switch between consecutive tools.
|
||||
"""
|
||||
if (
|
||||
delta_message.tool_calls
|
||||
and delta_message.tool_calls[0].function is not None
|
||||
):
|
||||
return _StateType.TOOL_CALL, delta_message.tool_calls[0]
|
||||
if delta_message.reasoning is not None:
|
||||
return _StateType.REASONING, None
|
||||
if delta_message.content:
|
||||
return _StateType.CONTENT, None
|
||||
return _StateType.NONE, None
|
||||
|
||||
def needs_transition(self, target_state: _StateType, tool_call: Any) -> bool:
|
||||
"""
|
||||
Return True when we must close the current state and open a new one.
|
||||
|
||||
Two cases trigger a transition:
|
||||
1. The target state differs from the current state
|
||||
(e.g. CONTENT -> TOOL_CALL).
|
||||
2. We are already in TOOL_CALL but the next tool_call has a
|
||||
different index (multiple consecutive tool calls).
|
||||
"""
|
||||
if self.state.current_state != target_state:
|
||||
return True
|
||||
return (
|
||||
target_state == _StateType.TOOL_CALL
|
||||
and tool_call is not None
|
||||
and self.state.tool_call_index is not None
|
||||
and tool_call.index is not None
|
||||
and self.state.tool_call_index != tool_call.index
|
||||
)
|
||||
|
||||
def close_current(self) -> list[StreamingResponsesResponse]:
|
||||
"""Close the current state and emit its 'done' event sequence."""
|
||||
handlers = self._STATE_HANDLERS.get(self.state.current_state)
|
||||
if handlers is None:
|
||||
return []
|
||||
return handlers.done_fn(self.state)
|
||||
|
||||
def open(
|
||||
self, target_state: _StateType, tool_call: Any = None
|
||||
) -> list[StreamingResponsesResponse]:
|
||||
"""Open a new state and emit its 'added' / 'open' event sequence."""
|
||||
handlers = self._STATE_HANDLERS[target_state]
|
||||
if target_state == _StateType.TOOL_CALL:
|
||||
assert tool_call is not None
|
||||
return handlers.open_fn(
|
||||
self.state, tool_call.function.name, tool_call.index
|
||||
)
|
||||
return handlers.open_fn(self.state)
|
||||
|
||||
def emit_delta(
|
||||
self,
|
||||
delta_message: DeltaMessage,
|
||||
output: CompletionOutput,
|
||||
get_logprobs: Callable[
|
||||
[CompletionOutput], list[response_text_delta_event.Logprob]
|
||||
]
|
||||
| None = None,
|
||||
) -> list[StreamingResponsesResponse]:
|
||||
"""
|
||||
Emit incremental events for the current state from the delta.
|
||||
|
||||
Special case: when already in REASONING and the same delta also
|
||||
carries content, we emit the reasoning delta, close reasoning,
|
||||
open content, and then emit the content delta.
|
||||
"""
|
||||
handlers = self._STATE_HANDLERS[self.state.current_state]
|
||||
events: list[StreamingResponsesResponse] = []
|
||||
|
||||
# Special case: reasoning -> content inside a single delta.
|
||||
if (
|
||||
self.state.current_state == _StateType.REASONING
|
||||
and delta_message.reasoning is not None
|
||||
and delta_message.content is not None
|
||||
):
|
||||
events.extend(handlers.delta_fn(self.state, delta_message.reasoning))
|
||||
events.extend(self.close_current())
|
||||
events.extend(self.open(_StateType.CONTENT))
|
||||
content_handlers = self._STATE_HANDLERS[_StateType.CONTENT]
|
||||
logprobs = get_logprobs(output) if get_logprobs else []
|
||||
events.extend(
|
||||
content_handlers.delta_fn(self.state, delta_message.content, logprobs)
|
||||
)
|
||||
return events
|
||||
|
||||
if self.state.current_state == _StateType.TOOL_CALL:
|
||||
assert delta_message.tool_calls is not None
|
||||
tool_call_function = delta_message.tool_calls[0].function
|
||||
assert tool_call_function is not None
|
||||
if tool_call_function.arguments:
|
||||
return handlers.delta_fn(self.state, tool_call_function.arguments)
|
||||
return []
|
||||
elif self.state.current_state == _StateType.REASONING:
|
||||
assert delta_message.reasoning is not None
|
||||
return handlers.delta_fn(self.state, delta_message.reasoning)
|
||||
elif self.state.current_state == _StateType.CONTENT:
|
||||
assert delta_message.content is not None
|
||||
logprobs = get_logprobs(output) if get_logprobs else []
|
||||
return handlers.delta_fn(self.state, delta_message.content, logprobs)
|
||||
return []
|
||||
|
||||
@@ -38,6 +38,7 @@ from vllm.logger import init_logger
|
||||
from vllm.reasoning.abs_reasoning_parsers import ReasoningParser
|
||||
from vllm.tokenizers import TokenizerLike
|
||||
from vllm.tool_parsers.abstract_tool_parser import ToolParser
|
||||
from vllm.tool_parsers.streaming import extract_required_tool_call_streaming
|
||||
from vllm.tool_parsers.utils import Tool
|
||||
from vllm.utils import random_uuid
|
||||
|
||||
@@ -53,6 +54,11 @@ class StreamState:
|
||||
prompt_reasoning_checked: bool = False
|
||||
previous_text: str = ""
|
||||
previous_token_ids: list[int] = field(default_factory=list)
|
||||
history_tool_call_cnt: int = 0
|
||||
tool_call_id_type: str = "random"
|
||||
# only used for "required" and "named tool" choices,
|
||||
# tracks whether function name has been fully returned in the stream yet
|
||||
function_name_returned: bool = False
|
||||
|
||||
|
||||
class Parser:
|
||||
@@ -558,6 +564,47 @@ class DelegatingParser(Parser):
|
||||
request,
|
||||
)
|
||||
|
||||
def _extract_tool_calls_streaming(
|
||||
self,
|
||||
previous_text: str,
|
||||
current_text: str,
|
||||
delta_text: str,
|
||||
previous_token_ids: Sequence[int],
|
||||
current_token_ids: Sequence[int],
|
||||
delta_token_ids: Sequence[int],
|
||||
request: ChatCompletionRequest,
|
||||
# The following parameters are used for "required" tool choice parsing and are
|
||||
# tracked in StreamState for streaming parsing.
|
||||
tool_call_idx: int | None = None,
|
||||
tool_call_id_type: str = "random",
|
||||
function_name_returned: bool = False,
|
||||
) -> tuple[DeltaMessage | None, bool]:
|
||||
if request.tool_choice and isinstance(
|
||||
request.tool_choice, ChatCompletionNamedToolChoiceParam
|
||||
):
|
||||
return None, False
|
||||
if request.tool_choice == "required":
|
||||
delta_message, function_name_returned = (
|
||||
extract_required_tool_call_streaming(
|
||||
previous_text=previous_text,
|
||||
current_text=current_text,
|
||||
delta_text=delta_text,
|
||||
function_name_returned=function_name_returned,
|
||||
tool_call_idx=tool_call_idx,
|
||||
tool_call_id_type=tool_call_id_type,
|
||||
)
|
||||
)
|
||||
return delta_message, function_name_returned
|
||||
return self.extract_tool_calls_streaming(
|
||||
previous_text,
|
||||
current_text,
|
||||
delta_text,
|
||||
previous_token_ids,
|
||||
current_token_ids,
|
||||
delta_token_ids,
|
||||
request,
|
||||
), False
|
||||
|
||||
def is_reasoning_end(self, input_ids: list[int]) -> bool:
|
||||
if self._reasoning_parser is None:
|
||||
return False
|
||||
@@ -628,14 +675,20 @@ class DelegatingParser(Parser):
|
||||
state.previous_token_ids = []
|
||||
delta_text = current_text
|
||||
delta_token_ids = current_token_ids
|
||||
delta_message = self.extract_tool_calls_streaming(
|
||||
previous_text=state.previous_text,
|
||||
current_text=current_text,
|
||||
delta_text=delta_text,
|
||||
previous_token_ids=state.previous_token_ids,
|
||||
current_token_ids=current_token_ids,
|
||||
delta_token_ids=delta_token_ids,
|
||||
request=request, # type: ignore[arg-type]
|
||||
|
||||
delta_message, state.function_name_returned = (
|
||||
self._extract_tool_calls_streaming(
|
||||
previous_text=state.previous_text,
|
||||
current_text=current_text,
|
||||
delta_text=delta_text,
|
||||
previous_token_ids=state.previous_token_ids,
|
||||
current_token_ids=current_token_ids,
|
||||
delta_token_ids=delta_token_ids,
|
||||
request=request, # type: ignore[arg-type]
|
||||
tool_call_idx=state.history_tool_call_cnt,
|
||||
tool_call_id_type=state.tool_call_id_type,
|
||||
function_name_returned=state.function_name_returned,
|
||||
)
|
||||
)
|
||||
|
||||
# No parsers: pass through as content
|
||||
|
||||
@@ -0,0 +1,199 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
|
||||
|
||||
import json
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import partial_json_parser
|
||||
import regex as re
|
||||
from partial_json_parser.core.options import Allow
|
||||
|
||||
from vllm.entrypoints.chat_utils import make_tool_call_id
|
||||
from vllm.entrypoints.openai.engine.protocol import (
|
||||
DeltaFunctionCall,
|
||||
DeltaMessage,
|
||||
DeltaToolCall,
|
||||
)
|
||||
from vllm.tool_parsers.mistral_tool_parser import MistralToolCall
|
||||
from vllm.tool_parsers.utils import partial_json_loads
|
||||
from vllm.utils.mistral import is_mistral_tokenizer
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from vllm.tokenizers import TokenizerLike
|
||||
else:
|
||||
TokenizerLike = object
|
||||
|
||||
|
||||
def _bracket_level(s: str, opening: str = "{", closing: str = "}") -> int:
|
||||
"""Calculate the current level of nested brackets in a string."""
|
||||
level = 0
|
||||
for char in s:
|
||||
if char == opening:
|
||||
level += 1
|
||||
elif char == closing:
|
||||
level -= 1
|
||||
return level
|
||||
|
||||
|
||||
def filter_delta_text(
|
||||
delta_text: str,
|
||||
previous_text: str,
|
||||
) -> tuple[str, bool]:
|
||||
"""Trim trailing tool-list delimiters from required-tool streaming text."""
|
||||
bracket_level = _bracket_level(previous_text)
|
||||
updated_delta = ""
|
||||
passed_zero = False
|
||||
for char in delta_text:
|
||||
if char == "{":
|
||||
bracket_level += 1
|
||||
passed_zero = bracket_level == 0
|
||||
elif char == "}":
|
||||
bracket_level -= 1
|
||||
passed_zero = bracket_level == 0
|
||||
|
||||
if bracket_level != 0:
|
||||
updated_delta += char
|
||||
else:
|
||||
if char == ",":
|
||||
break
|
||||
return updated_delta, passed_zero
|
||||
|
||||
|
||||
def extract_named_tool_call_streaming(
|
||||
*,
|
||||
delta_text: str,
|
||||
function_name: str,
|
||||
function_name_returned: bool,
|
||||
tool_call_idx: int | None,
|
||||
tool_call_id_type: str,
|
||||
tokenizer: "TokenizerLike",
|
||||
tool_call_array_index: int,
|
||||
) -> tuple[DeltaMessage, bool, bool]:
|
||||
"""Build a streaming tool-call delta for forced named tool choice."""
|
||||
created_new_tool_call = False
|
||||
if function_name_returned:
|
||||
delta_tool_call = DeltaToolCall(
|
||||
function=DeltaFunctionCall(arguments=delta_text),
|
||||
index=tool_call_array_index,
|
||||
)
|
||||
else:
|
||||
if is_mistral_tokenizer(tokenizer):
|
||||
tool_call_id = MistralToolCall.generate_random_id()
|
||||
else:
|
||||
tool_call_id = make_tool_call_id(
|
||||
id_type=tool_call_id_type,
|
||||
func_name=function_name,
|
||||
idx=tool_call_idx,
|
||||
)
|
||||
delta_tool_call = DeltaToolCall(
|
||||
id=tool_call_id,
|
||||
type="function",
|
||||
function=DeltaFunctionCall(
|
||||
name=function_name,
|
||||
arguments=delta_text,
|
||||
),
|
||||
index=tool_call_array_index,
|
||||
)
|
||||
function_name_returned = True
|
||||
created_new_tool_call = True
|
||||
|
||||
return (
|
||||
DeltaMessage(tool_calls=[delta_tool_call]),
|
||||
function_name_returned,
|
||||
created_new_tool_call,
|
||||
)
|
||||
|
||||
|
||||
def extract_required_tool_call_streaming(
|
||||
*,
|
||||
previous_text: str,
|
||||
current_text: str | None,
|
||||
delta_text: str,
|
||||
function_name_returned: bool,
|
||||
tool_call_idx: int | None,
|
||||
tool_call_id_type: str,
|
||||
) -> tuple[DeltaMessage | None, bool]:
|
||||
if current_text is None or current_text == "":
|
||||
# if the current text is empty, we cannot parse it
|
||||
return None, function_name_returned
|
||||
try:
|
||||
flags = Allow.ALL
|
||||
obj, _ = partial_json_loads(current_text, flags)
|
||||
except (
|
||||
partial_json_parser.core.exceptions.MalformedJSON,
|
||||
json.JSONDecodeError,
|
||||
):
|
||||
obj = None
|
||||
|
||||
# check if the current text is a valid array
|
||||
# containing a partial tool calling object
|
||||
# if not repeat
|
||||
if obj is None or not isinstance(obj, list) or not len(obj) > 0:
|
||||
function_name_returned = False
|
||||
delta_message = None
|
||||
else:
|
||||
_, finishes_previous_tool = filter_delta_text(delta_text, previous_text)
|
||||
# take the last tool call from the generated list
|
||||
current_tool_call = obj[-1]
|
||||
|
||||
# once parameters have been generated the name is complete as well
|
||||
if not finishes_previous_tool and (
|
||||
"name" not in current_tool_call or "parameters" not in current_tool_call
|
||||
):
|
||||
function_name_returned = False
|
||||
delta_message = None
|
||||
else:
|
||||
if not function_name_returned:
|
||||
# get partly generated arguments from the latest tool call
|
||||
param_match = re.search(
|
||||
r'.*"parameters":\s*(.*)', current_text, re.DOTALL
|
||||
)
|
||||
arguments = param_match.group(1) if param_match else ""
|
||||
arguments, _ = filter_delta_text(arguments, previous_text)
|
||||
|
||||
# if this iteration finishes a previous tool call but a
|
||||
# new incomplete tool is already generated, take the
|
||||
# previous from the list
|
||||
if finishes_previous_tool and "parameters" not in current_tool_call:
|
||||
current_tool_call = obj[-2]
|
||||
|
||||
function_name_returned = True
|
||||
tool_call_id = make_tool_call_id(
|
||||
id_type=tool_call_id_type,
|
||||
func_name=current_tool_call["name"],
|
||||
idx=tool_call_idx,
|
||||
)
|
||||
delta_message = DeltaMessage(
|
||||
tool_calls=[
|
||||
DeltaToolCall(
|
||||
id=tool_call_id,
|
||||
function=DeltaFunctionCall(
|
||||
name=current_tool_call["name"], arguments=arguments
|
||||
),
|
||||
index=len(obj) - 1,
|
||||
type="function",
|
||||
)
|
||||
]
|
||||
)
|
||||
|
||||
else:
|
||||
delta_text, _ = filter_delta_text(delta_text, previous_text)
|
||||
|
||||
if delta_text != "":
|
||||
delta_message = DeltaMessage(
|
||||
tool_calls=[
|
||||
DeltaToolCall(
|
||||
function=DeltaFunctionCall(
|
||||
# OpenAI API returns None
|
||||
# instead of name every time
|
||||
name=None,
|
||||
arguments=delta_text,
|
||||
),
|
||||
index=len(obj) - 1,
|
||||
)
|
||||
]
|
||||
)
|
||||
else:
|
||||
delta_message = None
|
||||
|
||||
return delta_message, function_name_returned
|
||||
Reference in New Issue
Block a user