가이드: 일괄 처리

일괄 작업(상품 카탈로그 번역, 지원 티켓 분류, 수천 개의 문서에서 필드 추출)은 같은 종류의 요청을 여러 번 보냅니다. 작업 자체는 단순하고, 어려운 부분은 처리량, 실패, 비용입니다. 이 가이드는 공식 openai 패키지와 base URL https://api14.com/api/v1을 사용해 세 가지를 모두 다룹니다.

별도의 batch 엔드포인트는 없습니다. OpenAI의 Batch API는 api14에서 사용할 수 없습니다(OpenAI SDK 호환성 참고). 일괄 작업은 동시성을 제어하면서 일반 요청으로 실행합니다.

속도 제한 파악하기

제한은 API 키별 분당 요청 수(RPM)로 계산되며, 요금제에 따라 상한이 정해집니다.

자체 RPM 제한이 있는 키는 둘 중 낮은 값을 적용받습니다. GET /key로 확인하세요(키가 요금제 제한을 따르면 rate_limit.requests는 null). 잔액 및 키 정보를 참고하세요. 제한을 초과하면 Retry-After: 60과 함께 429 rate_limited가 반환됩니다.

RPM은 분당 시작되는 요청 수를 제한할 뿐, 동시에 실행되는 요청 수를 제한하지 않습니다. 요청 하나가 평균 4초 걸리고 제한이 100 RPM이라면, 동시에 약 7개만 진행 중이어도 이미 제한에 도달합니다. 시작 시각을 기준으로 요청 간격을 조절하고 동시성은 적당히 유지하세요.

간격을 조절하는 동시 실행기 (Python)

import asyncio, json, os, time
from openai import AsyncOpenAI, APIStatusError

client = AsyncOpenAI(base_url="https://api14.com/api/v1", api_key=os.environ["API14_API_KEY"], max_retries=4)

RPM = 100           # your plan or key limit
CONCURRENCY = 10
FATAL = {"daily_limit_reached", "insufficient_balance", "invalid_api_key", "model_not_allowed"}

class Pacer:
    """Spaces request starts evenly so we never exceed RPM."""
    def __init__(self, rpm):
        self.interval, self.next, self.lock = 60 / rpm, 0.0, asyncio.Lock()
    async def wait(self):
        async with self.lock:
            now = time.monotonic()
            self.next = max(self.next, now)
            delay, self.next = self.next - now, self.next + self.interval
        await asyncio.sleep(delay)

SYSTEM = "Translate the user's text to German. Keep product names and units unchanged. Reply with the translation only."

async def translate(item, pacer, sem):
    async with sem:
        await pacer.wait()
        r = await client.chat.completions.create(
            model="openai/gpt-5-mini",
            messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": item["text"]}],
            max_tokens=500,
        )
        return {"id": item["id"], "text": r.choices[0].message.content, "cost": r.usage.cost}

async def main(items, out_path="out.jsonl"):
    done = set()
    if os.path.exists(out_path):
        done = {json.loads(line)["id"] for line in open(out_path)}
    todo = [i for i in items if i["id"] not in done]

    pacer, sem = Pacer(RPM), asyncio.Semaphore(CONCURRENCY)
    with open(out_path, "a") as out:
        for task in asyncio.as_completed([translate(i, pacer, sem) for i in todo]):
            try:
                out.write(json.dumps(await task) + "\n")
            except APIStatusError as e:
                if e.code in FATAL:
                    raise SystemExit(f"Stopping batch: {e.code}")
                print("failed:", e.status_code, e.code)  # logged; rerun to retry

items = [json.loads(line) for line in open("items.jsonl")]  # [{"id": "...", "text": "..."}]
asyncio.run(main(items))

이 코드가 하는 일은 다음과 같습니다.

• 간격 조절: Pacer는 분당 최대 RPM개의 요청만 시작하고, 세마포어는 동시에 진행되는 요청 수를 제한합니다.
• 재시도: SDK는 Retry-After를 따르면서 429와 5xx를 지수 백오프로 재시도합니다(max_retries=4). 업스트림 제공자의 429도 같은 방식으로 재시도합니다.
• 치명적인 코드는 작업을 중단: daily_limit_reached와 insufficient_balance는 재시도해도 해소되지 않으므로, 시도를 낭비하는 대신 작업을 종료합니다.
• 체크포인트: 결과는 끝나는 대로 JSONL 파일에 추가됩니다. 스크립트를 다시 실행하면 이미 끝난 id는 건너뛰므로, 비정상 종료나 중단이 있어도 잃는 것이 없습니다.

