parallel-agent-orchestration

并行 Agent 编排

属于 Agent Skills™，由 googleadsagent.ai™ 提供

描述

并行 Agent 编排是一门调度、协调多个并发子 Agent 并汇总其结果的实践，可显著加速复杂任务。大多数 AI 工作流默认采用顺序式单 Agent 执行，但这会浪费大量性能。当任务可以拆分为相互独立的子任务时——例如分析多个广告系列、审查多个文件、搜索多个数据源——并行调度可将实际耗时缩短 3–10 倍，同时保持结果质量。

这项技能将为 Superpowers 扩展生态系统开发的子 Agent 编排模式编码，并已在 googleadsagent.ai™ 投入生产使用。在该平台上，Buddy™ 会定期调度多个并行子 Agent，同时分析 Google Ads 账户的不同方面。一个子 Agent 分析出价策略，另一个评估关键词效果，第三个评估创意质量——所有任务同时运行。随后，编排器会汇总各项发现，形成统一的建议集，并解决各项独立分析之间的冲突。

并行编排的主要挑战包括任务划分（将工作分解为真正独立的单元）、结果汇总（合并可能相互冲突或重叠的输出）、资源管理（在并行 Agent 之间遵守速率限制和成本预算）以及进度监控（跟踪多个并发任务流，同时保持全局可见性）。

适用场景

• 任务可自然拆分为 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

编排器接收复杂任务，并使用任务分解器将其拆分为独立子任务（对于已知模式，可采用基于规则的方法；对于新颖任务，可采用模型辅助的方法）。每个子任务都会分派给一个独立执行的子 Agent，同时由资源监控器执行共享的速率限制和预算约束。子 Agent 完成任务后，其结果会传递给汇总器，由汇总器合并输出并检测冲突。若结果相互冲突（例如，子 Agent 1 建议提高出价，而子 Agent 2 建议降低出价），冲突解决器会应用领域规则来解决，或将问题升级给编排 Agent 判断。

实现

任务分解器：

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. 限制并发数——无限制并行会触发速率限制并增加成本；大多数场景下，将并发子 Agent 数量限制在 3–5 个。
3. 设置每个子任务和总体预算——限制每个子 Agent 的令牌数和编排过程中的总令牌数，避免成本失控。
4. 妥善处理部分失败——如果 10 个子 Agent 中有 8 个成功，应汇总这 8 个结果，而不是让整个编排失败。
5. 明确检测并解决冲突——如果子 Agent 给出相互矛盾的建议，应指出冲突并采用解决策略（基于置信度、少数服从多数或升级处理）。
6. 实时监控进度——提供仪表板或状态更新，显示哪些子 Agent 正在运行、已完成或失败。
7. 并行子任务使用相同的模型级别——同一并行批次混用不同模型级别会导致质量不一致；对于可比较的结果，应使用统一的模型。

平台兼容性

功能	Claude Code	Cursor	Codex	Gemini CLI
子 Agent 分派	✅ 原生子 Agent	✅ Task 工具	✅ 异步任务	✅ 异步任务
并行执行	✅ 完整支持	✅ 完整支持	✅ 完整支持	✅ 完整支持
并发控制	✅ 自定义	✅ 自定义	✅ 自定义	✅ 自定义
结果汇总	✅ 完整支持	✅ 完整支持	✅ 完整支持	✅ 完整支持
进度监控	✅ 状态更新	✅ 后台任务	✅ 自定义	✅ 自定义

Mythos Preview 参考

Anthropic 的 Mythos Preview 工作负载通过并行运行多个 Claude 实例来扩展规模，每个实例专注于不同的文件，使并行运行能够探索不同的范围，而不是反复发现同一个问题。随后会在下游汇总并去重结果（包括验证环节）。

在编排时，应将文件（或模块）边界视为自然的分片键，限制并发数以控制预算，并制定标准化的汇总约定，以便干净地合并部分输出。来源：Mythos Preview。

相关技能

• 长周期工作流 - 并行分派可加速多阶段工作流中的各个阶段
• 多模型路由 - 为并行子 Agent 选择最优模型，兼顾成本与质量
• 令牌优化 - 每个子任务和总体预算限制可防止并行执行时成本失控
• Google Ads 审计 - 多广告系列审计分析是典型的并行编排用例

关键词

并行编排、子 Agent、任务分解、结果汇总、冲突解决、并发、资源管理、批处理、集成推理、Agent 技能

---

© 2026 googleadsagent.ai™ | Agent Skills™ | MIT 许可证