Claude API Messages Batches 実装ガイド：非同期バッチ処理でコストを最大50%削減

import anthropic
import json
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
# バッチに含めるリクエストを準備
# 各リクエストには一意の custom_id が必要
requests = []
documents = [
    {"id": "doc_001", "content": "AIの倫理に関する最新の議論について述べる文書..."},
    {"id": "doc_002", "content": "機械学習の基礎と応用に関する技術論文..."},
    {"id": "doc_003", "content": "スタートアップの資金調達戦略についての解説..."},
]
 
for doc in documents:
    requests.append(
        anthropic.types.message_create_params.MessageCreateParamsNonStreaming(
            model="claude-sonnet-4-6",
            max_tokens=500,
            messages=[
                {
                    "role": "user",
                    "content": f"以下の文書を3行で要約してください。\n\n{doc['content']}"
                }
            ],
            # custom_id: 結果と元データを紐付けるためのID（必須）
            # 最大64文字の英数字・ハイフン・アンダースコア
        )
    )
    # custom_id は別途 BatchMessageParam 形式で指定
 
# 正しい形式でリクエストを構築
batch_requests = []
for doc in documents:
    batch_requests.append({
        "custom_id": doc["id"],  # 後で結果と紐付けるためのID
        "params": {
            "model": "claude-sonnet-4-6",
            "max_tokens": 500,
            "messages": [
                {
                    "role": "user",
                    "content": f"以下の文書を3行で要約してください。\n\n{doc['content']}"
                }
            ]
        }
    })
 
# バッチを作成
batch = client.beta.messages.batches.create(
    requests=batch_requests
)
 
print(f"バッチID: {batch.id}")
print(f"ステータス: {batch.processing_status}")
print(f"リクエスト数: {batch.request_counts.processing}")
# 出力例:
# バッチID: msgbatch_01HxxxxxxxxxxxxxxxxxxXXXX
# ステータス: in_progress
# リクエスト数: 3

import time
import anthropic
 
def wait_for_batch_completion(
    client: anthropic.Anthropic,
    batch_id: str,
    poll_interval_seconds: int = 60,
    max_wait_minutes: int = 120,
) -> anthropic.types.beta.BetaMessageBatch:
    """
    バッチ処理の完了を待機する関数。
 
    Args:
        client: Anthropic クライアント
        batch_id: 待機するバッチのID
        poll_interval_seconds: ポーリング間隔（秒）
        max_wait_minutes: 最大待機時間（分）
 
    Returns:
        完了したバッチオブジェクト
 
    Raises:
        TimeoutError: 最大待機時間を超過した場合
        RuntimeError: バッチが失敗した場合
    """
    max_attempts = (max_wait_minutes * 60) // poll_interval_seconds
 
    for attempt in range(max_attempts):
        batch = client.beta.messages.batches.retrieve(batch_id)
 
        status = batch.processing_status
 
        if status == "ended":
            # 処理完了（成功・失敗含む）
            print(f"バッチ処理完了: {batch_id}")
            print(f"  成功: {batch.request_counts.succeeded}")
            print(f"  失敗: {batch.request_counts.errored}")
            print(f"  期限切れ: {batch.request_counts.expired}")
            return batch
 
        elif status == "canceling":
            raise RuntimeError(f"バッチがキャンセルされています: {batch_id}")
 
        else:
            # in_progress の場合は継続待機
            processed = batch.request_counts.processing
            total = (
                batch.request_counts.processing +
                batch.request_counts.succeeded +
                batch.request_counts.errored
            )
            print(
                f"[{attempt + 1}/{max_attempts}] "
                f"処理中... ({processed}/{total} 件残り)"
            )
            time.sleep(poll_interval_seconds)
 
    raise TimeoutError(
        f"バッチ処理が {max_wait_minutes} 分以内に完了しませんでした: {batch_id}"
    )

import random
 
