batch-processing

バッチ処理スキル

Anthropic の Message Batches API を活用して、大量処理ワークロードのコストを 50％削減します。

このスキルを使う場面

• 1 万件以上のドキュメントの処理
• モデルの評価とベンチマーク
• ETL とデータのエンリッチメント
• トレーニングデータの生成
• 大量コンテンツの分析
• 時間に制約のないワークロード全般

コスト削減

処理タイプ	コスト
標準 API	100%
Batch API	50%

例： 100 万トークン @ $3/100 万 = $3（標準）対 $1.50（バッチ）

バッチのライフサイクル

Created → Processing → Ended (completed/failed/expired/canceled)

• 処理時間： 最大 24 時間
• 結果の保持期間： 29 日間
• 全機能をサポート： ツール、ビジョン、システムプロンプト

基本的な実装

ステップ 1：JSONL ファイルを作成する

import json

# Each line is a complete request
requests = [
    {
        "custom_id": "request-1",
        "params": {
            "model": "claude-sonnet-5",
            "max_tokens": 1024,
            "messages": [{"role": "user", "content": "Summarize: Document 1..."}]
        }
    },
    {
        "custom_id": "request-2",
        "params": {
            "model": "claude-sonnet-5",
            "max_tokens": 1024,
            "messages": [{"role": "user", "content": "Summarize: Document 2..."}]
        }
    }
]

# Write JSONL file
with open("batch_requests.jsonl", "w") as f:
    for request in requests:
        f.write(json.dumps(request) + "\n")

ステップ 2：バッチを送信する

import anthropic

client = anthropic.Anthropic()

# Create batch from file
with open("batch_requests.jsonl", "rb") as f:
    batch = client.beta.messages.batches.create(
        requests=f
    )

print(f"Batch ID: {batch.id}")
print(f"Status: {batch.processing_status}")

ステップ 3：完了するまでポーリングする

import time

def wait_for_batch(client, batch_id, poll_interval=60):
    """Poll until batch completes"""
    while True:
        batch = client.beta.messages.batches.retrieve(batch_id)

        if batch.processing_status == "ended":
            print(f"Batch completed!")
            print(f"  Succeeded: {batch.request_counts.succeeded}")
            print(f"  Errored: {batch.request_counts.errored}")
            print(f"  Expired: {batch.request_counts.expired}")
            return batch

        print(f"Status: {batch.processing_status} - waiting...")
        time.sleep(poll_interval)

ステップ 4：結果をストリーミングする

def process_results(client, batch_id):
    """Stream and process batch results"""
    results = {}

    for result in client.beta.messages.batches.results(batch_id):
        custom_id = result.custom_id

        if result.result.type == "succeeded":
            message = result.result.message
            results[custom_id] = {
                "status": "success",
                "content": message.content[0].text,
                "usage": message.usage
            }
        elif result.result.type == "errored":
            results[custom_id] = {
                "status": "error",
                "error": result.result.error
            }
        elif result.result.type == "expired":
            results[custom_id] = {
                "status": "expired"
            }

    return results

一連のワークフロー

import anthropic
import json
import time

def run_batch_job(documents):
    """Complete batch processing workflow"""
    client = anthropic.Anthropic()

    # 1. Create requests
    requests = []
    for i, doc in enumerate(documents):
        requests.append({
            "custom_id": f"doc-{i}",
            "params": {
                "model": "claude-sonnet-5",
                "max_tokens": 1024,
                "messages": [{"role": "user", "content": f"Summarize: {doc}"}]
            }
        })

    # 2. Write JSONL
    with open("batch.jsonl", "w") as f:
        for req in requests:
            f.write(json.dumps(req) + "\n")

    # 3. Submit batch
    with open("batch.jsonl", "rb") as f:
        batch = client.beta.messages.batches.create(requests=f)

    print(f"Batch submitted: {batch.id}")

    # 4. Wait for completion
    while True:
        batch = client.beta.messages.batches.retrieve(batch.id)
        if batch.processing_status == "ended":
            break
        time.sleep(60)

    # 5. Collect results
    results = {}
    for result in client.beta.messages.batches.results(batch.id):
        if result.result.type == "succeeded":
            results[result.custom_id] = result.result.message.content[0].text

    return results

エラー処理

失敗したリクエストを再試行する

def retry_failed(client, original_batch_id):
    """Retry failed and expired requests"""
    retry_requests = []

    for result in client.beta.messages.batches.results(original_batch_id):
        if result.result.type in ["errored", "expired"]:
            # Re-create the request (you'll need to store original params)
            retry_requests.append({
                "custom_id": result.custom_id,
                "params": get_original_params(result.custom_id)  # Your implementation
            })

    if retry_requests:
        # Write and submit retry batch
        with open("retry.jsonl", "w") as f:
            for req in retry_requests:
                f.write(json.dumps(req) + "\n")

        with open("retry.jsonl", "rb") as f:
            return client.beta.messages.batches.create(requests=f)

    return None

エラーの種類

結果の種類	対応
succeeded	通常どおり処理します
errored	エラーを確認し、必要に応じて再試行します
expired	リクエストに 24 時間以上かかりました。再試行します
canceled	バッチはキャンセルされました

API リファレンス

エンドポイント

メソッド	エンドポイント	用途
POST	/v1/messages/batches	バッチを作成します
GET	/v1/messages/batches/{id}	バッチのステータスを取得します
GET	/v1/messages/batches	バッチの一覧を取得します
POST	/v1/messages/batches/{id}/cancel	バッチをキャンセルします
GET	/v1/messages/batches/{id}/results	結果をストリーミングします

リクエスト形式

{
    "custom_id": "unique-identifier",
    "params": {
        "model": "claude-sonnet-5",
        "max_tokens": 1024,
        "system": "Optional system prompt",
        "messages": [{"role": "user", "content": "..."}],
        "tools": []
    }
}

ベストプラクティス

推奨事項：

• 100 件を超えるリクエストのワークロードに使用します
• 追跡用に一意の custom_id を含めます
• 再試行ロジック用に元のパラメーターを保存します
• ポーリングでバッチのステータスを監視します
• 結果をストリーミング JSONL として処理します

非推奨事項：

• 時間に制約のあるリクエストには使用しないでください
• 即時に結果が返ることを期待しないでください
• 期限切れのリクエストへの対応を忘れないでください
• エラー結果を無視しないでください

コスト最適化のヒント

1. 類似したリクエストをバッチ処理する - 同じモデル、類似したプロンプト
2. キャッシュと組み合わせる - バッチ処理の前に静的なコンテキストをキャッシュします
3. 適切なモデルを使用する - 単純なタスクには Haiku、複雑なタスクには Sonnet
4. プロンプトを最適化する - プロンプトが短いほどコストを抑えられます

ユースケース：ドキュメント処理

# Process 10,000 documents at 50% cost
documents = load_documents("data/*.txt")  # 10K files

# Create batch with consistent format
requests = [{
    "custom_id": doc.id,
    "params": {
        "model": "claude-haiku-4-5",  # Cheapest for bulk
        "max_tokens": 512,
        "system": "Extract key information as JSON.",
        "messages": [{"role": "user", "content": doc.content}]
    }
} for doc in documents]

# Run batch job
results = run_batch_job(requests)
# Cost: ~50% of standard API!

関連項目

• llm-integration - API の基本
• prompt-caching - コンテキストをキャッシュ
• streaming - リアルタイム出力