diff --git a/studio/backend/core/inference/llama_cpp.py b/studio/backend/core/inference/llama_cpp.py index 7fa117980b..3cc7514ff5 100644 --- a/studio/backend/core/inference/llama_cpp.py +++ b/studio/backend/core/inference/llama_cpp.py @@ -990,8 +990,10 @@ class LlamaCppBackend: # enable_thinking / reasoning_effort -- skip. if self._supports_reasoning and not self._reasoning_always_on: if self._reasoning_style == "reasoning_effort": - if reasoning_effort in ("low", "medium", "high"): + if reasoning_effort in ("none", "low", "medium", "high"): kwargs["reasoning_effort"] = reasoning_effort + elif reasoning_effort == "minimal": + kwargs["reasoning_effort"] = "low" elif enable_thinking is not None: kwargs["reasoning_effort"] = "high" if enable_thinking else "low" else: diff --git a/studio/backend/models/inference.py b/studio/backend/models/inference.py index 5a648bc999..5a414384d2 100644 --- a/studio/backend/models/inference.py +++ b/studio/backend/models/inference.py @@ -1331,6 +1331,23 @@ class ResponsesOutputMessage(BaseModel): content: list[ResponsesOutputTextContent] = Field(default_factory = list) +class ResponsesOutputReasoningContent(BaseModel): + """A reasoning text content block inside a reasoning output item.""" + + type: Literal["reasoning_text"] = "reasoning_text" + text: str + + +class ResponsesOutputReasoning(BaseModel): + """A top-level reasoning output item in the Responses API response.""" + + type: Literal["reasoning"] = "reasoning" + id: str = Field(default_factory = lambda: f"rs_{uuid.uuid4().hex[:12]}") + status: Literal["completed", "in_progress", "incomplete"] = "completed" + summary: list = Field(default_factory = list) + content: Optional[list[ResponsesOutputReasoningContent]] = None + + class ResponsesOutputFunctionCall(BaseModel): """A function-call output item in the Responses API response. @@ -1345,7 +1362,11 @@ class ResponsesOutputFunctionCall(BaseModel): status: Literal["completed", "in_progress", "incomplete"] = "completed" -ResponsesOutputItem = Union[ResponsesOutputMessage, ResponsesOutputFunctionCall] +ResponsesOutputItem = Union[ + ResponsesOutputMessage, + ResponsesOutputReasoning, + ResponsesOutputFunctionCall, +] class ResponsesUsage(BaseModel): diff --git a/studio/backend/routes/inference.py b/studio/backend/routes/inference.py index 7b17e450d1..21390d7a3e 100644 --- a/studio/backend/routes/inference.py +++ b/studio/backend/routes/inference.py @@ -682,6 +682,8 @@ from models.inference import ( ResponsesFunctionCallOutputInputItem, ResponsesOutputTextContent, ResponsesOutputMessage, + ResponsesOutputReasoning, + ResponsesOutputReasoningContent, ResponsesOutputFunctionCall, ResponsesUsage, ResponsesResponse, @@ -5074,6 +5076,149 @@ def _responses_tool_output_text(output: Union[str, list]) -> str: return "(no output)" +_RESPONSES_THINK_OPEN = "" +_RESPONSES_THINK_CLOSE = "" +_RESPONSES_REASONING_EFFORTS = {"none", "minimal", "low", "medium", "high", "max", "xhigh"} + + +def _coerce_responses_reasoning_text(value: Any) -> str: + if value is None: + return "" + if isinstance(value, str): + return value + if isinstance(value, list): + return "".join(_coerce_responses_reasoning_text(part) for part in value) + if isinstance(value, dict): + for key in ("text", "reasoning_text", "content"): + text = _coerce_responses_reasoning_text(value.get(key)) + if text: + return text + return "" + return json.dumps(value) + + +def _responses_marker_holdback(text: str, markers: tuple[str, ...]) -> int: + """Number of trailing chars to retain because they may start a marker.""" + for size in range(min(len(text), max(len(m) for m in markers) - 1), 0, -1): + suffix = text[-size:] + if any(marker.startswith(suffix) for marker in markers): + return size + return 0 + + +class _ResponsesReasoningExtractor: + """Split local markup into Responses reasoning and visible text.""" + + def __init__(self, *, parse_think_markers: bool = False) -> None: + self._buffer = "" + self._in_reasoning = False + self._parse_think_markers = parse_think_markers + + def feed( + self, + text: str = "", + reasoning_content: Any = None, + ) -> tuple[str, str]: + reasoning_parts: list[str] = [] + visible_parts: list[str] = [] + structured_reasoning = _coerce_responses_reasoning_text(reasoning_content) + if structured_reasoning: + reasoning_parts.append(structured_reasoning) + if text: + self._buffer += text + if not self._parse_think_markers: + visible_parts.append(self._buffer) + self._buffer = "" + return "".join(reasoning_parts), "".join(visible_parts) + + while self._buffer: + if self._in_reasoning: + close_idx = self._buffer.find(_RESPONSES_THINK_CLOSE) + if close_idx != -1: + reasoning_parts.append(self._buffer[:close_idx]) + self._buffer = self._buffer[close_idx + len(_RESPONSES_THINK_CLOSE) :] + self._in_reasoning = False + continue + keep = _responses_marker_holdback(self._buffer, (_RESPONSES_THINK_CLOSE,)) + if keep == len(self._buffer): + break + reasoning_parts.append(self._buffer[:-keep] if keep else self._buffer) + self._buffer = self._buffer[-keep:] if keep else "" + break + + open_idx = self._buffer.find(_RESPONSES_THINK_OPEN) + close_idx = self._buffer.find(_RESPONSES_THINK_CLOSE) + if close_idx != -1 and (open_idx == -1 or close_idx < open_idx): + visible_parts.append(self._buffer[:close_idx]) + self._buffer = self._buffer[close_idx + len(_RESPONSES_THINK_CLOSE) :] + continue + if open_idx != -1: + visible_parts.append(self._buffer[:open_idx]) + self._buffer = self._buffer[open_idx + len(_RESPONSES_THINK_OPEN) :] + self._in_reasoning = True + continue + + keep = _responses_marker_holdback( + self._buffer, + (_RESPONSES_THINK_OPEN, _RESPONSES_THINK_CLOSE), + ) + if keep == len(self._buffer): + break + visible_parts.append(self._buffer[:-keep] if keep else self._buffer) + self._buffer = self._buffer[-keep:] if keep else "" + break + + return "".join(reasoning_parts), "".join(visible_parts) + + def finish(self) -> tuple[str, str]: + if not self._buffer: + return "", "" + remaining = self._buffer + self._buffer = "" + if not self._parse_think_markers: + return "", remaining + if self._in_reasoning: + self._in_reasoning = False + return remaining, "" + return "", remaining.replace(_RESPONSES_THINK_CLOSE, "") + + +def _extract_responses_reasoning( + text: str = "", + reasoning_content: Any = None, + *, + parse_think_markers: bool = False, +) -> tuple[str, str]: + extractor = _ResponsesReasoningExtractor(parse_think_markers = parse_think_markers) + reasoning, visible = extractor.feed(text, reasoning_content) + final_reasoning, final_visible = extractor.finish() + return reasoning + final_reasoning, visible + final_visible + + +def _responses_should_parse_think_markers( + chat_req: ChatCompletionRequest, llama_backend: Any = None +) -> bool: + if llama_backend is not None and getattr(llama_backend, "is_loaded", False): + if getattr(llama_backend, "reasoning_always_on", False): + return True + if not getattr(llama_backend, "supports_reasoning", False): + return False + if chat_req.enable_thinking is True: + return True + return chat_req.enable_thinking is None and chat_req.reasoning_effort not in (None, "none") + + +def _responses_reasoning_output_item(reasoning_text: str, item_id: Optional[str] = None) -> dict: + kwargs: dict[str, Any] = { + "status": "completed", + "summary": [], + "content": [ResponsesOutputReasoningContent(text = reasoning_text)], + } + if item_id is not None: + kwargs["id"] = item_id + return ResponsesOutputReasoning(**kwargs).model_dump() + + def _normalise_responses_input(payload: ResponsesRequest) -> list[ChatMessage]: """Convert a ResponsesRequest's ``input`` into a Chat-format ``ChatMessage`` list. @@ -5232,6 +5377,33 @@ def _build_chat_request( if payload.parallel_tool_calls is not None: chat_kwargs["parallel_tool_calls"] = payload.parallel_tool_calls + # ``chat_template_kwargs`` (e.g. ``{"enable_thinking": true}``) arrives via + # the Responses extra-body: ResponsesRequest has ``extra="allow"``, so the + # OpenAI SDK's ``extra_body`` spread lands the dict in ``model_extra``. The + # downstream Chat Completions paths consume the typed ``enable_thinking`` + # field -- the non-streaming path lifts it in ``openai_chat_completions`` + # only when it is still ``None``, and the streaming pass-through reads + # ``payload.enable_thinking`` directly -- so lift it here, mirroring that + # handler, to cover both Responses paths. + explicit_enable_thinking = False + _extra = getattr(payload, "model_extra", None) + if isinstance(_extra, dict): + _tpl_kw = _extra.get("chat_template_kwargs") + if isinstance(_tpl_kw, dict) and "enable_thinking" in _tpl_kw: + chat_kwargs["enable_thinking"] = bool(_tpl_kw["enable_thinking"]) + explicit_enable_thinking = True + + if isinstance(payload.reasoning, dict): + effort = payload.reasoning.get("effort") + if isinstance(effort, str) and effort in _RESPONSES_REASONING_EFFORTS: + if not explicit_enable_thinking: + chat_kwargs["reasoning_effort"] = effort + chat_kwargs["enable_thinking"] = effort != "none" + elif chat_kwargs.get("enable_thinking") is False: + chat_kwargs["reasoning_effort"] = "none" + elif effort != "none": + chat_kwargs["reasoning_effort"] = effort + return ChatCompletionRequest(**chat_kwargs) @@ -5275,10 +5447,18 @@ async def _responses_non_streaming( choices = body.get("choices", []) text = "" + reasoning_text = "" tool_calls: list[dict] = [] if choices: msg = choices[0].get("message", {}) or {} - text = msg.get("content", "") or "" + raw_content = msg.get("content", "") or "" + raw_text = raw_content if isinstance(raw_content, str) else json.dumps(raw_content) + llama_backend = get_llama_cpp_backend() + reasoning_text, text = _extract_responses_reasoning( + raw_text, + msg.get("reasoning_content"), + parse_think_markers = _responses_should_parse_think_markers(chat_req, llama_backend), + ) tool_calls = msg.get("tool_calls") or [] usage_data = body.get("usage", {}) @@ -5292,6 +5472,10 @@ async def _responses_non_streaming( # the model produced content, so clients expecting a pure tool-call turn # (finish_reason="tool_calls") don't see a spurious empty message item. output_items: list[dict] = [] + if reasoning_text and not text and not tool_calls: + text = reasoning_text + if reasoning_text: + output_items.append(_responses_reasoning_output_item(reasoning_text)) if text: msg_id = f"msg_{uuid.uuid4().hex[:12]}" output_items.append( @@ -5339,16 +5523,15 @@ async def _responses_stream( avoids that. 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 + Output items are allocated as upstream deltas appear. Reasoning/text deltas + open top-level ``reasoning`` / ``message`` items; 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 + 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 next turn. """ resp_id = f"resp_{uuid.uuid4().hex[:12]}" - msg_id = f"msg_{uuid.uuid4().hex[:12]}" created_at = int(time.time()) chat_req = _build_chat_request(payload, messages, stream = True) @@ -5388,61 +5571,166 @@ async def _responses_stream( async def event_generator(): full_text = "" + full_reasoning = "" input_tokens = 0 output_tokens = 0 + extractor = _ResponsesReasoningExtractor( + parse_think_markers = _responses_should_parse_think_markers(chat_req, llama_backend) + ) + reasoning_state: dict[str, Any] = {"output_index": None, "item_id": None, "opened": False} + message_state: dict[str, Any] = {"output_index": None, "item_id": None, "opened": False} # Per-tool-call state keyed by Chat Completions `tool_calls[].index`, # stable across chunks for the same call. Values: # {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 + next_output_index = 0 + + def _sse(event_name: str, payload: dict) -> str: + return f"event: {event_name}\ndata: {json.dumps(payload)}\n\n" + + def _claim_output_index() -> int: + nonlocal next_output_index + output_index = next_output_index + next_output_index += 1 + return output_index + + def _ensure_reasoning_open() -> list[str]: + if reasoning_state["opened"]: + return [] + reasoning_state["output_index"] = _claim_output_index() + reasoning_state["item_id"] = f"rs_{uuid.uuid4().hex[:12]}" + reasoning_state["opened"] = True + output_index = reasoning_state["output_index"] + item_id = reasoning_state["item_id"] + return [ + _sse( + "response.output_item.added", + { + "type": "response.output_item.added", + "output_index": output_index, + "item": { + "type": "reasoning", + "id": item_id, + "status": "in_progress", + "summary": [], + "content": [], + }, + }, + ), + _sse( + "response.content_part.added", + { + "type": "response.content_part.added", + "item_id": item_id, + "output_index": output_index, + "content_index": 0, + "part": {"type": "reasoning_text", "text": ""}, + }, + ), + ] + + def _ensure_message_open() -> list[str]: + if message_state["opened"]: + return [] + message_state["output_index"] = _claim_output_index() + message_state["item_id"] = f"msg_{uuid.uuid4().hex[:12]}" + message_state["opened"] = True + output_index = message_state["output_index"] + item_id = message_state["item_id"] + return [ + _sse( + "response.output_item.added", + { + "type": "response.output_item.added", + "output_index": output_index, + "item": { + "type": "message", + "id": item_id, + "status": "in_progress", + "role": "assistant", + "content": [], + }, + }, + ), + _sse( + "response.content_part.added", + { + "type": "response.content_part.added", + "item_id": item_id, + "output_index": output_index, + "content_index": 0, + "part": {"type": "output_text", "text": "", "annotations": []}, + }, + ), + ] 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": [ + indexed_items: list[tuple[int, dict]] = [] + if reasoning_state["opened"]: + indexed_items.append( + ( + reasoning_state["output_index"], { - "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"], - } + "type": "reasoning", + "id": reasoning_state["item_id"], + "status": "completed", + "summary": [], + "content": [{"type": "reasoning_text", "text": full_reasoning}], + }, + ) ) - return items + if message_state["opened"]: + indexed_items.append( + ( + message_state["output_index"], + { + "type": "message", + "id": message_state["item_id"], + "status": "completed", + "role": "assistant", + "content": [ + { + "type": "output_text", + "text": full_text, + "annotations": [], + } + ], + }, + ) + ) + for st in tool_call_state.values(): + indexed_items.append( + ( + st["output_index"], + { + "type": "function_call", + "id": st["item_id"], + "status": "completed", + "call_id": st["call_id"], + "name": st["name"], + "arguments": st["arguments"], + }, + ) + ) + return [item for _, item in sorted(indexed_items, key = lambda pair: pair[0])] # ── 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 (text message at output_index 0) - output_item = { - "type": "message", - "id": msg_id, - "status": "in_progress", - "role": "assistant", - "content": [], - } - yield f"event: response.output_item.added\ndata: {json.dumps({'type': 'response.output_item.added', 'output_index': 0, 'item': output_item})}\n\n" - - # content_part.added - 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" + yield _sse( + "response.created", + { + "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}, + }, + }, + ) # ── Direct httpx lifecycle to llama-server ── # Full same-task open + close, same pattern as @@ -5459,7 +5747,21 @@ async def _responses_stream( 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" + yield _sse( + "response.failed", + { + "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)}, + }, + }, + ) return if resp.status_code != 200: @@ -5470,7 +5772,24 @@ async def _responses_stream( 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" + yield _sse( + "response.failed", + { + "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]}", + }, + }, + }, + ) return lines_iter = resp.aiter_lines() @@ -5500,17 +5819,38 @@ async def _responses_stream( continue 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" + reasoning_delta, visible_delta = extractor.feed( + delta.get("content") or "", + delta.get("reasoning_content"), + ) + if reasoning_delta: + for event in _ensure_reasoning_open(): + yield event + full_reasoning += reasoning_delta + yield _sse( + "response.reasoning_text.delta", + { + "type": "response.reasoning_text.delta", + "item_id": reasoning_state["item_id"], + "output_index": reasoning_state["output_index"], + "content_index": 0, + "delta": reasoning_delta, + }, + ) + if visible_delta: + for event in _ensure_message_open(): + yield event + full_text += visible_delta + yield _sse( + "response.output_text.delta", + { + "type": "response.output_text.delta", + "item_id": message_state["item_id"], + "output_index": message_state["output_index"], + "content_index": 0, + "delta": visible_delta, + }, + ) for tc in delta.get("tool_calls") or []: idx = tc.get("index", 0) @@ -5520,14 +5860,13 @@ async def _responses_stream( # First chunk for this tool call -- allocate an # output_index and emit output_item.added. st = { - "output_index": next_output_index, + "output_index": _claim_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 id/name only once; merge @@ -5550,7 +5889,7 @@ async def _responses_stream( "arguments": "", }, } - yield f"event: response.output_item.added\ndata: {json.dumps(item_added)}\n\n" + yield _sse("response.output_item.added", item_added) st["opened"] = True arg_delta = fn.get("arguments") or "" @@ -5562,7 +5901,7 @@ async def _responses_stream( "output_index": st["output_index"], "delta": arg_delta, } - yield f"event: response.function_call_arguments.delta\ndata: {json.dumps(args_delta_event)}\n\n" + yield _sse("response.function_call_arguments.delta", args_delta_event) elif arg_delta: # Buffer args until we can open the item (some models # send id/name in the same chunk as the first arg delta; @@ -5591,8 +5930,134 @@ async def _responses_stream( except Exception: pass - # ── Closing events for tool calls ── - for st in sorted(tool_call_state.values(), key = lambda s: s["output_index"]): + final_reasoning, final_visible = extractor.finish() + if final_reasoning: + for event in _ensure_reasoning_open(): + yield event + full_reasoning += final_reasoning + yield _sse( + "response.reasoning_text.delta", + { + "type": "response.reasoning_text.delta", + "item_id": reasoning_state["item_id"], + "output_index": reasoning_state["output_index"], + "content_index": 0, + "delta": final_reasoning, + }, + ) + if final_visible: + for event in _ensure_message_open(): + yield event + full_text += final_visible + yield _sse( + "response.output_text.delta", + { + "type": "response.output_text.delta", + "item_id": message_state["item_id"], + "output_index": message_state["output_index"], + "content_index": 0, + "delta": final_visible, + }, + ) + if full_reasoning and not full_text and not tool_call_state: + for event in _ensure_message_open(): + yield event + full_text = full_reasoning + yield _sse( + "response.output_text.delta", + { + "type": "response.output_text.delta", + "item_id": message_state["item_id"], + "output_index": message_state["output_index"], + "content_index": 0, + "delta": full_text, + }, + ) + + close_items: list[tuple[int, str, dict[str, Any]]] = [] + if reasoning_state["opened"]: + close_items.append((reasoning_state["output_index"], "reasoning", reasoning_state)) + if message_state["opened"]: + close_items.append((message_state["output_index"], "message", message_state)) + close_items.extend((st["output_index"], "tool", st) for st in tool_call_state.values()) + + for _, kind, st in sorted(close_items, key = lambda item: item[0]): + if kind == "reasoning": + yield _sse( + "response.reasoning_text.done", + { + "type": "response.reasoning_text.done", + "item_id": st["item_id"], + "output_index": st["output_index"], + "content_index": 0, + "text": full_reasoning, + }, + ) + yield _sse( + "response.content_part.done", + { + "type": "response.content_part.done", + "item_id": st["item_id"], + "output_index": st["output_index"], + "content_index": 0, + "part": {"type": "reasoning_text", "text": full_reasoning}, + }, + ) + yield _sse( + "response.output_item.done", + { + "type": "response.output_item.done", + "output_index": st["output_index"], + "item": { + "type": "reasoning", + "id": st["item_id"], + "status": "completed", + "summary": [], + "content": [{"type": "reasoning_text", "text": full_reasoning}], + }, + }, + ) + continue + + if kind == "message": + yield _sse( + "response.output_text.done", + { + "type": "response.output_text.done", + "item_id": st["item_id"], + "output_index": st["output_index"], + "content_index": 0, + "text": full_text, + }, + ) + yield _sse( + "response.content_part.done", + { + "type": "response.content_part.done", + "item_id": st["item_id"], + "output_index": st["output_index"], + "content_index": 0, + "part": {"type": "output_text", "text": full_text, "annotations": []}, + }, + ) + yield _sse( + "response.output_item.done", + { + "type": "response.output_item.done", + "output_index": st["output_index"], + "item": { + "type": "message", + "id": st["item_id"], + "status": "completed", + "role": "assistant", + "content": [ + {"type": "output_text", "text": full_text, "annotations": []} + ], + }, + }, + ) + continue + # If id/name never arrived (malformed upstream), synthesise so the # client still sees a coherent frame sequence. if not st["opened"]: @@ -5610,20 +6075,16 @@ async def _responses_stream( "arguments": "", }, } - yield f"event: response.output_item.added\ndata: {json.dumps(item_added)}\n\n" + yield _sse("response.output_item.added", item_added) 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" + yield _sse( + "response.function_call_arguments.delta", + { + "type": "response.function_call_arguments.delta", + "item_id": st["item_id"], + "output_index": st["output_index"], + "delta": st["arguments"], + }, ) st["opened"] = True @@ -5634,7 +6095,7 @@ async def _responses_stream( "name": st["name"], "arguments": st["arguments"], } - yield f"event: response.function_call_arguments.done\ndata: {json.dumps(args_done)}\n\n" + yield _sse("response.function_call_arguments.done", args_done) item_done = { "type": "response.output_item.done", @@ -5648,14 +6109,7 @@ async def _responses_stream( "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" - - 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" - - 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" + yield _sse("response.output_item.done", item_done) # response.completed total_tokens = input_tokens + output_tokens @@ -5675,7 +6129,7 @@ async def _responses_stream( }, }, } - yield f"event: response.completed\ndata: {json.dumps(completed_response)}\n\n" + yield _sse("response.completed", completed_response) return StreamingResponse( event_generator(), diff --git a/studio/backend/tests/test_openai_tool_passthrough.py b/studio/backend/tests/test_openai_tool_passthrough.py index 79e8c9ae49..baa3e50b95 100644 --- a/studio/backend/tests/test_openai_tool_passthrough.py +++ b/studio/backend/tests/test_openai_tool_passthrough.py @@ -28,11 +28,13 @@ from models.inference import ( ChatMessage, CompletionChoice, CompletionMessage, + ResponsesRequest, ) from core.inference.anthropic_compat import ( anthropic_tool_choice_to_openai, ) from routes.inference import ( + _build_chat_request, _build_openai_passthrough_body, _build_passthrough_payload, _clamp_finish_reason, @@ -786,6 +788,22 @@ class TestPassthroughReasoningKwargs: ) assert body["chat_template_kwargs"] == {"reasoning_effort": "high"} + def test_reasoning_effort_none_forwarded_for_effort_style_models(self): + body = _build_openai_passthrough_body( + self._payload(enable_thinking = False, reasoning_effort = "none"), + backend_ctx = 4096, + llama_backend = _reasoning_backend(reasoning_style = "reasoning_effort"), + ) + assert body["chat_template_kwargs"] == {"reasoning_effort": "none"} + + def test_reasoning_effort_minimal_maps_to_low_for_effort_style_models(self): + body = _build_openai_passthrough_body( + self._payload(enable_thinking = True, reasoning_effort = "minimal"), + backend_ctx = 4096, + llama_backend = _reasoning_backend(reasoning_style = "reasoning_effort"), + ) + assert body["chat_template_kwargs"] == {"reasoning_effort": "low"} + def test_enable_thinking_maps_to_effort_for_effort_style_models(self): body = _build_openai_passthrough_body( self._payload(enable_thinking = False), @@ -1396,3 +1414,47 @@ class TestGgufVisionToolRouting: assert seen_seeds == expected assert [choice["index"] for choice in body["choices"]] == [0, 1, 2] + + +# ===================================================================== +# Responses API -> Chat Completions translation: chat_template_kwargs +# (e.g. {"enable_thinking": true}) sent via the Responses extra-body must +# reach the built ChatCompletionRequest's typed ``enable_thinking`` field, +# otherwise /v1/responses silently ignores reasoning control (issue #6198). +# ===================================================================== + + +class TestResponsesChatTemplateKwargs: + _messages = [ChatMessage(role = "user", content = "What is 100 - 67?")] + + def test_enable_thinking_lifted_from_extra_body(self): + payload = ResponsesRequest( + model = "qwen-local", + input = "What is 100 - 67?", + chat_template_kwargs = {"enable_thinking": True}, + ) + chat_req = _build_chat_request(payload, self._messages, stream = False) + assert chat_req.enable_thinking is True + + def test_enable_thinking_false_lifted_from_extra_body(self): + payload = ResponsesRequest( + model = "qwen-local", + input = "hi", + chat_template_kwargs = {"enable_thinking": False}, + ) + chat_req = _build_chat_request(payload, self._messages, stream = True) + assert chat_req.enable_thinking is False + + def test_no_chat_template_kwargs_leaves_enable_thinking_unset(self): + payload = ResponsesRequest(model = "qwen-local", input = "hi") + chat_req = _build_chat_request(payload, self._messages, stream = False) + assert chat_req.enable_thinking is None + + def test_chat_template_kwargs_without_enable_thinking_is_ignored(self): + payload = ResponsesRequest( + model = "qwen-local", + input = "hi", + chat_template_kwargs = {"some_other_flag": True}, + ) + chat_req = _build_chat_request(payload, self._messages, stream = False) + assert chat_req.enable_thinking is None diff --git a/studio/backend/tests/test_responses_tool_passthrough.py b/studio/backend/tests/test_responses_tool_passthrough.py index 69fb0a78c2..ae7ff729bd 100644 --- a/studio/backend/tests/test_responses_tool_passthrough.py +++ b/studio/backend/tests/test_responses_tool_passthrough.py @@ -36,6 +36,7 @@ import json import httpx import pytest +from fastapi.responses import JSONResponse from pydantic import ValidationError from models.inference import ( @@ -46,6 +47,7 @@ from models.inference import ( ResponsesInputMessage, ResponsesOutputFunctionCall, ResponsesOutputMessage, + ResponsesOutputReasoning, ResponsesOutputTextContent, ResponsesOutputTextPart, ResponsesRequest, @@ -59,6 +61,7 @@ from routes.inference import ( _chat_tool_calls_to_responses_output, _normalise_responses_input, _responses_tool_output_text, + _responses_non_streaming, _responses_stream, _translate_responses_tool_choice_to_chat, _translate_responses_tools_to_chat, @@ -284,6 +287,59 @@ class TestBuildChatRequest: assert chat_req.parallel_tool_calls is False + def test_chat_template_kwargs_enable_thinking_true_is_lifted(self): + payload = ResponsesRequest( + input = "hi", + chat_template_kwargs = {"enable_thinking": True}, + ) + messages = [ChatMessage(role = "user", content = "hi")] + + chat_req = _build_chat_request(payload, messages, stream = False) + + assert chat_req.enable_thinking is True + + def test_chat_template_kwargs_enable_thinking_false_is_lifted(self): + payload = ResponsesRequest( + input = "hi", + chat_template_kwargs = {"enable_thinking": False}, + ) + messages = [ChatMessage(role = "user", content = "hi")] + + chat_req = _build_chat_request(payload, messages, stream = False) + + assert chat_req.enable_thinking is False + + def test_reasoning_effort_high_enables_local_thinking(self): + payload = ResponsesRequest(input = "hi", reasoning = {"effort": "high"}) + messages = [ChatMessage(role = "user", content = "hi")] + + chat_req = _build_chat_request(payload, messages, stream = False) + + assert chat_req.reasoning_effort == "high" + assert chat_req.enable_thinking is True + + def test_reasoning_effort_none_disables_local_thinking(self): + payload = ResponsesRequest(input = "hi", reasoning = {"effort": "none"}) + messages = [ChatMessage(role = "user", content = "hi")] + + chat_req = _build_chat_request(payload, messages, stream = False) + + assert chat_req.reasoning_effort == "none" + assert chat_req.enable_thinking is False + + def test_explicit_enable_thinking_false_disables_reasoning_effort(self): + payload = ResponsesRequest( + input = "hi", + reasoning = {"effort": "high"}, + chat_template_kwargs = {"enable_thinking": False}, + ) + messages = [ChatMessage(role = "user", content = "hi")] + + chat_req = _build_chat_request(payload, messages, stream = False) + + assert chat_req.reasoning_effort == "none" + assert chat_req.enable_thinking is False + # ===================================================================== # _normalise_responses_input — multi-turn tool mapping @@ -544,6 +600,119 @@ class TestChatToolCallsToResponsesOutput: assert items[0]["arguments"] == "" +# ===================================================================== +# Non-streaming Responses adapter +# ===================================================================== + + +class TestResponsesNonStreamingAdapter: + class _Request: + pass + + @staticmethod + def _run_with_message( + monkeypatch, + message, + payload = None, + llama_backend = None, + ): + import routes.inference as inf_mod + + async def fake_chat_completions(chat_req, request): + return JSONResponse( + content = { + "model": "test-model", + "choices": [{"message": message}], + "usage": {"prompt_tokens": 2, "completion_tokens": 3}, + } + ) + + monkeypatch.setattr(inf_mod, "openai_chat_completions", fake_chat_completions) + if llama_backend is not None: + monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: llama_backend) + payload = payload or ResponsesRequest(input = "hi") + messages = [ChatMessage(role = "user", content = "hi")] + + async def run(): + response = await _responses_non_streaming( + payload, messages, TestResponsesNonStreamingAdapter._Request() + ) + return json.loads(response.body.decode()) + + return asyncio.run(run()) + + def test_think_block_becomes_reasoning_item_before_message(self, monkeypatch): + payload = ResponsesRequest(input = "hi", reasoning = {"effort": "high"}) + body = self._run_with_message( + monkeypatch, + {"content": "plan33"}, + payload = payload, + ) + + assert [item["type"] for item in body["output"]] == ["reasoning", "message"] + assert body["output"][0]["content"] == [{"type": "reasoning_text", "text": "plan"}] + assert body["output"][0]["summary"] == [] + assert body["output"][1]["content"][0]["text"] == "33" + assert "" not in body["output"][1]["content"][0]["text"] + assert "" not in body["output"][1]["content"][0]["text"] + + def test_literal_think_tags_remain_visible_without_reasoning_request(self, monkeypatch): + body = self._run_with_message(monkeypatch, {"content": "show x tags"}) + + assert [item["type"] for item in body["output"]] == ["message"] + assert body["output"][0]["content"][0]["text"] == "show x tags" + + def test_non_reasoning_gguf_keeps_literal_think_tags_visible(self, monkeypatch): + payload = ResponsesRequest(input = "hi", reasoning = {"effort": "high"}) + body = self._run_with_message( + monkeypatch, + {"content": "show x tags"}, + payload = payload, + llama_backend = SimpleNamespace( + is_loaded = True, + reasoning_always_on = False, + supports_reasoning = False, + ), + ) + + assert [item["type"] for item in body["output"]] == ["message"] + assert body["output"][0]["content"][0]["text"] == "show x tags" + + def test_structured_reasoning_content_extracts_text_parts(self, monkeypatch): + body = self._run_with_message( + monkeypatch, + { + "content": "33", + "reasoning_content": [ + {"type": "reasoning_text", "text": "plan"}, + {"type": "reasoning_text", "text": " next"}, + ], + }, + ) + + assert [item["type"] for item in body["output"]] == ["reasoning", "message"] + assert body["output"][0]["content"] == [{"type": "reasoning_text", "text": "plan next"}] + assert body["output"][1]["content"][0]["text"] == "33" + + def test_plain_content_remains_message_only(self, monkeypatch): + body = self._run_with_message(monkeypatch, {"content": "33"}) + + assert [item["type"] for item in body["output"]] == ["message"] + assert body["output"][0]["content"][0]["text"] == "33" + + def test_reasoning_only_is_also_visible_message_text(self, monkeypatch): + payload = ResponsesRequest(input = "hi", reasoning = {"effort": "high"}) + body = self._run_with_message( + monkeypatch, + {"content": "plan"}, + payload = payload, + ) + + assert [item["type"] for item in body["output"]] == ["reasoning", "message"] + assert body["output"][0]["content"][0]["text"] == "plan" + assert body["output"][1]["content"][0]["text"] == "plan" + + # ===================================================================== # Streaming Responses adapter # ===================================================================== @@ -570,6 +739,262 @@ class TestResponsesStreamAdapter: if line.startswith(prefix) ] + @staticmethod + def _install_stream_mock( + monkeypatch, + chunks, + *, + supports_reasoning = True, + reasoning_always_on = False, + ): + import routes.inference as inf_mod + + def handler(request: httpx.Request) -> httpx.Response: + content = "".join(f"data: {json.dumps(chunk)}\n\n" for chunk in chunks) + content += "data: [DONE]\n\n" + return httpx.Response( + 200, + content = content.encode(), + headers = {"content-type": "text/event-stream"}, + ) + + transport = httpx.MockTransport(handler) + real_async_client = httpx.AsyncClient + + def _client(*args, **kwargs): + return real_async_client( + transport = transport, + timeout = kwargs.get("timeout", 600), + ) + + monkeypatch.setattr(inf_mod.httpx, "AsyncClient", _client) + monkeypatch.setattr( + inf_mod, + "get_llama_cpp_backend", + lambda: SimpleNamespace( + is_loaded = True, + is_vision = False, + context_length = 4096, + base_url = "http://llama.test", + supports_reasoning = supports_reasoning, + reasoning_always_on = reasoning_always_on, + _request_reasoning_kwargs = ( + lambda enable_thinking = None, reasoning_effort = None, preserve_thinking = None: None + ), + ), + ) + + def test_split_think_markers_stream_as_reasoning_and_visible_text(self, monkeypatch): + chunks = [ + {"choices": [{"delta": {"content": "pla"}}]}, + {"choices": [{"delta": {"content": "n33"}}]}, + {"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}}, + ] + self._install_stream_mock(monkeypatch, chunks) + payload = ResponsesRequest(input = "hi", stream = True, reasoning = {"effort": "high"}) + messages = [ChatMessage(role = "user", content = "hi")] + + async def run(): + response = await _responses_stream(payload, messages, self._Request()) + return await self._collect(response) + + lines = asyncio.run(run()) + + reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta") + text_deltas = self._payloads(lines, "response.output_text.delta") + assert "".join(event["delta"] for event in reasoning_deltas) == "plan" + assert "".join(event["delta"] for event in text_deltas) == "33" + completed = self._payloads(lines, "response.completed")[0] + assert [item["type"] for item in completed["response"]["output"]] == [ + "reasoning", + "message", + ] + assert completed["response"]["output"][0]["content"][0]["text"] == "plan" + assert completed["response"]["output"][1]["content"][0]["text"] == "33" + + def test_literal_think_tags_stream_as_visible_text_without_reasoning_request(self, monkeypatch): + chunks = [ + {"choices": [{"delta": {"content": "show x tags"}}]}, + {"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}}, + ] + self._install_stream_mock(monkeypatch, chunks) + payload = ResponsesRequest(input = "hi", stream = True) + messages = [ChatMessage(role = "user", content = "hi")] + + async def run(): + response = await _responses_stream(payload, messages, self._Request()) + return await self._collect(response) + + lines = asyncio.run(run()) + + reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta") + text_deltas = self._payloads(lines, "response.output_text.delta") + assert reasoning_deltas == [] + assert "".join(event["delta"] for event in text_deltas) == "show x tags" + completed = self._payloads(lines, "response.completed")[0] + assert [item["type"] for item in completed["response"]["output"]] == ["message"] + assert completed["response"]["output"][0]["content"][0]["text"] == ( + "show x tags" + ) + + def test_non_reasoning_gguf_stream_keeps_literal_think_tags_visible(self, monkeypatch): + chunks = [ + {"choices": [{"delta": {"content": "show x tags"}}]}, + {"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}}, + ] + self._install_stream_mock(monkeypatch, chunks, supports_reasoning = False) + payload = ResponsesRequest(input = "hi", stream = True, reasoning = {"effort": "high"}) + messages = [ChatMessage(role = "user", content = "hi")] + + async def run(): + response = await _responses_stream(payload, messages, self._Request()) + return await self._collect(response) + + lines = asyncio.run(run()) + + reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta") + text_deltas = self._payloads(lines, "response.output_text.delta") + assert reasoning_deltas == [] + assert "".join(event["delta"] for event in text_deltas) == "show x tags" + completed = self._payloads(lines, "response.completed")[0] + assert [item["type"] for item in completed["response"]["output"]] == ["message"] + assert completed["response"]["output"][0]["content"][0]["text"] == ( + "show x tags" + ) + + def test_reasoning_only_streams_as_visible_message_text(self, monkeypatch): + chunks = [ + {"choices": [{"delta": {"content": "plan"}}]}, + {"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}}, + ] + self._install_stream_mock(monkeypatch, chunks) + payload = ResponsesRequest(input = "hi", stream = True, reasoning = {"effort": "high"}) + messages = [ChatMessage(role = "user", content = "hi")] + + async def run(): + response = await _responses_stream(payload, messages, self._Request()) + return await self._collect(response) + + lines = asyncio.run(run()) + + reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta") + text_deltas = self._payloads(lines, "response.output_text.delta") + assert "".join(event["delta"] for event in reasoning_deltas) == "plan" + assert "".join(event["delta"] for event in text_deltas) == "plan" + completed = self._payloads(lines, "response.completed")[0] + assert [item["type"] for item in completed["response"]["output"]] == [ + "reasoning", + "message", + ] + assert completed["response"]["output"][0]["content"][0]["text"] == "plan" + assert completed["response"]["output"][1]["content"][0]["text"] == "plan" + + def test_structured_reasoning_content_streams_as_reasoning(self, monkeypatch): + chunks = [ + {"choices": [{"delta": {"reasoning_content": "plan"}}]}, + {"choices": [{"delta": {"content": "33"}}]}, + {"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}}, + ] + self._install_stream_mock(monkeypatch, chunks) + payload = ResponsesRequest(input = "hi", stream = True) + messages = [ChatMessage(role = "user", content = "hi")] + + async def run(): + response = await _responses_stream(payload, messages, self._Request()) + return await self._collect(response) + + lines = asyncio.run(run()) + + reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta") + text_deltas = self._payloads(lines, "response.output_text.delta") + assert "".join(event["delta"] for event in reasoning_deltas) == "plan" + assert "".join(event["delta"] for event in text_deltas) == "33" + completed = self._payloads(lines, "response.completed")[0] + assert completed["response"]["output"][0]["type"] == "reasoning" + assert completed["response"]["output"][1]["type"] == "message" + + def test_structured_reasoning_content_parts_stream_as_reasoning(self, monkeypatch): + chunks = [ + { + "choices": [ + { + "delta": { + "reasoning_content": { + "content": [ + {"type": "reasoning_text", "text": "plan"}, + {"type": "reasoning_text", "text": " next"}, + ] + } + } + } + ] + }, + {"choices": [{"delta": {"content": "33"}}]}, + {"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}}, + ] + self._install_stream_mock(monkeypatch, chunks) + payload = ResponsesRequest(input = "hi", stream = True) + messages = [ChatMessage(role = "user", content = "hi")] + + async def run(): + response = await _responses_stream(payload, messages, self._Request()) + return await self._collect(response) + + lines = asyncio.run(run()) + + reasoning_deltas = self._payloads(lines, "response.reasoning_text.delta") + text_deltas = self._payloads(lines, "response.output_text.delta") + assert "".join(event["delta"] for event in reasoning_deltas) == "plan next" + assert "".join(event["delta"] for event in text_deltas) == "33" + assert "reasoning_text" not in "".join(event["delta"] for event in reasoning_deltas) + completed = self._payloads(lines, "response.completed")[0] + assert completed["response"]["output"][0]["content"][0]["text"] == "plan next" + assert completed["response"]["output"][1]["content"][0]["text"] == "33" + + def test_tool_first_stream_closes_items_in_output_index_order(self, monkeypatch): + chunks = [ + { + "choices": [ + { + "delta": { + "tool_calls": [ + { + "index": 0, + "id": "call_0", + "type": "function", + "function": {"name": "lookup", "arguments": "{}"}, + } + ] + } + } + ] + }, + {"choices": [{"delta": {"content": "done"}}]}, + {"choices": [], "usage": {"prompt_tokens": 2, "completion_tokens": 3}}, + ] + self._install_stream_mock(monkeypatch, chunks) + payload = ResponsesRequest(input = "hi", stream = True) + messages = [ChatMessage(role = "user", content = "hi")] + + async def run(): + response = await _responses_stream(payload, messages, self._Request()) + return await self._collect(response) + + lines = asyncio.run(run()) + + done_events = self._payloads(lines, "response.output_item.done") + assert [event["output_index"] for event in done_events] == [0, 1] + assert [event["item"]["type"] for event in done_events] == ["function_call", "message"] + completed = self._payloads(lines, "response.completed")[0] + assert [item["type"] for item in completed["response"]["output"]] == [ + "function_call", + "message", + ] + def test_requests_usage_and_caps_parallel_tool_calls(self, monkeypatch): import routes.inference as inf_mod @@ -678,6 +1103,15 @@ class TestResponsesStreamAdapter: class TestResponsesOutputFunctionCall: + def test_reasoning_output_item_serialises_full_reasoning_content(self): + item = ResponsesOutputReasoning(content = [{"type": "reasoning_text", "text": "plan"}]) + d = item.model_dump() + assert d["type"] == "reasoning" + assert d["id"].startswith("rs_") + assert d["status"] == "completed" + assert d["summary"] == [] + assert d["content"] == [{"type": "reasoning_text", "text": "plan"}] + def test_direct_construction(self): fc = ResponsesOutputFunctionCall( call_id = "call_1", @@ -778,6 +1212,26 @@ class TestCodexStyleRequestShapes: assert len(req.input) == 3 assert isinstance(req.input[1], ResponsesUnknownInputItem) + def test_emitted_reasoning_item_replay_is_dropped_for_local_chat(self): + payload = ResponsesRequest( + input = [ + {"role": "user", "content": "Hi"}, + { + "type": "reasoning", + "id": "rs_1", + "summary": [], + "content": [{"type": "reasoning_text", "text": "plan"}], + }, + {"role": "assistant", "content": "33"}, + {"role": "user", "content": "Continue"}, + ], + ) + + msgs = _normalise_responses_input(payload) + + assert [m.role for m in msgs] == ["user", "assistant", "user"] + assert all("plan" not in (m.content or "") for m in msgs if isinstance(m.content, str)) + def test_unknown_content_part_type_accepted(self): """Unknown content-part types (e.g. future input_audio) validate as ResponsesUnknownContentPart so the request doesn't 422."""