Claude Code で実装する高度なマルチエージェント・オーケストレーション完全マニュアル

マルチエージェント・オーケストレーションの必要性

企業のデジタル変革では、単一の AI エージェントでは対応しきれない複雑なタスクが増えています：

金融分析 — 複数の市場データソースを並列取得し、リアルタイムで分析・リスク評価
Webスクレイピング — 異なるサイト構造に対応した複数スクレイパーの並列実行・結果統合
マルチステップワークフロー — 営業提案作成、顧客オンボーディング、契約作成まで一連のタスクを自動化
エラー耐性 — 一部のエージェントが失敗しても全体ワークフローは継続

従来のアプローチでは、複雑な状態管理コードやエラーハンドリングロジックが肥大化していました。Claude Code と Agent SDK の組み合わせで、こうした課題を体系的に解決できます。

基本概念：マルチエージェント・オーケストレーション

マスター・ワーカーパターン

// マスターエージェント（Orchestrator）の実装例
const Anthropic = require("@anthropic-ai/sdk");
 
class AgentOrchestrator {
  constructor() {
    this.client = new Anthropic();
    this.agents = [];
  }
 
  // ワーカーエージェントを登録
  registerAgent(name, role, capabilities) {
    this.agents.push({ name, role, capabilities });
  }
 
  // タスク分配エンジン
  async distributeTask(task, context) {
    // タスク分析
    const analysis = await this.analyzeTask(task);
    
    // 適切なエージェントを選択
    const assignedAgents = this.selectAgents(analysis, context);
    
    // 並列実行
    const results = await Promise.all(
      assignedAgents.map(agent => this.executeAgent(agent, task, context))
    );
 
    return this.aggregateResults(results);
  }
 
  // タスク分析（専門性の判定）
  async analyzeTask(task) {
    const response = await this.client.messages.create({
      model: "claude-opus-4-6",
      max_tokens: 1024,
      messages: [
        {
          role: "user",
          content: `Analyze this task and identify required skills: "${task}". 
          Return JSON: {skills: [string], complexity: number (1-10), estimatedTime: string}`
        }
      ]
    });
 
    const content = response.content[0];
    if (content.type === "text") {
      return JSON.parse(content.text);
    }
    return { skills: [], complexity: 5, estimatedTime: "unknown" };
  }
 
  // 最適なエージェント選択
  selectAgents(analysis, context) {
    return this.agents.filter(agent => {
      const hasSkills = analysis.skills.some(skill => 
        agent.capabilities.includes(skill)
      );
      return hasSkills && (context.agentPreferences ? 
        context.agentPreferences.includes(agent.name) : true);
    });
  }
 
  // エージェント実行
  async executeAgent(agent, task, context) {
    try {
      const response = await this.client.messages.create({
        model: "claude-opus-4-6",
        max_tokens: 4096,
        messages: [
          {
            role: "user",
            content: `You are a ${agent.role} agent with expertise in: ${agent.capabilities.join(", ")}.
            Task: ${task}
            Context: ${JSON.stringify(context)}
            
            Execute the task and provide detailed results with reasoning.`
          }
        ]
      });
 
      return {
        agentName: agent.name,
        result: response.content[0].type === "text" ? response.content[0].text : null,
        status: "success",
        tokensUsed: response.usage
      };
    } catch (error) {
      return {
        agentName: agent.name,
        error: error.message,
        status: "failed",
        tokensUsed: 0
      };
    }
  }
 
  // 結果の統合
  aggregateResults(results) {
    const successful = results.filter(r => r.status === "success");
    const failed = results.filter(r => r.status === "failed");
 
    return {
      successCount: successful.length,
      failureCount: failed.length,
      results: successful.map(r => ({
        agent: r.agentName,
        output: r.result
      })),
      errors: failed
    };
  }
}
 
// 使用例
const orchestrator = new AgentOrchestrator();
orchestrator.registerAgent("data-analyst", "Financial Data Analyst", [
  "financial-analysis", "data-visualization", "trend-detection"
]);
orchestrator.registerAgent("web-scraper", "Web Content Specialist", [
  "web-scraping", "html-parsing", "data-extraction"
]);
 
const result = await orchestrator.distributeTask(
  "Analyze current market trends and extract competitor pricing",
  { agentPreferences: ["data-analyst", "web-scraper"] }
);
 
console.log(result);

実装パターン 1：並列実行 × エラーハンドリング

ユースケース：複数ソースからのリアルタイムデータ取得

金融データ（株価・暗号資産・外国為替）を複数のソースから並列取得し、結果をマージ。単一ソースの障害時も全体処理は継続。

