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

"""Gemini 호출 — 이 프로젝트에서 Gemini 로 나가는 **유일한 통로**.

사진 분석(services/external/gemini.py)도 소개문·FAQ 생성(services/external/gemini_text.py)도
전부 여기를 통한다. 두 기능이 각자 HTTP 코드를 들고 있던 시절에는 `_post` 와 `_extract_text` 가
글자 그대로 두 벌 복사돼 있었고, 타임아웃·재시도·인증 처리를 고치려면 두 곳을 다 찾아야 했다.

여기가 책임지는 것: 주소·인증 헤더·재시도·응답 파싱·토큰 집계·비용 계산.
여기가 책임지지 않는 것: 무엇을 물을지(services/prompts/), 답을 믿을지(services/grounding/).
"""
import asyncio
from dataclasses import dataclass

import httpx

from config.server_configs import external_api_config

_BASE_URL = "https://generativelanguage.googleapis.com/v1beta/models"

DEFAULT_MODEL = "gemini-3.7-flash"

# 100만 토큰당 USD. 모르는 모델은 0 으로 잡는다 — 비용을 지어내느니 0 이 낫다(로그가 이상하면 눈에 띈다).
_PRICE_PER_1M_INPUT = {"gemini-3.7-flash": 0.75, "gemini-3.6-flash": 0.75, "gemini-2.5-flash": 0.30}
_PRICE_PER_1M_OUTPUT = {"gemini-3.7-flash": 3.75, "gemini-3.6-flash": 3.75, "gemini-2.5-flash": 2.50}

# 일시적 장애만 재시도한다. 4xx 는 요청 자체가 잘못된 것이라 다시 보내도 같다.
_RETRYABLE_STATUS = {408, 429, 500, 502, 503, 504}


class GeminiError(RuntimeError):
    """Gemini 호출 실패 — 호출측은 ErrorType.GENERATOR_CALL_FAILED 로 매핑한다."""


class GeminiNotConfigured(GeminiError):
    """GEMINI_API_KEY 미설정 또는 인증 실패.

    ★ 서버 부팅은 막지 않는다 — 이 어댑터만 비활성이고 나머지 파이프라인은 돈다."""


class GeminiInvalidOutput(GeminiError):
    """응답이 기대한 모양이 아니다 — ErrorType.GENERATOR_INVALID_OUTPUT."""


@dataclass
class Usage:
    """호출 1회(또는 여러 회 합산)의 토큰 사용량. 비용 로그의 근거다."""

    input_tokens: int = 0
    output_tokens: int = 0


def is_configured() -> bool:
    """키가 있는지 — 어댑터 등록/스킵 판단용. 예외를 던지지 않는다."""
    return bool(external_api_config.gemini_api_key)


async def call(
    client: httpx.AsyncClient,
    model: str,
    body: dict,
    max_retries: int = 2,
) -> dict:
    """★ LLM 이 실제로 불리는 지점. generateContent 1회 + 지수 백오프 재시도.

    재시도는 5xx·429·타임아웃만 한다. 401/403 은 키 문제이므로 GeminiNotConfigured 로
    구분해 올린다 — 호출측이 "설정이 없어서 못 한 것"과 "불렀는데 실패한 것"을 다르게 다룬다.

    응답 본문(dict)을 그대로 돌려준다. 해석은 부르는 쪽 몫이다 — 사진 분석과 문장 생성이
    같은 응답 구조에서 서로 다른 것을 꺼내 쓰기 때문이다.
    """
    url = f"{_BASE_URL}/{model}:generateContent"
    headers = {"Content-Type": "application/json", "x-goog-api-key": external_api_config.gemini_api_key}
    last = None
    for attempt in range(max_retries + 1):
        try:
            resp = await client.post(url, json=body, headers=headers)
        except (httpx.TimeoutException, httpx.TransportError) as ex:
            last = f"{type(ex).__name__}: {ex}"
        else:
            if resp.status_code == 200:
                try:
                    return resp.json()
                except ValueError as ex:
                    raise GeminiInvalidOutput(f"JSON 이 아닌 응답: {ex}") from ex
            if resp.status_code in (401, 403):
                raise GeminiNotConfigured(f"인증 실패 status={resp.status_code} — API 키를 확인하세요")
            if resp.status_code not in _RETRYABLE_STATUS:
                raise GeminiError(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)))
    raise GeminiError(f"{max_retries + 1}회 시도 실패: {last}")


def extract_text(payload: dict) -> str:
    """응답에서 텍스트 파트만 이어붙인다.

    ★ 파트에 thoughtSignature 가 함께 실려 오므로(실호출에서 확인) text 키가 있는 것만 고른다.
      전부 이어붙이면 모델의 사고 흔적이 결과 문자열에 섞인다."""
    candidates = payload.get("candidates") or []
    if not candidates:
        raise GeminiInvalidOutput("candidates 가 비었다(안전 필터 차단 가능)")
    parts = (candidates[0].get("content") or {}).get("parts") or []
    text = "".join(p["text"] for p in parts if isinstance(p, dict) and "text" in p)
    if not text.strip():
        raise GeminiInvalidOutput(f"텍스트 파트가 없다 finishReason={candidates[0].get('finishReason')}")
    return text


def read_usage(payload: dict) -> Usage:
    """응답의 usageMetadata → Usage. 없으면 0 이다(과금 안 된 호출도 있다)."""
    meta = payload.get("usageMetadata") or {}
    return Usage(
        input_tokens=int(meta.get("promptTokenCount") or 0),
        output_tokens=int(meta.get("candidatesTokenCount") or 0),
    )


def price(model: str, usage: Usage) -> float:
    """USD. 로그에만 쓴다 — 과금 근거가 아니라 "이 잡이 얼마짜리였나"를 눈으로 보는 값이다."""
    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)