diff --git a/python/sglang/srt/entrypoints/harmony_utils.py b/python/sglang/srt/entrypoints/harmony_utils.py index 1d1b4d8ed..dd040e07c 100644 --- a/python/sglang/srt/entrypoints/harmony_utils.py +++ b/python/sglang/srt/entrypoints/harmony_utils.py @@ -9,8 +9,11 @@ from typing import Optional, Union import orjson from openai.types.responses import ( + ResponseCodeInterpreterToolCall, ResponseOutputItem, - ResponseOutputMessage, +) +from openai.types.responses import ResponseOutputMessage as OpenAIResponseOutputMessage +from openai.types.responses import ( ResponseOutputText, ResponseReasoningItem, ) @@ -43,6 +46,11 @@ from openai_harmony import ( from sglang.srt.entrypoints.openai.protocol import ( ReasoningEffortTier, ResponseInputOutputItem, + ResponseOutputMessage, +) +from sglang.srt.entrypoints.openai.responses_adapters import ( + decode_reasoning_state, + encode_custom_tool_input, ) from sglang.srt.utils import random_uuid @@ -163,35 +171,66 @@ def parse_response_input( # 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") + c + for c in content + if c.get("type") in ("text", "input_text", "output_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": + if role == "assistant" and response_msg.get("phase") is not None: + msg = msg.with_channel( + "final" if response_msg["phase"] == "final_answer" else "commentary" + ) + elif response_msg["type"] in ("function_call_output", "custom_tool_call_output"): call_id = response_msg["call_id"] - call_response: Optional[ResponseFunctionToolCall] = None - for prev_response in reversed(prev_responses): + call_response = None + for previous in reversed(prev_responses): + if not isinstance(previous, dict): + previous = previous.model_dump() if ( - isinstance(prev_response, ResponseFunctionToolCall) - and prev_response.call_id == call_id + previous.get("type") in ("function_call", "custom_tool_call") + and previous.get("call_id") == call_id ): - call_response = prev_response + call_response = previous break if call_response is None: raise ValueError(f"No call message found for {call_id}") + output = response_msg.get("output", "") + if isinstance(output, list): + output = "".join( + part.get("text", "") for part in output if isinstance(part, dict) + ) msg = Message.from_author_and_content( - Author.new(Role.TOOL, f"functions.{call_response.name}"), - response_msg["output"], + Author.new(Role.TOOL, f"functions.{call_response['name']}"), + output, ) elif response_msg["type"] == "reasoning": - content = response_msg["content"] - assert len(content) == 1 - msg = Message.from_role_and_content(Role.ASSISTANT, content[0]["text"]) - elif response_msg["type"] == "function_call": - msg = Message.from_role_and_content(Role.ASSISTANT, response_msg["arguments"]) + text = "" + for field in ("summary", "content"): + text = "\n".join( + part.get("text", "") + for part in response_msg.get(field) or [] + if isinstance(part, dict) + ) + if text: + break + if not text: + text = decode_reasoning_state(response_msg.get("encrypted_content")) or "" + msg = Message.from_role_and_content(Role.ASSISTANT, text).with_channel( + "analysis" + ) + elif response_msg["type"] in ("function_call", "custom_tool_call"): + arguments = ( + encode_custom_tool_input(response_msg.get("input") or "") + if response_msg["type"] == "custom_tool_call" + else response_msg.get("arguments") or "{}" + ) + if isinstance(arguments, dict): + arguments = orjson.dumps(arguments).decode() + msg = Message.from_role_and_content(Role.ASSISTANT, arguments) msg = msg.with_channel("commentary") msg = msg.with_recipient(f"functions.{response_msg['name']}") msg = msg.with_content_type("json") @@ -201,10 +240,13 @@ def parse_response_input( def parse_response_output(output: ResponseOutputItem) -> Message: - if isinstance(output, ResponseOutputMessage): + if isinstance(output, OpenAIResponseOutputMessage): role = output.role contents = [TextContent(text=c.text) for c in output.content] msg = Message.from_role_and_contents(role, contents) + phase = getattr(output, "phase", None) + if phase is not None: + msg = msg.with_channel("final" if phase == "final_answer" else "commentary") return msg elif isinstance(output, ResponseFunctionToolCall): msg = Message.from_role_and_content(Role.ASSISTANT, output.arguments) @@ -282,6 +324,17 @@ def parse_output_message(message: Message): type="web_search_call", ) output_items.append(web_search_item) + elif recipient is not None and recipient.startswith("python"): + output_items.append( + ResponseCodeInterpreterToolCall( + id=f"ci_{random_uuid()}", + type="code_interpreter_call", + code="".join(content.text for content in message.content), + container_id="auto", + outputs=[], + status="completed", + ) + ) elif message.channel == "analysis": for content in message.content: reasoning_item = ResponseReasoningItem( @@ -296,9 +349,9 @@ def parse_output_message(message: Message): status=None, ) output_items.append(reasoning_item) - elif message.channel == "commentary": + elif message.channel == "commentary" and message.recipient is not None: if message.recipient.startswith("functions."): - function_name = message.recipient.split(".")[-1] + function_name = message.recipient.removeprefix("functions.") for content in message.content: random_id = random_uuid() response_item = ResponseFunctionToolCall( @@ -327,7 +380,9 @@ def parse_output_message(message: Message): output_items.append(reasoning_item) else: raise ValueError(f"Unknown recipient: {message.recipient}") - elif message.channel == "final": + elif message.channel == "final" or ( + message.channel == "commentary" and message.recipient is None + ): contents = [] for content in message.content: output_text = ResponseOutputText( @@ -338,6 +393,7 @@ def parse_output_message(message: Message): ) contents.append(output_text) text_item = ResponseOutputMessage( + phase="final_answer" if message.channel == "final" else "commentary", id=f"msg_{random_uuid()}", content=contents, role=message.author.role, @@ -359,6 +415,17 @@ def parse_remaining_state(parser: StreamableParser): if current_recipient is not None and current_recipient.startswith("browser."): return [] + if current_recipient is not None and current_recipient.startswith("python"): + return [ + ResponseCodeInterpreterToolCall( + id=f"ci_{random_uuid()}", + type="code_interpreter_call", + code=parser.current_content, + container_id="auto", + outputs=[], + status="in_progress", + ) + ] if parser.current_channel == "analysis": reasoning_item = ResponseReasoningItem( id=f"rs_{random_uuid()}", @@ -372,7 +439,21 @@ def parse_remaining_state(parser: StreamableParser): status=None, ) return [reasoning_item] - elif parser.current_channel == "final": + elif current_recipient is not None and current_recipient.startswith("functions."): + random_id = random_uuid() + return [ + ResponseFunctionToolCall( + id=f"ft_{random_id}", + call_id=f"call_{random_id}", + type="function_call", + name=current_recipient.removeprefix("functions."), + arguments=parser.current_content, + status="in_progress", + ) + ] + elif parser.current_channel == "final" or ( + parser.current_channel == "commentary" and current_recipient is None + ): output_text = ResponseOutputText( text=parser.current_content, annotations=[], # TODO @@ -380,6 +461,7 @@ def parse_remaining_state(parser: StreamableParser): logprobs=None, # TODO ) text_item = ResponseOutputMessage( + phase="final_answer" if parser.current_channel == "final" else "commentary", id=f"msg_{random_uuid()}", content=[output_text], role="assistant", diff --git a/python/sglang/srt/entrypoints/openai/protocol.py b/python/sglang/srt/entrypoints/openai/protocol.py index 5199cf7f8..b0825dfba 100644 --- a/python/sglang/srt/entrypoints/openai/protocol.py +++ b/python/sglang/srt/entrypoints/openai/protocol.py @@ -39,11 +39,14 @@ from openai.types.responses import ( ResponseFunctionToolCall, ResponseInputItemParam, ResponseOutputItem, - ResponseOutputMessage, +) +from openai.types.responses import ResponseOutputMessage as OpenAIResponseOutputMessage +from openai.types.responses import ( ResponseOutputText, ResponseReasoningItem, ResponseTextConfig, ) +from openai.types.responses.easy_input_message_param import EasyInputMessageParam from openai.types.responses.response import ToolChoice from openai.types.responses.response_format_text_json_schema_config import ( ResponseFormatTextJSONSchemaConfig, @@ -713,6 +716,7 @@ class ChatCompletionMessageGenericParam(BaseModel): ) tool_call_id: Optional[str] = None name: Optional[str] = None + phase: Optional[Literal["commentary", "final_answer"]] = None reasoning_content: Optional[str] = None tool_calls: Optional[List[ToolCall]] = Field(default=None, examples=[None]) tools: Optional[List[Tool]] = Field(default=None, examples=[None]) @@ -1581,6 +1585,9 @@ class ResponseTool(BaseModel): strict: bool = False # Inner schemas for ``namespace`` tools. tools: Optional[List[Dict[str, Any]]] = None + # Input format of a ``custom`` tool: {"type": "text"} or + # {"type": "grammar", "syntax": ..., "definition": ...}. + format: Optional[Dict[str, Any]] = None @model_validator(mode="after") def validate_function_tool(self) -> ResponseTool: @@ -1589,7 +1596,17 @@ class ResponseTool(BaseModel): return self +class ResponseInputMessageParam(EasyInputMessageParam, total=False): + phase: Optional[Literal["commentary", "final_answer"]] + + +class ResponseOutputMessage(OpenAIResponseOutputMessage): + phase: Optional[Literal["commentary", "final_answer"]] = None + + ResponseInputOutputItem: TypeAlias = Union[ + ResponseInputMessageParam, + ResponseOutputMessage, ResponseInputItemParam, "ResponseReasoningItem", ResponseFunctionToolCall, @@ -1782,20 +1799,24 @@ class ResponsesRequest(BaseModel): def is_include_output_logprobs(self) -> bool: return bool(self.include and "message.output_text.logprobs" in self.include) + def is_include_encrypted_reasoning(self) -> bool: + return bool(self.include and "reasoning.encrypted_content" 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.""" + object forms only a named ``function`` / ``custom`` tool 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: + if tool_choice.get("type") in ("function", "custom") and name: return {"type": "function", "name": name} return "auto" @@ -1897,7 +1918,12 @@ class ResponsesResponse(BaseModel): model: str output: List[ - Union[ResponseOutputItem, ResponseReasoningItem, ResponseFunctionToolCall] + Union[ + ResponseOutputMessage, + ResponseOutputItem, + ResponseReasoningItem, + ResponseFunctionToolCall, + ] ] = Field(default_factory=list) status: Literal[ "queued", "in_progress", "completed", "incomplete", "failed", "cancelled" @@ -1957,7 +1983,12 @@ class ResponsesResponse(BaseModel): model_name: str, created_time: int, output: List[ - Union[ResponseOutputItem, ResponseReasoningItem, ResponseFunctionToolCall] + Union[ + ResponseOutputMessage, + ResponseOutputItem, + ResponseReasoningItem, + ResponseFunctionToolCall, + ] ], status: str, usage: Optional[UsageInfo], @@ -1983,7 +2014,7 @@ class ResponsesResponse(BaseModel): try: if isinstance(it, ResponseOutputText): continue - elif isinstance(it, ResponseOutputMessage): + elif isinstance(it, OpenAIResponseOutputMessage): if not it.content: continue for c in it.content: @@ -2083,7 +2114,11 @@ class ResponseReasoningTextContent(BaseModel): ResponseInputOutputItem: TypeAlias = Union[ - ResponseInputItemParam, "ResponseReasoningItem", ResponseFunctionToolCall + ResponseInputMessageParam, + ResponseOutputMessage, + ResponseInputItemParam, + "ResponseReasoningItem", + ResponseFunctionToolCall, ] diff --git a/python/sglang/srt/entrypoints/openai/responses_adapters.py b/python/sglang/srt/entrypoints/openai/responses_adapters.py new file mode 100644 index 000000000..0d82012bb --- /dev/null +++ b/python/sglang/srt/entrypoints/openai/responses_adapters.py @@ -0,0 +1,215 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Translations between Responses-API wire shapes and SGLang's chat internals. + +Two things the Responses API expresses that the chat-completions path has no +representation for: + +* ``custom`` tools, whose payload is freeform text rather than JSON-object + arguments. Each is surfaced to the model as a function tool with a single + string property, and the resulting call is translated back into a + ``custom_tool_call``. +* ``reasoning.encrypted_content``, the opaque blob a ``store=false`` client + replays to hand a reasoning trace back to the server. +""" + +from __future__ import annotations + +import base64 +import json +import zlib +from typing import Any, Dict, Optional, Set, Tuple + +CUSTOM_TOOL_INPUT_KEY = "input" + +_SIMPLE_ESCAPES = { + '"': '"', + "\\": "\\", + "/": "/", + "b": "\b", + "f": "\f", + "n": "\n", + "r": "\r", + "t": "\t", +} + + +def custom_tool_parameters() -> Dict[str, Any]: + return { + "type": "object", + "properties": { + CUSTOM_TOOL_INPUT_KEY: { + "type": "string", + "description": "The freeform text payload for this tool.", + } + }, + "required": [CUSTOM_TOOL_INPUT_KEY], + "additionalProperties": False, + } + + +def custom_tool_description( + description: Optional[str], tool_format: Optional[Dict[str, Any]] +) -> Optional[str]: + """Fold a custom tool's description and its input format into one blob. + + A ``grammar`` format is described to the model rather than enforced by the + grammar backend: the payload is emitted inside the model's own tool-call + envelope, so constrained decoding cannot be scoped to it. + """ + parts = [] + if description: + parts.append(description) + if isinstance(tool_format, dict) and tool_format.get("type") == "grammar": + syntax = tool_format.get("syntax") or "lark" + definition = tool_format.get("definition") or "" + parts.append( + f"The payload must conform to this {syntax} grammar:\n{definition}" + ) + return "\n\n".join(parts) or None + + +def custom_tool_names(tools: Any) -> Set[str]: + return {tool.name for tool in tools or [] if tool.type == "custom" and tool.name} + + +def encode_custom_tool_input(payload: str) -> str: + return json.dumps({CUSTOM_TOOL_INPUT_KEY: payload}, ensure_ascii=False) + + +def decode_custom_tool_input(arguments: str) -> str: + """Recover the payload from a completed shim call's arguments. + + Falls back to the raw string so a model that ignored the wrapper and + emitted bare text still produces a usable call. + """ + try: + parsed = json.loads(arguments) + except ValueError: + return arguments + if isinstance(parsed, dict): + value = parsed.get(CUSTOM_TOOL_INPUT_KEY) + if isinstance(value, str): + return value + if len(parsed) == 1: + only = next(iter(parsed.values())) + if isinstance(only, str): + return only + return arguments + + +def _payload_start(buffer: str) -> Optional[int]: + key = f'"{CUSTOM_TOOL_INPUT_KEY}"' + key_at = buffer.find(key) + if key_at < 0: + return None + colon = buffer.find(":", key_at + len(key)) + if colon < 0: + return None + quote = buffer.find('"', colon + 1) + return quote + 1 if quote >= 0 else None + + +def _unicode_escape(buffer: str, i: int) -> Optional[Tuple[str, int]]: + if i + 6 > len(buffer): + return None + try: + code = int(buffer[i + 2 : i + 6], 16) + except ValueError: + return None + if not 0xD800 <= code <= 0xDBFF: + return chr(code), i + 6 + if i + 12 > len(buffer) or buffer[i + 6 : i + 8] != "\\u": + return None + try: + low = int(buffer[i + 8 : i + 12], 16) + except ValueError: + return None + if not 0xDC00 <= low <= 0xDFFF: + return None + return chr(0x10000 + ((code - 0xD800) << 10) + (low - 0xDC00)), i + 12 + + +def decode_custom_tool_input_prefix(buffer: str) -> str: + """Longest fully decodable prefix of the payload in a partial argument buffer. + + Streaming tool-call parsers hand out argument JSON in fragments, so the + payload is un-escaped incrementally to keep ``custom_tool_call_input`` + deltas in step with the value the item finally reports. + """ + start = _payload_start(buffer) + if start is None: + return "" + out = [] + i, n = start, len(buffer) + while i < n: + ch = buffer[i] + if ch == '"': + break + if ch != "\\": + out.append(ch) + i += 1 + continue + if i + 1 >= n: + break + simple = _SIMPLE_ESCAPES.get(buffer[i + 1]) + if simple is not None: + out.append(simple) + i += 2 + continue + if buffer[i + 1] == "u": + decoded = _unicode_escape(buffer, i) + if decoded is None: + break + out.append(decoded[0]) + i = decoded[1] + continue + break + return "".join(out) + + +DEVELOPER_BLOCK_LABEL = "Developer instructions:" + + +def label_developer_content(content: Any) -> Any: + """Prefix a ``developer`` message's text with its tier label. + + Chat templates recognize system/user/assistant/tool, so a developer message + collapses to ``system``. Labelling the block keeps the tier legible instead + of flattening it into the surrounding instructions, where a model can read + it as one more conversational turn and let a later user message win. + """ + if isinstance(content, str): + return f"{DEVELOPER_BLOCK_LABEL}\n{content}" + if isinstance(content, list): + for index, part in enumerate(content): + if isinstance(part, dict) and isinstance(part.get("text"), str): + labelled = dict(part) + labelled["text"] = f"{DEVELOPER_BLOCK_LABEL}\n{part['text']}" + return [*content[:index], labelled, *content[index + 1 :]] + return [{"type": "input_text", "text": DEVELOPER_BLOCK_LABEL}, *content] + return content + + +_REASONING_STATE_PREFIX = "sglang-reasoning-v1." + + +def encode_reasoning_state(text: str) -> str: + """Pack a reasoning trace into the blob ``reasoning.encrypted_content`` carries. + + A ``store=false`` client replays reasoning items verbatim, so the trace has + to survive the round trip without server-side state. SGLang holds no key, + so this is an encoded -- not cryptographically protected -- payload: treat + it as opaque to clients, not as a confidentiality boundary. + """ + packed = base64.urlsafe_b64encode(zlib.compress(text.encode("utf-8"))) + return _REASONING_STATE_PREFIX + packed.decode("ascii") + + +def decode_reasoning_state(blob: Any) -> Optional[str]: + if not isinstance(blob, str) or not blob.startswith(_REASONING_STATE_PREFIX): + return None + try: + raw = base64.urlsafe_b64decode(blob[len(_REASONING_STATE_PREFIX) :]) + return zlib.decompress(raw).decode("utf-8") + except (ValueError, zlib.error, UnicodeDecodeError): + return None diff --git a/python/sglang/srt/entrypoints/openai/serving_responses.py b/python/sglang/srt/entrypoints/openai/serving_responses.py index 052f4c3f4..d82033393 100644 --- a/python/sglang/srt/entrypoints/openai/serving_responses.py +++ b/python/sglang/srt/entrypoints/openai/serving_responses.py @@ -18,10 +18,10 @@ import orjson from fastapi import Request from fastapi.responses import ORJSONResponse from openai.types.responses import ( - ResponseOutputMessage, ResponseOutputText, ResponseReasoningItem, ) +from openai.types.responses.response_custom_tool_call import ResponseCustomToolCall 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 ( @@ -61,11 +61,23 @@ from sglang.srt.entrypoints.openai.protocol import ( MessageProcessingResult, PromptTokenUsageInfo, RequestResponseMetadata, + ResponseOutputMessage, ResponsesRequest, ResponsesResponse, Tool, UsageInfo, ) +from sglang.srt.entrypoints.openai.responses_adapters import ( + custom_tool_description, + custom_tool_names, + custom_tool_parameters, + decode_custom_tool_input, + decode_custom_tool_input_prefix, + decode_reasoning_state, + encode_custom_tool_input, + encode_reasoning_state, + label_developer_content, +) 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 @@ -189,9 +201,7 @@ class OpenAIServingResponses(OpenAIServingChat): # Message storage for conversation continuity # Note: In production, this should use a proper storage backend (Redis, database) # with TTL/expiration to prevent memory leaks - self.msg_store: dict[ - str, Union[list[ChatCompletionMessageParam], list[OpenAIMessage]] - ] = {} + self.msg_store: dict[str, Union[list[dict], list[OpenAIMessage]]] = {} self.background_tasks: dict[str, asyncio.Task] = {} @@ -235,6 +245,33 @@ class OpenAIServingResponses(OpenAIServingChat): def _request_id_prefix(self) -> str: return "resp_" + def _known_model_names(self) -> set[str]: + """Model ids a caller may address, mirroring what ``/v1/models`` lists.""" + names = {self.tokenizer_manager.served_model_name} + registry = getattr(self.tokenizer_manager, "lora_registry", None) + if registry is not None: + names.update(registry.get_all_adapters().keys()) + return names + + def _validate_model(self, model: Optional[str]) -> Optional[ORJSONResponse]: + """Reject an unknown ``model``, as the Responses API does. + + ``None`` means "whatever is loaded". A LoRA adapter may be addressed + either by name or through the ``base-model:adapter`` form. + """ + if model is None: + return None + base_model, _ = self._parse_model_parameter(model) + known = self._known_model_names() + if model in known or base_model in known: + return None + return self.create_error_response( + message=f"The model '{model}' does not exist", + err_type="invalid_request_error", + status_code=HTTPStatus.NOT_FOUND, + param="model", + ) + async def create_responses( self, request: ResponsesRequest, @@ -244,16 +281,31 @@ class OpenAIServingResponses(OpenAIServingChat): if not self.tokenizer_manager: return self.create_error_response("Model not loaded") + model_error = self._validate_model(request.model) + if model_error is not None: + return model_error + # FIXME: If the engine is dead, raise an error # This is required for the streaming case - # ``tool_choice="required"`` only works with ``function`` tools. + # ``tool_choice="required"`` needs a tool the parser can actually force. if request.tool_choice == "required" and not any( - tool.type == "function" for tool in (request.tools or []) + tool.type in ("function", "custom") 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.' + 'type="function" or type="custom"; other built-in tool types ' + "cannot be forced." + ) + + tool_choice = request.effective_tool_choice() + if isinstance(tool_choice, dict) and not any( + tool.type in ("function", "custom") and tool.name == tool_choice["name"] + for tool in request.tools or [] + ): + return self.create_error_response( + f"Tool {tool_choice['name']!r} is not declared in tools", + param="tool_choice", ) # harmony emits raw tokens; per-token logprobs aren't wired there. @@ -496,7 +548,11 @@ class OpenAIServingResponses(OpenAIServingChat): # Store the input messages if request.store: - self.msg_store[request.request_id] = messages + self.msg_store[request.request_id] = ( + messages[2:] + if self.use_harmony + else self._response_input_history(request) + ) if request.background and not request.stream: created_time = int(time.time()) @@ -630,9 +686,19 @@ class OpenAIServingResponses(OpenAIServingChat): request: ResponsesRequest, prev_response: Optional[ResponsesResponse], ): - if request.tool_choice != "auto": - raise NotImplementedError( - "Only 'auto' tool_choice is supported in response API" + tool_choice = request.effective_tool_choice() + if tool_choice != "auto": + requirement = ( + "restrict the recipient to the named tool" + if isinstance(tool_choice, dict) + else "require a tool recipient" + if tool_choice == "required" + else "exclude tool recipients" + ) + raise ValueError( + f"Harmony tool_choice={request.tool_choice!r} cannot {requirement}: " + "recipient-header decoding constraints are not implemented; " + "use tool_choice='auto'." ) messages = self._construct_input_messages_with_harmony(request, prev_response) prompt_token_ids = render_for_completion(messages) @@ -651,6 +717,7 @@ class OpenAIServingResponses(OpenAIServingChat): created_time: Optional[int] = None, *, require_reasoning: bool, + output_items: Optional[list] = None, ) -> Union[ResponsesResponse, ORJSONResponse]: if created_time is None: created_time = int(time.time()) @@ -664,15 +731,21 @@ class OpenAIServingResponses(OpenAIServingChat): return self.create_error_response(str(e)) status = "completed" + finish_reason = None if self.use_harmony: assert isinstance(context, HarmonyContext) - output = self._make_response_output_items_with_harmony(context) + output = ( + output_items + if output_items is not None + else self._make_response_output_items_with_harmony(context) + ) # 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 num_reasoning_tokens = context.num_reasoning_tokens - status = self._status_from_finish_reason(context.finish_reason) + finish_reason = context.finish_reason + status = self._status_from_finish_reason(finish_reason) else: assert isinstance(context, SimpleContext) final_res = context.last_output @@ -705,7 +778,8 @@ class OpenAIServingResponses(OpenAIServingChat): 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")) + finish_reason = meta_info.get("finish_reason") + status = self._status_from_finish_reason(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 @@ -757,12 +831,26 @@ class OpenAIServingResponses(OpenAIServingChat): usage=usage, ) + response.error = self._error_from_finish_reason(finish_reason) if request.store: async with self.response_store_lock: stored_response = self.response_store.get(response.id) # If the response is already cancelled, don't update it if stored_response is None or stored_response.status != "cancelled": self.response_store[response.id] = response + if self.use_harmony: + self.msg_store[response.id] = ( + self.msg_store.get(response.id, []) + context.messages + if isinstance(context, StreamingHarmonyContext) + else context.messages[2:] + ) + else: + self.msg_store[response.id] = self._response_input_history( + request + ) + [ + item.model_dump(exclude_none=True) + for item in response.output + ] return response @@ -770,16 +858,93 @@ class OpenAIServingResponses(OpenAIServingChat): def _wants_reasoning_summary(request: ResponsesRequest) -> bool: return request.reasoning is not None and request.reasoning.summary is not None + @classmethod + def _make_reasoning_item( + cls, + request: ResponsesRequest, + text: str, + *, + item_id: str, + status: Optional[str], + ) -> ResponseReasoningItem: + """Build a reasoning item, mirroring the trace into ``summary`` when the + caller opted in via ``reasoning.summary``; the full trace stays in + ``content``.""" + wants_summary = cls._wants_reasoning_summary(request) + return ResponseReasoningItem( + id=item_id, + type="reasoning", + summary=( + [ResponseReasoningSummary(type="summary_text", text=text)] + if wants_summary + else [] + ), + content=[ResponseReasoningTextContent(type="reasoning_text", text=text)], + encrypted_content=( + encode_reasoning_state(text) + if request.is_include_encrypted_reasoning() + else None + ), + status=status, + ) + + @staticmethod + def _make_tool_call_item( + name: str, arguments: str, custom_names: set[str] + ) -> Union[ResponseFunctionToolCall, ResponseCustomToolCall]: + """A call against a ``custom`` tool reports its freeform payload rather + than the JSON arguments of the shim function tool.""" + call_id = f"call_{random_uuid()[:24]}" + if name in custom_names: + return ResponseCustomToolCall( + type="custom_tool_call", + id=f"ctc_{random_uuid()[:8]}", + call_id=call_id, + name=name, + input=decode_custom_tool_input(arguments), + ) + return ResponseFunctionToolCall( + arguments=arguments, + call_id=call_id, + type="function_call", + name=name, + id=f"fc_{random_uuid()[:8]}", + status="completed", + ) + @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" + if reason == "length": + return "incomplete" + if reason in ("abort", "error"): + return "failed" + return "completed" + + @classmethod + def _error_from_finish_reason(cls, finish_reason: Any) -> Optional[dict]: + if cls._status_from_finish_reason(finish_reason) != "failed": + return None + message = ( + finish_reason.get("message") if isinstance(finish_reason, dict) else None + ) + return {"code": "server_error", "message": message or "Generation aborted"} + + @staticmethod + def _terminal_stream_event(response: dict): + status = response["status"] + event_cls = { + "completed": openai_responses_types.ResponseCompletedEvent, + "incomplete": openai_responses_types.ResponseIncompleteEvent, + "failed": openai_responses_types.ResponseFailedEvent, + }[status] + return event_cls( + type=f"response.{status}", sequence_number=-1, response=response + ) def _is_thinking_enabled_for_request(self, request: ResponsesRequest) -> bool: if not self.reasoning_parser: @@ -855,32 +1020,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=( - [ - ResponseReasoningSummary( - type="summary_text", text=reasoning_content - ) - ] - if wants_summary - else [] - ), - content=[ - ResponseReasoningTextContent( - type="reasoning_text", text=reasoning_content - ), - ], - status=None, + output_items.append( + self._make_reasoning_item( + request, + reasoning_content, + item_id=f"rs_{random_uuid()}", + status=None, + ) ) - output_items.append(reasoning_item) - is_required = request.tool_choice == "required" - tool_call_items: list[ResponseFunctionToolCall] = [] + tool_choice = request.effective_tool_choice() + is_required = tool_choice == "required" or isinstance(tool_choice, dict) + custom_names = custom_tool_names(request.tools) + tool_call_items: list[ + Union[ResponseFunctionToolCall, ResponseCustomToolCall] + ] = [] parsed_via_native = False detector_owns_format = False if ( @@ -894,23 +1048,17 @@ class OpenAIServingResponses(OpenAIServingChat): self.tool_call_parser, tokenizer=self.tokenizer_manager.tokenizer, ) - detector_owns_format = ( - parser.detector.supports_structural_tag() - or parser.detector.parses_required_natively() - ) + detector_owns_format = self._tool_parser_owns_format(parser) should_try_native = not is_required or detector_owns_format 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", + self._make_tool_call_item( + call_info.name, + call_info.parameters or "", + custom_names, ) ) parsed_via_native = bool(call_info_list) @@ -936,13 +1084,8 @@ class OpenAIServingResponses(OpenAIServingChat): 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", + self._make_tool_call_item( + tool["name"], arguments, custom_names ) ) content = "" @@ -958,6 +1101,7 @@ class OpenAIServingResponses(OpenAIServingChat): logprobs=output_logprobs, ) message = ResponseOutputMessage( + phase="commentary" if tool_call_items else "final_answer", id=f"msg_{random_uuid()}", content=[output_text], role="assistant", @@ -973,7 +1117,11 @@ class OpenAIServingResponses(OpenAIServingChat): context: HarmonyContext, ): output_items = [] - num_init_messages = context.num_init_messages + num_init_messages = ( + 0 + if isinstance(context, StreamingHarmonyContext) + else context.num_init_messages + ) for msg in context.messages[num_init_messages:]: output_items.extend(parse_output_message(msg)) # Handle the generation stopped in the middle (if any). @@ -982,6 +1130,13 @@ class OpenAIServingResponses(OpenAIServingChat): output_items.extend(last_items) return output_items + @staticmethod + def _tool_parser_owns_format(parser: FunctionCallParser) -> bool: + return ( + parser.detector.supports_structural_tag() + or parser.detector.parses_required_natively() + ) + @staticmethod def _chat_tool_choice(tool_choice: Any) -> Any: """Nest an ``effective_tool_choice()`` result the way chat expects: @@ -992,18 +1147,24 @@ class OpenAIServingResponses(OpenAIServingChat): @staticmethod def _response_tools_to_chat_tools(request: ResponsesRequest) -> list[Tool]: - # Only ``function`` tools flow to chat; built-ins go through harmony. + # ``function`` and ``custom`` tools flow to chat; built-ins go through + # harmony. A custom tool is shimmed into a single-string function tool. chat_tools = [] for tool in request.tools: - if tool.type != "function": + if tool.type == "function": + description, parameters = tool.description, tool.parameters + elif tool.type == "custom" and tool.name: + description = custom_tool_description(tool.description, tool.format) + parameters = custom_tool_parameters() + else: continue chat_tools.append( Tool( type="function", function=Function( name=tool.name, - description=tool.description, - parameters=tool.parameters, + description=description, + parameters=parameters, strict=tool.strict, ), ) @@ -1054,6 +1215,27 @@ class OpenAIServingResponses(OpenAIServingChat): return content_part + @staticmethod + def _flatten_tool_output(output: Any) -> str: + """``output`` may be a string or an array of content parts (OpenAI allows + both); the chat tool message needs a string.""" + if isinstance(output, list): + return "".join(p.get("text", "") for p in output if isinstance(p, dict)) + return output + + @staticmethod + def _chat_tool_call_message(message: dict, name: Any, arguments: str) -> dict: + return { + "role": "assistant", + "tool_calls": [ + { + "id": message.get("call_id") or message.get("id"), + "type": "function", + "function": {"name": name, "arguments": arguments}, + } + ], + } + @classmethod def _normalize_response_message_for_chat(cls, message: Any) -> Any: """Convert one Responses-API input item to a chat-completions message.""" @@ -1062,10 +1244,15 @@ class OpenAIServingResponses(OpenAIServingChat): if not isinstance(message, dict): return message - # Most chat templates only recognize system/user/assistant/tool; - # collapse ``developer`` to ``system`` at the boundary. + # Most chat templates only recognize system/user/assistant/tool; collapse + # ``developer`` to ``system`` at the boundary, labelled so the instruction + # tier survives the collapse. if message.get("role") == "developer": - message = {**message, "role": "system"} + message = { + **message, + "role": "system", + "content": label_developer_content(message.get("content")), + } msg_type = message.get("type") if msg_type == "function_call": @@ -1084,29 +1271,19 @@ class OpenAIServingResponses(OpenAIServingChat): 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": - # ``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 cls._chat_tool_call_message(message, message.get("name"), raw) + if msg_type == "custom_tool_call": + # Replay through the same single-string shim the tool was offered as. + return cls._chat_tool_call_message( + message, + message.get("name"), + encode_custom_tool_input(message.get("input") or ""), + ) + if msg_type in ("function_call_output", "custom_tool_call_output"): return { "role": "tool", "tool_call_id": message.get("call_id"), - "content": out, + "content": cls._flatten_tool_output(message.get("output", "")), } # Reasoning items render as {role: assistant, reasoning_content}; # empty ones drop instead of injecting an empty assistant block. @@ -1125,6 +1302,11 @@ class OpenAIServingResponses(OpenAIServingChat): text_parts = _collect(message.get("summary")) if not text_parts: text_parts = _collect(message.get("content")) + if not text_parts: + # A store=false client may replay only the opaque blob. + restored = decode_reasoning_state(message.get("encrypted_content")) + if restored: + text_parts = [restored] if not text_parts: return None return { @@ -1154,33 +1336,6 @@ class OpenAIServingResponses(OpenAIServingChat): 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, @@ -1196,6 +1351,7 @@ class OpenAIServingResponses(OpenAIServingChat): and merged and isinstance(merged[-1], dict) and merged[-1].get("role") == "assistant" + and merged[-1].get("phase") == msg.get("phase") ): prev = merged[-1] = dict(merged[-1]) # Lift mixed str/list content to list parts so non-text parts @@ -1236,6 +1392,18 @@ class OpenAIServingResponses(OpenAIServingChat): merged.append(msg) return merged + def _response_input_history(self, request: ResponsesRequest) -> list: + history = ( + list(self.msg_store[request.previous_response_id]) + if request.previous_response_id is not None + else [] + ) + if isinstance(request.input, str): + history.append({"role": "user", "content": request.input}) + else: + history.extend(request.input) + return history + def _construct_input_messages( self, request: ResponsesRequest, @@ -1250,27 +1418,10 @@ class OpenAIServingResponses(OpenAIServingChat): } ) - # Prepend the conversation history - if prev_response is not None: - # Add the previous messages - prev_msg = self.msg_store[prev_response.id] - messages.extend(prev_msg) - - for output_item in prev_response.output: - assistant_text = self._output_message_text(output_item) - if assistant_text is None: - continue - 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: - 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 + for input_item in self._response_input_history(request): + normalized = self._normalize_response_message_for_chat(input_item) + if normalized is not None: + messages.append(normalized) # One Responses-API assistant turn maps to multiple input items # (message + function_call(s)); collapse them into one chat message @@ -1306,59 +1457,33 @@ class OpenAIServingResponses(OpenAIServingChat): prev_response: Optional[ResponsesResponse], ) -> list[OpenAIMessage]: messages: list[OpenAIMessage] = [] - if prev_response is None: - # New conversation. - reasoning_effort = request.reasoning.effort if request.reasoning else None - tool_types = [tool.type for tool in request.tools] - enable_browser = ( - 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 - ) - sys_msg = get_system_message( - reasoning_effort=reasoning_effort, - browser_description=( - self.tool_server.get_tool_description("browser") - if self.tool_server and enable_browser - else None - ), - python_description=( - self.tool_server.get_tool_description("python") - if self.tool_server and enable_code_interpreter - else None - ), - ) - messages.append(sys_msg) - dev_msg = get_developer_message(request.instructions, request.tools) - messages.append(dev_msg) - else: - # Continue the previous conversation. - # FIXME: Currently, request params like reasoning and - # instructions are ignored. - prev_msgs = self.msg_store[prev_response.id] - # Remove the previous chain-of-thoughts if there is a new "final" - # message. - if ( - len(prev_msgs) > 0 - and hasattr(prev_msgs[-1], "channel") - and prev_msgs[-1].channel == "final" - ): # type: ignore[union-attr] - prev_final_msg_idx = -1 - for i in range(len(prev_msgs) - 2, -1, -1): - if ( - hasattr(prev_msgs[i], "channel") - and prev_msgs[i].channel == "final" - ): # type: ignore[union-attr] - prev_final_msg_idx = i - break - recent_turn_msgs = prev_msgs[prev_final_msg_idx + 1 :] - del prev_msgs[prev_final_msg_idx + 1 :] - for msg in recent_turn_msgs: - if hasattr(msg, "channel") and msg.channel != "analysis": # type: ignore[union-attr] - prev_msgs.append(msg) - messages.extend(prev_msgs) + reasoning_effort = request.reasoning.effort if request.reasoning else None + tool_types = [tool.type for tool in request.tools] + enable_browser = ( + 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 + ) + sys_msg = get_system_message( + reasoning_effort=reasoning_effort, + browser_description=( + self.tool_server.get_tool_description("browser") + if self.tool_server and enable_browser + else None + ), + python_description=( + self.tool_server.get_tool_description("python") + if self.tool_server and enable_code_interpreter + else None + ), + ) + messages.append(sys_msg) + dev_msg = get_developer_message(request.instructions, request.tools) + messages.append(dev_msg) + if prev_response is not None: + messages.extend(self.msg_store[prev_response.id]) # Append the new input. # Responses API supports simple text inputs without chat format. if isinstance(request.input, str): @@ -1370,7 +1495,12 @@ class OpenAIServingResponses(OpenAIServingChat): prev_outputs = [] for response_msg in request.input: messages.append(parse_response_input(response_msg, prev_outputs)) - if isinstance(response_msg, ResponseFunctionToolCall): + item_type = ( + response_msg.get("type") + if isinstance(response_msg, dict) + else response_msg.type + ) + if item_type in ("function_call", "custom_tool_call"): prev_outputs.append(response_msg) return messages @@ -1494,29 +1624,170 @@ class OpenAIServingResponses(OpenAIServingChat): *, require_reasoning: bool, ) -> AsyncGenerator[str, None]: - # TODO: - # 1. Handle disconnect - created_time = created_time or int(time.time()) - sequence_number = 0 + emitted_items = [] + active_item = None + active_model = None + num_messages = 0 - def _send_event(event): + def _send_event(event_type: str, **fields): nonlocal sequence_number - # Set sequence_number if the event has this attribute - if hasattr(event, "sequence_number"): - event.sequence_number = sequence_number + payload = {"type": event_type, "sequence_number": sequence_number, **fields} sequence_number += 1 - # Get event type from the event's type field if it exists - event_type = getattr(event, "type", "unknown") - return ( - f"event: {event_type}\ndata: {event.model_dump_json(indent=None)}\n\n" - ) + return f"event: {event_type}\ndata: {orjson.dumps(payload).decode()}\n\n" - current_content_index = 0 - current_output_index = 0 - current_item_id = f"item_{random_uuid()}" - sent_output_item_added = False + def _update_item(item, *, done=False, status="completed"): + nonlocal active_item, active_model + if item.type == "reasoning": + item = self._make_reasoning_item( + request, + "\n".join(part.text for part in item.content or []), + item_id=item.id, + status=item.status, + ) + snapshot = item.model_dump() + output_index = len(emitted_items) + if active_item is None: + active_item = dict(snapshot) + if "status" in active_item: + active_item["status"] = "in_progress" + for field in ("content", "summary"): + if field in active_item: + active_item[field] = [] + for field in ("arguments", "code"): + if field in active_item: + active_item[field] = "" + if "encrypted_content" in active_item: + active_item["encrypted_content"] = None + yield _send_event( + "response.output_item.added", + output_index=output_index, + item=active_item, + ) + if item.type in ("web_search_call", "code_interpreter_call"): + yield _send_event( + f"response.{item.type}.in_progress", + output_index=output_index, + item_id=active_item["id"], + ) + elif active_item["type"] != snapshot["type"]: + raise ValueError( + f"Harmony changed item type before closing the previous item, rid={request.request_id}" + ) + + snapshot["id"] = active_item["id"] + if "call_id" in active_item: + snapshot["call_id"] = active_item["call_id"] + active_model = type(item).model_validate(snapshot) + common = {"output_index": output_index, "item_id": active_item["id"]} + + for field in ("content", "summary"): + parts = snapshot.get(field) or [] + if not parts: + continue + is_summary = field == "summary" + event_prefix = ( + "response.reasoning_summary_text" + if is_summary + else "response.output_text" + if item.type == "message" + else "response.reasoning_text" + ) + index_key = "summary_index" if is_summary else "content_index" + part_prefix = ( + "response.reasoning_summary_part" + if is_summary + else "response.content_part" + if item.type == "message" + else None + ) + for index, part in enumerate(parts): + part_fields = {**common, index_key: index} + if index == len(active_item[field]): + empty_part = {**part, "text": ""} + active_item[field].append(empty_part) + if part_prefix: + yield _send_event( + f"{part_prefix}.added", **part_fields, part=empty_part + ) + previous_text = active_item[field][index]["text"] + if not part["text"].startswith(previous_text): + raise ValueError( + f"Harmony rewrote previously emitted text, rid={request.request_id}" + ) + delta = part["text"][len(previous_text) :] + if delta: + extra = ( + {"logprobs": []} + if item.type == "message" and not is_summary + else {} + ) + yield _send_event( + f"{event_prefix}.delta", **part_fields, delta=delta, **extra + ) + active_item[field][index] = part + if done: + extra = ( + {"logprobs": []} + if item.type == "message" and not is_summary + else {} + ) + yield _send_event( + f"{event_prefix}.done", + **part_fields, + text=part["text"], + **extra, + ) + if part_prefix: + yield _send_event( + f"{part_prefix}.done", **part_fields, part=part + ) + + for field, event_prefix in ( + ("arguments", "response.function_call_arguments"), + ("code", "response.code_interpreter_call_code"), + ): + if field not in snapshot: + continue + value = snapshot[field] or "" + previous_value = active_item[field] or "" + if not value.startswith(previous_value): + raise ValueError( + f"Harmony rewrote previously emitted tool input, rid={request.request_id}" + ) + delta = value[len(previous_value) :] + if delta: + yield _send_event(f"{event_prefix}.delta", **common, delta=delta) + active_item[field] = value + if done: + yield _send_event( + f"{event_prefix}.done", **common, **{field: value} + ) + + if done: + if "status" in snapshot: + snapshot["status"] = status + if status == "completed" and item.type in ( + "web_search_call", + "code_interpreter_call", + ): + action = ( + "searching" + if item.type == "web_search_call" + else "interpreting" + ) + yield _send_event(f"response.{item.type}.{action}", **common) + yield _send_event(f"response.{item.type}.completed", **common) + completed_item = type(item).model_validate(snapshot) + yield _send_event( + "response.output_item.done", + output_index=output_index, + item=completed_item.model_dump(), + ) + emitted_items.append(completed_item) + active_item = None + active_model = None initial_response = ResponsesResponse.from_request( request, @@ -1527,362 +1798,28 @@ class OpenAIServingResponses(OpenAIServingChat): 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, - ) - ) + initial_response["tools"] = [] + yield _send_event("response.created", response=initial_response) + yield _send_event("response.in_progress", response=initial_response) 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 - # or are not compatible with per-turn handling in the /v1/responses endpoint. - if not hasattr(ctx, "is_expecting_start"): - continue + for message in ctx.messages[num_messages:]: + for item in parse_output_message(message): + for event in _update_item(item, done=True): + yield event + num_messages = len(ctx.messages) + for item in parse_remaining_state(ctx.parser): + for event in _update_item(item): + yield event - if ctx.is_expecting_start(): - current_output_index += 1 - sent_output_item_added = False - - if len(ctx.parser.messages) > 0: - previous_item = ctx.parser.messages[-1] - if previous_item.recipient is not None: - # Deal with tool call here - pass - elif previous_item.channel == "analysis": - reasoning_item = ResponseReasoningItem( - id=f"rs_{random_uuid()}", - type="reasoning", - summary=[], - content=[ - ResponseReasoningTextContent( - text=previous_item.content[0].text, - type="reasoning_text", - ), - ], - status="completed", - ) - yield _send_event( - openai_responses_types.ResponseReasoningTextDoneEvent( - type="response.reasoning_text.done", - item_id=current_item_id, - sequence_number=-1, - output_index=current_output_index, - content_index=current_content_index, - text=previous_item.content[0].text, - ) - ) - yield _send_event( - openai_responses_types.ResponseOutputItemDoneEvent( - type="response.output_item.done", - sequence_number=-1, - output_index=current_output_index, - item=reasoning_item, - ) - ) - elif previous_item.channel == "final": - text_content = openai_responses_types.ResponseOutputText( - type="output_text", - text=previous_item.content[0].text, - annotations=[], - ) - yield _send_event( - openai_responses_types.ResponseTextDoneEvent( - type="response.output_text.done", - sequence_number=-1, - output_index=current_output_index, - content_index=current_content_index, - text=previous_item.content[0].text, - logprobs=[], - item_id=current_item_id, - ) - ) - yield _send_event( - openai_responses_types.ResponseContentPartDoneEvent( - type="response.content_part.done", - sequence_number=-1, - item_id=current_item_id, - output_index=current_output_index, - content_index=current_content_index, - part=text_content, - ) - ) - yield _send_event( - openai_responses_types.ResponseOutputItemDoneEvent( - type="response.output_item.done", - sequence_number=-1, - output_index=current_output_index, - item=openai_responses_types.ResponseOutputMessage( - id=current_item_id, - type="message", - role="assistant", - content=[text_content], - status="completed", - ), - ) - ) - - if ctx.parser.last_content_delta: - if ( - ctx.parser.current_channel == "final" - and ctx.parser.current_recipient is None - ): - if not sent_output_item_added: - sent_output_item_added = True - yield _send_event( - openai_responses_types.ResponseOutputItemAddedEvent( - type="response.output_item.added", - sequence_number=-1, - output_index=current_output_index, - item=openai_responses_types.ResponseOutputMessage( - id=current_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=current_output_index, - item_id=current_item_id, - content_index=current_content_index, - part=openai_responses_types.ResponseOutputText( - type="output_text", - text="", - annotations=[], - logprobs=None, - ), - ) - ) - yield _send_event( - openai_responses_types.ResponseTextDeltaEvent( - type="response.output_text.delta", - sequence_number=-1, - content_index=current_content_index, - output_index=current_output_index, - item_id=current_item_id, - delta=ctx.parser.last_content_delta, - # TODO, use logprobs from ctx.last_request_output - logprobs=[], - ) - ) - elif ( - ctx.parser.current_channel == "analysis" - and ctx.parser.current_recipient is None - ): - if not sent_output_item_added: - sent_output_item_added = True - yield _send_event( - openai_responses_types.ResponseOutputItemAddedEvent( - type="response.output_item.added", - sequence_number=-1, - output_index=current_output_index, - item=openai_responses_types.ResponseReasoningItem( - type="reasoning", - id=current_item_id, - summary=[], - status="in_progress", - ), - ) - ) - yield _send_event( - openai_responses_types.ResponseContentPartAddedEvent( - type="response.content_part.added", - sequence_number=-1, - output_index=current_output_index, - item_id=current_item_id, - content_index=current_content_index, - # TODO: migrate this to - # ResponseReasoningTextContent for now - part=openai_responses_types.ResponseOutputText( - type="output_text", - text="", - annotations=[], - logprobs=None, - ), - ) - ) - # TODO: migrate to OpenAI types once updated. - yield _send_event( - openai_responses_types.ResponseReasoningTextDeltaEvent( - type="response.reasoning_text.delta", - item_id=current_item_id, - output_index=current_output_index, - content_index=current_content_index, - delta=ctx.parser.last_content_delta, - sequence_number=-1, - ) - ) - - if ctx.is_assistant_action_turn() and len(ctx.parser.messages) > 0: - previous_item = ctx.parser.messages[-1] - if ( - self.supports_browsing - and previous_item.recipient is not None - and previous_item.recipient.startswith("browser.") - ): - function_name = previous_item.recipient[len("browser.") :] - action = None - parsed_args = orjson.loads(previous_item.content[0].text) - if function_name == "search": - action = openai_responses_types.response_function_web_search.ActionSearch( - type="search", - query=parsed_args["query"], - ) - elif function_name == "open": - action = openai_responses_types.response_function_web_search.ActionOpenPage( - type="open_page", - # TODO: translate to url - url=f"cursor:{parsed_args.get('cursor', '')}", - ) - elif function_name == "find": - action = openai_responses_types.response_function_web_search.ActionFind( - type="find", - pattern=parsed_args["pattern"], - # TODO: translate to url - url=f"cursor:{parsed_args.get('cursor', '')}", - ) - else: - raise ValueError(f"Unknown function name: {function_name}") - - yield _send_event( - openai_responses_types.ResponseOutputItemAddedEvent( - type="response.output_item.added", - sequence_number=-1, - output_index=current_output_index, - item=openai_responses_types.response_function_web_search.ResponseFunctionWebSearch( - # TODO: generate a unique id for web search call - type="web_search_call", - id=current_item_id, - action=action, - status="in_progress", - ), - ) - ) - yield _send_event( - openai_responses_types.ResponseWebSearchCallInProgressEvent( - type="response.web_search_call.in_progress", - sequence_number=-1, - output_index=current_output_index, - item_id=current_item_id, - ) - ) - yield _send_event( - openai_responses_types.ResponseWebSearchCallSearchingEvent( - type="response.web_search_call.searching", - sequence_number=-1, - output_index=current_output_index, - item_id=current_item_id, - ) - ) - - # enqueue - yield _send_event( - openai_responses_types.ResponseWebSearchCallCompletedEvent( - type="response.web_search_call.completed", - sequence_number=-1, - output_index=current_output_index, - item_id=current_item_id, - ) - ) - yield _send_event( - openai_responses_types.ResponseOutputItemDoneEvent( - type="response.output_item.done", - sequence_number=-1, - output_index=current_output_index, - item=openai_responses_types.ResponseFunctionWebSearch( - type="web_search_call", - id=current_item_id, - action=action, - status="completed", - ), - ) - ) - - if ( - self.supports_code_interpreter - and previous_item.recipient is not None - and previous_item.recipient.startswith("python") - ): - yield _send_event( - openai_responses_types.ResponseOutputItemAddedEvent( - type="response.output_item.added", - sequence_number=-1, - output_index=current_output_index, - item=openai_responses_types.ResponseCodeInterpreterToolCallParam( - type="code_interpreter_call", - id=current_item_id, - code="", - container_id="auto", - outputs=[], - status="in_progress", - ), - ) - ) - yield _send_event( - openai_responses_types.ResponseCodeInterpreterCallInProgressEvent( - type="response.code_interpreter_call.in_progress", - sequence_number=-1, - output_index=current_output_index, - item_id=current_item_id, - ) - ) - # TODO: do we need to add delta event here? - yield _send_event( - openai_responses_types.ResponseCodeInterpreterCallCodeDoneEvent( - type="response.code_interpreter_call_code.done", - sequence_number=-1, - output_index=current_output_index, - item_id=current_item_id, - code=previous_item.content[0].text, - ) - ) - yield _send_event( - openai_responses_types.ResponseCodeInterpreterCallInterpretingEvent( - type="response.code_interpreter_call.interpreting", - sequence_number=-1, - output_index=current_output_index, - item_id=current_item_id, - ) - ) - yield _send_event( - openai_responses_types.ResponseCodeInterpreterCallCompletedEvent( - type="response.code_interpreter_call.completed", - sequence_number=-1, - output_index=current_output_index, - item_id=current_item_id, - ) - ) - yield _send_event( - openai_responses_types.ResponseOutputItemDoneEvent( - type="response.output_item.done", - sequence_number=-1, - output_index=current_output_index, - item=openai_responses_types.ResponseCodeInterpreterToolCallParam( - type="code_interpreter_call", - id=current_item_id, - code=previous_item.content[0].text, - container_id="auto", - # TODO: add outputs here - outputs=[], - status="completed", - ), - ) - ) + if active_model is not None: + status = ( + "incomplete" + if self._status_from_finish_reason(context.finish_reason) != "completed" + else "completed" + ) + for event in _update_item(active_model, done=True, status=status): + yield event async def empty_async_generator(): for _ in (): @@ -1898,18 +1835,11 @@ class OpenAIServingResponses(OpenAIServingChat): request_metadata, created_time=created_time, require_reasoning=require_reasoning, + output_items=emitted_items, ) response_dict = final_response.model_dump() - # OpenAI SDK's Tool union may not know extended types; drop echo. response_dict["tools"] = [] - - yield _send_event( - openai_responses_types.ResponseCompletedEvent( - type="response.completed", - sequence_number=-1, - response=response_dict, - ) - ) + yield _send_event(f"response.{final_response.status}", response=response_dict) async def responses_stream_generator_non_harmony( self, @@ -1922,6 +1852,61 @@ class OpenAIServingResponses(OpenAIServingChat): created_time: Optional[int] = None, *, require_reasoning: bool, + ) -> AsyncGenerator[str, None]: + pending: list[dict] = [] + can_call_tools = ( + bool(self._response_tools_to_chat_tools(request)) + and request.effective_tool_choice() != "none" + ) + async for event in self._responses_stream_generator_non_harmony( + request, + sampling_params, + result_generator, + model_name, + tokenizer, + request_metadata, + created_time, + require_reasoning=require_reasoning, + ): + if not can_call_tools: + yield event + continue + payload = orjson.loads(event.split("data: ", 1)[1]) + event_type = payload["type"] + item_type = payload.get("item", {}).get("type") + opens_item = event_type == "response.output_item.added" + if pending or (opens_item and item_type == "message"): + pending.append(payload) + tool_follows = opens_item and item_type in ( + "function_call", + "custom_tool_call", + ) + terminal = event_type in ( + "response.completed", + "response.incomplete", + "response.failed", + ) + if tool_follows or terminal: + phase = "commentary" if tool_follows else "final_answer" + for buffered in pending: + if buffered.get("item", {}).get("type") == "message": + buffered["item"]["phase"] = phase + yield f"event: {buffered['type']}\ndata: {orjson.dumps(buffered).decode()}\n\n" + pending.clear() + else: + yield event + + 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, + *, + require_reasoning: bool, ) -> 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 @@ -1938,9 +1923,10 @@ class OpenAIServingResponses(OpenAIServingChat): event.sequence_number = sequence_number sequence_number += 1 event_type = getattr(event, "type", "unknown") - return ( - f"event: {event_type}\ndata: {event.model_dump_json(indent=None)}\n\n" - ) + payload = event.model_dump() + if isinstance(getattr(event, "item", None), ResponseOutputMessage): + payload["item"] = event.item.model_dump() + return f"event: {event_type}\ndata: {orjson.dumps(payload).decode()}\n\n" # The streaming Response* event models echo ``tools`` through a # narrower OpenAI SDK Tool union; strip it to avoid pydantic @@ -1976,20 +1962,20 @@ class OpenAIServingResponses(OpenAIServingChat): ) chat_tools = self._response_tools_to_chat_tools(request) - is_required = request.tool_choice == "required" + custom_names = custom_tool_names(request.tools) + tool_choice = request.effective_tool_choice() + is_required = tool_choice == "required" or isinstance(tool_choice, dict) tool_parser: Optional[Union[FunctionCallParser, JsonArrayParser]] = None if chat_tools and request.tool_choice != "none": - native_supports_structural_tag = False + detector_owns_format = False if self.tool_call_parser: probe = FunctionCallParser( chat_tools, self.tool_call_parser, tokenizer=self.tokenizer_manager.tokenizer, ) - native_supports_structural_tag = ( - probe.detector.supports_structural_tag() - ) - if is_required and not native_supports_structural_tag: + detector_owns_format = self._tool_parser_owns_format(probe) + if is_required and not detector_owns_format: tool_parser = JsonArrayParser() elif self.tool_call_parser: tool_parser = FunctionCallParser( @@ -2055,17 +2041,10 @@ class OpenAIServingResponses(OpenAIServingChat): 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), - ], + completed_item = self._make_reasoning_item( + request, + text, + item_id=reasoning_state["item_id"], status="completed", ) events: list = [] @@ -2145,6 +2124,7 @@ class OpenAIServingResponses(OpenAIServingChat): role="assistant", content=[text_content], status="completed", + phase="final_answer", ) events = [ _send_event( @@ -2186,25 +2166,67 @@ class OpenAIServingResponses(OpenAIServingChat): 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 "", + events: list = [] + if state["custom"]: + payload = decode_custom_tool_input(arguments) + # Deltas cannot be retracted, so a payload that no longer + # extends what was already streamed defers to the streamed text. + if not payload.startswith(state["payload"]): + payload = state["payload"] or payload + remainder = payload[len(state["payload"]) :] + if remainder: + state["payload"] = payload + events.append( + _send_event( + openai_responses_types.ResponseCustomToolCallInputDeltaEvent( + type="response.custom_tool_call_input.delta", + sequence_number=-1, + item_id=state["item_id"], + output_index=state["output_index"], + delta=remainder, + ) + ) ) - ), + completed_item = ResponseCustomToolCall( + type="custom_tool_call", + id=state["item_id"], + call_id=state["call_id"], + name=state["name"] or "", + input=payload, + ) + events.append( + _send_event( + openai_responses_types.ResponseCustomToolCallInputDoneEvent( + type="response.custom_tool_call_input.done", + sequence_number=-1, + item_id=state["item_id"], + output_index=state["output_index"], + input=payload, + ) + ) + ) + else: + completed_item = ResponseFunctionToolCall( + arguments=arguments, + call_id=state["call_id"], + name=state["name"] or "", + type="function_call", + id=state["item_id"], + status="completed", + ) + events.append( + _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 "", + ) + ) + ) + events.append( _send_event( openai_responses_types.ResponseOutputItemDoneEvent( type="response.output_item.done", @@ -2212,8 +2234,8 @@ class OpenAIServingResponses(OpenAIServingChat): output_index=state["output_index"], item=completed_item, ) - ), - ] + ) + ) emitted_items.append(completed_item) state["done"] = True return events @@ -2352,6 +2374,11 @@ class OpenAIServingResponses(OpenAIServingChat): for ev in _close_message_item(): yield ev + if calls: + for item in emitted_items: + if isinstance(item, ResponseOutputMessage): + item.phase = "commentary" + for call in calls: tool_index = call.tool_index state = tool_call_states.get(tool_index) @@ -2363,46 +2390,84 @@ class OpenAIServingResponses(OpenAIServingChat): 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]}" + name = call.name or "" + is_custom = name in custom_names state = { - "item_id": item_id, - "call_id": call_id, + "item_id": ( + f"ctc_{random_uuid()[:8]}" + if is_custom + else f"fc_{random_uuid()[:8]}" + ), + "call_id": f"call_{random_uuid()[:24]}", "output_index": current_output_index, - "name": call.name or "", + "name": name, "arguments": "", + "custom": is_custom, + "payload": "", "added": False, "done": False, } tool_call_states[tool_index] = state if not state["added"]: state["added"] = True + if state["custom"]: + added_item = ResponseCustomToolCall( + type="custom_tool_call", + id=state["item_id"], + call_id=state["call_id"], + name=state["name"], + input="", + ) + else: + added_item = ResponseFunctionToolCall( + arguments="", + call_id=state["call_id"], + name=state["name"], + type="function_call", + id=state["item_id"], + status="in_progress", + ) 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", - ), + item=added_item, ) ) 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, + if state["custom"]: + # The payload is a JSON string value inside the + # shim arguments, so un-escape what has arrived + # so far and stream only the new suffix. + decoded = decode_custom_tool_input_prefix( + state["arguments"] + ) + if decoded.startswith(state["payload"]) and len( + decoded + ) > len(state["payload"]): + delta = decoded[len(state["payload"]) :] + state["payload"] = decoded + yield _send_event( + openai_responses_types.ResponseCustomToolCallInputDeltaEvent( + type="response.custom_tool_call_input.delta", + sequence_number=-1, + item_id=state["item_id"], + output_index=state["output_index"], + delta=delta, + ) + ) + else: + 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( @@ -2430,6 +2495,7 @@ class OpenAIServingResponses(OpenAIServingChat): role="assistant", content=[], status="in_progress", + phase="final_answer", ), ) ) @@ -2478,8 +2544,10 @@ class OpenAIServingResponses(OpenAIServingChat): yield ev for ev in _emit_tool_calls(opening): yield ev - except Exception: - logger.exception("Error while streaming /v1/responses") + except Exception as e: + logger.exception( + "Error while streaming /v1/responses %s", request.request_id + ) failed = _sanitize_response_dict( ResponsesResponse.from_request( request, @@ -2491,6 +2559,7 @@ class OpenAIServingResponses(OpenAIServingChat): usage=None, ).model_dump() ) + failed["error"] = {"code": "server_error", "message": str(e)} yield _send_event( openai_responses_types.ResponseFailedEvent( type="response.failed", @@ -2531,21 +2600,22 @@ class OpenAIServingResponses(OpenAIServingChat): status=self._status_from_finish_reason(finish_reason), usage=usage, ) + final_response.error = self._error_from_finish_reason(finish_reason) 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 + self.msg_store[final_response.id] = self._response_input_history( + request + ) + [ + item.model_dump(exclude_none=True) + for item in final_response.output + ] response_dict = _sanitize_response_dict(final_response.model_dump()) - yield _send_event( - openai_responses_types.ResponseCompletedEvent( - type="response.completed", - sequence_number=-1, - response=response_dict, - ) - ) + yield _send_event(self._terminal_stream_event(response_dict)) async def _generate_with_builtin_tools( self, diff --git a/test/registered/unit/entrypoints/openai/test_responses_custom_tools.py b/test/registered/unit/entrypoints/openai/test_responses_custom_tools.py new file mode 100644 index 000000000..da8a3116d --- /dev/null +++ b/test/registered/unit/entrypoints/openai/test_responses_custom_tools.py @@ -0,0 +1,442 @@ +import asyncio +import unittest +from unittest.mock import Mock + +from utils import ( + StreamFixture, + engine_chunk, + event_payloads, + event_types, + find_completed_event, + make_serving, +) + +from sglang.srt.entrypoints.openai.protocol import ResponsesRequest +from sglang.srt.entrypoints.openai.responses_adapters import ( + decode_custom_tool_input, + decode_custom_tool_input_prefix, + decode_reasoning_state, + encode_custom_tool_input, + encode_reasoning_state, + label_developer_content, +) +from sglang.srt.entrypoints.openai.serving_responses import OpenAIServingResponses +from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase + +register_cpu_ci(est_time=10, suite="base-a-test-cpu") + +CUSTOM_TOOL = { + "type": "custom", + "name": "emit_command", + "description": "Emit a shell command.", + "format": {"type": "text"}, +} + + +def _custom_request(**kwargs) -> ResponsesRequest: + payload = { + "model": "x", + "input": "run pwd", + "tools": [CUSTOM_TOOL], + "tool_choice": "required", + "store": False, + } + payload.update(kwargs) + return ResponsesRequest(**payload) + + +class CustomToolAdapterTestCase(CustomTestCase): + def test_payload_survives_encode_decode(self): + for payload in ("pwd", 'echo "hi"', "a\nb\tc", "naïve 😀", "{not json}"): + self.assertEqual( + decode_custom_tool_input(encode_custom_tool_input(payload)), payload + ) + self.assertEqual(decode_custom_tool_input("pwd"), "pwd") + + def test_prefix_decode_tracks_a_growing_buffer(self): + payload = 'echo "a\nb" 😀' + arguments = encode_custom_tool_input(payload) + seen = "" + for end in range(len(arguments) + 1): + prefix = decode_custom_tool_input_prefix(arguments[:end]) + # Monotonic, and never runs ahead of the finished value. + self.assertTrue(prefix.startswith(seen), (seen, prefix)) + self.assertTrue(payload.startswith(prefix), (payload, prefix)) + seen = prefix + self.assertEqual(seen, payload) + self.assertEqual(decode_custom_tool_input_prefix('{"city": "Beijing"}'), "") + + +class CustomToolShimTestCase(CustomTestCase): + def test_custom_tool_becomes_a_single_string_function_tool(self): + request = _custom_request() + (tool,) = OpenAIServingResponses._response_tools_to_chat_tools(request) + self.assertEqual(tool.function.name, "emit_command") + self.assertEqual(list(tool.function.parameters["properties"]), ["input"]) + self.assertEqual(tool.function.parameters["required"], ["input"]) + + nameless = ResponsesRequest( + model="x", input="hi", tools=[{"type": "custom"}], store=False + ) + self.assertEqual( + OpenAIServingResponses._response_tools_to_chat_tools(nameless), [] + ) + + def test_grammar_format_is_described_to_the_model(self): + request = _custom_request( + tools=[ + { + **CUSTOM_TOOL, + "format": { + "type": "grammar", + "syntax": "lark", + "definition": 'start: "pwd"', + }, + } + ] + ) + (tool,) = OpenAIServingResponses._response_tools_to_chat_tools(request) + self.assertIn("lark", tool.function.description) + self.assertIn('start: "pwd"', tool.function.description) + + def test_required_tool_choice_accepts_a_custom_tool(self): + serving = make_serving() + serving.reasoning_parser = None + serving.tool_call_parser = None + request = _custom_request() + output_items = serving._make_response_output_items( + request, + '[{"name": "emit_command", "parameters": {"input": "pwd"}}]', + tokenizer=Mock(), + require_reasoning=False, + ) + (item,) = output_items + self.assertEqual(item.type, "custom_tool_call") + self.assertEqual(item.name, "emit_command") + self.assertEqual(item.input, "pwd") + self.assertTrue(item.call_id) + + def test_named_choice_parses_json_in_full_and_stream_responses(self): + serving = make_serving() + serving.reasoning_parser = None + serving.tool_call_parser = None + for tool_type in ("function", "custom"): + for nested in (False, True): + with self.subTest(tool_type=tool_type, nested=nested): + name = "emit_command" + choice = {"type": tool_type} + choice.update( + {"function": {"name": name}} if nested else {"name": name} + ) + request = _custom_request( + tools=[{"type": tool_type, "name": name}], + tool_choice=choice, + stream=True, + ) + raw = '[{"name":"emit_command","parameters":{"input":"pwd"}}]' + (item,) = serving._make_response_output_items( + request, raw, tokenizer=Mock(), require_reasoning=False + ) + self.assertEqual( + item.type, + f"{tool_type}_tool_call" + if tool_type == "custom" + else "function_call", + ) + events = StreamFixture(serving, request).run( + [engine_chunk(raw[:30]), engine_chunk(raw, 2, finish=True)] + ) + (stream_item,) = find_completed_event(events)["response"]["output"] + self.assertEqual(stream_item["type"], item.type) + self.assertEqual(stream_item["name"], name) + field = "input" if tool_type == "custom" else "arguments" + self.assertEqual(stream_item[field], getattr(item, field)) + self.assertNotIn("response.output_text.delta", event_types(events)) + + def test_named_choice_rejects_an_undeclared_tool_before_generation(self): + serving = make_serving() + for stream in (False, True): + request = _custom_request( + tool_choice={"type": "custom", "name": "missing"}, stream=stream + ) + result = asyncio.run(serving.create_responses(request)) + self.assertEqual(result.status_code, 400) + self.assertIn(b"tool_choice", result.body) + serving.tokenizer_manager.generate_request.assert_not_called() + + +class CustomToolReplayTestCase(CustomTestCase): + def test_custom_tool_call_replays_through_the_shim(self): + message = OpenAIServingResponses._normalize_response_message_for_chat( + { + "type": "custom_tool_call", + "call_id": "call_1", + "name": "emit_command", + "input": "pwd", + } + ) + self.assertEqual(message["role"], "assistant") + (call,) = message["tool_calls"] + self.assertEqual(call["id"], "call_1") + self.assertEqual(call["function"]["name"], "emit_command") + self.assertEqual(decode_custom_tool_input(call["function"]["arguments"]), "pwd") + + def test_custom_tool_call_output_becomes_a_tool_message(self): + message = OpenAIServingResponses._normalize_response_message_for_chat( + { + "type": "custom_tool_call_output", + "call_id": "call_1", + "output": "/workspace", + } + ) + self.assertEqual( + message, + {"role": "tool", "tool_call_id": "call_1", "content": "/workspace"}, + ) + + parts = OpenAIServingResponses._normalize_response_message_for_chat( + { + "type": "custom_tool_call_output", + "call_id": "call_1", + "output": [{"type": "output_text", "text": "/work"}, {"text": "space"}], + } + ) + self.assertEqual(parts["content"], "/workspace") + + +class CustomToolStreamTestCase(CustomTestCase): + def _stream(self, chunks): + serving = make_serving() + serving.reasoning_parser = None + serving.tool_call_parser = None + request = _custom_request(stream=True) + return StreamFixture(serving, request).run(chunks) + + def test_input_deltas_reconstruct_the_final_payload(self): + emitted = '[{"name": "emit_command", "parameters": {"input": "pwd"}}]' + chunks = [engine_chunk(emitted[:i], i) for i in range(1, len(emitted))] + [ + engine_chunk(emitted, len(emitted), finish=True) + ] + + events = self._stream(chunks) + pairs = list(zip(event_types(events), event_payloads(events))) + deltas = "".join( + p["delta"] for t, p in pairs if t == "response.custom_tool_call_input.delta" + ) + done = [p for t, p in pairs if t == "response.custom_tool_call_input.done"] + + self.assertEqual(len(done), 1) + self.assertEqual(done[0]["input"], "pwd") + self.assertEqual(deltas, "pwd") + + added = [p for t, p in pairs if t == "response.output_item.added"] + item_done = [p for t, p in pairs if t == "response.output_item.done"] + self.assertEqual(len(added), 1) + self.assertEqual(len(item_done), 1) + self.assertEqual(added[0]["item"]["type"], "custom_tool_call") + self.assertEqual(added[0]["item"]["id"], item_done[0]["item"]["id"]) + + final = find_completed_event(events)["response"] + (item,) = [i for i in final["output"] if i["type"] == "custom_tool_call"] + self.assertEqual(item["name"], "emit_command") + self.assertEqual(item["input"], "pwd") + self.assertTrue(item["call_id"]) + self.assertNotIn( + "response.function_call_arguments.delta", [t for t, _ in pairs] + ) + + +GLM47_CALL = ( + "emit_command" + "inputpwd" + "" +) + + +class CustomToolGlm47FormatTestCase(CustomTestCase): + """The shim has to survive a real model-native tool-call format, not just the + JSON array the ``required`` constraint produces.""" + + def _serving(self): + serving = make_serving() + serving.reasoning_parser = None + serving.tool_call_parser = "glm47" + return serving + + def test_non_streaming_glm47_call_becomes_a_custom_tool_call(self): + serving = self._serving() + request = _custom_request(tool_choice="auto") + (item,) = serving._make_response_output_items( + request, GLM47_CALL, tokenizer=Mock(), require_reasoning=False + ) + self.assertEqual(item.type, "custom_tool_call") + self.assertEqual(item.name, "emit_command") + self.assertEqual(item.input, "pwd") + + def test_streaming_glm47_call_reconstructs_the_payload(self): + serving = self._serving() + request = _custom_request(tool_choice="auto", stream=True) + chunks = [engine_chunk(GLM47_CALL[:i], i) for i in range(1, len(GLM47_CALL))] + chunks.append(engine_chunk(GLM47_CALL, len(GLM47_CALL), finish=True)) + + events = StreamFixture(serving, request).run(chunks) + pairs = list(zip(event_types(events), event_payloads(events))) + deltas = "".join( + p["delta"] for t, p in pairs if t == "response.custom_tool_call_input.delta" + ) + done = [p for t, p in pairs if t == "response.custom_tool_call_input.done"] + + self.assertEqual(len(done), 1) + self.assertEqual(done[0]["input"], "pwd") + self.assertEqual(deltas, done[0]["input"]) + + final = find_completed_event(events)["response"] + (item,) = [i for i in final["output"] if i["type"] == "custom_tool_call"] + self.assertEqual(item["input"], "pwd") + + +class ReasoningEncryptedContentTestCase(CustomTestCase): + def test_state_survives_encode_decode(self): + for text in ("", "step one\nstep two", "naïve 😀"): + self.assertEqual(decode_reasoning_state(encode_reasoning_state(text)), text) + self.assertIsNone(decode_reasoning_state("not-ours")) + self.assertIsNone(decode_reasoning_state(None)) + + def test_reasoning_item_carries_the_blob_only_when_included(self): + without = OpenAIServingResponses._make_reasoning_item( + ResponsesRequest(model="x", input="hi", store=False), + "because", + item_id="rs_1", + status=None, + ) + self.assertIsNone(without.encrypted_content) + + with_blob = OpenAIServingResponses._make_reasoning_item( + ResponsesRequest( + model="x", + input="hi", + store=False, + include=["reasoning.encrypted_content"], + ), + "because", + item_id="rs_1", + status=None, + ) + self.assertEqual(decode_reasoning_state(with_blob.encrypted_content), "because") + + def test_blob_only_reasoning_item_replays(self): + message = OpenAIServingResponses._normalize_response_message_for_chat( + { + "type": "reasoning", + "summary": [], + "content": [], + "encrypted_content": encode_reasoning_state("because the sky"), + } + ) + self.assertEqual( + message, {"role": "assistant", "reasoning_content": "because the sky"} + ) + + def test_streamed_reasoning_item_carries_the_blob(self): + serving = make_serving() + serving.reasoning_parser = "deepseek-r1" + serving.tool_call_parser = None + request = ResponsesRequest( + model="x", + input="hi", + stream=True, + store=False, + include=["reasoning.encrypted_content"], + ) + events = StreamFixture(serving, request, require_reasoning=True).run( + [ + engine_chunk("because", 1), + engine_chunk("becauseanswer", 2, finish=True), + ] + ) + final = find_completed_event(events)["response"] + (item,) = [i for i in final["output"] if i["type"] == "reasoning"] + self.assertEqual(decode_reasoning_state(item["encrypted_content"]), "because") + + def test_non_streaming_reasoning_item_carries_the_blob(self): + serving = make_serving() + serving.reasoning_parser = "deepseek-r1" + serving.tool_call_parser = None + request = ResponsesRequest( + model="x", + input="hi", + store=False, + include=["reasoning.encrypted_content"], + ) + output_items = serving._make_response_output_items( + request, + "becauseanswer", + tokenizer=Mock(), + require_reasoning=True, + ) + self.assertEqual( + decode_reasoning_state(output_items[0].encrypted_content), "because" + ) + + +class DeveloperMessageTestCase(CustomTestCase): + def test_content_is_labelled(self): + self.assertEqual( + label_developer_content("Be terse."), + "Developer instructions:\nBe terse.", + ) + self.assertEqual( + label_developer_content( + [{"type": "input_text", "text": "Be terse."}, {"type": "input_image"}] + ), + [ + {"type": "input_text", "text": "Developer instructions:\nBe terse."}, + {"type": "input_image"}, + ], + ) + + def test_developer_block_follows_instructions_in_the_system_message(self): + serving = make_serving() + request = ResponsesRequest( + model="x", + store=False, + instructions="Respond in English.", + input=[ + { + "type": "message", + "role": "developer", + "content": [ + {"type": "input_text", "text": "Reply with exactly OK."} + ], + }, + {"role": "user", "content": "Reply with exactly NO."}, + ], + ) + messages = serving._construct_input_messages(request, None) + self.assertEqual( + messages[0], + { + "role": "system", + "content": ( + "Respond in English.\n\n" + "Developer instructions:\nReply with exactly OK." + ), + }, + ) + self.assertEqual(messages[1]["role"], "user") + + +class ModelValidationTestCase(CustomTestCase): + def test_model_validation(self): + serving = make_serving() + error = serving._validate_model("__no_such_model__") + self.assertIsNotNone(error) + self.assertEqual(error.status_code, 404) + self.assertIsNone(serving._validate_model(None)) + self.assertIsNone(serving._validate_model("x")) + self.assertIsNone(serving._validate_model("x:my-adapter")) + + +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 c3b27c9ff..e60f3193a 100644 --- a/test/registered/unit/entrypoints/openai/test_serving_responses.py +++ b/test/registered/unit/entrypoints/openai/test_serving_responses.py @@ -8,7 +8,7 @@ from openai.types.responses import ( ResponseReasoningItem, ) from openai.types.responses.response_function_tool_call import ResponseFunctionToolCall -from utils import make_serving +from utils import StreamFixture, engine_chunk, make_serving from sglang.srt.entrypoints.context import SimpleContext from sglang.srt.entrypoints.openai.protocol import ( @@ -63,7 +63,10 @@ class InputMessageConstructionTestCase(CustomTestCase): type="message", ), ] - serving.msg_store["resp_prev"] = [{"role": "user", "content": "old input"}] + serving.msg_store["resp_prev"] = [ + {"role": "user", "content": "old input"}, + *[item.model_dump(exclude_none=True) for item in prev_response.output], + ] request = ResponsesRequest( model="x", @@ -82,12 +85,198 @@ class InputMessageConstructionTestCase(CustomTestCase): {"role": "user", "content": "old input"}, { "role": "assistant", - "content": "first answer part\nsecond answer part", + "content": [ + {"type": "text", "text": "first answer part"}, + {"type": "text", "text": "second answer part"}, + ], }, {"role": "user", "content": "new input"}, ], ) + def test_stored_tool_turn_matches_client_replay_without_old_instructions(self): + for stream in (False, True): + with self.subTest(stream=stream): + serving = make_serving() + serving.reasoning_parser = "deepseek-r1" + serving.tool_call_parser = None + request = ResponsesRequest( + model="x", + input="old input", + instructions="OLD INSTRUCTION", + tools=[{"type": "function", "name": "lookup"}], + tool_choice="required", + store=True, + stream=stream, + ) + chunk = engine_chunk( + 'secret plan[{"name":"lookup","parameters":{}}]', + finish=True, + ) + if stream: + StreamFixture(serving, request).run([chunk]) + response = serving.response_store[request.request_id] + else: + context = SimpleContext() + context.append_output(chunk) + + async def empty(): + if False: + yield + + response = asyncio.run( + serving.responses_full_generator( + request, + {}, + empty(), + context, + "x", + Mock(), + RequestResponseMetadata(request_id=request.request_id), + require_reasoning=False, + ) + ) + call = next( + item for item in response.output if item.type == "function_call" + ) + result = { + "type": "function_call_output", + "call_id": call.call_id, + "output": "answer 42", + } + for instructions in ("NEW INSTRUCTION", None): + followup = ResponsesRequest( + model="x", + previous_response_id=response.id, + input=[result], + instructions=instructions, + store=False, + ) + explicit = ResponsesRequest( + model="x", + input=[{"role": "user", "content": "old input"}] + + [item.model_dump() for item in response.output] + + [result], + instructions=instructions, + store=False, + ) + messages = serving._construct_input_messages(followup, response) + self.assertEqual( + messages, serving._construct_input_messages(explicit) + ) + self.assertNotIn("OLD INSTRUCTION", str(messages)) + self.assertIn("secret plan", str(messages)) + self.assertIn(call.call_id, str(messages)) + self.assertEqual( + [item["type"] for item in serving.msg_store[response.id][1:]], + ["reasoning", "function_call"], + ) + + def test_harmony_instructions_are_rebuilt_for_each_request(self): + serving = make_serving() + serving.use_harmony = True + previous = Mock(id="resp_previous", output=[]) + first = ResponsesRequest( + model="x", input="old input", instructions="OLD INSTRUCTION" + ) + messages = serving._construct_input_messages_with_harmony(first, None) + serving.msg_store[previous.id] = messages[2:] + for instructions in ("NEW INSTRUCTION", None): + request = ResponsesRequest( + model="x", + input="next", + instructions=instructions, + previous_response_id=previous.id, + ) + actual = serving._construct_input_messages_with_harmony(request, previous) + expected = serving._construct_input_messages_with_harmony(request, None) + self.assertEqual(actual[:2], expected[:2]) + self.assertEqual(actual[2:], messages[2:] + expected[2:]) + + def test_harmony_replays_output_text_and_encoded_reasoning(self): + from sglang.srt.entrypoints.harmony_utils import parse_response_input + from sglang.srt.entrypoints.openai.responses_adapters import ( + encode_reasoning_state, + ) + + message = parse_response_input( + { + "type": "message", + "role": "assistant", + "content": [ + {"type": "output_text", "text": "assistant-only secret 42"}, + ], + "phase": "final_answer", + }, + [], + ) + self.assertEqual(message.content[0].text, "assistant-only secret 42") + reasoning = parse_response_input( + { + "type": "reasoning", + "encrypted_content": encode_reasoning_state("private plan"), + }, + [], + ) + self.assertEqual(reasoning.content[0].text, "private plan") + self.assertEqual(reasoning.channel, "analysis") + + def test_harmony_replays_dict_tool_calls_and_results_in_one_input(self): + serving = make_serving() + request = ResponsesRequest( + model="x", + input=[ + { + "type": "function_call", + "name": "lookup", + "call_id": "call_1", + "arguments": "{}", + }, + { + "type": "function_call_output", + "call_id": "call_1", + "output": [{"type": "output_text", "text": "result"}], + }, + ], + store=False, + ) + messages = serving._construct_input_messages_with_harmony(request, None) + self.assertEqual(messages[-1].author.name, "functions.lookup") + self.assertEqual(messages[-1].content[0].text, "result") + + def test_harmony_message_channels_map_to_phases(self): + from sglang.srt.entrypoints.harmony_utils import ( + parse_output_message, + parse_response_input, + ) + + for phase in ("commentary", "final_answer"): + message = parse_response_input( + { + "role": "assistant", + "content": "answer", + "phase": phase, + }, + [], + ) + (item,) = parse_output_message(message) + self.assertEqual(item.phase, phase) + self.assertEqual(item.content[0].text, "answer") + + def test_replay_preserves_different_assistant_phases(self): + serving = make_serving() + request = ResponsesRequest( + model="x", + input=[ + {"role": "assistant", "content": "working", "phase": "commentary"}, + {"role": "assistant", "content": "answer", "phase": "final_answer"}, + ], + store=False, + ) + messages = serving._construct_input_messages(request) + self.assertEqual([m["phase"] for m in messages], ["commentary", "final_answer"]) + self.assertEqual([m["content"] for m in messages], ["working", "answer"]) + def test_input_parts_normalized_for_chat_templates(self): serving = make_serving() request = ResponsesRequest( @@ -204,6 +393,23 @@ class ChatToolForwardingTestCase(CustomTestCase): self.assertFalse(seen["parallel_tool_calls"]) self.assertEqual(processed.tool_call_constraint[0], "json_schema") + def test_harmony_forced_choices_explain_missing_routing_constraints(self): + serving = make_serving() + serving.use_harmony = True + for choice in ("none", "required", {"type": "function", "name": "lookup"}): + request = ResponsesRequest( + model="x", + input="hi", + tool_choice=choice, + tools=[{"type": "function", "name": "lookup"}], + store=False, + ) + response = asyncio.run(serving.create_responses(request)) + self.assertEqual(response.status_code, 400) + self.assertIn(b"recipient", response.body) + self.assertIn(b"tool_choice", response.body) + serving.tokenizer_manager.generate_request.assert_not_called() + def test_required_tool_choice_without_function_tool_returns_400(self): serving = make_serving() request = ResponsesRequest( @@ -458,11 +664,14 @@ class InputItemNormalizationTestCase(CustomTestCase): }, ) - def test_developer_role_becomes_system(self): + def test_developer_role_becomes_labelled_system(self): normalized = OpenAIServingResponses._normalize_response_message_for_chat( {"role": "developer", "content": "Be terse."} ) - self.assertEqual(normalized, {"role": "system", "content": "Be terse."}) + self.assertEqual( + normalized, + {"role": "system", "content": "Developer instructions:\nBe terse."}, + ) def test_function_call_output_becomes_tool_message(self): normalized = OpenAIServingResponses._normalize_response_message_for_chat( @@ -734,6 +943,7 @@ class OutputItemsTestCase(CustomTestCase): types = [type(item).__name__ for item in output_items] self.assertEqual(types, ["ResponseOutputMessage", "ResponseFunctionToolCall"]) + self.assertEqual(output_items[0].phase, "commentary") def test_required_tool_choice_parses_json_array_without_native_parser(self): serving = self.serving 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 b01cb24f8..3fdbccad6 100644 --- a/test/registered/unit/entrypoints/openai/test_serving_responses_stream.py +++ b/test/registered/unit/entrypoints/openai/test_serving_responses_stream.py @@ -1,8 +1,11 @@ +import asyncio import unittest -from unittest.mock import patch +from types import SimpleNamespace +from unittest.mock import Mock, patch from utils import ( StreamFixture, + collect_stream_events, engine_chunk, event_payloads, event_types, @@ -10,7 +13,10 @@ from utils import ( make_serving, ) -from sglang.srt.entrypoints.openai.protocol import ResponsesRequest +from sglang.srt.entrypoints.openai.protocol import ( + RequestResponseMetadata, + ResponsesRequest, +) from sglang.srt.runtime_context import publish, reset_context from sglang.srt.server_args import ServerArgs from sglang.test.ci.ci_register import register_cpu_ci @@ -97,6 +103,53 @@ class NonHarmonyStreamTestCase(CustomTestCase): seqs = [p["sequence_number"] for p in event_payloads(events)] self.assertEqual(seqs, list(range(len(seqs)))) + for payload in event_payloads(events): + if payload["type"] in ( + "response.output_item.added", + "response.output_item.done", + ): + self.assertEqual(payload["item"]["phase"], "final_answer") + self.assertEqual( + find_completed_event(events)["response"]["output"][0]["phase"], + "final_answer", + ) + + def test_truncated_and_aborted_streams_have_matching_terminal_events(self): + serving = make_serving() + for finish_reason, status in ( + ({"type": "length"}, "incomplete"), + ( + {"type": "abort", "status_code": 503, "message": "Worker unavailable"}, + "failed", + ), + ): + with self.subTest(status=status): + request = ResponsesRequest( + model="x", input="hi", stream=True, store=True + ) + chunk = engine_chunk("partial answer", finish=True) + chunk["meta_info"]["finish_reason"] = finish_reason + events = StreamFixture(serving, request).run([chunk]) + terminal = event_payloads(events)[-1] + self.assertEqual(terminal["type"], f"response.{status}") + self.assertEqual(terminal["response"]["status"], status) + self.assertNotIn("response.completed", event_types(events)) + stored = serving.response_store[request.request_id] + self.assertEqual(stored.status, status) + if status == "incomplete": + self.assertEqual( + terminal["response"]["incomplete_details"], + {"reason": "max_output_tokens"}, + ) + else: + self.assertEqual( + terminal["response"]["error"]["message"], "Worker unavailable" + ) + self.assertEqual( + [p["sequence_number"] for p in event_payloads(events)], + list(range(len(events))), + ) + def test_required_tool_choice_emits_function_call_events(self): serving = make_serving() serving.reasoning_parser = None @@ -143,6 +196,58 @@ class NonHarmonyStreamTestCase(CustomTestCase): ] self.assertIn("function_call", added_kinds) + def test_required_native_parser_matches_full_response(self): + serving = make_serving() + serving.reasoning_parser = None + serving.tool_call_parser = "hunyuan" + serving.tokenizer_manager.tokenizer.get_vocab.return_value = {"": 1} + raw = ( + "get_weather" + "cityBeijing" + "" + ) + for choice in ("required", {"type": "function", "name": "get_weather"}): + with self.subTest(choice=choice): + request = ResponsesRequest( + model="x", + input="hi", + stream=True, + store=False, + tool_choice=choice, + tools=[ + { + "type": "function", + "name": "get_weather", + "parameters": { + "type": "object", + "properties": {"city": {"type": "string"}}, + }, + } + ], + ) + (full_item,) = serving._make_response_output_items( + request, + raw, + serving.tokenizer_manager.tokenizer, + require_reasoning=False, + ) + events = StreamFixture(serving, request).run( + [ + engine_chunk(raw[:i], i, finish=i == len(raw)) + for i in range(1, len(raw) + 1) + ] + ) + (stream_item,) = find_completed_event(events)["response"]["output"] + self.assertEqual(stream_item["type"], "function_call") + self.assertEqual(stream_item["name"], full_item.name) + self.assertEqual(stream_item["arguments"], full_item.arguments) + deltas = "".join( + p["delta"] + for p in event_payloads(events) + if p["type"] == "response.function_call_arguments.delta" + ) + self.assertEqual(deltas, full_item.arguments) + def test_final_output_preserves_text_tool_text_order(self): from sglang.srt.function_call.core_types import ( StreamingParseResult, @@ -211,6 +316,17 @@ class NonHarmonyStreamTestCase(CustomTestCase): 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.") + self.assertEqual(output[0]["phase"], "commentary") + self.assertEqual(output[2]["phase"], "final_answer") + for payload in event_payloads(events): + if ( + payload["type"] + in ("response.output_item.added", "response.output_item.done") + and payload["item"]["type"] == "message" + ): + self.assertEqual( + payload["item"]["phase"], output[payload["output_index"]]["phase"] + ) def test_reasoning_parser_flushed_at_stream_end(self): """Bug regression: the stream loop never drained text the reasoning @@ -242,6 +358,210 @@ class NonHarmonyStreamTestCase(CustomTestCase): self.assertEqual(streamed, "Answer<|e") +class HarmonyStreamLifecycleTestCase(CustomTestCase): + def test_truncated_harmony_arguments_close_the_emitted_item(self): + from openai_harmony import Role + + from sglang.srt.entrypoints.context import StreamingHarmonyContext + + serving = make_serving() + serving.use_harmony = True + request = ResponsesRequest(model="x", input="hi", stream=True, store=False) + context = Mock(spec=StreamingHarmonyContext) + context.messages = [] + context.parser = SimpleNamespace( + current_content='{"city":', + current_role=Role.ASSISTANT, + current_channel="commentary", + current_recipient="functions.lookup", + ) + context.num_prompt_tokens = 5 + context.num_output_tokens = 3 + context.num_cached_tokens = 0 + context.num_reasoning_tokens = 0 + context.finish_reason = {"type": "length"} + + async def generate(): + yield context + + events = asyncio.run( + collect_stream_events( + serving.responses_stream_generator( + request, + {}, + generate(), + context, + "x", + Mock(), + RequestResponseMetadata(request_id=request.request_id), + require_reasoning=False, + ) + ) + ) + payloads = event_payloads(events) + self.assertEqual(payloads[-1]["type"], "response.incomplete") + self.assertEqual( + payloads[-1]["response"]["incomplete_details"], + {"reason": "max_output_tokens"}, + ) + output = payloads[-1]["response"]["output"] + self.assertEqual(output[0]["arguments"], '{"city":') + self.assertEqual(output[0]["status"], "incomplete") + added = next( + p["item"] for p in payloads if p["type"] == "response.output_item.added" + ) + done = next( + p["item"] for p in payloads if p["type"] == "response.output_item.done" + ) + self.assertEqual(added["id"], done["id"]) + self.assertEqual(done, output[0]) + self.assertEqual(added["call_id"], done["call_id"]) + + def test_split_and_coalesced_messages_preserve_stream_items(self): + from openai_harmony import Message, Role, StreamState + + from sglang.srt.entrypoints.context import StreamingHarmonyContext + + reasoning = Message.from_role_and_content(Role.ASSISTANT, "plan").with_channel( + "analysis" + ) + commentary = Message.from_role_and_content( + Role.ASSISTANT, "checking" + ).with_channel("commentary") + call = ( + Message.from_role_and_content(Role.ASSISTANT, '{"city":"Beijing"}') + .with_channel("commentary") + .with_recipient("functions.lookup") + ) + code = ( + Message.from_role_and_content(Role.ASSISTANT, "print(42)") + .with_channel("commentary") + .with_recipient("python") + ) + search = ( + Message.from_role_and_content(Role.ASSISTANT, '{"query":"weather"}') + .with_channel("commentary") + .with_recipient("browser.search") + ) + final = Message.from_role_and_content(Role.ASSISTANT, "answer").with_channel( + "final" + ) + snapshots = [ + ([], "analysis", None, "pl"), + ([reasoning], "commentary", None, "checking"), + ([reasoning, commentary], "commentary", "functions.lookup", '{"city":'), + ([reasoning, commentary, call], "commentary", "python", "print("), + ([reasoning, commentary, call, code, search], "final", None, "ans"), + ([reasoning, commentary, call, code, search, final], None, None, ""), + ] + for chunks in (snapshots, [snapshots[0], snapshots[-1]]): + with self.subTest(chunk_count=len(chunks)): + serving = make_serving() + serving.use_harmony = True + request = ResponsesRequest( + model="x", + input="hi", + stream=True, + store=True, + include=["reasoning.encrypted_content"], + reasoning={"summary": "auto"}, + ) + context = StreamingHarmonyContext.__new__(StreamingHarmonyContext) + context.num_init_messages = 2 + context.num_prompt_tokens = 5 + context.num_output_tokens = 10 + context.num_cached_tokens = 0 + context.num_reasoning_tokens = 0 + context.finish_reason = {"type": "stop"} + context.last_tok = None + context.encoding = Mock() + context.encoding.stop_tokens_for_assistant_actions.return_value = [] + + async def generate(): + for messages, channel, recipient, text in chunks: + context.parser = SimpleNamespace( + messages=messages, + current_role=Role.ASSISTANT, + current_channel=channel, + current_recipient=recipient, + current_content=text, + last_content_delta=text, + state=StreamState.CONTENT, + ) + yield context + + events = asyncio.run( + collect_stream_events( + serving.responses_stream_generator( + request, + {}, + generate(), + context, + "x", + Mock(), + RequestResponseMetadata(request_id=request.request_id), + require_reasoning=False, + ) + ) + ) + payloads = event_payloads(events) + output = find_completed_event(events)["response"]["output"] + self.assertEqual( + [item["type"] for item in output], + [ + "reasoning", + "message", + "function_call", + "code_interpreter_call", + "web_search_call", + "message", + ], + ) + added = [ + p for p in payloads if p["type"] == "response.output_item.added" + ] + done = [p for p in payloads if p["type"] == "response.output_item.done"] + self.assertEqual([p["output_index"] for p in added], list(range(6))) + self.assertEqual([p["output_index"] for p in done], list(range(6))) + self.assertEqual( + [p["item"]["id"] for p in added], [i["id"] for i in output] + ) + self.assertEqual(len({i["id"] for i in output}), 6) + self.assertEqual([p["item"] for p in done], output) + self.assertEqual( + [p["sequence_number"] for p in payloads], list(range(len(payloads))) + ) + for index, field, event_type in ( + (0, "content", "response.reasoning_text.delta"), + (1, "content", "response.output_text.delta"), + (2, "arguments", "response.function_call_arguments.delta"), + (5, "content", "response.output_text.delta"), + ): + text = "".join( + p["delta"] + for p in payloads + if p["type"] == event_type and p["output_index"] == index + ) + expected = output[index][field] + self.assertEqual( + text, expected[0]["text"] if field == "content" else expected + ) + self.assertEqual(output[1]["phase"], "commentary") + self.assertEqual(output[5]["phase"], "final_answer") + self.assertTrue(output[0]["encrypted_content"]) + self.assertEqual(output[3]["code"], "print(42)") + for event_type in ( + "response.code_interpreter_call_code.done", + "response.code_interpreter_call.completed", + "response.web_search_call.completed", + ): + self.assertIn(event_type, event_types(events)) + self.assertEqual( + serving.response_store[request.request_id].model_dump()["output"], + output, + ) + + class MultiToolCallStreamingOrderTestCase(CustomTestCase): """The wire order of message / function_call items across tool-call deltas.""" diff --git a/test/registered/unit/entrypoints/openai/utils.py b/test/registered/unit/entrypoints/openai/utils.py index 60909de2d..1d248945b 100644 --- a/test/registered/unit/entrypoints/openai/utils.py +++ b/test/registered/unit/entrypoints/openai/utils.py @@ -42,12 +42,17 @@ if torch is not None: class MockTokenizerManager: + # The model id the cases address; /v1/responses validates ``model`` against it. + SERVED_MODEL_NAME = "x" + 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.served_model_name = self.SERVED_MODEL_NAME + self.lora_registry = None self.server_args = Mock( enable_cache_report=False, reasoning_parser=None,