ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

studio/backend: route OpenAI traffic through /v1/responses

Browse source 
OpenAI's new flagship models (gpt-5.x) return 404 'This is not a chat
model' on /v1/chat/completions and are only reachable via /v1/responses.
Add a dedicated _stream_openai_responses path in ExternalProviderClient
that:

- Translates outbound messages into the Responses shape: system messages
  are folded into the top-level 'instructions' field, user/assistant
  messages become {role, content} items with input_text / input_image
  content parts (data URLs and https URLs both pass through).
- Drops presence_penalty / top_k / frequency_penalty, none of which the
  Responses contract accepts.
- Translates inbound SSE events back into OpenAI Chat Completions
  chunks so the frontend keeps a single SSE shape:
    response.output_text.delta  -> delta chunk with content
    response.completed          -> chunk with finish_reason='stop'
    response.incomplete         -> chunk with finish_reason='length'
    response.failed / error     -> propagated error SSE line
  Stream terminates with data: [DONE] (Responses emits this verbatim).

stream_chat_completion dispatches all provider_type='openai' calls to
this path; other OpenAI-compatible providers (mistral, gemini, etc.)
continue to use /v1/chat/completions.

Frontend provider-capabilities map updated to hide presence_penalty for
OpenAI in the chat settings panel, matching the new request contract.

Includes unit coverage in tests/test_openai_responses_translation.py
exercising the request body translation, image-part rewriting, and
SSE-to-chat-completions translation via httpx.MockTransport.
This commit is contained in:
Roland Tannous 2026-05-12 10:52:30 +04:00
commit f32b626b54
3 changed files with 487 additions and 1 deletions
Download patch file Download diff file Expand all files Collapse all files

232
studio/backend/core/inference/external_provider.py
View file

