parallel-agent-orchestration

並列エージェントオーケストレーション

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

説明

並列エージェントオーケストレーションとは、複雑なタスクを大幅に高速化するために、複数のサブエージェントを同時に起動し、連携させ、その結果を集約する技術です。多くの AI ワークフローでは、単一エージェントによる逐次実行がデフォルトのモードですが、これでは大きなパフォーマンスの可能性が活用されません。複数のキャンペーンの分析、複数のファイルのレビュー、複数のデータソースの検索など、タスクを独立したサブタスクに分割できる場合、並列に処理を割り当てることで、結果の品質を保ちながら実時間を 3～10 倍短縮できます。

このスキルには、Superpowers 拡張エコシステム向けに開発され、googleadsagent.ai™ で本番運用されているサブエージェントオーケストレーションのパターンが組み込まれています。Buddy™ はそこで、Google Ads アカウントのさまざまな側面を同時に分析するため、日常的にサブエージェントを並列で起動しています。あるサブエージェントは入札戦略を分析し、別のサブエージェントはキーワードのパフォーマンスを評価し、3 つ目はクリエイティブの品質を評価します。すべてが同時に実行されます。オーケストレーターはその後、各分析間の矛盾を解決しながら、調査結果を統合された推奨事項のセットに集約します。

並列オーケストレーションにおける主な課題は、タスクの分割（作業を真に独立した単位に分解すること）、結果の集約（矛盾または重複する可能性のある出力を組み合わせること）、リソース管理（並列エージェント間でレート制限とコスト予算を遵守すること）、進捗の監視（可視性を失わずに複数の同時進行の処理を追跡すること）です。

使用する場面

• タスクを自然に 3 つ以上の独立したサブタスクに分割できる場合
• 実時間が重要な制約となる場合（ユーザーの待ち時間、SLA 要件）
• 複数のデータソースやドキュメントを同時に分析する必要がある場合
• 独立してレビューできる多数のファイルにまたがるコードレビュー
• 複数のファイルやサービスにまたがる一括操作（移行、リファクタリング）
• 同じ問題について多様な視点が必要な場合（アンサンブル推論）

仕組み

graph TD
    A[Complex Task] --> B[Task Decomposer]
    B --> C[Subtask 1]
    B --> D[Subtask 2]
    B --> E[Subtask 3]
    B --> F[Subtask N]
    
    C --> G[Subagent 1]
    D --> H[Subagent 2]
    E --> I[Subagent 3]
    F --> J[Subagent N]
    
    G --> K[Result Aggregator]
    H --> K
    I --> K
    J --> K
    
    K --> L{Conflicts?}
    L -->|Yes| M[Conflict Resolver]
    L -->|No| N[Unified Result]
    M --> N
    
    O[Resource Monitor] --> G
    O --> H
    O --> I
    O --> J

オーケストレーターは複雑なタスクを受け取り、タスク分解器（既知のパターンにはルールベース、新しいタスクにはモデル支援型）を使って、独立したサブタスクに分解します。各サブタスクは、それぞれ独立して実行するサブエージェントに割り当てられ、リソースモニターが共有のレート制限と予算の制約を適用します。サブエージェントが完了すると、その結果は集約器に送られ、出力の統合と矛盾の検出が行われます。矛盾する結果（例：サブエージェント 1 が入札額の引き上げを推奨する一方、サブエージェント 2 は引き下げを推奨する場合）は、ドメインルールを適用する競合解決器によって解決されるか、判断を仰ぐためオーケストレーションを担当するエージェントにエスカレーションされます。

実装

タスク分解器：

interface SubTask {
  id: string;
  description: string;
  context: Record<string, unknown>;
  dependencies: string[];
  priority: number;
}

