⚡

编写智能体中继工作流Skill writing-agent-relay-workflows

本技能文档详细介绍了如何使用中继代理SDK构建和编排多智能体工作流系统。涵盖WorkflowBuilder API、DAG步骤依赖管理、智能体定义配置、步骤输出链式传递、验证门机制、专用通道设置、多种集群模式（如扇出、管道、Map-Reduce等）、错误处理策略和事件监听。适用于需要协调多个AI智能体（Claude、Codex、Gemini等）完成复杂任务的开发场景，提供完整的多智能体工作流编排解决方案。关键词：多智能体工作流、AI智能体编排、DAG依赖管理、集群模式、中继代理SDK、工作流自动化、智能体协同、验证门机制、错误处理、事件监听

AI智能体 0 次安装 2 次浏览 更新于 2/28/2026

name: 编写智能体中继工作流 description: 在使用中继代理SDK构建多智能体工作流时使用 - 涵盖WorkflowBuilder API、DAG步骤依赖、智能体定义、通过{{steps.X.output}}进行步骤输出链式传递、验证门、专用通道、集群模式、错误处理和事件监听器

编写智能体中继工作流

概述

中继代理SDK工作流系统通过类型化的基于DAG的工作流编排多个AI智能体（Claude、Codex、Gemini、Aider、Goose）。工作流通过流畅的构建器API或YAML文件定义。

使用场景

  • 构建具有步骤依赖关系的多智能体工作流
  • 编排不同的AI CLI工具（claude、codex、gemini、aider、goose）
  • 创建DAG、管道、扇出或其他集群模式
  • 需要验证门、重试或步骤输出链式传递

快速参考

import { workflow } from '../workflows/builder.js';

const result = await workflow('my-workflow')
  .description('此工作流的功能描述')
  .pattern('dag')                          // 或 'pipeline', 'fan-out' 等
  .channel('wf-my-workflow')               // 专用通道（如果省略则自动生成）
  .maxConcurrency(3)
  .timeout(3_600_000)                      // 全局超时（毫秒）

  .agent('lead',      { cli: 'claude', role: '架构师', retries: 2 })
  .agent('worker',    { cli: 'codex',  role: '实施者', retries: 2 })

  .step('plan', {
    agent: 'lead',
    task: `分析代码库并制定计划。
以以下内容结束：PLAN_COMPLETE`,
    retries: 2,
    verification: { type: 'output_contains', value: 'PLAN_COMPLETE' },
  })
  .step('implement', {
    agent: 'worker',
    task: `基于此计划实施：
{{steps.plan.output}}`,
    dependsOn: ['plan'],
  })

  .onError('retry', { maxRetries: 2, retryDelayMs: 10_000 })
  .run({ onEvent: (e) => console.log(e.type), vars: { task: '添加认证' } });

核心概念

步骤输出链式传递

在下游步骤的任务中使用{{steps.步骤名称.output}}来注入前一步骤的终端输出。运行器会自动捕获PTY输出。

验证门

步骤可以要求在智能体输出中包含特定字符串后才能标记为完成：

verification: { type: 'output_contains', value: '完成' }

其他类型：file_exists、exit_code、custom。

DAG依赖关系

具有dependsOn的步骤等待所有列出的步骤完成。没有依赖关系的步骤立即开始。具有相同dependsOn的步骤并行运行：

// 这两个步骤在'review'完成后并行运行：
.step('修复类型',  { agent: 'worker', dependsOn: ['review'], ... })
.step('修复测试',  { agent: 'worker', dependsOn: ['review'], ... })
// 此步骤等待两者都完成：
.step('最终步骤',  { agent: 'lead',   dependsOn: ['修复类型', '修复测试'], ... })

专用通道

始终设置.channel('wf-工作流名称')以实现工作流隔离。如果省略，运行器会自动生成wf-{名称}-{id}。切勿依赖general通道。

自终止

不要在任务字符串中添加退出指令。运行器会在spawnAndWait()中自动附加带有智能体运行时名称的自终止指令。

避免每个智能体的超时设置

除非有特定原因，否则避免在智能体/步骤上设置timeoutMs。全局.timeout()是安全网。每个智能体的超时设置会导致需要更多时间的步骤被过早终止。

智能体定义

.agent('名称', {
  cli: 'claude' | 'codex' | 'gemini' | 'aider' | 'goose' | 'opencode' | 'droid',
  role?: string,        // 描述智能体用途（用于模式自动选择）
  retries?: number,     // 使用此智能体的步骤的默认重试次数
  model?: string,       // 模型覆盖
  interactive?: boolean, // 默认：true。设置为false以使用非交互式子进程模式
})

