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

Merge branch 'main' into fix_tokenizer_save_gemma

Browse source
This commit is contained in:
Datta Nimmaturi 2026-04-22 08:58:52 +05:30 • committed by GitHub
commit 576c62edf0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1876 additions and 152 deletions
Download patch file Download diff file Expand all files Collapse all files

139
studio/backend/core/inference/anthropic_compat.py
View file

@ -13,11 +13,44 @@ import json
from typing import Any, Optional, Union
def _anthropic_image_block_to_openai_part(block: dict) -> Optional[dict]:
"""Translate one Anthropic ``image`` block to an OpenAI ``image_url`` part.
Accepts both source shapes:
- ``{"type": "base64", "media_type": "image/jpeg", "data": "..."}``
- ``{"type": "url", "url": "https://..."}``
Returns ``None`` when the source is malformed so the caller can skip it.
"""
source = block.get("source") or {}
stype = source.get("type")
if stype == "base64":
data = source.get("data")
if not data:
return None
media_type = source.get("media_type") or "image/jpeg"
return {
"type": "image_url",
"image_url": {"url": f"data:{media_type};base64,{data}"},
}
if stype == "url":
url = source.get("url")
if not url:
return None
return {"type": "image_url", "image_url": {"url": url}}
return None
def anthropic_messages_to_openai(
messages: list[dict],
system: Optional[Union[str, list]] = None,
) -> list[dict]:
"""Convert Anthropic messages + system to OpenAI-format message dicts."""
"""Convert Anthropic messages + system to OpenAI-format message dicts.
User messages that carry ``image`` blocks are emitted as OpenAI
multimodal content arrays (``[{type: "text", ...}, {type: "image_url", ...}]``)
so they flow through llama-server's native vision pathway.
"""
result: list[dict] = []
# System prompt
@ -42,54 +75,76 @@ def anthropic_messages_to_openai(
result.append({"role": role, "content": content})
continue
# Content is a list of blocks
text_parts: list[str] = []
tool_calls: list[dict] = []
tool_results: list[dict] = []
for block in content:
b = block if isinstance(block, dict) else block.model_dump()
btype = b.get("type", "")
if btype == "text":
text_parts.append(b["text"])
elif btype == "tool_use":
tool_calls.append(
{
"id": b["id"],
"type": "function",
"function": {
"name": b["name"],
"arguments": json.dumps(b["input"]),
},
}
)
elif btype == "tool_result":
tc = b.get("content", "")
if isinstance(tc, list):
tc = " ".join(
p["text"]
for p in tc
if isinstance(p, dict) and p.get("type") == "text"
)
tool_results.append(
{
"role": "tool",
"tool_call_id": b["tool_use_id"],
"content": str(tc),
}
)
if role == "assistant":
# Assistant content carries text + tool_use; images aren't
# part of Anthropic's assistant content model.
text_parts: list[str] = []
tool_calls: list[dict] = []
for block in content:
b = block if isinstance(block, dict) else block.model_dump()
btype = b.get("type", "")
if btype == "text":
text_parts.append(b["text"])
elif btype == "tool_use":
tool_calls.append(
{
"id": b["id"],
"type": "function",
"function": {
"name": b["name"],
"arguments": json.dumps(b["input"]),
},
}
)
msg_dict: dict[str, Any] = {"role": "assistant"}
if text_parts:
msg_dict["content"] = "\n".join(text_parts)
if tool_calls:
msg_dict["tool_calls"] = tool_calls
result.append(msg_dict)
elif role == "user":
if text_parts:
result.append({"role": "user", "content": "\n".join(text_parts)})
continue
if role == "user":
# Build an ordered part list so text/image interleaving is
# preserved (e.g. [text, image, text, image]). tool_result
# blocks become their own OpenAI "tool" role messages.
user_parts: list[dict] = []
has_image = False
tool_results: list[dict] = []
for block in content:
b = block if isinstance(block, dict) else block.model_dump()
btype = b.get("type", "")
if btype == "text":
user_parts.append({"type": "text", "text": b["text"]})
elif btype == "image":
part = _anthropic_image_block_to_openai_part(b)
if part is not None:
user_parts.append(part)
has_image = True
elif btype == "tool_result":
tc = b.get("content", "")
if isinstance(tc, list):
tc = " ".join(
p["text"]
for p in tc
if isinstance(p, dict) and p.get("type") == "text"
)
tool_results.append(
{
"role": "tool",
"tool_call_id": b["tool_use_id"],
"content": str(tc),
}
)
if has_image:
result.append({"role": "user", "content": user_parts})
else:
# No images — collapse text parts to a plain string so
# existing text-only callers keep their simple shape.
text = "\n".join(p["text"] for p in user_parts)
if text:
result.append({"role": "user", "content": text})
for tr in tool_results:
result.append(tr)

207
studio/backend/models/inference.py
View file

@ -599,23 +599,172 @@ class ResponsesInputImagePart(BaseModel):
detail: Optional[Literal["auto", "low", "high"]] = "auto"
ResponsesContentPart = Union[ResponsesInputTextPart, ResponsesInputImagePart]
class ResponsesOutputTextPart(BaseModel):
"""Assistant ``output_text`` content part replayed on subsequent turns.
When a client (OpenAI Codex CLI, OpenAI Python SDK agents) loops on a
stateless Responses endpoint, prior assistant messages are round-tripped
as ``{"role":"assistant","content":[{"type":"output_text","text":...,
"annotations":[],"logprobs":[]}]}``. We preserve the text and ignore
the annotations/logprobs metadata when flattening into Chat Completions.
"""
type: Literal["output_text"]
text: str
annotations: Optional[list] = None
logprobs: Optional[list] = None
model_config = {"extra": "allow"}
class ResponsesUnknownContentPart(BaseModel):
"""Catch-all for content-part types we don't model explicitly.
Keeps validation green when a client sends newer part types (e.g.
``input_audio``, ``input_file``) we haven't mapped; these are silently
skipped during normalisation rather than rejected with a 422.
"""
type: str
model_config = {"extra": "allow"}
ResponsesContentPart = Union[
ResponsesInputTextPart,
ResponsesInputImagePart,
ResponsesOutputTextPart,
ResponsesUnknownContentPart,
]
class ResponsesInputMessage(BaseModel):
"""A single message in the Responses API input array."""
type: Optional[Literal["message"]] = None
role: Literal["system", "user", "assistant", "developer"]
content: Union[str, list[ResponsesContentPart]]
# Codex (gpt-5.3-codex+) attaches a `phase` field ("commentary" |
# "final_answer") to assistant messages and requires clients to preserve
# it on subsequent turns. We accept and round-trip it; llama-server does
# not care about it.
model_config = {"extra": "allow"}
class ResponsesFunctionCallInputItem(BaseModel):
"""A prior assistant function_call being replayed in a multi-turn Responses input.
The Responses API represents tool calls as top-level input items (not
nested inside assistant messages), correlated across turns by ``call_id``.
"""
type: Literal["function_call"]
id: Optional[str] = Field(
None, description = "Item id assigned by the server (e.g. fc_...)"
)
call_id: str = Field(
...,
description = "Correlation id matching a function_call_output on the next turn.",
)
name: str
arguments: str = Field(
..., description = "JSON string of the arguments the model produced."
)
status: Optional[Literal["in_progress", "completed", "incomplete"]] = None
class ResponsesFunctionCallOutputInputItem(BaseModel):
"""A tool result supplied by the client for a prior function_call.
Replaces Chat Completions' ``role="tool"`` message. Correlated to the
originating call by ``call_id``.
"""
type: Literal["function_call_output"]
id: Optional[str] = None
call_id: str
output: Union[str, list] = Field(
..., description = "String or content-array result of the tool call."
)
status: Optional[Literal["in_progress", "completed", "incomplete"]] = None
class ResponsesUnknownInputItem(BaseModel):
"""Catch-all for Responses input item types we don't model explicitly.
Covers ``reasoning`` items (replayed from prior o-series / gpt-5 turns)
and any future item types the client may send. These items are dropped
during normalisation — llama-server-backed GGUFs cannot consume them —
but keeping them in the request-model union stops unrelated turns from
failing validation with a 422.
"""
type: str
model_config = {"extra": "allow"}
def _responses_input_item_discriminator(v: Any) -> str:
"""Route a Responses input item to the correct tagged variant.
Pydantic's default smart-union matching fails when one variant in the
union is tagged with a strict ``Literal`` (``function_call`` /
``function_call_output``) and the incoming dict uses a different
``type`` — the other variants' validation errors are hidden and the
outer ``Union[str, list[...]]`` reports a misleading "Input should be a
valid string" error. An explicit discriminator makes the routing
deterministic and lets us fall through to the catch-all.
"""
if isinstance(v, dict):
t = v.get("type")
r = v.get("role")
else:
t = getattr(v, "type", None)
r = getattr(v, "role", None)
if t == "function_call":
return "function_call"
if t == "function_call_output":
return "function_call_output"
if r is not None or t == "message":
return "message"
return "unknown"
ResponsesInputItem = Annotated[
Union[
Annotated[ResponsesInputMessage, Tag("message")],
Annotated[ResponsesFunctionCallInputItem, Tag("function_call")],
Annotated[ResponsesFunctionCallOutputInputItem, Tag("function_call_output")],
Annotated[ResponsesUnknownInputItem, Tag("unknown")],
],
Discriminator(_responses_input_item_discriminator),
]
class ResponsesFunctionTool(BaseModel):
"""Flat function-tool definition used by the Responses API request.
Unlike Chat Completions (which nests ``{"name": ..., "parameters": ...}``
inside a ``"function"`` key), the Responses API uses a flat shape with
``type``, ``name``, ``description``, ``parameters``, and ``strict`` at the
top level of each tool entry.
"""
type: Literal["function"]
name: str
description: Optional[str] = None
parameters: Optional[dict] = None
strict: Optional[bool] = None
class ResponsesRequest(BaseModel):
"""OpenAI Responses API request."""
model: str = Field("default", description = "Model identifier")
input: Union[str, list[ResponsesInputMessage]] = Field(
input: Union[str, list[ResponsesInputItem]] = Field(
default = [],
description = "Input text or message list",
description = "Input text or list of messages / function_call / function_call_output items",
)
instructions: Optional[str] = Field(
None, description = "System / developer instructions"
@ -625,9 +774,31 @@ class ResponsesRequest(BaseModel):
max_output_tokens: Optional[int] = Field(None, ge = 1)
stream: bool = Field(False, description = "Whether to stream the response via SSE")
# Accepted but ignored -- keeps SDK clients from failing on unsupported fields
tools: Optional[list] = None
tool_choice: Optional[Any] = None
# OpenAI function-calling fields — forwarded to llama-server via the
# Chat Completions pass-through (see routes/inference.py). Typed as a
# plain list so built-in tool shapes (``web_search``, ``file_search``,
# ``mcp``, ...) round-trip without validation errors — the translator
# picks out only ``type=="function"`` entries for forwarding.
tools: Optional[list[dict]] = Field(
None,
description = (
"Responses-shape function tool definitions. Entries with "
'`type="function"` are translated to the Chat Completions nested '
"shape before being forwarded to llama-server; other tool types "
"(built-in web_search, file_search, mcp, ...) are accepted for SDK "
"compatibility but ignored on the llama-server passthrough."
),
)
tool_choice: Optional[Any] = Field(
None,
description = (
"'auto' | 'required' | 'none' | {'type': 'function', 'name': ...} — "
"the Responses-shape forcing object is translated to the Chat "
"Completions nested shape internally."
),
)
parallel_tool_calls: Optional[bool] = None
previous_response_id: Optional[str] = None
store: Optional[bool] = None
metadata: Optional[dict] = None
@ -660,6 +831,28 @@ class ResponsesOutputMessage(BaseModel):
content: list[ResponsesOutputTextContent] = Field(default_factory = list)
class ResponsesOutputFunctionCall(BaseModel):
"""A function-call output item in the Responses API response.
Unlike Chat Completions (which nests tool calls inside the assistant
message), the Responses API emits each tool call as its own top-level
``output`` item so clients can correlate results via ``call_id`` on the
next turn.
"""
type: Literal["function_call"] = "function_call"
id: str = Field(default_factory = lambda: f"fc_{uuid.uuid4().hex[:12]}")
call_id: str
name: str
arguments: str = Field(
..., description = "JSON string of the arguments the model produced."
)
status: Literal["completed", "in_progress", "incomplete"] = "completed"
ResponsesOutputItem = Union[ResponsesOutputMessage, ResponsesOutputFunctionCall]
class ResponsesUsage(BaseModel):
"""Token usage for a Responses API response (input_tokens, not prompt_tokens)."""
@ -676,7 +869,7 @@ class ResponsesResponse(BaseModel):
created_at: int = Field(default_factory = lambda: int(time.time()))
status: Literal["completed", "in_progress", "failed"] = "completed"
model: str = "default"
output: list[ResponsesOutputMessage] = Field(default_factory = list)
output: list[ResponsesOutputItem] = Field(default_factory = list)
usage: ResponsesUsage = Field(default_factory = ResponsesUsage)
error: Optional[Any] = None
incomplete_details: Optional[Any] = None

773
studio/backend/routes/inference.py
View file

@ -12,7 +12,7 @@ import uuid
from pathlib import Path
from fastapi import APIRouter, Depends, HTTPException, Request, status
from fastapi.responses import StreamingResponse, JSONResponse, Response
from typing import Optional
from typing import Any, Optional, Union
import json
import httpx
import structlog
@ -27,6 +27,58 @@ import re as _re
from utils.models import extract_model_size_b as _extract_model_size_b
def _install_httpcore_asyncgen_silencer() -> None:
"""Silence benign httpx/httpcore asyncgen GC noise on Python 3.13.
When Studio proxies a streaming response from llama-server via httpx,
the innermost ``HTTP11ConnectionByteStream.__aiter__`` async generator
is finalised by Python's asyncgen GC hook on a task different from the
one that opened it. Its ``aclose`` path then calls
``anyio.Lock.acquire`` → ``cancel_shielded_checkpoint`` which enters a
``CancelScope`` on the finaliser task — Python 3.13 flags the
cross-task exit as ``"Attempted to exit cancel scope in a different
task"`` and prints ``"async generator ignored GeneratorExit"`` as an
unraisable warning.
This is a known httpx + httpcore + anyio interaction (see MCP SDK
python-sdk#831, agno #3556, chainlit #2361, langchain-mcp-adapters
#254). It is benign: the response has already been delivered with a
200. The streaming pass-throughs (``/v1/chat/completions``,
``/v1/messages``, ``/v1/responses``, ``/v1/completions``) already
manage their httpx lifecycle inside a single task with explicit
``aclose()`` of the lines iterator, response, and client; the errant
generator is not one we hold a reference to and therefore cannot
close ourselves.
We install a single process-wide unraisable hook that swallows just
this specific interaction — identified by the tuple of (RuntimeError
mentioning cancel scope / GeneratorExit) + (object repr referencing
HTTP11ConnectionByteStream) — and defers to the default hook for
everything else. The filter is idempotent.
"""
prior_hook = sys.unraisablehook
if getattr(prior_hook, "_unsloth_httpcore_silencer", False):
return
def _hook(unraisable):
exc_value = getattr(unraisable, "exc_value", None)
obj = getattr(unraisable, "object", None)
obj_repr = repr(obj) if obj is not None else ""
if (
isinstance(exc_value, RuntimeError)
and "HTTP11ConnectionByteStream" in obj_repr
and ("cancel scope" in str(exc_value) or "GeneratorExit" in str(exc_value))
):
return
prior_hook(unraisable)
_hook._unsloth_httpcore_silencer = True # type: ignore[attr-defined]
sys.unraisablehook = _hook
_install_httpcore_asyncgen_silencer()
def _friendly_error(exc: Exception) -> str:
"""Extract a user-friendly message from known llama-server errors."""
# httpx transport-layer failures reaching the managed llama-server —
@ -101,8 +153,14 @@ from models.inference import (
ResponsesInputMessage,
ResponsesInputTextPart,
ResponsesInputImagePart,
ResponsesOutputTextPart,
ResponsesUnknownContentPart,
ResponsesUnknownInputItem,
ResponsesFunctionCallInputItem,
ResponsesFunctionCallOutputInputItem,
ResponsesOutputTextContent,
ResponsesOutputMessage,
ResponsesOutputFunctionCall,
ResponsesUsage,
ResponsesResponse,
AnthropicMessagesRequest,
@ -2083,49 +2141,229 @@ async def openai_embeddings(
# =====================================================================
def _translate_responses_tools_to_chat(
tools: Optional[list[dict]],
) -> Optional[list[dict]]:
"""Translate Responses-shape function tools to the Chat Completions nested shape.
Responses uses a flat shape per tool entry::
{"type": "function", "name": "...", "description": "...",
"parameters": {...}, "strict": true}
The Chat Completions / llama-server passthrough expects the nested shape::
{"type": "function",
"function": {"name": "...", "description": "...",
"parameters": {...}, "strict": true}}
Only ``type=="function"`` entries are forwarded. Built-in Responses tools
(``web_search``, ``file_search``, ``mcp``, ...) are dropped because
llama-server does not implement them server-side; keeping them in the
request would produce an opaque upstream 400.
"""
if not tools:
return None
out: list[dict] = []
for tool in tools:
if not isinstance(tool, dict):
continue
if tool.get("type") != "function":
continue
fn: dict = {}
if "name" in tool:
fn["name"] = tool["name"]
if tool.get("description") is not None:
fn["description"] = tool["description"]
if tool.get("parameters") is not None:
fn["parameters"] = tool["parameters"]
if tool.get("strict") is not None:
fn["strict"] = tool["strict"]
out.append({"type": "function", "function": fn})
return out or None
def _translate_responses_tool_choice_to_chat(tool_choice: Any) -> Any:
"""Translate a Responses-shape ``tool_choice`` to the Chat Completions shape.
String values (``"auto"``/``"none"``/``"required"``) pass through unchanged.
The Responses forcing object ``{"type": "function", "name": "X"}`` is
converted to Chat Completions' ``{"type": "function", "function": {"name": "X"}}``.
Unknown / built-in tool choices are forwarded as-is; llama-server ignores
what it doesn't recognise.
"""
if tool_choice is None:
return None
if isinstance(tool_choice, str):
return tool_choice
if (
isinstance(tool_choice, dict)
and tool_choice.get("type") == "function"
and "name" in tool_choice
and "function" not in tool_choice
):
return {"type": "function", "function": {"name": tool_choice["name"]}}
return tool_choice
def _responses_message_text(content: Union[str, list]) -> str:
"""Flatten a ResponsesInputMessage ``content`` into a plain text string.
Used for system/developer message hoisting and for assistant-replay
(``output_text``) messages when images/unknown parts are irrelevant.
Returns an empty string for empty input.
"""
if isinstance(content, str):
return content
parts: list[str] = []
for part in content or []:
if isinstance(part, (ResponsesInputTextPart, ResponsesOutputTextPart)):
parts.append(part.text)
return "\n".join(parts)
def _normalise_responses_input(payload: ResponsesRequest) -> list[ChatMessage]:
"""Convert a ResponsesRequest into a list of ChatMessage for the completions backend."""
"""Convert a ResponsesRequest's ``input`` into Chat-format ``ChatMessage`` list.
Handles the three input item shapes allowed by the Responses API:
- ``ResponsesInputMessage`` — regular chat messages (text or multimodal).
- ``ResponsesFunctionCallInputItem`` — a prior assistant tool call replayed
on a follow-up turn. Converted into an assistant message carrying a
Chat Completions ``tool_calls`` entry keyed by ``call_id``.
- ``ResponsesFunctionCallOutputInputItem`` — a tool result the client is
returning. Converted into a ``role="tool"`` message with ``tool_call_id``
set to the originating ``call_id`` so llama-server can reconcile the
call with its result.
System / developer content is collected from ``instructions`` *and* from
any ``role="system"`` / ``role="developer"`` entries in ``input``, then
merged into a single ``role="system"`` message placed at the top of the
returned list. This satisfies strict chat templates (harmony / gpt-oss,
Qwen3, ...) whose Jinja raises ``"System message must be at the
beginning."`` when more than one system message is present or when a
system message appears after a user turn — the exact pattern the OpenAI
Codex CLI hits, since Codex sets ``instructions`` *and* also sends a
developer message in ``input``.
"""
system_parts: list[str] = []
messages: list[ChatMessage] = []
# System / developer instructions
if payload.instructions:
messages.append(ChatMessage(role = "system", content = payload.instructions))
system_parts.append(payload.instructions)
# Simple string input
if isinstance(payload.input, str):
if payload.input:
messages.append(ChatMessage(role = "user", content = payload.input))
if system_parts:
merged = "\n\n".join(p for p in system_parts if p)
return [ChatMessage(role = "system", content = merged), *messages]
return messages
# List of ResponsesInputMessage
for msg in payload.input:
role = "system" if msg.role == "developer" else msg.role
for item in payload.input:
if isinstance(item, ResponsesFunctionCallInputItem):
messages.append(
ChatMessage(
role = "assistant",
content = None,
tool_calls = [
{
"id": item.call_id,
"type": "function",
"function": {
"name": item.name,
"arguments": item.arguments,
},
}
],
)
)
continue
if isinstance(msg.content, str):
messages.append(ChatMessage(role = role, content = msg.content))
else:
# Convert Responses content parts -> Chat content parts
parts = []
for part in msg.content:
if isinstance(part, ResponsesInputTextPart):
parts.append(TextContentPart(type = "text", text = part.text))
elif isinstance(part, ResponsesInputImagePart):
parts.append(
ImageContentPart(
type = "image_url",
image_url = ImageUrl(url = part.image_url, detail = part.detail),
)
if isinstance(item, ResponsesFunctionCallOutputInputItem):
# Chat Completions `role="tool"` requires a string content; if a
# Responses client sends a content-array output, serialize it.
output = item.output
if not isinstance(output, str):
output = json.dumps(output)
messages.append(
ChatMessage(
role = "tool",
tool_call_id = item.call_id,
content = output,
)
)
continue
if isinstance(item, ResponsesUnknownInputItem):
# Reasoning items and any other unmodelled top-level Responses
# item types are silently dropped — llama-server-backed GGUFs
# cannot consume them and our lenient validation let them in so
# unrelated turns don't 422.
continue
# ResponsesInputMessage — hoist system/developer to the top, merge.
if item.role in ("system", "developer"):
hoisted = _responses_message_text(item.content)
if hoisted:
system_parts.append(hoisted)
continue
if isinstance(item.content, str):
messages.append(ChatMessage(role = item.role, content = item.content))
continue
# Assistant-replay turns come back as content = [output_text, ...].
# Chat Completions' assistant role expects a plain string, not a
# multimodal content array, so flatten output_text (and any stray
# input_text / unknown text) to a single string.
if item.role == "assistant":
text = _responses_message_text(item.content)
if text:
messages.append(ChatMessage(role = "assistant", content = text))
continue
# User (and any other remaining roles) — keep multimodal when
# present, drop unknown content parts silently.
parts: list = []
for part in item.content:
if isinstance(part, (ResponsesInputTextPart, ResponsesOutputTextPart)):
parts.append(TextContentPart(type = "text", text = part.text))
elif isinstance(part, ResponsesInputImagePart):
parts.append(
ImageContentPart(
type = "image_url",
image_url = ImageUrl(url = part.image_url, detail = part.detail),
)
messages.append(ChatMessage(role = role, content = parts if parts else ""))
)
# ResponsesUnknownContentPart and anything else: drop.
if parts:
# Collapse single-text-part content to a plain string so roles
# that reject multimodal arrays (e.g. legacy templates) still
# accept the message.
if len(parts) == 1 and isinstance(parts[0], TextContentPart):
messages.append(ChatMessage(role = item.role, content = parts[0].text))
else:
messages.append(ChatMessage(role = item.role, content = parts))
if system_parts:
merged = "\n\n".join(p for p in system_parts if p)
return [ChatMessage(role = "system", content = merged), *messages]
return messages
def _build_chat_request(
payload: ResponsesRequest, messages: list[ChatMessage], stream: bool
) -> ChatCompletionRequest:
"""Build a ChatCompletionRequest from a ResponsesRequest."""
chat_kwargs = dict(
"""Build a ChatCompletionRequest from a ResponsesRequest.
Tools and ``tool_choice`` are translated from the flat Responses shape to
the nested Chat Completions shape here so the existing #5099
``/v1/chat/completions`` client-side pass-through picks them up without
further modification.
"""
chat_kwargs: dict = dict(
model = payload.model,
messages = messages,
stream = stream,
@ -2136,7 +2374,46 @@ def _build_chat_request(
chat_kwargs["top_p"] = payload.top_p
if payload.max_output_tokens is not None:
chat_kwargs["max_tokens"] = payload.max_output_tokens
return ChatCompletionRequest(**chat_kwargs)
chat_tools = _translate_responses_tools_to_chat(payload.tools)
if chat_tools is not None:
chat_kwargs["tools"] = chat_tools
chat_tool_choice = _translate_responses_tool_choice_to_chat(payload.tool_choice)
if chat_tool_choice is not None:
chat_kwargs["tool_choice"] = chat_tool_choice
req = ChatCompletionRequest(**chat_kwargs)
# `parallel_tool_calls` is not a first-class field on ChatCompletionRequest,
# but the model allows extras and _build_openai_passthrough_body forwards
# only explicitly-known fields. Llama-server does not currently implement
# parallel_tool_calls semantics, so we accept-and-ignore it on the
# Responses side to avoid breaking SDK clients that always send it.
return req
def _chat_tool_calls_to_responses_output(tool_calls: list[dict]) -> list[dict]:
"""Map Chat Completions ``tool_calls`` into Responses ``function_call`` output items.
The Chat Completions id (``call_xxx``) is the shared correlation key across
turns in the OpenAI Responses API — it is stored as ``call_id`` on the
output item and must be echoed back by the client as
``function_call_output.call_id`` on the next turn.
"""
items: list[dict] = []
for tc in tool_calls:
if tc.get("type") != "function":
continue
fn = tc.get("function") or {}
items.append(
ResponsesOutputFunctionCall(
call_id = tc.get("id", ""),
name = fn.get("name", ""),
arguments = fn.get("arguments", "") or "",
status = "completed",
).model_dump()
)
return items
async def _responses_non_streaming(
@ -2156,35 +2433,44 @@ async def _responses_non_streaming(
else:
body = result
# Extract content and usage from the Chat Completions response
choices = body.get("choices", [])
text = ""
tool_calls: list[dict] = []
if choices:
msg = choices[0].get("message", {})
msg = choices[0].get("message", {}) or {}
text = msg.get("content", "") or ""
tool_calls = msg.get("tool_calls") or []
usage_data = body.get("usage", {})
input_tokens = usage_data.get("prompt_tokens", 0)
output_tokens = usage_data.get("completion_tokens", 0)
resp_id = f"resp_{uuid.uuid4().hex[:12]}"
msg_id = f"msg_{uuid.uuid4().hex[:12]}"
# Responses API emits each tool call as its own top-level output item,
# alongside an optional assistant text message. Emit the text message
# only when the model actually produced content, so clients that expect
# a pure tool-call turn (finish_reason="tool_calls") don't see a spurious
# empty message item.
output_items: list[dict] = []
if text:
msg_id = f"msg_{uuid.uuid4().hex[:12]}"
output_items.append(
ResponsesOutputMessage(
id = msg_id,
status = "completed",
role = "assistant",
content = [ResponsesOutputTextContent(text = text)],
).model_dump()
)
output_items.extend(_chat_tool_calls_to_responses_output(tool_calls))
response = ResponsesResponse(
id = resp_id,
created_at = int(time.time()),
status = "completed",
model = body.get("model", payload.model),
output = [
ResponsesOutputMessage(
id = msg_id,
status = "completed",
role = "assistant",
content = [
ResponsesOutputTextContent(text = text),
],
),
],
output = output_items,
usage = ResponsesUsage(
input_tokens = input_tokens,
output_tokens = output_tokens,
@ -2203,24 +2489,100 @@ async def _responses_stream(
messages: list[ChatMessage],
request: Request,
):
"""Handle a streaming Responses API call, emitting named SSE events."""
"""Handle a streaming Responses API call, emitting named SSE events.
For GGUF models the request goes directly to llama-server's
``/v1/chat/completions`` endpoint from inside the StreamingResponse
child task — a single httpx lifecycle, a single async generator.
Wrapping the existing ``openai_chat_completions`` pass-through (which
already does its own httpx lifecycle) stacks two generators: Python
3.13 + httpcore 1.0.x then loses the close-propagation chain on the
innermost ``HTTP11ConnectionByteStream`` at asyncgen finalisation,
tripping "Attempted to exit cancel scope in a different task" /
"async generator ignored GeneratorExit". The direct path avoids that
altogether. Non-GGUF falls back to the wrapper (which doesn't use
httpx, so the issue doesn't apply).
Text deltas arrive as ``response.output_text.delta`` on a single
``message`` output item at ``output_index=0``. Each tool call from
``delta.tool_calls[]`` is promoted to its own top-level ``function_call``
output item (one per distinct ``tool_calls[].index``), and relayed as
``response.function_call_arguments.delta`` / ``.done`` events so clients
(Codex, OpenAI Python SDK) can reconstruct the call incrementally and
reply with a ``function_call_output`` item on the next turn.
"""
resp_id = f"resp_{uuid.uuid4().hex[:12]}"
msg_id = f"msg_{uuid.uuid4().hex[:12]}"
item_id = f"item_{uuid.uuid4().hex[:12]}"
created_at = int(time.time())
chat_req = _build_chat_request(payload, messages, stream = True)
result = await openai_chat_completions(chat_req, request)
llama_backend = get_llama_cpp_backend()
if not llama_backend.is_loaded:
# The direct pass-through is GGUF-only. Non-GGUF /v1/responses
# streaming isn't a Codex-compatible path today and wrapping the
# transformers backend's streaming generator here would re-
# introduce the double-layer asyncgen close pattern that produces
# "Attempted to exit cancel scope in a different task" on Python
# 3.13. Surface a typed 400 so the client sees a useful error
# instead of a dangling stream.
raise HTTPException(
status_code = 400,
detail = (
"Streaming /v1/responses requires a GGUF model loaded via "
"llama-server. Use non-streaming /v1/responses, "
"/v1/chat/completions, or load a GGUF model."
),
)
body = _build_openai_passthrough_body(chat_req)
target_url = f"{llama_backend.base_url}/v1/chat/completions"
async def event_generator():
full_text = ""
input_tokens = 0
output_tokens = 0
# Per-tool-call state keyed by the Chat Completions `tool_calls[].index`
# which stays stable across chunks for the same call. Values are:
# {output_index, item_id, call_id, name, arguments, opened}
tool_call_state: dict[int, dict] = {}
# Text message lives at output_index 0; tool calls claim 1, 2, ...
next_output_index = 1
def _snapshot_output() -> list[dict]:
"""Snapshot of all completed output items for response.completed."""
items: list[dict] = [
{
"type": "message",
"id": msg_id,
"status": "completed",
"role": "assistant",
"content": [
{
"type": "output_text",
"text": full_text,
"annotations": [],
}
],
}
]
for st in sorted(tool_call_state.values(), key = lambda s: s["output_index"]):
items.append(
{
"type": "function_call",
"id": st["item_id"],
"status": "completed",
"call_id": st["call_id"],
"name": st["name"],
"arguments": st["arguments"],
}
)
return items
# ── Preamble events ──
yield f"event: response.created\ndata: {json.dumps({'type': 'response.created', 'response': {'id': resp_id, 'object': 'response', 'created_at': created_at, 'status': 'in_progress', 'model': payload.model, 'output': [], 'usage': {'input_tokens': 0, 'output_tokens': 0, 'total_tokens': 0}}})}\n\n"
# output_item.added
# output_item.added (text message at output_index 0)
output_item = {
"type": "message",
"id": msg_id,
@ -2234,62 +2596,215 @@ async def _responses_stream(
content_part = {"type": "output_text", "text": "", "annotations": []}
yield f"event: response.content_part.added\ndata: {json.dumps({'type': 'response.content_part.added', 'item_id': msg_id, 'output_index': 0, 'content_index': 0, 'part': content_part})}\n\n"
# ── Stream delta events from the inner chat completions stream ──
if isinstance(result, StreamingResponse):
async for raw_chunk in result.body_iterator:
if isinstance(raw_chunk, bytes):
raw_chunk = raw_chunk.decode("utf-8", errors = "replace")
# ── Direct httpx lifecycle to llama-server ──
# Full same-task open + close, identical pattern to
# _openai_passthrough_stream and _anthropic_passthrough_stream:
# no `async with`, explicit aclose of lines_iter BEFORE resp /
# client so the innermost httpcore byte stream is finalised in
# this task (not via Python's asyncgen GC in a sibling task).
client = httpx.AsyncClient(timeout = 600)
resp = None
lines_iter = None
try:
req = client.build_request("POST", target_url, json = body)
try:
resp = await client.send(req, stream = True)
except httpx.RequestError as e:
logger.error("responses stream: upstream unreachable: %s", e)
yield f"event: response.failed\ndata: {json.dumps({'type': 'response.failed', 'response': {'id': resp_id, 'object': 'response', 'created_at': created_at, 'status': 'failed', 'model': payload.model, 'output': [], 'error': {'code': 502, 'message': _friendly_error(e)}}})}\n\n"
return
for line in raw_chunk.split("\n"):
line = line.strip()
if not line.startswith("data: "):
continue
data_str = line[6:]
if data_str == "[DONE]":
continue
try:
chunk_data = json.loads(data_str)
except json.JSONDecodeError:
continue
if resp.status_code != 200:
err_bytes = await resp.aread()
err_text = err_bytes.decode("utf-8", errors = "replace")
logger.error(
"responses stream upstream error: status=%s body=%s",
resp.status_code,
err_text[:500],
)
yield f"event: response.failed\ndata: {json.dumps({'type': 'response.failed', 'response': {'id': resp_id, 'object': 'response', 'created_at': created_at, 'status': 'failed', 'model': payload.model, 'output': [], 'error': {'code': resp.status_code, 'message': f'llama-server error: {err_text[:500]}'}}})}\n\n"
return
choices = chunk_data.get("choices", [])
if not choices:
# Check for usage in final chunk
usage = chunk_data.get("usage")
if usage:
input_tokens = usage.get("prompt_tokens", input_tokens)
output_tokens = usage.get(
"completion_tokens", output_tokens
)
continue
lines_iter = resp.aiter_lines()
async for raw_line in lines_iter:
if await request.is_disconnected():
break
if not raw_line:
continue
if not raw_line.startswith("data: "):
continue
data_str = raw_line[6:]
if data_str.strip() == "[DONE]":
break
try:
chunk_data = json.loads(data_str)
except json.JSONDecodeError:
continue
delta = choices[0].get("delta", {})
content = delta.get("content")
if content:
full_text += content
delta_event = {
"type": "response.output_text.delta",
"item_id": msg_id,
"output_index": 0,
"content_index": 0,
"delta": content,
}
yield f"event: response.output_text.delta\ndata: {json.dumps(delta_event)}\n\n"
# Check for usage in chunk
choices = chunk_data.get("choices", [])
if not choices:
usage = chunk_data.get("usage")
if usage:
input_tokens = usage.get("prompt_tokens", input_tokens)
output_tokens = usage.get("completion_tokens", output_tokens)
continue
# ── Closing events ──
# output_text.done
delta = choices[0].get("delta", {}) or {}
content = delta.get("content")
if content:
full_text += content
delta_event = {
"type": "response.output_text.delta",
"item_id": msg_id,
"output_index": 0,
"content_index": 0,
"delta": content,
}
yield f"event: response.output_text.delta\ndata: {json.dumps(delta_event)}\n\n"
for tc in delta.get("tool_calls") or []:
idx = tc.get("index", 0)
st = tool_call_state.get(idx)
fn = tc.get("function") or {}
if st is None:
# First chunk for this tool call — allocate an
# output_index and emit output_item.added.
st = {
"output_index": next_output_index,
"item_id": f"fc_{uuid.uuid4().hex[:12]}",
"call_id": tc.get("id") or "",
"name": fn.get("name") or "",
"arguments": "",
"opened": False,
}
next_output_index += 1
tool_call_state[idx] = st
else:
# Later chunks sometimes carry the id/name only
# once; merge when present.
if tc.get("id") and not st["call_id"]:
st["call_id"] = tc["id"]
if fn.get("name") and not st["name"]:
st["name"] = fn["name"]
if not st["opened"] and st["call_id"] and st["name"]:
item_added = {
"type": "response.output_item.added",
"output_index": st["output_index"],
"item": {
"type": "function_call",
"id": st["item_id"],
"status": "in_progress",
"call_id": st["call_id"],
"name": st["name"],
"arguments": "",
},
}
yield f"event: response.output_item.added\ndata: {json.dumps(item_added)}\n\n"
st["opened"] = True
arg_delta = fn.get("arguments") or ""
if arg_delta and st["opened"]:
st["arguments"] += arg_delta
args_delta_event = {
"type": "response.function_call_arguments.delta",
"item_id": st["item_id"],
"output_index": st["output_index"],
"delta": arg_delta,
}
yield f"event: response.function_call_arguments.delta\ndata: {json.dumps(args_delta_event)}\n\n"
elif arg_delta:
# Buffer the args until we can open the item
# (id/name arrive in the same chunk as the first
# arg delta for some models — but if not, stash).
st["arguments"] += arg_delta
usage = chunk_data.get("usage")
if usage:
input_tokens = usage.get("prompt_tokens", input_tokens)
output_tokens = usage.get("completion_tokens", output_tokens)
except Exception as e:
logger.error("responses stream error: %s", e)
finally:
if lines_iter is not None:
try:
await lines_iter.aclose()
except Exception:
pass
if resp is not None:
try:
await resp.aclose()
except Exception:
pass
try:
await client.aclose()
except Exception:
pass
# ── Closing events for tool calls ──
for st in sorted(tool_call_state.values(), key = lambda s: s["output_index"]):
# If id/name never arrived (malformed upstream), synthesise so
# the client still sees a coherent frame sequence.
if not st["opened"]:
if not st["call_id"]:
st["call_id"] = f"call_{uuid.uuid4().hex[:12]}"
item_added = {
"type": "response.output_item.added",
"output_index": st["output_index"],
"item": {
"type": "function_call",
"id": st["item_id"],
"status": "in_progress",
"call_id": st["call_id"],
"name": st["name"],
"arguments": "",
},
}
yield f"event: response.output_item.added\ndata: {json.dumps(item_added)}\n\n"
if st["arguments"]:
yield (
"event: response.function_call_arguments.delta\n"
"data: "
+ json.dumps(
{
"type": "response.function_call_arguments.delta",
"item_id": st["item_id"],
"output_index": st["output_index"],
"delta": st["arguments"],
}
)
+ "\n\n"
)
st["opened"] = True
args_done = {
"type": "response.function_call_arguments.done",
"item_id": st["item_id"],
"output_index": st["output_index"],
"name": st["name"],
"arguments": st["arguments"],
}
yield f"event: response.function_call_arguments.done\ndata: {json.dumps(args_done)}\n\n"
item_done = {
"type": "response.output_item.done",
"output_index": st["output_index"],
"item": {
"type": "function_call",
"id": st["item_id"],
"status": "completed",
"call_id": st["call_id"],
"name": st["name"],
"arguments": st["arguments"],
},
}
yield f"event: response.output_item.done\ndata: {json.dumps(item_done)}\n\n"
# ── Closing events for text message ──
yield f"event: response.output_text.done\ndata: {json.dumps({'type': 'response.output_text.done', 'item_id': msg_id, 'output_index': 0, 'content_index': 0, 'text': full_text})}\n\n"
# content_part.done
yield f"event: response.content_part.done\ndata: {json.dumps({'type': 'response.content_part.done', 'item_id': msg_id, 'output_index': 0, 'content_index': 0, 'part': {'type': 'output_text', 'text': full_text, 'annotations': []}})}\n\n"
# output_item.done
yield f"event: response.output_item.done\ndata: {json.dumps({'type': 'response.output_item.done', 'output_index': 0, 'item': {'type': 'message', 'id': msg_id, 'status': 'completed', 'role': 'assistant', 'content': [{'type': 'output_text', 'text': full_text, 'annotations': []}]}})}\n\n"
# response.completed
@ -2302,21 +2817,7 @@ async def _responses_stream(
"created_at": created_at,
"status": "completed",
"model": payload.model,
"output": [
{
"type": "message",
"id": msg_id,
"status": "completed",
"role": "assistant",
"content": [
{
"type": "output_text",
"text": full_text,
"annotations": [],
}
],
}
],
"output": _snapshot_output(),
"usage": {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
@ -2365,6 +2866,64 @@ async def openai_responses(
# =====================================================================
def _normalize_anthropic_openai_images(
openai_messages: list[dict], is_vision: bool
) -> bool:
"""Enforce the vision guard on translated Anthropic messages and
normalize any ``image_url`` parts with base64 data URLs to PNG.
llama-server's stb_image only handles a few formats (JPEG/PNG/BMP/…);
Anthropic clients commonly send JPEG or WebP, and Claude Code sends
WebP. Re-encoding everything to PNG mirrors the behavior of
`_openai_messages_for_passthrough` / the GGUF branch of
`/v1/chat/completions` so the two endpoints agree.
Mutates ``openai_messages`` in place. Returns ``True`` when any
image part was seen (so the caller can skip a second scan). Raises
HTTPException(400) when images are present but the active model is
not a vision model, or when an image cannot be decoded.
"""
from PIL import Image
has_image = False
for msg in openai_messages:
content = msg.get("content")
if not isinstance(content, list):
continue
for part in content:
if part.get("type") != "image_url":
continue
has_image = True
if not is_vision:
raise HTTPException(
status_code = 400,
detail = "Image provided but current GGUF model does not support vision.",
)
url = (part.get("image_url") or {}).get("url", "")
if not url.startswith("data:"):
# Remote URLs are forwarded as-is; llama-server will
# fetch (or fail) per its own support matrix.
continue
try:
_, b64data = url.split(",", 1)
raw = base64.b64decode(b64data)
img = Image.open(io.BytesIO(raw)).convert("RGB")
buf = io.BytesIO()
img.save(buf, format = "PNG")
png_b64 = base64.b64encode(buf.getvalue()).decode("ascii")
except Exception as e:
raise HTTPException(
status_code = 400,
detail = f"Failed to process image: {e}",
)
part["image_url"] = {"url": f"data:image/png;base64,{png_b64}"}
return has_image
@router.post("/messages")
async def anthropic_messages(
payload: AnthropicMessagesRequest,
@ -2395,6 +2954,12 @@ async def anthropic_messages(
payload.system,
)
# Enforce vision guard + re-encode embedded images to PNG so the
# Anthropic endpoint matches the behavior of /v1/chat/completions.
_has_image = _normalize_anthropic_openai_images(
openai_messages, llama_backend.is_vision
)
temperature = payload.temperature if payload.temperature is not None else 0.6
top_p = payload.top_p if payload.top_p is not None else 0.95
top_k = payload.top_k if payload.top_k is not None else 20
@ -2421,7 +2986,11 @@ async def anthropic_messages(
# 1. enable_tools=true → server-side execution of built-in tools (Unsloth shorthand)
# 2. tools=[...] only → client-side pass-through (standard Anthropic behavior)
# 3. neither → plain chat
server_tools = payload.enable_tools and llama_backend.supports_tools
# Server-side agentic loop doesn't support multimodal input — matches
# the `not image_b64` gate in /v1/chat/completions.
server_tools = (
payload.enable_tools and llama_backend.supports_tools and not _has_image
)
client_tools = (
not server_tools
and payload.tools

239
studio/backend/tests/test_anthropic_messages.py
View file

@ -10,6 +10,8 @@ import sys
import os
import json
import pytest
_backend = os.path.join(os.path.dirname(__file__), "..")
sys.path.insert(0, _backend)
@ -32,6 +34,10 @@ from core.inference.anthropic_compat import (
AnthropicStreamEmitter,
AnthropicPassthroughEmitter,
)
from routes.inference import _normalize_anthropic_openai_images
from fastapi import HTTPException
import base64 as _b64
from io import BytesIO as _BytesIO
# =====================================================================
@ -246,6 +252,141 @@ class TestAnthropicMessagesToOpenAI:
result = anthropic_messages_to_openai(msgs)
assert result[0]["content"] == "Line 1 Line 2"
def test_image_base64_block_becomes_multimodal_part(self):
msgs = [
{
"role": "user",
"content": [
{"type": "text", "text": "What is this?"},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/jpeg",
"data": "AAAA",
},
},
],
}
]
result = anthropic_messages_to_openai(msgs)
assert len(result) == 1
assert result[0]["role"] == "user"
parts = result[0]["content"]
assert isinstance(parts, list)
assert parts[0] == {"type": "text", "text": "What is this?"}
assert parts[1]["type"] == "image_url"
assert parts[1]["image_url"]["url"] == "data:image/jpeg;base64,AAAA"
def test_image_url_block_forwarded_as_url(self):
msgs = [
{
"role": "user",
"content": [
{"type": "text", "text": "Describe it"},
{
"type": "image",
"source": {"type": "url", "url": "https://x/y.png"},
},
],
}
]
result = anthropic_messages_to_openai(msgs)
parts = result[0]["content"]
assert parts[1] == {
"type": "image_url",
"image_url": {"url": "https://x/y.png"},
}
def test_image_only_user_message_emits_no_text_part(self):
msgs = [
{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": "ZZ",
},
},
],
}
]
result = anthropic_messages_to_openai(msgs)
parts = result[0]["content"]
assert len(parts) == 1
assert parts[0]["type"] == "image_url"
def test_image_default_media_type_when_missing(self):
msgs = [
{
"role": "user",
"content": [
{
"type": "image",
"source": {"type": "base64", "data": "BB"},
},
],
}
]
result = anthropic_messages_to_openai(msgs)
parts = result[0]["content"]
assert parts[0]["image_url"]["url"].startswith("data:image/jpeg;base64,")
def test_image_text_order_preserved(self):
# [text1, image1, text2, image2] must not collapse to
# [text1+text2, image1, image2].
msgs = [
{
"role": "user",
"content": [
{"type": "text", "text": "before"},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": "AA",
},
},
{"type": "text", "text": "after"},
{
"type": "image",
"source": {"type": "url", "url": "https://x/y.png"},
},
],
}
]
result = anthropic_messages_to_openai(msgs)
parts = result[0]["content"]
assert [p["type"] for p in parts] == [
"text",
"image_url",
"text",
"image_url",
]
assert parts[0]["text"] == "before"
assert parts[2]["text"] == "after"
assert parts[1]["image_url"]["url"] == "data:image/png;base64,AA"
assert parts[3]["image_url"]["url"] == "https://x/y.png"
def test_malformed_image_block_is_skipped(self):
msgs = [
{
"role": "user",
"content": [
{"type": "text", "text": "Hi"},
{"type": "image", "source": {"type": "base64"}},
{"type": "image", "source": {"type": "url"}},
],
}
]
result = anthropic_messages_to_openai(msgs)
# No image parts emitted; message falls back to plain text.
assert result[0] == {"role": "user", "content": "Hi"}
# =====================================================================
# Tool translation tests
@ -772,3 +913,101 @@ class TestAnthropicPassthroughEmitter:
parsed = self._parse(events[1])
assert parsed["content_block"]["name"] == "Read"
assert parsed["content_block"]["id"] == "c2"
# =====================================================================
# Vision guard + PNG normalization (/v1/messages)
# =====================================================================
def _jpeg_data_url() -> str:
from PIL import Image
img = Image.new("RGB", (2, 2), (255, 0, 0))
buf = _BytesIO()
img.save(buf, format = "JPEG")
b64 = _b64.b64encode(buf.getvalue()).decode("ascii")
return f"data:image/jpeg;base64,{b64}"
class TestNormalizeAnthropicOpenAIImages:
def test_noop_when_no_images(self):
msgs = [{"role": "user", "content": "hi"}]
has_image = _normalize_anthropic_openai_images(msgs, is_vision = False)
assert has_image is False
assert msgs == [{"role": "user", "content": "hi"}]
def test_returns_true_when_image_present(self):
msgs = [
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url": _jpeg_data_url()}},
],
}
]
assert _normalize_anthropic_openai_images(msgs, is_vision = True) is True
def test_rejects_image_when_model_not_vision(self):
msgs = [
{
"role": "user",
"content": [
{"type": "text", "text": "?"},
{
"type": "image_url",
"image_url": {"url": _jpeg_data_url()},
},
],
}
]
with pytest.raises(HTTPException) as exc:
_normalize_anthropic_openai_images(msgs, is_vision = False)
assert exc.value.status_code == 400
def test_reencodes_jpeg_data_url_to_png(self):
original_url = _jpeg_data_url()
msgs = [
{
"role": "user",
"content": [
{"type": "text", "text": "?"},
{"type": "image_url", "image_url": {"url": original_url}},
],
}
]
_normalize_anthropic_openai_images(msgs, is_vision = True)
new_url = msgs[0]["content"][1]["image_url"]["url"]
assert new_url.startswith("data:image/png;base64,")
assert new_url != original_url
def test_remote_url_left_unchanged(self):
msgs = [
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {"url": "https://x.example/y.png"},
},
],
}
]
_normalize_anthropic_openai_images(msgs, is_vision = True)
assert msgs[0]["content"][0]["image_url"]["url"] == "https://x.example/y.png"
def test_bad_base64_raises_400(self):
msgs = [
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {"url": "data:image/jpeg;base64,!!!not-b64!!!"},
},
],
}
]
with pytest.raises(HTTPException) as exc:
_normalize_anthropic_openai_images(msgs, is_vision = True)
assert exc.value.status_code == 400

667
studio/backend/tests/test_responses_tool_passthrough.py Normal file
View file

@ -0,0 +1,667 @@
# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved.
"""
Tests for the OpenAI /v1/responses client-side function-calling pass-through.
Covers:
- ResponsesRequest accepts Responses-shape `tools`, `tool_choice`,
`parallel_tool_calls`, and the `function_call` / `function_call_output`
input items used for multi-turn tool loops.
- _translate_responses_tools_to_chat() converts the flat Responses tool
shape to the nested Chat Completions shape, drops non-function built-in
tools, and returns None for empty lists.
- _translate_responses_tool_choice_to_chat() passes string choices through
and converts {type:function,name:X} to Chat Completions' nested shape.
- _normalise_responses_input() maps function_call_output items to
role="tool" ChatMessages with tool_call_id, and function_call items to
assistant messages with tool_calls.
- _chat_tool_calls_to_responses_output() preserves call_id and drops
non-function tool calls.
- ResponsesOutputFunctionCall and ResponsesResponse round-trip tool-call
outputs without losing fields.
No running server or GPU required.
"""
import os
import sys
_backend = os.path.join(os.path.dirname(__file__), "..")
sys.path.insert(0, _backend)
import json
import pytest
from pydantic import ValidationError
from models.inference import (
ChatMessage,
ResponsesFunctionCallInputItem,
ResponsesFunctionCallOutputInputItem,
ResponsesFunctionTool,
ResponsesInputMessage,
ResponsesOutputFunctionCall,
ResponsesOutputMessage,
ResponsesOutputTextContent,
ResponsesOutputTextPart,
ResponsesRequest,
ResponsesResponse,
ResponsesUnknownContentPart,
ResponsesUnknownInputItem,
ResponsesUsage,
)
from routes.inference import (
_chat_tool_calls_to_responses_output,
_normalise_responses_input,
_translate_responses_tool_choice_to_chat,
_translate_responses_tools_to_chat,
)
# =====================================================================
# Request model — tools / tool_choice / parallel_tool_calls
# =====================================================================
class TestResponsesRequestTools:
def test_flat_function_tool_accepted(self):
req = ResponsesRequest(
input = "hi",
tools = [
{
"type": "function",
"name": "get_weather",
"description": "Get the weather for a city.",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
"strict": True,
}
],
)
assert req.tools is not None
assert req.tools[0]["name"] == "get_weather"
assert req.tools[0]["type"] == "function"
assert req.tools[0]["strict"] is True
def test_tool_choice_string_values(self):
for choice in ("auto", "required", "none"):
req = ResponsesRequest(input = "hi", tool_choice = choice)
assert req.tool_choice == choice
def test_tool_choice_forcing_object(self):
req = ResponsesRequest(
input = "hi",
tool_choice = {"type": "function", "name": "get_weather"},
)
assert req.tool_choice == {"type": "function", "name": "get_weather"}
def test_parallel_tool_calls(self):
req = ResponsesRequest(input = "hi", parallel_tool_calls = True)
assert req.parallel_tool_calls is True
def test_builtin_tool_type_passes_validation(self):
"""Non-function built-in tools (web_search, file_search, mcp, ...) must
not raise at request validation so SDKs that default to them don't
fail on Studio; they are filtered out during translation."""
req = ResponsesRequest(
input = "hi",
tools = [{"type": "web_search_preview"}],
)
assert req.tools == [{"type": "web_search_preview"}]
def test_function_tool_model_direct(self):
tool = ResponsesFunctionTool(
type = "function",
name = "send_email",
parameters = {"type": "object", "properties": {}},
)
assert tool.name == "send_email"
assert tool.description is None
def test_function_tool_rejects_other_type(self):
with pytest.raises(ValidationError):
ResponsesFunctionTool(type = "web_search", name = "x")
# =====================================================================
# Request model — function_call / function_call_output input items
# =====================================================================
class TestResponsesMultiTurnInput:
def test_function_call_input_item(self):
req = ResponsesRequest(
input = [
{"role": "user", "content": "Weather in Paris?"},
{
"type": "function_call",
"id": "fc_abc",
"call_id": "call_abc",
"name": "get_weather",
"arguments": '{"city": "Paris"}',
},
{
"type": "function_call_output",
"call_id": "call_abc",
"output": '{"temp": 12}',
},
],
)
assert len(req.input) == 3
assert isinstance(req.input[1], ResponsesFunctionCallInputItem)
assert req.input[1].call_id == "call_abc"
assert isinstance(req.input[2], ResponsesFunctionCallOutputInputItem)
assert req.input[2].call_id == "call_abc"
assert req.input[2].output == '{"temp": 12}'
def test_function_call_output_missing_call_id_rejected(self):
with pytest.raises(ValidationError):
ResponsesFunctionCallOutputInputItem(
type = "function_call_output", output = "x"
)
def test_function_call_output_accepts_content_array(self):
item = ResponsesFunctionCallOutputInputItem(
type = "function_call_output",
call_id = "call_1",
output = [{"type": "output_text", "text": "done"}],
)
assert isinstance(item.output, list)
# =====================================================================
# Translators — tools, tool_choice
# =====================================================================
class TestToolsTranslation:
def test_flat_to_nested(self):
tools = [
{
"type": "function",
"name": "get_weather",
"description": "Returns weather.",
"parameters": {"type": "object"},
"strict": True,
}
]
out = _translate_responses_tools_to_chat(tools)
assert out == [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Returns weather.",
"parameters": {"type": "object"},
"strict": True,
},
}
]
def test_builtin_tools_dropped(self):
out = _translate_responses_tools_to_chat(
[
{"type": "web_search_preview"},
{"type": "file_search"},
{
"type": "function",
"name": "search",
"parameters": {"type": "object"},
},
]
)
assert len(out) == 1
assert out[0]["function"]["name"] == "search"
def test_empty_returns_none(self):
assert _translate_responses_tools_to_chat(None) is None
assert _translate_responses_tools_to_chat([]) is None
def test_only_builtin_tools_returns_none(self):
assert (
_translate_responses_tools_to_chat([{"type": "web_search_preview"}]) is None
)
def test_description_optional(self):
out = _translate_responses_tools_to_chat(
[
{
"type": "function",
"name": "noop",
"parameters": {"type": "object"},
}
]
)
assert "description" not in out[0]["function"]
class TestToolChoiceTranslation:
def test_string_passthrough(self):
for v in ("auto", "required", "none"):
assert _translate_responses_tool_choice_to_chat(v) == v
def test_none_passthrough(self):
assert _translate_responses_tool_choice_to_chat(None) is None
def test_forcing_object_converted(self):
assert _translate_responses_tool_choice_to_chat(
{"type": "function", "name": "get_weather"}
) == {"type": "function", "function": {"name": "get_weather"}}
def test_already_chat_nested_shape_passes_through(self):
"""If a client happens to send the Chat Completions nested shape,
we don't double-wrap it."""
already_nested = {"type": "function", "function": {"name": "get_weather"}}
assert (
_translate_responses_tool_choice_to_chat(already_nested) == already_nested
)
def test_unknown_shape_passes_through(self):
obj = {"type": "allowed_tools", "tools": [{"type": "function", "name": "x"}]}
assert _translate_responses_tool_choice_to_chat(obj) == obj
# =====================================================================
# _normalise_responses_input — multi-turn tool mapping
# =====================================================================
class TestNormaliseResponsesInputWithTools:
def test_function_call_output_maps_to_tool_role(self):
payload = ResponsesRequest(
input = [
{"role": "user", "content": "Weather?"},
{
"type": "function_call",
"call_id": "call_1",
"name": "get_weather",
"arguments": "{}",
},
{
"type": "function_call_output",
"call_id": "call_1",
"output": '{"temp": 20}',
},
],
)
msgs = _normalise_responses_input(payload)
assert len(msgs) == 3
assert msgs[0].role == "user"
assert msgs[1].role == "assistant"
assert msgs[1].tool_calls is not None
assert msgs[1].tool_calls[0]["id"] == "call_1"
assert msgs[1].tool_calls[0]["function"]["name"] == "get_weather"
assert msgs[2].role == "tool"
assert msgs[2].tool_call_id == "call_1"
assert msgs[2].content == '{"temp": 20}'
def test_instructions_plus_developer_message_are_merged(self):
"""Codex CLI sends `instructions` (system prompt) AND a developer
message in `input`. Strict chat templates (harmony / gpt-oss, Qwen3,
...) raise "System message must be at the beginning" when two
separate system-role messages appear, so we must emit exactly one
merged system message at the top.
"""
payload = ResponsesRequest(
instructions = "Base instructions.",
input = [
{"role": "developer", "content": "Developer override."},
{"role": "user", "content": "Hi"},
],
)
msgs = _normalise_responses_input(payload)
system_roles = [m for m in msgs if m.role == "system"]
assert len(system_roles) == 1
assert "Base instructions." in system_roles[0].content
assert "Developer override." in system_roles[0].content
# System must be the very first message for strict templates.
assert msgs[0].role == "system"
assert msgs[1].role == "user"
def test_developer_message_after_user_is_still_hoisted(self):
"""Multi-turn conversations where a developer message appears after
user turns must still produce a single leading system message, not
a mid-conversation system that strict templates reject."""
payload = ResponsesRequest(
input = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi!"},
{"role": "developer", "content": "Updated rules."},
{"role": "user", "content": "Continue"},
],
)
msgs = _normalise_responses_input(payload)
assert msgs[0].role == "system"
assert "Updated rules." in msgs[0].content
for m in msgs[1:]:
assert m.role != "system", "no trailing system message permitted"
def test_no_system_output_when_no_system_input(self):
payload = ResponsesRequest(input = "Hi")
msgs = _normalise_responses_input(payload)
assert all(m.role != "system" for m in msgs)
def test_multiple_system_messages_in_input_are_merged(self):
payload = ResponsesRequest(
input = [
{"role": "system", "content": "A"},
{"role": "system", "content": "B"},
{"role": "user", "content": "Hi"},
],
)
msgs = _normalise_responses_input(payload)
assert sum(1 for m in msgs if m.role == "system") == 1
assert "A" in msgs[0].content and "B" in msgs[0].content
def test_content_array_output_serialised_to_json_string(self):
payload = ResponsesRequest(
input = [
{
"type": "function_call_output",
"call_id": "call_1",
"output": [{"type": "output_text", "text": "ok"}],
}
],
)
msgs = _normalise_responses_input(payload)
assert msgs[0].role == "tool"
# Content is serialised so llama-server sees a string.
assert json.loads(msgs[0].content) == [{"type": "output_text", "text": "ok"}]
# =====================================================================
# Response mapping — tool_calls → function_call output items
# =====================================================================
class TestChatToolCallsToResponsesOutput:
def test_basic_mapping(self):
items = _chat_tool_calls_to_responses_output(
[
{
"id": "call_abc",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city":"Paris"}',
},
}
]
)
assert len(items) == 1
assert items[0]["type"] == "function_call"
assert items[0]["call_id"] == "call_abc"
assert items[0]["name"] == "get_weather"
assert items[0]["arguments"] == '{"city":"Paris"}'
assert items[0]["status"] == "completed"
assert items[0]["id"].startswith("fc_")
def test_multiple_tool_calls_preserved(self):
items = _chat_tool_calls_to_responses_output(
[
{
"id": "call_1",
"type": "function",
"function": {"name": "a", "arguments": "{}"},
},
{
"id": "call_2",
"type": "function",
"function": {"name": "b", "arguments": "{}"},
},
]
)
assert [it["call_id"] for it in items] == ["call_1", "call_2"]
def test_non_function_tool_call_dropped(self):
items = _chat_tool_calls_to_responses_output([{"id": "x", "type": "retrieval"}])
assert items == []
def test_missing_arguments_coerced_to_empty_string(self):
items = _chat_tool_calls_to_responses_output(
[{"id": "call_1", "type": "function", "function": {"name": "x"}}]
)
assert items[0]["arguments"] == ""
# =====================================================================
# Response model — ResponsesOutputFunctionCall / mixed output
# =====================================================================
class TestResponsesOutputFunctionCall:
def test_direct_construction(self):
fc = ResponsesOutputFunctionCall(
call_id = "call_1",
name = "get_weather",
arguments = '{"city":"Paris"}',
)
d = fc.model_dump()
assert d["type"] == "function_call"
assert d["call_id"] == "call_1"
assert d["status"] == "completed"
assert d["id"].startswith("fc_")
def test_response_with_tool_call_output(self):
resp = ResponsesResponse(
model = "test",
output = [
ResponsesOutputFunctionCall(
call_id = "call_1",
name = "get_weather",
arguments = "{}",
)
],
usage = ResponsesUsage(input_tokens = 1, output_tokens = 1, total_tokens = 2),
)
d = json.loads(resp.model_dump_json())
assert d["output"][0]["type"] == "function_call"
assert d["output"][0]["call_id"] == "call_1"
def test_response_with_mixed_output(self):
resp = ResponsesResponse(
model = "test",
output = [
ResponsesOutputMessage(
content = [ResponsesOutputTextContent(text = "Calling...")],
),
ResponsesOutputFunctionCall(
call_id = "call_1",
name = "get_weather",
arguments = '{"city":"Paris"}',
),
],
)
d = resp.model_dump()
assert d["output"][0]["type"] == "message"
assert d["output"][1]["type"] == "function_call"
# =====================================================================
# Regression: ChatMessage validator still accepts mapped tool messages
# =====================================================================
class TestCodexStyleRequestShapes:
"""Regression tests for the request shapes OpenAI Codex CLI sends."""
def test_assistant_replay_output_text_accepted(self):
"""Codex replays prior assistant turns with `output_text` content.
Before, this triggered a 422 on every turn after the first."""
req = ResponsesRequest(
input = [
{"role": "user", "content": "Hi"},
{
"type": "message",
"role": "assistant",
"content": [
{
"type": "output_text",
"text": "Hello!",
"annotations": [],
"logprobs": [],
}
],
},
{"role": "user", "content": "Continue"},
],
)
assert len(req.input) == 3
parts = req.input[1].content
assert isinstance(parts, list)
assert isinstance(parts[0], ResponsesOutputTextPart)
assert parts[0].text == "Hello!"
def test_reasoning_item_accepted_as_unknown(self):
"""`reasoning` items replayed from prior o-series turns must not
fail validation — Codex preserves them in multi-turn."""
req = ResponsesRequest(
input = [
{"role": "user", "content": "Hi"},
{
"type": "reasoning",
"id": "rs_1",
"summary": [],
"encrypted_content": "opaque",
},
{"role": "assistant", "content": "Hello!"},
],
)
assert len(req.input) == 3
assert isinstance(req.input[1], ResponsesUnknownInputItem)
def test_unknown_content_part_type_accepted(self):
"""Unknown content-part types (e.g. future input_audio) validate as
ResponsesUnknownContentPart so the whole request doesn't 422."""
req = ResponsesRequest(
input = [
{
"role": "user",
"content": [
{"type": "input_text", "text": "See:"},
{"type": "input_audio", "audio": {"data": "..."}},
],
}
],
)
parts = req.input[0].content
assert isinstance(parts[1], ResponsesUnknownContentPart)
assert parts[1].type == "input_audio"
def test_codex_full_shape_roundtrip(self):
"""End-to-end: developer + user + assistant(output_text) +
function_call + function_call_output + reasoning in one request."""
payload = ResponsesRequest(
instructions = "Base instructions.",
input = [
{
"type": "message",
"role": "developer",
"content": [{"type": "input_text", "text": "Dev override."}],
},
{
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "Weather?"}],
},
{
"type": "reasoning",
"id": "rs_1",
"summary": [],
},
{
"type": "function_call",
"call_id": "call_1",
"name": "get_weather",
"arguments": "{}",
},
{
"type": "function_call_output",
"call_id": "call_1",
"output": '{"temp":20}',
},
{
"type": "message",
"role": "assistant",
"content": [
{
"type": "output_text",
"text": "It's 20°C.",
"annotations": [],
"logprobs": [],
}
],
},
{"role": "user", "content": "And tomorrow?"},
],
)
msgs = _normalise_responses_input(payload)
# Single leading merged system; no mid-conversation system.
assert msgs[0].role == "system"
assert sum(1 for m in msgs if m.role == "system") == 1
assert "Base instructions." in msgs[0].content
assert "Dev override." in msgs[0].content
roles = [m.role for m in msgs[1:]]
# Reasoning item is dropped. Order: user, assistant(tool_calls),
# tool, assistant(text), user.
assert roles == ["user", "assistant", "tool", "assistant", "user"]
assert msgs[2].tool_calls is not None
assert msgs[3].role == "tool"
assert msgs[3].tool_call_id == "call_1"
assert msgs[4].content == "It's 20°C."
def test_single_output_text_part_flattens_to_string(self):
"""ChatMessage assistant role prefers plain string content — tests
confirm we don't forward a single-part array that would otherwise
force legacy chat templates into multimodal handling."""
payload = ResponsesRequest(
input = [
{
"role": "assistant",
"content": [
{"type": "output_text", "text": "ok", "annotations": []}
],
},
{"role": "user", "content": "next"},
],
)
msgs = _normalise_responses_input(payload)
assert msgs[0].role == "assistant"
assert msgs[0].content == "ok"
class TestTranslatedMessagesValidate:
"""Verify that the messages produced by _normalise_responses_input
satisfy ChatMessage's role-shape validator so the downstream /v1/chat/
completions pass-through does not reject them."""
def test_round_trip_multi_turn(self):
payload = ResponsesRequest(
input = [
{"role": "user", "content": "Weather in Paris?"},
{
"type": "function_call",
"call_id": "call_1",
"name": "get_weather",
"arguments": '{"city": "Paris"}',
},
{
"type": "function_call_output",
"call_id": "call_1",
"output": '{"temp": 20}',
},
{"role": "user", "content": "Thanks!"},
],
)
msgs = _normalise_responses_input(payload)
for m in msgs:
# Constructing a fresh ChatMessage from the dump round-trips the
# role-shape validator — the key invariant for the passthrough.
ChatMessage(**m.model_dump(exclude_none = True))

1
studio/backend/utils/datasets/model_mappings.py
View file

@ -369,6 +369,7 @@ TEMPLATE_TO_MODEL_MAPPER = {
"unsloth/Qwen3.5-0.8B",
"unsloth/Qwen3.5-2B",
"unsloth/Qwen3.5-4B",
"unsloth/Qwen3.5-9B",
"unsloth/Qwen3.5-27B",
"unsloth/Qwen3.5-35B-A3B",
),