Claude Batch API로 야간 문서 파이프라인의 비용을 절반으로 낮춘 이야기

월 Claude API 청구서를 보고 멈칫한 적이 있다면 이 글이 조금은 도움이 될 것 같습니다. 저도 처음에는 수천 건의 계약서 파싱을 동기 API로 처리하다가 청구액이 예상치를 훌쩍 넘어버리는 걸 경험했습니다. 근본 원인은 구조였습니다. 모든 요청을 즉시 처리해야 한다고 가정하고 있었는데, 사실 대부분의 문서 처리는 당장 답이 필요하지 않았거든요.

Claude Message Batches API는 비동기 배치 처리 인터페이스로, 표준 API 대비 50% 저렴한 비용으로 최대 10,000건의 요청을 한 번에 처리할 수 있습니다. 24시간 이내에 결과가 돌아오는 대신 가격을 절반으로 낮추는 구조입니다. 여기에 프롬프트 캐싱을 조합하면 입력 토큰 부분에서 추가 할인이 붙습니다.

이 API가 모든 상황에 맞는 것은 아닙니다. 하지만 계약서 추출, 인보이스 파싱, 야간 콘텐츠 분류처럼 오늘 밤 안에만 결과가 있으면 되는 워크로드라면 이야기가 달라집니다. 그 전제 조건과 실제 아키텍처 설계를 지금부터 풀어봅니다.

배치 API의 동작 방식

핵심 흐름

배치 API는 단순합니다. POST /v1/messages/batches로 요청 묶음을 제출하면 즉시 batch_id가 돌아옵니다. 실제 처리는 Anthropic 측에서 비동기로 진행되고, 완료를 확인하는 방법은 두 가지입니다.

여기서 짚어둘 점 하나. 웹훅은 완료 알림만 전달합니다. 실제 결과는 여전히 GET /v1/messages/batches/{id}/results로 별도 다운로드해야 합니다. 처음에는 웹훅 페이로드에 결과가 들어있을 거라 생각하기 쉬운데, 미리 알아두면 헛수고를 줄일 수 있습니다.

각 요청은 custom_id로 식별됩니다. 결과가 들어올 때 어떤 요청에 대한 응답인지 이 ID로 매핑합니다. 멀티테넌트 환경이라면 tenant_a::doc_001 형태로 접두어를 붙여두면 결과 분리가 훨씬 수월합니다.

비용 레버가 중첩되는 방식

프롬프트 캐싱과 배치 할인은 독립적으로 적용됩니다. 동일한 시스템 프롬프트나 문서 템플릿을 반복 사용하는 파이프라인이라면 두 할인이 동시에 붙습니다.

여기서 오해하기 쉬운 부분이 있습니다. 배치 50%와 캐시 히트를 단순히 곱해서 "표준 대비 5~10%"라고 말하는 사례를 자주 보는데, 캐시 할인은 입력 토큰에만 적용됩니다. 시스템 프롬프트가 수천 토큰이고 출력은 짧은 JSON 몇 백 토큰인 계약서 파싱 같은 워크로드에서는 실효 절감이 매우 크지만, 출력이 긴 워크로드에서는 그만큼 극적이지 않습니다. 자신의 입출력 비율을 먼저 계산해보는 것을 권합니다.

워크로드 적합성 판단 기준

배치 내 각 요청은 완전히 격리됩니다. 한 요청의 출력이 다음 요청의 입력이 되는 순차적 에이전트 워크플로는 배치 API로 처리할 수 없습니다. 제약이라기보다 설계 원칙이라고 이해하는 게 편합니다.

FastAPI + PostgreSQL + 배치 API 파이프라인 예시

아래 코드는 anthropic Python SDK를 사용하며, 개념적 예시로 이해해 주세요. SDK 내부 타입 경로는 버전에 따라 자주 바뀌므로, 실제 적용 시에는 자신이 설치한 버전에서 pip show anthropic으로 확인 후 import 경로를 조정하는 편이 안전합니다. 또한 Redis 클라이언트는 redis.asyncio (redis-py 4.2+)를 기준으로 작성했습니다.

요청 수락과 잡 큐 분리

사용자가 문서를 업로드하면 즉시 job_id를 반환하고, 실제 처리는 백그라운드로 넘기는 패턴입니다.

from fastapi import FastAPI
from uuid import uuid4
import json
 
app = FastAPI()
 
@app.post("/documents/process")
async def submit_document(document: DocumentRequest):
    job_id = str(uuid4())
 
    await db.execute(
        "INSERT INTO processing_jobs (job_id, status, document_content) "
        "VALUES ($1, 'pending', $2)",
        job_id, document.content
    )
 
    await redis.rpush("batch_queue", json.dumps({
        "job_id": job_id,
        "content": document.content,
        "custom_id": f"job::{job_id}"
    }))
 
    return {"job_id": job_id, "status": "queued"}

