"""OpenAI API 클라이언트 — urllib 기반 구현 + 자동 DB 로깅.""" import json import socket import time import urllib.error import urllib.request from typing import Any, Dict, List from app.core.config import settings from app.modules.llm.image_tracer import traced_call from app.modules.llm.base import BaseLLMClient from app.modules.llm.llm_logger import log_llm_call OPENAI_API_URL = "https://api.openai.com/v1/responses" def _response_text_from_payload(payload: Dict[str, Any]) -> str: """Extract text from OpenAI Responses API payload.""" output_text = payload.get("output_text") if isinstance(output_text, str) and output_text.strip(): return output_text output = payload.get("output") if isinstance(output, list): for item in output: if not isinstance(item, dict): continue content = item.get("content") if not isinstance(content, list): continue for part in content: if isinstance(part, dict) and part.get("type") == "output_text": text = part.get("text") if isinstance(text, str) and text.strip(): return text raise RuntimeError("OpenAI response did not include output_text.") class OpenAIClient(BaseLLMClient): """OpenAI Responses API client using urllib.request + 자동 DB 로깅.""" def __init__( self, api_key: str | None = None, model: str | None = None, ) -> None: # ★생성 시점에 키를 복사하면 이후 슬롯 전환이 이 인스턴스에 닿지 # 않는다(Codex 재리뷰 BLOCKING-2 실측). 명시 키만 보관하고, 실제 # 키는 호출 시점에 브로커에게 묻는다. self._explicit_key = api_key self._model = model or settings.openai_model self._ctx: Dict[str, Any] = {} def set_context(self, **kwargs) -> "OpenAIClient": """로깅 컨텍스트 설정: project_id, episode_id, operation_type, step_name.""" self._ctx.update(kwargs) return self @traced_call(operation="structured_output", provider="openai", model_of=lambda self, *a, **k: getattr(self, "_model", "unknown"), prompt_of=lambda self, system_prompt, user_prompt, *a, **k: user_prompt, output_text="[structured output]") def generate_structured( self, system_prompt: str, user_prompt: str, response_schema: Dict[str, Any], schema_name: str = "response", max_tokens: int = None, _api_key: str | None = None, ) -> dict: """★기록: urllib 직접 호출이라 litellm 콜백 밖이다 (2026-08-07).""" if _api_key is None: # 슬롯을 브로커가 정하고, 키 수준 실패면 다음 슬롯으로 다시 부른다. # 명시 키(`api_key=` 인자)는 고정 계약이라 옮기지 않는다. from app.core.openai_keys import call_with_key_failover return call_with_key_failover( lambda k: self.generate_structured( system_prompt, user_prompt, response_schema, schema_name=schema_name, max_tokens=max_tokens, _api_key=k), where="openai_client.responses", fixed_key=self._explicit_key) if max_tokens is None: max_tokens = settings.llm_max_output_tokens if not _api_key: raise RuntimeError("OpenAI API key is not configured.") body: Dict[str, Any] = { "model": self._model, "instructions": system_prompt, "input": user_prompt, "temperature": 0.2, "store": False, "text": { "format": { "type": "json_schema", "name": schema_name, "strict": True, "schema": response_schema, } }, } if max_tokens is not None: body["max_output_tokens"] = max_tokens request = urllib.request.Request( OPENAI_API_URL, data=json.dumps(body).encode("utf-8"), headers={ "Authorization": f"Bearer {_api_key}", "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(request, timeout=timeout) as resp: payload = json.loads(resp.read().decode("utf-8")) break except urllib.error.HTTPError as exc: error_text = exc.read().decode("utf-8", errors="replace") last_error = RuntimeError( f"OpenAI API error {exc.code}: {error_text}") # urllib 은 provider 를 싣지 않아 브로커가 OpenAI 호출인지 # 모른다 — 표시만 붙이면 기존 판정 계약이 그대로 적용된다. from app.core.openai_keys import mark_openai_failure mark_openai_failure(last_error, status=exc.code, body=error_text) # ★키 수준 실패는 같은 키로 재시도해도 풀리지 않는다 — quota 소진이 # 429 로 올 때 죽은 키를 max_retries 번 더 태우고 backoff 까지 # 걸었다(실측 13.8초). 전환은 브로커가 판단하게 넘긴다. from app.core.openai_keys import is_key_level_failure if (exc.code in {429, 500, 502, 503, 504} and attempt <= max_retries and not is_key_level_failure(last_error)): time.sleep(2 * attempt) continue duration = int((time.monotonic() - start_ms) * 1000) log_llm_call( model_name=self._model, user_prompt=user_prompt, system_prompt=system_prompt, status="error", error_message=str(last_error), duration_ms=duration, **self._ctx, ) 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_prompt, system_prompt=system_prompt, status="error", error_message=str(last_error), duration_ms=duration, **self._ctx, ) raise RuntimeError(f"OpenAI API failed after {max_retries} retries: {exc}") from exc else: duration = int((time.monotonic() - start_ms) * 1000) log_llm_call( model_name=self._model, user_prompt=user_prompt, system_prompt=system_prompt, status="error", error_message=str(last_error), duration_ms=duration, **self._ctx, ) raise RuntimeError(f"OpenAI API failed: {last_error}") duration = int((time.monotonic() - start_ms) * 1000) text = _response_text_from_payload(payload) # 토큰 사용량 usage = payload.get("usage", {}) input_tokens = usage.get("input_tokens") output_tokens = usage.get("output_tokens") # 자동 DB 로깅 log_llm_call( model_name=self._model, system_prompt=system_prompt, user_prompt=user_prompt, output_text=text, duration_ms=duration, input_tokens=input_tokens, output_tokens=output_tokens, status="success", **self._ctx, ) try: return json.loads(text) except json.JSONDecodeError as exc: raise RuntimeError(f"Structured output was not valid JSON:\n{text}") from exc def _active_openai_key() -> str: """활성 키 슬롯 — 1차 필드를 직접 보면 보조 키가 안 쓰인다.""" from app.core.openai_keys import active_key return active_key()