# go-workflow 中文文档

规范地址：https://npmcn.com/package/go-workflow/

版本：0\.7\.0
npm 元数据更新时间：2026-11-03T16:47:06.939Z
中文译文更新时间：2026-11-03T18:55:52.355Z
软件包许可：MIT

本文由上游 README 机器翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/go-workflow
原文对照：https://npmcn.com/package/go-workflow/?lang=original
项目主页：https://go-workflow.vercel.app/
源码仓库：https://github.com/eriicafes/go-workflow

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：262 kB，5 个文件
- 模块格式：ES Module
- TypeScript 类型：包内自带
- 运行依赖：1 个
- 首次发布：2026-10-23T09:08:46.288Z
- 最新版本发布：2026-11-03T16:47:06.576Z
- 已发布版本数：9（近 12 个月 9 个）

## 中文简介

面向Go世界的事件源持久工作流。

## 安装

```sh
npm install go-workflow
```

也可以使用：`pnpm add go-workflow`、`yarn add go-workflow`、`bun add go-workflow`

## README

# go-workflow

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

```sh
go get github.com/eriicafes/go-workflow
```

阅读以下文档 [go-workflow.vercel.app](https://go-workflow.vercel.app/)。

代理技能可用于 [编写工作流程](https://npmcn.com/package/go-workflow/skills/go-workflow/SKILL.md) 和 [实现世界后台](https://npmcn.com/package/go-workflow/skills/go-workflow-worlds/SKILL.md)。

## 工作流

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

```go
package workflows

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

type GenerateReportResult struct {
    Summary string
}
```

## 跑步者和台阶

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

```go
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`。

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

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

如果部署的更改使运行程序重播与其运行的内容不同 记录，运行时检测到偏差并使用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。

```go
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的句柄。

## 世界

| 包 | 储存和交付 | 入口点 |
| --- | --- | --- |
| [`worlds/local`](https://npmcn.com/package/go-workflow/worlds/local) | JSON文件，进程内交付 | `World.Prepare` |
| [`worlds/postgres`](https://npmcn.com/package/go-workflow/worlds/postgres) | PostgreSQL，监听/通知工人 | `postgresworld.Worker` |
| [`worlds/vercel`](https://npmcn.com/package/go-workflow/worlds/vercel) | Redis， Vercel队列，可选的QStash | `vercelworld.Queue` |
| [`worlds/aws`](https://npmcn.com/package/go-workflow/worlds/aws) | DynamoDB， SQS，可选的EventBridge调度器 | `awsworld.Queue` |

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

### Postgres

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

```go
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` 在每一个 请求。

```go
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。这个 [起程拓殖模块](https://npmcn.com/package/go-workflow/worlds/aws/terraform) 创建 表、队列、调度器组和IAM策略 `awsworld.CreateTable` 为测试和DynamoDB Local创建表。

```go
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` 将它设置为每个工作流，其中为零 保持工作流的运行永远：

```go
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` 把 客户端在每个请求上下文中。

```go
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`](https://npmcn.com/package/go-workflow/examples/basic) 有完整的与工人的申请吗 每一个世界。

## 并发性和时间

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

```go
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` 在跑步和儿童跑步中进行练习。

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

type LimitError struct{ Limit int }

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

检查流道中记录的错误：

```go
_, 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的延迟顺序作为持久步骤运行，最后一个延迟在前面。

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

## 流

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

```go
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 负索引从末尾开始计数。读取等待新的数据，直到流 关闭。

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

## 钩和网钩

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

```go
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` 通过恐慌传播失败，包括延迟失败。

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

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

```go
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中一样。 调用其他工作流的运行器直接在当前运行中运行其步骤。

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

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

```go
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`](https://npmcn.com/package/go-workflow/workflowai) 包运行AI SDK样式代理为 持久运行：记录每个模型调用和工具步骤，响应流 通过运行流，工具审批等待用户跨轮。它匹配 `@ai-sdk/workflow`的 `WorkflowAgent` ,是 建立在 [Go AI SDK](https://goaisdk.com/)。 [`examples/durable-agent`](https://npmcn.com/package/go-workflow/examples/durable-agent) 将它提供给React `useChat` 客户端，以及 [AI代理指南](https://go-workflow.vercel.app/guide/ai-agents) 覆盖它 深度。

## 发展

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

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

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

```sh
npm install --save-dev workflow go-workflow
```

```go
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())
```

## 常见问题

### go-workflow 是什么？

面向Go世界的事件源持久工作流。

### 如何安装 go-workflow？

在项目目录运行 npm install go-workflow，或使用 pnpm add / yarn add / bun add。

### go-workflow 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package\.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

### go-workflow 支持 ESM 还是 CommonJS？

以 ES Module 形式发布，推荐使用 import 引入。在 CommonJS 项目中，Node\.js 20\.19\+ / 22\.12\+ 可以直接 require\(\) 不含顶层 await 的 ES Module，更早的版本需要改用动态 import\(\)。

### go-workflow 可以商用吗？

go-workflow 使用 MIT 许可。宽松许可：允许商业使用、修改和再分发，通常只需在分发时保留原版权与许可声明。以上为概括，不构成法律意见，请以许可原文为准。

### go-workflow 还在维护吗？

最新版本 0\.7\.0 发布于 2026/11/04（今天）；至今共发布 9 个版本，其中过去 12 个月发布了 9 个；首次发布于 2026/10/23。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。