OpenAI Responses API 500·503 대응 재시도 폴백 서킷브레이커

import random
import time
from dataclasses import dataclass

import httpx

RETRYABLE_STATUS = {500, 502, 503, 504}

@dataclass
class RetryPolicy:
    max_attempts: int = 4
    base_delay: float = 0.15        # seconds
    max_delay: float = 1.2          # seconds
    total_budget: float = 2.5       # seconds

    def backoff(self, attempt: int) -> float:
        exp = self.base_delay * (2 ** (attempt - 1))
        jitter = random.uniform(0, exp * 0.25)
        return min(self.max_delay, exp + jitter)


def call_responses_with_retry(
    api_key: str,
    payload: dict,
    policy: RetryPolicy = RetryPolicy(),
    timeout: float = 10.0,
) -> dict:
    url = "https://api.openai.com/v1/responses"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    start = time.monotonic()
    last_exc = None

    with httpx.Client(timeout=timeout) as client:
        for attempt in range(1, policy.max_attempts + 1):
            elapsed = time.monotonic() - start
            if elapsed > policy.total_budget:
                break

            try:
                r = client.post(url, headers=headers, json=payload)

                # 성공
                if 200 <= r.status_code < 300:
                    return r.json()

                # 재시도 가능한 서버 오류
                if r.status_code in RETRYABLE_STATUS:
                    retry_after = r.headers.get("retry-after")
                    if retry_after:
                        try:
                            delay = min(policy.max_delay, float(retry_after))
                        except ValueError:
                            delay = policy.backoff(attempt)
                    else:
                        delay = policy.backoff(attempt)

                    # 마지막 시도면 종료
                    if attempt == policy.max_attempts:
                        r.raise_for_status()

                    time.sleep(delay)
                    continue

                # 그 외는 즉시 실패(400/401/403/429 등)
                r.raise_for_status()

            except (httpx.TimeoutException, httpx.NetworkError) as e:
                last_exc = e
                if attempt == policy.max_attempts:
                    raise
                time.sleep(policy.backoff(attempt))

    # 타임버짓 초과 또는 반복 실패
    if last_exc:
        raise last_exc
    raise RuntimeError("Responses API retry budget exceeded")

import hashlib
import json
from typing import Optional

# 예시용 인메모리 캐시(실전은 Redis 권장)
CACHE = {}


def cache_key(payload: dict) -> str:
    raw = json.dumps(payload, sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(raw.encode("utf-8")).hexdigest()


def get_cached_response(key: str) -> Optional[dict]:
    return CACHE.get(key)


def set_cached_response(key: str, value: dict) -> None:
    CACHE[key] = value


def create_payload(model: str, user_text: str) -> dict:
    return {
        "model": model,
        "input": user_text,
        # 필요 시 response_format, tools 등 추가
    }


def robust_generate(api_key: str, user_text: str) -> dict:
    primary = create_payload("gpt-4.1", user_text)
    fallback = create_payload("gpt-4.1-mini", user_text)

    key = cache_key(primary)

    # 1) 캐시 힌트(이전 성공 결과)
    cached = get_cached_response(key)
    if cached:
        return {"source": "cache", "data": cached}

    # 2) 1차 모델 시도
    try:
        data = call_responses_with_retry(api_key, primary)
        set_cached_response(key, data)
        return {"source": "primary", "data": data}
    except Exception:
        pass

    # 3) 폴백 모델 시도(재시도 정책은 더 짧게)
    try:
        data = call_responses_with_retry(
            api_key,
            fallback,
            policy=RetryPolicy(max_attempts=3, total_budget=1.5),
        )
        return {"source": "fallback_model", "data": data}
    except Exception:
        # 4) 최후 폴백
        return {
            "source": "degraded",
            "data": {
                "message": "현재 생성 요청이 많아 답변이 지연되고 있습니다. 잠시 후 다시 시도해 주세요.",
            },
        }

import time
from dataclasses import dataclass

@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_time: float = 15.0  # seconds

    state: str = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
    failures: int = 0
    opened_at: float = 0.0

    def allow(self) -> bool:
        if self.state == "CLOSED":
            return True
        if self.state == "OPEN":
            if time.monotonic() - self.opened_at >= self.recovery_time:
                self.state = "HALF_OPEN"
                return True
            return False
        # HALF_OPEN
        return True

    def on_success(self):
        self.failures = 0
        self.state = "CLOSED"

    def on_failure(self):
        self.failures += 1
        if self.failures >= self.failure_threshold:
            self.state = "OPEN"
            self.opened_at = time.monotonic()


cb = CircuitBreaker()


def generate_with_cb(api_key: str, payload: dict) -> dict:
    if not cb.allow():
        return {"source": "circuit_open", "data": {"message": "일시적으로 요청이 많습니다. 잠시 후 재시도해 주세요."}}

    try:
        data = call_responses_with_retry(api_key, payload)
        cb.on_success()
        return {"source": "primary", "data": data}
    except Exception:
        cb.on_failure()
        raise