Node에서는 공유 큐에서 작업을 가져오는 고정된 수의 워커에 Promise.all을 사용하는 같은 구조로 구현할 수 있으며, 각 워커는 시작 사이에 60_000 / RPM ms를 기다립니다.

여러 항목을 한 요청에 담기

짧은 항목은 항목마다 요청을 하나씩 보내면 RPM을 낭비하고 매번 지시문을 반복하게 됩니다. 구조화된 출력으로 요청 하나에 20~50개 항목을 분류하면 요청 수가 같은 비율로 줄어듭니다.

schema = {
    "type": "object",
    "properties": {"labels": {"type": "array", "items": {
        "type": "object",
        "properties": {"id": {"type": "string"}, "label": {"type": "string", "enum": ["billing", "bug", "feature", "other"]}},
        "required": ["id", "label"], "additionalProperties": False,
    }}},
    "required": ["labels"], "additionalProperties": False,
}

tickets = "\n".join(f"{t['id']}: {t['text']}" for t in batch_of_30)
r = await client.chat.completions.create(
    model="openai/gpt-5-mini",
    messages=[{"role": "system", "content": "Label each ticket."}, {"role": "user", "content": tickets}],
    response_format={"type": "json_schema", "json_schema": {"name": "labels", "strict": True, "schema": schema}},
)
labels = json.loads(r.choices[0].message.content)["labels"]

입력한 id가 모두 돌아왔는지 확인하고, 빠진 것은 다시 보내세요. finish_reason이 "length"라면 묶음이 max_tokens에 비해 너무 큰 것이므로 나누세요. Structured outputs를 참고하세요. 임베딩은 input 배열 하나에 최대 100개의 문자열을 담을 수 있습니다. Embeddings를 참고하세요.

프롬프트 캐싱으로 비용 줄이기

일괄 작업에서는 지시문, 예시, 스키마가 모든 요청에서 동일합니다. 이들을 앞에 두고 달라지는 항목을 마지막에 두면 제공자가 캐시된 접두부를 재사용할 수 있습니다.

• OpenAI, DeepSeek, Gemini 2.5 등은 접두부가 충분히 길면(OpenAI는 1,024 토큰) 자동으로 캐시합니다. 캐시 읽기 비용은 일반 입력의 일부에 불과합니다.
• Anthropic 모델은 프롬프트의 고정 부분에 명시적인 cache_control 중단점이 필요합니다.

따라서 몇 가지 예시를 담은 길고 자세한 시스템 프롬프트도 짧은 프롬프트와 비용이 거의 비슷하면서 품질은 대개 더 좋아집니다. 적중 여부는 usage.prompt_tokens_details.cached_tokens로 확인하세요. 프롬프트 캐싱을 참고하세요.

저렴한 모델을 고르고 검증하기

번역, 분류, 추출에는 작은 모델이 훨씬 낮은 가격으로 충분한 경우가 많습니다. 텍스트 모델을 입력 가격순으로 나열합니다.

curl -s "https://api14.com/api/v1/models?category=text" \
  | jq -r '.data | sort_by(.pricing.prompt | tonumber) | .[:15][] | "\(.id)  \(.pricing.prompt)  \(.pricing.completion)"'

확정하기 전에 측정하세요.

1. 대표적인 항목 50~100개를 골라 기대 출력을 직접 라벨링합니다.
2. 후보 모델 두세 개에 실행해 정확도와 usage.cost 합계를 비교합니다.
3. 항목당 비용에 전체 항목 수를 곱해 전체 작업 비용을 추정하고, GET /credits로 잔액이 충분한지 확인합니다.

:free 변형은 모델의 무료 엔드포인트가 있으면 이를 사용하지만 업스트림 속도 제한이 더 엄격해서, 대규모 일괄 작업보다는 테스트에 적합합니다. 모델 id와 변형을 참고하세요.

대규모 실행 전에

• 버그로 잔액이 바닥나는 일이 없도록 일일 USD 한도가 있는 전용 키를 만드세요. 한도에 도달하면 작업이 429 daily_limit_reached로 멈춥니다. 속도 제한을 참고하세요.
• 모든 요청에 max_tokens를 설정하세요.
• 먼저 항목의 1%만 실행해 출력과 비용을 확인한 뒤 나머지를 시작하세요.
• 실패한 요청(모든 4xx 또는 5xx)은 과금되지 않으므로 실패분을 다시 실행해도 안전합니다. 오류를 참고하세요.