@ -88,6 +88,18 @@ class ExternalProviderClient:
yield line
return
# OpenAI moved their flagship models (gpt-5.x) off /v1/chat/completions
# — those endpoints return 404 with "This is not a chat model" for the
# new families. Route all OpenAI traffic through /v1/responses instead;
# we translate the Responses SSE back into Chat Completions chunks so
# the frontend stays endpoint-agnostic.
if self.provider_type == "openai":
async for line in self._stream_openai_responses(
messages, model, temperature, top_p, max_tokens
):
yield line
return
body: dict[str, Any] = {
"model": model,
"messages": messages,
@ -395,6 +407,226 @@ class ExternalProviderClient:
self.provider_type,
)
async def _stream_openai_responses(
self,
messages: list[dict[str, Any]],
model: str,
temperature: float,
top_p: float,
max_tokens: Optional[int],
) -> AsyncGenerator[str, None]:
"""
Call OpenAI's /v1/responses endpoint and translate its SSE stream back
into OpenAI Chat Completions chunk format.
The Responses API uses a different request shape (``input`` instead of
``messages``, ``instructions`` for system prompts, ``max_output_tokens``
for the budget) and emits event-typed SSE frames (e.g.
``response.output_text.delta``) rather than chat-completion chunks.
``presence_penalty`` / ``top_k`` are not part of the Responses contract
and are dropped here intentionally.
"""
import json as _json
# Split system messages out into a single `instructions` string and
# translate user/assistant messages into the Responses input shape.
instructions_parts: list[str] = []
input_items: list[dict[str, Any]] = []
for msg in messages:
role = msg.get("role")
content = msg.get("content", "")
if role == "system":
if isinstance(content, str):
if content:
instructions_parts.append(content)
elif isinstance(content, list):
for part in content:
if part.get("type") == "text" and part.get("text"):
instructions_parts.append(part["text"])
continue
if isinstance(content, str):
input_items.append({"role": role, "content": content})
continue
if isinstance(content, list):
translated_parts: list[dict[str, Any]] = []
for part in content:
part_type = part.get("type")
if part_type == "text":
translated_parts.append(
{"type": "input_text", "text": part.get("text", "")}
)
elif part_type == "image_url":
url = part.get("image_url", {}).get("url", "")
if url:
# Responses takes image_url as a flat string (both
# https:// URLs and data: URLs are accepted).
translated_parts.append(
{"type": "input_image", "image_url": url}
)
if translated_parts:
input_items.append({"role": role, "content": translated_parts})
body: dict[str, Any] = {
"model": model,
"input": input_items,
"stream": True,
"temperature": temperature,
"top_p": top_p,
}
if instructions_parts:
body["instructions"] = "\n\n".join(instructions_parts)
if max_tokens is not None:
body["max_output_tokens"] = max_tokens
url = f"{self.base_url}/responses"
completion_id = f"chatcmpl-openai-{model.replace('/', '-')}"
logger.info("Proxying OpenAI Responses API to %s (model=%s)", url, model)
try:
async with _http_client.stream(
"POST",
url,
json = body,
headers = self._auth_headers(),
timeout = self._timeout,
) as response:
if response.status_code != 200:
error_body = await response.aread()
error_text = error_body.decode("utf-8", errors = "replace")
logger.error(
"OpenAI Responses returned %d: %s",
response.status_code,
error_text[:500],
)
yield _error_sse_line(
response.status_code, error_text, self.provider_type
)
return
# NOTE: same manual __anext__ loop as stream_chat_completion —
# see comment there for the GeneratorExit / aclose ordering.
lines_gen = response.aiter_lines().__aiter__()
done_emitted = False
try:
while True:
try:
line = await lines_gen.__anext__()
except StopAsyncIteration:
break
if not line or line.startswith("event:"):
continue
if not line.startswith("data:"):
continue
data_str = line[len("data:") :].strip()
if not data_str:
continue
if data_str == "[DONE]":
if not done_emitted:
yield "data: [DONE]"
done_emitted = True
break
try:
event = _json.loads(data_str)
except _json.JSONDecodeError:
continue
event_type = event.get("type")
if event_type == "response.output_text.delta":
delta_text = event.get("delta", "")
if delta_text:
chunk = {
"id": completion_id,
"object": "chat.completion.chunk",
"choices": [
{
"index": 0,
"delta": {"content": delta_text},
"finish_reason": None,
}
],
}
yield f"data: {_json.dumps(chunk)}"
elif event_type == "response.completed":
chunk = {
"id": completion_id,
"object": "chat.completion.chunk",
"choices": [
{
"index": 0,
"delta": {},
"finish_reason": "stop",
}
],
}
yield f"data: {_json.dumps(chunk)}"
elif event_type == "response.incomplete":
chunk = {
"id": completion_id,
"object": "chat.completion.chunk",
"choices": [
{
"index": 0,
"delta": {},
"finish_reason": "length",
}
],
}
yield f"data: {_json.dumps(chunk)}"
elif event_type in ("response.failed", "error"):
# Surface the failure to the client; let the
# outer route emit [DONE] as part of its cleanup.
error_payload = event.get("response", {}).get(
"error", {}
) or {
"message": event.get("message", "Unknown error"),
"code": event.get("code"),
}
yield _error_sse_line(
502,
_json.dumps(error_payload),
self.provider_type,
)
break
except GeneratorExit:
await response.aclose()
await lines_gen.aclose()
raise
finally:
await response.aclose()
await lines_gen.aclose()
except httpx.ConnectError as exc:
logger.error("Connection error to %s: %s", self.provider_type, exc)
yield _error_sse_line(
502,
f"Failed to connect to {self.provider_type}: {exc}",
self.provider_type,
)
except httpx.ReadTimeout as exc:
logger.error("Read timeout from %s: %s", self.provider_type, exc)
yield _error_sse_line(
504,
f"Timeout waiting for {self.provider_type} response",
self.provider_type,
)
except httpx.HTTPError as exc:
logger.error("HTTP error from %s: %s", self.provider_type, exc)
yield _error_sse_line(
502,
f"Error communicating with {self.provider_type}: {exc}",
self.provider_type,
)
async def chat_completion(
self,
messages: list[dict[str, Any]],

246
studio/backend/tests/test_openai_responses_translation.py Normal file
View file

@ -0,0 +1,246 @@
# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
"""
Unit tests for the OpenAI `/v1/responses` translation in external_provider.
Covers:
- Request body shape: system messages collapse into `instructions`, user/
assistant messages go into `input`, sampling knobs Responses does not
support (presence_penalty, top_k) are not forwarded.
- SSE translation: `response.output_text.delta` events become OpenAI Chat
Completions chunks, `response.completed` emits a `finish_reason: stop`
chunk, the stream terminates with `data: [DONE]`.
- Image parts in user content are rewritten from Chat Completions
`{type: image_url, image_url: {url}}` into Responses
`{type: input_image, image_url: <url>}`.
"""
import asyncio
import json
import httpx
from core.inference import external_provider as ep_mod
from core.inference.external_provider import ExternalProviderClient
def _drive(coro):
return asyncio.new_event_loop().run_until_complete(coro)
async def _collect(agen):
out = []
async for line in agen:
out.append(line)
return out
def _mock_http_client(monkeypatch, handler):
transport = httpx.MockTransport(handler)
monkeypatch.setattr(ep_mod, "_http_client", httpx.AsyncClient(transport = transport))
def _make_client() -> ExternalProviderClient:
return ExternalProviderClient(
provider_type = "openai",
base_url = "https://api.openai.com/v1",
api_key = "sk-test",
)
def _responses_sse(events: list[dict]) -> bytes:
"""Serialize a list of Responses-API event dicts as an SSE byte stream."""
chunks: list[str] = []
for event in events:
chunks.append(f"event: {event['type']}")
chunks.append(f"data: {json.dumps(event)}")
chunks.append("")
chunks.append("data: [DONE]")
chunks.append("")
return ("\n".join(chunks) + "\n").encode("utf-8")
def test_responses_request_body_uses_input_and_instructions(monkeypatch):
captured: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
captured["url"] = str(request.url)
captured["body"] = json.loads(request.content.decode("utf-8"))
return httpx.Response(
200,
content = _responses_sse([{"type": "response.completed", "response": {}}]),
headers = {"content-type": "text/event-stream"},
)
_mock_http_client(monkeypatch, handler)
async def run():
client = _make_client()
async for _ in client._stream_openai_responses(
messages = [
{"role": "system", "content": "You are concise."},
{"role": "user", "content": "Hi"},
],
model = "gpt-5.5",
temperature = 0.5,
top_p = 0.9,
max_tokens = 512,
):
pass
await client.close()
_drive(run())
assert captured["url"] == "https://api.openai.com/v1/responses"
body = captured["body"]
assert body["model"] == "gpt-5.5"
assert body["instructions"] == "You are concise."
assert body["input"] == [{"role": "user", "content": "Hi"}]
assert body["temperature"] == 0.5
assert body["top_p"] == 0.9
assert body["max_output_tokens"] == 512
assert body["stream"] is True
# Responses API does not accept these — frontend caps + backend path both
# strip them; make sure we never silently forward them.
assert "presence_penalty" not in body
assert "frequency_penalty" not in body
assert "top_k" not in body
assert "messages" not in body
def test_responses_translates_image_parts(monkeypatch):
captured: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
captured["body"] = json.loads(request.content.decode("utf-8"))
return httpx.Response(
200,
content = _responses_sse([{"type": "response.completed", "response": {}}]),
headers = {"content-type": "text/event-stream"},
)
_mock_http_client(monkeypatch, handler)
async def run():
client = _make_client()
async for _ in client._stream_openai_responses(
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "What is this?"},
{
"type": "image_url",
"image_url": {"url": "data:image/png;base64,AAA"},
},
],
}
],
model = "gpt-5.5",
temperature = 0.7,
top_p = 0.95,
max_tokens = None,
):
pass
await client.close()
_drive(run())
parts = captured["body"]["input"][0]["content"]
assert parts[0] == {"type": "input_text", "text": "What is this?"}
assert parts[1] == {
"type": "input_image",
"image_url": "data:image/png;base64,AAA",
}
# No max_output_tokens key when caller passes max_tokens=None.
assert "max_output_tokens" not in captured["body"]
def test_responses_sse_translates_to_chat_completions_chunks(monkeypatch):
def handler(request: httpx.Request) -> httpx.Response:
events = [
{"type": "response.created"},
{"type": "response.output_text.delta", "delta": "Hello"},
{"type": "response.output_text.delta", "delta": ", world"},
{"type": "response.completed", "response": {}},
]
return httpx.Response(
200,
content = _responses_sse(events),
headers = {"content-type": "text/event-stream"},
)
_mock_http_client(monkeypatch, handler)
async def run():
client = _make_client()
lines = await _collect(
client._stream_openai_responses(
messages = [{"role": "user", "content": "hi"}],
model = "gpt-5.5",
temperature = 0.7,
top_p = 0.95,
max_tokens = None,
)
)
await client.close()
return lines
lines = _drive(run())
# Drop empty / non-data lines for assertion clarity.
data_lines = [line for line in lines if line.startswith("data:")]
payloads = []
for line in data_lines:
raw = line[len("data:") :].strip()
if raw == "[DONE]":
payloads.append("[DONE]")
else:
payloads.append(json.loads(raw))
# Two text deltas, one terminal chunk, then [DONE].
assert payloads[0]["choices"][0]["delta"]["content"] == "Hello"
assert payloads[0]["choices"][0]["finish_reason"] is None
assert payloads[1]["choices"][0]["delta"]["content"] == ", world"
assert payloads[2]["choices"][0]["delta"] == {}
assert payloads[2]["choices"][0]["finish_reason"] == "stop"
assert payloads[-1] == "[DONE]"
def test_responses_response_incomplete_maps_to_length_finish_reason(monkeypatch):
def handler(request: httpx.Request) -> httpx.Response:
events = [
{"type": "response.output_text.delta", "delta": "partial"},
{"type": "response.incomplete", "response": {}},
]
return httpx.Response(
200,
content = _responses_sse(events),
headers = {"content-type": "text/event-stream"},
)
_mock_http_client(monkeypatch, handler)
async def run():
client = _make_client()
lines = await _collect(
client._stream_openai_responses(
messages = [{"role": "user", "content": "hi"}],
model = "gpt-5.5",
temperature = 0.7,
top_p = 0.95,
max_tokens = 4,
)
)
await client.close()
return lines
lines = _drive(run())
finish_reasons = [
json.loads(line[len("data:") :].strip())["choices"][0]["finish_reason"]
for line in lines
if line.startswith("data:")
and line[len("data:") :].strip() not in ("", "[DONE]")
]
assert "length" in finish_reasons

10
studio/frontend/src/features/chat/provider-capabilities.ts
View file

@ -39,7 +39,15 @@ const ALL_SUPPORTED: ProviderCapabilities = {
};
const PROVIDER_CAPABILITIES: Record<string, ProviderCapabilities> = {
openai: OPENAI_COMPAT_BASE,
// OpenAI's flagship models (gpt-5.x) now require /v1/responses, and the
// Responses API drops presence/frequency penalty from the contract — see
// backend external_provider._stream_openai_responses for the proxy.
openai: {
topK: false,
minP: false,
repetitionPenalty: false,
presencePenalty: false,
},
// Anthropic's Messages API accepts top_k but not presence/frequency penalty.
anthropic: {
topK: true,