OpenAI 429 Rate Limit 해결 - 백오프·큐·배치

type RetryableError = {
  status?: number;
  headers?: Record<string, string>;
  message?: string;
};

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

function parseRetryAfterMs(headers?: Record<string, string>) {
  const v = headers?.["retry-after"];
  if (!v) return null;
  const sec = Number(v);
  if (Number.isFinite(sec)) return Math.max(0, sec * 1000);
  return null;
}

function isRetryable(status?: number) {
  return status === 429 || status === 500 || status === 502 || status === 503 || status === 504;
}

export async function withBackoff<T>(
  fn: () => Promise<T>,
  opts?: { maxRetries?: number; baseDelayMs?: number; maxDelayMs?: number }
): Promise<T> {
  const maxRetries = opts?.maxRetries ?? 5;
  const baseDelayMs = opts?.baseDelayMs ?? 500;
  const maxDelayMs = opts?.maxDelayMs ?? 10_000;

  let attempt = 0;
  while (true) {
    try {
      return await fn();
    } catch (e: any) {
      const err: RetryableError = {
        status: e?.status ?? e?.response?.status,
        headers: e?.headers ?? e?.response?.headers,
        message: e?.message,
      };

      if (!isRetryable(err.status) || attempt >= maxRetries) {
        throw e;
      }

      const retryAfterMs = parseRetryAfterMs(err.headers);
      const exp = Math.min(maxDelayMs, baseDelayMs * Math.pow(2, attempt));
      const jitter = Math.floor(Math.random() * 250); // 0~249ms
      const delay = retryAfterMs ?? exp + jitter;

      attempt += 1;
      await sleep(delay);
    }
  }
}

type Job<T> = {
  run: () => Promise<T>;
  resolve: (v: T) => void;
  reject: (e: unknown) => void;
};

export class RateLimitedQueue {
  private concurrency: number;
  private intervalMs: number;
  private running = 0;
  private q: Job<any>[] = [];
  private lastStartAt = 0;

  constructor(opts: { concurrency: number; intervalMs: number }) {
    this.concurrency = opts.concurrency;
    this.intervalMs = opts.intervalMs;
  }

  add<T>(run: () => Promise<T>): Promise<T> {
    return new Promise<T>((resolve, reject) => {
      this.q.push({ run, resolve, reject });
      this.pump();
    });
  }

  private async pump() {
    if (this.running >= this.concurrency) return;
    const job = this.q.shift();
    if (!job) return;

    const now = Date.now();
    const wait = Math.max(0, this.intervalMs - (now - this.lastStartAt));
    if (wait > 0) {
      this.q.unshift(job);
      setTimeout(() => this.pump(), wait);
      return;
    }

    this.lastStartAt = Date.now();
    this.running += 1;

    try {
      const v = await job.run();
      job.resolve(v);
    } catch (e) {
      job.reject(e);
    } finally {
      this.running -= 1;
      this.pump();
    }
  }
}

import { RateLimitedQueue } from "./RateLimitedQueue";
import { withBackoff } from "./withBackoff";

const queue = new RateLimitedQueue({ concurrency: 3, intervalMs: 400 });

async function callOpenAI(payload: any) {
  return withBackoff(async () => {
    // 여기에 OpenAI 호출 로직
    // 예: await client.responses.create(payload)
    return { ok: true };
  });
}

export function enqueueOpenAI(payload: any) {
  return queue.add(() => callOpenAI(payload));
}

import asyncio
import time
from typing import Any, List, Tuple

class MicroBatcher:
    def __init__(self, max_batch_size: int = 10, max_wait_ms: int = 80):
        self.max_batch_size = max_batch_size
        self.max_wait_ms = max_wait_ms
        self.buffer: List[Tuple[Any, asyncio.Future]] = []
        self.lock = asyncio.Lock()

    async def add(self, item: Any):
        fut = asyncio.get_event_loop().create_future()
        async with self.lock:
            self.buffer.append((item, fut))
            if len(self.buffer) == 1:
                asyncio.create_task(self._flush_soon())
            if len(self.buffer) >= self.max_batch_size:
                asyncio.create_task(self._flush())
        return await fut

    async def _flush_soon(self):
        await asyncio.sleep(self.max_wait_ms / 1000)
        await self._flush()

    async def _flush(self):
        async with self.lock:
            if not self.buffer:
                return
            batch = self.buffer[: self.max_batch_size]
            self.buffer = self.buffer[self.max_batch_size :]

        items = [x for x, _ in batch]
        futures = [f for _, f in batch]

        try:
            # 여기에 OpenAI 호출을 1번만 수행하도록 구성
            # 예: responses.create(input=items를 묶어 전달)
            results = ["ok" for _ in items]
            for fut, r in zip(futures, results):
                fut.set_result(r)
        except Exception as e:
            for fut in futures:
                fut.set_exception(e)

        # 남은 버퍼가 있으면 계속 플러시
        if self.buffer:
            asyncio.create_task(self._flush())