Anthropic Claude 429 레이트리밋 재시도 설계법

import os
import time
import random
import threading
from dataclasses import dataclass

# 예: anthropic SDK를 쓴다고 가정
# pip install anthropic
from anthropic import Anthropic

client = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])

# 프로세스 내 동시성 제한 (파드/프로세스 단위)
CLAUDE_MAX_CONCURRENCY = int(os.getenv("CLAUDE_MAX_CONCURRENCY", "8"))
_sema = threading.Semaphore(CLAUDE_MAX_CONCURRENCY)

@dataclass
class RetryPolicy:
    max_attempts: int = 6
    base_delay_s: float = 0.3
    max_delay_s: float = 12.0
    factor: float = 2.0
    total_deadline_s: float = 8.0

class RateLimitError(Exception):
    pass

class RetryBudgetExceeded(Exception):
    pass


def _full_jitter_delay(cap: float) -> float:
    # AWS architecture blog에서 자주 권장하는 full jitter
    return random.uniform(0, cap)


def call_claude_with_retry(prompt: str, policy: RetryPolicy = RetryPolicy()):
    start = time.monotonic()
    attempt = 0
    cap = policy.base_delay_s

    while True:
        attempt += 1
        elapsed = time.monotonic() - start
        if elapsed > policy.total_deadline_s:
            raise RetryBudgetExceeded(f"deadline exceeded after {elapsed:.2f}s")

        with _sema:
            try:
                # 예시: Messages API 형태(개념)
                resp = client.messages.create(
                    model="claude-3-5-sonnet-latest",
                    max_tokens=512,
                    temperature=0.2,
                    messages=[{"role": "user", "content": prompt}],
                )
                return resp

            except Exception as e:
                # 실제로는 SDK가 제공하는 예외 타입(예: RateLimitError 등)로 분기 권장
                msg = str(e).lower()
                is_429 = "429" in msg or "rate limit" in msg
                is_retryable_5xx = any(code in msg for code in ["500", "502", "503", "504"])

                if not (is_429 or is_retryable_5xx):
                    raise

                if attempt >= policy.max_attempts:
                    raise RateLimitError(f"max attempts reached: {attempt}") from e

        # 세마포어 밖에서 대기(중요: 대기 중에 동시성 슬롯을 점유하지 않음)
        cap = min(policy.max_delay_s, cap * policy.factor)
        sleep_s = _full_jitter_delay(cap)

        # deadline을 넘기지 않도록 마지막에 클램프
        remaining = policy.total_deadline_s - (time.monotonic() - start)
        if remaining <= 0:
            raise RetryBudgetExceeded("no remaining budget")
        time.sleep(min(sleep_s, remaining))

type Task<T> = () => Promise<T>;

class ConcurrencyQueue {
  private running = 0;
  private queue: Array<() => void> = [];

  constructor(private readonly concurrency: number) {}

  async run<T>(task: Task<T>): Promise<T> {
    if (this.running >= this.concurrency) {
      await new Promise<void>((resolve) => this.queue.push(resolve));
    }

    this.running++;
    try {
      return await task();
    } finally {
      this.running--;
      const next = this.queue.shift();
      if (next) next();
    }
  }
}

function sleep(ms: number) {
  return new Promise((r) => setTimeout(r, ms));
}

function fullJitterDelay(capMs: number) {
  return Math.floor(Math.random() * capMs);
}

async function withRetry<T>(fn: () => Promise<T>) {
  const maxAttempts = 6;
  const deadlineMs = 8000;
  const baseMs = 300;
  const maxMs = 12000;
  let capMs = baseMs;

  const start = Date.now();

  for (let attempt = 1; attempt <= maxAttempts; attempt++) {
    try {
      return await fn();
    } catch (err: any) {
      const msg = String(err?.message ?? err).toLowerCase();
      const is429 = msg.includes("429") || msg.includes("rate limit");
      const is5xx = ["500", "502", "503", "504"].some((c) => msg.includes(c));

      if (!(is429 || is5xx)) throw err;
      if (attempt === maxAttempts) throw err;

      capMs = Math.min(maxMs, capMs * 2);
      const delay = fullJitterDelay(capMs);
      const remaining = deadlineMs - (Date.now() - start);
      if (remaining <= 0) throw err;
      await sleep(Math.min(delay, remaining));
    }
  }

  throw new Error("unreachable");
}

// 사용 예: Claude 호출을 큐로 감싸기
const q = new ConcurrencyQueue(Number(process.env.CLAUDE_MAX_CONCURRENCY ?? 8));

async function callClaude(prompt: string) {
  return q.run(() =>
    withRetry(async () => {
      // 여기에 실제 Claude SDK 호출을 넣는다.
      // 예: await anthropic.messages.create(...)
      return { ok: true, prompt };
    })
  );
}