workflow

重要：始终使用正确的 workflow 文档

你对 workflow 的了解已经过时。

下面列出的 workflow 文档与已安装的 Workflow DevKit 版本一致。 开始任何与 workflow 相关的任务前，请遵循以下说明：

搜索 node_modules/workflow/docs/ 中捆绑的文档：

1. 查找文档：glob "node_modules/workflow/docs/**/*.mdx"
2. 搜索内容：grep "your query" node_modules/workflow/docs/

node_modules/workflow/docs/ 中的文档结构：

• getting-started/ - 框架设置（next.mdx、express.mdx、hono.mdx 等）
• foundations/ - 核心概念（workflows-and-steps.mdx、hooks.mdx、streaming.mdx 等）
• api-reference/workflow/ - API 文档（sleep.mdx、create-hook.mdx、fatal-error.mdx 等）
• api-reference/workflow-api/ - 客户端 API（start.mdx、get-run.mdx、resume-hook.mdx 等）
• ai/ - AI SDK 集成文档
• errors/ - 错误代码文档

相关软件包也包含捆绑文档：

• @workflow/ai：node_modules/@workflow/ai/docs/ - DurableAgent 和 AI 集成
• @workflow/core：node_modules/@workflow/core/docs/ - 核心运行时（foundations、how-it-works）
• @workflow/next：node_modules/@workflow/next/docs/ - Next.js 集成

如有疑问，请更新到最新版本的 Workflow DevKit。

官方资源

• 网站：https://useworkflow.dev
• GitHub：https://github.com/vercel/workflow

快速参考

指令：

"use workflow"; // First line - makes async function durable
"use step"; // First line - makes function a cached, retryable unit

必要导入：

// Workflow primitives
import { sleep, fetch, createHook, createWebhook, getWritable } from "workflow";
import { FatalError, RetryableError } from "workflow";
import { getWorkflowMetadata, getStepMetadata } from "workflow";

// API operations
import { start, getRun, resumeHook, resumeWebhook } from "workflow/api";

// Framework integrations
import { withWorkflow } from "workflow/next";
import { workflow } from "workflow/vite";
import { workflow } from "workflow/astro";
// Or use modules: ["workflow/nitro"] for Nitro/Nuxt

// AI agent
import { DurableAgent } from "@workflow/ai/agent";

优先使用 Step 函数，避免沙箱错误

"use workflow" 函数在沙箱化的 VM 中运行。"use step" 函数拥有完整的 Node.js 访问权限。将逻辑放入 Step 中，并让 Workflow 函数专门负责编排。

// Steps have full Node.js and npm access
async function fetchUserData(userId: string) {
  "use step";
  const response = await fetch(`https://api.example.com/users/${userId}`);
  return response.json();
}

async function processWithAI(data: any) {
  "use step";
  // AI SDK works in steps without workarounds
  return await generateText({
    model: openai("gpt-4"),
    prompt: `Process: ${JSON.stringify(data)}`,
  });
}

// Workflow orchestrates steps - no sandbox issues
export async function dataProcessingWorkflow(userId: string) {
  "use workflow";
  const data = await fetchUserData(userId);
  const processed = await processWithAI(data);
  return { success: true, processed };
}

**优点：**Step 会自动重试，结果会持久化以便重放，而且不受沙箱限制。

Workflow 沙箱限制

需要在 Workflow 函数中直接编写逻辑（而不是放入 Step）时，需遵守以下限制：

限制	解决方法
不支持 fetch()	import { fetch } from "workflow" 然后 globalThis.fetch = fetch
不支持 setTimeout/setInterval	使用 "workflow" 中的 sleep("5s")
不支持 Node.js 模块（fs、crypto 等）	移至 Step 函数

示例 - 在 Workflow 上下文中使用 fetch：

import { fetch } from "workflow";

export async function myWorkflow() {
  "use workflow";
  globalThis.fetch = fetch; // Required for AI SDK and HTTP libraries
  // Now generateText() and other libraries work
}

注意：@workflow/ai 中的 DurableAgent 会自动处理 fetch 赋值。

DurableAgent — Workflow 中的 AI Agent

使用 DurableAgent 构建能够维护状态并从中断中恢复的 AI Agent。它会自动处理 Workflow 沙箱（无需手动设置 globalThis.fetch）。

import { DurableAgent } from "@workflow/ai/agent";
import { getWritable } from "workflow";
import { z } from "zod";
import type { UIMessageChunk } from "ai";

async function lookupData({ query }: { query: string }) {
  "use step";
  // Step functions have full Node.js access
  return `Results for "${query}"`;
}

export async function myAgentWorkflow(userMessage: string) {
  "use workflow";

  const agent = new DurableAgent({
    model: "anthropic/claude-sonnet-4-5",
    system: "You are a helpful assistant.",
    tools: {
      lookupData: {
        description: "Search for information",
        inputSchema: z.object({ query: z.string() }),
        execute: lookupData,
      },
    },
  });

  const result = await agent.stream({
    messages: [{ role: "user", content: userMessage }],
    writable: getWritable<UIMessageChunk>(),
    maxSteps: 10,
  });

  return result.messages;
}

要点：

• getWritable<UIMessageChunk>() 会将输出流式传输到 Workflow 运行的默认流
• 需要访问 Node.js/npm 的工具 execute 函数应使用 "use step"
• 使用 Workflow 原语（sleep()、createHook()）的工具 execute 函数不应使用 "use step"——它们在 Workflow 层级运行
• maxSteps 限制 LLM 调用次数（默认不设上限）
• 多轮对话：将 result.messages 与新的用户消息一起传入后续的 agent.stream() 调用

