workflow

关键提示：务必使用正确的 workflow 文档

你关于 workflow 的知识库可能已过期。

下面列出的 workflow 文档与当前安装的 Workflow SDK 版本完全匹配。
在开始执行任何与 workflow 相关的任务前，请先遵循以下指引：

检索项目自带的文档（位于 node_modules/workflow/docs/）：

1. 查找文档文件：glob "node_modules/workflow/docs/**/*.mdx"
2. 搜索文档内容：grep "查询关键词" 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 等）
• api-reference/workflow-runtime/ - 运行时 API（get-world.mdx）以及 world/ World SDK（storage.mdx, streams.mdx, queue.mdx）
• api-reference/workflow-observability/ - 数据反序列化/注水（Hydration）与名称解析工具（hydrate-resource-io.mdx, parse-workflow-name.mdx 等）
• ai/ - AI SDK 集成文档
• errors/ - 错误码文档

相关依赖包同样内置了文档：

• @workflow/ai: node_modules/@workflow/ai/docs/ - DurableAgent 与 AI 集成
• @workflow/core: node_modules/@workflow/core/docs/ - 核心运行时（基础概念、工作原理）
• @workflow/next: node_modules/@workflow/next/docs/ - Next.js 集成

如有疑问，请直接更新至最新版本的 Workflow SDK。

官方资源

• 官网：https://workflow-sdk.dev
• GitHub：https://github.com/vercel/workflow

快速参考

指令声明（Directives）：

"use workflow";  // 首行声明 - 将异步函数转为持久化工作流
"use step";      // 首行声明 - 将函数转为具备缓存与自动重试特性的独立单元

常用 Core API 导入：

// Workflow 原生方法
import { sleep, fetch, createHook, createWebhook, getWritable } from "workflow";
import { FatalError, RetryableError } from "workflow";
import { getWorkflowMetadata, getStepMetadata } from "workflow";

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

// 可观测性与数据反序列化（Hydration）
import { hydrateResourceIO, observabilityRevivers, parseStepName, parseWorkflowName } from "workflow/observability";

// 框架集成
import { withWorkflow } from "workflow/next";
import { workflow } from "workflow/vite";
import { workflow } from "workflow/astro";
// 或在 Nitro/Nuxt 中使用 modules: ["workflow/nitro"]

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

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

带有 "use workflow" 的函数会在沙箱 VM 中运行，而带有 "use step" 的函数拥有 完整的 Node.js 运行时访问权限。因此，请将具体业务逻辑拆分放到 Step 中，主 workflow 函数仅用于控制流编排。

// Step 函数拥有完整的 Node.js 与 npm 模块访问权限
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";
  // 在 Step 中可以直接使用 AI SDK，无需任何 Hack 或绕过方案
  return await generateText({
    model: openai("gpt-4"),
    prompt: `Process: ${JSON.stringify(data)}`,
  });
}

// Workflow 函数仅负责编排 Step —— 完全不会遇到沙箱限制
export async function dataProcessingWorkflow(userId: string) {
  "use workflow";
  const data = await fetchUserData(userId);
  const processed = await processWithAI(data);
  return { success: true, processed };
}

核心优势： Step 具备失败自动重试机制、执行结果自动持久化（支持状态重放/Replay），且不受沙箱限制。

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;  // AI SDK 和 HTTP 请求库运行所必需
  // 赋值后，generateText() 及其他第三方库即可正常工作
}

注意： 来自 @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 函数拥有完整的 Node.js 访问权限
  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 实例的默认输出流中
• 工具的 execute 函数若需要使用 Node.js / npm 依赖，必须加上 "use step"
• 工具的 execute 函数若使用了 workflow 的原生原语（如 sleep()、createHook()），则绝对不能加 "use step" —— 因为它们必须运行在 workflow 层级
• maxSteps 用于限制 LLM 调用的最大轮数（默认无限制）
• 多轮对话：在后续调用 agent.stream() 时，将 result.messages 与新的用户消息拼合传入即可

更多关于 DurableAgent 的详细信息，请查阅 node_modules/@workflow/ai/docs/ 中的 AI 专属文档。

启动 Workflow 与子 Workflow（Child Workflows）

可以在 API Route 中直接使用 start() 启动 Workflow。但注意：start() 不能在 workflow 上下文中直接调用 —— 如果想在 Workflow 内部启动子 Workflow，必须将其包裹在 step 函数中。

