AIコーディング 2026.05.12
Claude Codeで複数チーム同時管理【2026年版】企画→開発→承認を1セッションで完結するワークフロー
マルチチームワークフローのアーキテクチャ
タスク受付
↓
企画ロール（Claude）: 要件定義・優先順位付け
↓（並列）
開発ロール（Claude）: コード実装
QAロール（Claude）: テスト・品質確認
↓
レビュー・承認
↓
state.jsonに状態保存 → 次セッションへ引き継ぎ
役割別エージェントの実装
# role_based_agent.py
# 企画・開発・QAの3役割を1つのスクリプトで管理する
import anthropic
import json
from pathlib import Path
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
client = anthropic.Anthropic()
ROLES = {
"planner": {
"model": "claude-opus-5", # 企画には高品質モデル
"system": "あなたはプロジェクト企画担当です。ビジネス要件を技術仕様に落とし込み、優先順位と見積もりを提供してください。",
},
"developer": {
"model": "claude-sonnet-4-5", # 開発は汎用モデル
"system": "あなたはシニアエンジニアです。実装コードを書き、コードの品質・拡張性・テスト可能性を考慮してください。",
},
"qa": {
"model": "claude-haiku-4-5-20251001", # QA確認は安いモデル
"system": "あなたはQAエンジニアです。実装のバグ・セキュリティリスク・エッジケースを確認し、テストケースを提案してください。",
},
}
def ask_role(role: str, task: str) -> dict:
"""特定の役割としてClaudeに質問する"""
config = ROLES[role]
response = client.messages.create(
model=config["model"],
max_tokens=1024,
system=config["system"],
messages=[{"role": "user", "content": task}]
)
return {
"role": role,
"response": response.content[0].text,
"tokens": response.usage.input_tokens + response.usage.output_tokens
}
def parallel_team_review(task: str) -> dict:
"""企画・開発・QAを並列実行して結果を統合する"""
with ThreadPoolExecutor(max_workers=3) as executor:
futures = {
executor.submit(ask_role, role, task): role
for role in ROLES
}
results = {}
for future in futures:
result = future.result()
results[result["role"]] = result
total_tokens = sum(r["tokens"] for r in results.values())
print(f"並列処理完了: 合計{total_tokens}トークン使用")
return results
# 使用例
results = parallel_team_review(
"ユーザーのCSVファイルを読み込んでグラフを表示するWeb機能を追加してください"
)
for role, data in results.items():
print(f"\n=== {role.upper()} ===\n{data['response'][:300]}")
state-as-files パターン（セッション間の状態引き継ぎ）
# state_manager.py
# プロジェクト状態をJSONで永続化して次のセッションに引き継ぐ
import json
from pathlib import Path
from datetime import datetime
STATE_FILE = Path(".claude_project_state.json")
def save_state(task: str, results: dict, decisions: list) -> None:
"""タスク結果と決定事項を保存する"""
state = {
"last_updated": datetime.now().isoformat(),
"current_task": task,
"team_outputs": {
role: data["response"] for role, data in results.items()
},
"decisions": decisions,
"pending": [] # 未解決の課題
}
STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2))
print(f"状態保存: {STATE_FILE}")
def load_state_summary() -> str:
"""前のセッションの状態を要約してClaudeに渡す"""
if not STATE_FILE.exists():
return "前のセッションの状態なし（新規プロジェクト）"
state = json.loads(STATE_FILE.read_text())
return f"""
前回の作業（{state['last_updated'][:10]}）:
タスク: {state['current_task']}
決定事項: {', '.join(state['decisions'][:3])}
継続課題: {len(state.get('pending', []))}件
"""
# 新セッション開始時
previous_context = load_state_summary()
print(f"引き継ぎ情報:\n{previous_context}")
コスト試算
COSTS = {
"claude-opus-5": (15.00, 75.00),
"claude-sonnet-4-5": (3.00, 15.00),
"claude-haiku-4-5-20251001": (0.80, 4.00),
}
def estimate_team_cost(daily_tasks: int) -> dict:
"""1日N件のチームレビューのコストを試算"""
per_task = 0
for role, (inp, out) in zip(
["planner", "developer", "qa"],
[COSTS["claude-opus-5"], COSTS["claude-sonnet-4-5"], COSTS["claude-haiku-4-5-20251001"]]
):
# 各役割: 入力3000+出力1000トークン
per_task += (3000 * inp + 1000 * out) / 1_000_000
return {
"per_task_usd": round(per_task, 4),
"monthly_usd": round(per_task * daily_tasks * 22, 2), # 月22営業日
}
print("月間コスト試算:")
for daily in [5, 10, 20]:
result = estimate_team_cost(daily)
print(f" 1日{daily}タスク: ${result['monthly_usd']}/月")
