Batch는 기다릴 수 있는 작업을 위한 별도 흐름이야

24시간 안이면 되는 일을 모아

야간 분류, 주기적 콘텐츠 채우기, 큰 평가 실행, 데이터셋 표지, 콘텐츠 감사처럼 즉답이 필요 없는 작업이 Batch API의 자리야. 완료 보장은 24시간 안이고, 기다림을 받아들이는 대신 토큰 가격을 약 50% 아낄 수 있어.

결과는 위치가 아니라 custom_id로 연결해

각 요청에 custom_id를 붙이면 결과도 그 ID를 달고 임의 순서로 돌아와. 입력 배열의 n번째 요청이 출력의 n번째일 거라고 가정하지 마. 입력 모음과 결과 스트림을 ID로 결합하는 두 자료 집합으로 다뤄.

완료 확인과 결과 반영은 반복 가능하게

작은 작업은 30~60초 간격의 확인으로도 충분하고, 운영에서는 웹훅이 제공되면 쓰거나 예약 작업으로 상태를 살펴. 완료된 Batch를 다시 읽더라도 하류 저장소에 같은 결과가 두 번 쓰이지 않게 처리 단계를 멱등하게 만들어.

원칙: Batch는 느린 Messages가 아니라 기다릴 수 있는 일을 싸게 처리하는 별도 API야. 그 차이를 설계에 드러내.

Code

Submit, poll, idempotent write·python

from anthropic import Anthropic
import time, json

client = Anthropic()

batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": row["id"],
            "params": {
                "model": "claude-haiku-4-5-20251001",
                "max_tokens": 64,
                "system": "Reply with one word: positive, negative, or neutral.",
                "messages": [{"role": "user", "content": row["text"]}],
            },
        }
        for row in rows
    ]
)

while True:
    b = client.messages.batches.retrieve(batch.id)
    if b.processing_status == "ended":
        break
    time.sleep(60)

# Idempotent write: custom_id로 upsert
for item in client.messages.batches.results(batch.id):
    label = item.result.message.content[0].text.strip()
    db.upsert("sentiment", {"id": item.custom_id, "label": label})

Per-row 에러 처리·python

for item in client.messages.batches.results(batch.id):
    if item.result.type == "succeeded":
        save(item.custom_id, item.result.message)
    elif item.result.type == "errored":
        log.warning("batch row failed", id=item.custom_id, error=item.result.error)
        retry_queue.add(item.custom_id)

24시간 안이면 되는 일을 모아

결과는 위치가 아니라 custom_id로 연결해

완료 확인과 결과 반영은 반복 가능하게

Code

External links

Exercise

Progress

댓글 0