def wait_for_batch_with_backoff(
    client: anthropic.Anthropic,
    batch_id: str,
    initial_interval: float = 30.0,
    max_interval: float = 300.0,  # 最大5分
    multiplier: float = 1.5,
    jitter: bool = True,
) -> anthropic.types.beta.BetaMessageBatch:
    """
    指数バックオフによるポーリング実装。
    APIレート制限への配慮とリソース効率化に有効。
    """
    interval = initial_interval
 
    while True:
        batch = client.beta.messages.batches.retrieve(batch_id)
 
        if batch.processing_status == "ended":
            return batch
 
        # ジッターを加えてサンダリングハード問題を防ぐ
        wait_time = interval
        if jitter:
            wait_time = interval * (0.5 + random.random())
 
        print(f"次のポーリングまで {wait_time:.1f} 秒待機...")
        time.sleep(wait_time)
 
        # 次回の待機時間を計算（上限を設ける）
        interval = min(interval * multiplier, max_interval)

def process_batch_results(
    client: anthropic.Anthropic,
    batch_id: str,
    original_documents: dict[str, dict],
) -> list[dict]:
    """
    バッチ処理結果を取得し、元データと紐付けて返す。
 
    Args:
        client: Anthropic クライアント
        batch_id: 処理済みバッチのID
        original_documents: custom_id -> ドキュメントデータ の辞書
 
    Returns:
        処理結果のリスト（成功・失敗情報を含む）
    """
    results = []
 
    # 結果をストリーミングで取得（大量データでもメモリ効率よく処理）
    for result in client.beta.messages.batches.results(batch_id):
        custom_id = result.custom_id
        original_doc = original_documents.get(custom_id, {})
 
        if result.result.type == "succeeded":
            # 成功したリクエスト
            message = result.result.message
            content_text = ""
 
            for block in message.content:
                if block.type == "text":
                    content_text = block.text
                    break
 
            results.append({
                "id": custom_id,
                "status": "success",
                "original": original_doc,
                "summary": content_text,
                "input_tokens": message.usage.input_tokens,
                "output_tokens": message.usage.output_tokens,
            })
 
        elif result.result.type == "errored":
            # エラーが発生したリクエスト
            error = result.result.error
            results.append({
                "id": custom_id,
                "status": "error",
                "original": original_doc,
                "error_type": error.type,
                "error_message": str(error),
            })
 
        elif result.result.type == "expired":
            # 24時間以内に処理されなかったリクエスト
            results.append({
                "id": custom_id,
                "status": "expired",
                "original": original_doc,
            })
 
    return results
 
# 実際の使用例
documents_map = {doc["id"]: doc for doc in documents}
completed_batch = wait_for_batch_completion(client, batch.id)
results = process_batch_results(client, batch.id, documents_map)
 
# 結果をJSONファイルに保存
with open("batch_results.json", "w", encoding="utf-8") as f:
    json.dump(results, f, ensure_ascii=False, indent=2)
 
# 統計レポート
success_count = sum(1 for r in results if r["status"] == "success")
error_count = sum(1 for r in results if r["status"] == "error")
print(f"処理完了: 成功 {success_count}件 / 失敗 {error_count}件")

import asyncio
import logging
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Callable
import anthropic
 
logger = logging.getLogger(__name__)
 
@dataclass
class BatchJob:
    """バッチジョブの管理クラス"""
    job_id: str
    batch_id: str | None = None
    status: str = "pending"  # pending / submitted / completed / failed
    submitted_at: datetime | None = None
    completed_at: datetime | None = None
    request_count: int = 0
    success_count: int = 0
    error_count: int = 0
    metadata: dict = field(default_factory=dict)
 
