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

import anthropic
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming
from anthropic.types.messages.batch_create_params import Request
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
documents = [
    {"id": "doc_001", "content": "AIの倫理に関する最新の議論について述べる文書..."},
    {"id": "doc_002", "content": "機械学習の基礎と応用に関する技術論文..."},
    {"id": "doc_003", "content": "スタートアップの資金調達戦略についての解説..."},
]
 
MODEL = "claude-sonnet-4-6"  # モデル ID は 1 か所にまとめる
 
batch_requests = [
    Request(
        # custom_id: 結果と元データを紐付けるための ID（必須）
        # 1〜64 文字の英数字・ハイフン・アンダースコアのみ
        custom_id=doc["id"],
        params=MessageCreateParamsNonStreaming(
            model=MODEL,
            max_tokens=500,  # バッチでは 1 以上が必須（0 は使えない）
            messages=[
                {
                    "role": "user",
                    "content": f"以下の文書を3行で要約してください。\n\n{doc['content']}",
                }
            ],
        ),
    )
    for doc in documents
]
 
batch = client.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.messages.MessageBatch:
    """
    バッチ処理の完了を待機する関数。
 
    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.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.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.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":
            # エラーが発生したリクエスト（課金されません）
            # Python SDK では error はエラーレスポンスなので、
            # 実際の種別は error.error.type にあります（.error が二段）
            err = result.result.error.error
            results.append({
                "id": custom_id,
                "status": "error",
                "original": original_doc,
                "error_type": err.type,       # 例: invalid_request_error
                "error_message": err.message,
            })
 
        elif result.result.type == "expired":
            # 24 時間の処理枠に入らなかったリクエスト（課金されません）
            results.append({
                "id": custom_id,
                "status": "expired",
                "original": original_doc,
            })
 
        elif result.result.type == "canceled":
            # キャンセルしたバッチには部分結果が残るため、
            # canceled を取りこぼすと「消えたリクエスト」になります
            results.append({
                "id": custom_id,
                "status": "canceled",
                "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,  # 上限は 100,000 だが 256 MB 側で先に詰まる
    ):
        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]:
        """
        大量のリクエストを複数バッチに分割して送信。
 
        件数の上限は 100,000 だが、実務で先に当たるのは 256 MB のサイズ上限。
        長いシステムプロンプトを全リクエストに複製すると、
        数千件でサイズ側に触れることがあるため保守的に分割する。
        """
        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.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・いずれも Batches 割引後 = 公開価格の半額）
        # 退役したモデルの価格を残すと、移行後に実費と合わなくなる
        pricing = {
            "claude-sonnet-4-6": {"input": 1.50, "output": 7.50},
            "claude-sonnet-5": {"input": 1.00, "output": 5.00},
            "claude-haiku-4-5-20251001": {"input": 0.50, "output": 2.50},
            "claude-opus-5": {"input": 2.50, "output": 12.50},
        }
 
        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],
    documents_map: dict[str, 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.messages.batches.create(requests=retry_requests)
        wait_for_batch_completion(client, batch.id)
        # 元データの辞書を渡さないと original が空になり、
        # 再送で成功した分だけ突合できない結果が混ざります
        retry_results = process_batch_results(client, batch.id, documents_map)
 
        # 成功したものを結果に追加
        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": "あなたは文書分析の専門家です。以下のルールに従って...(長い共通指示)",
                # 既定の ephemeral は有効期限 5 分。バッチは 5 分を超えて
                # 処理されることが普通なので、1 時間の TTL を明示する
                "cache_control": {"type": "ephemeral", "ttl": "1h"}
            }
        ],
        "messages": [
            {"role": "user", "content": "文書の内容..."}
        ]
    }
}

def adaptive_polling(client, batch_id, base_interval=60, fast_interval=15,
                     deadline_minutes=180):
    """進捗 95% 超で短間隔に切り替える適応ポーリング"""
    deadline = time.monotonic() + deadline_minutes * 60
    while time.monotonic() < deadline:
        batch = client.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)
 
    # 24 時間枠の中でも、後続処理の締切は自分で持つ
    raise TimeoutError(f"{deadline_minutes} 分を超えても ended になりませんでした: {batch_id}")

[project]
dependencies = [
  "anthropic>=0.40.0,<1.0.0",  # 型の置き場所が動くため minor もピン留め
]

import re
 
import json
 
MAX_REQUESTS_PER_BATCH = 100_000      # 公式上限
MAX_BATCH_BYTES = 256 * 1024 * 1024   # 256 MB。実務ではこちらに先に当たる
CUSTOM_ID_RE = re.compile(r"^[a-zA-Z0-9_-]{1,64}$")
UNSUPPORTED_PARAMS = ("stream", "speed")  # バッチでは検証エラーになる
 
def preflight(requests: list[dict], budget_usd: float | None = None) -> None:
    """バッチ投入前の静的検証。1つでも落ちたら投入しない"""
    if not requests:
        raise ValueError("空のバッチは投入しない（空振りのポーリングが残る）")
    if len(requests) > MAX_REQUESTS_PER_BATCH:
        raise ValueError(f"1バッチ上限超過: {len(requests)} 件")
 
    size = len(json.dumps(requests, ensure_ascii=False).encode("utf-8"))
    if size > MAX_BATCH_BYTES:
        raise ValueError(f"サイズ上限超過: {size / 1024 / 1024:.1f} MB（413 になる）")
 
    seen: set[str] = set()
    for r in requests:
        cid = r["custom_id"]
        if not CUSTOM_ID_RE.match(cid):
            raise ValueError(f"custom_id が仕様外: {cid}")
        if cid in seen:
            raise ValueError(f"custom_id 重複: {cid}")
        seen.add(cid)
 
        params = r["params"]
        for key in UNSUPPORTED_PARAMS:
            if key in params:
                raise ValueError(f"バッチでは使えないパラメータ: {key} ({cid})")
        if params.get("max_tokens", 0) < 1:
            raise ValueError(f"max_tokens は 1 以上が必須: {cid}")
 
    # max_tokens の積み上げから上限コストを概算し、想定外の膨張を止める
    if budget_usd is not None:
        est = sum(estimate_max_cost(r) for r in requests)
        if est > budget_usd:
            raise ValueError(f"想定コスト超過: ${est:.2f} > ${budget_usd:.2f}")