diff --git a/python/sglang/srt/entrypoints/harmony_utils.py b/python/sglang/srt/entrypoints/harmony_utils.py index 86d4356dc..33bd88783 100644 --- a/python/sglang/srt/entrypoints/harmony_utils.py +++ b/python/sglang/srt/entrypoints/harmony_utils.py @@ -3,6 +3,7 @@ # Adapted from vLLM: https://github.com/vllm-project/vllm/blob/1b9902806915040ac9b3029f2ab7522ec505afc3/vllm/entrypoints/harmony_utils.py # Slight differences in processing chat messages import datetime +import logging from collections.abc import Iterable from typing import Literal, Optional, Union @@ -42,6 +43,8 @@ from openai_harmony import ( from sglang.srt.entrypoints.openai.protocol import ResponseInputOutputItem from sglang.srt.utils import random_uuid +logger = logging.getLogger(__name__) + REASONING_EFFORT = { "high": ReasoningEffort.HIGH, "medium": ReasoningEffort.MEDIUM, @@ -92,13 +95,22 @@ def get_developer_message( if tools is not None: function_tools = [] for tool in tools: - if tool.type in ("web_search_preview", "code_interpreter"): + if tool.type in ( + "web_search", + "web_search_preview", + "code_interpreter", + ): # These are built-in tools that are added to the system message. pass elif tool.type == "function": function_tools.append(tool) else: - raise ValueError(f"tool type {tool.type} not supported") + # No harmony prompt template for the remaining built-ins; + # drop them so the request still runs. + logger.debug( + "harmony: ignoring unsupported response tool type %r", + tool.type, + ) if function_tools: function_tool_descriptions = [ ToolDescription.new( @@ -139,7 +151,16 @@ def parse_response_input( if isinstance(content, str): msg = Message.from_role_and_content(role, text_prefix + content) else: - contents = [TextContent(text=text_prefix + c["text"]) for c in content] + # Filter to text parts first, then enumerate, so the surviving first + # text chunk always carries the system→developer text_prefix even if + # earlier parts were non-text (image/audio) and got dropped. + text_chunks = [ + c for c in content if c.get("type") in ("text", "input_text") + ] + contents = [ + TextContent(text=(text_prefix if i == 0 else "") + c.get("text", "")) + for i, c in enumerate(text_chunks) + ] msg = Message.from_role_and_contents(role, contents) elif response_msg["type"] == "function_call_output": call_id = response_msg["call_id"] diff --git a/python/sglang/srt/entrypoints/http_server.py b/python/sglang/srt/entrypoints/http_server.py index cc1002084..86efbd085 100644 --- a/python/sglang/srt/entrypoints/http_server.py +++ b/python/sglang/srt/entrypoints/http_server.py @@ -1799,12 +1799,11 @@ async def v1_score_request(request: ScoringRequest, raw_request: Request): @app.post("/v1/responses", dependencies=[Depends(validate_json_request)]) -async def v1_responses_request(request: dict, raw_request: Request): +async def v1_responses_request(request: ResponsesRequest, raw_request: Request): """Endpoint for the responses API with reasoning support.""" - request_obj = ResponsesRequest(**request) result = await raw_request.app.state.openai_serving_responses.create_responses( - request_obj, raw_request + request, raw_request ) # Handle streaming responses diff --git a/python/sglang/srt/entrypoints/openai/protocol.py b/python/sglang/srt/entrypoints/openai/protocol.py index eae4a312f..d23366372 100644 --- a/python/sglang/srt/entrypoints/openai/protocol.py +++ b/python/sglang/srt/entrypoints/openai/protocol.py @@ -1284,14 +1284,46 @@ class ResponseReasoningParam(BaseModel): default="medium", description="Constrains effort on reasoning for reasoning models.", ) + summary: Optional[Literal["auto", "concise", "detailed"]] = Field( + default=None, + description="Include a summary of the model's reasoning trace on the response.", + ) + + +# Only ``function`` / ``web_search*`` / ``code_interpreter`` are wired to +# execution paths; the rest pass validation so clients aren't rejected. +RESPONSE_TOOL_TYPES = Literal[ + "function", + "web_search", + "web_search_preview", + "code_interpreter", + "file_search", + "image_generation", + "computer_use_preview", + "local_shell", + "mcp", + "custom", + "namespace", + "tool_search", +] class ResponseTool(BaseModel): """Tool definition for responses.""" - type: Literal["web_search_preview", "code_interpreter"] = Field( - description="Type of tool to enable" - ) + type: RESPONSE_TOOL_TYPES = Field(description="Type of tool to enable") + name: Optional[str] = None + description: Optional[str] = None + parameters: Optional[Dict[str, Any]] = None + strict: bool = False + # Inner schemas for ``namespace`` tools. + tools: Optional[List[Dict[str, Any]]] = None + + @model_validator(mode="after") + def validate_function_tool(self) -> "ResponseTool": + if self.type == "function" and not self.name: + raise ValueError("Function tools must include a name.") + return self ResponseInputOutputItem: TypeAlias = Union[ @@ -1318,7 +1350,9 @@ class ResponsesRequest(BaseModel): ] ] ] = None - input: Union[str, List[ResponseInputOutputItem]] + # Accept dict-shaped items as the loose arm; downstream normalization + # handles replayed shapes that don't satisfy every openai TypedDict. + input: Union[str, List[ResponseInputOutputItem], List[Dict[str, Any]]] instructions: Optional[str] = None max_output_tokens: Optional[int] = None max_tool_calls: Optional[int] = None @@ -1352,13 +1386,13 @@ class ResponsesRequest(BaseModel): default=None, description="Cache salt for request caching" ) - # SGLang-specific sampling parameters + # SGLang sampling extras. ``None`` defers to ``--preferred-sampling-params``. frequency_penalty: float = 0.0 presence_penalty: float = 0.0 stop: Optional[Union[str, List[str]]] = None - top_k: int = -1 - min_p: float = 0.0 - repetition_penalty: float = 1.0 + top_k: Optional[int] = None + min_p: Optional[float] = None + repetition_penalty: Optional[float] = None # Default sampling parameters _DEFAULT_SAMPLING_PARAMS = { @@ -1369,8 +1403,57 @@ class ResponsesRequest(BaseModel): "repetition_penalty": 1.0, } + @model_validator(mode="before") + @classmethod + def normalize_responses_input(cls, values): + if not isinstance(values, dict): + return values + + input_value = values.get("input") + if not isinstance(input_value, list): + return values + + values = values.copy() + values["input"] = [ + cls._normalize_input_item_for_validation(item) for item in input_value + ] + return values + + @staticmethod + def _normalize_input_item_for_validation(item): + if not isinstance(item, dict): + return item + + content = item.get("content") + if not isinstance(content, list): + return item + + item = item.copy() + item["content"] = [ + ResponsesRequest._normalize_content_part_for_validation(part) + for part in content + ] + return item + + @staticmethod + def _normalize_content_part_for_validation(part): + if not isinstance(part, dict): + return part + + part_type = part.get("type") + if part_type != "input_image" or part.get("detail") is not None: + return part + + part = part.copy() + part["detail"] = "auto" + return part + def to_sampling_params( - self, default_max_tokens: int, default_params: Optional[Dict] = None + self, + default_max_tokens: int, + default_params: Optional[Dict] = None, + stop: Optional[Union[str, List[str]]] = None, + tool_call_constraint: Optional[ToolCallConstraint] = None, ) -> Dict[str, Any]: """Convert to sampling parameters for generation.""" if default_params is None: @@ -1382,10 +1465,9 @@ class ResponsesRequest(BaseModel): else: max_tokens = default_max_tokens - # Avoid exceed the context length by minus 2 token + # Headroom for BOS/EOS the engine appends on top of prompt+budget. max_tokens -= 2 - # Get parameters with defaults temperature = self.temperature if temperature is None: temperature = default_params.get( @@ -1396,23 +1478,51 @@ class ResponsesRequest(BaseModel): if top_p is None: top_p = default_params.get("top_p", self._DEFAULT_SAMPLING_PARAMS["top_p"]) - params = { + # Omit None entries so they fall through to ``--preferred-sampling-params`` + # rather than overriding it with a literal default. + params: dict[str, Any] = { "max_new_tokens": max_tokens, "temperature": temperature, "top_p": top_p, "frequency_penalty": self.frequency_penalty, "presence_penalty": self.presence_penalty, - "stop": self.stop, - "top_k": self.top_k, - "min_p": self.min_p, - "repetition_penalty": self.repetition_penalty, + "stop": self.stop if stop is None else stop, } + if self.top_k is not None: + params["top_k"] = self.top_k + if self.min_p is not None: + params["min_p"] = self.min_p + if self.repetition_penalty is not None: + params["repetition_penalty"] = self.repetition_penalty # Apply any additional default parameters for key, value in default_params.items(): if key not in params or params[key] is None: params[key] = value + has_existing_constraints = ( + params.get("regex") + or params.get("ebnf") + or params.get("structural_tag") + or params.get("json_schema") + ) + if tool_call_constraint and has_existing_constraints: + # Refuse rather than silently drop the tool-call grammar. + raise ValueError( + "Cannot combine tool calls with constrained decoding " + "(regex / ebnf / structural_tag / json_schema). Remove one." + ) + if tool_call_constraint: + constraint_type, constraint_value = tool_call_constraint + if constraint_type in ("structural_tag", "json_schema"): + params[constraint_type] = convert_json_schema_to_str( + constraint_value.model_dump(by_alias=True) + if hasattr(constraint_value, "model_dump") + else constraint_value + ) + else: + params[constraint_type] = constraint_value + return params @@ -1515,7 +1625,11 @@ class ResponsesResponse(BaseModel): output=output, status=status, usage=usage, - parallel_tool_calls=request.parallel_tool_calls or True, + parallel_tool_calls=( + request.parallel_tool_calls + if request.parallel_tool_calls is not None + else True + ), tool_choice=request.tool_choice, tools=request.tools, # fields for parity with v1/responses diff --git a/python/sglang/srt/entrypoints/openai/serving_responses.py b/python/sglang/srt/entrypoints/openai/serving_responses.py index edda5a471..86ed083fc 100644 --- a/python/sglang/srt/entrypoints/openai/serving_responses.py +++ b/python/sglang/srt/entrypoints/openai/serving_responses.py @@ -5,7 +5,6 @@ from __future__ import annotations import asyncio -import copy import json import logging import time @@ -27,6 +26,15 @@ from openai.types.responses.response_function_tool_call import ResponseFunctionT from openai.types.responses.response_reasoning_item import ( Content as ResponseReasoningTextContent, ) +from openai.types.responses.response_reasoning_item import ( + Summary as ResponseReasoningSummary, +) +from openai.types.responses.response_reasoning_summary_part_added_event import ( + Part as ResponseReasoningSummaryAddedPart, +) +from openai.types.responses.response_reasoning_summary_part_done_event import ( + Part as ResponseReasoningSummaryDonePart, +) from openai_harmony import Message as OpenAIMessage from sglang.srt.entrypoints.context import ( @@ -48,14 +56,19 @@ from sglang.srt.entrypoints.harmony_utils import ( from sglang.srt.entrypoints.openai.protocol import ( ChatCompletionMessageParam, ChatCompletionRequest, + Function, + MessageProcessingResult, PromptTokenUsageInfo, RequestResponseMetadata, ResponsesRequest, ResponsesResponse, + Tool, UsageInfo, ) from sglang.srt.entrypoints.openai.serving_chat import OpenAIServingChat from sglang.srt.entrypoints.openai.tool_server import MCPToolServer, ToolServer +from sglang.srt.function_call.function_call_parser import FunctionCallParser +from sglang.srt.function_call.json_array_parser import JsonArrayParser from sglang.srt.managers.io_struct import GenerateReqInput from sglang.srt.parser.reasoning_parser import ReasoningParser from sglang.srt.utils import random_uuid @@ -84,8 +97,9 @@ class OpenAIServingResponses(OpenAIServingChat): self.reasoning_parser = self.tokenizer_manager.server_args.reasoning_parser self.enable_prompt_tokens_details = enable_prompt_tokens_details - # Get default sampling params from model config if available - self.default_sampling_params = {} + # Parent OpenAIServingChat.__init__ already populated default_sampling_params. + if not isinstance(self.default_sampling_params, dict): + self.default_sampling_params = {} self.supports_browsing = ( tool_server.has_tool("browser") if tool_server else False @@ -171,6 +185,15 @@ class OpenAIServingResponses(OpenAIServingChat): # FIXME: If the engine is dead, raise an error # This is required for the streaming case + # ``tool_choice="required"`` only works with ``function`` tools. + if request.tool_choice == "required" and not any( + tool.type == "function" for tool in (request.tools or []) + ): + return self.create_error_response( + 'tool_choice="required" requires at least one tool with ' + 'type="function"; other built-in tool types cannot be forced.' + ) + # Handle the previous response ID prev_response_id = request.previous_response_id if prev_response_id is not None: @@ -186,15 +209,19 @@ class OpenAIServingResponses(OpenAIServingChat): try: model_name = request.model tokenizer = self.tokenizer_manager.tokenizer + processed_messages: Optional[MessageProcessingResult] = None if self.use_harmony: messages, request_prompts, engine_prompts = ( self._make_request_with_harmony(request, prev_response) ) else: - messages, request_prompts, engine_prompts = await self._make_request( - request, prev_response, tokenizer - ) + ( + messages, + request_prompts, + engine_prompts, + processed_messages, + ) = await self._make_request(request, prev_response, tokenizer) except (ValueError, TypeError, RuntimeError, jinja2.TemplateError) as e: logger.exception("Error in preprocessing prompt inputs") @@ -210,7 +237,7 @@ class OpenAIServingResponses(OpenAIServingChat): and (request.background or request.stream) and request.tools and any( - tool.type in ["web_search_preview", "code_interpreter"] + tool.type in ("web_search", "web_search_preview", "code_interpreter") for tool in request.tools ) ): @@ -244,10 +271,10 @@ class OpenAIServingResponses(OpenAIServingChat): tool_sessions = {} for i, engine_prompt in enumerate(engine_prompts): # Calculate default max tokens from context length minus prompt length - if hasattr(engine_prompt, "__len__"): - prompt_length = len(engine_prompt) - elif isinstance(engine_prompt, list): + if isinstance(engine_prompt, list): prompt_length = len(engine_prompt) + elif isinstance(engine_prompt, str): + prompt_length = len(tokenizer.encode(engine_prompt)) else: prompt_length = 0 @@ -263,7 +290,18 @@ class OpenAIServingResponses(OpenAIServingChat): context_len - prompt_length - num_reserved_tokens, 512 ) # Ensure minimum 512 tokens sampling_params = request.to_sampling_params( - default_max_tokens, self.default_sampling_params + default_max_tokens, + self.default_sampling_params, + stop=( + processed_messages.stop + if processed_messages + else request.stop + ), + tool_call_constraint=( + processed_messages.tool_call_constraint + if processed_messages + else None + ), ) context: ConversationContext @@ -276,8 +314,33 @@ class OpenAIServingResponses(OpenAIServingChat): context = SimpleContext() # Create GenerateReqInput for SGLang + if isinstance(engine_prompt, str): + prompt_kwargs = {"text": engine_prompt} + else: + prompt_kwargs = {"input_ids": engine_prompt} + adapted_request = GenerateReqInput( - input_ids=engine_prompt, + **prompt_kwargs, + image_data=( + processed_messages.image_data + if processed_messages + else None + ), + video_data=( + processed_messages.video_data + if processed_messages + else None + ), + audio_data=( + processed_messages.audio_data + if processed_messages + else None + ), + modalities=( + processed_messages.modalities + if processed_messages + else None + ), sampling_params=sampling_params, stream=request.stream, rid=request.request_id, @@ -342,11 +405,20 @@ class OpenAIServingResponses(OpenAIServingChat): return response if request.stream: - return self.responses_stream_generator( + if self.use_harmony: + return self.responses_stream_generator( + request, + sampling_params, + result_generator, + context, + model_name, + tokenizer, + request_metadata, + ) + return self.responses_stream_generator_non_harmony( request, sampling_params, result_generator, - context, model_name, tokenizer, request_metadata, @@ -374,43 +446,34 @@ class OpenAIServingResponses(OpenAIServingChat): prev_response: Optional[ResponsesResponse], tokenizer: Any, ): - # Construct the input messages messages = self._construct_input_messages(request, prev_response) - # Follow SGLang's pattern: create a ChatCompletionRequest and process messages - try: - # Convert ResponsesRequest to ChatCompletionRequest for processing - chat_request = ChatCompletionRequest( - model=request.model, - messages=messages, - stream=request.stream, - ) + chat_tools = self._response_tools_to_chat_tools(request) + chat_request = ChatCompletionRequest( + model=request.model, + messages=messages, + stream=request.stream, + tools=chat_tools or None, + tool_choice=request.tool_choice if chat_tools else "none", + parallel_tool_calls=( + request.parallel_tool_calls + if request.parallel_tool_calls is not None + else True + ), + stop=request.stop, + ) - # Follow SGLang's _process_messages pattern - is_multimodal = self.tokenizer_manager.model_config.is_multimodal - processed_messages = self._process_messages(chat_request, is_multimodal) + is_multimodal = self.tokenizer_manager.model_config.is_multimodal + processed_messages = self._process_messages(chat_request, is_multimodal) - # Extract the results - if is_multimodal: - request_prompts = [processed_messages.prompt] - engine_prompts = [processed_messages.prompt] - else: - request_prompts = [processed_messages.prompt_ids] - engine_prompts = [processed_messages.prompt_ids] + if is_multimodal: + request_prompts = [processed_messages.prompt] + engine_prompts = [processed_messages.prompt] + else: + request_prompts = [processed_messages.prompt_ids] + engine_prompts = [processed_messages.prompt_ids] - except Exception as e: - logger.warning(f"Chat processing failed, using fallback: {e}") - # Fallback to simple encoding - prompt_text = "" - for msg in messages: - role = msg.get("role", "user") - content = msg.get("content", "") - prompt_text += f"{role}: {content}\n" - prompt_ids = tokenizer.encode(prompt_text) - request_prompts = [prompt_ids] - engine_prompts = [prompt_ids] - - return messages, request_prompts, engine_prompts + return messages, request_prompts, engine_prompts, processed_messages def _make_request_with_harmony( self, @@ -451,7 +514,7 @@ class OpenAIServingResponses(OpenAIServingChat): if self.use_harmony: assert isinstance(context, HarmonyContext) output = self._make_response_output_items_with_harmony(context) - # TODO: these are all 0 for now! + # num_reasoning_tokens isn't wired through HarmonyContext yet; stays 0. num_prompt_tokens = context.num_prompt_tokens num_generated_tokens = context.num_output_tokens num_cached_tokens = context.num_cached_tokens @@ -466,10 +529,29 @@ class OpenAIServingResponses(OpenAIServingChat): ) # Calculate usage from actual output - if hasattr(final_res, "meta_info"): - num_prompt_tokens = final_res.meta_info.get("prompt_tokens", 0) - num_generated_tokens = final_res.meta_info.get("completion_tokens", 0) - num_cached_tokens = final_res.meta_info.get("cached_tokens", 0) + num_reasoning_tokens = 0 + meta_info = None + if isinstance(final_res, dict) and isinstance( + final_res.get("meta_info"), dict + ): + meta_info = final_res["meta_info"] + elif hasattr(final_res, "meta_info"): + meta_info = final_res.meta_info + + if meta_info is not None: + num_prompt_tokens = meta_info.get("prompt_tokens", 0) + num_generated_tokens = meta_info.get("completion_tokens", 0) + num_cached_tokens = meta_info.get("cached_tokens", 0) + num_reasoning_tokens = meta_info.get("reasoning_tokens", 0) + elif isinstance(final_res, dict) and ( + final_res.get("prompt_token_ids") is not None + or final_res.get("output_ids") is not None + ): + prompt_token_ids = final_res.get("prompt_token_ids") or [] + output_token_ids = final_res.get("output_ids") or [] + num_prompt_tokens = len(prompt_token_ids) + num_generated_tokens = len(output_token_ids) + num_cached_tokens = final_res.get("num_cached_tokens", 0) elif hasattr(final_res, "prompt_token_ids") and hasattr( final_res, "outputs" ): @@ -483,7 +565,6 @@ class OpenAIServingResponses(OpenAIServingChat): else 0 ) num_cached_tokens = getattr(final_res, "num_cached_tokens", 0) - num_reasoning_tokens = 0 else: # Final fallback num_prompt_tokens = 0 @@ -522,18 +603,57 @@ class OpenAIServingResponses(OpenAIServingChat): return response + @staticmethod + def _wants_reasoning_summary(request: ResponsesRequest) -> bool: + return request.reasoning is not None and request.reasoning.summary is not None + + def _is_thinking_enabled_for_request(self, request: ResponsesRequest) -> bool: + """Whether to start the reasoning detector in thinking mode.""" + if not self.reasoning_parser: + return False + effort = request.reasoning.effort if request.reasoning is not None else None + if self.reasoning_parser == "hunyuan": + return effort not in (None, "none", "no_think") + if self.template_manager.force_reasoning: + return True + config = self.template_manager.reasoning_config + if config is None: + # Parser-only models (DeepSeek-R1, …) carry the thinking default in + # the detector itself. + detector = getattr(self, "_reasoning_detector", None) + mode = getattr(detector, "reasoning_default", None) if detector else None + if mode is None or mode == "always": + return mode == "always" + if mode == "mistral": + return effort is not None and effort != "none" + if mode in ("thinking", "enable_thinking"): + return effort != "none" + if mode in ("explicit_thinking", "explicit_enable_thinking"): + return False + return False + if config.special_case == "always": + return True + if config.special_case == "mistral": + return effort is not None and effort != "none" + if config.toggle_param is None or config.default_enabled is None: + return False + if effort == "none": + return False + return bool(config.default_enabled) + def _make_response_output_items( self, request: ResponsesRequest, final_output: Any, tokenizer: Any, ): - # Handle reasoning parsing if enabled if self.reasoning_parser: - # Use standard reasoning parser (openai maps to T4Detector internally) + # Templates that prefill ```` only emit the close tag, so + # start the detector in thinking mode. reasoning_parser = ReasoningParser( model_type=self.reasoning_parser, stream_reasoning=False, + force_reasoning=self._is_thinking_enabled_for_request(request), request=request, ) reasoning_content, content = reasoning_parser.parse_non_stream(final_output) @@ -543,10 +663,21 @@ class OpenAIServingResponses(OpenAIServingChat): output_items = [] if reasoning_content: + # Mirror the single parsed blob into ``summary`` when the caller opts + # in via ``reasoning.summary``; full trace stays in ``content``. + wants_summary = self._wants_reasoning_summary(request) reasoning_item = ResponseReasoningItem( id=f"rs_{random_uuid()}", type="reasoning", - summary=[], + summary=( + [ + ResponseReasoningSummary( + type="summary_text", text=reasoning_content + ) + ] + if wants_summary + else [] + ), content=[ ResponseReasoningTextContent( type="reasoning_text", text=reasoning_content @@ -555,6 +686,65 @@ class OpenAIServingResponses(OpenAIServingChat): status=None, ) output_items.append(reasoning_item) + + chat_tools = self._response_tools_to_chat_tools(request) + is_required = request.tool_choice == "required" + tool_call_items: list[ResponseFunctionToolCall] = [] + parsed_via_native = False + if ( + content + and chat_tools + and self.tool_call_parser + and request.tool_choice != "none" + ): + parser = FunctionCallParser(chat_tools, self.tool_call_parser) + should_try_native = ( + not is_required or parser.detector.supports_structural_tag() + ) + if should_try_native and parser.has_tool_call(content): + try: + content, call_info_list = parser.parse_non_stream(content) + for call_info in call_info_list: + tool_call_items.append( + ResponseFunctionToolCall( + arguments=call_info.parameters or "", + call_id=f"call_{random_uuid()[:24]}", + type="function_call", + name=call_info.name, + id=f"fc_{random_uuid()[:8]}", + status="completed", + ) + ) + parsed_via_native = bool(call_info_list) + except Exception as e: + logger.error("Tool call parsing error: %s", e) + + if content and chat_tools and is_required and not parsed_via_native: + try: + tool_call_data = orjson.loads(content) + if isinstance(tool_call_data, dict): + tool_call_data = [tool_call_data] + if isinstance(tool_call_data, list): + for tool in tool_call_data: + if not isinstance(tool, dict) or "name" not in tool: + continue + arguments = json.dumps( + tool.get("parameters", {}), ensure_ascii=False + ) + tool_call_items.append( + ResponseFunctionToolCall( + arguments=arguments, + call_id=f"call_{random_uuid()[:24]}", + type="function_call", + name=tool["name"], + id=f"fc_{random_uuid()[:8]}", + status="completed", + ) + ) + content = "" + except Exception as e: + logger.error("Required tool JSON parse error: %s", e) + if content: output_text = ResponseOutputText( text=content, @@ -570,6 +760,7 @@ class OpenAIServingResponses(OpenAIServingChat): type="message", ) output_items.append(message) + output_items.extend(tool_call_items) return output_items def _make_response_output_items_with_harmony( @@ -586,6 +777,247 @@ class OpenAIServingResponses(OpenAIServingChat): output_items.extend(last_items) return output_items + @staticmethod + def _response_tools_to_chat_tools(request: ResponsesRequest) -> list[Tool]: + # Only ``function`` tools flow to chat; built-ins go through harmony. + chat_tools = [] + for tool in request.tools: + if tool.type != "function": + continue + chat_tools.append( + Tool( + type="function", + function=Function( + name=tool.name, + description=tool.description, + parameters=tool.parameters, + strict=tool.strict, + ), + ) + ) + return chat_tools + + @staticmethod + def _normalize_response_content_part_for_chat(content_part: Any) -> Any: + # Default detail=\"auto\" and lift flat min/max_dynamic_patch onto + # image_url so the image preprocessor sees them. + if hasattr(content_part, "model_dump"): + content_part = content_part.model_dump(exclude_none=True) + if not isinstance(content_part, dict): + return content_part + + part_type = content_part.get("type") + if part_type in ("input_text", "output_text"): + return {"type": "text", "text": content_part.get("text", "")} + + if part_type == "input_image": + image_url = content_part.get("image_url") + if isinstance(image_url, dict): + image_url_obj = image_url.copy() + else: + image_url_obj = {"url": image_url} + if not image_url_obj.get("detail"): + image_url_obj["detail"] = content_part.get("detail") or "auto" + for key in ("min_dynamic_patch", "max_dynamic_patch"): + if key in content_part and key not in image_url_obj: + image_url_obj[key] = content_part[key] + return {"type": "image_url", "image_url": image_url_obj} + + if part_type == "text": + return content_part + + if part_type == "image_url": + image_url = content_part.get("image_url") + if isinstance(image_url, str): + image_url = { + "url": image_url, + "detail": content_part.get("detail", "auto"), + } + elif isinstance(image_url, dict): + image_url = image_url.copy() + if not image_url.get("detail"): + image_url["detail"] = content_part.get("detail") or "auto" + return {**content_part, "image_url": image_url} + + return content_part + + @classmethod + def _normalize_response_message_for_chat(cls, message: Any) -> Any: + """Convert one Responses-API input item to a chat-completions message.""" + if hasattr(message, "model_dump"): + message = message.model_dump(exclude_none=True) + if not isinstance(message, dict): + return message + + # Most chat templates only recognize system/user/assistant/tool; + # collapse ``developer`` to ``system`` at the boundary. + if message.get("role") == "developer": + message = {**message, "role": "system"} + + msg_type = message.get("type") + if msg_type == "function_call": + # Coerce ``arguments`` to a valid JSON-object string so the chat + # template's unconditional ``orjson.loads`` survives truncated or + # dict-shaped echoes. + raw = message.get("arguments") + if isinstance(raw, str): + try: + parsed = orjson.loads(raw) if raw else None + except orjson.JSONDecodeError: + parsed = None + if not isinstance(parsed, dict): + raw = "{}" + elif isinstance(raw, dict): + raw = orjson.dumps(raw).decode("utf-8") + else: + raw = "{}" + return { + "role": "assistant", + "tool_calls": [ + { + "id": message.get("call_id") or message.get("id"), + "type": "function", + "function": { + "name": message.get("name"), + "arguments": raw, + }, + } + ], + } + if msg_type == "function_call_output": + return { + "role": "tool", + "tool_call_id": message.get("call_id"), + "content": message.get("output", ""), + } + # Reasoning items render as {role: assistant, reasoning_content}; + # empty ones drop instead of injecting an empty assistant block. + if msg_type == "reasoning": + # Prefer ``summary``; fall back to ``content`` only when summary + # is empty, since clients often populate both with the same text. + def _collect(parts): + out: list[str] = [] + for entry in parts or []: + if isinstance(entry, dict): + text = entry.get("text") + if text: + out.append(text) + return out + + text_parts = _collect(message.get("summary")) + if not text_parts: + text_parts = _collect(message.get("content")) + if not text_parts: + return None + return { + "role": "assistant", + "reasoning_content": "\n".join(text_parts), + } + if msg_type not in (None, "message"): + raise ValueError(f"Unsupported Responses API input item type: {msg_type!r}") + + content = message.get("content") + if not isinstance(content, list): + return { + k: v + for k, v in message.items() + if v is not None and k not in ("id", "status", "type") + } + + return { + k: v + for k, v in { + **message, + "content": [ + cls._normalize_response_content_part_for_chat(part) + for part in content + ], + }.items() + if v is not None and k not in ("id", "status", "type") + } + + @staticmethod + def _output_message_text(output_item: Any) -> Optional[str]: + """Return assistant text from a ``message`` output item (joining + ``output_text`` parts with newlines), or None for non-message items.""" + if isinstance(output_item, ResponseReasoningItem): + return None + if hasattr(output_item, "model_dump"): + output_item = output_item.model_dump(exclude_none=True) + if not isinstance(output_item, dict): + return None + if output_item.get("type") != "message": + return None + + text_parts = [] + for content in output_item.get("content") or []: + if isinstance(content, ResponseOutputText): + text_parts.append(content.text) + continue + if hasattr(content, "model_dump"): + content = content.model_dump(exclude_none=True) + if isinstance(content, dict) and content.get("type") == "output_text": + text = content.get("text") + if text is not None: + text_parts.append(text) + + return "\n".join(text_parts) if text_parts else None + + @staticmethod + def _merge_consecutive_assistant_messages( + messages: list, + ) -> list: + """Collapse runs of consecutive ``assistant`` dicts into one entry, + joining ``content`` and concatenating ``tool_calls`` and + ``reasoning_content`` so a logical turn renders as a single block.""" + merged: list = [] + for msg in messages: + if ( + isinstance(msg, dict) + and msg.get("role") == "assistant" + and merged + and isinstance(merged[-1], dict) + and merged[-1].get("role") == "assistant" + ): + prev = merged[-1] = dict(merged[-1]) + # Lift mixed str/list content to list parts so non-text parts + # (e.g. image_url) survive when the two sides differ in shape. + new_content = msg.get("content") + if new_content is not None and new_content != "": + prev_content = prev.get("content") + if prev_content is None or prev_content == "": + prev["content"] = new_content + elif isinstance(prev_content, str) and isinstance(new_content, str): + sep = "\n\n" if prev_content and new_content else "" + prev["content"] = prev_content + sep + new_content + else: + + def _as_parts(c): + if isinstance(c, list): + return list(c) + if isinstance(c, str) and c: + return [{"type": "text", "text": c}] + return [] + + prev["content"] = _as_parts(prev_content) + _as_parts( + new_content + ) + new_calls = msg.get("tool_calls") + if new_calls: + prev_calls = prev.get("tool_calls") or [] + prev["tool_calls"] = prev_calls + list(new_calls) + new_reasoning = msg.get("reasoning_content") + if new_reasoning: + prev_reasoning = prev.get("reasoning_content") + prev["reasoning_content"] = ( + f"{prev_reasoning}\n{new_reasoning}" + if prev_reasoning + else new_reasoning + ) + continue + merged.append(msg) + return merged + def _construct_input_messages( self, request: ResponsesRequest, @@ -606,26 +1038,49 @@ class OpenAIServingResponses(OpenAIServingChat): prev_msg = self.msg_store[prev_response.id] messages.extend(prev_msg) - # Add the previous output for output_item in prev_response.output: - # NOTE: We skip the reasoning output of the previous response - if isinstance(output_item, ResponseReasoningItem): + assistant_text = self._output_message_text(output_item) + if assistant_text is None: continue - for content in output_item.content: - messages.append( - { - "role": "system", - "content": request.instructions, - } - ) + messages.append({"role": "assistant", "content": assistant_text}) # Append the new input # Responses API supports simple text inputs without chat format if isinstance(request.input, str): messages.append({"role": "user", "content": request.input}) else: - messages.extend(request.input) # type: ignore - return messages + for input_item in request.input: + normalized = self._normalize_response_message_for_chat(input_item) + if normalized is not None: + messages.append(normalized) # type: ignore + + # One Responses-API assistant turn maps to multiple input items + # (message + function_call(s)); collapse them into one chat message + # so chat templates render a single assistant block per turn. + messages = self._merge_consecutive_assistant_messages(messages) + + # Most chat templates expect a single leading ``system`` message; + # coalesce any ``instructions`` + interleaved ``developer`` entries. + system_chunks: list[str] = [] + other_msgs: list = [] + for m in messages: + if isinstance(m, dict) and m.get("role") == "system": + content = m.get("content") + if isinstance(content, str): + system_chunks.append(content) + elif isinstance(content, list): + for part in content: + if isinstance(part, dict): + text = part.get("text") + if isinstance(text, str): + system_chunks.append(text) + else: + other_msgs.append(m) + if system_chunks: + return [ + {"role": "system", "content": "\n\n".join(system_chunks)} + ] + other_msgs + return other_msgs def _construct_input_messages_with_harmony( self, @@ -638,7 +1093,8 @@ class OpenAIServingResponses(OpenAIServingChat): reasoning_effort = request.reasoning.effort if request.reasoning else None tool_types = [tool.type for tool in request.tools] enable_browser = ( - "web_search_preview" in tool_types and self.tool_server is not None + any(t in tool_types for t in ("web_search", "web_search_preview")) + and self.tool_server is not None ) enable_code_interpreter = ( "code_interpreter" in tool_types and self.tool_server is not None @@ -693,7 +1149,7 @@ class OpenAIServingResponses(OpenAIServingChat): messages.append(get_user_message(request.input)) else: if prev_response is not None: - prev_outputs = copy(prev_response.output) + prev_outputs = list(prev_response.output) else: prev_outputs = [] for response_msg in request.input: @@ -1231,6 +1687,8 @@ class OpenAIServingResponses(OpenAIServingChat): ) # Convert final_response to the format expected by ResponseCompletedEvent response_dict = final_response.model_dump() + # OpenAI SDK's Tool union may not know extended types; drop echo. + response_dict["tools"] = [] # Convert UsageInfo to ResponseUsage format if response_dict.get("usage"): @@ -1255,6 +1713,596 @@ class OpenAIServingResponses(OpenAIServingChat): ) ) + async def responses_stream_generator_non_harmony( + self, + request: ResponsesRequest, + sampling_params: Any, + result_generator: AsyncIterator[Any], + model_name: str, + tokenizer: Any, + request_metadata: RequestResponseMetadata, + created_time: Optional[int] = None, + ) -> AsyncGenerator[str, None]: + """Stream a /v1/responses response as typed OpenAI SSE events for + non-harmony models. Each engine chunk is run through the reasoning + and function-call parsers; leftover text becomes + ``response.output_text.delta``. + """ + + created_time = created_time or int(time.time()) + sequence_number = 0 + + def _send_event(event): + nonlocal sequence_number + if hasattr(event, "sequence_number"): + event.sequence_number = sequence_number + sequence_number += 1 + event_type = getattr(event, "type", "unknown") + return ( + f"event: {event_type}\n" + f"data: {event.model_dump_json(indent=None)}\n\n" + ) + + # The streaming Response* event models echo ``tools`` through a + # narrower OpenAI SDK Tool union; strip it to avoid pydantic + # validation failures on extended tool types. + def _sanitize_response_dict(d: dict) -> dict: + d["tools"] = [] + return d + + initial_response = _sanitize_response_dict( + ResponsesResponse.from_request( + request, + sampling_params, + model_name=model_name, + created_time=created_time, + output=[], + status="in_progress", + usage=None, + ).model_dump() + ) + yield _send_event( + openai_responses_types.ResponseCreatedEvent( + type="response.created", + sequence_number=-1, + response=initial_response, + ) + ) + yield _send_event( + openai_responses_types.ResponseInProgressEvent( + type="response.in_progress", + sequence_number=-1, + response=initial_response, + ) + ) + + chat_tools = self._response_tools_to_chat_tools(request) + is_required = request.tool_choice == "required" + tool_parser: Optional[Union[FunctionCallParser, JsonArrayParser]] = None + if chat_tools and request.tool_choice != "none": + native_supports_structural_tag = False + if self.tool_call_parser: + probe = FunctionCallParser(chat_tools, self.tool_call_parser) + native_supports_structural_tag = ( + probe.detector.supports_structural_tag() + ) + if is_required and not native_supports_structural_tag: + tool_parser = JsonArrayParser() + elif self.tool_call_parser: + tool_parser = FunctionCallParser(chat_tools, self.tool_call_parser) + reasoning_parser_obj: Optional[ReasoningParser] = None + if self.reasoning_parser: + reasoning_parser_obj = ReasoningParser( + model_type=self.reasoning_parser, + stream_reasoning=True, + force_reasoning=self._is_thinking_enabled_for_request(request), + request=request, + ) + + current_output_index = -1 + reasoning_state = { + "open": False, + "item_id": "", + "output_index": -1, + "text": "", + } + message_state = { + "open": False, + "item_id": "", + "output_index": -1, + "text": "", + } + tool_call_states: dict[int, dict[str, Any]] = {} + # Items closed during the stream, in wire order. Feeds the final + # ``response.completed`` snapshot and the stored response. + emitted_items: list = [] + + prompt_tokens = 0 + completion_tokens = 0 + cached_tokens = 0 + total_tokens_meta = 0 + reasoning_tokens_meta = 0 + finish_reason: Optional[dict[str, Any]] = None + stream_offset = 0 + incremental = self.tokenizer_manager.server_args.incremental_streaming_output + + def _open_reasoning_item() -> str: + nonlocal current_output_index + current_output_index += 1 + item_id = f"rs_{random_uuid()}" + reasoning_state.update( + open=True, item_id=item_id, output_index=current_output_index, text="" + ) + return item_id + + wants_summary = self._wants_reasoning_summary(request) + + def _close_reasoning_item(): + if not reasoning_state["open"]: + return [] + text = reasoning_state["text"] + completed_item = ResponseReasoningItem( + id=reasoning_state["item_id"], + type="reasoning", + summary=( + [ResponseReasoningSummary(type="summary_text", text=text)] + if wants_summary + else [] + ), + content=[ + ResponseReasoningTextContent(type="reasoning_text", text=text), + ], + status="completed", + ) + events: list = [] + if wants_summary: + events.append( + _send_event( + openai_responses_types.ResponseReasoningSummaryTextDoneEvent( + type="response.reasoning_summary_text.done", + item_id=reasoning_state["item_id"], + sequence_number=-1, + output_index=reasoning_state["output_index"], + summary_index=0, + text=text, + ) + ) + ) + events.append( + _send_event( + openai_responses_types.ResponseReasoningSummaryPartDoneEvent( + type="response.reasoning_summary_part.done", + item_id=reasoning_state["item_id"], + sequence_number=-1, + output_index=reasoning_state["output_index"], + summary_index=0, + part=ResponseReasoningSummaryDonePart( + type="summary_text", text=text + ), + ) + ) + ) + else: + events.append( + _send_event( + openai_responses_types.ResponseReasoningTextDoneEvent( + type="response.reasoning_text.done", + item_id=reasoning_state["item_id"], + sequence_number=-1, + output_index=reasoning_state["output_index"], + content_index=0, + text=text, + ) + ) + ) + events += [ + _send_event( + openai_responses_types.ResponseOutputItemDoneEvent( + type="response.output_item.done", + sequence_number=-1, + output_index=reasoning_state["output_index"], + item=completed_item, + ) + ), + ] + emitted_items.append(completed_item) + reasoning_state["open"] = False + return events + + def _open_message_item() -> str: + nonlocal current_output_index + current_output_index += 1 + item_id = f"msg_{random_uuid()}" + message_state.update( + open=True, item_id=item_id, output_index=current_output_index, text="" + ) + return item_id + + def _close_message_item(): + if not message_state["open"]: + return [] + text = message_state["text"] + text_content = openai_responses_types.ResponseOutputText( + type="output_text", text=text, annotations=[], logprobs=None + ) + completed_item = ResponseOutputMessage( + id=message_state["item_id"], + type="message", + role="assistant", + content=[text_content], + status="completed", + ) + events = [ + _send_event( + openai_responses_types.ResponseTextDoneEvent( + type="response.output_text.done", + sequence_number=-1, + output_index=message_state["output_index"], + content_index=0, + text=text, + logprobs=[], + item_id=message_state["item_id"], + ) + ), + _send_event( + openai_responses_types.ResponseContentPartDoneEvent( + type="response.content_part.done", + sequence_number=-1, + item_id=message_state["item_id"], + output_index=message_state["output_index"], + content_index=0, + part=text_content, + ) + ), + _send_event( + openai_responses_types.ResponseOutputItemDoneEvent( + type="response.output_item.done", + sequence_number=-1, + output_index=message_state["output_index"], + item=completed_item, + ) + ), + ] + emitted_items.append(completed_item) + message_state["open"] = False + return events + + def _close_tool_call_state(tool_index: int): + state = tool_call_states.get(tool_index) + if state is None or state.get("done"): + return [] + arguments = state["arguments"] + completed_item = ResponseFunctionToolCall( + arguments=arguments, + call_id=state["call_id"], + name=state["name"] or "", + type="function_call", + id=state["item_id"], + status="completed", + ) + events = [ + _send_event( + openai_responses_types.ResponseFunctionCallArgumentsDoneEvent( + type="response.function_call_arguments.done", + sequence_number=-1, + item_id=state["item_id"], + output_index=state["output_index"], + arguments=arguments, + name=state["name"] or "", + ) + ), + _send_event( + openai_responses_types.ResponseOutputItemDoneEvent( + type="response.output_item.done", + sequence_number=-1, + output_index=state["output_index"], + item=completed_item, + ) + ), + ] + emitted_items.append(completed_item) + state["done"] = True + return events + + try: + async for ctx in result_generator: + if isinstance(ctx, dict): + chunk = ctx + else: + chunk = getattr(ctx, "last_output", None) + if not isinstance(chunk, dict): + continue + meta = chunk.get("meta_info") or {} + prompt_tokens = meta.get("prompt_tokens", prompt_tokens) + completion_tokens = meta.get("completion_tokens", completion_tokens) + cached_tokens = meta.get("cached_tokens", cached_tokens) + total_tokens_meta = meta.get("total_tokens", total_tokens_meta) + reasoning_tokens_meta = meta.get( + "reasoning_tokens", reasoning_tokens_meta + ) + finish_reason = meta.get("finish_reason") or finish_reason + + text = chunk.get("text", "") or "" + if incremental: + delta = text + else: + delta = text[stream_offset:] + stream_offset = len(text) + if not delta and finish_reason is None: + continue + + if reasoning_parser_obj is not None: + reasoning_chunk, delta = reasoning_parser_obj.parse_stream_chunk( + delta + ) + else: + reasoning_chunk = None + + if reasoning_chunk: + if message_state["open"]: + for ev in _close_message_item(): + yield ev + if not reasoning_state["open"]: + item_id = _open_reasoning_item() + yield _send_event( + openai_responses_types.ResponseOutputItemAddedEvent( + type="response.output_item.added", + sequence_number=-1, + output_index=reasoning_state["output_index"], + item=ResponseReasoningItem( + id=item_id, + type="reasoning", + summary=[], + content=[], + status="in_progress", + ), + ) + ) + # Clients that opt into ``reasoning.summary`` render + # off the ``reasoning_summary_text.*`` event stream, + # so mirror the trace into a summary part. + if wants_summary: + yield _send_event( + openai_responses_types.ResponseReasoningSummaryPartAddedEvent( + type="response.reasoning_summary_part.added", + item_id=item_id, + output_index=reasoning_state["output_index"], + summary_index=0, + part=ResponseReasoningSummaryAddedPart( + type="summary_text", text="" + ), + sequence_number=-1, + ) + ) + reasoning_state["text"] += reasoning_chunk + if wants_summary: + yield _send_event( + openai_responses_types.ResponseReasoningSummaryTextDeltaEvent( + type="response.reasoning_summary_text.delta", + item_id=reasoning_state["item_id"], + output_index=reasoning_state["output_index"], + summary_index=0, + delta=reasoning_chunk, + sequence_number=-1, + ) + ) + else: + yield _send_event( + openai_responses_types.ResponseReasoningTextDeltaEvent( + type="response.reasoning_text.delta", + item_id=reasoning_state["item_id"], + output_index=reasoning_state["output_index"], + content_index=0, + delta=reasoning_chunk, + sequence_number=-1, + ) + ) + + if not delta: + continue + + if isinstance(tool_parser, JsonArrayParser): + sp = tool_parser.parse_streaming_increment(delta, chat_tools) + normal_text, tool_calls = sp.normal_text or "", sp.calls + elif tool_parser is not None: + normal_text, tool_calls = tool_parser.parse_stream_chunk(delta) + else: + normal_text, tool_calls = delta, [] + + # Close any open tool-call item before opening a message so + # ``output_item.done`` lands before the next ``added``. + if normal_text: + if reasoning_state["open"]: + for ev in _close_reasoning_item(): + yield ev + for tool_index in list(tool_call_states): + for ev in _close_tool_call_state(tool_index): + yield ev + if not message_state["open"]: + item_id = _open_message_item() + yield _send_event( + openai_responses_types.ResponseOutputItemAddedEvent( + type="response.output_item.added", + sequence_number=-1, + output_index=message_state["output_index"], + item=ResponseOutputMessage( + id=item_id, + type="message", + role="assistant", + content=[], + status="in_progress", + ), + ) + ) + yield _send_event( + openai_responses_types.ResponseContentPartAddedEvent( + type="response.content_part.added", + sequence_number=-1, + output_index=message_state["output_index"], + item_id=message_state["item_id"], + content_index=0, + part=openai_responses_types.ResponseOutputText( + type="output_text", + text="", + annotations=[], + logprobs=None, + ), + ) + ) + message_state["text"] += normal_text + yield _send_event( + openai_responses_types.ResponseTextDeltaEvent( + type="response.output_text.delta", + sequence_number=-1, + content_index=0, + output_index=message_state["output_index"], + item_id=message_state["item_id"], + delta=normal_text, + logprobs=[], + ) + ) + + if not tool_calls: + continue + + if reasoning_state["open"]: + for ev in _close_reasoning_item(): + yield ev + if message_state["open"]: + for ev in _close_message_item(): + yield ev + + for call in tool_calls: + tool_index = call.tool_index + state = tool_call_states.get(tool_index) + if state is None or state.get("done"): + current_output_index += 1 + item_id = f"fc_{random_uuid()[:8]}" + call_id = f"call_{random_uuid()[:24]}" + state = { + "item_id": item_id, + "call_id": call_id, + "output_index": current_output_index, + "name": call.name or "", + "arguments": "", + "added": False, + "done": False, + } + tool_call_states[tool_index] = state + if not state["added"]: + state["added"] = True + # Capture ``call.name`` before the ``added`` event so + # the name is set on the first emitted item. + if call.name and not state["name"]: + state["name"] = call.name + yield _send_event( + openai_responses_types.ResponseOutputItemAddedEvent( + type="response.output_item.added", + sequence_number=-1, + output_index=state["output_index"], + item=ResponseFunctionToolCall( + arguments="", + call_id=state["call_id"], + name=state["name"], + type="function_call", + id=state["item_id"], + status="in_progress", + ), + ) + ) + if call.parameters: + state["arguments"] += call.parameters + yield _send_event( + openai_responses_types.ResponseFunctionCallArgumentsDeltaEvent( + type="response.function_call_arguments.delta", + sequence_number=-1, + item_id=state["item_id"], + output_index=state["output_index"], + delta=call.parameters, + ) + ) + except Exception: + logger.exception("Error while streaming /v1/responses") + failed = _sanitize_response_dict( + ResponsesResponse.from_request( + request, + sampling_params, + model_name=model_name, + created_time=created_time, + output=[], + status="failed", + usage=None, + ).model_dump() + ) + yield _send_event( + openai_responses_types.ResponseFailedEvent( + type="response.failed", + sequence_number=-1, + response=failed, + ) + ) + return + + for ev in _close_reasoning_item(): + yield ev + for ev in _close_message_item(): + yield ev + for tool_index in list(tool_call_states): + for ev in _close_tool_call_state(tool_index): + yield ev + + final_output_items = list(emitted_items) + + usage = UsageInfo( + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + total_tokens=total_tokens_meta or (prompt_tokens + completion_tokens), + reasoning_tokens=reasoning_tokens_meta, + ) + if self.enable_prompt_tokens_details and cached_tokens: + usage.prompt_tokens_details = PromptTokenUsageInfo( + cached_tokens=cached_tokens + ) + request_metadata.final_usage_info = usage + + final_response = ResponsesResponse.from_request( + request, + sampling_params, + model_name=model_name, + created_time=created_time, + output=final_output_items, + status="completed", + usage=usage, + ) + if request.store: + async with self.response_store_lock: + stored = self.response_store.get(final_response.id) + if stored is None or stored.status != "cancelled": + self.response_store[final_response.id] = final_response + + response_dict = _sanitize_response_dict(final_response.model_dump()) + if response_dict.get("usage"): + usage_info = response_dict["usage"] + response_dict["usage"] = { + "input_tokens": usage_info.get("prompt_tokens", 0), + "input_tokens_details": { + "cached_tokens": cached_tokens, + }, + "output_tokens": usage_info.get("completion_tokens", 0), + "output_tokens_details": { + "reasoning_tokens": reasoning_tokens_meta, + }, + "total_tokens": usage_info.get("total_tokens", 0), + } + + yield _send_event( + openai_responses_types.ResponseCompletedEvent( + type="response.completed", + sequence_number=-1, + response=response_dict, + ) + ) + async def _generate_with_builtin_tools( self, request_id: str, diff --git a/python/sglang/srt/parser/jinja_template_utils.py b/python/sglang/srt/parser/jinja_template_utils.py index dd2b1397f..c187219dc 100644 --- a/python/sglang/srt/parser/jinja_template_utils.py +++ b/python/sglang/srt/parser/jinja_template_utils.py @@ -157,14 +157,16 @@ def process_content_for_template_format( if isinstance(chunk, dict): chunk_type = chunk.get("type") - if chunk_type == "image_url": + if chunk_type in ("image_url", "input_image"): image_obj = chunk.get("image_url") or {} + if isinstance(image_obj, str): + image_obj = {"url": image_obj, "detail": chunk.get("detail")} mdp = image_obj.get("max_dynamic_patch", None) # Also allow flat style: chunk["max_dynamic_patch"] image_data.append( ImageData( url=image_obj["url"], - detail=image_obj.get("detail", "auto"), + detail=image_obj.get("detail") or "auto", max_dynamic_patch=mdp, ) ) @@ -194,13 +196,15 @@ def process_content_for_template_format( audio_data.append(chunk["audio_url"]["url"]) # Normalize to simple 'audio' type processed_content_parts.append({"type": "audio"}) - elif chunk_type == "text": + elif chunk_type in ("text", "input_text"): # For v32 encoding, collect text parts separately if use_dpsk_v32_encoding: text_parts.append(chunk["text"]) else: # Keep text content as-is for openai format - processed_content_parts.append(chunk) + processed_content_parts.append( + {"type": "text", "text": chunk["text"]} + ) elif chunk_type == "tool_reference": # GLM-specific extension: pass through so the chat template # can match tool_reference.name against tools[*].function.name @@ -220,7 +224,7 @@ def process_content_for_template_format( # String format: flatten to text only (for templates like DeepSeek) text_parts = [] for chunk in msg_dict["content"]: - if isinstance(chunk, dict) and chunk.get("type") == "text": + if isinstance(chunk, dict) and chunk.get("type") in ("text", "input_text"): text_parts.append(chunk["text"]) # Note: For string format, we ignore images/audio since the template # doesn't expect structured content - multimodal placeholders would diff --git a/python/sglang/srt/parser/reasoning_parser.py b/python/sglang/srt/parser/reasoning_parser.py index b36a8fa8f..d4fe2eacf 100644 --- a/python/sglang/srt/parser/reasoning_parser.py +++ b/python/sglang/srt/parser/reasoning_parser.py @@ -257,6 +257,9 @@ class Qwen3Detector(BaseReasoningFormatDetector): think_excluded_tokens=think_excluded_tokens, force_reasoning=force_reasoning, stream_reasoning=stream_reasoning, + # Qwen3.5 sometimes opens ```` without closing + # ````; treat it as an implicit reasoning close. + tool_start_token="", continue_final_message=continue_final_message, previous_content=previous_content, thinks_internally=True, diff --git a/test/registered/unit/entrypoints/openai/test_responses_protocol.py b/test/registered/unit/entrypoints/openai/test_responses_protocol.py new file mode 100644 index 000000000..19a11ad13 --- /dev/null +++ b/test/registered/unit/entrypoints/openai/test_responses_protocol.py @@ -0,0 +1,146 @@ +import unittest + +from utils import make_serving # noqa: F401 — bootstrap import + +from sglang.srt.entrypoints.openai.protocol import ResponsesRequest, UsageInfo +from sglang.test.ci.ci_register import register_cpu_ci + +register_cpu_ci(est_time=4, suite="base-a-test-cpu") + + +class ResponsesRequestTestCase(unittest.TestCase): + def test_function_tool_accepted(self): + request = ResponsesRequest( + model="x", + input="call the tool", + tools=[ + { + "type": "function", + "name": "lookup", + "description": "Look up a value.", + "parameters": { + "type": "object", + "properties": {"key": {"type": "string"}}, + "required": ["key"], + }, + "strict": True, + } + ], + store=False, + ) + + self.assertEqual(request.tools[0].type, "function") + self.assertEqual(request.tools[0].name, "lookup") + self.assertTrue(request.tools[0].strict) + + def test_function_tool_requires_name(self): + with self.assertRaises(ValueError): + ResponsesRequest( + model="x", input="hi", tools=[{"type": "function"}], store=False + ) + with self.assertRaises(ValueError): + ResponsesRequest( + model="x", + input="hi", + tools=[{"type": "function", "name": ""}], + store=False, + ) + + def test_extended_tool_types_accepted(self): + for tool_type in ( + "web_search", + "web_search_preview", + "code_interpreter", + "file_search", + "image_generation", + "computer_use_preview", + "local_shell", + "mcp", + "custom", + "namespace", + ): + request = ResponsesRequest( + model="x", + input="hi", + tools=[{"type": tool_type}], + store=False, + ) + self.assertEqual(request.tools[0].type, tool_type) + + def test_namespace_tool_carries_inner_tools_list(self): + request = ResponsesRequest( + model="x", + input="hi", + tools=[ + { + "type": "namespace", + "name": "codex", + "tools": [ + {"type": "function", "name": "apply_patch"}, + {"type": "function", "name": "shell"}, + ], + } + ], + store=False, + ) + self.assertEqual(request.tools[0].type, "namespace") + self.assertEqual(len(request.tools[0].tools), 2) + self.assertEqual(request.tools[0].tools[0]["name"], "apply_patch") + + +class ResponsesSamplingParamsTestCase(unittest.TestCase): + def test_processed_stop_and_tool_constraint_propagate(self): + request = ResponsesRequest(model="x", input="call the tool", store=False) + params = request.to_sampling_params( + default_max_tokens=128, + default_params={}, + stop=[""], + tool_call_constraint=("json_schema", {"type": "object"}), + ) + self.assertEqual(params["stop"], [""]) + self.assertEqual(params["json_schema"], '{"type": "object"}') + + def test_constraint_conflict_raises(self): + request = ResponsesRequest(model="x", input="hi", store=False) + with self.assertRaises(ValueError): + request.to_sampling_params( + default_max_tokens=128, + default_params={"json_schema": '{"type": "object"}'}, + tool_call_constraint=("json_schema", {"type": "object"}), + ) + + def test_structural_tag_with_model_dump(self): + class _FakeStructuralTag: + def model_dump(self, by_alias=False): + return {"type": "structural_tag"} + + request = ResponsesRequest(model="x", input="hi", store=False) + params = request.to_sampling_params( + default_max_tokens=128, + default_params={}, + tool_call_constraint=("structural_tag", _FakeStructuralTag()), + ) + self.assertEqual(params["structural_tag"], '{"type": "structural_tag"}') + + +class ResponsesResponseFromRequestTestCase(unittest.TestCase): + def test_parallel_tool_calls_false_preserved(self): + from sglang.srt.entrypoints.openai.protocol import ResponsesResponse + + request = ResponsesRequest( + model="x", input="hi", parallel_tool_calls=False, store=False + ) + response = ResponsesResponse.from_request( + request, + sampling_params={}, + model_name="x", + created_time=0, + output=[], + status="completed", + usage=UsageInfo(prompt_tokens=1, completion_tokens=1, total_tokens=2), + ) + self.assertFalse(response.parallel_tool_calls) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/registered/unit/entrypoints/openai/test_serving_responses.py b/test/registered/unit/entrypoints/openai/test_serving_responses.py new file mode 100644 index 000000000..8a410cbbf --- /dev/null +++ b/test/registered/unit/entrypoints/openai/test_serving_responses.py @@ -0,0 +1,529 @@ +import asyncio +import unittest +from unittest.mock import Mock, patch + +from openai.types.responses import ( + ResponseOutputMessage, + ResponseOutputText, + ResponseReasoningItem, +) +from openai.types.responses.response_function_tool_call import ResponseFunctionToolCall +from utils import make_serving + +from sglang.srt.entrypoints.context import SimpleContext +from sglang.srt.entrypoints.openai.protocol import ( + MessageProcessingResult, + RequestResponseMetadata, + ResponsesRequest, +) +from sglang.srt.entrypoints.openai.serving_responses import OpenAIServingResponses +from sglang.srt.function_call.core_types import ToolCallItem +from sglang.test.ci.ci_register import register_cpu_ci + +register_cpu_ci(est_time=8, suite="base-a-test-cpu") + + +class InputMessageConstructionTestCase(unittest.TestCase): + def test_previous_response_replays_assistant_text_not_instructions(self): + serving = make_serving() + prev_response = Mock(id="resp_prev") + prev_response.output = [ + ResponseReasoningItem( + id="rs_prev", summary=[], type="reasoning", content=None, status=None + ), + ResponseOutputMessage( + id="msg_prev", + content=[ + ResponseOutputText( + text="first answer part", + annotations=[], + type="output_text", + logprobs=None, + ), + ResponseOutputText( + text="second answer part", + annotations=[], + type="output_text", + logprobs=None, + ), + ], + role="assistant", + status="completed", + type="message", + ), + ] + serving.msg_store["resp_prev"] = [{"role": "user", "content": "old input"}] + + request = ResponsesRequest( + model="x", + instructions="Be brief", + previous_response_id="resp_prev", + input="new input", + store=False, + ) + + messages = serving._construct_input_messages(request, prev_response) + + self.assertEqual( + messages, + [ + {"role": "system", "content": "Be brief"}, + {"role": "user", "content": "old input"}, + { + "role": "assistant", + "content": "first answer part\nsecond answer part", + }, + {"role": "user", "content": "new input"}, + ], + ) + + def test_input_parts_normalized_for_chat_templates(self): + serving = make_serving() + request = ResponsesRequest( + model="x", + input=[ + { + "role": "user", + "content": [ + {"type": "input_text", "text": "what is this?"}, + { + "type": "input_image", + "image_url": "http://example.com/cat.png", + }, + ], + } + ], + store=False, + ) + + messages = serving._construct_input_messages(request) + + self.assertEqual( + messages, + [ + { + "role": "user", + "content": [ + {"type": "text", "text": "what is this?"}, + { + "type": "image_url", + "image_url": { + "url": "http://example.com/cat.png", + "detail": "auto", + }, + }, + ], + } + ], + ) + + def test_previous_response_id_input_list_does_not_call_copy_module(self): + serving = make_serving() + serving.use_harmony = True + prev = Mock(id="resp_prev") + prev.output = [ + ResponseFunctionToolCall( + arguments="{}", + call_id="call_x", + name="t", + type="function_call", + id="fc_x", + status="completed", + ) + ] + request = ResponsesRequest( + model="x", + input=[{"role": "user", "content": "hi"}], + previous_response_id="resp_prev", + store=False, + ) + try: + serving._construct_input_messages_with_harmony(request, prev) + except TypeError as exc: + self.fail(f"copy() module-call regression: {exc}") + except Exception: + pass + + +class ChatToolForwardingTestCase(unittest.TestCase): + def test_make_request_passes_function_tools_to_chat_processing(self): + serving = make_serving() + seen = {} + + def fake_process(chat_request, is_multimodal): + seen["tools"] = chat_request.tools + seen["tool_choice"] = chat_request.tool_choice + seen["parallel_tool_calls"] = chat_request.parallel_tool_calls + return MessageProcessingResult( + prompt="prompt", + prompt_ids=[1, 2, 3], + image_data=None, + audio_data=None, + video_data=None, + modalities=[], + stop=[""], + tool_call_constraint=("json_schema", {"type": "object"}), + ) + + serving._process_messages = Mock(side_effect=fake_process) + request = ResponsesRequest( + model="x", + input="call the tool", + tools=[ + { + "type": "function", + "name": "lookup", + "parameters": {"type": "object"}, + } + ], + tool_choice="required", + parallel_tool_calls=False, + store=False, + ) + + messages, request_prompts, engine_prompts, processed = asyncio.run( + serving._make_request(request, None, serving.tokenizer_manager.tokenizer) + ) + + self.assertEqual(messages, [{"role": "user", "content": "call the tool"}]) + self.assertEqual(request_prompts, [[1, 2, 3]]) + self.assertEqual(engine_prompts, [[1, 2, 3]]) + self.assertEqual(seen["tools"][0].function.name, "lookup") + self.assertEqual(seen["tool_choice"], "required") + self.assertFalse(seen["parallel_tool_calls"]) + self.assertEqual(processed.tool_call_constraint[0], "json_schema") + + def test_required_tool_choice_without_function_tool_returns_400(self): + serving = make_serving() + request = ResponsesRequest( + model="x", + input="hi", + tool_choice="required", + tools=[{"type": "web_search"}, {"type": "mcp"}], + store=False, + ) + result = asyncio.run(serving.create_responses(request, raw_request=None)) + self.assertEqual(getattr(result, "status_code", None), 400) + + +class InputItemNormalizationTestCase(unittest.TestCase): + def test_function_call_becomes_assistant_tool_call(self): + normalized = OpenAIServingResponses._normalize_response_message_for_chat( + { + "type": "function_call", + "id": "fc_1", + "call_id": "call_abc", + "name": "lookup", + "arguments": '{"key": "val"}', + "status": "completed", + } + ) + self.assertEqual( + normalized, + { + "role": "assistant", + "tool_calls": [ + { + "id": "call_abc", + "type": "function", + "function": { + "name": "lookup", + "arguments": '{"key": "val"}', + }, + } + ], + }, + ) + + def test_developer_role_becomes_system(self): + normalized = OpenAIServingResponses._normalize_response_message_for_chat( + {"role": "developer", "content": "Be terse."} + ) + self.assertEqual(normalized, {"role": "system", "content": "Be terse."}) + + def test_function_call_output_becomes_tool_message(self): + normalized = OpenAIServingResponses._normalize_response_message_for_chat( + { + "type": "function_call_output", + "call_id": "call_abc", + "output": "42", + } + ) + self.assertEqual( + normalized, + {"role": "tool", "tool_call_id": "call_abc", "content": "42"}, + ) + + def test_unknown_input_item_type_raises(self): + with self.assertRaises(ValueError): + OpenAIServingResponses._normalize_response_message_for_chat( + {"type": "web_search_call", "id": "ws_1"} + ) + + +class FullResponseUsageTestCase(unittest.TestCase): + def test_full_response_uses_dict_meta_info_for_usage(self): + serving = make_serving() + context = SimpleContext() + context.last_output = { + "text": "done", + "meta_info": { + "prompt_tokens": 11, + "completion_tokens": 7, + "cached_tokens": 3, + "reasoning_tokens": 2, + }, + } + request = ResponsesRequest( + model="x", input="hello", request_id="resp_usage", store=False + ) + metadata = RequestResponseMetadata(request_id=request.request_id) + + async def empty_generator(): + if False: + yield None + + response = asyncio.run( + serving.responses_full_generator( + request, + sampling_params={}, + result_generator=empty_generator(), + context=context, + model_name="x", + tokenizer=serving.tokenizer_manager.tokenizer, + request_metadata=metadata, + created_time=123, + ) + ) + + self.assertEqual(response.usage.prompt_tokens, 11) + self.assertEqual(response.usage.completion_tokens, 7) + self.assertEqual(response.usage.reasoning_tokens, 2) + self.assertEqual(metadata.final_usage_info, response.usage) + + +class MultimodalRequestTestCase(unittest.TestCase): + def test_multimodal_create_responses_sends_text_and_media_to_engine(self): + serving = make_serving(is_multimodal=True) + captured = {} + + serving._process_messages = Mock( + return_value=MessageProcessingResult( + prompt="rendered multimodal prompt", + prompt_ids=[9, 9, 9], + image_data=["http://example.com/cat.png"], + audio_data=None, + video_data=None, + modalities=["image"], + stop=[], + ) + ) + + async def fake_generate( + request_id, + request_prompt, + adapted_request, + sampling_params, + context, + **kwargs, + ): + captured["request_prompt"] = request_prompt + captured["adapted_request"] = adapted_request + context.append_output( + { + "text": "looks like a cat", + "meta_info": { + "prompt_tokens": 5, + "completion_tokens": 4, + "cached_tokens": 0, + }, + } + ) + yield context + + serving._generate_with_builtin_tools = fake_generate + request = ResponsesRequest( + model="x", + input=[ + { + "role": "user", + "content": [ + {"type": "input_text", "text": "describe it"}, + { + "type": "input_image", + "image_url": "http://example.com/cat.png", + }, + ], + } + ], + request_id="resp_mm", + store=False, + ) + + response = asyncio.run(serving.create_responses(request)) + + self.assertEqual(response.status, "completed") + self.assertEqual(captured["request_prompt"], "rendered multimodal prompt") + self.assertEqual(captured["adapted_request"].text, "rendered multimodal prompt") + self.assertIsNone(captured["adapted_request"].input_ids) + self.assertEqual( + captured["adapted_request"].image_data, ["http://example.com/cat.png"] + ) + self.assertEqual(captured["adapted_request"].modalities, ["image"]) + + +class OutputItemsTestCase(unittest.TestCase): + def _function_tool_request(self): + return ResponsesRequest( + model="x", + input="weather?", + store=False, + tools=[ + { + "type": "function", + "name": "get_weather", + "description": "Get weather", + "parameters": {"type": "object"}, + } + ], + ) + + def test_function_tool_call_extracted_via_parser(self): + serving = make_serving() + serving.tool_call_parser = "qwen3_coder" + fake_call = ToolCallItem( + tool_index=0, name="get_weather", parameters='{"city": "Beijing"}' + ) + + with patch( + "sglang.srt.entrypoints.openai.serving_responses.FunctionCallParser" + ) as parser_cls: + instance = parser_cls.return_value + instance.has_tool_call.return_value = True + instance.parse_non_stream.return_value = ("trailing text", [fake_call]) + output_items = serving._make_response_output_items( + self._function_tool_request(), + "raw model output with ", + tokenizer=Mock(), + ) + + tool_calls = [ + item for item in output_items if isinstance(item, ResponseFunctionToolCall) + ] + self.assertEqual(len(tool_calls), 1) + self.assertEqual(tool_calls[0].name, "get_weather") + self.assertEqual(tool_calls[0].arguments, '{"city": "Beijing"}') + + message_items = [ + item for item in output_items if isinstance(item, ResponseOutputMessage) + ] + self.assertEqual(len(message_items), 1) + self.assertEqual(message_items[0].content[0].text, "trailing text") + + def test_prose_emitted_before_tool_call_item(self): + serving = make_serving() + serving.tool_call_parser = "qwen3_coder" + fake_call = ToolCallItem( + tool_index=0, name="get_weather", parameters='{"city": "Beijing"}' + ) + + with patch( + "sglang.srt.entrypoints.openai.serving_responses.FunctionCallParser" + ) as parser_cls: + instance = parser_cls.return_value + instance.has_tool_call.return_value = True + instance.parse_non_stream.return_value = ( + "I'll check the weather.", + [fake_call], + ) + output_items = serving._make_response_output_items( + self._function_tool_request(), "raw model output", tokenizer=Mock() + ) + + types = [type(item).__name__ for item in output_items] + self.assertEqual(types, ["ResponseOutputMessage", "ResponseFunctionToolCall"]) + + def test_required_tool_choice_parses_json_array_without_native_parser(self): + serving = make_serving() + serving.tool_call_parser = None + request = ResponsesRequest( + model="x", + input="hi", + tool_choice="required", + tools=[ + { + "type": "function", + "name": "get_weather", + "parameters": {"type": "object"}, + } + ], + store=False, + ) + raw = '[{"name": "get_weather", "parameters": {"city": "Beijing"}}]' + + output_items = serving._make_response_output_items( + request, raw, tokenizer=Mock() + ) + + tool_calls = [ + item for item in output_items if isinstance(item, ResponseFunctionToolCall) + ] + self.assertEqual(len(tool_calls), 1) + self.assertEqual(tool_calls[0].name, "get_weather") + self.assertEqual(tool_calls[0].arguments, '{"city": "Beijing"}') + self.assertEqual( + [item for item in output_items if isinstance(item, ResponseOutputMessage)], + [], + ) + + def test_no_tool_call_extraction_when_tool_choice_none(self): + serving = make_serving() + serving.tool_call_parser = "qwen3_coder" + request = ResponsesRequest( + model="x", + input="hi", + store=False, + tool_choice="none", + tools=[ + { + "type": "function", + "name": "get_weather", + "parameters": {"type": "object"}, + } + ], + ) + + with patch( + "sglang.srt.entrypoints.openai.serving_responses.FunctionCallParser" + ) as parser_cls: + output_items = serving._make_response_output_items( + request, "just a plain answer", tokenizer=Mock() + ) + parser_cls.assert_not_called() + + self.assertEqual(len(output_items), 1) + self.assertIsInstance(output_items[0], ResponseOutputMessage) + + +class HarmonyResponsesTestCase(unittest.TestCase): + def test_developer_message_skips_unsupported_tool_types(self): + from sglang.srt.entrypoints.harmony_utils import get_developer_message + from sglang.srt.entrypoints.openai.protocol import ResponseTool + + tools = [ + ResponseTool( + type="function", + name="get_weather", + description="Look up weather.", + parameters={"type": "object"}, + ), + ResponseTool(type="web_search"), + ResponseTool(type="namespace", name="codex"), + ResponseTool(type="mcp"), + ] + msg = get_developer_message(instructions="be helpful", tools=tools) + self.assertIsNotNone(msg) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/registered/unit/entrypoints/openai/test_serving_responses_stream.py b/test/registered/unit/entrypoints/openai/test_serving_responses_stream.py new file mode 100644 index 000000000..d62cb1629 --- /dev/null +++ b/test/registered/unit/entrypoints/openai/test_serving_responses_stream.py @@ -0,0 +1,211 @@ +import asyncio +import unittest +from unittest.mock import Mock, patch + +from utils import ( + collect_stream_events, + event_payloads, + event_types, + find_completed_event, + make_serving, +) + +from sglang.srt.entrypoints.openai.protocol import ( + RequestResponseMetadata, + ResponsesRequest, +) +from sglang.test.ci.ci_register import register_cpu_ci + +register_cpu_ci(est_time=4, suite="base-a-test-cpu") + + +class _StreamFixture: + def __init__(self, serving, request): + self.serving = serving + self.request = request + self.request_metadata = RequestResponseMetadata(request_id=request.request_id) + + def run(self, chunks): + async def gen(): + for ch in chunks: + yield ch + + async def collect(): + return await collect_stream_events( + self.serving.responses_stream_generator_non_harmony( + self.request, + sampling_params={}, + result_generator=gen(), + model_name="x", + tokenizer=Mock(), + request_metadata=self.request_metadata, + ) + ) + + return asyncio.run(collect()) + + +def _engine_chunk(text, completion_tokens, *, finish=False): + return { + "text": text, + "meta_info": { + "id": "rid", + "prompt_tokens": 5, + "completion_tokens": completion_tokens, + "cached_tokens": 0, + "reasoning_tokens": 0, + "finish_reason": {"type": "stop"} if finish else None, + }, + } + + +class NonHarmonyStreamTestCase(unittest.TestCase): + def test_emits_typed_sse_events_in_order(self): + serving = make_serving() + serving.reasoning_parser = None + serving.tool_call_parser = None + + request = ResponsesRequest(model="x", input="hi", stream=True, store=False) + fixture = _StreamFixture(serving, request) + events = fixture.run( + [ + _engine_chunk("Hel", 1), + _engine_chunk("Hello", 2), + _engine_chunk("Hello world", 4, finish=True), + ] + ) + + types = event_types(events) + self.assertEqual(types[0], "response.created") + self.assertEqual(types[1], "response.in_progress") + for ev in ( + "response.output_item.added", + "response.content_part.added", + "response.output_text.delta", + "response.output_text.done", + "response.content_part.done", + "response.output_item.done", + ): + self.assertIn(ev, types) + self.assertEqual(types[-1], "response.completed") + + seqs = [p["sequence_number"] for p in event_payloads(events)] + self.assertEqual(seqs, list(range(len(seqs)))) + + def test_required_tool_choice_emits_function_call_events(self): + serving = make_serving() + serving.reasoning_parser = None + serving.tool_call_parser = None + + request = ResponsesRequest( + model="x", + input="hi", + stream=True, + store=False, + tool_choice="required", + tools=[ + { + "type": "function", + "name": "get_weather", + "parameters": {"type": "object"}, + } + ], + ) + payload = '[{"name": "get_weather", "parameters": {"city": "Beijing"}}]' + + chunks = [] + sent = 0 + while sent < len(payload): + sent += min(8, len(payload) - sent) + chunks.append( + _engine_chunk(payload[:sent], sent, finish=sent == len(payload)) + ) + + fixture = _StreamFixture(serving, request) + events = fixture.run(chunks) + types = event_types(events) + + self.assertIn("response.function_call_arguments.delta", types) + self.assertIn("response.function_call_arguments.done", types) + self.assertIn("response.output_item.added", types) + self.assertIn("response.output_item.done", types) + self.assertNotIn("response.output_text.delta", types) + + added_kinds = [ + payload["item"]["type"] + for payload in event_payloads(events) + if payload.get("type") == "response.output_item.added" + ] + self.assertIn("function_call", added_kinds) + + def test_final_output_preserves_text_tool_text_order(self): + from sglang.srt.function_call.core_types import ( + StreamingParseResult, + ToolCallItem, + ) + + serving = make_serving() + serving.reasoning_parser = None + serving.tool_call_parser = "qwen3_coder" + + request = ResponsesRequest( + model="x", + input="hi", + stream=True, + store=False, + tools=[ + { + "type": "function", + "name": "get_weather", + "parameters": {"type": "object"}, + } + ], + ) + + scripted = [ + StreamingParseResult(normal_text="I'll check.", calls=[]), + StreamingParseResult( + normal_text="", + calls=[ + ToolCallItem( + tool_index=0, + name="get_weather", + parameters='{"city": "Beijing"}', + ) + ], + ), + StreamingParseResult(normal_text="It's sunny.", calls=[]), + ] + chunks = [ + _engine_chunk(" " * 3, 3), + _engine_chunk(" " * 10, 10), + _engine_chunk(" " * 14, 14, finish=True), + ] + + script_iter = iter(scripted) + + def fake_parse_stream_chunk(delta): + sp = next(script_iter) + return sp.normal_text, sp.calls + + with patch( + "sglang.srt.entrypoints.openai.serving_responses.FunctionCallParser" + ) as parser_cls: + parser_cls.return_value.detector.supports_structural_tag.return_value = True + parser_cls.return_value.parse_stream_chunk.side_effect = ( + fake_parse_stream_chunk + ) + fixture = _StreamFixture(serving, request) + events = fixture.run(chunks) + + completed = find_completed_event(events) + output = completed["response"]["output"] + kinds = [item["type"] for item in output] + self.assertEqual(kinds, ["message", "function_call", "message"]) + self.assertEqual(output[0]["content"][0]["text"], "I'll check.") + self.assertEqual(output[1]["name"], "get_weather") + self.assertEqual(output[2]["content"][0]["text"], "It's sunny.") + + +if __name__ == "__main__": + unittest.main() diff --git a/test/registered/unit/entrypoints/openai/utils.py b/test/registered/unit/entrypoints/openai/utils.py new file mode 100644 index 000000000..8f2d8872a --- /dev/null +++ b/test/registered/unit/entrypoints/openai/utils.py @@ -0,0 +1,109 @@ +"""Stub CUDA-only deps before importing sglang.srt serving modules. Must +be imported first by every /v1/responses test that runs on CPU.""" + +try: + import torch + + _ORIGINAL_TORCH_COMPILE = torch.compile + + def _identity_compile(fn=None, **kwargs): + if fn is None: + return lambda inner_fn: inner_fn + return fn + + torch.compile = _identity_compile +except ImportError: + torch = None + _ORIGINAL_TORCH_COMPILE = None + +from sglang.test.test_utils import maybe_stub_sgl_kernel + +maybe_stub_sgl_kernel() + +import json +from typing import AsyncIterator +from unittest.mock import Mock + +from sglang.srt.entrypoints.openai.serving_responses import OpenAIServingResponses +from sglang.test.ci.ci_register import register_cpu_ci + +register_cpu_ci( + est_time=0, + suite="base-a-test-cpu", + disabled="helper module — exported fixtures, not a test", +) + +if torch is not None: + torch.compile = _ORIGINAL_TORCH_COMPILE + + +class MockTokenizerManager: + def __init__(self, *, is_multimodal: bool = False): + self.model_config = Mock(is_multimodal=is_multimodal, context_len=4096) + self.model_config.get_default_sampling_params.return_value = {} + self.model_config.hf_config = Mock( + model_type="llama", architectures=["LlamaForCausalLM"] + ) + self.server_args = Mock( + enable_cache_report=False, + reasoning_parser=None, + stream_response_default_include_usage=False, + tokenizer_metrics_allowed_custom_labels=None, + tool_call_parser=None, + incremental_streaming_output=False, + ) + self.tokenizer = Mock() + self.tokenizer.encode.return_value = [1, 2, 3] + self.tokenizer.chat_template = None + self.tokenizer.bos_token_id = 1 + self.num_reserved_tokens = 0 + self.generate_request = Mock() + self.create_abort_task = Mock() + + +class MockTemplateManager: + def __init__(self): + self.chat_template_name = "llama-3" + self.jinja_template_content_format = None + self.completion_template_name = None + self.reasoning_config = None + self.force_reasoning = False + + +def make_serving(*, is_multimodal: bool = False) -> OpenAIServingResponses: + return OpenAIServingResponses( + MockTokenizerManager(is_multimodal=is_multimodal), MockTemplateManager() + ) + + +async def collect_stream_events(stream: AsyncIterator[str]) -> list[str]: + events = [] + async for chunk in stream: + events.append(chunk) + return events + + +def event_types(events: list[str]) -> list[str]: + return [ + line[len("event: ") :].strip() + for chunk in events + for line in chunk.splitlines() + if line.startswith("event: ") + ] + + +def event_payloads(events: list[str]) -> list[dict]: + return [ + json.loads(line[len("data: ") :]) + for chunk in events + for line in chunk.splitlines() + if line.startswith("data: ") + ] + + +def find_completed_event(events: list[str]) -> dict: + for chunk in events: + lines = chunk.splitlines() + if lines and lines[0] == "event: response.completed": + return json.loads(lines[1][len("data: ") :]) + raise AssertionError("response.completed event missing from stream") diff --git a/test/registered/unit/parser/test_jinja_template_utils.py b/test/registered/unit/parser/test_jinja_template_utils.py index 34b2dfacc..3a25d9aca 100644 --- a/test/registered/unit/parser/test_jinja_template_utils.py +++ b/test/registered/unit/parser/test_jinja_template_utils.py @@ -383,6 +383,37 @@ class TestTemplateContentFormatDetection(CustomTestCase): # Image data is still extracted self.assertEqual(len(image_data), 1) + def test_process_content_v32_encoding_accepts_responses_input_text(self): + msg_dict = { + "role": "user", + "content": [ + {"type": "input_text", "text": "Hello"}, + { + "type": "input_image", + "image_url": "http://example.com/img.jpg", + "detail": "auto", + }, + {"type": "input_text", "text": "World"}, + ], + } + image_data = [] + video_data = [] + audio_data = [] + modalities = [] + result = process_content_for_template_format( + msg_dict, + "openai", + image_data, + video_data, + audio_data, + modalities, + use_dpsk_v32_encoding=True, + ) + + self.assertEqual(result["content"], "Hello World") + self.assertEqual(len(image_data), 1) + self.assertEqual(image_data[0].url, "http://example.com/img.jpg") + def test_process_content_invalid_format_raises(self): """Test that invalid content_format raises ValueError.""" msg_dict = {