class TaskDecomposer {
  decompose(task: string, context: Record<string, unknown>): SubTask[] {
    const patterns: Record<string, (ctx: Record<string, unknown>) => SubTask[]> = {
      account_audit: (ctx) => {
        const campaigns = ctx.campaigns as string[];
        return campaigns.map((campaign, i) => ({
          id: `campaign_${i}`,
          description: `Analyze campaign: ${campaign}`,
          context: { campaign, metrics: ctx.metrics },
          dependencies: [],
          priority: 1,
        }));
      },
      code_review: (ctx) => {
        const files = ctx.changedFiles as string[];
        return files.map((file, i) => ({
          id: `review_${i}`,
          description: `Review changes in ${file}`,
          context: { file, diff: ctx.diffs?.[file] },
          dependencies: [],
          priority: file.includes("test") ? 2 : 1,
        }));
      },
    };

    const taskType = this.classifyTask(task);
    const decomposer = patterns[taskType];
    return decomposer ? decomposer(context) : [{ id: "single", description: task, context, dependencies: [], priority: 1 }];
  }

  private classifyTask(task: string): string {
    if (task.includes("audit") || task.includes("account")) return "account_audit";
    if (task.includes("review") || task.includes("PR")) return "code_review";
    return "generic";
  }
}

並列オーケストレーター：

import asyncio

class ParallelOrchestrator:
    def __init__(self, max_concurrency=5, budget_limit=None):
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self.budget_limit = budget_limit
        self.total_tokens = 0
        self.results = {}

    async def execute(self, subtasks: list[dict], agent_factory) -> dict:
        dependency_graph = self.build_dependency_graph(subtasks)
        ready = [t for t in subtasks if not t["dependencies"]]
        pending = [t for t in subtasks if t["dependencies"]]

        while ready or pending:
            batch_results = await asyncio.gather(*[
                self.run_subtask(task, agent_factory) for task in ready
            ], return_exceptions=True)

            for task, result in zip(ready, batch_results):
                if isinstance(result, Exception):
                    self.results[task["id"]] = {"success": False, "error": str(result)}
                else:
                    self.results[task["id"]] = {"success": True, "result": result}

            completed_ids = set(self.results.keys())
            ready = [t for t in pending if all(d in completed_ids for d in t["dependencies"])]
            pending = [t for t in pending if t not in ready]

        return self.results

    async def run_subtask(self, task: dict, agent_factory):
        async with self.semaphore:
            if self.budget_limit and self.total_tokens >= self.budget_limit:
                raise BudgetExceededError(f"Token budget {self.budget_limit} exceeded")
            agent = agent_factory(task)
            result = await agent.execute(task["description"], task["context"])
            self.total_tokens += result.get("tokens_used", 0)
            return result

    def build_dependency_graph(self, subtasks):
        return {t["id"]: t["dependencies"] for t in subtasks}

競合解決機能を備えた結果集約器：

class ResultAggregator:
    def aggregate(self, results: dict, strategy: str = "merge") -> dict:
        successful = {k: v for k, v in results.items() if v["success"]}
        failed = {k: v for k, v in results.items() if not v["success"]}

        if strategy == "merge":
            merged = self.merge_results(successful)
        elif strategy == "vote":
            merged = self.majority_vote(successful)
        else:
            merged = self.concatenate_results(successful)

        conflicts = self.detect_conflicts(successful)
        if conflicts:
            merged = self.resolve_conflicts(merged, conflicts)

        return {
            "aggregated_result": merged,
            "subtask_count": len(results),
            "success_count": len(successful),
            "failure_count": len(failed),
            "conflicts_resolved": len(conflicts),
            "failed_tasks": list(failed.keys()),
        }

    def detect_conflicts(self, results: dict) -> list[dict]:
        conflicts = []
        recommendations = {}
        for task_id, result in results.items():
            for rec in result.get("result", {}).get("recommendations", []):
                key = rec.get("target")
                if key in recommendations:
                    if recommendations[key]["action"] != rec["action"]:
                        conflicts.append({
                            "target": key,
                            "conflict": [recommendations[key], rec],
                            "tasks": [recommendations[key]["source"], task_id],
                        })
                recommendations[key] = {**rec, "source": task_id}
        return conflicts

    def resolve_conflicts(self, merged: dict, conflicts: list) -> dict:
        for conflict in conflicts:
            higher_confidence = max(conflict["conflict"], key=lambda c: c.get("confidence", 0))
            merged["recommendations"] = [
                r for r in merged.get("recommendations", [])
                if r.get("target") != conflict["target"]
            ]
            merged["recommendations"].append(higher_confidence)
            merged.setdefault("conflict_notes", []).append(
                f"Conflict on {conflict['target']}: chose {higher_confidence['action']} (confidence: {higher_confidence.get('confidence', 'N/A')})"
            )
        return merged

