Batch API로 비동기 워크로드 비용 50% 절감하기

왜 Batch API인가: 트레이드오프 먼저

Batch API는 24시간 이내 완료를 보장하지만 즉각 응답은 없다. 동기 API 대비 50% 비용 절감이 핵심 이점이다. 적합한 워크로드는 대규모 문서 분류, 야간 데이터 파이프라인, A/B 평가 셋 생성이다. 부적합한 경우는 사용자 대면 챗봇, SLA 1분 이내 요구 작업이다.

실패 모드: 배치 내 일부 요청만 실패해도 전체 결과 파일에 포함되므로, 각 요청 단위로 error 필드를 반드시 확인해야 한다. 배치 자체가 만료(24시간 초과)되면 부분 결과조차 유실될 수 있다.

프로덕션 구현 패턴

import anthropic
import json
import time

client = anthropic.Anthropic()

# 1. 요청 목록 구성 (최대 10,000건 / 배치)
requests = [
    {
        "custom_id": f"doc-{i}",
        "params": {
            "model": "claude-opus-4-5",
            "max_tokens": 256,
            "messages": [{"role": "user", "content": f"Classify sentiment: {doc}"}],
        },
    }
    for i, doc in enumerate(["Great product!", "Terrible experience.", "It's okay."])
]

# 2. 배치 제출
batch = client.messages.batches.create(requests=requests)
print(f"Batch ID: {batch.id} | Status: {batch.processing_status}")

# 3. 폴링 (프로덕션에서는 웹훅 또는 별도 워커 권장)
while True:
    batch = client.messages.batches.retrieve(batch.id)
    if batch.processing_status == "ended":
        break
    time.sleep(60)  # 60초 간격 폴링

# 4. 결과 수집 및 에러 분리
results, errors = [], []
for result in client.messages.batches.results(batch.id):
    if result.result.type == "succeeded":
        results.append({"id": result.custom_id, "text": result.result.message.content[0].text})
    else:
        errors.append({"id": result.custom_id, "error": result.result.error.type})

print(f"성공: {len(results)}, 실패: {len(errors)}")

핵심 포인트: custom_id를 내부 DB PK와 1:1 매핑하면 결과 재조합이 단순해진다. 실패 항목은 동기 API로 즉시 재처리하는 폴백 로직을 붙여야 SLA를 지킬 수 있다.

비용 관측 체크리스트

배치 실행 후 batch.request_counts에서 succeeded / errored / expired 수치를 메트릭 시스템(Datadog, Grafana)으로 내보낼 것. 에러율이 5% 초과 시 알림을 설정하고 원인을 분류한다(토큰 초과, 콘텐츠 필터, 네트워크 타임아웃).

운영 체크리스트

• [ ] 배치당 요청 수 10,000건 한도 확인
• [ ] custom_id 중복 방지 (UUID 권장)
• [ ] 폴링 간격 ≥ 60초 (rate limit 회피)
• [ ] 만료 전 결과 수집 크론잡 등록 (23시간 이내)
• [ ] 실패 요청 자동 재처리 큐 구성
• [ ] 월별 배치 vs 동기 비용 대비 리포트 생성