diff --git a/tests/entrypoints/openai/responses/test_harmony.py b/tests/entrypoints/openai/responses/test_harmony.py index 88dd2d38457d..749ff0fa7838 100644 --- a/tests/entrypoints/openai/responses/test_harmony.py +++ b/tests/entrypoints/openai/responses/test_harmony.py @@ -13,7 +13,7 @@ import pytest import pytest_asyncio import requests -from openai import InternalServerError, NotFoundError, OpenAI +from openai import BadRequestError, NotFoundError, OpenAI from openai_harmony import Message from tests.utils import RemoteOpenAIServer @@ -699,11 +699,97 @@ async def test_function_calling_multi_turn(client: OpenAI, model_name: str): async def test_function_calling_required(client: OpenAI, model_name: str): tools = [GET_WEATHER_SCHEMA] - with pytest.raises(InternalServerError): + response = await client.responses.create( + model=model_name, + input="What's the weather like in Paris today?", + tools=tools, + tool_choice="required", + ) + + assert response.status == "completed" + assert has_output_type(response, "function_call"), ( + f"Expected function_call in output, got: {[o.type for o in response.output]}" + ) + assert not has_output_type(response, "message"), ( + "Required tool-choice JSON should not be returned as output_text." + ) + tool_call = next(o for o in response.output if o.type == "function_call") + assert tool_call.name == "get_weather" + args = json.loads(tool_call.arguments) + assert "latitude" in args + assert "longitude" in args + + +@pytest.mark.asyncio +@pytest.mark.parametrize("model_name", [MODEL_NAME]) +async def test_function_calling_required_with_stream( + pairs_of_event_types: dict[str, str], + client: OpenAI, + model_name: str, +): + tools = [GET_WEATHER_SCHEMA] + + stream = await client.responses.create( + model=model_name, + input="What's the weather like in Paris today?", + tools=tools, + tool_choice="required", + stream=True, + ) + + function_call_item = None + completed_event = None + completed_response = None + events = [] + text_deltas = [] + async for event in stream: + events.append(event) + if ( + event.type == "response.output_item.added" + and event.item.type == "function_call" + ): + function_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 function_call_item is not None + ): + function_call_item.arguments += event.delta + elif ( + event.type == "response.output_item.done" + and event.item.type == "function_call" + ): + completed_event = event + elif event.type == "response.completed": + completed_response = event.response + + validate_streaming_event_stack(events, pairs_of_event_types) + assert function_call_item is not None + assert function_call_item.type == "function_call" + assert function_call_item.name == "get_weather" + assert completed_event is not None + assert function_call_item.arguments == completed_event.item.arguments + assert completed_response is not None + assert has_output_type(completed_response, "function_call") + assert not has_output_type(completed_response, "message") + args = json.loads(function_call_item.arguments) + assert "latitude" in args + assert "longitude" in args + assert "".join(text_deltas).strip() == "" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("model_name", [MODEL_NAME]) +async def test_function_calling_required_builtin_tools_still_unsupported( + client: OpenAI, + model_name: str, +): + with pytest.raises(BadRequestError): await client.responses.create( model=model_name, - input="What's the weather like in Paris today?", - tools=tools, + input="Search for the latest vLLM release.", + tools=[{"type": "web_search_preview"}], tool_choice="required", ) diff --git a/tests/entrypoints/openai/responses/test_serving_responses.py b/tests/entrypoints/openai/responses/test_serving_responses.py index 25b00ff19278..13782851f413 100644 --- a/tests/entrypoints/openai/responses/test_serving_responses.py +++ b/tests/entrypoints/openai/responses/test_serving_responses.py @@ -1,12 +1,14 @@ # SPDX-License-Identifier: Apache-2.0 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project +import json from contextlib import AsyncExitStack from unittest.mock import MagicMock import pytest import pytest_asyncio from openai.types.responses import ( + FunctionTool, ResponseOutputItemDoneEvent, ResponseReasoningItem, ResponseReasoningTextDeltaEvent, @@ -51,7 +53,8 @@ ) from vllm.inputs import tokens_input from vllm.outputs import CompletionOutput, RequestOutput -from vllm.sampling_params import SamplingParams +from vllm.sampling_params import SamplingParams, StructuredOutputsParams +from vllm.tool_parsers.streaming import extract_required_tool_call_streaming class MockConversationContext(ConversationContext): @@ -154,6 +157,254 @@ def test_extract_tool_types(monkeypatch: pytest.MonkeyPatch) -> None: } +def _function_tool() -> FunctionTool: + return FunctionTool( + type="function", + name="get_weather", + description="Get weather.", + parameters={ + "type": "object", + "properties": { + "location": {"type": "string"}, + }, + "required": ["location"], + "additionalProperties": False, + }, + strict=True, + ) + + +def _harmony_required_tool_serving() -> OpenAIServingResponses: + # Bypass __init__ to test internal helpers without engine wiring. + serving = object.__new__(OpenAIServingResponses) + serving.use_harmony = True + serving.enable_log_outputs = False + serving.enable_store = True + serving.request_logger = None + serving.tool_call_id_type = "random" + serving.msg_store = {} + return serving + + +def test_harmony_required_function_call_output_replays() -> None: + serving = _harmony_required_tool_serving() + request = ResponsesRequest( + input="Call get_weather for Paris.", + tools=[_function_tool()], + tool_choice="required", + store=True, + ) + serving.msg_store[request.request_id] = [] + output = CompletionOutput( + index=0, + text='[{"name":"get_weather","parameters":{"location":"Paris"}}]', + token_ids=[], + cumulative_logprob=0.0, + logprobs=None, + finish_reason="stop", + stop_reason=None, + ) + + items = serving._make_harmony_required_function_call_items(request, output) + + assert len(items) == 1 + item = items[0] + assert item.type == "function_call" + assert item.name == "get_weather" + assert json.loads(item.arguments) == {"location": "Paris"} + + stored_msg = serving.msg_store[request.request_id][0] + assert stored_msg.channel == "commentary" + assert stored_msg.recipient == "functions.get_weather" + assert stored_msg.content[0].text == '{"location": "Paris"}' + + prev_response = MagicMock() + prev_response.id = request.request_id + next_request = ResponsesRequest( + input="Continue.", + tools=[_function_tool()], + ) + + messages = serving._construct_input_messages_with_harmony( + next_request, + prev_response, + ) + + assert messages[0].channel == "commentary" + assert messages[0].recipient == "functions.get_weather" + assert messages[0].content[0].text == '{"location": "Paris"}' + assert messages[-1].author.role == "user" + assert messages[-1].content[0].text == "Continue." + + +@pytest.mark.parametrize( + ("text", "match"), + [ + ('[{"name":"get_weather"}]', "did not include object parameters"), + ('[{"name":"get_time","parameters":{}}]', "unknown tool"), + ], +) +def test_harmony_required_function_call_rejects_invalid_output( + text: str, + match: str, +) -> None: + request = ResponsesRequest( + input="Call get_weather for Paris.", + tools=[_function_tool()], + tool_choice="required", + ) + + with pytest.raises(ValueError, match=match): + OpenAIServingResponses._parse_required_function_tool_output(request, text) + + +@pytest.mark.parametrize( + ("request_kwargs", "param"), + [ + ({"max_tool_calls": 0}, "max_tool_calls"), + ( + {"structured_outputs": StructuredOutputsParams(json={"type": "object"})}, + "structured_outputs", + ), + ], +) +def test_harmony_required_function_call_rejects_conflicting_options( + request_kwargs: dict, + param: str, +) -> None: + serving = _harmony_required_tool_serving() + request = ResponsesRequest( + input="Call get_weather for Paris.", + tools=[_function_tool()], + tool_choice="required", + **request_kwargs, + ) + + error = serving._validate_create_responses_input(request) + + assert error is not None + assert error.error.type == "invalid_request_error" + assert error.error.param == param + + +def test_harmony_required_function_call_allows_text_format() -> None: + serving = _harmony_required_tool_serving() + text_schema = { + "type": "object", + "properties": {"answer": {"type": "string"}}, + "required": ["answer"], + "additionalProperties": False, + } + request = ResponsesRequest( + input="Call get_weather for Paris.", + tools=[_function_tool()], + tool_choice="required", + text=ResponseTextConfig( + format=ResponseFormatTextJSONSchemaConfig( + type="json_schema", + name="weather_answer", + schema=text_schema, + ) + ), + ) + + assert serving._validate_create_responses_input(request) is None + sampling_params = request.to_sampling_params(default_max_tokens=128) + assert sampling_params.structured_outputs is not None + assert sampling_params.structured_outputs.json == text_schema + + serving._apply_harmony_required_function_tool_schema( + request, sampling_params, is_required_function_call=True + ) + + assert sampling_params.structured_outputs is not None + tool_schema = sampling_params.structured_outputs.json + assert isinstance(tool_schema, dict) + assert tool_schema["type"] == "array" + assert tool_schema["items"]["anyOf"][0]["properties"]["name"]["enum"] == [ + "get_weather" + ] + + +def test_harmony_required_function_tool_schema_honors_limits() -> None: + single_tool_schema = OpenAIServingResponses._required_function_tool_json_schema( + ResponsesRequest( + input="Call get_weather for Paris.", + tools=[_function_tool()], + tool_choice="required", + parallel_tool_calls=False, + ) + ) + limited_schema = OpenAIServingResponses._required_function_tool_json_schema( + ResponsesRequest( + input="Call get_weather for Paris.", + tools=[_function_tool()], + tool_choice="required", + max_tool_calls=3, + ) + ) + + assert isinstance(single_tool_schema, dict) + assert single_tool_schema["maxItems"] == 1 + assert isinstance(limited_schema, dict) + assert limited_schema["maxItems"] == 3 + + +def test_harmony_required_function_tool_schema_preserves_tool_parameters() -> None: + tool = _function_tool() + tool.parameters["$defs"] = {"city": {"type": "string"}} + request = ResponsesRequest( + input="Call get_weather for Paris.", + tools=[tool], + tool_choice="required", + ) + + OpenAIServingResponses._required_function_tool_json_schema(request) + + assert "$defs" in tool.parameters + + +@pytest.mark.parametrize( + "chunks", + [ + ['[{"name":"get_weather","parameters":{"location":"Paris"}}]'], + [ + '[{"name":"get_weather",', + '"parameters":{"location":"Paris"}}]', + ], + ], +) +def test_harmony_required_function_call_streaming_reconstructs_arguments( + chunks: list[str], +) -> None: + previous_text = "" + function_name_returned = False + name = None + arguments = "" + + for chunk in chunks: + current_text = previous_text + chunk + delta_message, function_name_returned = extract_required_tool_call_streaming( + previous_text=previous_text, + current_text=current_text, + delta_text=chunk, + function_name_returned=function_name_returned, + tool_call_idx=0, + tool_call_id_type="random", + ) + previous_text = current_text + + if delta_message is None: + continue + tool_call = delta_message.tool_calls[0] + if tool_call.function.name: + name = tool_call.function.name + arguments += tool_call.function.arguments or "" + + assert name == "get_weather" + assert json.loads(arguments) == {"location": "Paris"} + + @pytest.mark.skip_global_cleanup def test_response_created_event_uses_public_json_schema_alias() -> None: schema = { diff --git a/vllm/entrypoints/openai/responses/serving.py b/vllm/entrypoints/openai/responses/serving.py index 112328def214..d9e0f59b5823 100644 --- a/vllm/entrypoints/openai/responses/serving.py +++ b/vllm/entrypoints/openai/responses/serving.py @@ -2,16 +2,18 @@ # SPDX-FileCopyrightText: Copyright contributors to the vLLM project import asyncio +import json import time from collections import deque from collections.abc import AsyncGenerator, AsyncIterator, Callable, Mapping, Sequence from contextlib import AsyncExitStack -from copy import copy +from copy import copy, deepcopy from http import HTTPStatus -from typing import Any, Final +from typing import Any, Final, cast from fastapi import Request from openai.types.responses import ( + FunctionTool, ResponseFunctionToolCall, ResponseOutputItem, ResponseOutputMessage, @@ -22,7 +24,7 @@ from openai.types.responses.response_output_text import Logprob, LogprobTopLogprob from openai.types.responses.tool import Mcp, Tool from openai_harmony import Message as OpenAIHarmonyMessage -from pydantic import TypeAdapter +from pydantic import TypeAdapter, ValidationError from vllm import envs from vllm.config.utils import replace @@ -31,11 +33,13 @@ ChatCompletionMessageParam, ChatTemplateContentFormatOption, get_tool_call_id_type, + make_tool_call_id, ) from vllm.entrypoints.mcp.tool_server import ToolServer from vllm.entrypoints.openai.engine.protocol import ( DeltaMessage, ErrorResponse, + FunctionDefinition, RequestResponseMetadata, ) from vllm.entrypoints.openai.engine.serving import ( @@ -45,6 +49,7 @@ from vllm.entrypoints.openai.models.serving import OpenAIServingModels from vllm.entrypoints.openai.parser.harmony_utils import ( get_developer_message, + get_encoding, get_system_message, get_user_message, has_custom_tools, @@ -105,11 +110,16 @@ from vllm.sampling_params import SamplingParams, StructuredOutputsParams from vllm.tokenizers import TokenizerLike from vllm.tool_parsers import ToolParser +from vllm.tool_parsers.streaming import extract_required_tool_call_streaming +from vllm.tool_parsers.utils import get_json_schema_from_tools from vllm.utils import random_uuid from vllm.utils.collection_utils import as_list logger = init_logger(__name__) +# Matches the Harmony final-channel header emitted before final text content. +_HARMONY_FINAL_MESSAGE_PREFILL = "<|channel|>final<|message|>" + def _extract_allowed_tools_from_mcp_requests( tools: list[Tool], @@ -313,6 +323,45 @@ def _validate_create_responses_input( status_code=HTTPStatus.BAD_REQUEST, param="previous_response_id", ) + is_required_function_call = self._is_harmony_required_function_call_request( + request + ) + if ( + self.use_harmony + and request.tool_choice == "required" + and request.tools + and not is_required_function_call + ): + return self.create_error_response( + err_type="invalid_request_error", + message=( + "tool_choice='required' for Harmony Responses currently " + "supports function tools only." + ), + status_code=HTTPStatus.BAD_REQUEST, + param="tool_choice", + ) + if is_required_function_call: + if request.max_tool_calls is not None and request.max_tool_calls < 1: + return self.create_error_response( + err_type="invalid_request_error", + message=( + "max_tool_calls must be greater than 0 when " + "tool_choice='required' for Harmony Responses." + ), + status_code=HTTPStatus.BAD_REQUEST, + param="max_tool_calls", + ) + if request.structured_outputs is not None: + return self.create_error_response( + err_type="invalid_request_error", + message=( + "structured_outputs cannot be combined with " + "tool_choice='required' for Harmony Responses." + ), + status_code=HTTPStatus.BAD_REQUEST, + param="structured_outputs", + ) return None async def create_responses( @@ -371,9 +420,12 @@ async def _create_responses( lora_request = self._maybe_get_adapters(request) model_name = self.models.model_name(lora_request) + is_required_function_call = self._is_harmony_required_function_call_request( + request + ) if self.use_harmony: messages, engine_inputs = self._make_request_with_harmony( - request, prev_response + request, prev_response, is_required_function_call ) else: messages, engine_inputs = await self._make_request(request, prev_response) @@ -435,6 +487,9 @@ async def _create_responses( sampling_params = request.to_sampling_params( default_max_tokens, self.default_sampling_params ) + self._apply_harmony_required_function_tool_schema( + request, sampling_params, is_required_function_call + ) trace_headers = ( None @@ -443,8 +498,10 @@ async def _create_responses( ) context: ConversationContext - function_tool_names = extract_function_tool_names(request.tools) - if self.use_harmony: + if is_required_function_call: + context = SimpleContext() + elif self.use_harmony: + function_tool_names = extract_function_tool_names(request.tools) if request.stream: context = StreamingHarmonyContext( messages, available_tools, function_tool_names @@ -729,21 +786,162 @@ def _make_request_with_harmony( self, request: ResponsesRequest, prev_response: ResponsesResponse | None, + is_required_function_call: bool, ): - if request.tool_choice not in ("auto", "none"): + has_unsupported_tool_choice = request.tool_choice not in ("auto", "none") + if has_unsupported_tool_choice and not is_required_function_call: raise NotImplementedError( - "Only 'auto' or 'none' tool_choice is supported " - "in response API with Harmony" + "Only 'auto', 'none', or 'required' (with function tools) " + "tool_choice is supported in response API with Harmony" ) arrival_time = time.time() messages = self._construct_input_messages_with_harmony(request, prev_response) prompt_token_ids = render_for_completion(messages) + if is_required_function_call: + prompt_token_ids += get_encoding().encode( + _HARMONY_FINAL_MESSAGE_PREFILL, allowed_special="all" + ) engine_input = tokens_input(prompt_token_ids, cache_salt=request.cache_salt) engine_input["arrival_time"] = arrival_time return messages, [engine_input] + def _is_harmony_required_function_call_request( + self, + request: ResponsesRequest, + ) -> bool: + """Return True when Harmony Responses should force function calls.""" + return ( + self.use_harmony + and request.tool_choice == "required" + and bool(request.tools) + and all(isinstance(tool, FunctionTool) for tool in request.tools) + ) + + @staticmethod + def _required_function_tool_json_schema(request: ResponsesRequest) -> dict: + """Build the JSON schema used to force required function calls.""" + json_schema = get_json_schema_from_tools( + tool_choice=request.tool_choice, + tools=deepcopy(request.tools), + ) + if json_schema is None: + raise ValueError("tool_choice='required' requires function tools.") + json_schema = cast(dict[str, Any], json_schema) + + max_items = request.max_tool_calls + if request.parallel_tool_calls is False: + max_items = min(max_items, 1) if max_items is not None else 1 + if max_items is not None: + json_schema["maxItems"] = max_items + return json_schema + + def _apply_harmony_required_function_tool_schema( + self, + request: ResponsesRequest, + sampling_params: SamplingParams, + is_required_function_call: bool, + ) -> None: + """Attach required-function constrained decoding to sampling params.""" + if not is_required_function_call: + return + sampling_params.structured_outputs = StructuredOutputsParams( + json=self._required_function_tool_json_schema(request) + ) + + @staticmethod + def _parse_required_function_tool_output( + request: ResponsesRequest, + text: str, + ) -> list[FunctionDefinition]: + """Parse constrained JSON into function-call definitions.""" + try: + tool_calls = TypeAdapter(list[FunctionDefinition]).validate_json(text) + except ValidationError as exc: + raise ValueError( + "Failed to parse required tool-choice output as tool-call JSON." + ) from exc + if not tool_calls: + raise ValueError( + "Required tool-choice output did not contain any tool calls." + ) + + allowed_tool_names = { + tool.name for tool in request.tools if isinstance(tool, FunctionTool) + } + for tool_call in tool_calls: + if tool_call.name not in allowed_tool_names: + raise ValueError( + "Required tool-choice output referenced an unknown tool: " + f"{tool_call.name!r}." + ) + if tool_call.parameters is None: + raise ValueError( + "Required tool-choice output did not include object " + f"parameters for tool {tool_call.name!r}." + ) + return tool_calls + + def _make_harmony_required_function_call_items( + self, + request: ResponsesRequest, + final_output: CompletionOutput, + ) -> list[ResponseOutputItem]: + """Convert constrained JSON output into public function_call items.""" + if self.enable_log_outputs and self.request_logger: + self.request_logger.log_outputs( + request_id=request.request_id, + outputs=final_output.text, + output_token_ids=final_output.token_ids, + finish_reason=final_output.finish_reason, + is_streaming=False, + delta=False, + ) + + tool_calls = self._parse_required_function_tool_output( + request=request, + text=final_output.text, + ) + output_items: list[ResponseOutputItem] = [] + for idx, tool_call in enumerate(tool_calls): + output_items.append( + ResponseFunctionToolCall( + id=f"fc_{random_uuid()}", + call_id=make_tool_call_id( + id_type=self.tool_call_id_type, + func_name=tool_call.name, + idx=idx, + ), + type="function_call", + status="completed", + name=tool_call.name, + arguments=json.dumps( + tool_call.parameters, + ensure_ascii=False, + ), + ) + ) + self._append_required_function_calls_to_harmony_store( + request, + output_items, + ) + return output_items + + def _append_required_function_calls_to_harmony_store( + self, + request: ResponsesRequest, + output_items: list[ResponseOutputItem], + ) -> None: + """Store equivalent Harmony calls so previous_response_id can replay.""" + if not request.store or request.request_id not in self.msg_store: + return + for item in output_items: + # function_call conversion does not use previous outputs. + msg = response_input_to_harmony(item, []) + if msg is not None: + self.msg_store[request.request_id].append(msg) + async def _initialize_tool_sessions( self, request: ResponsesRequest, @@ -789,7 +987,31 @@ async def responses_full_generator( input_messages: ResponseInputOutputMessage | None = None output_messages: ResponseInputOutputMessage | None = None - if self.use_harmony: + if self._is_harmony_required_function_call_request(request): + assert isinstance(context, SimpleContext) + final_res = context.final_output + assert final_res is not None + assert len(final_res.outputs) == 1 + final_output = final_res.outputs[0] + + self._raise_if_error(final_output.finish_reason, request.request_id) + + if final_output.finish_reason == "length": + status = "incomplete" + output = [] + else: + output = self._make_harmony_required_function_call_items( + request, + final_output, + ) + + if request.enable_response_messages: + input_messages = context.input_messages + output_messages = context.output_messages + + assert final_res.prompt_token_ids is not None + num_tool_output_tokens = 0 + elif self.use_harmony: assert isinstance(context, HarmonyContext) output = self._make_response_output_items_with_harmony(context) if request.enable_response_messages: @@ -1370,13 +1592,19 @@ async def _process_simple_streaming_events( ], ) -> AsyncGenerator[StreamingResponsesResponse, None]: processor = SimpleStreamingEventProcessor() + is_required_function_call = self._is_harmony_required_function_call_request( + request + ) + previous_tool_json_text = "" + history_tool_call_cnt = 0 + function_name_returned = False parser = ( self.parser( tokenizer, request.tools, chat_template_kwargs=self._effective_chat_template_kwargs(request), ) - if self.parser + if self.parser and not is_required_function_call else None ) @@ -1402,7 +1630,41 @@ def _get_logprobs( delta_text = output.text delta_token_ids = as_list(output.token_ids) - if parser: + if is_required_function_call: + current_tool_json_text = previous_tool_json_text + delta_text + delta_message, function_name_returned = ( + extract_required_tool_call_streaming( + previous_text=previous_tool_json_text, + current_text=current_tool_json_text, + delta_text=delta_text, + function_name_returned=function_name_returned, + tool_call_idx=history_tool_call_cnt, + tool_call_id_type=self.tool_call_id_type, + ) + ) + if ( + delta_message + and delta_message.tool_calls + and delta_message.tool_calls[0].id is not None + ): + history_tool_call_cnt += 1 + previous_tool_json_text = current_tool_json_text + if ( + output.finish_reason is not None + and output.finish_reason != "length" + ): + try: + # Late validation checks schema-level rules that are not + # knowable while streaming partial JSON shards. + self._parse_required_function_tool_output( + request, + previous_tool_json_text, + ) + except ValueError as exc: + raise GenerationError( + "Failed to parse required tool-choice output." + ) from exc + elif parser: delta_message = parser.parse_delta( delta_text=delta_text, delta_token_ids=delta_token_ids, @@ -1500,8 +1762,11 @@ def _increment_sequence_number_and_return( sequence_number += 1 return event + is_required_function_call = self._is_harmony_required_function_call_request( + request + ) async with AsyncExitStack() as exit_stack: - if self.use_harmony: + if self.use_harmony and not is_required_function_call: # TODO: in streaming, we noticed this bug: # https://github.com/vllm-project/vllm/issues/25697 await self._initialize_tool_sessions(request, context, exit_stack) diff --git a/vllm/tool_parsers/streaming.py b/vllm/tool_parsers/streaming.py index 7f6638dcb94e..d45303af7c04 100644 --- a/vllm/tool_parsers/streaming.py +++ b/vllm/tool_parsers/streaming.py @@ -146,6 +146,11 @@ def extract_required_tool_call_streaming( ) arguments = param_match.group(1) if param_match else "" arguments, _ = filter_delta_text(arguments, previous_text) + try: + _, end = json.JSONDecoder().raw_decode(arguments) + arguments = arguments[:end] + except json.JSONDecodeError: + pass # if this iteration finishes a previous tool call but a # new incomplete tool is already generated, take the