Batch API 비용 관측: 청구 단위 분해와 이상 감지 파이프라인

Batch API 청구 구조의 실제 분해

Batch API는 동기 API 대비 50% 할인을 제공하지만, 비용 폭발의 원인은 대부분 input_tokens가 아닌 숨겨진 항목에 있다. 실제 청구는 세 가지로 구성된다.

1. input_tokens: 시스템 프롬프트 + 유저 메시지. Prompt Caching 미적용 시 요청마다 전액 청구.
2. cache_read_input_tokens: 캐시 히트 토큰. 기본 input 대비 약 10% 요금.
3. output_tokens: 가장 예측 불가. max_tokens 설정 미비 시 응답 길이가 요청별로 10배 이상 차이 발생.

프로덕션에서 관찰된 패형: 배치 1만 건 중 상위 2%의 요청이 전체 output_tokens의 40%를 소비. max_tokens를 전역 500으로 고정하지 않고 태스크 유형별로 분리 설정해야 한다.

비용 이상 감지 파이프라인 설계

배치 완료 후 message_batches.results를 스트리밍으로 읽어 토큰 분포를 집계하고, p95 output_tokens가 임계치를 초과하면 Slack 알림을 트리거하는 패턴이 효과적이다.

import anthropic
import statistics

client = anthropic.Anthropic()

def analyze_batch_costs(batch_id: str, output_token_threshold_p95: int = 400):
    results = client.beta.messages.batches.results(batch_id)
    
    token_stats = {"input": [], "output": [], "cache_read": []}
    errors = []

    for result in results:
        if result.result.type == "succeeded":
            usage = result.result.message.usage
            token_stats["input"].append(usage.input_tokens)
            token_stats["output"].append(usage.output_tokens)
            token_stats["cache_read"].append(
                getattr(usage, "cache_read_input_tokens", 0)
            )
        else:
            errors.append(result.custom_id)

    if not token_stats["output"]:
        return

    p95_output = statistics.quantiles(token_stats["output"], n=20)[18]  # 95th
    total_cost_estimate = (
        sum(token_stats["input"]) * 0.0000015
        + sum(token_stats["cache_read"]) * 0.00000015
        + sum(token_stats["output"]) * 0.0000075
    )  # claude-3-5-sonnet 기준 달러

    report = {
        "batch_id": batch_id,
        "total_requests": len(token_stats["output"]) + len(errors),
        "error_count": len(errors),
        "p95_output_tokens": round(p95_output),
        "estimated_cost_usd": round(total_cost_estimate, 4),
        "alert": p95_output > output_token_threshold_p95,
    }
    return report

실패 모드: results() 스트림은 배치 완료 전 호출 시 BetaMessageBatchNotFoundError가 아닌 빈 이터레이터를 반환할 수 있다. batches.retrieve(batch_id).processing_status == "ended" 확인 후 호출해야 한다.

운영 체크리스트

• [ ] 태스크 유형별 max_tokens 프로파일 분리 (요약 256 / 분류 64 / 생성 512)
• [ ] 배치 완료 후 p95 output_tokens 자동 집계 및 임계치 알림
• [ ] 에러율 > 5% 시 해당 custom_id 재처리 큐 분리
• [ ] 월별 cache_read 비율 추적 (목표: input_tokens의 60% 이상)
• [ ] Prompt Caching + Batch API 동시 적용 여부 확인 (두 할인 중첩 가능)
• [ ] 배치 ID와 내부 job_id 매핑 테이블 유지 (24시간 후 결과 만료 대비)