사용자가 화면을 보며 기다리나요? 동기 API를 쓰세요. 사용자가 제출하고 다른 일로 넘어가나요? 배치 API를 쓰세요. 이 한 줄 판단으로 Claude API 비용이 절반이 됩니다. 여기에 프롬프트 캐싱까지 스태킹하면 표준 대비 최대 85%까지 절감이 가능합니다. 이 글에서는 기본 사용법부터 100K 요청 청크 분할, 300K 출력 토큰 베타, 에러 재시도, 프로덕션 주의사항까지 실전 패턴을 전부 정리합니다.
핵심 요약
Message Batches API는 최대 100,000개 요청을 단일 배치로 제출하며 24시간 내에 결과를 반환합니다. 비용은 입력·출력 토큰 모두 표준 가격의 정확히 50%이며, 요청 수와 무관합니다. 10개만 보내도 50% 할인이 적용됩니다.
출력 토큰 한도는 동기 API 대비 대폭 확대됐으며, 베타 헤더로 요청당 최대 300K까지 사용할 수 있습니다. 배치 전용 rate limit이 별도로 운영되기 때문에 배치가 일반 API 한도에 영향을 주지 않습니다. 실전 사례 기준으로 782개 파일을 8개 배치로 처리했을 때 25분 만에 100% 성공률을 기록했습니다. 단, 진행 상황 실시간 추적이 불가하고 개별 취소가 불가하다는 점은 프로덕션에서 반드시 고려해야 합니다. 반복 시스템 프롬프트에 캐싱 스태킹을 적용하면 추가로 85~90% 절감이 가능합니다.
언제 배치 API를 쓰나 — 단 하나의 판단 기준
핵심 규칙은 하나입니다. 사용자가 화면을 보며 결과를 기다린다면 표준 Messages API, 사용자가 작업을 제출하고 다른 일로 넘어간다면 배치 API입니다. 아래 판단 트리로 결정하세요.
def should_use_batch_api(task: dict) -> tuple[bool, str]:
# 배치 API 금지 케이스
if task.get("user_waiting_realtime"):
return False, "사용자가 실시간 대기 중 → 표준 API"
if task.get("request_count", 0) < 100:
return False, "100개 미만 → 폴링 오버헤드가 할인보다 클 수 있음"
if task.get("max_latency_hours", 24) < 1:
return False, "1시간 이내 결과 필요 → 표준 API"
# 배치 API 최적 케이스
batch_use_cases = [
"document_processing", # 문서 대량 처리
"data_enrichment", # 데이터 보강 파이프라인
"nightly_analytics", # 야간 분석 배치
"offline_evaluation", # LLM Eval 오프라인 실행
"content_generation_queue", # 콘텐츠 생성 큐
"translation_pipeline", # 대량 번역
"classification_at_scale", # 대규모 분류
]
if task.get("type") in batch_use_cases:
return True, f"{task['type']} → 배치 API (50% 절감)"
return True, "비동기 허용 워크로드 → 배치 API 권장"
100개 미만이라면 폴링 오버헤드가 절감 효과보다 커질 수 있습니다. 1시간 이내 결과가 필요한 경우도 배치보다 asyncio 병렬 처리가 더 적합합니다.
1. 기본 사용법 — 배치 생성부터 결과 수신까지
전체 파이프라인은 배치 생성, 완료 대기(폴링), 결과 수신 세 단계로 구성됩니다. 각 단계를 함수로 분리해두면 재사용과 에러 처리가 쉬워집니다.
1단계: 배치 생성
custom_id는 나중에 결과와 매칭할 식별자입니다. 결과가 제출 순서와 다르게 돌아올 수 있기 때문에 반드시 의미 있는 ID를 붙여야 합니다.
import anthropic
from anthropic.types.beta.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.beta.messages.batch_create_params import Request
client = anthropic.Anthropic()
def create_batch(texts: list[str], model: str = "claude-sonnet-4-6") -> str:
requests = [
Request(
custom_id=f"item-{i:06d}",
params=MessageCreateParamsNonStreaming(
model=model,
max_tokens=1024,
messages=[{
"role": "user",
"content": f"다음 텍스트의 감정을 분석해줘 (positive/negative/neutral):\n\n{text}"
}]
)
)
for i, text in enumerate(texts)
]
batch = client.beta.messages.batches.create(requests=requests)
print(f"배치 생성 완료: {batch.id}")
return batch.id
2단계: 완료 대기 (폴링)
대부분 1시간 이내에 완료되지만 최대 24시간이 걸릴 수 있습니다. processing_status == "ended"가 될 때까지 주기적으로 상태를 확인합니다.
import time
def wait_for_batch(batch_id: str, poll_interval: int = 60) -> object:
while True:
batch = client.beta.messages.batches.retrieve(batch_id)
counts = batch.request_counts
total = counts.processing + counts.succeeded + counts.errored + counts.canceled + counts.expired
print(
f" 진행: {counts.succeeded}✅ {counts.errored}❌ "
f"{counts.processing}⏳ / {total}개 "
f"({counts.succeeded/total*100:.0f}%)"
)
if batch.processing_status == "ended":
return batch
time.sleep(poll_interval)
3단계: 결과 수신
custom_id를 키로 결과를 딕셔너리에 담아 반환합니다. errored 타입은 별도로 처리해야 하며, 이 항목들이 나중에 재시도 대상이 됩니다.
def get_batch_results(batch_id: str) -> dict:
results = {}
for result in client.beta.messages.batches.results(batch_id):
cid = result.custom_id
if result.result.type == "succeeded":
results[cid] = {
"status": "success",
"text": result.result.message.content[0].text,
"input_tokens": result.result.message.usage.input_tokens,
"output_tokens": result.result.message.usage.output_tokens,
}
elif result.result.type == "errored":
results[cid] = {
"status": "error",
"error": str(result.result.error),
}
return results
세 단계를 묶은 전체 파이프라인은 아래처럼 한 함수로 구성하면 됩니다.
def process_large_dataset(texts: list[str]) -> dict:
batch_id = create_batch(texts)
wait_for_batch(batch_id)
results = get_batch_results(batch_id)
success_count = sum(1 for r in results.values() if r["status"] == "success")
print(f"\n완료: {success_count}/{len(texts)} 성공")
return results
# 실행
texts = [f"고객 리뷰 {i}: ..." for i in range(10_000)]
results = process_large_dataset(texts)
2. 비용 계산 — 실제 절감 규모
배치 API는 정확히 50% 할인이며, 여기에 프롬프트 캐싱을 스태킹하면 추가 절감이 가능합니다. 아래 함수로 월 절감 규모를 계산할 수 있습니다.
STANDARD_PRICES = {
"claude-haiku-4-5": {"input": 0.80, "output": 4.00},
"claude-sonnet-4-6": {"input": 3.00, "output": 15.00},
"claude-opus-4-7": {"input": 5.00, "output": 25.00},
}
def calculate_batch_savings(
model: str,
monthly_requests: int,
avg_input_tokens: int = 500,
avg_output_tokens: int = 200,
) -> dict:
prices = STANDARD_PRICES[model]
monthly_input_tokens = monthly_requests * avg_input_tokens
monthly_output_tokens = monthly_requests * avg_output_tokens
standard_cost = (
monthly_input_tokens / 1_000_000 * prices["input"] +
monthly_output_tokens / 1_000_000 * prices["output"]
)
batch_cost = standard_cost * 0.5
# 캐싱 스태킹: 시스템 프롬프트 1000 토큰 반복 재사용 시 90% 절감
cached_portion = 1000 * monthly_requests / 1_000_000 * prices["input"]
cache_savings = cached_portion * 0.90
stacked_cost = batch_cost - cache_savings
return {
"standard_monthly": f"${standard_cost:,.0f}",
"batch_monthly": f"${batch_cost:,.0f}",
"stacked_monthly": f"${stacked_cost:,.0f}",
"savings_pct": f"{(1 - stacked_cost/standard_cost)*100:.0f}%",
}
시나리오별 결과를 보면 절감 규모가 체감됩니다. claude-sonnet-4-6으로 월 10만 건 처리 시 표준 $1,900 → 배치 $950 → 배치+캐싱 $445(77% 절감), 월 50만 건은 $9,500 → $4,750 → $2,225(77% 절감)입니다. claude-opus-4-7 월 5만 건은 $6,250 → $3,125 → $1,838(71% 절감)입니다.
3. 100,000개 요청 처리 — 청크 분할 패턴
배치 1개당 최대 100,000개 요청이 가능하지만 안전 마진을 위해 10,000개 단위로 분할하는 것이 권장됩니다. 아래 BatchProcessor 클래스는 자동으로 청크를 나누고 모든 배치의 완료를 병렬로 대기합니다.
import asyncio
from dataclasses import dataclass, field
@dataclass
class BatchProcessor:
client: anthropic.Anthropic
model: str = "claude-sonnet-4-6"
chunk_size: int = 10_000
poll_interval: int = 60
batch_ids: list[str] = field(default_factory=list)
results: dict = field(default_factory=dict)
async def submit_all(self, items: dict[str, str]) -> list[str]:
items_list = list(items.items())
chunks = [
items_list[i:i+self.chunk_size]
for i in range(0, len(items_list), self.chunk_size)
]
print(f"총 {len(items)}개 요청 → {len(chunks)}개 배치로 분할")
for chunk_idx, chunk in enumerate(chunks):
requests = [
Request(
custom_id=cid,
params=MessageCreateParamsNonStreaming(
model=self.model,
max_tokens=512,
system="당신은 데이터 분석 전문가입니다.",
messages=[{"role": "user", "content": content}]
)
)
for cid, content in chunk
]
batch = self.client.beta.messages.batches.create(requests=requests)
self.batch_ids.append(batch.id)
print(f" 배치 {chunk_idx+1}/{len(chunks)} 제출: {batch.id}")
return self.batch_ids
async def wait_all(self) -> None:
pending = set(self.batch_ids)
while pending:
for batch_id in list(pending):
batch = self.client.beta.messages.batches.retrieve(batch_id)
if batch.processing_status == "ended":
pending.discard(batch_id)
print(f" 배치 완료: {batch_id}")
if pending:
print(f" 대기 중: {len(pending)}개 배치...")
await asyncio.sleep(self.poll_interval)
def collect_results(self) -> dict:
for batch_id in self.batch_ids:
for result in self.client.beta.messages.batches.results(batch_id):
if result.result.type == "succeeded":
self.results[result.custom_id] = {
"text": result.result.message.content[0].text,
"tokens": {
"input": result.result.message.usage.input_tokens,
"output": result.result.message.usage.output_tokens,
}
}
else:
self.results[result.custom_id] = {"error": str(result.result.error)}
return self.results
10만 개 문서를 처리하는 전체 흐름은 제출 → 대기 → 수집 세 줄로 완성됩니다.
async def process_100k_documents():
client = anthropic.Anthropic()
processor = BatchProcessor(client=client, chunk_size=10_000)
items = {f"doc-{i:06d}": f"문서 {i}: ..." for i in range(100_000)}
await processor.submit_all(items)
await processor.wait_all()
results = processor.collect_results()
success = sum(1 for r in results.values() if "text" in r)
print(f"최종: {success}/{len(items)} 성공")
return results
4. 300K 출력 토큰 베타 — 장문 생성 워크로드
2026년 3월에 추가된 기능으로, output-300k-2026-03-24 베타 헤더를 사용하면 배치 API에서 요청당 최대 300K 출력 토큰이 가능합니다. 동기 Messages API에서는 지원하지 않습니다. 기술 문서, 코드 스캐폴딩, 장문 보고서 생성에 적합합니다.
def create_long_form_batch(prompts: list[dict]) -> str:
"""
단일 300K 요청 완료에 1시간 이상 소요될 수 있으므로
타임라인 여유가 충분한 워크로드에만 사용하세요.
"""
requests = [
Request(
custom_id=p["id"],
params=MessageCreateParamsNonStreaming(
model="claude-opus-4-7",
max_tokens=300_000,
messages=[{"role": "user", "content": p["prompt"]}]
)
)
for p in prompts
]
batch = client.beta.messages.batches.create(
requests=requests,
betas=["output-300k-2026-03-24"] # 베타 헤더 필수
)
print(f"300K 배치 생성: {batch.id}")
return batch.id
장문 생성은 Opus를 권장합니다. 단일 요청 완료에 1시간 이상 소요될 수 있기 때문에 24시간 타임라인 여유가 있는 워크로드에만 적용하세요.
5. 에러 처리 + 재시도 패턴
배치 파이프라인이 중단됐다가 재시작될 때, 이미 완료된 배치를 다시 제출하는 낭비를 막으려면 상태 파일 기반 관리가 필요합니다. 아래 패턴은 JSON 파일에 배치 상태를 영속화하고, 실패한 항목만 새 배치로 재시도합니다.
from enum import Enum
import json
from pathlib import Path
class BatchState(Enum):
SUBMITTED = "submitted"
COMPLETED = "completed"
FAILED = "failed"
def robust_batch_pipeline(
items: dict[str, str],
state_file: str = "batch_state.json",
max_retries: int = 3,
) -> dict:
state_path = Path(state_file)
if state_path.exists():
with open(state_path) as f:
state = json.load(f)
print(f"기존 상태 로드: 배치 {len(state['batches'])}개")
else:
state = {"batches": {}, "results": {}, "retry_counts": {}}
for batch_id, batch_state in list(state["batches"].items()):
if batch_state == BatchState.SUBMITTED.value:
batch = client.beta.messages.batches.retrieve(batch_id)
if batch.processing_status == "ended":
for result in client.beta.messages.batches.results(batch_id):
if result.result.type == "succeeded":
state["results"][result.custom_id] = (
result.result.message.content[0].text
)
else:
cid = result.custom_id
retry_count = state["retry_counts"].get(cid, 0)
if retry_count < max_retries:
state["retry_counts"][cid] = retry_count + 1
print(f" 재시도 예약: {cid} (시도 {retry_count+1}/{max_retries})")
state["batches"][batch_id] = BatchState.COMPLETED.value
with open(state_path, "w") as f:
json.dump(state, f, ensure_ascii=False, indent=2)
return state["results"]
실패한 항목만 모아 새 배치로 재제출하는 함수도 별도로 두면 관리가 쉽습니다. failed_ids 리스트를 받아 새 배치 ID를 반환합니다.
def retry_failed_items(
failed_ids: list[str],
original_items: dict[str, str],
model: str = "claude-sonnet-4-6"
) -> str:
retry_items = {
cid: original_items[cid]
for cid in failed_ids
if cid in original_items
}
if not retry_items:
return None
print(f"재시도 배치 생성: {len(retry_items)}개 항목")
requests = [
Request(
custom_id=cid,
params=MessageCreateParamsNonStreaming(
model=model,
max_tokens=512,
messages=[{"role": "user", "content": content}]
)
)
for cid, content in retry_items.items()
]
batch = client.beta.messages.batches.create(requests=requests)
return batch.id
6. 캐싱 + 배치 스태킹 — 최대 절감
반복되는 시스템 프롬프트나 공유 문서가 있다면 배치 50% 할인 위에 캐싱을 스태킹해서 추가 절감이 가능합니다. 캐시 히트 시 해당 토큰 비용이 90% 절감되는 구조입니다.
핵심은 cache_control: {"type": "ephemeral"} 마커를 시스템 프롬프트와 공유 문서에 붙이는 것입니다. 개별 요청마다 달라지는 텍스트에는 캐시 마커를 붙이지 않습니다.
def create_cached_batch(
texts: list[str],
system_prompt: str,
shared_document: str = "",
) -> str:
requests = []
for i, text in enumerate(texts):
system_content = [{
"type": "text",
"text": system_prompt,
"cache_control": {"type": "ephemeral"} # 캐시 마킹
}]
user_content = []
if shared_document:
user_content.append({
"type": "text",
"text": f"참조 문서:\n{shared_document}",
"cache_control": {"type": "ephemeral"} # 공유 문서도 캐시
})
user_content.append({
"type": "text",
"text": f"분석 대상:\n{text}" # 개별 텍스트는 캐시 안 함
})
requests.append(
Request(
custom_id=f"item-{i:06d}",
params=MessageCreateParamsNonStreaming(
model="claude-sonnet-4-6",
max_tokens=512,
system=system_content,
messages=[{"role": "user", "content": user_content}]
)
)
)
batch = client.beta.messages.batches.create(requests=requests)
return batch.id
시스템 프롬프트 1,000 토큰과 공유 문서 5,000 토큰을 10,000건에 반복 사용하는 경우, 배치 전용이라면 약 $90, 배치+캐싱 스태킹이라면 약 $27까지 내려갑니다. 표준 대비 약 86% 절감입니다.
7. Webhook 대안 — 폴링 없이 완료 알림
2026년 5월 기준으로 Anthropic Batch API는 Webhook을 아직 지원하지 않습니다. 대신 세 가지 대안 패턴으로 폴링 부담을 줄일 수 있습니다.
첫 번째는 스마트 폴링입니다. 초반에는 30초 간격으로 빠르게 확인하다가 시간이 지날수록 최대 10분까지 간격을 늘리는 방식입니다. 불필요한 API 호출을 줄이면서도 완료를 빠르게 감지할 수 있습니다.
def smart_poll(batch_id: str) -> object:
intervals = [30, 60, 120, 300, 600] # 30초 → 10분
attempt = 0
while True:
batch = client.beta.messages.batches.retrieve(batch_id)
if batch.processing_status == "ended":
return batch
interval = intervals[min(attempt, len(intervals)-1)]
print(f"대기 중... {interval}초 후 재확인")
time.sleep(interval)
attempt += 1
두 번째는 Temporal 같은 워크플로우 엔진을 활용하는 패턴입니다. 서버가 재시작돼도 폴링 상태가 영속화되기 때문에 프로덕션 환경에서 안정적입니다. 세 번째는 배치 ID를 DB에 저장하고 크론잡이 주기적으로 미완료 배치를 확인하는 패턴입니다. 특별한 인프라 없이도 구현할 수 있습니다.
def submit_and_store(items: dict, db_connection) -> None:
batch_id = create_batch(list(items.values()))
db_connection.execute("""
INSERT INTO batch_jobs (batch_id, status, created_at, item_count)
VALUES (?, 'submitted', datetime('now'), ?)
""", (batch_id, len(items)))
print(f"배치 제출 완료. ID 저장됨: {batch_id}")
# 크론잡이 30분마다 미완료 배치 확인
8. 실전 주의사항 — 프로덕션에서 배운 것
프로덕션에서 Batch API를 사용할 때 놓치기 쉬운 함정을 정리했습니다.
완료 시간 불확실성은 가장 중요한 주의사항입니다. 대부분 1시간 이내에 완료되지만 최대 24시간까지 걸릴 수 있습니다. 실제 프로덕션 경험에서 4시간 이상 소요된 사례가 있었습니다. 타임라인이 확실한 워크로드에만 사용하세요.
개별 항목 진행 추적이 불가합니다. 배치 전체의 성공·실패 수만 확인할 수 있으며, 어떤 항목이 실패했는지는 배치가 완료된 후에만 custom_id로 확인할 수 있습니다.
개별 취소가 불가합니다. 제출된 배치는 요청 단위 취소가 없으며 배치 전체만 취소할 수 있습니다(batches.cancel(batch_id)).
결과는 29일 후 자동 삭제됩니다. 완료 즉시 결과를 수집해서 자체 저장소에 보관해야 합니다.
실시간 UI에 절대 사용하면 안 됩니다. 사용자가 기다리는 시나리오에 배치를 쓰는 것은 최악의 UX입니다. "제출 완료, 결과는 이메일로 드립니다" 패턴이 맞는 구조입니다.
실시간성이 필요하다면 asyncio 병렬 처리가 대안입니다. 50% 할인은 없지만 실시간 진행 상황 추적, 개별 재시도, 즉시 결과 확인이 가능합니다.
async def parallel_process(items: list[str], concurrency: int = 50) -> list:
semaphore = asyncio.Semaphore(concurrency)
async def process_one(item: str) -> str:
async with semaphore:
async_client = anthropic.AsyncAnthropic()
response = await async_client.messages.create(
model="claude-sonnet-4-6",
max_tokens=512,
messages=[{"role": "user", "content": item}]
)
return response.content[0].text
return await asyncio.gather(*[process_one(item) for item in items])
9. 워크로드별 최적 선택
어떤 워크로드에 무엇을 써야 할지 정리했습니다.
배치 API가 최적인 경우는 월 10만 건 이상 문서 처리, LLM Eval 오프라인 실행, 야간 데이터 보강 크론잡, 콘텐츠 생성 대기열처럼 실시간 응답이 필요 없고 비용이 중요한 워크로드입니다. 장문 기술 문서 생성이라면 배치 + 300K 베타 조합이 유일한 선택입니다.
반면 사용자가 대기하는 채팅, 빠른 프로토타입(100건 미만), 진행 상황 UI가 필요한 경우, 취소 기능이 필요한 경우에는 동기 API 또는 asyncio 병렬 처리가 적합합니다.
결론
배치 API 도입 여부를 판단하는 기준은 하나입니다. 비동기를 허용할 수 있고 100건 이상이라면, 이유를 막론하고 50% 할인이 적용됩니다. 여기에 반복 시스템 프롬프트가 있다면 캐싱을 스태킹해서 최대 85%까지 절감할 수 있습니다. 장문 콘텐츠 생성이라면 동기 API에서는 불가능한 300K 출력 토큰이 배치에서만 가능합니다.
지금 바로 적용할 수 있는 워크로드는 야간 분석 크론잡, LLM Eval 오프라인 실행, 대량 문서·피드백 분류입니다. 표준 API를 쓰고 있다면 코드 변경 없이 배치로 전환하는 것만으로 즉시 비용이 절반이 됩니다.
반드시 기억해야 할 것이 세 가지 있습니다. 실시간 UI에 배치를 쓰는 것은 절대 안 됩니다. 결과는 29일 후 자동 삭제되기 때문에 완료 즉시 자체 저장소에 보관해야 합니다. 그리고 4시간 이상 완료 시간이 걸린 경험이 실제로 있기 때문에, 24시간 여유가 없는 워크로드라면 asyncio 병렬 처리를 대안으로 고려해야 합니다.
관련 글
- Claude Code Hooks 완전가이드 — 프롬프트 요청이 아닌 보장된 실행
- Claude Opus 4.8 Fast Mode 완전 분석 — 2.5배 빠르고 3배 싸다는 게 실제로 맞는가
'Claude' 카테고리의 다른 글
| Claude Opus 4.8 Mid-conversation System Messages 실전 — 에이전트 루프 중간에 지시 바꾸는 법 (0) | 2026.06.01 |
|---|---|
| Claude Code Dynamic Workflows 실전 — 병렬 서브에이전트로 대규모 리팩토링 하는 법 (0) | 2026.06.01 |
| 일본 정부 + 3대 메가뱅크 Claude Mythos 도입 — 왜 하필 일본이 첫 번째 비(非)영미권 파트너인가 (1) | 2026.05.29 |
| Anthropic이 월스트리트를 노린다 — 10개 금융 에이전트 + $1.5B JV, 무엇이 바뀌나 (0) | 2026.05.28 |
| Anthropic이 공개를 거부한 AI — Claude Mythos 완전 분석 (0) | 2026.05.27 |