Claude API Messages Batches API でコストを半額にする — 個人開発者が学んだ非同期処理設計

import anthropic
import time
 
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
 
def create_batch(items: list[dict]) -> str:
    """
    複数のテキストをバッチ処理するジョブを作成する
 
    Args:
        items: [{"id": "item_1", "text": "分析対象テキスト"}, ...]
 
    Returns:
        batch_id: バッチのID（ステータス確認・結果取得に使用）
    """
    requests = [
        anthropic.types.message_create_params.Request(
            custom_id=item["id"],
            params=anthropic.types.message_create_params.MessageCreateParamsNonStreaming(
                model="claude-haiku-4-5-20251001",  # バッチ処理は Haiku が費用対効果◎
                max_tokens=512,
                messages=[
                    {
                        "role": "user",
                        "content": (
                            "以下のテキストの感情を「ポジティブ/ネガティブ/ニュートラル」"
                            "の3値で分類し、その理由を1文で述べてください。\n\n"
                            + item["text"]
                        )
                    }
                ]
            )
        )
        for item in items
    ]
 
    batch = client.messages.batches.create(requests=requests)
 
    print(f"✅ バッチ作成完了")
    print(f"   Batch ID: {batch.id}")
    print(f"   リクエスト数: {batch.request_counts.processing}")
    print(f"   ステータス: {batch.processing_status}")
 
    return batch.id
 
# 使用例
items = [
    {"id": "review_001", "text": "UIが直感的で使いやすい。ただ起動が少し遅い。"},
    {"id": "review_002", "text": "広告が多すぎてうんざりしている。有料版は高すぎる。"},
    {"id": "review_003", "text": "シンプルで良い。必要な機能は揃っている。"},
]
 
batch_id = create_batch(items)
# 出力例 → Batch ID: msgbatch_01Abc123...

def wait_for_batch(
    batch_id: str,
    poll_interval: int = 30,
    max_wait_seconds: int = 7200  # 最大2時間待機
) -> str:
    """
    バッチの完了を待ち、最終ステータスを返す
 
    Returns:
        "ended" | "errored" | "canceled" | "expired" | "timeout"
    """
    start_time = time.time()
 
    while True:
        elapsed = int(time.time() - start_time)
 
        if elapsed > max_wait_seconds:
            print(f"⚠️ タイムアウト（{max_wait_seconds}秒経過）")
            return "timeout"
 
        batch = client.messages.batches.retrieve(batch_id)
        status = batch.processing_status
 
        counts = batch.request_counts
        total = (
            counts.processing + counts.succeeded
            + counts.errored + counts.canceled + counts.expired
        )
        completed = counts.succeeded + counts.errored + counts.canceled + counts.expired
 
        print(f"⏳ [{elapsed}s] {status}: {completed}/{total} 完了")
 
        if status == "ended":
            print(
                f"✅ バッチ完了 "
                f"(成功: {counts.succeeded}, "
                f"エラー: {counts.errored}, "
                f"期限切れ: {counts.expired})"
            )
            return status
 
        if status in ("errored", "canceled"):
            print(f"❌ バッチ失敗: {status}")
            return status
 
        time.sleep(poll_interval)

def retrieve_batch_results(batch_id: str) -> dict[str, dict]:
    """
    バッチの結果を取得し、custom_id をキーにした辞書で返す
 
    Returns:
        {"review_001": {"type": "succeeded", "content": "..."}, ...}
    """
    results = {}
 
    for result in client.messages.batches.results(batch_id):
        custom_id = result.custom_id
 
        if result.result.type == "succeeded":
            message = result.result.message
            content_text = message.content[0].text if message.content else ""
 
            results[custom_id] = {
                "type": "succeeded",
                "content": content_text,
                "input_tokens": message.usage.input_tokens,
                "output_tokens": message.usage.output_tokens,
            }
 
        elif result.result.type == "errored":
            error = result.result.error
            results[custom_id] = {
                "type": "errored",
                "error_type": error.type,
                "error_message": str(error),
            }
 
        elif result.result.type == "expired":
            results[custom_id] = {"type": "expired"}
 
    succeeded = sum(1 for r in results.values() if r["type"] == "succeeded")
    errored = sum(1 for r in results.values() if r["type"] != "succeeded")
    print(f"📊 取得完了: 成功={succeeded}, 失敗/期限切れ={errored}")
 
    return results
 
# フルフロー
batch_id = create_batch(items)
final_status = wait_for_batch(batch_id)
 
if final_status == "ended":
    results = retrieve_batch_results(batch_id)
    for item_id, result in results.items():
        if result["type"] == "succeeded":
            print(f"{item_id}: {result['content'][:100]}...")

import sqlite3
from datetime import datetime
 
