AI Agent与Cadence入门实战：从零构建高可靠工作流引擎

1次阅读

没有评论

共计 3923 个字符，预计需要花费 10 分钟才能阅读完成。

在开发 AI Agent 时，任务调度和状态管理是两大核心挑战。传统方案如 Celery 虽然简单易用，但在处理长周期、多步骤的复杂工作流时存在明显不足：

状态管理困难：Celery 默认不持久化任务状态，需要额外开发
错误恢复复杂：任务失败后需要手动实现重试和补偿逻辑
协调能力有限：难以实现跨任务的复杂编排逻辑

Cadence 作为 Uber 开源的分布式工作流引擎，提供了以下优势：

内置持久化状态存储
自动错误恢复和重试机制
原生支持复杂工作流模式
完善的版本管理和信号机制

使用 Docker 快速搭建 Cadence 开发环境是最佳选择。以下是 docker-compose.yml 配置片段：

version: '3'
services:
  cadence:
    image: ubercadence/server:0.22.4
    ports:
      - "7933:7933" # 前端端口
      - "7934:7934" # 后端端口
    environment:
      - CASSANDRA_SEEDS=cassandra
      - DYNAMIC_CONFIG_FILE_PATH=config/dynamicconfig/development.yaml

  cadence-web:
    image: ubercadence/web:3.10.2
    ports:
      - "8088:8088"
    environment:
      - CADENCE_TCHANNEL_PEERS=cadence:7933

  cassandra:
    image: cassandra:3.11
    ports:
      - "9042:9042"

启动命令：

docker-compose up -d

package main

import (
    "go.uber.org/cadence/workflow"
    "go.uber.org/zap"
)

// 定义 Activity
func SampleActivity(ctx workflow.Context, input string) (string, error) {logger := workflow.GetLogger(ctx)
    logger.Info("Activity started", zap.String("input", input))
    // 模拟业务逻辑
    return "Processed:" + input, nil
}

// 定义 Workflow
func SampleWorkflow(ctx workflow.Context, name string) (string, error) {
    // 设置工作流选项
    ao := workflow.ActivityOptions{
        ScheduleToStartTimeout: time.Minute,
        StartToCloseTimeout:    time.Minute,
        HeartbeatTimeout:       time.Second * 20,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)

    var result string
    err := workflow.ExecuteActivity(ctx, SampleActivity, name).Get(ctx, &result)
    if err != nil {return "", err}

    return result, nil
}

func main() {
    // 注册 Workflow 和 Activity
    worker := worker.New(...)
    worker.RegisterWorkflow(SampleWorkflow)
    worker.RegisterActivity(SampleActivity)
    worker.Start()}

// Workflow 中接收信号
func SignalWorkflow(ctx workflow.Context) error {
    var signalVal string
    signalChan := workflow.GetSignalChannel(ctx, "signal-name")

    selector := workflow.NewSelector(ctx)
    selector.AddReceive(signalChan, func(c workflow.Channel, more bool) {c.Receive(ctx, &signalVal)
        // 处理信号
    })

    // 等待信号
    selector.Select(ctx)
    return nil
}

// 客户端发送信号
client.SignalWorkflow(ctx, "workflow-id", "","signal-name", data)

// Workflow 中定义查询处理器
func QueryWorkflow(ctx workflow.Context) error {
    queryType := "state"
    err := workflow.SetQueryHandler(ctx, queryType, func() (string, error) {return "current state", nil})
    if err != nil {return err}
    // ...
}

// 客户端查询
response, err := client.QueryWorkflow(ctx, "workflow-id", "","state")

ao := workflow.ActivityOptions{
    ScheduleToStartTimeout: time.Minute * 5,
    StartToCloseTimeout:    time.Minute * 10,
    HeartbeatTimeout:       time.Second * 30,
    RetryPolicy: &cadence.RetryPolicy{
        InitialInterval:    time.Second,
        BackoffCoefficient: 2.0,
        MaximumInterval:    time.Minute,
        ExpirationInterval: time.Minute * 10,
        MaximumAttempts:    5,
    },
}

使用 workflow.GetVersion 检查版本
为新版本工作流添加标记
分阶段逐步迁移

version := workflow.GetVersion(ctx, "change-id", workflow.DefaultVersion, 1)
if version == workflow.DefaultVersion {// 旧逻辑} else {// 新逻辑}

Cadence 原生支持 Prometheus 指标，配置示例：

statsd:
  hostPort: "prometheus:9125"
  prefix: "cadence"
  flushInterval: "1s"

Activity 心跳超时
问题：长时间运行的 Activity 未发送心跳导致超时
解决：定期调用workflow.RecordHeartbeat
工作流无限循环
问题：工作流逻辑陷入死循环
解决：设置合理的workflow.ExecutionStartToCloseTimeout
信号丢失
问题：工作流未运行时发送信号导致丢失
解决：先启动工作流再发送信号
版本管理混乱
问题：多版本工作流并存导致逻辑错误
解决：使用 workflow.GetVersion 进行版本切换
资源泄漏
问题：未正确关闭 Worker 导致资源泄漏
解决：确保 worker.Stop() 被调用

实现带人工审批节点的 AI 工作流

任务要求：
1. 工作流执行到特定步骤暂停，等待人工审批
2. 审批通过后继续执行，拒绝则终止
3. 审批超时 (如 24 小时) 自动处理

提示：
– 使用 Signal 机制接收审批结果
– 结合 Timer 实现超时处理
– 考虑审批记录的持久化

示例框架：

func ApprovalWorkflow(ctx workflow.Context, input string) error {
    // ...

    // 等待审批
    var approvalResult bool
    signalChan := workflow.GetSignalChannel(ctx, "approval-signal")

    // 设置 24 小时超时
    timerCtx, cancel := workflow.WithCancel(ctx)
    _ = workflow.NewTimer(timerCtx, time.Hour*24)

    selector := workflow.NewSelector(ctx)
    selector.AddReceive(signalChan, func(c workflow.Channel, more bool) {c.Receive(ctx, &approvalResult)
        cancel() // 取消定时器})
    selector.AddFuture(timerCtx, func(f workflow.Future) {
        // 超时处理
        approvalResult = false
    })

    selector.Select(ctx)

    if !approvalResult {return errors.New("审批未通过")
    }

    // 继续执行...
    return nil
}

Cadence 为 AI Agent 开发提供了强大的工作流引擎，通过本文的实战指南，你应该已经掌握了：