workflow-orchestration

工作流编排

Agent Skills™ 的一部分，由 googleadsagent.ai™ 提供

描述

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

单独的 AI 调用很有用；而编排后的工作流则能带来变革。内容流水线可以从单次输入触发，依次起草文本、核查事实、生成图像、按多个平台的要求进行格式整理并安排发布，从而省去数小时的人工协调工作。此技能编码了构建这类工作流的模式：基于 DAG 的执行、条件路由、重试逻辑和人在回路中的检查点。

MCP（模型上下文协议）插件系统通过数据库查询、文件操作、API 调用和浏览器自动化等外部功能扩展工作流。每个 MCP 工具都会成为工作流图中可重复使用的节点，使 Agent 能够与任何提供 MCP 接口的系统交互。工作流将这些节点组合成复杂的自动化流程，无需编写自定义集成代码。

适用场景

• 构建多步骤 AI 自动化流水线
• 将大模型调用与数据转换和 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——出现循环意味着设计存在缺陷，而非具备某种功能
• 为每个节点设置超时，防止外部服务导致无限期挂起
• 针对暂时性故障（网络问题、速率限制）实现指数退避重试
• 在不可逆操作（发布、删除、发送）之前设置人工审核检查点
• 记录每个节点执行的输入、输出、耗时和尝试次数
• 使用 MCP 工具进行外部集成，而非使用硬编码的 API 客户端

平台兼容性

平台	支持情况	备注
Cursor	完全支持	MCP 集成 + 工作流设计
VS Code	完全支持	基于扩展的工作流
Windsurf	完全支持	支持流程构建器
Claude Code	完全支持	工作流代码生成
Cline	完全支持	流水线编排
aider	部分支持	代码级工作流支持

相关技能

• 批处理
• AI 聊天工作室
• 并行 Agent 编排
• MCP 服务创建

关键词

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

---

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