class ResilientDataFetcher {
  constructor(maxRetries = 3, timeoutMs = 30000) {
    this.client = new Anthropic();
    this.maxRetries = maxRetries;
    this.timeoutMs = timeoutMs;
  }
 
  // ソースごとの取得タスク
  async fetchFromSource(source, query) {
    let lastError;
    for (let attempt = 1; attempt <= this.maxRetries; attempt++) {
      try {
        const response = await Promise.race([
          this.client.messages.create({
            model: "claude-opus-4-6",
            max_tokens: 2048,
            messages: [
              {
                role: "user",
                content: `Fetch data from ${source} for query: "${query}". 
                Return JSON with: {source, timestamp, data, confidence}`
              }
            ]
          }),
          new Promise((_, reject) =>
            setTimeout(() => reject(new Error("Timeout")), this.timeoutMs)
          )
        ]);
 
        return JSON.parse(response.content[0].text);
      } catch (error) {
        lastError = error;
        const backoffMs = Math.pow(2, attempt) * 1000; // Exponential backoff
        console.log(`Attempt ${attempt} failed. Retrying in ${backoffMs}ms...`);
        await new Promise(resolve => setTimeout(resolve, backoffMs));
      }
    }
 
    return {
      source,
      status: "failed",
      error: lastError.message,
      confidence: 0
    };
  }
 
  // 複数ソースの並列取得
  async fetchMultipleSources(sources, query) {
    const fetchTasks = sources.map(source => 
      this.fetchFromSource(source, query)
    );
 
    const results = await Promise.allSettled(fetchTasks);
 
    return {
      timestamp: new Date().toISOString(),
      query,
      results: results.map((result, index) => ({
        source: sources[index],
        status: result.status,
        data: result.status === "fulfilled" ? result.value : { error: result.reason }
      })),
      successRate: results.filter(r => r.status === "fulfilled").length / sources.length
    };
  }
}
 
// 使用例
const fetcher = new ResilientDataFetcher(3, 30000);
const data = await fetcher.fetchMultipleSources(
  ["AlphaVantage", "CoinGecko", "OANDA"],
  "AAPL stock price"
);
 
console.log(`成功率: ${(data.successRate * 100).toFixed(1)}%`);
console.log("取得結果:", data.results);

実装パターン 2：依存関係グラフと順序実行

ユースケース：営業提案の自動生成ワークフロー

複数のステップ（顧客情報取得 → ニーズ分析 → 提案書作成 → 見積もり生成）が依存関係を持つ場合、適切な順序で実行し、各ステップの結果を次のステップに引き継ぐ。

class DependencyGraphExecutor {
  constructor() {
    this.client = new Anthropic();
    this.executionLog = {};
    this.resultCache = {};
  }
 
  // 依存関係グラフの定義
  async executeWorkflow(graph, inputs) {
    // トポロジカルソート
    const executionOrder = this.topologicalSort(graph);
 
    for (const taskId of executionOrder) {
      const task = graph[taskId];
 
      // 依存タスクの完了を待つ
      const dependencyResults = {};
      for (const depId of task.dependencies || []) {
        dependencyResults[depId] = this.resultCache[depId];
      }
 
      // タスク実行
      console.log(`Executing task: ${taskId}`);
      const result = await this.executeTask(
        taskId,
        task,
        { ...inputs, dependencies: dependencyResults }
      );
 
      // 結果キャッシュ
      this.resultCache[taskId] = result;
      this.executionLog[taskId] = {
        timestamp: new Date().toISOString(),
        status: "completed",
        result
      };
    }
 
    return this.resultCache;
  }
 
  // トポロジカルソート
  topologicalSort(graph) {
    const visited = new Set();
    const stack = [];
 
    const visit = (node) => {
      if (visited.has(node)) return;
      visited.add(node);
 
      const task = graph[node];
      for (const dep of task.dependencies || []) {
        visit(dep);
      }
 
      stack.push(node);
    };
 
    for (const node of Object.keys(graph)) {
      visit(node);
    }
 
    return stack;
  }
 
  // 個別タスク実行
  async executeTask(taskId, task, context) {
    const response = await this.client.messages.create({
      model: "claude-opus-4-6",
      max_tokens: 4096,
      messages: [
        {
          role: "user",
          content: `
            Execute task: ${task.description}
            Context: ${JSON.stringify(context, null, 2)}
            
            Output format: ${task.outputFormat || "natural language"}
          `
        }
      ]
    });
 
    return response.content[0].type === "text" ? response.content[0].text : null;
  }
}
 
