Batch API + 비용 관측 파이프라인으로 대규모 추론 비용 절반으로 줄이기

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

Message Batches API는 동일 모델에 대해 최대 50% 할인을 제공하지만, 결과 반환까지 최대 24시간이 소요된다. 따라서 실시간 응답이 필요 없는 워크로드(데이터 레이블링, 오프라인 평가, 대규모 문서 분류)에만 적합하다. 레이턴시 SLA가 3초 이내라면 Batch는 잘못된 선택이다.

실패 모드: 배치 제출 후 polling을 하지 않으면 실패한 요청을 조용히 놓친다. errored_request_count를 반드시 확인해야 한다.

실전 구현 — 제출·폴링·비용 추적

import anthropic, time, json
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class CostTracker:
    input_tokens: int = 0
    output_tokens: int = 0
    # claude-opus-4: $15/$75 per 1M tokens (2026 Q3 기준)
    INPUT_RATE: float = 15.0 / 1_000_000
    OUTPUT_RATE: float = 75.0 / 1_000_000

    def add(self, usage: dict):
        self.input_tokens += usage.get("input_tokens", 0)
        self.output_tokens += usage.get("output_tokens", 0)

    @property
    def total_usd(self) -> float:
        return (self.input_tokens * self.INPUT_RATE +
                self.output_tokens * self.OUTPUT_RATE)

client = anthropic.Anthropic()
tracker = CostTracker()

# 1. 배치 제출
requests = [
    {"custom_id": f"doc-{i}",
     "params": {"model": "claude-opus-4-5",
                "max_tokens": 256,
                "messages": [{"role": "user",
                              "content": f"Classify sentiment: doc {i}"}]}}
    for i in range(500)
]
batch = client.beta.messages.batches.create(requests=requests)
print(f"Batch ID: {batch.id}")

# 2. 폴링 — 지수 백오프
delay = 10
while True:
    batch = client.beta.messages.batches.retrieve(batch.id)
    rc = batch.request_counts
    print(f"processing={rc.processing} succeeded={rc.succeeded} errored={rc.errored}")
    if batch.processing_status == "ended":
        break
    time.sleep(min(delay, 300))
    delay *= 1.5

# 3. 결과 수집 + 비용 집계
results = []
for result in client.beta.messages.batches.results(batch.id):
    if result.result.type == "succeeded":
        msg = result.result.message
        tracker.add({"input_tokens": msg.usage.input_tokens,
                     "output_tokens": msg.usage.output_tokens})
        results.append({"id": result.custom_id,
                        "text": msg.content[0].text})
    else:
        print(f"FAILED: {result.custom_id} — {result.result.error}")

print(f"총 비용: ${tracker.total_usd:.4f} | 성공: {len(results)}/500")

운영 체크리스트

• [ ] errored_request_count > 0 시 Slack/PagerDuty 알림 연동
• [ ] 배치 ID를 DB에 저장해 중복 제출 방지 (idempotency key)
• [ ] 결과 스트리밍 중 중단 시 커서 위치 저장 후 재개 가능하게 설계
• [ ] 토큰 비용을 요청 단위로 BigQuery/Datadog에 적재해 서비스별 chargeback 구현
• [ ] 24시간 초과 시 expires_at 필드 체크 후 자동 재제출 로직 추가
• [ ] 배치 사이즈는 10,000 요청 / 32MB 제한 내에서 분할 처리