diff --git a/tests/entrypoints/openai/responses/test_function_call.py b/tests/entrypoints/openai/responses/test_function_call.py index f198e1cfa7..515f31b399 100644 --- a/tests/entrypoints/openai/responses/test_function_call.py +++ b/tests/entrypoints/openai/responses/test_function_call.py @@ -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() == "" diff --git a/vllm/entrypoints/openai/chat_completion/serving.py b/vllm/entrypoints/openai/chat_completion/serving.py index b916379f3f..f001aac15a 100644 --- a/vllm/entrypoints/openai/chat_completion/serving.py +++ b/vllm/entrypoints/openai/chat_completion/serving.py @@ -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, diff --git a/vllm/entrypoints/openai/responses/serving.py b/vllm/entrypoints/openai/responses/serving.py index 6d018f9c5b..ed9204ccaf 100644 --- a/vllm/entrypoints/openai/responses/serving.py +++ b/vllm/entrypoints/openai/responses/serving.py @@ -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, diff --git a/vllm/entrypoints/openai/responses/streaming_events.py b/vllm/entrypoints/openai/responses/streaming_events.py index cc242e7baa..425ae379b6 100644 --- a/vllm/entrypoints/openai/responses/streaming_events.py +++ b/vllm/entrypoints/openai/responses/streaming_events.py @@ -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 [] diff --git a/vllm/parser/abstract_parser.py b/vllm/parser/abstract_parser.py index b87455ab6f..5943e5aafb 100644 --- a/vllm/parser/abstract_parser.py +++ b/vllm/parser/abstract_parser.py @@ -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 diff --git a/vllm/tool_parsers/streaming.py b/vllm/tool_parsers/streaming.py new file mode 100644 index 0000000000..fc903328e3 --- /dev/null +++ b/vllm/tool_parsers/streaming.py @@ -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