batch-processing

バッチ処理

googleadsagent.ai™ による Agent Skills™ の一部です。

説明

バッチ処理では、進捗追跡、エラー処理、レート制限、結果の集約を行いながら、AI タスクを並列実行できます。エージェントは、ドキュメント、画像、コードファイル、API リクエストなどの大量の項目を、AI パイプラインで並行して処理し、スループット、失敗、部分的な結果を管理して、作業を失うことなく処理します。

単一項目の AI 処理は簡単ですが、大規模なバッチ処理では、失敗、レート制限、メモリ制約に加え、処理を再開できることが必要になります。このスキルは、本番環境で直面するこうした課題に対応します。設定可能な同時実行数の上限、レート制限エラー発生時の指数バックオフ、クラッシュ後に再開できるチェックポイント方式、成功・失敗・保留中の項目を正確に示す構造化された進捗レポートを備えています。

このスキルは、同種のバッチ（すべての項目で同じ処理を行う）と異種のバッチ（項目の種類ごとに異なる処理を振り分ける）の両方に対応しています。結果は、項目ごとのステータス、所要時間、エラーの詳細を含む構造化レポートに集約されます。失敗した項目はバックオフを伴って自動的に再試行され、恒久的に失敗した項目は、手動で確認できるようデッドレターキューに収集されます。

使用する場面

• AI パイプラインで数百から数千の項目を処理する場合
• 大規模なドキュメント群を翻訳、要約、分類する場合
• ドキュメントコーパスの埋め込みを生成する場合
• リポジトリ全体のコード分析を実行する場合
• 画像、説明、メタデータを一括生成する場合
• 項目を順次処理するタスクで、並列化による効果が見込める場合

仕組み

graph TD
    A[Input Batch: N Items] --> B[Load Checkpoint: Resume if Exists]
    B --> C[Partition into Work Chunks]
    C --> D[Parallel Workers: Concurrency Limit]
    D --> E[Worker 1: Process Item]
    D --> F[Worker 2: Process Item]
    D --> G[Worker K: Process Item]
    E --> H{Success?}
    F --> H
    G --> H
    H -->|Yes| I[Record Result]
    H -->|Rate Limited| J[Backoff + Retry]
    H -->|Failed| K[Retry Queue]
    J --> D
    K --> L{Retries Exhausted?}
    L -->|No| D
    L -->|Yes| M[Dead Letter Queue]
    I --> N[Save Checkpoint]
    N --> O[Progress Report]
    O --> P{All Done?}
    P -->|No| D
    P -->|Yes| Q[Final Aggregation Report]

エンジンは、レート制限と再試行回数の上限を守りながら、複数の同時実行ワーカーに作業を分配します。各チャンクの処理後にチェックポイントを保存し、クラッシュ後の復旧を可能にします。デッドレターキューには、恒久的に失敗した項目が記録され、人による確認に回されます。

実装

import asyncio
from dataclasses import dataclass, field
from time import time
import json

@dataclass
class BatchItem:
    id: str
    input: dict
    status: str = "pending"
    result: dict | None = None
    error: str | None = None
    attempts: int = 0
    duration_ms: float = 0

@dataclass
class BatchResult:
    total: int
    succeeded: int
    failed: int
    dead_letter: int
    duration_s: float
    items: list[BatchItem] = field(default_factory=list)

class BatchProcessor:
    def __init__(self, concurrency: int = 5, max_retries: int = 3, checkpoint_file: str = "batch_checkpoint.json"):
        self.concurrency = concurrency
        self.max_retries = max_retries
        self.checkpoint_file = checkpoint_file
        self.semaphore = asyncio.Semaphore(concurrency)

    async def process(self, items: list[BatchItem], processor_fn) -> BatchResult:
        items = self._load_checkpoint(items)
        start = time()
        pending = [i for i in items if i.status == "pending"]

        tasks = [self._process_item(item, processor_fn) for item in pending]
        await asyncio.gather(*tasks, return_exceptions=True)

        return BatchResult(
            total=len(items),
            succeeded=sum(1 for i in items if i.status == "succeeded"),
            failed=sum(1 for i in items if i.status == "failed"),
            dead_letter=sum(1 for i in items if i.status == "dead_letter"),
            duration_s=time() - start,
            items=items,
        )

    async def _process_item(self, item: BatchItem, processor_fn):
        async with self.semaphore:
            while item.attempts < self.max_retries:
                item.attempts += 1
                start = time()
                try:
                    item.result = await processor_fn(item.input)
                    item.status = "succeeded"
                    item.duration_ms = (time() - start) * 1000
                    self._save_checkpoint_item(item)
                    return
                except RateLimitError:
                    await asyncio.sleep(2 ** item.attempts)
                except Exception as e:
                    item.error = str(e)
                    item.duration_ms = (time() - start) * 1000

            item.status = "dead_letter"
            self._save_checkpoint_item(item)

    def _save_checkpoint_item(self, item: BatchItem):
        try:
            data = json.loads(open(self.checkpoint_file).read()) if Path(self.checkpoint_file).exists() else {}
            data[item.id] = {"status": item.status, "result": item.result, "error": item.error}
            with open(self.checkpoint_file, "w") as f:
                json.dump(data, f)
        except Exception:
            pass

    def _load_checkpoint(self, items: list[BatchItem]) -> list[BatchItem]:
        try:
            data = json.loads(open(self.checkpoint_file).read())
            for item in items:
                if item.id in data:
                    item.status = data[item.id]["status"]
                    item.result = data[item.id]["result"]
        except FileNotFoundError:
            pass
        return items

ベストプラクティス

• CPU コア数だけでなく、API のレート制限に基づいて同時実行数の上限を設定します
• 100 項目を超えるバッチでは、チェックポイント方式で処理を再開できるようにします
• レート制限による再試行には指数バックオフ（2^attempt 秒）を使用します
• 定期的に（10％ごと、または30秒ごとに）進捗を報告します
• 手動確認と再処理のために、デッドレター項目を別途収集します
• スループットを低下させる処理の遅い項目を特定するため、項目ごとの所要時間を記録します

プラットフォーム互換性

プラットフォーム	対応状況	備考
Cursor	完全対応	Python/TS の非同期実行
VS Code	完全対応	ターミナルベースのバッチ実行
Windsurf	完全対応	バッチワークフローに対応
Claude Code	完全対応	スクリプト実行
Cline	完全対応	バッチタスク管理
aider	一部対応	順次実行のみ

関連スキル

• ワークフローオーケストレーション
• ローコード生成
• 並列エージェントオーケストレーション
• トークン最適化

キーワード

batch-processing parallel-execution rate-limiting retry-logic checkpointing dead-letter-queue progress-tracking concurrency

---

© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License