更多关于 DurableAgent 的详情，请查看 node_modules/@workflow/ai/docs/ 中的 AI 文档。

启动 Workflow 和子 Workflow

使用 start() 从 API 路由启动 Workflow。不能直接在 Workflow 上下文中调用 start()——请将其封装在 Step 函数中。

import { start } from "workflow/api";

// From an API route — works directly
export async function POST() {
  const run = await start(myWorkflow, [arg1, arg2]);
  return Response.json({ runId: run.runId });
}

// No-args workflow
const run = await start(noArgWorkflow);

在 Workflow 内部启动子 Workflow——必须使用 Step：

import { start } from "workflow/api";

// Wrap start() in a step function
async function triggerChild(data: string) {
  "use step";
  const run = await start(childWorkflow, [data]);
  return run.runId;
}

export async function parentWorkflow() {
  "use workflow";
  const childRunId = await triggerChild("some data"); // Fire-and-forget via step
  await sleep("1h");
}

start() 会立即返回——它不会等待 Workflow 完成。使用 run.returnValue 等待完成。

Hooks — 使用外部事件暂停和恢复

Hooks 让工作流能够等待外部数据。在工作流中使用 createHook()，并从 API 路由调用 resumeHook()。

单个事件

import { createHook } from "workflow";

export async function approvalWorkflow() {
  "use workflow";

  const hook = createHook<{ approved: boolean }>({
    token: "approval-123", // deterministic token for external systems
  });

  const result = await hook; // Workflow suspends here
  return result.approved;
}

多个事件（可迭代 Hooks）

Hooks 实现了 AsyncIterable——使用 for await...of 接收多个事件：

import { createHook } from "workflow";

export async function chatWorkflow(channelId: string) {
  "use workflow";

  const hook = createHook<{ text: string; done?: boolean }>({
    token: `chat-${channelId}`,
  });

  for await (const event of hook) {
    await processMessage(event.text);
    if (event.done) break;
  }
}

每次调用 resumeHook(token, payload)，都会向循环传递下一个值。

从 API 路由恢复

import { resumeHook } from "workflow/api";

export async function POST(req: Request) {
  const { token, data } = await req.json();
  await resumeHook(token, data);
  return new Response("ok");
}

错误处理

对于永久性故障（不重试），使用 FatalError；对于暂时性故障，使用 RetryableError：

import { FatalError, RetryableError } from "workflow";

if (res.status >= 400 && res.status < 500) {
  throw new FatalError(`Client error: ${res.status}`);
}
if (res.status === 429) {
  throw new RetryableError("Rate limited", { retryAfter: "5m" });
}

序列化

传入工作流和步骤以及从中传出的所有数据都必须可序列化。

支持的类型： string、number、boolean、null、undefined、bigint、普通对象、数组、Date、RegExp、URL、URLSearchParams、Map、Set、Headers、ArrayBuffer、类型化数组、Request、Response、ReadableStream、WritableStream。

不支持： 函数、类实例、Symbols、WeakMap/WeakSet。传递数据，而不是回调。

流式传输

使用 getWritable() 从工作流中流式传输数据。getWritable() 可在工作流和步骤上下文中调用，但不能在工作流函数中直接与流交互（调用 getWriter()、write()、close()）。必须将流传递给步骤函数以执行实际 I/O，或者由步骤自行调用 getWritable()。

在工作流中获取流，并将其传递给步骤：

import { getWritable } from "workflow";

export async function myWorkflow() {
  "use workflow";
  const writable = getWritable();
  await writeData(writable, "hello world");
}

async function writeData(writable: WritableStream, chunk: string) {
  "use step";
  const writer = writable.getWriter();
  try {
    await writer.write(chunk);
  } finally {
    writer.releaseLock();
  }
}

直接在步骤中调用 getWritable()（无需传递）：

import { getWritable } from "workflow";

async function streamData(chunk: string) {
  "use step";
  const writer = getWritable().getWriter();
  try {
    await writer.write(chunk);
  } finally {
    writer.releaseLock();
  }
}

调试

# Check workflow endpoints are reachable
npx workflow health
npx workflow health --port 3001  # Non-default port

# Visual dashboard for runs
npx workflow web
npx workflow web <run_id>

# CLI inspection (use --json for machine-readable output, --help for full usage)
npx workflow inspect runs
npx workflow inspect run <run_id>

# For Vercel-deployed projects, specify backend and project
npx workflow inspect runs --backend vercel --project <project-name> --team <team-slug>
npx workflow inspect run <run_id> --backend vercel --project <project-name> --team <team-slug>

# Open Vercel dashboard in browser for a specific run
npx workflow inspect run <run_id> --web
npx workflow web <run_id> --backend vercel --project <project-name> --team <team-slug>

# Cancel a running workflow
npx workflow cancel <run_id>
npx workflow cancel <run_id> --backend vercel --project <project-name> --team <team-slug>
# --env defaults to "production"; use --env preview for preview deployments

调试提示：

• 在任何命令中使用 --json（-j）以获取机器可读的输出
• 使用 --web 在浏览器中打开 Vercel Observability 仪表板
• 在任何命令中使用 --help 查看完整的用法详情
• 只导入实际使用的工作流 API。未使用的导入可能导致 500 错误。