workflow-orchestration

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Workflow Orchestration

工作流编排

Part of Agent Skills™ by googleadsagent.ai™

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

Description

描述

Workflow Orchestration provides a visual AI flow-building framework with MCP plugin support, conditional branching, parallel execution, and error recovery. The agent designs and implements multi-step workflows that chain AI operations, data transformations, and external service calls into reliable, repeatable automation pipelines.

Individual AI calls are useful; orchestrated workflows are transformative. A content pipeline that drafts text, fact-checks claims, generates images, formats for multiple platforms, and schedules publication—all triggered by a single input—replaces hours of manual coordination. This skill encodes the patterns for building such workflows: DAG-based execution, conditional routing, retry logic, and human-in-the-loop checkpoints.

The MCP (Model Context Protocol) plugin system extends workflows with external capabilities: database queries, file operations, API calls, and browser automation. Each MCP tool becomes a reusable node in the workflow graph, enabling agents to interact with any system that exposes an MCP interface. Workflows compose these nodes into complex automations without custom integration code.

工作流编排提供支持 MCP 插件的可视化 AI 工作流构建框架，并具备条件分支、并行执行和错误恢复能力。Agent 负责设计并实现多步骤工作流，将 AI 操作、数据转换和外部服务调用串联成可靠、可重复执行的自动化流水线。

单独进行 AI 调用固然有用；经过编排的工作流则能带来根本性变革。一个内容流水线可以完成文本起草、事实核查、图像生成、针对多个平台进行格式化以及安排发布——所有这些操作都由单一输入触发，从而取代数小时的手动协调。该 skill 编码了构建此类工作流的模式：基于 DAG 的执行、条件路由、重试逻辑和 human-in-the-loop 检查点。

MCP（Model Context Protocol）插件系统使用外部能力扩展工作流，包括数据库查询、文件操作、API 调用和浏览器自动化。每个 MCP 工具都会成为工作流图中的可复用节点，使 Agent 能够与任何暴露 MCP 接口的系统交互。工作流可以将这些节点组合成复杂的自动化流程，而无需编写自定义集成代码。

Use When

适用场景

Building multi-step AI automation pipelines
Chaining LLM calls with data transformations and API integrations
Implementing conditional logic in AI workflows (if/else, switch)
Adding human-in-the-loop approval steps to automated processes
Integrating MCP tools into repeatable workflows
Orchestrating parallel AI tasks with dependency management

构建多步骤 AI 自动化流水线
将 LLM 调用与数据转换和 API 集成串联起来
在 AI 工作流中实现条件逻辑（if/else、switch）
为自动化流程添加 human-in-the-loop 审批步骤
将 MCP 工具集成到可重复执行的工作流中
编排并行 AI 任务并管理依赖关系

How It Works

工作原理

mermaid

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]

Workflows are directed acyclic graphs (DAGs) where each node is a discrete operation. The orchestrator manages execution order, passes data between nodes, handles retries on failure, and pauses at human checkpoints.

mermaid

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），其中每个节点都是一个离散操作。编排器负责管理执行顺序、在节点之间传递数据、处理失败重试，并在 human-in-the-loop 检查点暂停。

Implementation

实现

typescript

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] ?? "");
      })
    );
  }
}