o2o-site-AEO/solution/backend/services/llm/openai.py

"""OpenAI Chat Completions 호출 — services/llm/gemini.py 와 같은 자리, 다른 공급자."""
import asyncio
import base64
import json

import httpx

from config.server_configs import external_api_config
from services.llm.errors import LlmError, LlmInvalidOutput, LlmNotConfigured
from services.llm.types import ImagePart, LlmResult, Usage

_BASE_URL = "https://api.openai.com/v1/chat/completions"

DEFAULT_MODEL = "gpt-5.6-luna"

_PRICE_PER_1M_INPUT = {"gpt-5.6-luna": 0.20, "gpt-5.6-terra": 2.00, "gpt-5.6-sol": 5.00}
_PRICE_PER_1M_OUTPUT = {"gpt-5.6-luna": 1.20, "gpt-5.6-terra": 12.00, "gpt-5.6-sol": 30.00}

_RETRYABLE_STATUS = {408, 429, 500, 502, 503, 504}


def is_configured() -> bool:
    return bool(external_api_config.openai_api_key)


def _to_strict_schema(schema: dict) -> dict:
    """OpenAI strict 모드 요구사항(모든 object 에 additionalProperties:false, 모든 property 가 required)을 만족하도록 재귀 변환한다."""
    schema = dict(schema)
    if schema.get("type") == "object" and "properties" in schema:
        schema["properties"] = {k: _to_strict_schema(v) for k, v in schema["properties"].items()}
        schema["required"] = list(schema["properties"].keys())
        schema["additionalProperties"] = False
    elif schema.get("type") == "array" and "items" in schema:
        schema["items"] = _to_strict_schema(schema["items"])
    return schema


def _build_messages(prompt: str, images: list[ImagePart] | None) -> list[dict]:
    content: list[dict] = [{"type": "text", "text": prompt}]
    for image in images or []:
        if image.label:
            content.append({"type": "text", "text": image.label})
        b64 = base64.b64encode(image.data).decode()
        content.append({"type": "image_url", "image_url": {"url": f"data:{image.mime_type};base64,{b64}"}})
    return [{"role": "user", "content": content}]


async def generate(
    client: httpx.AsyncClient,
    model: str,
    *,
    prompt: str,
    images: list[ImagePart] | None = None,
    response_schema: dict | None = None,
    temperature: float = 0.2,
    max_retries: int = 2,
) -> LlmResult:
    body: dict = {
        "model": model,
        "messages": _build_messages(prompt, images),
    }
    if response_schema is not None:
        body["response_format"] = {
            "type": "json_schema",
            "json_schema": {"name": "result", "strict": True, "schema": _to_strict_schema(response_schema)},
        }

    headers = {"Content-Type": "application/json", "Authorization": f"Bearer {external_api_config.openai_api_key}"}
    last = None
    payload = None
    for attempt in range(max_retries + 1):
        try:
            resp = await client.post(_BASE_URL, json=body, headers=headers)
        except (httpx.TimeoutException, httpx.TransportError) as ex:
            last = f"{type(ex).__name__}: {ex}"
        else:
            if resp.status_code == 200:
                payload = resp.json()
                break
            if resp.status_code in (401, 403):
                raise LlmNotConfigured(f"인증 실패 status={resp.status_code} — API 키를 확인하세요")
            if resp.status_code not in _RETRYABLE_STATUS:
                raise LlmError(f"status={resp.status_code} body={resp.text[:200]}")
            last = f"status={resp.status_code}"

        if attempt < max_retries:
            await asyncio.sleep(min(8.0, 1.0 * (2 ** attempt)))

    if payload is None:
        raise LlmError(f"{max_retries + 1}회 시도 실패: {last}")

    choices = payload.get("choices") or []
    if not choices:
        raise LlmInvalidOutput("choices 가 비었다(안전 필터 차단 가능)")
    text = (choices[0].get("message") or {}).get("content") or ""
    if not text.strip():
        raise LlmInvalidOutput(f"텍스트가 없다 finish_reason={choices[0].get('finish_reason')}")

    parsed = None
    if response_schema is not None:
        try:
            parsed = json.loads(text)
        except json.JSONDecodeError as ex:
            raise LlmInvalidOutput(f"구조화 출력 파싱 실패: {ex}") from ex

    usage_raw = payload.get("usage") or {}
    usage = Usage(
        input_tokens=int(usage_raw.get("prompt_tokens") or 0),
        output_tokens=int(usage_raw.get("completion_tokens") or 0),
    )
    return LlmResult(json=parsed, text=text, usage=usage)


def price(model: str, usage: Usage) -> float:
    inp = _PRICE_PER_1M_INPUT.get(model, 0.0) * usage.input_tokens / 1_000_000
    out = _PRICE_PER_1M_OUTPUT.get(model, 0.0) * usage.output_tokens / 1_000_000
    return round(inp + out, 4)