Claude API バッチ処理 — API費用を最大90%削減する非同期バッチ実装パターン

{"custom_id": "req-1", "params": {"model": "claude-3-5-sonnet-20241022", "max_tokens": 100, "messages": [{"role": "user", "content": "日本語で「こんにちは」と言ってください"}]}}
{"custom_id": "req-2", "params": {"model": "claude-3-5-sonnet-20241022", "max_tokens": 100, "messages": [{"role": "user", "content": "英語で「Hello」に関する短編を書いてください"}]}}

const fs = require('fs');
 
// サンプルリクエスト群
const requests = [
  { id: 'review-1', text: 'この製品のレビューを分析してください：「とても良い製品ですが、送料が高い」' },
  { id: 'review-2', text: 'このレビュー分析：「価格は妥当だが、配送が遅い」' },
  { id: 'review-3', text: 'レビュー分析：「品質は最高。次も買う！」' },
];
 
// JSONL に変換
const jsonl = requests
  .map(req => JSON.stringify({
    custom_id: req.id,
    params: {
      model: 'claude-3-5-sonnet-20241022',
      max_tokens: 150,
      messages: [{
        role: 'user',
        content: req.text
      }]
    }
  }))
  .join('\n');
 
// ファイルに保存
fs.writeFileSync('batch_requests.jsonl', jsonl);
console.log('✅ JSONL ファイル生成完了');

const Anthropic = require('@anthropic-ai/sdk');
const fs = require('fs');
 
const client = new Anthropic({
  apiKey: process.env.ANTHROPIC_API_KEY
});
 
async function submitBatch() {
  // JSONL ファイルを読み込み
  const fileContent = fs.readFileSync('batch_requests.jsonl');
  
  // ファイルをアップロード
  const response = await client.beta.files.upload(
    { file: new File([fileContent], 'batch_requests.jsonl') },
    { headers: { 'anthropic-beta': 'files-api-2025-04-14' } }
  );
  
  const fileId = response.id;
  console.log(`📤 ファイルアップロード完了: ${fileId}`);
  
  // バッチを作成・送信
  const batch = await client.beta.batches.create(
    { input_file_id: fileId },
    { headers: { 'anthropic-beta': 'files-api-2025-04-14' } }
  );
  
  console.log(`📋 バッチ作成完了: ${batch.id}`);
  console.log(`⏳ ステータス: ${batch.processing_status}`);
  
  return batch.id;
}
 
submitBatch().catch(console.error);

async function checkBatchStatus(batchId) {
  const batch = await client.beta.batches.retrieve(batchId, {
    headers: { 'anthropic-beta': 'files-api-2025-04-14' }
  });
  
  console.log(`📊 ステータス: ${batch.processing_status}`);
  console.log(`✅ 完了: ${batch.request_counts.succeeded} 件`);
  console.log(`❌ 失敗: ${batch.request_counts.failed} 件`);
  
  // 処理中の場合は後で再確認
  if (batch.processing_status === 'in_progress') {
    console.log('⏳ まだ処理中です。1分後に再確認してください。');
    return;
  }
  
  // 結果ファイルをダウンロード
  if (batch.result_file_id) {
    const results = await client.beta.files.retrieveContent(batch.result_file_id, {
      headers: { 'anthropic-beta': 'files-api-2025-04-14' }
    });
    console.log('✅ 結果ファイルを取得しました');
  }
}
 
// バッチID を指定して確認
checkBatchStatus('batch_xxxxx').catch(console.error);

// ❌ 効率が低い：1リクエスト = 1件分析
{ custom_id: 'review-1', params: { messages: [{ role: 'user', content: 'レビューA を分析' }] } }
 
// ✅ 効率的：1リクエスト = 複数件分析
{ custom_id: 'batch-1-10', params: { messages: [{ role: 'user', content: 'レビューA〜J を分析\nレビューA: ...\nレビューB: ...' }] } }

// テスト: 10件だけバッチ送信して結果を確認
const testBatch = requests.slice(0, 10);
// → 結果がOK なら、全 1,000 件バッチを送信

import anthropic
 
client = anthropic.Anthropic()
 
batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": "wallpaper_0001.jpg",
            "params": {
                "model": "claude-haiku-4-5-20251001",
                "max_tokens": 512,
                "messages": [
                    {"role": "user", "content": "この壁紙画像の雰囲気を一語のカテゴリで分類してください: 夜空に淡い光の輪"}
                ],
            },
        },
        {
            "custom_id": "wallpaper_0002.jpg",
            "params": {
                "model": "claude-haiku-4-5-20251001",
                "max_tokens": 512,
                "messages": [
                    {"role": "user", "content": "この壁紙画像の雰囲気を一語のカテゴリで分類してください: 朝もやの中の山並み"}
                ],
            },
        },
    ]
)
 
print(f"Batch ID: {batch.id}")

import time
 
while True:
    batch = client.messages.batches.retrieve(batch.id)
    if batch.processing_status == "ended":
        break
    counts = batch.request_counts
    print(f"処理中… 成功 {counts.succeeded} / 失敗 {counts.errored} / 処理中 {counts.processing}")
    time.sleep(60)
 
print("バッチ完了")

results = {}
 
for item in client.messages.batches.results(batch.id):
    cid = item.custom_id
    if item.result.type == "succeeded":
        results[cid] = item.result.message.content[0].text
    elif item.result.type == "errored":
        # ここで握りつぶさず、再投入リストに積む
        print(f"失敗: {cid} -> {item.result.error}")
        results[cid] = None
    elif item.result.type == "expired":
        # 24時間以内に処理されなかった分
        print(f"期限切れ: {cid}")
        results[cid] = None
 
# custom_id をファイル名にしておいたので、そのまま元画像に紐づく
ok = sum(1 for v in results.values() if v is not None)
print(f"取得 {ok} / {len(results)} 件")

def chunk(items, size=2000):
    for i in range(0, len(items), size):
        yield items[i:i + size]
 
for n, group in enumerate(chunk(all_requests)):
    b = client.messages.batches.create(requests=group)
    print(f"chunk {n}: {b.id} ({len(group)}件)")