ベストプラクティス

1. 本当に独立したタスクのみを並列化する — 可変状態を共有するサブタスクや、相互の出力に依存するサブタスクは、並列化せずに順番に実行する必要があります。
2. 同時実行数に上限を設ける — 無制限の並列処理はレート制限に達し、コストも増加します。ほとんどのユースケースでは、同時に実行するサブエージェントを3〜5個に制限してください。
3. サブタスクごとと全体の予算を設定する — サブエージェントごとのトークン数と、オーケストレーション全体のトークン数に上限を設け、コストの急増を防ぎます。
4. 部分的な失敗を適切に処理する — 10個中8個のサブエージェントが成功した場合は、オーケストレーション全体を失敗させず、8個の結果を集約します。
5. 競合を明示的に検出して解決する — サブエージェントが矛盾する推奨事項を提示した場合は、競合を明らかにし、解決方針（信頼度に基づく判断、多数決、またはエスカレーション）を適用します。
6. 進行状況をリアルタイムで監視する — どのサブエージェントが実行中、完了、または失敗したかを示すダッシュボードやステータス更新を提供します。
7. 並列サブタスクには同じモデルのティアを使用する — 1つの並列バッチ内でモデルのティアを混在させると、品質に一貫性がなくなります。比較可能な結果を得るには、統一したモデルを使用してください。

プラットフォームの互換性

機能	Claude Code	Cursor	Codex	Gemini CLI
サブエージェントのディスパッチ	✅ ネイティブのサブエージェント	✅ Task ツール	✅ 非同期タスク	✅ 非同期タスク
並列実行	✅ 完全対応	✅ 完全対応	✅ 完全対応	✅ 完全対応
同時実行の制御	✅ カスタム	✅ カスタム	✅ カスタム	✅ カスタム
結果の集約	✅ 完全対応	✅ 完全対応	✅ 完全対応	✅ 完全対応
進行状況の監視	✅ ステータス更新	✅ バックグラウンドタスク	✅ カスタム	✅ カスタム

Mythos Preview リファレンス

Anthropic の Mythos Preview のワークロードは、多数の Claude インスタンスを並列実行し、それぞれを異なるファイルに集中させることで拡張します。これにより、並列実行ごとに同じ問題を再発見するのではなく、異なる範囲を調査できます。結果はその後、下流工程で（検証パスを含めて）集約され、重複が排除されます。

オーケストレーションでは、ファイル（またはモジュール）の境界を自然な分割キーとして扱い、予算を守るために同時実行数を制限し、部分的な出力をスムーズに統合できるよう、集約の契約を標準化してください。出典：Mythos Preview。

関連スキル

• 長期ワークフロー - 並列ディスパッチにより、複数フェーズのワークフローにおける各フェーズの実行を高速化します
• マルチモデルルーティング - 並列サブエージェントに最適なモデルを選択し、コストと品質のバランスを取ります
• トークン最適化 - サブタスクごとと全体の予算上限により、並列実行時のコストの急増を防ぎます
• Google Ads 監査 - 複数キャンペーンの監査分析は、代表的な並列オーケストレーションのユースケースです

キーワード

parallel-orchestration, subagents, task-decomposition, result-aggregation, conflict-resolution, concurrency, resource-management, batch-processing, ensemble-reasoning, agent-skills

---

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