AIエージェント入門

Claude Code SDKサブエージェントで大量ファイルを並列処理する方法

この記事の結論

Claude Code SDKのサブエージェントとgit worktreeを組み合わせ、数百ファイルを並列処理するパイプラインを構築する実践ガイド。/batchコマンドとBatch APIの使い分けも解説。



「/batchは知ってる。でも、もっと細かく制御したい」——そんな声をよく聞きます。

/batchコマンドは強力ですが、「Worktree単位でのオーケストレーションが固定」「カスタムロジックを挟めない」という制限があります。Claude Code SDKのサブエージェントとgit worktreeを組み合わせると、数百ファイルを同時並行で処理する独自パイプラインが作れます。この記事では、よく出る疑問形式で実践パターンを解説します。

なお、/batchコマンドと/simplifyコマンドの基礎についてはClaude Code /batch + /simplify実践ガイドで詳しく解説しています。この記事はその先、「SDK経由での自前オーケストレーション」に特化した内容です。

そもそもサブエージェントとは何か

Claude Code SDKでは、親エージェントがTaskツールを使って子プロセス(サブエージェント)をスポーンできます。各サブエージェントは独立したコンテキストウィンドウを持ち、独自のツールアクセス権を持ちます。

方式 制御の細かさ カスタムロジック 向くケース 制限
/batchコマンド 低(自動分解) なし コードリファクタ、パターン適用 Worktree必須、Claude Code内のみ
SDK Subagent + Task 高(手動オーケスト) Python任意処理を挟める 大量ファイル変換、バッチ評価 SDK設定が必要
Batch API(Claude API) 中(リクエスト単位) 前処理・後処理OK 大量推論、データ加工 最大100K件、24h以内
git worktree直接管理 最高(完全制御) 全て自由 マルチブランチ並列開発 merge戦略の設計が必要

何が新しいのか — SDK v1.5以降のサブエージェント機能

Claude Code SDK v1.5以降、サブエージェントにisolation: worktreeを指定すると、各サブエージェントが独立したgit worktree(別ブランチの一時コピー)上で動作します。並列編集によるコンフリクトが原理的に起きません。

# 動作環境: Python 3.11+, claude-code-sdk>=1.5
# pip install claude-code-sdk
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

import asyncio
from claude_code_sdk import ClaudeCode, SubagentConfig

async def process_files_in_parallel(file_paths: list[str]) -> list[str]:
    """複数ファイルをサブエージェントで並列処理する"""
    cc = ClaudeCode()

    # ファイルを5件ずつのチャンクに分割(サブエージェント数の上限を制御)
    chunk_size = 5
    chunks = [file_paths[i:i+chunk_size] for i in range(0, len(file_paths), chunk_size)]

    results = []
    for chunk in chunks:
        # チャンク内のファイルを並列スポーン
        tasks = [
            cc.spawn_subagent(
                prompt=f"以下のファイルのdocstringをGoogle形式に変換してください: {fp}",
                config=SubagentConfig(
                    isolation="worktree",          # worktree分離で競合ゼロ
                    model="claude-haiku-4-5",       # コスト削減:Haikuを使用
                    allowed_tools=["Read", "Edit"], # 権限を最小限に
                    timeout_seconds=120,
                ),
            )
            for fp in chunk
        ]
        chunk_results = await asyncio.gather(*tasks)
        results.extend(chunk_results)

    return results

# 実行例
import glob
py_files = glob.glob("src/**/*.py", recursive=True)
print(f"処理対象: {len(py_files)}ファイル")
asyncio.run(process_files_in_parallel(py_files))

ポイント: model="claude-haiku-4-5"を使うと、大量並列時のコストをOpus比で約20分の1に抑えられます。単純な変換タスクには十分な性能です。

具体的に何ができるようになるのか

パターン1:大量ファイルの一括テスト生成

# 動作環境: Python 3.11+, claude-code-sdk>=1.5, pytest
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

import asyncio
from pathlib import Path
from claude_code_sdk import ClaudeCode, SubagentConfig

async def generate_tests_for_module(cc: ClaudeCode, module_path: str) -> str:
    """1ファイル分のテストをサブエージェントで生成"""
    test_path = module_path.replace("src/", "tests/").replace(".py", "_test.py")
    return await cc.spawn_subagent(
        prompt=f"""
        {module_path} を読み、pytest形式のユニットテストを {test_path} に生成してください。
        - 正常系・異常系・境界値の3パターン以上
        - モックが必要な外部依存はpytest-mockを使う
        - カバレッジ目標: 80%以上
        """,
        config=SubagentConfig(
            isolation="worktree",
            model="claude-sonnet-4-6",  # テスト生成はSonnetで品質重視
            allowed_tools=["Read", "Write"],
        ),
    )

