go-workflow

Go -workflow是一个持久的Go语言工作流库，灵感来自于 工作流SDK． 工作流是Go函数，其 进步是一步一步坚持下来的。运行可以挂起定时器或钩子 不保留进程，并在可以继续时恢复。跑步可以是 使用Workflow SDK web UI进行检查。

go get github.com/eriicafes/go-workflow

阅读以下文档 go-workflow.vercel.app。

代理技能可用于 编写工作流程 和 实现世界后台。

工作流

工作流是一种嵌入的结构类型 workflow.Struct[Out] 宣布其 输出 Out． 它的包路径和类型名称是工作流名称，它的值是 运行的输入。保持工作流在一个 workflows 包裹和他们的跑步者 flows 里面的包装， workflows/flows，以便启动请求处理程序 没有导入运行程序的工作流。

package workflows

type GenerateReport struct {
    workflow.Struct[GenerateReportResult]
    Topic string
}

type GenerateReportResult struct {
    Summary string
}

跑步者和台阶

跑步者执行工作流。它是一个简单的函数，将工作流作为 输入和调用步骤，它们完成实际的工作和副作用。保持每个跑步者 在包含调用步骤的文件中 workflows/flows。

package flows

func GenerateReport(ctx context.Context, input workflows.GenerateReport) (workflows.GenerateReportResult, error) {
    sources, err := workflow.Step(ctx, findSources, input.Topic)
    if err != nil {
        return workflows.GenerateReportResult{}, err
    }
    return workflows.GenerateReportResult{Summary: strings.Join(sources, ", ")}, nil
}

func findSources(ctx context.Context, topic string) ([]string, error) {
    return search(ctx, topic)
}

登记跑步者 workflow.Use 然后用 worlds.Runners。

runners := worlds.Runners(
    workflow.Use(flows.GenerateReport),
)

每个步骤的结果一旦完成就会被持久化，因此运行将从其 最后完成的步骤，从不重复已完成的工作。等在台阶上，一 定时器，或者钩子挂起运行而不消耗任何资源，直到它可以 继续下去。为了达到这个效果，跑步者只会通过步数和 只能读时间 workflow.Now，和 分析仪 检查 这些规则。

如果部署的更改使运行程序重播与其运行的内容不同 记录，运行时检测到偏差并使用a失败运行 worlds.ErrCorruptedEventLog 而不是将结果匹配到错误的调用。 还可以将运行固定到使用DeploymentID启动它们的部署上， 因此，在工作流程中，飞行运行会不断回放记录它们的代码 ,嵌入 workflow.StartOnLatest 在最新部署上开始新的运行。

步骤至少运行一次，并在出现错误时重试，默认情况下运行三次。集 workflow.StepOptions{MaxRetries: n} 要改变这一点，返回 workflow.FatalError(err) 没有重试或返回就失败了 workflow.RetryableError(err, after) 延迟后重试。这些包装器 仅适用于由步骤、内联步骤和回滚返回的错误：运行程序 返回一个错误，它的运行失败而不重试，无论是否包装。步骤 函数应该是幂等的，并且 workflow.StepID(ctx) 是一个马厩 幂等性的关键。

workflow.InlineStep(ctx, fn, input) 接受相同的参数 Step 但 在流本身中运行一个简短的步骤，例如生成ID，而不需要 分步交付。失败的尝试会像台阶一样重新尝试。

运行工作流

每次运行都记录在一个World中。本地World存储作为JSON文件和 交付进程中的工作，因此不需要数据库或worker。

world, err := localworld.Open(localworld.Config{Dir: ".workflows"}, runners)
if err != nil {
    log.Fatal(err)
}
defer world.Close()
if err := world.Prepare(ctx); err != nil {
    log.Fatal(err)
}

ctx = workflow.NewContext(ctx, workflow.NewClient(world))
run, err := workflow.Start(ctx, workflows.GenerateReport{Topic: "workflows"})
if err != nil {
    log.Fatal(err)
}
result, err := run.Value(ctx)

workflow.Start 只需要工作流类型和ctx中的客户端。一个 Run 句柄读取并控制运行 Value， Status， Cancel, WakeUp, workflow.Get 返回已存在的run by ID的句柄。

世界

