workflow-orchestration

ワークフローオーケストレーション

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

説明

ワークフローオーケストレーションは、MCP プラグインのサポート、条件分岐、並列実行、エラー回復機能を備えた、視覚的な AI フロー構築フレームワークを提供します。エージェントは、AI 操作、データ変換、外部サービス呼び出しを連鎖させ、信頼性が高く繰り返し実行できる自動化パイプラインを構築する、複数ステップのワークフローを設計・実装します。

個々の AI 呼び出しも有用ですが、オーケストレーションされたワークフローは大きな変革をもたらします。テキストの下書き、主張のファクトチェック、画像の生成、複数のプラットフォーム向けのフォーマット、公開日時の設定を一度に行うコンテンツパイプラインは、単一の入力で起動し、何時間もの手作業による調整を置き換えます。このスキルでは、こうしたワークフローを構築するためのパターン、すなわち DAG ベースの実行、条件付きルーティング、再試行ロジック、ヒューマン・イン・ザ・ループのチェックポイントを体系化しています。

MCP（Model Context Protocol）プラグインシステムは、データベースクエリ、ファイル操作、API 呼び出し、ブラウザー自動化といった外部機能でワークフローを拡張します。各 MCP ツールはワークフローグラフ内で再利用可能なノードとなり、エージェントは MCP インターフェースを公開するあらゆるシステムを操作できます。ワークフローでは、カスタムの統合コードを使わずに、これらのノードを組み合わせて複雑な自動化を構成できます。

使用する場面

• 複数ステップの AI 自動化パイプラインを構築するとき
• LLM 呼び出しをデータ変換や API 統合と連鎖させるとき
• AI ワークフローに条件ロジック（if/else、switch）を実装するとき
• 自動化されたプロセスにヒューマン・イン・ザ・ループの承認ステップを追加するとき
• MCP ツールを再現可能なワークフローに統合するとき
• 依存関係を管理しながら、AI タスクを並列にオーケストレーションするとき

動作の仕組み

graph TD
    A[Trigger: Manual / Schedule / Webhook] --> B[Node 1: Input Parser]
    B --> C{Conditional Router}
    C -->|Type A| D[Node 2a: LLM Analysis]
    C -->|Type B| E[Node 2b: Database Query via MCP]
    D --> F[Node 3: Parallel Execution]
    E --> F
    F --> G[Node 3a: Image Generation]
    F --> H[Node 3b: Content Formatting]
    G --> I[Node 4: Human Review Checkpoint]
    H --> I
    I -->|Approved| J[Node 5: Publish via MCP]
    I -->|Rejected| K[Node 5: Return to Step 2]
    J --> L[Node 6: Log + Notify]

ワークフローは、有向非巡回グラフ（DAG）で構成され、各ノードは個別の処理を表します。オーケストレーターは実行順序を管理し、ノード間でデータを受け渡し、失敗時の再試行を処理し、ヒューマンチェックポイントで一時停止します。

実装

interface WorkflowNode {
  id: string;
  type: "llm" | "mcp" | "transform" | "condition" | "human_review" | "parallel";
  config: Record<string, unknown>;
  next: string | string[] | ConditionalNext[];
  retries?: number;
  timeout_ms?: number;
}

interface ConditionalNext {
  condition: string;
  target: string;
}

interface Workflow {
  id: string;
  name: string;
  trigger: { type: "manual" | "schedule" | "webhook"; config: Record<string, unknown> };
  nodes: WorkflowNode[];
}

class WorkflowEngine {
  private state = new Map<string, unknown>();

  async execute(workflow: Workflow, input: unknown): Promise<unknown> {
    this.state.set("input", input);
    let currentNode = workflow.nodes[0];

    while (currentNode) {
      const result = await this.executeNode(currentNode);
      this.state.set(currentNode.id, result);

      const nextId = this.resolveNext(currentNode, result);
      currentNode = nextId ? workflow.nodes.find(n => n.id === nextId)! : undefined!;
    }

    return Object.fromEntries(this.state);
  }

  private async executeNode(node: WorkflowNode): Promise<unknown> {
    for (let attempt = 0; attempt <= (node.retries ?? 0); attempt++) {
      try {
        switch (node.type) {
          case "llm": return await this.executeLLM(node.config);
          case "mcp": return await this.executeMCP(node.config);
          case "transform": return this.executeTransform(node.config);
          case "parallel": return await this.executeParallel(node.config);
          case "human_review": return await this.waitForHumanReview(node.config);
          default: throw new Error(`Unknown node type: ${node.type}`);
        }
      } catch (error) {
        if (attempt === (node.retries ?? 0)) throw error;
        await this.delay(1000 * Math.pow(2, attempt));
      }
    }
  }

  private async executeMCP(config: Record<string, unknown>): Promise<unknown> {
    const { server, tool, arguments: args } = config as {
      server: string; tool: string; arguments: Record<string, unknown>;
    };
    return callMcpTool(server, tool, this.interpolate(args));
  }

  private async executeParallel(config: Record<string, unknown>): Promise<unknown[]> {
    const { nodeIds } = config as { nodeIds: string[] };
    const nodes = nodeIds.map(id => this.findNode(id));
    return Promise.all(nodes.map(n => this.executeNode(n)));
  }

  private interpolate(obj: Record<string, unknown>): Record<string, unknown> {
    return JSON.parse(
      JSON.stringify(obj).replace(/\{\{(\w+)\.(\w+)\}\}/g, (_, nodeId, key) => {
        const nodeResult = this.state.get(nodeId) as Record<string, unknown>;
        return String(nodeResult?.[key] ?? "");
      })
    );
  }
}

ベストプラクティス

• ワークフローは DAG として設計します。循環は機能ではなく、設計上の欠陥を示します
• 外部サービスが無期限に応答しなくなるのを防ぐため、すべてのノードにタイムアウトを設定します
• 一時的な障害（ネットワーク、レート制限）には、指数バックオフによる再試行を実装します
• 取り消せない操作（公開、削除、送信）の前に、人間によるレビューのチェックポイントを設けます
• 入力、出力、所要時間、試行回数を含め、すべてのノード実行をログに記録します
• 外部サービスの統合には、ハードコードされた API クライアントではなく MCP ツールを使用します

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

プラットフォーム	サポート	備考
Cursor	完全	MCP 統合とワークフロー設計
VS Code	完全	拡張機能ベースのワークフロー
Windsurf	完全	フロービルダーをサポート
Claude Code	完全	ワークフローコードの生成
Cline	完全	パイプラインのオーケストレーション
aider	一部	コードレベルのワークフローをサポート

関連スキル

• バッチ処理
• AI チャットスタジオ
• 並列エージェントオーケストレーション
• MCP サーバーの作成

キーワード

workflow orchestration mcp dag parallel-execution conditional-routing human-in-the-loop automation

---

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