async def main():
    cc = ClaudeCode()
    modules = list(Path("src").rglob("*.py"))

    # 最大10並列でテスト生成
    sem = asyncio.Semaphore(10)
    async def bounded_generate(mod):
        async with sem:
            return await generate_tests_for_module(cc, str(mod))

    results = await asyncio.gather(*[bounded_generate(m) for m in modules])
    print(f"テスト生成完了: {len([r for r in results if r])} / {len(modules)} ファイル")

asyncio.run(main())

パターン2:Anthropic Message Batch APIで大量推論

Claude Code SDKとは別に、Anthropic API直接呼び出しでバッチ処理する方法もあります。1バッチあたり最大10,000件のリクエストを送信でき、通常料金の50%で実行できます(2026年4月時点)。大量処理の場合は複数バッチを並行送信します。

# 動作環境: Python 3.10+, anthropic>=0.40
# pip install anthropic
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。

import anthropic
import json
from pathlib import Path

client = anthropic.Anthropic()

def create_batch_from_files(file_paths: list[str]) -> str:
    """ファイルリストからバッチリクエストを作成して送信"""
    requests = []
    for fp in file_paths:
        content = Path(fp).read_text(encoding="utf-8")
        requests.append({
            "custom_id": fp,  # 後で結果を紐付けるID
            "params": {
                "model": "claude-haiku-4-5",
                "max_tokens": 1024,
                "messages": [{
                    "role": "user",
                    "content": f"以下のコードを日本語でコメントしてください:\n\n{content[:2000]}",
                }],
            },
        })

    # バッチ送信(非同期処理、最大24時間以内に完了)
    # 注意: 1バッチあたり最大10,000リクエスト。大量の場合は複数バッチに分割
    batch = client.messages.batches.create(requests=requests)
    print(f"バッチ作成: {batch.id} / リクエスト数: {len(requests)}")
    return batch.id

def poll_batch_results(batch_id: str) -> dict:
    """バッチ結果をポーリングして取得"""
    import time
    while True:
        batch = client.messages.batches.retrieve(batch_id)
        if batch.processing_status == "ended":
            break
        print(f"処理中... {batch.request_counts.processing}件残り")
        time.sleep(30)

    # 結果を辞書に変換
    results = {}
    for result in client.messages.batches.results(batch_id):
        if result.result.type == "succeeded":
            results[result.custom_id] = result.result.message.content[0].text
    return results

# 使用例
file_paths = list(Path("src").rglob("*.py"))
batch_id = create_batch_from_files([str(f) for f in file_paths[:500]])
results = poll_batch_results(batch_id)
print(f"処理完了: {len(results)} ファイル")

よくある誤解

誤解1:「/batchとBatch APIは同じもの」

違います。/batchはClaude Code コマンドでgit worktreeを使ったコード変更のオーケストレーター、Batch APIはAnthropicのHTTP APIでLLM推論をまとめて非同期実行するもの。用途も仕組みも別物です。

誤解2:「並列数を増やすほど速い」

サブエージェントが同じファイルに触れると、worktree分離があってもmerge時にコンフリクトします。「各サブエージェントが別々のファイルを担当する」という分解設計が並列化の鍵です。

誤解3:「全タスクをOpusで実行すべき」

ファイル変換や定型的なコメント生成にOpusを使うのはコストの無駄です。「調査・計画→Sonnet、単純変換→Haiku」のモデル使い分けが大量処理の経済合理性を左右します。

結局どうすればいいのか

3つの指針でシナリオを選んでください。

  1. Claude Code内で完結させたい/batchコマンドを使う(設定なしで即使える)
  2. Python処理を挟みたい・モデルを細かく制御したい → SDK SubagentをPythonで直接オーケストレート
  3. コスト最優先・Claude Code外から呼びたい → Anthropic Message Batch APIをPythonスクリプトから実行

参考・出典

あわせて読みたい


著者: 佐藤傑(さとう・すぐる)
株式会社Uravation代表取締役。X(@SuguruKun_ai)フォロワー10万人超。100社以上の企業向けAI研修・導入支援。著書累計3万部突破。SoftBank IT連載7回執筆(NewsPicks最大1,125ピックス)。
ご質問・ご相談は お問い合わせフォーム からお気軽にどうぞ。

Need help moving from reading to rollout?

この記事を読んで導入イメージが固まってきた方へ

Uravationでは、AIエージェントの要件整理、PoC設計、社内導入、研修まで一気通貫で支援しています。

この記事をシェア

X Facebook LINE

※ 本記事の情報は2026年4月時点のものです。サービスの料金・仕様は変更される可能性があります。最新情報は各サービスの公式サイトをご確認ください。

関連記事