「/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つの指針でシナリオを選んでください。
- Claude Code内で完結させたい →
/batchコマンドを使う(設定なしで即使える) - Python処理を挟みたい・モデルを細かく制御したい → SDK SubagentをPythonで直接オーケストレート
- コスト最優先・Claude Code外から呼びたい → Anthropic Message Batch APIをPythonスクリプトから実行
参考・出典
- Subagents in the SDK — Anthropic API公式ドキュメント(参照日: 2026-04-11)
- Batch processing — Claude API Docs(参照日: 2026-04-11)
- Create custom subagents — Claude Code Docs(参照日: 2026-04-11)
- Claude Code Git Worktrees: Parallel Coding — Popular AI Tools(参照日: 2026-04-11)
あわせて読みたい
- Claude Code /batch + /simplify実践ガイド — コマンド起点で始める並列処理の入門
- AIエージェント構築完全ガイド — サブエージェント設計の前提知識を整理
著者: 佐藤傑(さとう・すぐる)
株式会社Uravation代表取締役。X(@SuguruKun_ai)フォロワー10万人超。100社以上の企業向けAI研修・導入支援。著書累計3万部突破。SoftBank IT連載7回執筆(NewsPicks最大1,125ピックス)。
ご質問・ご相談は お問い合わせフォーム からお気軽にどうぞ。