每个World包也输出 HTTP(world)，它为RPC端点提供服务 客户端和web UI使用的webhook端点 /v1/workflow。 HTTPPrefix(world, prefix) 在另一个前缀下提供它们，和 RPC(world) 和 Webhook(world) 为每个端点单独提供服务。集 Config.BaseURL 来 该前缀的公共URL，因为webhook URL是从它构建的。

Postgres

任意数量的应用程序和工作进程共享一个数据库。 Worker 运行执行交付的循环。

world, err := postgresworld.Open(ctx, postgresworld.Config{BaseURL: os.Getenv("WORKFLOW_BASE_URL"), URL: os.Getenv("DATABASE_URL")})
if err != nil {
    log.Fatal(err)
}
defer world.Close()

go func() {
    if err := postgresworld.Worker(ctx, world, runners); err != nil {
        log.Printf("workflow worker stopped: %v", err)
    }
}()
log.Fatal(http.ListenAndServe(":8081", postgresworld.HTTP(world)))

Vercel

部署 Queue 作为具有Vercel Queues触发器的函数 Config.Topic， 和 HTTP 作为客户端的独立功能。Redis是通过TCP与 RedisURL 或通过Upstash REST API RedisRESTURL 和 RedisRESTToken。 OIDCToken 默认为读取 VERCEL_OIDC_TOKEN 在每一个 请求。

world, err := vercelworld.New(vercelworld.Config{
    BaseURL:  os.Getenv("WORKFLOW_BASE_URL"),
    RedisURL: os.Getenv("REDIS_URL"),
    QueueURL: os.Getenv("VERCEL_QUEUE_URL"),
    Topic:    os.Getenv("VERCEL_WORKFLOW_QUEUE_TOPIC"),
})
if err != nil {
    log.Fatal(err)
}

// The queue consumer function.
log.Fatal(http.ListenAndServe(":8081", vercelworld.Queue(world, runners)))

// The client-facing function.
log.Fatal(http.ListenAndServe(":8081", vercelworld.HTTP(world)))

Vercel Queues最多将交货延迟7天。集 Config.QStash 来 迟交工作。

AWS

安装 Queue 作为一个Lambda函数的SQS处理程序和 HTTP 后面 命令功能另一个URL。这个 起程拓殖模块 创建 表、队列、调度器组和IAM策略 awsworld.CreateTable 为测试和DynamoDB Local创建表。

world, err := awsworld.New(awsworld.Config{BaseURL: os.Getenv("WORKFLOW_BASE_URL"), TableName: os.Getenv("WORKFLOW_TABLE"), QueueURL: os.Getenv("WORKFLOW_QUEUE_URL")}, dynamodb.NewFromConfig(cfg), sqs.NewFromConfig(cfg))
if err != nil {
    log.Fatal(err)
}

// The queue consumer function.
lambda.Start(awsworld.Queue(world, runners))

// The client-facing function.
lambdaurl.Start(awsworld.HTTP(world))

SQS最多延迟交货15分钟。集 Config.Scheduler 交付 稍后通过EventBridge Scheduler完成的工作。

保留

默认情况下，运行是永久保存的。支持保留的World删除A 运行的数据 Config.Retention 在它完成、失败或被取消之后 存储是有限的。 Config.Retentions 将它设置为每个工作流，其中为零 保持工作流的运行永远：

config.Retention = 24 * time.Hour
config.Retentions = worlds.Retentions(
    worlds.Retain[workflows.Chat](10 * time.Minute),
    worlds.Retain[workflows.Invoice](0),
)

仍然挂起或正在运行的运行永远不会被删除，钩子的令牌保留 保持运行直到结束。几分钟是最短的 保留，因为调用者读取运行结果并紧跟其后 完成。每个World的自述文件描述了如何以及何时删除。

申请和工作人员

只启动和读取运行的应用程序需要工作流类型和一个 客户，不是选手。 NewHTTPClient 与世界对话 HTTP,在 服务器的URL后跟处理程序的前缀，以及 Middleware 把 客户端在每个请求上下文中。

client := workflow.NewHTTPClient("http://workflow-worker:8081/v1/workflow")

mux := http.NewServeMux()
mux.HandleFunc("POST /reports", func(w http.ResponseWriter, r *http.Request) {
    run, err := workflow.Start(r.Context(), workflows.GenerateReport{Topic: r.FormValue("topic")})
    if err != nil {
        http.Error(w, err.Error(), http.StatusInternalServerError)
        return
    }
    w.WriteHeader(http.StatusAccepted)
    w.Write([]byte(run.ID()))
})
log.Fatal(http.ListenAndServe(":8080", workflow.Middleware(mux, client)))