핵심은 요청을 누적한다는 점입니다. 건당 즉시 제출이 아니라, 일정 수량이 쌓이거나 일정 시간이 지나면 한 번에 배치로 제출합니다.

배치 제출 워커

비동기 환경에서는 반드시 AsyncAnthropic 클라이언트를 써야 합니다. 동기 클라이언트를 async def 안에서 호출하면 이벤트 루프가 블로킹되고, 결과 이터레이션에서 async for도 성립하지 않습니다.

import anthropic
import asyncio
import json
 
client = anthropic.AsyncAnthropic()
 
BATCH_SIZE = 500
POLL_INTERVAL_SEC = 300
 
async def drain_queue_atomically(count: int) -> list[str]:
    # LRANGE + LTRIM은 원자적이지 않아 경합 시 중복/누락 위험이 있음
    # 실전에서는 아래 Lua 스크립트 또는 BLMOVE로 원자적 처리 필요
    lua = """
    local items = redis.call('LRANGE', KEYS[1], 0, tonumber(ARGV[1]) - 1)
    if #items > 0 then
        redis.call('LTRIM', KEYS[1], #items, -1)
    end
    return items
    """
    return await redis.eval(lua, 1, "batch_queue", BATCH_SIZE)
 
async def submit_batch_worker():
    while True:
        pending = await drain_queue_atomically(BATCH_SIZE)
 
        if not pending:
            await asyncio.sleep(60)
            continue
 
        requests = []
        job_ids = []
 
        for raw in pending:
            data = json.loads(raw)
            job_ids.append(data["job_id"])
 
            requests.append({
                "custom_id": data["custom_id"],
                "params": {
                    "model": "claude-sonnet-4-6",
                    "max_tokens": 4096,
                    "system": [{
                        "type": "text",
                        "text": CONTRACT_EXTRACTION_PROMPT,
                        "cache_control": {"type": "ephemeral"}
                    }],
                    "messages": [{
                        "role": "user",
                        "content": data["content"]
                    }]
                }
            })
 
        batch = await client.beta.messages.batches.create(requests=requests)
 
        await db.executemany(
            "UPDATE processing_jobs SET batch_id = $1, status = 'submitted' "
            "WHERE job_id = $2",
            [(batch.id, jid) for jid in job_ids]
        )
 
        print(f"배치 제출 완료: {batch.id}, 요청 수: {len(requests)}")
        await asyncio.sleep(POLL_INTERVAL_SEC)

원자적 큐 드레이닝을 Lua로 처리한 점이 중요합니다. LRANGE 후 별도로 LTRIM을 부르면 그 사이에 새 항목이 들어오거나, 다른 워커가 같은 구간을 읽어 중복/누락이 발생합니다. EVAL로 한 번에 처리하거나, 리스트 대신 스트림(XADD/XREADGROUP)을 쓰는 것도 방법입니다.

시스템 프롬프트에 cache_control을 붙인 부분도 눈여겨보면 좋습니다. 동일한 시스템 프롬프트가 배치 내 수백 건에 반복 사용되면 캐시 히트가 발생해 입력 토큰 비용이 크게 줄어듭니다.

결과 폴링과 DB 반영

async def poll_batch_results():
    active_batches = await db.fetch(
        "SELECT DISTINCT batch_id FROM processing_jobs WHERE status = 'submitted'"
    )
 
    for row in active_batches:
        batch_id = row["batch_id"]
        batch = await client.beta.messages.batches.retrieve(batch_id)
 
        if batch.processing_status != "ended":
            continue
 
        async for result in await client.beta.messages.batches.results(batch_id):
            job_id = result.custom_id.replace("job::", "")
 
            if result.result.type == "succeeded":
                extracted = result.result.message.content[0].text
 
                await db.execute(
                    "UPDATE processing_jobs SET status = 'completed', result = $1 "
                    "WHERE job_id = $2",
                    extracted, job_id
                )
 
                await db.execute(
                    "SELECT pg_notify('job_completed', $1)",
                    json.dumps({"job_id": job_id})
                )
 
            elif result.result.type == "errored":
                await db.execute(
                    "UPDATE processing_jobs SET status = 'failed' WHERE job_id = $1",
                    job_id
                )

결과 스트림은 이터레이션 방식으로 소비하므로 수만 건의 결과를 한 번에 메모리에 올릴 필요가 없습니다.

대용량 출력이 필요한 경우

2026년 기준으로 배치 요청의 출력 토큰 상한을 확장하는 베타 기능이 순차 공개되고 있습니다. 다만 정확한 베타 헤더 이름이나 상한 값은 시점에 따라 달라지므로, 프로덕션 적용 전에 공식 문서의 최신 릴리즈 노트를 확인하고 실제 헤더를 그대로 사용하는 편이 안전합니다. 개념적으로는 다음과 같은 형태로 베타 헤더를 지정합니다.