// ワークフロー定義例
const proposalWorkflow = {
  "fetch_customer": {
    description: "Fetch customer data from CRM",
    dependencies: [],
    outputFormat: "JSON"
  },
  "analyze_needs": {
    description: "Analyze customer needs based on their profile",
    dependencies: ["fetch_customer"],
    outputFormat: "JSON"
  },
  "create_proposal": {
    description: "Generate proposal document",
    dependencies: ["fetch_customer", "analyze_needs"],
    outputFormat: "markdown"
  },
  "generate_quote": {
    description: "Generate pricing quote",
    dependencies: ["analyze_needs"],
    outputFormat: "JSON"
  }
};
 
const executor = new DependencyGraphExecutor();
const workflowResults = await executor.executeWorkflow(proposalWorkflow, {
  customerId: "CUST-001"
});
 
console.log("Workflow execution complete");
console.log("Proposal:", workflowResults.create_proposal);
console.log("Quote:", workflowResults.generate_quote);

実装パターン 3：動的なタスク分配と負荷分散

ユースケース：Webスクレイピングタスクの動的割り当て

複数のスクレイピング対象サイトがあり、各エージェントの処理能力や完了時間を考慮して、最適なタスク分配を行います。

class LoadBalancedTaskDistributor {
  constructor() {
    this.client = new Anthropic();
    this.agents = [];
    this.taskQueue = [];
    this.metrics = {}; // エージェント別メトリクス
  }
 
  // エージェント登録と初期化
  registerAgent(id, specialization, maxConcurrentTasks = 3) {
    const agent = {
      id,
      specialization,
      maxConcurrentTasks,
      currentTasks: 0,
      completedTasks: 0,
      averageExecutionTime: 0,
      successRate: 1.0
    };
    this.agents.push(agent);
    this.metrics[id] = agent;
  }
 
  // タスクをキューに追加
  enqueueTask(task) {
    this.taskQueue.push({
      id: Math.random().toString(36),
      ...task,
      createdAt: Date.now(),
      attempts: 0
    });
  }
 
  // 最適なエージェントを選択（スコアベース）
  selectBestAgent(task) {
    let bestAgent = null;
    let bestScore = -Infinity;
 
    for (const agent of this.agents) {
      // 利用可能性チェック
      if (agent.currentTasks >= agent.maxConcurrentTasks) continue;
 
      // スコア計算：能力適性 + 負荷 + パフォーマンス
      const loadScore = (agent.maxConcurrentTasks - agent.currentTasks) / agent.maxConcurrentTasks;
      const performanceScore = agent.successRate * (1 / (agent.averageExecutionTime || 1));
      
      let compatibilityScore = 0;
      if (task.requiredSkills) {
        const matchingSkills = task.requiredSkills.filter(skill =>
          agent.specialization.includes(skill)
        ).length;
        compatibilityScore = matchingSkills / task.requiredSkills.length;
      }
 
      const totalScore = (loadScore * 0.3) + (performanceScore * 0.4) + (compatibilityScore * 0.3);
 
      if (totalScore > bestScore) {
        bestScore = totalScore;
        bestAgent = agent;
      }
    }
 
    return bestAgent;
  }
 
  // タスク処理メインループ
  async processTaskQueue() {
    while (this.taskQueue.length > 0) {
      const task = this.taskQueue.shift();
      const agent = this.selectBestAgent(task);
 
      if (!agent) {
        console.log("No available agents. Re-queuing task.");
        this.taskQueue.push(task);
        await new Promise(resolve => setTimeout(resolve, 1000));
        continue;
      }
 
      // タスク実行
      agent.currentTasks++;
      const startTime = Date.now();
 
      this.executeTaskOnAgent(agent, task)
        .then(result => {
          const executionTime = Date.now() - startTime;
 
          // メトリクス更新
          agent.completedTasks++;
          agent.averageExecutionTime = 
            (agent.averageExecutionTime * (agent.completedTasks - 1) + executionTime) / 
            agent.completedTasks;
          agent.successRate = 0.95 * agent.successRate + 0.05; // 成功時
 
          console.log(`✓ Task ${task.id} completed by ${agent.id} in ${executionTime}ms`);
        })
        .catch(error => {
          // リトライ
          task.attempts++;
          if (task.attempts < 3) {
            console.log(`✗ Task ${task.id} failed. Retrying...`);
            this.taskQueue.push(task);
          }
          agent.successRate = 0.95 * agent.successRate; // 失敗時
        })
        .finally(() => {
          agent.currentTasks--;
        });
 
      // ワーカーに処理を委ねるため、少し待つ
      await new Promise(resolve => setTimeout(resolve, 100));
    }
  }
 
