lingma-openai-gateway/app/http/responses_adapter.py

from __future__ import annotations

import json
import time
import uuid
from typing import Any

from fastapi import HTTPException

from ..openai_schema import ChatCompletionsRequest, ResponsesRequest, flatten_content


def _responses_input_to_messages(req: ResponsesRequest) -> list[dict[str, Any]]:
    messages: list[dict[str, Any]] = []
    if req.instructions:
        messages.append({"role": "system", "content": req.instructions})

    raw_input = req.input
    if raw_input is None:
        return messages

    valid_roles = {"system", "user", "assistant", "tool", "developer", "function"}

    def _append(role: str, content: Any, *, tool_call_id: str | None = None) -> None:
        msg: dict[str, Any] = {"role": role, "content": flatten_content(content)}
        if role == "tool" and tool_call_id:
            msg["tool_call_id"] = tool_call_id
        messages.append(msg)

    if isinstance(raw_input, str):
        _append("user", raw_input)
        return messages

    raw_items: list[Any]
    if isinstance(raw_input, dict):
        raw_items = [raw_input]
    elif isinstance(raw_input, list):
        raw_items = list(raw_input)
    else:
        _append("user", str(raw_input))
        return messages

    for item in raw_items:
        if isinstance(item, str):
            _append("user", item)
            continue
        if not isinstance(item, dict):
            _append("user", str(item))
            continue

        role = item.get("role")
        if isinstance(role, str) and role in valid_roles:
            tool_call_id = item.get("tool_call_id") or item.get("call_id")
            _append(role, item.get("content"), tool_call_id=str(tool_call_id) if tool_call_id else None)
            continue

        if item.get("type") == "function_call_output":
            output = item.get("output")
            if isinstance(output, (dict, list)):
                output = json.dumps(output, ensure_ascii=False)
            tool_call_id = item.get("call_id")
            _append("tool", output, tool_call_id=str(tool_call_id) if tool_call_id else None)
            continue

        if "content" in item:
            text = flatten_content(item.get("content"))
        else:
            text = flatten_content([item])
        if text:
            _append("user", text)

    return messages


def _responses_to_chat_request(req: ResponsesRequest) -> ChatCompletionsRequest:
    return ChatCompletionsRequest(
        model=req.model,
        messages=_responses_input_to_messages(req),
        stream=req.stream,
        temperature=req.temperature,
        top_p=req.top_p,
        max_tokens=req.max_output_tokens,
        user=req.user,
        tools=req.tools,
        tool_choice=req.tool_choice,
    )


def _responses_id_from_chat_id(chat_id: Any) -> str:
    if isinstance(chat_id, str) and chat_id:
        suffix = chat_id.removeprefix("chatcmpl-")
        return f"resp_{suffix}"
    return f"resp_{uuid.uuid4().hex}"


def _responses_usage_from_chat(usage: Any) -> dict[str, int]:
    if not isinstance(usage, dict):
        return {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
    input_tokens = int(usage.get("prompt_tokens") or 0)
    output_tokens = int(usage.get("completion_tokens") or 0)
    return {
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
        "total_tokens": int(usage.get("total_tokens") or (input_tokens + output_tokens)),
    }


def _responses_non_stream_from_chat_payload(chat_payload: Any) -> dict[str, Any]:
    if not isinstance(chat_payload, dict):
        raise HTTPException(
            status_code=502,
            detail={"error": {"message": "invalid upstream response", "type": "upstream_error"}},
        )
    choice = {}
    choices = chat_payload.get("choices")
    if isinstance(choices, list) and choices:
        choice = choices[0] if isinstance(choices[0], dict) else {}
    message = choice.get("message") if isinstance(choice.get("message"), dict) else {}

    output: list[dict[str, Any]] = []
    content = message.get("content")
    if isinstance(content, str) and content:
        output.append(
            {
                "type": "message",
                "id": f"msg_{uuid.uuid4().hex}",
                "status": "completed",
                "role": "assistant",
                "content": [{"type": "output_text", "text": content}],
            }
        )

    tool_calls = message.get("tool_calls")
    if isinstance(tool_calls, list):
        for idx, tool_call in enumerate(tool_calls):
            if not isinstance(tool_call, dict):
                continue
            fn = tool_call.get("function") if isinstance(tool_call.get("function"), dict) else {}
            call_id = str(tool_call.get("id") or f"call_{idx}")
            output.append(
                {
                    "type": "function_call",
                    "id": call_id,
                    "call_id": call_id,
                    "name": str(fn.get("name") or "tool"),
                    "arguments": str(fn.get("arguments") or "{}"),
                }
            )

    output_text_parts: list[str] = []
    for item in output:
        if item.get("type") == "message":
            blocks = item.get("content")
            if isinstance(blocks, list):
                for block in blocks:
                    if isinstance(block, dict) and block.get("type") == "output_text":
                        text = block.get("text")
                        if isinstance(text, str) and text:
                            output_text_parts.append(text)

    return {
        "id": _responses_id_from_chat_id(chat_payload.get("id")),
        "object": "response",
        "created_at": int(chat_payload.get("created") or time.time()),
        "status": "completed",
        "error": None,
        "incomplete_details": None,
        "model": chat_payload.get("model"),
        "output": output,
        "output_text": "".join(output_text_parts),
        "usage": _responses_usage_from_chat(chat_payload.get("usage")),
    }


def _sse_data(payload: dict[str, Any]) -> str:
    return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"