class BatchJobManager:
    """
    本番環境向けバッチジョブ管理クラス。
 
    特徴:
    - ジョブの永続化（DB/ファイル対応可能）
    - 自動リトライ（失敗リクエストの再送）
    - 進捗トラッキング
    - コスト計算
    """
 
    def __init__(
        self,
        client: anthropic.Anthropic,
        max_requests_per_batch: int = 5000,  # 安全マージンとして10,000の半分
    ):
        self.client = client
        self.max_requests_per_batch = max_requests_per_batch
        self.jobs: dict[str, BatchJob] = {}
 
    def create_batches_from_requests(
        self,
        all_requests: list[dict],
        job_id_prefix: str = "job",
    ) -> list[BatchJob]:
        """
        大量のリクエストを複数バッチに分割して送信。
        10,000件上限を超える場合に自動分割する。
        """
        jobs = []
 
        # リクエストをバッチサイズで分割
        for i in range(0, len(all_requests), self.max_requests_per_batch):
            chunk = all_requests[i:i + self.max_requests_per_batch]
            job_id = f"{job_id_prefix}_{i // self.max_requests_per_batch:04d}"
 
            try:
                logger.info(f"バッチ送信中: {job_id} ({len(chunk)} 件)")
 
                batch = self.client.beta.messages.batches.create(
                    requests=chunk
                )
 
                job = BatchJob(
                    job_id=job_id,
                    batch_id=batch.id,
                    status="submitted",
                    submitted_at=datetime.now(),
                    request_count=len(chunk),
                )
 
                self.jobs[job_id] = job
                jobs.append(job)
 
                logger.info(f"バッチ作成成功: {batch.id}")
 
            except anthropic.APIError as e:
                logger.error(f"バッチ作成失敗: {job_id} - {e}")
                job = BatchJob(
                    job_id=job_id,
                    status="failed",
                    metadata={"error": str(e)},
                )
                self.jobs[job_id] = job
                jobs.append(job)
 
        return jobs
 
    def calculate_cost_estimate(
        self,
        results: list[dict],
        model: str = "claude-sonnet-4-6",
    ) -> dict:
        """
        処理結果からコストを計算する。
        Batches API の50%割引を適用。
        """
        # モデルごとの料金設定（$/MTok）
        pricing = {
            "claude-sonnet-4-6": {"input": 1.50, "output": 7.50},  # Batches割引後
            "claude-haiku-4-5-20251001": {"input": 0.40, "output": 2.00},  # Batches割引後
            "claude-opus-4": {"input": 7.50, "output": 37.50},  # Batches割引後
        }
 
        model_pricing = pricing.get(model, pricing["claude-sonnet-4-6"])
 
        total_input_tokens = sum(
            r.get("input_tokens", 0)
            for r in results
            if r["status"] == "success"
        )
        total_output_tokens = sum(
            r.get("output_tokens", 0)
            for r in results
            if r["status"] == "success"
        )
 
        input_cost = (total_input_tokens / 1_000_000) * model_pricing["input"]
        output_cost = (total_output_tokens / 1_000_000) * model_pricing["output"]
 
        return {
            "total_input_tokens": total_input_tokens,
            "total_output_tokens": total_output_tokens,
            "input_cost_usd": round(input_cost, 4),
            "output_cost_usd": round(output_cost, 4),
            "total_cost_usd": round(input_cost + output_cost, 4),
            "model": model,
        }

def create_product_description_requests(products: list[dict]) -> list[dict]:
    """商品データからバッチリクエストを生成"""
    requests = []
 
    for product in products:
        prompt = f"""以下の商品情報から、SEOに最適化された魅力的な商品説明文を生成してください。
 
商品名: {product['name']}
カテゴリ: {product['category']}
スペック: {json.dumps(product['specs'], ensure_ascii=False)}
価格: ¥{product['price']:,}
 
要件:
- 150〜200文字
- 特徴・メリットを自然な言葉で表現
- ターゲット顧客が共感できる表現
- 購買意欲を高めるCTA的な締め"""
 
        requests.append({
            "custom_id": f"product_{product['id']}",
            "params": {
                "model": "claude-haiku-4-5-20251001",  # 大量処理にはHaikuが経済的
                "max_tokens": 300,
                "messages": [{"role": "user", "content": prompt}]
            }
        })
 
    return requests

def create_sentiment_analysis_requests(reviews: list[dict]) -> list[dict]:
    """レビューデータからセンチメント分析リクエストを生成"""
 
    ANALYSIS_SCHEMA = """
{
  "sentiment": "positive | neutral | negative",
  "score": 1-5の整数,
  "categories": ["品質", "価格", "配送", "サポート"] の中から該当するもの,
  "summary": "20文字以内の要約",
  "requires_action": true | false  // 返金・対応が必要な場合はtrue
}"""
 
    requests = []
    for review in reviews:
        requests.append({
            "custom_id": f"review_{review['id']}",
            "params": {
                "model": "claude-haiku-4-5-20251001",
                "max_tokens": 150,
                "messages": [
                    {
                        "role": "user",
                        "content": f"""以下のレビューを分析し、JSON形式で回答してください。
 
レビュー: {review['text']}
 
出力形式:
{ANALYSIS_SCHEMA}
 
JSONのみを出力し、説明は不要です。"""
                    }
                ]
            }
        })
 
    return requests