import { start } from "workflow/api";

// 在 API Route 中调用 —— 可以直接使用
export async function POST() {
  const run = await start(myWorkflow, [arg1, arg2]);
  return Response.json({ runId: run.runId });
}

// 无参数 Workflow 的启动方式
const run = await start(noArgWorkflow);

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

import { start } from "workflow/api";

// 将 start() 包裹在 step 函数中
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");  // 通过 Step 触发异步任务（Fire-and-forget）
  await sleep("1h");
}

start() 调用后会立即返回，不会阻塞等待工作流执行结束。如果需要等待其执行完成，请使用 run.returnValue。

Hooks —— 基于外部事件的暂停与恢复

Hooks 允许 Workflow 暂停并等待外部数据。在 workflow 内部调用 createHook() 创建挂起点，在 API Route 中使用 resumeHook() 触发恢复。确定性 token（Deterministic tokens）仅适用于 createHook() + resumeHook()（服务端联动）。createWebhook() 总是会生成随机 token —— 请切勿向 createWebhook() 传入 token 选项。

单次事件监听

import { createHook } from "workflow";

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

  const hook = createHook<{ approved: boolean }>({
    token: "approval-123",  // 供外部系统调用的确定性 token
  });

  const result = await hook;  // Workflow 将在此处挂起等待
  return result.approved;
}

多次事件监听（可迭代 Hook）

Hook 实现了 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 Route 中恢复运行

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(`客户端错误: ${res.status}`);
}
if (res.status === 429) {
  throw new RetryableError("已触发请求限流", { retryAfter: "5m" });
}

序列化规则

在 workflow 与 step 之间传递的所有输入输出数据，都必须是可序列化的。

支持的原生类型： string、number、boolean、null、undefined、bigint、纯对象（plain objects）、数组、Date、RegExp、URL、URLSearchParams、Map、Set、Headers、ArrayBuffer、TypedArray、Request、Response、ReadableStream、WritableStream。

不支持的类型： Function（函数）、Symbol、WeakMap / WeakSet。请确保只传递纯数据而非回调函数。

自定义 Class 的序列化

只要实现了 @workflow/serde 协议，Class 实例即可在 workflow/step 边界之间进行跨边界序列化传递。当 Class 内部包含挂载了 "use step" 的实例方法，或者需要在不同 step 之间传递对象实例时，该特性尤为关键。

安装依赖： 包含该 Class 的 package 必须将 @workflow/serde 列为其依赖项。

标准写法： 在类（Class）主体内部，使用计算属性语法定义两个静态方法：

import { WORKFLOW_SERIALIZE, WORKFLOW_DESERIALIZE } from "@workflow/serde";

export class Point {
  x: number;
  y: number;

  constructor(x: number, y: number) {
    this.x = x;
    this.y = y;
  }

  // 序列化：返回纯数据结构（必须仅包含 devalue 兼容的类型）
  static [WORKFLOW_SERIALIZE](instance: Point) {
    return { x: instance.x, y: instance.y };
  }

  // 反序列化：根据纯数据还原 Class 实例
  static [WORKFLOW_DESERIALIZE](data: { x: number; y: number }) {
    return new Point(data.x, data.y);
  }

  async computeDistance(other: Point) {
    "use step";
    return Math.sqrt((this.x - other.x) ** 2 + (this.y - other.y) ** 2);
  }
}

关键规则：

1. 必须在类（Class）的内部声明 serde 方法，且必须作为使用计算属性语法的静态方法（如 static [WORKFLOW_SERIALIZE](...)）。SWC 编译器插件会在扫描 Class 时进行提取。千万不要在外部进行动态赋值（比如 (MyClass as any)[WORKFLOW_SERIALIZE] = ...），否则编译器将完全无法检测到该方法。
2. Serde 方法的返回值必须仅包含 devalue 兼容的类型（如纯对象、数组、基础数据类型、Date、Map、Set、Uint8Array 等）。不能包含函数、Class 实例或 Node.js 专属对象。
3. 依赖 Node.js 的实例方法必须加上 "use step"。 SWC 插件会将带有 "use step" 的方法体从打包给 workflow 的 Bundle 中剥离出去。这样就能保证 Node.js 相关的导入（如 fs、crypto、child_process 等）不会混入 workflow 沙箱中。而类本身的骨架及 serde 方法则会保留在 workflow 侧。