  // エージェントでのタスク実行
  async executeTaskOnAgent(agent, task) {
    const response = await this.client.messages.create({
      model: "claude-opus-4-6",
      max_tokens: 2048,
      messages: [
        {
          role: "user",
          content: `You are specialized in: ${agent.specialization.join(", ")}.
          
          Execute this scraping task:
          URL: ${task.url}
          Target data: ${task.targetData}
          
          Return extracted data as JSON.`
        }
      ]
    });
 
    return JSON.parse(response.content[0].text);
  }
}
 
// 使用例
const distributor = new LoadBalancedTaskDistributor();
distributor.registerAgent("scraper-1", ["ecommerce", "json-parsing"], 2);
distributor.registerAgent("scraper-2", ["news", "article-parsing"], 3);
distributor.registerAgent("scraper-3", ["finance", "table-data"], 2);
 
// タスク追加
[
  { url: "example-ecommerce.com", targetData: "product-prices", requiredSkills: ["ecommerce"] },
  { url: "news-site.com", targetData: "articles", requiredSkills: ["news"] },
  { url: "finance-site.com", targetData: "stock-tables", requiredSkills: ["finance"] }
].forEach(task => distributor.enqueueTask(task));
 
// 処理開始
await distributor.processTaskQueue();

パフォーマンスの最適化

トークン消費削減

マルチエージェント実装ではトークン消費が増加しやすいため、以下の最適化を適用：

class TokenEfficientOrchestrator {
  constructor() {
    this.client = new Anthropic();
  }
 
  // キャッシュの活用（同じクエリの重複実行を回避）
  async getCachedResponse(cacheKey, generator) {
    if (this.responseCache?.has(cacheKey)) {
      return this.responseCache.get(cacheKey);
    }
 
    const response = await generator();
    this.responseCache.set(cacheKey, response);
    return response;
  }
 
  // バッチ処理でトークンを効率化
  async batchExecute(tasks) {
    // 複数タスクを1リクエストで処理
    const batchedPrompt = tasks.map((task, i) => 
      `Task ${i + 1}: ${task.description}`
    ).join("\n\n");
 
    const response = await this.client.messages.create({
      model: "claude-opus-4-6",
      max_tokens: 4096,
      messages: [
        {
          role: "user",
          content: `Process all these tasks and return results:\n\n${batchedPrompt}`
        }
      ]
    });
 
    // 結果をパース
    return response.content[0].text;
  }
}

セッションをまたいでも止まらない——状態を外部化し、失敗を学習に変える

ここまでのパターンは、どれも「一度の実行」を前提にしています。並列化も依存グラフも負荷分散も、オーケストレータが立ち上がってから終わるまでの話です。

ところが実運用では、パイプラインは一度で終わりません。夜間に走らせて翌朝続きを見る。前半のエージェント群だけ先に流し、後半は結果を確認してから回す。こうした「またぎ」が入った瞬間、LLM は前回の文脈を丸ごと忘れます。オーケストレータをもう一度動かすと、どのタスクが済んでいるのか分からないまま最初からやり直しになります。

私自身、個人開発で Dolice Labs の記事生成をエージェントに任せていて、夜間ジョブが途中で止まった翌朝にこれで手が止まりました。ログを遡って「どこまで進んだか」を人間が組み立て直す時間が、実行そのものより長くなっていたのです。

オーケストレータの状態をファイルに逃がす

解決は単純です。オーケストレータが持っている進行状態を、実行の外——ファイルに書き出します。人間が読めるサマリと、機械が読み込む構造化状態の2枚に分けておくと扱いやすくなります。

from datetime import datetime
from pathlib import Path
import yaml
 
def save_orchestrator_state(run_dir: str, state: dict) -> None:
    """オーケストレータの進行状態を2つのファイルに書き出す。"""
    run_path = Path(run_dir)
    run_path.mkdir(parents=True, exist_ok=True)
 
    # 1) 人間がひと目で追えるサマリ
    summary = f"""# 実行状態（最終更新: {datetime.now():%Y-%m-%d %H:%M}）
 
## 現在のフェーズ
{state.get("current_phase", "未設定")}
 
## 完了したエージェント
{", ".join(state.get("done_agents", [])) or "なし"}
 
## 残っているタスク
{", ".join(state.get("pending_tasks", [])) or "なし"}
 
## 重要な決定
{state.get("decisions", "なし")}
"""
    (run_path / "run_state.md").write_text(summary, encoding="utf-8")
 
    # 2) 再開時に機械が読み込む構造化状態
    machine = {
        "updated_at": datetime.now().isoformat(),
        "current_phase": state.get("current_phase"),
        "done_agents": state.get("done_agents", []),
        "pending_tasks": state.get("pending_tasks", []),
    }
    with open(run_path / "run_state.yaml", "w", encoding="utf-8") as f:
        yaml.dump(machine, f, allow_unicode=True, default_flow_style=False)
 
 
