본문 바로가기
C.W.K.
Stream
Lesson 06 of 06 · published

Files와 Batch API는 기다릴 수 있는 일을 싸게 만들어

~16 min · files-api, batch-api, async-workflows

Level 0Observer
0 XP0/64 lessons0/13 achievements
0/150 XP to next level150 XP to go0% complete

문서는 한 번 올리고 ID로 다시 써

Files API에 PDF나 이미지, 문서를 업로드하면 file_id를 받아 messages.create()에서 참조할 수 있어. 같은 자료를 여러 턴 분석할 때마다 바이트를 다시 올리지 않아도 되므로 전송 지연과 중복 처리가 줄어들어.

Batch는 기다리는 대신 약 50%를 아껴

Batch API는 서로 독립인 요청 목록을 비동기로 처리해. 완료 보장은 24시간 안이며 보통은 더 빠르고, 토큰 가격은 대략 50% 할인돼. 평가, 대량 분류, 콘텐츠 채우기처럼 사용자가 지금 답을 기다리지 않는 일에 잘 맞아.

실시간 요구부터 물어봐

대화 화면에서 즉시 스트리밍해야 한다면 Files도 Batch도 그 요구를 대신하지 못해. 반대로 내일까지 끝나면 되는 평가 100개는 Batch가 유리하고, 같은 50MB PDF를 일주일 동안 100번 참조한다면 Files가 반복 업로드를 없애 줘. 자료 재사용과 완료 기한을 따로 보고 선택해.

원칙: 실시간 응답에는 비용이 붙어. 기다릴 수 있는 작업을 비동기로 돌려 그 비용을 되찾아.

Code

PDF 한 번 업로드하고 턴마다 참조·python
from anthropic import Anthropic

client = Anthropic()

# 1. 업로드 (한 번)
uploaded = client.beta.files.upload(
    file=("contract.pdf", open("/path/to/contract.pdf", "rb"), "application/pdf"),
)
print("file id:", uploaded.id)

# 2. 후속 호출에서 file_id로 참조 (재업로드 X)
resp = client.beta.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "document", "source": {"type": "file", "file_id": uploaded.id}},
                {"type": "text", "text": "Summarize section 5."},
            ],
        }
    ],
    betas=["files-api-2025-04-14"],
)
100 분류를 반값에 batch·python
# Submit
batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": f"row-{i}",
            "params": {
                "model": "claude-haiku-4-5-20251001",
                "max_tokens": 32,
                "system": "Reply with one word: positive, negative, neutral.",
                "messages": [{"role": "user", "content": text}],
            },
        }
        for i, text in enumerate(corpus)
    ]
)
print("batch id:", batch.id, "status:", batch.processing_status)

# Poll (프로덕션이면 webhook이나 scheduled check 사용)
import time
while True:
    b = client.messages.batches.retrieve(batch.id)
    if b.processing_status == "ended":
        break
    time.sleep(30)

# 결과 stream — 요청당 한 JSON 라인
for item in client.messages.batches.results(batch.id):
    print(item.custom_id, item.result.message.content[0].text)

External links

Exercise

평가 실행이나 일일 분류, 데이터셋 표지 중 하나를 Batch API로 옮겨. 일반 호출과의 비용 차이, 24시간 보장에 견준 실제 완료 시간을 재.
Hint
대부분 한 시간 안에 끝나도 사업 흐름은 낙관값이 아니라 24시간 보장에 맞춰.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.