From 4de411d9f4ba57ccc0b5046b91cb66900a2b8d07 Mon Sep 17 00:00:00 2001 From: sufubao Date: Tue, 7 Jul 2026 10:00:38 +0800 Subject: [PATCH 1/3] feat: support /v1/responses --- lightllm/server/api_http.py | 21 ++ lightllm/server/api_responses.py | 558 +++++++++++++++++++++++++++++++ 2 files changed, 579 insertions(+) create mode 100644 lightllm/server/api_responses.py diff --git a/lightllm/server/api_http.py b/lightllm/server/api_http.py index e127f0931c..322df822ff 100755 --- a/lightllm/server/api_http.py +++ b/lightllm/server/api_http.py @@ -352,6 +352,27 @@ async def anthropic_messages(raw_request: Request) -> Response: return Response(status_code=499) +@app.post("/v1/responses") +async def openai_responses(raw_request: Request) -> Response: + if get_env_start_args().run_mode in ["prefill", "decode"]: + return create_error_response( + HTTPStatus.EXPECTATION_FAILED, "service in pd mode dont recv reqs from http interface" + ) + from .api_responses import responses_impl + + try: + return await responses_impl(raw_request) + except ServerBusyError as e: + logger.error("%s", str(e), exc_info=True) + return create_error_response(HTTPStatus.SERVICE_UNAVAILABLE, str(e)) + except ClientDisconnected as e: + logger.warning(str(e)) + return Response(status_code=499) + except Exception as e: + logger.error("An error occurred: %s", str(e), exc_info=True) + return create_error_response(HTTPStatus.EXPECTATION_FAILED, str(e)) + + @app.get("/v1/models", response_model=ModelListResponse) async def get_models(raw_request: Request): model_name = g_objs.args.model_name diff --git a/lightllm/server/api_responses.py b/lightllm/server/api_responses.py new file mode 100644 index 0000000000..4ce0e19673 --- /dev/null +++ b/lightllm/server/api_responses.py @@ -0,0 +1,558 @@ +from __future__ import annotations + +import time +import uuid +import ujson as json +from http import HTTPStatus +from typing import Any, AsyncGenerator, Dict, List, Optional + +from fastapi import Request +from fastapi.responses import JSONResponse, Response, StreamingResponse + +from lightllm.utils.log_utils import init_logger + +logger = init_logger(__name__) + + +def _content_parts_to_chat(parts: List[Any]) -> List[Dict[str, Any]]: + chat_parts = [] + for part in parts: + if isinstance(part, str): + chat_parts.append({"type": "text", "text": part}) + continue + ptype = part.get("type") + if ptype in ("input_text", "output_text", "text"): + chat_parts.append({"type": "text", "text": part.get("text", "")}) + elif ptype == "refusal": + chat_parts.append({"type": "text", "text": part.get("refusal", "")}) + elif ptype == "input_image": + url = part.get("image_url") + if isinstance(url, dict): + url = url.get("url") + if not url: + raise ValueError("input_image requires an image_url") + chat_parts.append({"type": "image_url", "image_url": {"url": url}}) + elif ptype == "input_audio": + audio = part.get("input_audio") or {} + url = audio.get("url") or part.get("audio_url") + if not url: + raise ValueError("input_audio requires a url (raw audio data is not supported)") + chat_parts.append({"type": "audio_url", "audio_url": {"url": url}}) + else: + raise ValueError(f"Unsupported input content type: {ptype}") + return chat_parts + + +def _input_items_to_messages(items: List[Any]) -> List[Dict[str, Any]]: + messages: List[Dict[str, Any]] = [] + for item in items: + if not isinstance(item, dict): + raise ValueError("input items must be objects") + itype = item.get("type") + if itype == "function_call": + messages.append( + { + "role": "assistant", + "content": None, + "tool_calls": [ + { + "id": item.get("call_id") or item.get("id"), + "type": "function", + "function": {"name": item.get("name"), "arguments": item.get("arguments") or ""}, + } + ], + } + ) + elif itype == "function_call_output": + output = item.get("output") + if not isinstance(output, str): + output = json.dumps(output, ensure_ascii=False) + messages.append({"role": "tool", "tool_call_id": item.get("call_id"), "content": output}) + elif itype == "reasoning": + continue + elif itype in (None, "message"): + role = item.get("role", "user") + if role == "developer": + role = "system" + content = item.get("content") + if isinstance(content, list): + content = _content_parts_to_chat(content) + messages.append({"role": role, "content": content}) + else: + raise ValueError(f"Unsupported input item type: {itype}") + return messages + + +def _tools_to_chat(tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + chat_tools = [] + for tool in tools: + if tool.get("type") != "function": + logger.warning("Ignoring unsupported tool type: %s", tool.get("type")) + continue + if "function" in tool: + chat_tools.append(tool) + continue + chat_tools.append( + { + "type": "function", + "function": { + "name": tool.get("name"), + "description": tool.get("description"), + "parameters": tool.get("parameters"), + }, + } + ) + return chat_tools + + +def _text_format_to_response_format(body: Dict[str, Any]) -> Optional[Dict[str, Any]]: + fmt = (body.get("text") or {}).get("format") + if not fmt: + return None + ftype = fmt.get("type") + if ftype == "text": + return None + if ftype == "json_object": + return {"type": "json_object"} + if ftype == "json_schema": + return { + "type": "json_schema", + "json_schema": { + "name": fmt.get("name", "response"), + "description": fmt.get("description"), + "schema": fmt.get("schema"), + "strict": fmt.get("strict"), + }, + } + raise ValueError(f"Unsupported text.format type: {ftype}") + + +def _responses_to_chat_request(body: Dict[str, Any]) -> Dict[str, Any]: + messages: List[Dict[str, Any]] = [] + if body.get("instructions"): + messages.append({"role": "system", "content": body["instructions"]}) + + inp = body.get("input") + if isinstance(inp, str): + messages.append({"role": "user", "content": inp}) + elif isinstance(inp, list): + messages.extend(_input_items_to_messages(inp)) + else: + raise ValueError("'input' must be a string or an array of input items") + + system_texts = [] + for m in messages: + if m["role"] == "system": + content = m["content"] + if isinstance(content, list): + content = "\n".join(p.get("text", "") for p in content) + if content: + system_texts.append(content) + messages = [m for m in messages if m["role"] != "system"] + if system_texts: + messages.insert(0, {"role": "system", "content": "\n\n".join(system_texts)}) + + chat: Dict[str, Any] = { + "model": body.get("model", "default"), + "messages": messages, + "stream": bool(body.get("stream")), + "n": 1, + } + for src, dst in ( + ("temperature", "temperature"), + ("top_p", "top_p"), + ("max_output_tokens", "max_completion_tokens"), + ("parallel_tool_calls", "parallel_tool_calls"), + ("user", "user"), + ): + if body.get(src) is not None: + chat[dst] = body[src] + + if body.get("tools"): + chat["tools"] = _tools_to_chat(body["tools"]) + tool_choice = body.get("tool_choice") + if tool_choice is not None: + if isinstance(tool_choice, dict) and tool_choice.get("type") == "function": + chat["tool_choice"] = {"type": "function", "function": {"name": tool_choice.get("name")}} + else: + chat["tool_choice"] = tool_choice + + effort = (body.get("reasoning") or {}).get("effort") + if effort in ("low", "medium", "high"): + chat["reasoning_effort"] = effort + + response_format = _text_format_to_response_format(body) + if response_format: + chat["response_format"] = response_format + + extra_body = body.get("extra_body") + if isinstance(extra_body, dict): + for k, v in extra_body.items(): + chat.setdefault(k, v) + + return chat + + +def _new_ids() -> Dict[str, str]: + return { + "response": f"resp_{uuid.uuid4().hex}", + "message": f"msg_{uuid.uuid4().hex}", + "reasoning": f"rs_{uuid.uuid4().hex}", + } + + +def _response_envelope(body: Dict[str, Any], response_id: str, created_at: int) -> Dict[str, Any]: + return { + "id": response_id, + "object": "response", + "created_at": created_at, + "status": "in_progress", + "background": False, + "error": None, + "incomplete_details": None, + "instructions": body.get("instructions"), + "max_output_tokens": body.get("max_output_tokens"), + "model": body.get("model", "default"), + "output": [], + "parallel_tool_calls": body.get("parallel_tool_calls", True), + "previous_response_id": None, + "reasoning": body.get("reasoning"), + "store": False, + "temperature": body.get("temperature"), + "text": body.get("text") or {"format": {"type": "text"}}, + "tool_choice": body.get("tool_choice", "auto"), + "tools": body.get("tools") or [], + "top_p": body.get("top_p"), + "truncation": body.get("truncation", "disabled"), + "usage": None, + "user": body.get("user"), + "metadata": body.get("metadata") or {}, + } + + +def _usage_to_responses(usage: Dict[str, Any]) -> Dict[str, Any]: + input_tokens = int(usage.get("prompt_tokens", 0)) + output_tokens = int(usage.get("completion_tokens", 0)) + cached = int((usage.get("prompt_tokens_details") or {}).get("cached_tokens", 0) or 0) + return { + "input_tokens": input_tokens, + "input_tokens_details": {"cached_tokens": cached}, + "output_tokens": output_tokens, + "output_tokens_details": {"reasoning_tokens": 0}, + "total_tokens": input_tokens + output_tokens, + } + + +def _chat_response_to_responses(chat_response: Any, body: Dict[str, Any]) -> Dict[str, Any]: + if hasattr(chat_response, "model_dump"): + openai_dict = chat_response.model_dump(exclude_none=True) + else: + openai_dict = dict(chat_response) + + ids = _new_ids() + created_at = int(openai_dict.get("created") or time.time()) + result = _response_envelope(body, ids["response"], created_at) + + choice = (openai_dict.get("choices") or [{}])[0] + message = choice.get("message") or {} + finish_reason = choice.get("finish_reason") + + output: List[Dict[str, Any]] = [] + reasoning_text = message.get("reasoning") or message.get("reasoning_content") + if reasoning_text: + output.append( + { + "type": "reasoning", + "id": ids["reasoning"], + "summary": [], + "content": [{"type": "reasoning_text", "text": reasoning_text}], + } + ) + text = message.get("content") + if text: + output.append( + { + "type": "message", + "id": ids["message"], + "status": "completed", + "role": "assistant", + "content": [{"type": "output_text", "text": text, "annotations": []}], + } + ) + for tc in message.get("tool_calls") or []: + fn = tc.get("function") or {} + output.append( + { + "type": "function_call", + "id": f"fc_{uuid.uuid4().hex}", + "call_id": tc.get("id"), + "name": fn.get("name"), + "arguments": fn.get("arguments") or "", + "status": "completed", + } + ) + + result["output"] = output + result["usage"] = _usage_to_responses(openai_dict.get("usage") or {}) + if finish_reason == "length": + result["status"] = "incomplete" + result["incomplete_details"] = {"reason": "max_output_tokens"} + else: + result["status"] = "completed" + return result + + +async def _openai_sse_to_responses_events( + openai_body_iterator, + body: Dict[str, Any], +) -> AsyncGenerator[bytes, None]: + seq = 0 + + def event(event_type: str, data: Dict[str, Any]) -> bytes: + nonlocal seq + seq += 1 + data = {"type": event_type, "sequence_number": seq, **data} + return f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n".encode("utf-8") + + ids = _new_ids() + response = _response_envelope(body, ids["response"], int(time.time())) + + yield event("response.created", {"response": response}) + yield event("response.in_progress", {"response": response}) + + output_index = -1 + current: Optional[tuple] = None + finish_reason = None + usage: Dict[str, Any] = {} + failed_error: Optional[Dict[str, Any]] = None + + def close_current(): + nonlocal current + if current is None: + return + kind, item = current + current = None + if kind == "message": + text = item["content"][0]["text"] + yield event( + "response.output_text.done", + {"item_id": item["id"], "output_index": output_index, "content_index": 0, "text": text}, + ) + yield event( + "response.content_part.done", + { + "item_id": item["id"], + "output_index": output_index, + "content_index": 0, + "part": {"type": "output_text", "text": text, "annotations": []}, + }, + ) + elif kind == "reasoning": + yield event( + "response.reasoning_text.done", + { + "item_id": item["id"], + "output_index": output_index, + "content_index": 0, + "text": item["content"][0]["text"], + }, + ) + elif kind == "function_call": + yield event( + "response.function_call_arguments.done", + {"item_id": item["id"], "output_index": output_index, "arguments": item["arguments"]}, + ) + item["status"] = "completed" + response["output"].append(item) + yield event("response.output_item.done", {"output_index": output_index, "item": item}) + + def open_item(kind: str, item: Dict[str, Any]): + nonlocal current, output_index + yield from close_current() + output_index += 1 + current = (kind, item) + yield event("response.output_item.added", {"output_index": output_index, "item": item}) + if kind == "message": + yield event( + "response.content_part.added", + { + "item_id": item["id"], + "output_index": output_index, + "content_index": 0, + "part": {"type": "output_text", "text": "", "annotations": []}, + }, + ) + + async for raw_chunk in openai_body_iterator: + if not raw_chunk: + continue + if isinstance(raw_chunk, (bytes, bytearray)): + raw_chunk = raw_chunk.decode("utf-8", errors="replace") + for line in raw_chunk.split("\n"): + line = line.strip() + if not line.startswith("data: "): + continue + payload = line[len("data: ") :] + if payload == "[DONE]": + continue + try: + chunk = json.loads(payload) + except Exception: + logger.debug("Skipping non-JSON SSE payload: %r", payload) + continue + + if "error" in chunk and "choices" not in chunk: + failed_error = chunk["error"] + break + + if chunk.get("usage"): + usage = chunk["usage"] + choices = chunk.get("choices") or [] + if not choices: + continue + choice = choices[0] + delta = choice.get("delta") or {} + if choice.get("finish_reason"): + finish_reason = choice["finish_reason"] + + reasoning_piece = delta.get("reasoning") or delta.get("reasoning_content") + if reasoning_piece: + if current is None or current[0] != "reasoning": + item = { + "type": "reasoning", + "id": f"rs_{uuid.uuid4().hex}", + "summary": [], + "content": [{"type": "reasoning_text", "text": ""}], + "status": "in_progress", + } + for e in open_item("reasoning", item): + yield e + item = current[1] + item["content"][0]["text"] += reasoning_piece + yield event( + "response.reasoning_text.delta", + { + "item_id": item["id"], + "output_index": output_index, + "content_index": 0, + "delta": reasoning_piece, + }, + ) + + content_piece = delta.get("content") + if content_piece: + if current is None or current[0] != "message": + item = { + "type": "message", + "id": f"msg_{uuid.uuid4().hex}", + "status": "in_progress", + "role": "assistant", + "content": [{"type": "output_text", "text": "", "annotations": []}], + } + for e in open_item("message", item): + yield e + item = current[1] + item["content"][0]["text"] += content_piece + yield event( + "response.output_text.delta", + { + "item_id": item["id"], + "output_index": output_index, + "content_index": 0, + "delta": content_piece, + }, + ) + + for tc in delta.get("tool_calls") or []: + fn = tc.get("function") or {} + if fn.get("name"): + item = { + "type": "function_call", + "id": f"fc_{uuid.uuid4().hex}", + "call_id": tc.get("id") or f"call_{uuid.uuid4().hex[:24]}", + "name": fn["name"], + "arguments": "", + "status": "in_progress", + } + for e in open_item("function_call", item): + yield e + args = fn.get("arguments") + if args and current is not None and current[0] == "function_call": + item = current[1] + item["arguments"] += args + yield event( + "response.function_call_arguments.delta", + {"item_id": item["id"], "output_index": output_index, "delta": args}, + ) + if failed_error is not None: + break + + for e in close_current(): + yield e + + if failed_error is not None: + response["status"] = "failed" + response["error"] = {"code": "server_error", "message": failed_error.get("message", "generation failed")} + yield event("response.failed", {"response": response}) + return + + response["usage"] = _usage_to_responses(usage) + if finish_reason == "length": + response["status"] = "incomplete" + response["incomplete_details"] = {"reason": "max_output_tokens"} + yield event("response.incomplete", {"response": response}) + else: + response["status"] = "completed" + yield event("response.completed", {"response": response}) + + +async def responses_impl(raw_request: Request) -> Response: + from .api_models import ChatCompletionRequest, ChatCompletionResponse + from .api_openai import chat_completions_impl, create_error_response + + try: + body = await raw_request.json() + except Exception as exc: + return create_error_response(HTTPStatus.BAD_REQUEST, f"Invalid JSON body: {exc}") + if not isinstance(body, dict): + return create_error_response(HTTPStatus.BAD_REQUEST, "Request body must be a JSON object") + + if body.get("previous_response_id"): + return create_error_response( + HTTPStatus.BAD_REQUEST, + "previous_response_id is not supported (this server is stateless); " + "resend the full conversation in 'input' instead", + param="previous_response_id", + ) + if body.get("background"): + return create_error_response( + HTTPStatus.BAD_REQUEST, "background responses are not supported", param="background" + ) + + try: + chat_dict = _responses_to_chat_request(body) + chat_request = ChatCompletionRequest(**chat_dict) + except ValueError as exc: + return create_error_response(HTTPStatus.BAD_REQUEST, str(exc)) + except Exception as exc: + logger.exception("Failed to translate Responses API request") + return create_error_response(HTTPStatus.BAD_REQUEST, f"Invalid request: {exc}") + + downstream = await chat_completions_impl(chat_request, raw_request) + + if chat_request.stream: + if not isinstance(downstream, StreamingResponse): + return downstream + return StreamingResponse( + _openai_sse_to_responses_events(downstream.body_iterator, body), + media_type="text/event-stream", + ) + + if not isinstance(downstream, ChatCompletionResponse): + return downstream + + try: + return JSONResponse(_chat_response_to_responses(downstream, body)) + except Exception as exc: + logger.error("Failed to translate response to Responses API format: %s", exc, exc_info=True) + return create_error_response(HTTPStatus.INTERNAL_SERVER_ERROR, str(exc)) From b386c36a48f9ed6d45ffbf1070e31100b17f2882 Mon Sep 17 00:00:00 2001 From: sufubao Date: Fri, 17 Jul 2026 17:47:57 +0800 Subject: [PATCH 2/3] fix: align Responses API behavior --- lightllm/server/api_responses.py | 24 +++++-- unit_tests/server/test_api_responses.py | 87 +++++++++++++++++++++++++ 2 files changed, 105 insertions(+), 6 deletions(-) create mode 100644 unit_tests/server/test_api_responses.py diff --git a/lightllm/server/api_responses.py b/lightllm/server/api_responses.py index 4ce0e19673..c09dd3bdd6 100644 --- a/lightllm/server/api_responses.py +++ b/lightllm/server/api_responses.py @@ -128,6 +128,14 @@ def _text_format_to_response_format(body: Dict[str, Any]) -> Optional[Dict[str, def _responses_to_chat_request(body: Dict[str, Any]) -> Dict[str, Any]: + truncation = body.get("truncation") + if truncation not in (None, "disabled"): + raise ValueError("Only truncation='disabled' is supported") + + effort = (body.get("reasoning") or {}).get("effort") + if effort is not None and effort not in ("low", "medium", "high"): + raise ValueError("reasoning.effort must be one of: low, medium, high") + messages: List[Dict[str, Any]] = [] if body.get("instructions"): messages.append({"role": "system", "content": body["instructions"]}) @@ -177,8 +185,7 @@ def _responses_to_chat_request(body: Dict[str, Any]) -> Dict[str, Any]: else: chat["tool_choice"] = tool_choice - effort = (body.get("reasoning") or {}).get("effort") - if effort in ("low", "medium", "high"): + if effort is not None: chat["reasoning_effort"] = effort response_format = _text_format_to_response_format(body) @@ -360,7 +367,12 @@ def close_current(): elif kind == "function_call": yield event( "response.function_call_arguments.done", - {"item_id": item["id"], "output_index": output_index, "arguments": item["arguments"]}, + { + "item_id": item["id"], + "output_index": output_index, + "name": item["name"], + "arguments": item["arguments"], + }, ) item["status"] = "completed" response["output"].append(item) @@ -487,15 +499,15 @@ def open_item(kind: str, item: Dict[str, Any]): if failed_error is not None: break - for e in close_current(): - yield e - if failed_error is not None: response["status"] = "failed" response["error"] = {"code": "server_error", "message": failed_error.get("message", "generation failed")} yield event("response.failed", {"response": response}) return + for e in close_current(): + yield e + response["usage"] = _usage_to_responses(usage) if finish_reason == "length": response["status"] = "incomplete" diff --git a/unit_tests/server/test_api_responses.py b/unit_tests/server/test_api_responses.py new file mode 100644 index 0000000000..71750f06cf --- /dev/null +++ b/unit_tests/server/test_api_responses.py @@ -0,0 +1,87 @@ +import asyncio + +import pytest +import ujson as json + +from lightllm.server.api_responses import _openai_sse_to_responses_events, _responses_to_chat_request + + +async def _chunks(*payloads): + for payload in payloads: + yield f"data: {json.dumps(payload)}\n\n" + + +def _collect_events(*payloads): + async def collect(): + return [event async for event in _openai_sse_to_responses_events(_chunks(*payloads), {"input": "hi"})] + + raw_events = asyncio.run(collect()) + return [json.loads(event.decode().split("data: ", 1)[1]) for event in raw_events] + + +def test_function_call_arguments_done_includes_name(): + events = _collect_events( + { + "choices": [ + { + "delta": { + "tool_calls": [ + { + "id": "call_1", + "function": {"name": "get_weather", "arguments": '{"city":"Paris"}'}, + } + ] + } + } + ] + } + ) + + done = next(event for event in events if event["type"] == "response.function_call_arguments.done") + assert done["name"] == "get_weather" + + +@pytest.mark.parametrize( + "partial_payload", + [ + {"choices": [{"delta": {"content": "partial"}}]}, + { + "choices": [ + { + "delta": { + "tool_calls": [ + { + "id": "call_1", + "function": {"name": "get_weather", "arguments": '{"city":'}, + } + ] + } + } + ] + }, + ], +) +def test_stream_failure_does_not_complete_partial_item(partial_payload): + events = _collect_events(partial_payload, {"error": {"message": "generation failed"}}) + event_types = [event["type"] for event in events] + + assert "response.output_text.done" not in event_types + assert "response.function_call_arguments.done" not in event_types + assert "response.output_item.done" not in event_types + assert event_types[-1] == "response.failed" + + +@pytest.mark.parametrize("effort", ["none", "minimal", "xhigh"]) +def test_unsupported_reasoning_effort_is_rejected(effort): + with pytest.raises(ValueError, match="reasoning.effort"): + _responses_to_chat_request({"input": "hi", "reasoning": {"effort": effort}}) + + +def test_supported_reasoning_effort_is_forwarded(): + request = _responses_to_chat_request({"input": "hi", "reasoning": {"effort": "high"}}) + assert request["reasoning_effort"] == "high" + + +def test_automatic_truncation_is_rejected(): + with pytest.raises(ValueError, match="truncation"): + _responses_to_chat_request({"input": "hi", "truncation": "auto"}) From e401f6be451943bbe692bd93a1a776cc27fafd6a Mon Sep 17 00:00:00 2001 From: sufubao Date: Fri, 17 Jul 2026 18:24:19 +0800 Subject: [PATCH 3/3] fix: correct Responses API edge cases --- lightllm/server/api_http.py | 2 + lightllm/server/api_responses.py | 14 +-- unit_tests/server/test_api_responses.py | 109 +++++++++++++++++++++++- 3 files changed, 119 insertions(+), 6 deletions(-) diff --git a/lightllm/server/api_http.py b/lightllm/server/api_http.py index 322df822ff..7633e8801d 100755 --- a/lightllm/server/api_http.py +++ b/lightllm/server/api_http.py @@ -365,6 +365,8 @@ async def openai_responses(raw_request: Request) -> Response: except ServerBusyError as e: logger.error("%s", str(e), exc_info=True) return create_error_response(HTTPStatus.SERVICE_UNAVAILABLE, str(e)) + except ValueError as e: + return create_error_response(HTTPStatus.BAD_REQUEST, str(e)) except ClientDisconnected as e: logger.warning(str(e)) return Response(status_code=499) diff --git a/lightllm/server/api_responses.py b/lightllm/server/api_responses.py index c09dd3bdd6..a0ca3613de 100644 --- a/lightllm/server/api_responses.py +++ b/lightllm/server/api_responses.py @@ -65,7 +65,9 @@ def _input_items_to_messages(items: List[Any]) -> List[Dict[str, Any]]: ) elif itype == "function_call_output": output = item.get("output") - if not isinstance(output, str): + if isinstance(output, list): + output = _content_parts_to_chat(output) + elif not isinstance(output, str): output = json.dumps(output, ensure_ascii=False) messages.append({"role": "tool", "tool_call_id": item.get("call_id"), "content": output}) elif itype == "reasoning": @@ -263,6 +265,7 @@ def _chat_response_to_responses(chat_response: Any, body: Dict[str, Any]) -> Dic choice = (openai_dict.get("choices") or [{}])[0] message = choice.get("message") or {} finish_reason = choice.get("finish_reason") + output_status = "incomplete" if finish_reason == "length" else "completed" output: List[Dict[str, Any]] = [] reasoning_text = message.get("reasoning") or message.get("reasoning_content") @@ -281,7 +284,7 @@ def _chat_response_to_responses(chat_response: Any, body: Dict[str, Any]) -> Dic { "type": "message", "id": ids["message"], - "status": "completed", + "status": output_status, "role": "assistant", "content": [{"type": "output_text", "text": text, "annotations": []}], } @@ -295,7 +298,7 @@ def _chat_response_to_responses(chat_response: Any, body: Dict[str, Any]) -> Dic "call_id": tc.get("id"), "name": fn.get("name"), "arguments": fn.get("arguments") or "", - "status": "completed", + "status": output_status, } ) @@ -374,7 +377,7 @@ def close_current(): "arguments": item["arguments"], }, ) - item["status"] = "completed" + item["status"] = "incomplete" if finish_reason == "length" else "completed" response["output"].append(item) yield event("response.output_item.done", {"output_index": output_index, "item": item}) @@ -383,7 +386,8 @@ def open_item(kind: str, item: Dict[str, Any]): yield from close_current() output_index += 1 current = (kind, item) - yield event("response.output_item.added", {"output_index": output_index, "item": item}) + added_item = {**item, "content": []} if kind == "message" else item + yield event("response.output_item.added", {"output_index": output_index, "item": added_item}) if kind == "message": yield event( "response.content_part.added", diff --git a/unit_tests/server/test_api_responses.py b/unit_tests/server/test_api_responses.py index 71750f06cf..70048be5a6 100644 --- a/unit_tests/server/test_api_responses.py +++ b/unit_tests/server/test_api_responses.py @@ -3,7 +3,11 @@ import pytest import ujson as json -from lightllm.server.api_responses import _openai_sse_to_responses_events, _responses_to_chat_request +from lightllm.server.api_responses import ( + _chat_response_to_responses, + _openai_sse_to_responses_events, + _responses_to_chat_request, +) async def _chunks(*payloads): @@ -41,6 +45,109 @@ def test_function_call_arguments_done_includes_name(): assert done["name"] == "get_weather" +def test_content_array_function_output_is_preserved(): + request = _responses_to_chat_request( + { + "input": [ + { + "type": "function_call_output", + "call_id": "call_1", + "output": [ + {"type": "input_text", "text": "sunny"}, + {"type": "input_image", "image_url": "https://example.com/weather.png"}, + ], + } + ] + } + ) + + assert request["messages"] == [ + { + "role": "tool", + "tool_call_id": "call_1", + "content": [ + {"type": "text", "text": "sunny"}, + {"type": "image_url", "image_url": {"url": "https://example.com/weather.png"}}, + ], + } + ] + + +def test_non_streamed_truncated_function_call_is_incomplete(): + response = _chat_response_to_responses( + { + "choices": [ + { + "finish_reason": "length", + "message": { + "tool_calls": [ + { + "id": "call_1", + "function": {"name": "get_weather", "arguments": '{"city":'}, + } + ] + }, + } + ] + }, + {"input": "hi"}, + ) + + assert response["status"] == "incomplete" + assert response["output"][0]["status"] == "incomplete" + + +def test_streamed_truncated_function_call_is_incomplete(): + events = _collect_events( + { + "choices": [ + { + "delta": { + "tool_calls": [ + { + "id": "call_1", + "function": {"name": "get_weather", "arguments": '{"city":'}, + } + ] + } + } + ] + }, + {"choices": [{"delta": {}, "finish_reason": "length"}]}, + ) + + item_done = next(event for event in events if event["type"] == "response.output_item.done") + response_done = events[-1] + assert item_done["item"]["status"] == "incomplete" + assert response_done["type"] == "response.incomplete" + assert response_done["response"]["output"][0]["status"] == "incomplete" + + +def test_streamed_message_adds_text_content_once(): + events = _collect_events({"choices": [{"delta": {"content": "hello"}, "finish_reason": "stop"}]}) + + item_added = next(event for event in events if event["type"] == "response.output_item.added") + part_added = next(event for event in events if event["type"] == "response.content_part.added") + assert item_added["item"]["content"] == [] + assert part_added["part"] == {"type": "output_text", "text": "", "annotations": []} + + +def test_route_maps_downstream_value_error_to_bad_request(monkeypatch): + from types import SimpleNamespace + + from lightllm.server import api_http, api_responses + + async def raise_value_error(raw_request): + raise ValueError("Unrecognized image input.") + + monkeypatch.setattr(api_http, "get_env_start_args", lambda: SimpleNamespace(run_mode="normal")) + monkeypatch.setattr(api_http.g_objs, "metric_client", SimpleNamespace(counter_inc=lambda *args: None)) + monkeypatch.setattr(api_responses, "responses_impl", raise_value_error) + + response = asyncio.run(api_http.openai_responses(None)) + assert response.status_code == 400 + + @pytest.mark.parametrize( "partial_payload", [