[Feat][Responses API] Support custom tools, encrypted reasoning replay, developer tier and model validation (#38690)
Co-authored-by: Xinyuan Tong <xinyuantong.cs@gmail.com> Co-authored-by: Xinyuan Tong <115166877+JustinTong0323@users.noreply.github.com>
This commit is contained in:
Shijin Zhang
co-authored by
Xinyuan Tong
Xinyuan Tong
Xinyuan Tong
Xinyuan Tong
parent
6dc7b3421b
commit
925e684a88
@@ -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 = (
|
||||
"<tool_call>emit_command"
|
||||
"<arg_key>input</arg_key><arg_value>pwd</arg_value>"
|
||||
"</tool_call>"
|
||||
)
|
||||
|
||||
|
||||
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("because</think>answer", 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,
|
||||
"because</think>answer",
|
||||
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()
|
||||
@@ -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(
|
||||
'<think>secret plan</think>[{"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
|
||||
|
||||
@@ -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 = {"<tool_sep>": 1}
|
||||
raw = (
|
||||
"<tool_calls><tool_call>get_weather<tool_sep>"
|
||||
"<arg_key>city</arg_key><arg_value>Beijing</arg_value>"
|
||||
"</tool_call></tool_calls>"
|
||||
)
|
||||
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."""
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user