TARGET_LANGUAGES = [
    ("en", "English"),
    ("zh", "Simplified Chinese"),
    ("ko", "Korean"),
    ("es", "Spanish"),
]
 
def create_translation_requests(articles: list[dict]) -> list[dict]:
    """記事を複数言語に翻訳するリクエストを生成"""
    requests = []
 
    for article in articles:
        for lang_code, lang_name in TARGET_LANGUAGES:
            requests.append({
                "custom_id": f"article_{article['id']}_{lang_code}",
                "params": {
                    "model": "claude-sonnet-4-6",  # 翻訳品質のためsonnetを使用
                    "max_tokens": len(article['content']) * 2,  # 余裕を持たせる
                    "messages": [
                        {
                            "role": "user",
                            "content": f"""Translate the following Japanese article to {lang_name}.
Maintain the original tone, formatting, and technical accuracy.
Return only the translated text.
 
---
{article['content']}"""
                        }
                    ]
                }
            })
 
    return requests

def retry_failed_requests(
    client: anthropic.Anthropic,
    original_requests: list[dict],
    results: list[dict],
    max_retries: int = 3,
) -> list[dict]:
    """
    失敗・期限切れのリクエストを再送信する。
    リトライ回数の上限を設けてループを防ぐ。
    """
    # 失敗・期限切れのIDを取得
    failed_ids = {
        r["id"] for r in results
        if r["status"] in ("error", "expired")
    }
 
    if not failed_ids:
        logger.info("再送信が必要なリクエストはありません")
        return results
 
    # 元のリクエストから失敗分を抽出
    retry_requests = [
        req for req in original_requests
        if req["custom_id"] in failed_ids
    ]
 
    logger.warning(
        f"{len(retry_requests)} 件のリクエストを再送信します "
        f"(試行回数: {max_retries})"
    )
 
    # 再送信（上限回数まで）
    all_results = [r for r in results if r["status"] == "success"]
 
    for attempt in range(max_retries):
        if not retry_requests:
            break
 
        batch = client.beta.messages.batches.create(requests=retry_requests)
        completed = wait_for_batch_completion(client, batch.id)
        retry_results = process_batch_results(client, batch.id, {})
 
        # 成功したものを結果に追加
        success_results = [r for r in retry_results if r["status"] == "success"]
        all_results.extend(success_results)
 
        # まだ失敗しているリクエストを次の試行対象に
        still_failed_ids = {
            r["id"] for r in retry_results
            if r["status"] in ("error", "expired")
        }
        retry_requests = [
            req for req in retry_requests
            if req["custom_id"] in still_failed_ids
        ]
 
        logger.info(f"リトライ {attempt + 1}/{max_retries}: 残り {len(retry_requests)} 件")
 
    # 最終的に失敗したリクエストも結果に含める
    if retry_requests:
        logger.error(f"{len(retry_requests)} 件が最終的に失敗しました")
        all_results.extend([
            {"id": req["custom_id"], "status": "permanent_failure"}
            for req in retry_requests
        ])
 
    return all_results

# プロンプトキャッシュ + Batches API の組み合わせ
request_with_cache = {
    "custom_id": "doc_001",
    "params": {
        "model": "claude-sonnet-4-6",
        "max_tokens": 500,
        "system": [
            {
                "type": "text",
                "text": "あなたは文書分析の専門家です。以下のルールに従って...(長い共通指示)",
                "cache_control": {"type": "ephemeral"}  # キャッシュ有効化
            }
        ],
        "messages": [
            {"role": "user", "content": "文書の内容..."}
        ]
    }
}

def adaptive_polling(client, batch_id, base_interval=60, fast_interval=15):
    """進捗 95% 超で短間隔に切り替える適応ポーリング"""
    while True:
        batch = client.beta.messages.batches.retrieve(batch_id)
        if batch.processing_status == "ended":
            return batch
 
        rc = batch.request_counts
        total = rc.processing + rc.succeeded + rc.errored + rc.expired
        done = rc.succeeded + rc.errored + rc.expired
        progress = done / total if total else 0
 
        # 95% 超えたら短間隔モード
        interval = fast_interval if progress >= 0.95 else base_interval
        time.sleep(interval)

[project]
dependencies = [
  "anthropic>=0.40.0,<1.0.0",  # Batches Beta が安定化するまで minor もピン留め
]