def resume_orchestrator_state(run_dir: str) -> dict:
    """再開時に前回の状態を読み込む。無ければ空で始める。"""
    state_file = Path(run_dir) / "run_state.yaml"
    if not state_file.exists():
        return {"current_phase": None, "done_agents": [], "pending_tasks": []}
    with open(state_file, encoding="utf-8") as f:
        return yaml.safe_load(f) or {}

各エージェントの完了を done_agents に積み、次に流すタスクだけを pending_tasks に残す。再開時はまず resume_orchestrator_state を読み、済んだエージェントはスキップします。冒頭の「誰が何を待っているか」を一意の ID でログに流す話と、ここでつながります。ログは実行中の可視化、状態ファイルは実行と実行のあいだの記憶です。

失敗を次の実行に持ち越す

状態を外に出せると、もう一歩進めます。人間が返した修正指示を捨てずに貯め、同じ間違いを次の実行に持ち越さない仕組みです。

import re
from collections import Counter
from datetime import datetime
from pathlib import Path
 
def collect_recurring_corrections(run_dir: str, threshold: int = 3) -> dict:
    """修正ログから、閾値以上くり返された指摘だけを拾う。"""
    log_dir = Path(run_dir) / "_feedback"
    if not log_dir.exists():
        return {}
 
    corrections = []
    for log in log_dir.glob("*.md"):
        text = log.read_text(encoding="utf-8")
        corrections += re.findall(r"##\s*修正\s*\n(.*?)(?=##|\Z)", text, re.DOTALL)
 
    normalized = [c.strip()[:100] for c in corrections]
    counts = Counter(normalized)
    return {k: v for k, v in counts.items() if v >= threshold}
 
 
def stage_improvements(run_dir: str, recurring: dict) -> None:
    """くり返しの指摘を pending_improvements.md に『提案』として記録する。
    エージェントの指示書へは自動反映せず、人間のレビューを挟む。"""
    if not recurring:
        return
    pending = Path(run_dir) / "pending_improvements.md"
    stamp = datetime.now().strftime("%Y-%m-%d")
    lines = [f"- [{stamp}] {count}回: {issue[:80]}" for issue, count in recurring.items()]
    body = pending.read_text(encoding="utf-8") if pending.exists() else ""
    pending.write_text(body + f"\n\n## {stamp}\n" + "\n".join(lines), encoding="utf-8")
    print(f"{len(recurring)} 件を提案として記録しました。レビューのうえ反映してください。")

ここで意図的に一段止めているのが要点です。検出した問題を、エージェントの指示書へいきなり書き込まない。いったん提案として置き、人間が目を通してから反映します。オーケストレータが自分の設計図を知らないうちに書き換える——それが一番こわい失敗の形だからです。取り返しのつく形にしてから自動化する。この順序だけは崩さないようにしています。

次の実行では、承認済みの「既知の落とし穴」をエージェントのプロンプト先頭に差し込みます。同じ指摘を三度は受けない。それだけで、レビューに戻ってくる回数が目に見えて減りました。

全体を振り返って

Claude Code × Agent SDK を組み合わせたマルチエージェント・オーケストレーションは、以下の課題を一挙に解決します：

複雑なビジネスロジック — 並列実行・依存関係・エラーハンドリングを体系化
スケーラビリティ — エージェント数の増加に耐える設計
保守性 — オーケストレータレイヤーで統一管理
コスト効率 — キャッシング・バッチ処理でトークン消費を最小化

これらのパターンを習得することで、AI 駆動のエンタープライズワークフローが現実のものになります。

個人開発で Dolice Labs の運用を自動化していると、エージェントを増やすほど「誰が何を待っているか」が見えなくなるのを痛感します。私自身、最初はオーケストレータ層を省いて痛い目に遭い、いまは状態共有とエラー伝播だけは必ず1か所に集約するようにしています。複雑さは機能ではなく、設計で抑えるものだと考えています。具体的には、各エージェントの開始・終了・待機状態に一意の ID を振ってログへ流すだけでも、どこで誰が止まっているのかが一目で追えるようになり、障害切り分けの時間が大きく短くなりました。

Claude Code エージェント完全ガイドや Agent SDK 入門も参考にして、より深い理解を目指してください。