# 개념적 예시 — 실제 베타 헤더 이름과 지원 상한은 공식 문서로 확인
client = anthropic.AsyncAnthropic(
    default_headers={"anthropic-beta": "<latest-long-output-beta-header>"}
)

장문 보고서 생성이나 계약서 전문 재작성처럼 128k 이상의 출력이 필요한 파이프라인에서 유용하지만, 베타 기능이므로 실제 동작을 스테이징에서 반드시 검증하고 넘어가는 것을 권합니다.

트레이드오프

장단점 한눈에

실무에서 자주 만나는 실수

1. 배치 크기를 너무 작게 잡는 경우 요청 100개짜리 배치를 수시로 제출하면 배치 API의 이점이 줄어듭니다. 가능하면 수백~수천 건 단위로 누적한 뒤 제출하는 편이 유리합니다.

2. 결과 저장을 미루는 경우 배치 결과는 29일 후 자동 삭제됩니다. 결과가 ended 상태가 되면 즉시 자체 저장소(S3, GCS, DB)에 옮기는 로직을 반드시 포함해야 합니다.

3. 에러 요청을 배치째 재시도하는 경우 배치 결과에는 성공(succeeded)과 에러(errored)가 섞여서 반환됩니다. 에러 건만 골라 재처리해야 하며, 배치 전체를 다시 제출하면 성공한 건에 비용이 이중으로 발생합니다.

4. 웹훅 도착 후 결과가 바로 있다고 가정하는 경우 웹훅은 완료 알림일 뿐이고, 결과 데이터는 별도 API 호출로 가져와야 합니다. 웹훅 핸들러에서 바로 결과를 파싱하려다 실패하는 패턴이 흔합니다.

5. 첫 워크로드 선정 실수 처음부터 사용자 대면 파이프라인을 배치로 전환하려 하면 24시간 지연 특성 때문에 UX가 무너집니다. 이미 야간에 돌아도 되는 작업(전날 쌓인 문서의 메타데이터 추출, 정기 콘텐츠 분류, 리포트 생성)부터 옮기는 것을 권합니다.

Temporal로 내구성 있는 워크플로 만들기

단순 폴링 스크립트는 서버가 재시작되면 상태를 잃습니다. 프로덕션에서는 Temporal을 활용해 크래시 이후 자동 재개와 멱등성을 확보할 수 있습니다.

각 Activity는 실패 시 자동 재시도되고, batch_id는 워크플로 상태에 영속적으로 저장되므로 워커가 죽어도 재시작 후 이어서 진행할 수 있습니다.

도입을 위한 체크리스트

배치 API를 실제로 도입한다고 결정했다면, 아래 순서로 점검하면서 옮겨가는 편이 시행착오를 줄이는 데 도움이 됩니다.

• 현재 워크로드의 입력 : 출력 토큰 비율을 측정해두었는가. 입력 비중이 클수록 캐싱 절감이 커진다.
• 대상 워크로드가 요청 간 독립성을 만족하는가. 순차 의존 관계가 있으면 배치 대신 워크플로 오케스트레이션이 필요하다.
• 사용자가 결과를 24시간 이내에만 받으면 되는 UX인가.
• 배치 제출 트리거는 크기 임계값과 시간 임계값 중 무엇을 우선할 것인가.
• 큐에서 배치로 뽑아내는 과정이 원자적인가(Redis라면 Lua/스트림, RDB라면 트랜잭션).
• ended 시점에 결과를 자체 스토리지로 즉시 이관하는 로직이 있는가(29일 삭제 정책 대응).
• errored 건만 선별 재처리하는 경로가 분리되어 있는가.
• 웹훅을 쓴다면 핸들러가 결과를 별도로 다운로드하는가.
• 워커 크래시 대비 batch_id 영속화가 되어 있는가(Temporal, DB 상태 컬럼 등).

간단한 비용 감각을 잡을 때는 다음 공식이 유용합니다.

월 예상 비용 ≈
  (월 입력 토큰 × 입력 단가 × (1 - 캐시 히트율 × 캐시 절감률) × 0.5)
+ (월 출력 토큰 × 출력 단가 × 0.5)

× 0.5가 배치 할인이고, 캐시는 입력 항에만 곱해집니다. 자신의 캐시 히트율을 로그에서 추정해 넣어보면, 이 파이프라인이 실제로 얼마나 이득인지 감이 잡힙니다.

참고 자료

• Introducing the Message Batches API | Anthropic Blog
• Batch processing | Claude Platform Docs
• Batch processing with Message Batches API | Claude Cookbook
• Claude Batch API in Practice | claudeapi.com
• Optimizing costs with Anthropic's API batching and caching | ai.moda
• Using Anthropic's Message Batches API with Temporal | Steve Kinney
• Anthropic Batch API in Production | Dotzlaw Consulting
• Anthropic API Pricing in 2026 | finout.io
• Claude Cost Optimization 2026 | pecollective.com
• Anthropic Batch API for Asynchronous Multi-Tenant AI Processing | DEV Community