def save_batch_results_to_db(
    batch_id: str,
    results: dict[str, dict],
    db_path: str = "batch_results.db"
) -> None:
    """バッチ結果を SQLite に保存する"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
 
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS batch_results (
            batch_id     TEXT,
            custom_id    TEXT,
            type         TEXT,
            content      TEXT,
            input_tokens INTEGER,
            output_tokens INTEGER,
            error_message TEXT,
            created_at   TEXT,
            PRIMARY KEY (batch_id, custom_id)
        )
    """)
 
    now = datetime.now().isoformat()
    for custom_id, result in results.items():
        cursor.execute("""
            INSERT OR REPLACE INTO batch_results
            (batch_id, custom_id, type, content,
             input_tokens, output_tokens, error_message, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            batch_id,
            custom_id,
            result.get("type"),
            result.get("content"),
            result.get("input_tokens"),
            result.get("output_tokens"),
            result.get("error_message"),
            now
        ))
 
    conn.commit()
    conn.close()
    print(f"💾 {len(results)} 件を DB に保存しました")

def retry_failed_requests(
    original_items: list[dict],
    results: dict[str, dict],
    max_retries: int = 2
) -> dict[str, dict]:
    """失敗したリクエストのみをリトライする"""
 
    failed_ids = {
        custom_id
        for custom_id, result in results.items()
        if result["type"] != "succeeded"
    }
 
    if not failed_ids:
        print("✅ 全リクエスト成功、リトライ不要")
        return results
 
    print(f"⚠️ {len(failed_ids)} 件が失敗、リトライします")
 
    failed_items = [
        item for item in original_items
        if item["id"] in failed_ids
    ]
 
    for attempt in range(max_retries):
        print(f"🔄 リトライ {attempt + 1}/{max_retries}")
 
        retry_batch_id = create_batch(failed_items)
        retry_status = wait_for_batch(retry_batch_id)
 
        if retry_status == "ended":
            retry_results = retrieve_batch_results(retry_batch_id)
 
            for custom_id, result in retry_results.items():
                if result["type"] == "succeeded":
                    results[custom_id] = result
                    failed_ids.discard(custom_id)
 
            if not failed_ids:
                print("✅ リトライ完了、全リクエスト成功")
                break
 
    if failed_ids:
        print(f"❌ {len(failed_ids)} 件が最終的に失敗")
 
    return results

import json
from pathlib import Path
from dataclasses import dataclass, field, asdict
 
@dataclass
class BatchJob:
    """バッチジョブの状態を追跡するクラス"""
    job_name: str
    batch_id: str | None = None
    status: str = "pending"  # pending | running | ended | failed
    total_items: int = 0
    succeeded: int = 0
    failed: int = 0
    retry_count: int = 0
    created_at: str = field(
        default_factory=lambda: datetime.now().isoformat()
    )
    completed_at: str | None = None
 
    def save(self, state_dir: str = "/tmp/batch_states") -> None:
        Path(state_dir).mkdir(exist_ok=True)
        path = f"{state_dir}/{self.job_name}.json"
        with open(path, "w") as f:
            json.dump(asdict(self), f, ensure_ascii=False, indent=2)
 
    @classmethod
    def load(cls, job_name: str, state_dir: str = "/tmp/batch_states") -> "BatchJob":
        path = f"{state_dir}/{job_name}.json"
        with open(path) as f:
            return cls(**json.load(f))

# ❌ 悪い例：ID を保存せずに続行
batch = client.messages.batches.create(requests=requests)
time.sleep(300)
# ...プロセスが途中でクラッシュしたら ID が失われる
 
# ✅ 良い例：作成直後に ID を永続化
batch = client.messages.batches.create(requests=requests)
job.batch_id = batch.id
job.status = "running"
job.save()  # ファイルまたは DB に保存してから処理継続

batch = client.messages.batches.retrieve(batch_id)
if batch.processing_status == "ended":
    counts = batch.request_counts
    if counts.errored > 0 or counts.expired > 0:
        print(
            f"⚠️ 一部失敗: "
            f"errored={counts.errored}, expired={counts.expired}"
        )
        # リトライ処理へ

import glob
 
def generate_seo_descriptions(articles_dir: str) -> str:
    """MDX ファイルの SEO description を一括生成する"""
    mdx_files = glob.glob(f"{articles_dir}/**/*.mdx", recursive=True)
 
    items = []
    for filepath in mdx_files[:200]:  # 1バッチ最大200件
        with open(filepath, encoding="utf-8") as f:
            content = f.read()
 
        # フロントマターを除いた本文の冒頭1,000文字
        body = content.split("---", 2)[-1][:1000]
        slug = filepath.split("/")[-1].replace(".mdx", "")
 
        items.append({
            "id": slug,
            "text": (
                "以下の技術記事本文を読み、160文字以内のSEO用description"
                "を日本語で生成してください。"
                "導入文フレーズ（「ここでは〜」等）は禁止。\n\n" + body
            )
        })
 
    print(f"📝 {len(items)} 件を処理します")
    return create_batch(items)

def analyze_app_reviews_daily(reviews: list[dict]) -> str:
    """
    アプリレビューを日次バッチで感情分析する
    reviews: [{"id": "rev_001", "text": "レビュー本文", "rating": 4}, ...]
    """
    items = []
    for review in reviews:
        prompt = (
            "以下のアプリレビューを分析し、JSON形式のみ返してください。\n"
            '{"sentiment":"positive|negative|neutral",'
            '"main_issue":"主な問題（なければnull）",'
            '"feature_request":"機能要望（なければnull）",'
            '"urgency":"high|medium|low"}\n\n'
            f"評価: {review['rating']}星\n"
            f"レビュー: {review['text']}"
        )
        items.append({"id": review["id"], "text": prompt})
 
    return create_batch(items)

def check_pending_batches(state_dir: str = "/var/lib/batch_states") -> None:
    """cron から15分おきに呼ばれ、未完了バッチを確認する"""
    for path in Path(state_dir).glob("*.json"):
        job = BatchJob.load(path.stem, state_dir)
        if job.status != "running":
            continue
        batch = client.messages.batches.retrieve(job.batch_id)
        if batch.processing_status == "ended":
            results = retrieve_batch_results(job.batch_id)
            save_batch_results_to_db(job.batch_id, results)
            job.status = "ended"
            job.completed_at = datetime.now().isoformat()
            job.save(state_dir)