步骤定义

.step('名称', {
  agent: string,                  // 必须匹配.agent()名称
  task: string,                   // 支持{{var}}和{{steps.名称.output}}
  dependsOn?: string[],           // DAG边
  verification?: VerificationCheck,
  retries?: number,               // 覆盖智能体级别的重试次数
})

事件监听器

.run({
  onEvent: (event) => {
    // event.type是以下之一：
    // 'run:started' | 'run:completed' | 'run:failed' | 'run:cancelled'
    // 'step:started' | 'step:completed' | 'step:failed' | 'step:skipped' | 'step:retrying'
  },
  vars: { key: 'value' },  // 用于{{key}}的模板变量
})

常见模式

并行评审（负责人和评审员同时运行）

.step('负责人评审', { agent: 'lead', dependsOn: ['implement'], ... })
.step('代码评审', { agent: 'reviewer', dependsOn: ['implement'], ... })
.step('下一阶段', { agent: 'worker', dependsOn: ['负责人评审', '代码评审'], ... })

管道（顺序交接）

.pattern('pipeline')
.step('分析', { agent: 'analyst', task: '...' })
.step('实施', { agent: 'dev', task: '{{steps.analyze.output}}', dependsOn: ['analyze'] })
.step('测试', { agent: 'tester', task: '{{steps.implement.output}}', dependsOn: ['implement'] })

错误处理策略

.onError('fail-fast')   // 在第一个失败时停止（默认）
.onError('continue')    // 跳过失败的分支，继续其他分支
.onError('retry', { maxRetries: 3, retryDelayMs: 5000 })

非交互式智能体

对于扇出和Map-Reduce等集群模式，只需要执行任务并返回输出的工作器不需要完整的PTY/中继消息传递开销。设置interactive: false以将它们作为简单的子进程运行：

.agent('worker', {
  cli: 'codex',
  interactive: false,  // 运行"codex exec <task>"，无PTY，无中继消息传递
  role: '后端工程师',
})

interactive: false的变化：

  • 智能体通过CLI一次性模式运行（例如，claude -p、codex exec、gemini -p）
  • 无PTY包装，无stdin透传，无/exit自终止
  • 无中继消息传递 - 智能体无法发送或接收消息
  • 输出从stdout捕获，可通过{{steps.X.output}}使用
  • 负责人智能体会自动获知哪些工作器是非交互式的
  • 比交互式模式启动更快，开销更低

使用时机：

  • 只需处理任务并返回结果的扇出工作器
  • 不需要任务中通信的Map-Reduce映射器
  • 任何不需要逐轮中继消息传递的智能体

不应使用的时机：

  • 需要与其他智能体通信的负责人/协调器智能体
  • 需要接收消息或参与通道的智能体
  • 参与辩论、共识或反思模式的智能体

常见错误

错误 修复方法
在任务中添加withExit()或退出指令 运行器会自动处理
在智能体上设置严格的timeoutMs 仅使用全局.timeout()
使用general通道 设置.channel('wf-名称')以实现隔离
引用{{steps.X.output}}而没有dependsOn: ['X'] 输出尚不可用
评审步骤可以并行时却设置为串行 两个评审员可以依赖相同的上游步骤
关键步骤未使用验证门 添加带有完成标记的output_contains

YAML替代方案

工作流也可以定义为.yaml文件：

version: "1.0"
name: 我的工作流
swarm:
  pattern: dag
  channel: wf-我的工作流
agents:
  - name: lead
    cli: claude
    role: 架构师
  - name: worker
    cli: codex
    role: 实施者
workflows:
  - name: default
    steps:
      - name: plan
        agent: lead
        task: "制定计划。以以下内容结束：PLAN_COMPLETE"
        verification:
          type: output_contains
          value: PLAN_COMPLETE
      - name: implement
        agent: worker
        task: "实施：{{steps.plan.output}}"
        dependsOn: [plan]

运行方式：agent-relay run 路径/到/workflow.yaml

可用集群模式

dag（默认）、fan-out、pipeline、hub-spoke、consensus、mesh、handoff、cascade、debate、hierarchical、map-reduce、scatter-gather、supervisor、reflection、red-team、verifier、auction、escalation、saga、circuit-breaker、blackboard、swarm

有关模式选择指南，请参阅技能choosing-swarm-patterns。