From 3c51e29debed99757a433d5e96f36f4dabb46734 Mon Sep 17 00:00:00 2001 From: Liangsheng Yin Date: Fri, 7 Aug 2026 13:21:46 -0700 Subject: [PATCH] Responses support (#32689) Co-authored-by: github-actions[bot] Co-authored-by: Harmya Bhatt Co-authored-by: harmya Co-authored-by: Xinyuan Co-authored-by: Xinyuan Tong Co-authored-by: Xinyuan Tong <115166877+JustinTong0323@users.noreply.github.com> --- python/sglang/srt/entrypoints/context.py | 11 + .../sglang/srt/entrypoints/harmony_utils.py | 20 +- .../sglang/srt/entrypoints/openai/protocol.py | 140 +++++- .../entrypoints/openai/serving_responses.py | 440 ++++++++++++------ .../openai_server/basic/test_openai_server.py | 23 +- .../openai/test_responses_protocol.py | 278 ++++++++++- .../openai/test_serving_responses.py | 249 +++++++++- .../openai/test_serving_responses_stream.py | 186 +++++--- .../unit/entrypoints/openai/utils.py | 51 ++ 9 files changed, 1135 insertions(+), 263 deletions(-) diff --git a/python/sglang/srt/entrypoints/context.py b/python/sglang/srt/entrypoints/context.py index dd6af3f89..652bc92a4 100644 --- a/python/sglang/srt/entrypoints/context.py +++ b/python/sglang/srt/entrypoints/context.py @@ -79,6 +79,7 @@ class HarmonyContext(ConversationContext): self.num_cached_tokens = 0 self.num_output_tokens = 0 self.num_reasoning_tokens = 0 + self.finish_reason = None def append_output(self, output) -> None: if isinstance(output, dict) and "output_ids" in output: @@ -97,12 +98,20 @@ class HarmonyContext(ConversationContext): self.num_cached_tokens = meta_info["cached_tokens"] if "completion_tokens" in meta_info: self.num_output_tokens += meta_info["completion_tokens"] + self._record_finish_reason(meta_info) else: output_msgs = output self._messages.extend(output_msgs) + def _record_finish_reason(self, meta_info: dict) -> None: + # Last non-null wins: a builtin-tool continuation turn supersedes the + # reason recorded for the turn before it. + reason = meta_info.get("finish_reason") + if reason is not None: + self.finish_reason = reason + @property def messages(self) -> list: return self._messages @@ -210,6 +219,8 @@ class StreamingHarmonyContext(HarmonyContext): new_token_ids = output_token_ids self.num_processed_tokens += len(output_token_ids) + self._record_finish_reason(meta_info) + for token_id in new_token_ids: self.parser.process(token_id) diff --git a/python/sglang/srt/entrypoints/harmony_utils.py b/python/sglang/srt/entrypoints/harmony_utils.py index 33bd88783..1d1b4d8ed 100644 --- a/python/sglang/srt/entrypoints/harmony_utils.py +++ b/python/sglang/srt/entrypoints/harmony_utils.py @@ -5,7 +5,7 @@ import datetime import logging from collections.abc import Iterable -from typing import Literal, Optional, Union +from typing import Optional, Union import orjson from openai.types.responses import ( @@ -40,7 +40,10 @@ from openai_harmony import ( load_harmony_encoding, ) -from sglang.srt.entrypoints.openai.protocol import ResponseInputOutputItem +from sglang.srt.entrypoints.openai.protocol import ( + ReasoningEffortTier, + ResponseInputOutputItem, +) from sglang.srt.utils import random_uuid logger = logging.getLogger(__name__) @@ -49,6 +52,11 @@ REASONING_EFFORT = { "high": ReasoningEffort.HIGH, "medium": ReasoningEffort.MEDIUM, "low": ReasoningEffort.LOW, + # harmony has only these three, so outer tiers clamp inward. "none" is left + # out so it falls through to the harmony default. + "minimal": ReasoningEffort.LOW, + "xhigh": ReasoningEffort.HIGH, + "max": ReasoningEffort.HIGH, } _harmony_encoding = None @@ -63,7 +71,7 @@ def get_encoding(): def get_system_message( model_identity: Optional[str] = None, - reasoning_effort: Optional[Literal["high", "medium", "low"]] = None, + reasoning_effort: Optional[ReasoningEffortTier] = None, start_date: Optional[str] = None, browser_description: Optional[str] = None, python_description: Optional[str] = None, @@ -72,9 +80,9 @@ def get_system_message( if model_identity is not None: sys_msg_content = sys_msg_content.with_model_identity(model_identity) if reasoning_effort is not None: - sys_msg_content = sys_msg_content.with_reasoning_effort( - REASONING_EFFORT[reasoning_effort] - ) + effort = REASONING_EFFORT.get(reasoning_effort) + if effort is not None: + sys_msg_content = sys_msg_content.with_reasoning_effort(effort) if start_date is None: start_date = datetime.datetime.now().strftime("%Y-%m-%d") sys_msg_content = sys_msg_content.with_conversation_start_date(start_date) diff --git a/python/sglang/srt/entrypoints/openai/protocol.py b/python/sglang/srt/entrypoints/openai/protocol.py index 2352e5531..ecb458fc6 100644 --- a/python/sglang/srt/entrypoints/openai/protocol.py +++ b/python/sglang/srt/entrypoints/openai/protocol.py @@ -42,12 +42,18 @@ from openai.types.responses import ( ResponseOutputMessage, ResponseOutputText, ResponseReasoningItem, + ResponseTextConfig, ) from openai.types.responses.response import ToolChoice +from openai.types.responses.response_format_text_json_schema_config import ( + ResponseFormatTextJSONSchemaConfig, +) +from openai.types.shared.response_format_json_object import ResponseFormatJSONObject from pydantic import ( BaseModel, ConfigDict, Field, + field_serializer, field_validator, model_serializer, model_validator, @@ -710,6 +716,9 @@ class ToolChoice(BaseModel): ReasoningEffortTier = Literal[ "none", "minimal", "low", "medium", "high", "xhigh", "max" ] +# The typed /v1/responses stream events validate against OpenAI's narrower +# ``Reasoning.effort``; echoing a tier outside this set kills the stream. +ECHOABLE_REASONING_EFFORTS = frozenset({"minimal", "low", "medium", "high"}) # Chat Completions and /v1/tokenize additionally accept a fine-grained float in # [0.0, 0.99] as an sglang extension (not part of the OpenAI schema, so the # /v1/responses surface deliberately keeps the string tiers only). Single-sourced @@ -1509,7 +1518,8 @@ class ResponsesRequest(BaseModel): store: Optional[bool] = True stream: Optional[bool] = False temperature: Optional[float] = None - tool_choice: Literal["auto", "required", "none"] = "auto" + text: Optional[ResponseTextConfig] = None + tool_choice: Union[Literal["auto", "required", "none"], Dict[str, Any]] = "auto" tools: List[ResponseTool] = Field(default_factory=list) top_logprobs: Optional[int] = 0 top_p: Optional[float] = None @@ -1517,6 +1527,7 @@ class ResponsesRequest(BaseModel): user: Optional[str] = None # Extra SGLang parameters + chat_template_kwargs: Optional[Dict[str, Any]] = None request_id: str = Field( default_factory=lambda: f"resp_{uuid.uuid4().hex}", description="The request_id related to this request. If the caller does not set it, a random uuid will be generated.", @@ -1548,6 +1559,34 @@ class ResponsesRequest(BaseModel): "repetition_penalty": 1.0, } + @model_validator(mode="before") + @classmethod + def normalize_reasoning_to_thinking(cls, values): + """Turn reasoning.effort == "none" into a thinking toggle, as + ChatCompletionRequest does. Both keys are set: families differ + ("thinking" for deepseek/kimi, "enable_thinking" for qwen3/glm).""" + if not isinstance(values, dict): + return values + # mode="before", so reasoning is still raw: a dict from JSON, or a + # built param object when constructed in Python. + r = values.get("reasoning") + effort = None + if isinstance(r, dict): + effort = r.get("effort") or r.get("reasoning_effort") + elif isinstance(r, ResponseReasoningParam): + effort = r.effort + + if effort == "none": + existing = values.get("chat_template_kwargs") + existing = existing if isinstance(existing, dict) else {} + # existing last: an explicit caller value wins. + values["chat_template_kwargs"] = { + "thinking": False, + "enable_thinking": False, + **existing, + } + return values + @model_validator(mode="before") @classmethod def normalize_responses_input(cls, values): @@ -1569,11 +1608,17 @@ class ResponsesRequest(BaseModel): if not isinstance(item, dict): return item + # an output item replayed into input carries a string id; without this it'd + # be read as an item-reference, drop its content, and fail as an empty {}. + # input-item ids aren't resolved server-side, so just drop it. + if isinstance(item.get("id"), str) and item.get("content") is not None: + item = {k: v for k, v in item.items() if k != "id"} + content = item.get("content") if not isinstance(content, list): return item - item = item.copy() + item = dict(item) item["content"] = [ ResponsesRequest._normalize_content_part_for_validation(part) for part in content @@ -1593,6 +1638,40 @@ class ResponsesRequest(BaseModel): part["detail"] = "auto" return part + @staticmethod + def _json_schema_from_text_format( + text: Optional[ResponseTextConfig], + ) -> Optional[str]: + """Map a Responses ``text.format`` to a json_schema string, or None when + no JSON constraint applies (``text`` and anything unrecognized).""" + response_format = text.format if text is not None else None + if isinstance(response_format, ResponseFormatJSONObject): + return '{"type": "object"}' + if not isinstance(response_format, ResponseFormatTextJSONSchemaConfig): + return None + schema = response_format.schema_ + return convert_json_schema_to_str(schema) if schema is not None else None + + def is_include_output_logprobs(self) -> bool: + return bool(self.include and "message.output_text.logprobs" in self.include) + + def has_json_schema_constraint(self) -> bool: + return self._json_schema_from_text_format(self.text) is not None + + def effective_tool_choice(self) -> Union[str, Dict[str, Any]]: + """``tool_choice`` reduced to what the server can actually honor: of the + object forms only a named ``function`` survives, the rest (web_search, + mcp, ...) can't be forced through the tool-call parser.""" + tool_choice = self.tool_choice + if not isinstance(tool_choice, dict): + return tool_choice + name = tool_choice.get("name") or (tool_choice.get("function") or {}).get( + "name" + ) + if tool_choice.get("type") == "function" and name: + return {"type": "function", "name": name} + return "auto" + def to_sampling_params( self, default_max_tokens: int, @@ -1645,6 +1724,10 @@ class ResponsesRequest(BaseModel): if key not in params or params[key] is None: params[key] = value + json_schema = self._json_schema_from_text_format(self.text) + if json_schema is not None: + params["json_schema"] = json_schema + has_existing_constraints = ( params.get("regex") or params.get("ebnf") @@ -1655,7 +1738,8 @@ class ResponsesRequest(BaseModel): # 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." + "(text.format / regex / ebnf / structural_tag / json_schema). " + "Remove one." ) if tool_call_constraint: constraint_type, constraint_value = tool_call_constraint @@ -1688,10 +1772,12 @@ class ResponsesResponse(BaseModel): output: List[ Union[ResponseOutputItem, ResponseReasoningItem, ResponseFunctionToolCall] ] = Field(default_factory=list) - status: Literal["queued", "in_progress", "completed", "failed", "cancelled"] + status: Literal[ + "queued", "in_progress", "completed", "incomplete", "failed", "cancelled" + ] usage: Optional[UsageInfo] = None parallel_tool_calls: bool = True - tool_choice: str = "auto" + tool_choice: Union[str, Dict[str, Any]] = "auto" tools: List[ResponseTool] = Field(default_factory=list) # OpenAI compatibility fields. not all are used at the moment. @@ -1714,6 +1800,28 @@ class ResponsesResponse(BaseModel): user: Optional[str] = None metadata: Optional[Dict[str, Any]] = None + @field_serializer("usage") + def _serialize_usage(self, usage: Optional[UsageInfo], _info): + """Emit the Responses usage shape, not the chat one UsageInfo carries.""" + if usage is None: + return None + cached = ( + usage.prompt_tokens_details.cached_tokens + if usage.prompt_tokens_details + else 0 + ) + return { + "input_tokens": usage.prompt_tokens, + "input_tokens_details": { + "cached_tokens": cached, + # required (no default) in the SDK's InputTokensDetails model + "cache_write_tokens": 0, + }, + "output_tokens": usage.completion_tokens or 0, + "output_tokens_details": {"reasoning_tokens": usage.reasoning_tokens or 0}, + "total_tokens": usage.total_tokens, + } + @classmethod def from_request( cls, @@ -1761,7 +1869,13 @@ class ResponsesResponse(BaseModel): return False return True - text_format = {"format": {"type": "text"}} if _is_text_only(output) else None + if request.text is not None: + # by_alias keeps the wire key "schema" rather than the attr schema_. + text_format = request.text.model_dump(by_alias=True, exclude_none=True) + else: + text_format = ( + {"format": {"type": "text"}} if _is_text_only(output) else None + ) return cls( id=request.request_id, @@ -1775,16 +1889,23 @@ class ResponsesResponse(BaseModel): if request.parallel_tool_calls is not None else True ), - tool_choice=request.tool_choice, + tool_choice=request.effective_tool_choice(), tools=request.tools, # fields for parity with v1/responses error=None, - incomplete_details=None, + incomplete_details=( + {"reason": "max_output_tokens"} if status == "incomplete" else None + ), instructions=request.instructions, max_output_tokens=request.max_output_tokens, previous_response_id=request.previous_response_id, # TODO(v): ensure this is propagated if retrieved from store reasoning={ - "effort": request.reasoning.effort if request.reasoning else None, + "effort": ( + request.reasoning.effort + if request.reasoning + and request.reasoning.effort in ECHOABLE_REASONING_EFFORTS + else None + ), "summary": None, # unused }, store=request.store, @@ -1814,6 +1935,7 @@ class MessageProcessingResult: modalities: List[str] stop: List[str] tool_call_constraint: Optional[ToolCallConstraint] = None + skip_special_tokens: bool = True require_reasoning: bool = False diff --git a/python/sglang/srt/entrypoints/openai/serving_responses.py b/python/sglang/srt/entrypoints/openai/serving_responses.py index ca4207e81..516467359 100644 --- a/python/sglang/srt/entrypoints/openai/serving_responses.py +++ b/python/sglang/srt/entrypoints/openai/serving_responses.py @@ -23,6 +23,7 @@ from openai.types.responses import ( ResponseReasoningItem, ) from openai.types.responses.response_function_tool_call import ResponseFunctionToolCall +from openai.types.responses.response_output_text import Logprob, LogprobTopLogprob from openai.types.responses.response_reasoning_item import ( Content as ResponseReasoningTextContent, ) @@ -67,6 +68,7 @@ from sglang.srt.entrypoints.openai.protocol import ( ) from sglang.srt.entrypoints.openai.serving_chat import OpenAIServingChat from sglang.srt.entrypoints.openai.tool_server import MCPToolServer, ToolServer +from sglang.srt.entrypoints.openai.utils import to_openai_style_logprobs 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 @@ -84,6 +86,54 @@ class _MediaInputValidationError(ValueError): pass +def _build_output_text_logprobs(meta_info: dict) -> list[Logprob]: + """Reshape decoded ``meta_info`` logprobs into the Responses logprob type, + covering every generated token.""" + decoded = to_openai_style_logprobs( + output_token_logprobs=meta_info.get("output_token_logprobs"), + output_top_logprobs=meta_info.get("output_top_logprobs"), + ) + top_lists = decoded.top_logprobs or [] + logprobs: list[Logprob] = [] + for index, (token, logprob) in enumerate( + zip(decoded.tokens, decoded.token_logprobs) + ): + top_entry = top_lists[index] if index < len(top_lists) else None + top_logprobs = [ + LogprobTopLogprob( + token=top_token, + logprob=top_logprob, + bytes=list(top_token.encode("utf-8")), + ) + for top_token, top_logprob in (top_entry or {}).items() + ] + logprobs.append( + Logprob( + token=token, + logprob=logprob, + bytes=list(token.encode("utf-8")), + top_logprobs=top_logprobs, + ) + ) + return logprobs + + +def _should_emit_normal_text_as_message( + text: str, *, any_tool_call_in_progress: bool +) -> bool: + """Whether ``text`` should open / extend a user-visible message item. + + qwen3-coder separates adjacent tool-call blocks with ``\\n``, which the + streaming detector cannot tell from real content -- so whitespace arriving + while a call is open is treated as a separator. + """ + if not text: + return False + if any_tool_call_in_progress and not text.strip(): + return False + return True + + class OpenAIServingResponses(OpenAIServingChat): """Handler for /v1/responses requests""" @@ -202,6 +252,23 @@ class OpenAIServingResponses(OpenAIServingChat): 'type="function"; other built-in tool types cannot be forced.' ) + # harmony emits raw tokens; per-token logprobs aren't wired there. + if self.use_harmony and request.is_include_output_logprobs(): + return self.create_error_response( + "logprobs are not supported with gpt-oss models", param="logprobs" + ) + # streaming skips the logprobs build path; reject so the include doesn't silently no-op. + if request.stream and request.is_include_output_logprobs(): + return self.create_error_response( + "logprobs are not supported in streaming mode", param="logprobs" + ) + # harmony output opens with <|channel|>analysis<|message|>, so a whole-output + # json_schema forces "{" at the first token and the harmony parse then fails. + if self.use_harmony and request.has_json_schema_constraint(): + return self.create_error_response( + "structured output (text.format) is not supported with gpt-oss models", + param="text", + ) if ( self.use_harmony and self._has_response_tool(request, "web_search", "web_search_preview") @@ -328,6 +395,13 @@ class OpenAIServingResponses(OpenAIServingChat): ), ) + # _process_messages set skip_special_tokens on a chat_request + # we then discard, so re-apply it to the engine sampling dict. + if processed_messages is not None and ( + not processed_messages.skip_special_tokens + ): + sampling_params["skip_special_tokens"] = False + context: ConversationContext if self.use_harmony: if request.stream: @@ -343,8 +417,20 @@ class OpenAIServingResponses(OpenAIServingChat): else: prompt_kwargs = {"input_ids": engine_prompt} + logprob_kwargs = ( + { + "return_logprob": True, + "logprob_start_len": -1, + "top_logprobs_num": request.top_logprobs or 0, + "return_text_in_logprobs": True, + } + if request.is_include_output_logprobs() + else {} + ) + adapted_request = GenerateReqInput( **prompt_kwargs, + **logprob_kwargs, image_data=( processed_messages.image_data if processed_messages @@ -370,7 +456,8 @@ class OpenAIServingResponses(OpenAIServingChat): rid=request.request_id, session_id=request.session_id, extra_key=self._compute_extra_key(request), - background=request.background, + # background+stream streams on this connection, so don't detach. + background=request.background and not request.stream, require_reasoning=require_reasoning, ) @@ -394,7 +481,7 @@ class OpenAIServingResponses(OpenAIServingChat): if request.store: self.msg_store[request.request_id] = messages - if request.background: + if request.background and not request.stream: created_time = int(time.time()) response = ResponsesResponse.from_request( request, @@ -484,7 +571,11 @@ class OpenAIServingResponses(OpenAIServingChat): messages=messages, stream=request.stream, tools=chat_tools or None, - tool_choice=request.tool_choice if chat_tools else "none", + tool_choice=( + self._chat_tool_choice(request.effective_tool_choice()) + if chat_tools + else "none" + ), parallel_tool_calls=( request.parallel_tool_calls if request.parallel_tool_calls is not None @@ -492,6 +583,7 @@ class OpenAIServingResponses(OpenAIServingChat): ), stop=request.stop, reasoning_effort=(request.reasoning.effort if request.reasoning else None), + chat_template_kwargs=request.chat_template_kwargs, ) media_error = self._validate_media_content(chat_request) @@ -500,6 +592,7 @@ class OpenAIServingResponses(OpenAIServingChat): is_multimodal = self.tokenizer_manager.model_config.is_multimodal processed_messages = self._process_messages(chat_request, is_multimodal) + processed_messages.skip_special_tokens = chat_request.skip_special_tokens if is_multimodal: request_prompts = [processed_messages.prompt] @@ -548,6 +641,7 @@ class OpenAIServingResponses(OpenAIServingChat): except ValueError as e: return self.create_error_response(str(e)) + status = "completed" if self.use_harmony: assert isinstance(context, HarmonyContext) output = self._make_response_output_items_with_harmony(context) @@ -556,19 +650,12 @@ class OpenAIServingResponses(OpenAIServingChat): num_generated_tokens = context.num_output_tokens num_cached_tokens = context.num_cached_tokens num_reasoning_tokens = context.num_reasoning_tokens + status = self._status_from_finish_reason(context.finish_reason) else: assert isinstance(context, SimpleContext) final_res = context.last_output assert final_res is not None - output = self._make_response_output_items( - request, - final_res["text"], - tokenizer, - require_reasoning=require_reasoning, - ) - - # Calculate usage from actual output num_reasoning_tokens = 0 meta_info = None if isinstance(final_res, dict) and isinstance( @@ -578,11 +665,25 @@ class OpenAIServingResponses(OpenAIServingChat): elif hasattr(final_res, "meta_info"): meta_info = final_res.meta_info + output_logprobs = ( + _build_output_text_logprobs(meta_info) + if request.is_include_output_logprobs() and isinstance(meta_info, dict) + else None + ) + output = self._make_response_output_items( + request, + final_res["text"], + tokenizer, + output_logprobs=output_logprobs, + require_reasoning=require_reasoning, + ) + 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) + status = self._status_from_finish_reason(meta_info.get("finish_reason")) elif isinstance(final_res, dict) and ( final_res.get("prompt_token_ids") is not None or final_res.get("output_ids") is not None @@ -630,7 +731,7 @@ class OpenAIServingResponses(OpenAIServingChat): model_name=model_name, created_time=created_time, output=output, - status="completed", + status=status, usage=usage, ) @@ -647,9 +748,27 @@ class OpenAIServingResponses(OpenAIServingChat): def _wants_reasoning_summary(request: ResponsesRequest) -> bool: return request.reasoning is not None and request.reasoning.summary is not None + @staticmethod + def _status_from_finish_reason(finish_reason: Any) -> str: + """Only a length-capped generation is ``incomplete``; anything that got + here otherwise finished normally.""" + reason = None + if isinstance(finish_reason, dict): + reason = finish_reason.get("type") + elif isinstance(finish_reason, str): + reason = finish_reason + return "incomplete" if reason == "length" else "completed" + def _is_thinking_enabled_for_request(self, request: ResponsesRequest) -> bool: if not self.reasoning_parser: return False + # an explicit toggle wins; the key differs by family (enable_thinking vs thinking). + ctk = request.chat_template_kwargs or {} + thinking_toggles = (ctk.get("enable_thinking"), ctk.get("thinking")) + if any(toggle is False for toggle in thinking_toggles): + return False + if any(toggle is True for toggle in thinking_toggles): + return True 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") @@ -685,6 +804,7 @@ class OpenAIServingResponses(OpenAIServingChat): request: ResponsesRequest, final_output: Any, tokenizer: Any, + output_logprobs: Optional[list] = None, *, require_reasoning: bool, ): @@ -692,7 +812,11 @@ class OpenAIServingResponses(OpenAIServingChat): reasoning_parser = ReasoningParser( model_type=self.reasoning_parser, stream_reasoning=False, - force_reasoning=require_reasoning, + # A template that prefills forces the parser open even + # when the request itself did not ask for reasoning, same as chat. + force_reasoning=( + self.template_manager.force_reasoning or require_reasoning + ), request=request, tokenizer=self.tokenizer_manager.tokenizer, ) @@ -794,7 +918,8 @@ class OpenAIServingResponses(OpenAIServingChat): text=content, annotations=[], # TODO type="output_text", - logprobs=None, # TODO + # logprobs cover all generated tokens, not just the stripped content. + logprobs=output_logprobs, ) message = ResponseOutputMessage( id=f"msg_{random_uuid()}", @@ -821,6 +946,14 @@ class OpenAIServingResponses(OpenAIServingChat): output_items.extend(last_items) return output_items + @staticmethod + def _chat_tool_choice(tool_choice: Any) -> Any: + """Nest an ``effective_tool_choice()`` result the way chat expects: + ``{"type":"function","name":X}`` -> ``{...,"function":{"name":X}}``.""" + if not isinstance(tool_choice, dict): + return tool_choice + return {"type": "function", "function": {"name": tool_choice["name"]}} + @staticmethod def _response_tools_to_chat_tools(request: ResponsesRequest) -> list[Tool]: # Only ``function`` tools flow to chat; built-ins go through harmony. @@ -929,10 +1062,15 @@ class OpenAIServingResponses(OpenAIServingChat): ], } if msg_type == "function_call_output": + # ``output`` may be a string or an array of content parts (OpenAI + # allows both); the chat tool message needs a string, so flatten. + out = message.get("output", "") + if isinstance(out, list): + out = "".join(p.get("text", "") for p in out if isinstance(p, dict)) return { "role": "tool", "tool_call_id": message.get("call_id"), - "content": message.get("output", ""), + "content": out, } # Reasoning items render as {role: assistant, reasoning_content}; # empty ones drop instead of injecting an empty assistant block. @@ -1274,10 +1412,8 @@ class OpenAIServingResponses(OpenAIServingChat): prev_status = response.status if prev_status not in ("queued", "in_progress"): - return self.create_error_response( - err_type="invalid_request_error", - message="Cannot cancel a synchronous response.", - ) + # already terminal; a second cancel is a no-op, return as-is. + return response # Update the status to "cancelled" response.status = "cancelled" @@ -1374,7 +1510,6 @@ class OpenAIServingResponses(OpenAIServingChat): ) async for ctx in result_generator: - # Only process context objects that implement the `is_expecting_start()` method, # which indicates they support per-turn streaming (e.g., StreamingHarmonyContext). # Contexts without this method are skipped, as they do not represent a new turn @@ -1731,26 +1866,10 @@ class OpenAIServingResponses(OpenAIServingChat): created_time=created_time, require_reasoning=require_reasoning, ) - # 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"): - usage_info = response_dict["usage"] - response_dict["usage"] = { - "input_tokens": usage_info.get("prompt_tokens", 0), - "input_tokens_details": { - "cached_tokens": usage_info.get("cached_tokens", 0) - }, - "output_tokens": usage_info.get("completion_tokens", 0), - "output_tokens_details": { - "reasoning_tokens": usage_info.get("reasoning_tokens", 0) - }, - "total_tokens": usage_info.get("total_tokens", 0), - } - yield _send_event( openai_responses_types.ResponseCompletedEvent( type="response.completed", @@ -1851,7 +1970,11 @@ class OpenAIServingResponses(OpenAIServingChat): reasoning_parser_obj = ReasoningParser( model_type=self.reasoning_parser, stream_reasoning=True, - force_reasoning=require_reasoning, + # A template that prefills forces the parser open even + # when the request itself did not ask for reasoning, same as chat. + force_reasoning=( + self.template_manager.force_reasoning or require_reasoning + ), request=request, tokenizer=self.tokenizer_manager.tokenizer, ) @@ -2166,118 +2289,142 @@ class OpenAIServingResponses(OpenAIServingChat): 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", - ), + def _emit_tool_calls(calls): + nonlocal current_output_index + if calls: + 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 calls: + tool_index = call.tool_index + state = tool_call_states.get(tool_index) + if state is None or state.get("done"): + # Close other open calls first, so their + # output_item.done precedes the next added. + for other_index in list(tool_call_states): + if other_index != tool_index: + for ev in _close_tool_call_state(other_index): + yield ev + 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 + 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, + ) + ) + + def _emit_normal_text(): + if normal_text and _should_emit_normal_text_as_message( + normal_text, + any_tool_call_in_progress=any( + not s.get("done") for s in tool_call_states.values() + ), + ): + 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.ResponseContentPartAddedEvent( - type="response.content_part.added", + 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"], - content_index=0, - part=openai_responses_types.ResponseOutputText( - type="output_text", - text="", - annotations=[], - logprobs=None, - ), + delta=normal_text, + logprobs=[], ) ) - 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 + # The parser's (text, calls) tuple is unordered, but positions + # are recoverable: continuing arguments precede this delta's + # text, a newly opened call follows it. Classify first -- + # emitting mutates tool_call_states. + def _is_continuing(call): + state = tool_call_states.get(call.tool_index) + return state is not None and not state.get("done") - if reasoning_state["open"]: - for ev in _close_reasoning_item(): - yield ev - if message_state["open"]: - for ev in _close_message_item(): - yield ev + continuing = [c for c in tool_calls if _is_continuing(c)] + opening = [c for c in tool_calls if not _is_continuing(c)] - 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, - ) - ) + for ev in _emit_tool_calls(continuing): + yield ev + for ev in _emit_normal_text(): + yield ev + for ev in _emit_tool_calls(opening): + yield ev except Exception: logger.exception("Error while streaming /v1/responses") failed = _sanitize_response_dict( @@ -2328,7 +2475,7 @@ class OpenAIServingResponses(OpenAIServingChat): model_name=model_name, created_time=created_time, output=final_output_items, - status="completed", + status=self._status_from_finish_reason(finish_reason), usage=usage, ) if request.store: @@ -2338,19 +2485,6 @@ class OpenAIServingResponses(OpenAIServingChat): 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( diff --git a/test/registered/openai_server/basic/test_openai_server.py b/test/registered/openai_server/basic/test_openai_server.py index d4042ce28..a3c517602 100644 --- a/test/registered/openai_server/basic/test_openai_server.py +++ b/test/registered/openai_server/basic/test_openai_server.py @@ -786,20 +786,24 @@ class TestOpenAIServerv1Responses(CustomTestCase): assert isinstance(resp.output, list) assert resp.status in ( "completed", + "incomplete", "in_progress", "queued", "failed", "cancelled", ) - if resp.status == "completed": + if resp.status in ("completed", "incomplete"): assert resp.usage is not None - assert resp.usage.prompt_tokens >= 0 - assert resp.usage.completion_tokens >= 0 + assert resp.usage.input_tokens >= 0 + assert resp.usage.output_tokens >= 0 assert resp.usage.total_tokens >= 0 if hasattr(resp, "error"): assert resp.error is None if hasattr(resp, "incomplete_details"): - assert resp.incomplete_details is None + if resp.status == "incomplete": + assert resp.incomplete_details.reason == "max_output_tokens" + else: + assert resp.incomplete_details is None if getattr(resp, "text", None): fmt = resp.text.get("format") if isinstance(resp.text, dict) else None if fmt: @@ -818,8 +822,8 @@ class TestOpenAIServerv1Responses(CustomTestCase): def test_response_completion(self): resp = self.run_response(temperature=0, max_output_tokens=16) - assert resp.status in ("completed", "in_progress", "queued") - if resp.status == "completed": + assert resp.status in ("completed", "incomplete", "in_progress", "queued") + if resp.status in ("completed", "incomplete"): assert resp.usage is not None assert resp.usage.total_tokens >= 0 @@ -900,9 +904,10 @@ class TestOpenAIServerv1Responses(CustomTestCase): self.assertEqual(body.get("object"), "response") self.assertIn("output", body) self.assertIn("status", body) - if "usage" in body: - self.assertIn("prompt_tokens", body["usage"]) - self.assertIn("total_tokens", body["usage"]) + self.assertIn("usage", body) + self.assertIn("input_tokens", body["usage"]) + self.assertIn("output_tokens", body["usage"]) + self.assertIn("total_tokens", body["usage"]) def test_response_prefill(self): client = openai.Client(api_key=self.api_key, base_url=self.base_url) diff --git a/test/registered/unit/entrypoints/openai/test_responses_protocol.py b/test/registered/unit/entrypoints/openai/test_responses_protocol.py index 19a11ad13..5faa3b19f 100644 --- a/test/registered/unit/entrypoints/openai/test_responses_protocol.py +++ b/test/registered/unit/entrypoints/openai/test_responses_protocol.py @@ -1,14 +1,33 @@ +import json import unittest from utils import make_serving # noqa: F401 — bootstrap import -from sglang.srt.entrypoints.openai.protocol import ResponsesRequest, UsageInfo +from sglang.srt.entrypoints.openai.protocol import ( + PromptTokensDetails, + ResponsesRequest, + ResponsesResponse, + UsageInfo, +) from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase -register_cpu_ci(est_time=4, suite="base-a-test-cpu") +register_cpu_ci(est_time=7, suite="base-a-test-cpu") -class ResponsesRequestTestCase(unittest.TestCase): +def _in_progress_response(request: ResponsesRequest) -> ResponsesResponse: + return ResponsesResponse.from_request( + request, + sampling_params={}, + model_name="x", + created_time=0, + output=[], + status="in_progress", + usage=None, + ) + + +class ResponsesRequestTestCase(CustomTestCase): def test_function_tool_accepted(self): request = ResponsesRequest( model="x", @@ -88,7 +107,7 @@ class ResponsesRequestTestCase(unittest.TestCase): self.assertEqual(request.tools[0].tools[0]["name"], "apply_patch") -class ResponsesSamplingParamsTestCase(unittest.TestCase): +class ResponsesSamplingParamsTestCase(CustomTestCase): def test_processed_stop_and_tool_constraint_propagate(self): request = ResponsesRequest(model="x", input="call the tool", store=False) params = request.to_sampling_params( @@ -122,11 +141,118 @@ class ResponsesSamplingParamsTestCase(unittest.TestCase): ) self.assertEqual(params["structural_tag"], '{"type": "structural_tag"}') + def test_text_format_maps_to_json_schema_constraint(self): + schema = {"type": "object", "properties": {"age": {"type": "integer"}}} + for fmt, expected in [ + ({"type": "json_schema", "name": "p", "schema": schema}, schema), + ({"type": "json_object"}, {"type": "object"}), + ]: + req = ResponsesRequest( + model="x", input="hi", store=False, text={"format": fmt} + ) + params = req.to_sampling_params(default_max_tokens=128, default_params={}) + self.assertEqual(json.loads(params["json_schema"]), expected, fmt) + plain = ResponsesRequest( + model="x", input="hi", store=False, text={"format": {"type": "text"}} + ).to_sampling_params(default_max_tokens=128, default_params={}) + self.assertNotIn("json_schema", plain) + + def test_text_format_conflicts_with_tool_constraint(self): + request = ResponsesRequest( + model="x", + input="hi", + store=False, + text={ + "format": { + "type": "json_schema", + "name": "p", + "schema": {"type": "object"}, + } + }, + ) + # The message must name text.format: it is the only source of the + # conflict a /v1/responses caller can actually set. + with self.assertRaisesRegex(ValueError, r"text\.format"): + request.to_sampling_params( + default_max_tokens=128, + default_params={}, + tool_call_constraint=("json_schema", {"type": "object"}), + ) + + +class IncludeOutputLogprobsTestCase(CustomTestCase): + def test_detected_only_for_logprobs_include(self): + def has(include): + return ResponsesRequest( + model="x", input="hi", store=False, include=include + ).is_include_output_logprobs() + + self.assertTrue(has(["message.output_text.logprobs"])) + self.assertFalse(has(None)) + self.assertFalse(has(["reasoning.encrypted_content"])) + + +class ThinkingControlTestCase(CustomTestCase): + def test_effort_none_disables_thinking(self): + ctk = ResponsesRequest( + model="x", input="hi", store=False, reasoning={"effort": "none"} + ).chat_template_kwargs + self.assertEqual((ctk["enable_thinking"], ctk["thinking"]), (False, False)) + + def test_thinking_untouched_otherwise(self): + # grammar-constrained requests keep thinking on; ReasonerGrammarBackend + # defers the grammar past so the two coexist. + for kw in ( + {"reasoning": {"effort": "medium"}}, + {"text": {"format": {"type": "text"}}}, + { + "text": { + "format": { + "type": "json_schema", + "name": "p", + "schema": {"type": "object"}, + } + } + }, + {"text": {"format": {"type": "json_object"}}}, + {"tool_choice": "required"}, + {"tool_choice": {"type": "function", "name": "f"}}, + {}, + ): + req = ResponsesRequest(model="x", input="hi", store=False, **kw) + self.assertIsNone(req.chat_template_kwargs, kw) + + def test_explicit_chat_template_kwargs_preserved(self): + req = ResponsesRequest( + model="x", + input="hi", + store=False, + chat_template_kwargs={"enable_thinking": True}, + ) + self.assertTrue(req.chat_template_kwargs["enable_thinking"]) + + +class ResponsesResponseFromRequestTestCase(CustomTestCase): + def test_requested_text_format_is_echoed(self): + schema = {"type": "object", "properties": {"x": {"type": "integer"}}} + request = ResponsesRequest( + model="x", + input="hi", + store=False, + text={"format": {"type": "json_schema", "name": "p", "schema": schema}}, + ) + 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.assertEqual(response.text["format"]["type"], "json_schema") -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 ) @@ -141,6 +267,144 @@ class ResponsesResponseFromRequestTestCase(unittest.TestCase): ) self.assertFalse(response.parallel_tool_calls) + def test_incomplete_status_sets_incomplete_details(self): + request = ResponsesRequest(model="x", input="hi", store=False) + incomplete = ResponsesResponse.from_request( + request, + sampling_params={}, + model_name="x", + created_time=0, + output=[], + status="incomplete", + usage=UsageInfo(prompt_tokens=1, completion_tokens=1, total_tokens=2), + ) + self.assertEqual(incomplete.status, "incomplete") + self.assertEqual(incomplete.incomplete_details, {"reason": "max_output_tokens"}) + + completed = 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.assertIsNone(completed.incomplete_details) + + def test_usage_serialized_in_responses_shape(self): + request = ResponsesRequest(model="x", input="hi", store=False) + resp = ResponsesResponse.from_request( + request, + sampling_params={}, + model_name="x", + created_time=0, + output=[], + status="completed", + usage=UsageInfo( + prompt_tokens=11, + completion_tokens=102, + total_tokens=113, + reasoning_tokens=7, + prompt_tokens_details=PromptTokensDetails(cached_tokens=3), + ), + ) + usage = resp.model_dump()["usage"] + self.assertEqual(usage["input_tokens"], 11) + self.assertEqual(usage["output_tokens"], 102) + self.assertEqual(usage["total_tokens"], 113) + self.assertEqual(usage["output_tokens_details"]["reasoning_tokens"], 7) + self.assertEqual(usage["input_tokens_details"]["cached_tokens"], 3) + # Chat-style keys must be gone. + self.assertNotIn("prompt_tokens", usage) + self.assertNotIn("completion_tokens", usage) + + def test_only_sdk_known_efforts_echoed_so_streaming_event_validates(self): + import openai.types.responses as ort + + # OpenAI's Reasoning.effort literal is narrower than our tier list: + # "none" is a request-side extension and xhigh/max postdate it. Echoing + # one of those raises a ValidationError in the typed event below, which + # the stream generator builds before its try block -- the connection + # then dies without a single SSE byte. + for effort in ("none", "minimal", "low", "medium", "high", "xhigh", "max"): + resp = _in_progress_response( + ResponsesRequest( + model="x", input="hi", store=False, reasoning={"effort": effort} + ) + ) + expected = ( + effort if effort in ("minimal", "low", "medium", "high") else None + ) + self.assertEqual(resp.reasoning["effort"], expected, effort) + ort.ResponseCreatedEvent( + type="response.created", sequence_number=0, response=resp.model_dump() + ) + + +class InputItemStringIdTestCase(CustomTestCase): + """A response.output item replayed into input (string id + content) must + keep its content rather than collapse to an item-reference.""" + + def test_string_id_dropped_only_for_content_items(self): + norm = ResponsesRequest._normalize_input_item_for_validation + kept = norm( + { + "id": "msg_x", + "role": "assistant", + "content": [{"type": "output_text", "text": "hi"}], + } + ) + self.assertNotIn("id", kept) + self.assertTrue(kept["content"]) + # int ids and bare item-references are left alone + self.assertEqual(norm({"id": 123, "content": [{"type": "text"}]})["id"], 123) + ref = {"type": "item_reference", "id": "msg_ref"} + self.assertEqual(norm(ref), ref) + + def test_request_accepts_replayed_output_item(self): + # Construction must not raise (the bug returned 400). + ResponsesRequest( + model="x", + store=False, + input=[ + { + "id": "msg_x", + "role": "assistant", + "content": [{"type": "output_text", "text": "hi"}], + } + ], + ) + + +class ToolChoiceObjectFormTestCase(CustomTestCase): + def test_echoed_choice_is_what_the_server_honors(self): + import openai.types.responses as ort + + named = {"type": "function", "name": "get_weather"} + # Object forms other than a named function cannot be forced through the + # tool-call parser, so the response must echo the "auto" we actually + # run -- which also keeps it inside the SDK's ToolChoice union that the + # typed event below validates against. + for tool_choice, expected in ( + ("auto", "auto"), + ("required", "required"), + ("none", "none"), + (named, named), + ({"type": "function", "function": {"name": "get_weather"}}, named), + ({"type": "web_search"}, "auto"), + ({"type": "mcp", "server_label": "s"}, "auto"), + ): + req = ResponsesRequest( + model="x", input="hi", store=False, tool_choice=tool_choice + ) + self.assertEqual(req.effective_tool_choice(), expected, tool_choice) + resp = _in_progress_response(req) + self.assertEqual(resp.tool_choice, expected, tool_choice) + ort.ResponseCreatedEvent( + type="response.created", sequence_number=0, response=resp.model_dump() + ) + 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 index 2a0e14223..932456998 100644 --- a/test/registered/unit/entrypoints/openai/test_serving_responses.py +++ b/test/registered/unit/entrypoints/openai/test_serving_responses.py @@ -16,15 +16,20 @@ from sglang.srt.entrypoints.openai.protocol import ( RequestResponseMetadata, ResponsesRequest, ) -from sglang.srt.entrypoints.openai.serving_responses import OpenAIServingResponses +from sglang.srt.entrypoints.openai.serving_responses import ( + OpenAIServingResponses, + _build_output_text_logprobs, + _should_emit_normal_text_as_message, +) from sglang.srt.function_call.core_types import ToolCallItem from sglang.srt.parser.template_detection import ReasoningToggleConfig from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase -register_cpu_ci(est_time=8, suite="base-a-test-cpu") +register_cpu_ci(est_time=7, suite="base-a-test-cpu") -class InputMessageConstructionTestCase(unittest.TestCase): +class InputMessageConstructionTestCase(CustomTestCase): def test_previous_response_replays_assistant_text_not_instructions(self): serving = make_serving() prev_response = Mock(id="resp_prev") @@ -146,7 +151,7 @@ class InputMessageConstructionTestCase(unittest.TestCase): pass -class ChatToolForwardingTestCase(unittest.TestCase): +class ChatToolForwardingTestCase(CustomTestCase): def test_make_request_passes_function_tools_to_chat_processing(self): serving = make_serving() seen = {} @@ -306,7 +311,7 @@ class ReasoningRequestForwardingTestCase(unittest.TestCase): self.assertFalse(parser_cls.call_args.kwargs["force_reasoning"]) -class InputItemNormalizationTestCase(unittest.TestCase): +class InputItemNormalizationTestCase(CustomTestCase): def test_function_call_becomes_assistant_tool_call(self): normalized = OpenAIServingResponses._normalize_response_message_for_chat( { @@ -361,7 +366,7 @@ class InputItemNormalizationTestCase(unittest.TestCase): ) -class FullResponseUsageTestCase(unittest.TestCase): +class FullResponseUsageTestCase(CustomTestCase): def test_full_response_uses_dict_meta_info_for_usage(self): serving = make_serving() context = SimpleContext() @@ -403,7 +408,7 @@ class FullResponseUsageTestCase(unittest.TestCase): self.assertEqual(metadata.final_usage_info, response.usage) -class MultimodalRequestTestCase(unittest.TestCase): +class MultimodalRequestTestCase(CustomTestCase): def test_text_only_create_responses_rejects_media_before_generation(self): serving = make_serving() serving._process_messages = Mock() @@ -500,7 +505,13 @@ class MultimodalRequestTestCase(unittest.TestCase): self.assertEqual(captured["adapted_request"].modalities, ["image"]) -class OutputItemsTestCase(unittest.TestCase): +class OutputItemsTestCase(CustomTestCase): + def setUp(self): + # qwen3_coder is the default for this class; the one no-native-parser + # case overrides it. + self.serving = make_serving() + self.serving.tool_call_parser = "qwen3_coder" + def _function_tool_request(self): return ResponsesRequest( model="x", @@ -517,8 +528,7 @@ class OutputItemsTestCase(unittest.TestCase): ) def test_function_tool_call_extracted_via_parser(self): - serving = make_serving() - serving.tool_call_parser = "qwen3_coder" + serving = self.serving fake_call = ToolCallItem( tool_index=0, name="get_weather", parameters='{"city": "Beijing"}' ) @@ -550,8 +560,7 @@ class OutputItemsTestCase(unittest.TestCase): 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" + serving = self.serving fake_call = ToolCallItem( tool_index=0, name="get_weather", parameters='{"city": "Beijing"}' ) @@ -576,7 +585,7 @@ class OutputItemsTestCase(unittest.TestCase): self.assertEqual(types, ["ResponseOutputMessage", "ResponseFunctionToolCall"]) def test_required_tool_choice_parses_json_array_without_native_parser(self): - serving = make_serving() + serving = self.serving serving.tool_call_parser = None request = ResponsesRequest( model="x", @@ -609,8 +618,7 @@ class OutputItemsTestCase(unittest.TestCase): ) def test_no_tool_call_extraction_when_tool_choice_none(self): - serving = make_serving() - serving.tool_call_parser = "qwen3_coder" + serving = self.serving request = ResponsesRequest( model="x", input="hi", @@ -640,7 +648,7 @@ class OutputItemsTestCase(unittest.TestCase): self.assertIsInstance(output_items[0], ResponseOutputMessage) -class HarmonyResponsesTestCase(unittest.TestCase): +class HarmonyResponsesTestCase(CustomTestCase): 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 @@ -660,5 +668,214 @@ class HarmonyResponsesTestCase(unittest.TestCase): self.assertIsNotNone(msg) +class StatusFromFinishReasonTestCase(CustomTestCase): + def test_only_length_maps_to_incomplete(self): + fn = OpenAIServingResponses._status_from_finish_reason + self.assertEqual(fn({"type": "length"}), "incomplete") + self.assertEqual(fn("length"), "incomplete") + for other in ({"type": "stop"}, {"type": "tool_calls"}, "stop", None): + self.assertEqual(fn(other), "completed", other) + + +class BuildOutputTextLogprobsTestCase(CustomTestCase): + def test_tokens_and_top_logprobs_are_converted(self): + meta_info = { + "output_token_logprobs": [(-0.1, 10, "Hello"), (-0.2, 11, " world")], + "output_top_logprobs": [ + [(-0.1, 10, "Hello"), (-2.0, 12, "Hi")], + [(-0.2, 11, " world"), (-3.0, 13, " earth")], + ], + } + out = _build_output_text_logprobs(meta_info) + self.assertEqual(len(out), 2) + self.assertEqual(out[0].token, "Hello") + self.assertEqual(out[0].logprob, -0.1) + self.assertEqual(out[0].bytes, list("Hello".encode("utf-8"))) + self.assertEqual(len(out[0].top_logprobs), 2) + self.assertEqual(out[0].top_logprobs[0].token, "Hello") + self.assertEqual(out[1].token, " world") + + def test_no_top_logprobs_yields_empty_lists(self): + meta_info = { + "output_token_logprobs": [(-0.5, 7, "hi")], + "output_top_logprobs": None, + } + out = _build_output_text_logprobs(meta_info) + self.assertEqual(len(out), 1) + self.assertEqual(out[0].top_logprobs, []) + + +class ChatToolChoiceConversionTestCase(CustomTestCase): + def test_conversion(self): + fn = OpenAIServingResponses._chat_tool_choice + for s in ("auto", "required", "none"): + self.assertEqual(fn(s), s) + # Input is an effective_tool_choice() result, so the only object form + # reaching here is a named function; degrading the rest to "auto" + # happens there, once, so the echoed and the honored value agree. + self.assertEqual( + fn({"type": "function", "name": "get_weather"}), + {"type": "function", "function": {"name": "get_weather"}}, + ) + + +class ShouldEmitNormalTextTestCase(CustomTestCase): + def test_whitespace_suppressed_only_while_a_tool_is_open(self): + emit = _should_emit_normal_text_as_message + self.assertFalse(emit("", any_tool_call_in_progress=False)) + # whitespace between tool blocks is an inter-call separator, not content + self.assertFalse(emit("\n", any_tool_call_in_progress=True)) + self.assertTrue(emit("\n", any_tool_call_in_progress=False)) + self.assertTrue(emit("hello", any_tool_call_in_progress=True)) + + +class EnginePassthroughTestCase(CustomTestCase): + """Both flags cross hops with no type contract, and dropping either fails + silently.""" + + def _capture(self, serving, request): + # Let the real _process_messages run: it is the hop that turns + # skip_special_tokens off, so mocking it would make that assertion vacuous. + # chat_template_name=None routes it through the tokenizer's template + # (mocked) instead of the conversation registry, which has no fixture entry. + serving.default_chat_template_kwargs = {} + serving.template_manager.chat_template_name = None + captured = {} + + async def fake_generate( + request_id, + request_prompt, + adapted_request, + sampling_params, + context, + **kwargs, + ): + captured["adapted_request"] = adapted_request + captured["sampling_params"] = sampling_params + context.append_output( + { + "text": "ok", + "meta_info": { + "prompt_tokens": 1, + "completion_tokens": 1, + "cached_tokens": 0, + }, + } + ) + yield context + + serving._generate_with_builtin_tools = fake_generate + asyncio.run(serving.create_responses(request)) + return captured + + def test_require_reasoning_forwarded_when_reasoning_parser_configured(self): + serving = make_serving() + serving.reasoning_parser = "deepseek-r1" + serving.template_manager.reasoning_config = ReasoningToggleConfig( + toggle_param="thinking", default_enabled=True + ) + + captured = self._capture( + serving, ResponsesRequest(model="x", input="hi", store=False) + ) + + self.assertTrue(captured["adapted_request"].require_reasoning) + + def test_prefilled_think_template_opens_the_parser(self): + """``force_reasoning`` is a template property, not a request one, so it + drives the parser but never the engine flag -- as on the chat path.""" + serving = make_serving() + serving.reasoning_parser = "deepseek-r1" + serving.template_manager.force_reasoning = True + + with patch( + "sglang.srt.entrypoints.openai.serving_responses.ReasoningParser" + ) as parser_cls: + parser_cls.return_value.parse_non_stream.return_value = (None, "hi") + serving._make_response_output_items( + ResponsesRequest(model="x", input="hi", store=False), + "hi", + tokenizer=Mock(), + require_reasoning=False, + ) + + self.assertTrue(parser_cls.call_args.kwargs["force_reasoning"]) + + def test_require_reasoning_false_without_reasoning_parser(self): + serving = make_serving() + serving.reasoning_parser = None + + captured = self._capture( + serving, ResponsesRequest(model="x", input="hi", store=False) + ) + + self.assertFalse(captured["adapted_request"].require_reasoning) + + def test_skip_special_tokens_disabled_for_tool_requests(self): + # _process_messages turns it off so tool-call markers survive detokenize; + # create_responses must re-apply it to the engine sampling dict. + serving = make_serving() + serving.tool_call_parser = "qwen25" + + captured = self._capture( + serving, + ResponsesRequest( + model="x", + input="weather", + store=False, + tools=[ + { + "type": "function", + "name": "get_weather", + "parameters": {"type": "object"}, + } + ], + ), + ) + + self.assertFalse(captured["sampling_params"]["skip_special_tokens"]) + + +class CancelIdempotencyTestCase(CustomTestCase): + def test_cancelling_a_terminal_response_returns_it_not_an_error(self): + from sglang.srt.entrypoints.openai.protocol import ResponsesResponse + + for status in ("cancelled", "completed"): + serving = make_serving() + resp = ResponsesResponse.from_request( + ResponsesRequest(model="x", input="hi", store=False), + sampling_params={}, + model_name="x", + created_time=0, + output=[], + status=status, + usage=None, + ) + serving.response_store[resp.id] = resp + + out = asyncio.run(serving.cancel_responses(resp.id)) + + self.assertIs(out, resp, status) + self.assertEqual(out.status, status) + + +class StreamingLogprobsRejectionTestCase(CustomTestCase): + def test_stream_with_logprobs_include_rejected(self): + import orjson + + serving = make_serving() + request = ResponsesRequest( + model="x", + input="hi", + store=False, + stream=True, + include=["message.output_text.logprobs"], + ) + result = asyncio.run(serving.create_responses(request)) + self.assertEqual(result.status_code, 400) + body = orjson.loads(result.body) + self.assertIn("streaming mode", body["error"]["message"]) + + 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 index 479b6bb5c..5c00ce3c5 100644 --- a/test/registered/unit/entrypoints/openai/test_serving_responses_stream.py +++ b/test/registered/unit/entrypoints/openai/test_serving_responses_stream.py @@ -1,67 +1,23 @@ -import asyncio import unittest -from unittest.mock import Mock, patch +from unittest.mock import patch from utils import ( - collect_stream_events, + StreamFixture, + engine_chunk, event_payloads, event_types, find_completed_event, make_serving, ) -from sglang.srt.entrypoints.openai.protocol import ( - RequestResponseMetadata, - ResponsesRequest, -) +from sglang.srt.entrypoints.openai.protocol import ResponsesRequest from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase -register_cpu_ci(est_time=4, suite="base-a-test-cpu") +register_cpu_ci(est_time=7, suite="base-a-test-cpu") -class _StreamFixture: - def __init__(self, serving, request, *, require_reasoning=False): - self.serving = serving - self.request = request - self.require_reasoning = require_reasoning - 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, - require_reasoning=self.require_reasoning, - ) - ) - - 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): +class NonHarmonyStreamTestCase(CustomTestCase): def test_reasoning_parser_uses_processed_reasoning_state(self): serving = make_serving() serving.reasoning_parser = "deepseek-r1" @@ -71,8 +27,8 @@ class NonHarmonyStreamTestCase(unittest.TestCase): "sglang.srt.entrypoints.openai.serving_responses.ReasoningParser" ) as parser_cls: parser_cls.return_value.parse_stream_chunk.return_value = (None, "done") - fixture = _StreamFixture(serving, request, require_reasoning=True) - fixture.run([_engine_chunk("done", 1, finish=True)]) + fixture = StreamFixture(serving, request, require_reasoning=True) + fixture.run([engine_chunk("done", 1, finish=True)]) self.assertTrue(parser_cls.call_args.kwargs["force_reasoning"]) @@ -82,12 +38,12 @@ class NonHarmonyStreamTestCase(unittest.TestCase): serving.tool_call_parser = None request = ResponsesRequest(model="x", input="hi", stream=True, store=False) - fixture = _StreamFixture(serving, request) + fixture = StreamFixture(serving, request) events = fixture.run( [ - _engine_chunk("Hel", 1), - _engine_chunk("Hello", 2), - _engine_chunk("Hello world", 4, finish=True), + engine_chunk("Hel", 1), + engine_chunk("Hello", 2), + engine_chunk("Hello world", 4, finish=True), ] ) @@ -134,10 +90,10 @@ class NonHarmonyStreamTestCase(unittest.TestCase): while sent < len(payload): sent += min(8, len(payload) - sent) chunks.append( - _engine_chunk(payload[:sent], sent, finish=sent == len(payload)) + engine_chunk(payload[:sent], sent, finish=sent == len(payload)) ) - fixture = _StreamFixture(serving, request) + fixture = StreamFixture(serving, request) events = fixture.run(chunks) types = event_types(events) @@ -193,9 +149,9 @@ class NonHarmonyStreamTestCase(unittest.TestCase): StreamingParseResult(normal_text="It's sunny.", calls=[]), ] chunks = [ - _engine_chunk(" " * 3, 3), - _engine_chunk(" " * 10, 10), - _engine_chunk(" " * 14, 14, finish=True), + engine_chunk(" " * 3, 3), + engine_chunk(" " * 10, 10), + engine_chunk(" " * 14, 14, finish=True), ] script_iter = iter(scripted) @@ -211,7 +167,7 @@ class NonHarmonyStreamTestCase(unittest.TestCase): parser_cls.return_value.parse_stream_chunk.side_effect = ( fake_parse_stream_chunk ) - fixture = _StreamFixture(serving, request) + fixture = StreamFixture(serving, request) events = fixture.run(chunks) completed = find_completed_event(events) @@ -223,5 +179,109 @@ class NonHarmonyStreamTestCase(unittest.TestCase): self.assertEqual(output[2]["content"][0]["text"], "It's sunny.") +class MultiToolCallStreamingOrderTestCase(CustomTestCase): + """The wire order of message / function_call items across tool-call deltas.""" + + def setUp(self): + from sglang.srt.function_call.qwen3_coder_detector import Qwen3CoderDetector + + self.serving = make_serving() + self.serving.tool_call_parser = "qwen3_coder" + self.serving.reasoning_parser = None + + det = Qwen3CoderDetector() + s, e = det.tool_call_start_token, det.tool_call_end_token + fp, fe = det.tool_call_prefix, det.function_end_token + pp, pe = det.parameter_prefix, det.parameter_end_token + self.weather = f"{s}{fp}get_weather>{pp}city>Beijing{pe}{fe}{e}" + self.time = f"{s}{fp}get_time>{pp}tz>UTC{pe}{fe}{e}" + # a prefix of ``weather`` that stops mid-arguments + self.weather_head = f"{s}{fp}get_weather>{pp}city>Beij" + + def _seq(self, texts, *names): + """Stream cumulative ``texts`` (last one final) and return (type, payload).""" + request = ResponsesRequest( + model="x", + input="weather and time", + store=False, + tools=[ + {"type": "function", "name": n, "parameters": {"type": "object"}} + for n in names + ], + ) + chunks = [engine_chunk(t) for t in texts] + chunks.append(engine_chunk(texts[-1], finish=True)) + return StreamFixture(self.serving, request).run_seq(chunks) + + @staticmethod + def _added(seq): + return [ + (p["output_index"], p["item"].get("type")) + for t, p in seq + if t == "response.output_item.added" + ] + + @staticmethod + def _done_calls(seq): + return [ + p["item"] + for t, p in seq + if t == "response.output_item.done" + and p["item"].get("type") == "function_call" + ] + + def test_prior_tool_call_done_before_next_added(self): + full = self.weather + "\n" + self.time + seq = self._seq( + [self.weather, self.weather + "\n", full], "get_weather", "get_time" + ) + + def position(pred): + return next(i for i, (t, p) in enumerate(seq) if pred(t, p)) + + done0 = position( + lambda t, p: t == "response.output_item.done" and p["output_index"] == 0 + ) + added1 = position( + lambda t, p: t == "response.output_item.added" and p["output_index"] == 1 + ) + self.assertLess(done0, added1) + + items = self._done_calls(seq) + self.assertEqual(sorted(i["name"] for i in items), ["get_time", "get_weather"]) + + def test_prose_before_tool_call_keeps_message_first(self): + """Prose and a tool-call start in one delta: the message item must come + first, since the prose preceded the call.""" + # One delta spanning prose + the whole call, as spec decoding or + # --stream-interval > 1 produces. + seq = self._seq(["Let me check." + self.weather], "get_weather") + + added = self._added(seq) + message_index = next(i for i, kind in added if kind == "message") + call_index = next(i for i, kind in added if kind == "function_call") + self.assertLess(message_index, call_index) + + # The call must not be split across two items by the reordering. + self.assertEqual(len([k for _, k in added if k == "function_call"]), 1) + + def test_call_tail_prose_and_next_call_in_one_delta(self): + """One delta closing tool1, carrying prose, and opening tool2 needs both + orders at once: tool1's trailing "}" must be drained before the prose + closes every open item, and tool2 must land after the message.""" + seq = self._seq( + [self.weather_head, self.weather + "Here you go." + self.time], + "get_weather", + "get_time", + ) + + items = self._done_calls(seq) + # No duplicate item invented for the already-closed call, and no call + # left nameless by being reopened from an args-only fragment. + self.assertEqual(len(items), 2) + self.assertTrue(all(i["name"] for i in items)) + self.assertEqual(items[0]["arguments"], '{"city": "Beijing"}') + + if __name__ == "__main__": unittest.main() diff --git a/test/registered/unit/entrypoints/openai/utils.py b/test/registered/unit/entrypoints/openai/utils.py index 36c2c1fc7..7917a5002 100644 --- a/test/registered/unit/entrypoints/openai/utils.py +++ b/test/registered/unit/entrypoints/openai/utils.py @@ -20,10 +20,12 @@ from sglang.test.test_utils import maybe_stub_sgl_kernel maybe_stub_sgl_kernel() +import asyncio import json from typing import AsyncIterator from unittest.mock import Mock +from sglang.srt.entrypoints.openai.protocol import RequestResponseMetadata from sglang.srt.entrypoints.openai.serving_responses import OpenAIServingResponses from sglang.test.ci.ci_register import register_cpu_ci @@ -115,3 +117,52 @@ def find_completed_event(events: list[str]) -> dict: if lines and lines[0] == "event: response.completed": return json.loads(lines[1][len("data: ") :]) raise AssertionError("response.completed event missing from stream") + + +def engine_chunk(text, completion_tokens=1, *, 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 StreamFixture: + """Drives ``responses_stream_generator_non_harmony`` over a chunk list.""" + + def __init__(self, serving, request, *, require_reasoning=False): + self.serving = serving + self.request = request + self.require_reasoning = require_reasoning + self.request_metadata = RequestResponseMetadata(request_id=request.request_id) + + def run(self, chunks) -> list[str]: + 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, + require_reasoning=self.require_reasoning, + ) + ) + + return asyncio.run(collect()) + + def run_seq(self, chunks) -> list[tuple]: + """``run`` plus (event type, payload) pairing, the common assertion shape.""" + events = self.run(chunks) + return list(zip(event_types(events), event_payloads(events)))