examples/basic 有完整的与工人的申请吗 每一个世界。

并发性和时间

workflow.Background 不等待地开始一个步骤并返回 *Result[T]。 Map 每个输入开始一个步骤。 All 按输入顺序收集输出，如果任何步骤失败，则取消其余步骤。 Join， JoinOk，和 Select 函数同时等待多个承诺。

results, err := workflow.All(ctx, findSources, []string{"go", "rust"})

MapResults 将映射结果片转换为 Join， Select，或组 Add。 WaitGroup 等待每一个结果； ErrGroup 等待每一个结果 返回第一个失败。使用 workflow.Go(ctx, &group, step, input) 开始 并跟踪步骤。 workflow.ErrGroupWithContext(ctx) 返回一个 ErrGroup和持久上下文在失败或Wait返回时被取消。

workflow.Sleep 暂停运行一段时间， Timer 返回一个承诺 在1之后安定下来，并且 WithTimeout 和 WithDeadline 返回取消上下文和取消函数，使用计时器取消使用该上下文的工作。睡眠跑不消耗资源，所以它可以睡上几天或几个月。

错误

失败的步骤的错误将与该步骤一起保存，并作为 *workflow.Error运行失败的错误也是如此 Run.Value． 它使 原始链中每个错误的消息、类型和导出字段，因此 errors.Is 和 errors.As 在跑步和儿童跑步中进行练习。

var ErrCardDeclined = errors.New("card declined")

type LimitError struct{ Limit int }

func (e *LimitError) Error() string { return fmt.Sprintf("over limit %d", e.Limit) }

检查流道中记录的错误：

_, err := workflow.Step(ctx, charge, order)
var limit *LimitError
switch {
case errors.Is(err, ErrCardDeclined):
    // ...
case errors.As(err, &limit):
    // limit.Limit is decoded from the recorded error.
}

回滚

在工作被逆转后推迟补偿。如果以后运行失败，则 补偿按照Go的延迟顺序作为持久步骤运行，最后一个延迟在前面。

invoice, err := workflow.Step(ctx, createInvoice, input)
if err != nil {
    return Report{}, err
}
defer workflow.Rollback(ctx, voidInvoice, invoice.ID)

流

步骤写入读取器在运行停止时使用的运行所有的字节流 走了。写入器和读取器使用运行的默认流，除非它们的选项 另一个名字。流一次只有一个写入器，另一个写入器等待 直到当前程序释放或关闭它。

func writeProgress(ctx context.Context, count int) (struct{}, error) {
    writer, release, err := workflow.Writer(ctx)
    if err != nil {
        return struct{}{}, err
    }
    defer release()
    _, err = fmt.Fprintf(writer, "found %d sources\n", count)
    return struct{}{}, err
}

在运行之外，比如在路由处理程序中， run.Reader 读取流。每一个 写是一个块，和 ReaderOption.StartIndex 选择第一个块，其中a 负索引从末尾开始计数。读取等待新的数据，直到流 关闭。

// The last three writes onward.
reader, err := run.Reader(ctx, workflow.ReaderOption{StartIndex: -3})

钩和网钩

钩子接收重复类型的有效载荷。在共享包的工作流结构旁边声明可重用的钩子定义。 传递令牌时，调用者必须提前知道它。

hook := workflow.CreateHook[Approval](ctx, workflow.HookOption{Token: "approve:" + input.OrderID})
defer hook.Dispose(ctx)

approval, err := hook.Next(ctx) // registers the hook and waits for a payload

创建是本地的，只返回一个句柄。从钩子接收或选择该钩子注册它；不相关的等待则不然。使用 GetConflict(ctx) 在开始需要所有权或公开代币的工作之前；它在不等待有效负载的情况下注册，并返回冲突的运行或nil。 Dispose 通过恐慌传播失败，包括延迟失败。

err := workflow.ResumeHook(ctx, "approve:"+orderID, Approval{Approved: true})

webhook是一个由HTTP请求恢复到其URL的钩子，带有JSON主体 解码为有效载荷。 WebhookOption 可以要求一个方法，设置一个静态吗 响应，或者保持请求直到一个步骤调用 WebhookRequest.RespondWith． 令牌 URL中的值是其唯一的授权，因此在此之前请验证提供者签名 信任有效负载。

