"""Gemini 텍스트 분석 클라이언트 — 멀티턴 대화 기반 REST API.""" import json import logging import socket import time import urllib.error import urllib.request from typing import Any, Dict, List, Optional from app.core.config import settings from app.modules.llm.gemini_key_pool import get_next_key from app.modules.llm.image_tracer import ( ambient_call_meta, record_provider_call, resolve_step_name) from app.modules.llm.llm_logger import log_llm_call logger = logging.getLogger(__name__) GEMINI_API_URL_TEMPLATE = ( "https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={api_key}" ) class GeminiTextClient: """Gemini REST API 클라이언트 — 멀티턴 대화 + 구조화 출력 + 자동 로깅.""" def __init__( self, api_key: str | None = None, model: str | None = None, ) -> None: self._fixed_api_key = api_key # 명시 전달 시 고정, None이면 라운드로빈 self._model = model or settings.gemini_text_model self._history: List[Dict[str, Any]] = [] # 로깅 컨텍스트 — set_context()로 설정 self._ctx: Dict[str, Any] = {} def set_context(self, **kwargs) -> "GeminiTextClient": """로깅 컨텍스트 설정: project_id, episode_id, operation_type, step_name.""" self._ctx.update(kwargs) return self def _get_api_key(self) -> str: if self._fixed_api_key: return self._fixed_api_key return get_next_key() def reset_history(self) -> None: self._history = [] def get_history(self) -> List[Dict[str, Any]]: return list(self._history) def _trace_opik(self, status: str, user_message: str, duration_ms: int, error: Optional[str] = None) -> None: """Opik 에만 보탠다 — DB 는 이 자리의 ``log_llm_call`` 이 이미 남긴다. ★이 클라이언트는 Gemini REST 직접 호출이라 **litellm 콜백이 안 닿는다** (2026-08-07). DB 에만 남고 Opik 에는 한 줄도 없었다. """ meta = dict(ambient_call_meta()) for key in ("project_id", "episode_id", "step_name"): value = self._ctx.get(key) if value is not None and key not in meta: meta[key] = value record_provider_call( step=resolve_step_name(self._ctx.get("step_name"), meta), model=self._model, prompt=user_message, status=status, duration_ms=duration_ms, meta=meta, error=error, to_db=False) def send( self, user_message: str, response_schema: Optional[Dict[str, Any]] = None, schema_name: str = "response", system_instruction: Optional[str] = None, temperature: float = 0.2, ) -> Dict[str, Any] | str: """멀티턴 대화에서 한 턴을 보내고 응답 받기.""" self._history.append({ "role": "user", "parts": [{"text": user_message}], }) body: Dict[str, Any] = { "contents": self._history, "generationConfig": { "temperature": temperature, "maxOutputTokens": settings.llm_max_output_tokens, }, } if system_instruction: body["systemInstruction"] = { "parts": [{"text": system_instruction}], } if response_schema: body["generationConfig"]["responseMimeType"] = "application/json" body["generationConfig"]["responseJsonSchema"] = response_schema api_key = self._get_api_key() url = GEMINI_API_URL_TEMPLATE.format(model=self._model, api_key=api_key) req = urllib.request.Request( url, data=json.dumps(body).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) timeout = settings.llm_timeout_text max_retries = settings.llm_max_retries last_error: Exception | None = None start_ms = time.monotonic() for attempt in range(1, max_retries + 2): try: with urllib.request.urlopen(req, timeout=timeout) as response: payload = json.loads(response.read().decode("utf-8")) break except urllib.error.HTTPError as exc: error_text = exc.read().decode("utf-8", errors="replace") last_error = RuntimeError( f"Gemini text API error {exc.code}: {error_text}" ) if exc.code in {429, 500, 502, 503, 504} and attempt <= max_retries: # 429면 다음 키로 재시도 if exc.code == 429 and not self._fixed_api_key: api_key = self._get_api_key() url = GEMINI_API_URL_TEMPLATE.format(model=self._model, api_key=api_key) req = urllib.request.Request( url, data=json.dumps(body).encode("utf-8"), headers={"Content-Type": "application/json"}, method="POST", ) time.sleep(2 * attempt) continue duration = int((time.monotonic() - start_ms) * 1000) log_llm_call( model_name=self._model, user_prompt=user_message, system_prompt=system_instruction, status="error", error_message=str(last_error), duration_ms=duration, **self._ctx, ) self._trace_opik("error", user_message, duration, error=str(last_error)) raise last_error from exc except (urllib.error.URLError, socket.timeout) as exc: last_error = exc if attempt <= max_retries: time.sleep(2 * attempt) continue duration = int((time.monotonic() - start_ms) * 1000) log_llm_call( model_name=self._model, user_prompt=user_message, system_prompt=system_instruction, status="error", error_message=str(last_error), duration_ms=duration, **self._ctx, ) self._trace_opik("error", user_message, duration, error=str(last_error)) raise RuntimeError( f"Gemini text API failed after retries: {exc}" ) from exc else: duration = int((time.monotonic() - start_ms) * 1000) log_llm_call( model_name=self._model, user_prompt=user_message, system_prompt=system_instruction, status="error", error_message=str(last_error), duration_ms=duration, **self._ctx, ) self._trace_opik("error", user_message, duration, error=str(last_error)) raise RuntimeError(f"Gemini text API failed: {last_error}") duration = int((time.monotonic() - start_ms) * 1000) text = self._extract_text(payload) self._history.append({ "role": "model", "parts": [{"text": text}], }) # 토큰 사용량 추출 usage = payload.get("usageMetadata", {}) input_tokens = usage.get("promptTokenCount") output_tokens = usage.get("candidatesTokenCount") # 자동 DB 로깅 log_llm_call( model_name=self._model, system_prompt=system_instruction, user_prompt=user_message, output_text=text, duration_ms=duration, input_tokens=input_tokens, output_tokens=output_tokens, status="success", **self._ctx, ) self._trace_opik("success", user_message, duration) if response_schema: try: return json.loads(text) except json.JSONDecodeError as exc: raise RuntimeError( f"Gemini structured output was not valid JSON:\n{text[:500]}" ) from exc return text def send_structured( self, user_message: str, response_schema: Dict[str, Any], schema_name: str = "response", system_instruction: Optional[str] = None, ) -> Dict[str, Any]: """구조화 출력 전용 편의 메서드.""" result = self.send( user_message=user_message, response_schema=response_schema, schema_name=schema_name, system_instruction=system_instruction, ) if not isinstance(result, dict): raise RuntimeError(f"Expected dict, got {type(result)}") return result @staticmethod def _extract_text(payload: Dict[str, Any]) -> str: candidates = payload.get("candidates", []) if not candidates: feedback = payload.get("promptFeedback", {}) block_reason = feedback.get("blockReason") if block_reason: raise RuntimeError(f"Gemini blocked: {block_reason}") raise RuntimeError(f"Gemini returned no candidates: {json.dumps(payload)[:500]}") content = candidates[0].get("content", {}) parts = content.get("parts", []) texts = [p.get("text", "") for p in parts if "text" in p] if not texts: raise RuntimeError(f"Gemini response has no text parts: {json.dumps(payload)[:500]}") return "".join(texts)