webhook, err := workflow.CreateWebhook[Payment](ctx)
if err != nil {
    return Result{}, err
}
defer webhook.Dispose(ctx)

if _, err := workflow.Step(ctx, registerCallback, webhook.URL()); err != nil {
    return Result{}, err
}
request, err := webhook.Request(ctx)
if err != nil {
    return Result{}, err
}
payment, err := request.JSON()

WithCancel 和 WithCancelCause 返回持久取消上下文、取消函数和创建错误。它们的cancel函数也会返回错误。将取消上下文传递给步骤；Cancel显式运行 Run.Cancel;使用 CancelAny(ctx, sources...) 在保留基值的同时结合消去。取消的步骤没有重试就失败了；取消跑步让跑步步骤自然结束。 Then 返回的结果 func(T, error) *Result[U];使用 NewResult(value, err) 对于即时值和失败。

组成工作流

运行者可以通过两种方式组成另一个工作流，就像在workflow SDK中一样。 调用其他工作流的运行器直接在当前运行中运行其步骤。

report, err := GenerateReport(ctx, workflows.GenerateReport{Topic: input.Topic})

workflow.Start 在工作流上下文中启动一个子运行，它携带 $parentRunId 和 $rootRunId 属性。在工作流上下文中 Run.Value 返回 workflow.ErrRunNotFinished 与其等待，不如等待一个孩子， 向它传递一个钩子令牌，并让它在钩子完成后恢复钩子。

func Parent(ctx context.Context, input workflows.Parent) (workflows.ParentResult, error) {
    hook := workflow.CreateHook[string](ctx)
    defer hook.Dispose(ctx)

    if owner, err := hook.GetConflict(ctx); err != nil {
        return workflows.ParentResult{}, err
    } else if owner != nil {
        return workflows.ParentResult{}, workflow.ErrHookConflict
    }
    if _, err := workflow.Start(ctx, workflows.Child{Token: hook.Token()}); err != nil {
        return workflows.ParentResult{}, err
    }
    value, err := hook.Next(ctx)
    return workflows.ParentResult{Value: value}, err
}

func Child(ctx context.Context, input workflows.Child) (workflows.ChildResult, error) {
    value, err := workflow.Step(ctx, work, input)
    if err != nil {
        return workflows.ChildResult{}, err
    }
    _, err = workflow.Step(ctx, resumeParent, completion{Token: input.Token, Value: value})
    return workflows.ChildResult{Value: value}, err
}

func resumeParent(ctx context.Context, c completion) (struct{}, error) {
    return struct{}{}, workflow.ResumeHook(ctx, c.Token, c.Value)
}

人工智能代理

这个 workflowai 包运行AI SDK样式代理为 持久运行：记录每个模型调用和工具步骤，响应流 通过运行流，工具审批等待用户跨轮。它匹配 @ai-sdk/workflow的 WorkflowAgent ,是 建立在 Go AI SDK。 examples/durable-agent 将它提供给React useChat 客户端，以及 AI代理指南 覆盖它 深度。

发展

分析程序检查每个跑者传递到的内容 Use 一个命名函数是谁的 输入是一个工作流，并报告程序， select，映射迭代； time.Now， I/O外部步骤，以及捕获变量的步骤，而不是 把它们作为输入。将其作为兽医工具运行。

go install github.com/eriicafes/go-workflow/cmd/workflowcheck@latest
go vet -vettool=$(which workflowcheck) ./...

这个 workflowdev 包集成了 go-dev。 workflowdev.Check 运行 分析器，每次启动或重新加载之前 workflowdev.Web 运行工作流 SDK web UI对一个世界的服务 HTTP． web UI需要Workflow SDK CLI和此存储库的World适配器，安装在目录中 Go-dev从。

npm install --save-dev workflow go-workflow

session := dev.New()
task := session.NewTask(dev.Cmd{
    Run:        dev.Package("./cmd/app"),
    Watch:      dev.Values("cmd", "workflows"),
    ServerAddr: ":8080",
    Commands: dev.Commands(
        dev.Cmd{Run: dev.Package("./cmd/worker/pg"), Prepare: dev.Hooks(workflowdev.Check("./...")), Watch: dev.Values("cmd", "workflows"), Phase: dev.Before},
        workflowdev.Web("http://localhost:8081/v1/workflow"),
    ),
})
session.Catch(task.Run())
session.Catch(session.Wait())