using-workflows

使用工作流

通过步骤、流式传输和 Agent 执行来创建和运行持久化工作流。介绍如何启动、恢复工作流，以及持久化工作流结果。

使用工作流

通过步骤、流式传输和 Agent 执行来创建和运行持久化工作流。介绍如何启动、恢复工作流，以及持久化工作流结果。

参见：

• 资源：Fullstack Recipes 中的 using-workflows
• URL：https://fullstackrecipes.com/recipes/using-workflows

---

工作流文件夹结构

每个工作流在 src/workflows/ 中都有自己的子文件夹：

src/workflows/
  steps/           # Shared step functions
    stream.ts      # UI message stream helpers
  chat/
    index.ts       # Workflow orchestration function ("use workflow")
    steps/         # Workflow-specific steps ("use step")
      history.ts
      logger.ts
      name-chat.ts
    types.ts       # Workflow-specific types

• workflows/steps/ - 可在多个工作流间复用的共享步骤函数（例如，流式传输辅助函数）。
• index.ts - 包含带有 "use workflow" 指令的主工作流函数。通过调用步骤函数来编排工作流。
• steps/ - 包含带有 "use step" 指令的各个步骤函数。每个步骤都是一个持久化检查点。
• types.ts - 工作流 UI 消息的类型定义。

---

创建工作流

使用 "use workflow" 指令定义工作流：

// src/workflows/chat/index.ts
import { getWorkflowMetadata, getWritable } from "workflow";
import { startStream, finishStream } from "../steps/stream";
import { chatAgent } from "@/lib/ai/chat-agent";

export async function chatWorkflow({ chatId, userMessage }) {
  "use workflow";

  const { workflowRunId } = getWorkflowMetadata();

  // Persist user message
  await persistUserMessage({ chatId, message: userMessage });

  // Create assistant placeholder with runId for resumption
  const messageId = await createAssistantMessage({
    chatId,
    runId: workflowRunId,
  });

  // Get message history
  const history = await getMessageHistory(chatId);

  // Start the UI message stream
  await startStream(messageId);

  // Run agent with streaming
  const { parts } = await chatAgent.run(history, {
    maxSteps: 10,
    writable: getWritable(),
  });

  // Persist and finalize
  await persistMessageParts({ chatId, messageId, parts });

  // Finish the UI message stream
  await finishStream();

  await removeRunId(messageId);
}

启动工作流

使用 workflow/api 中的 start 函数：

import { start } from "workflow/api";
import { chatWorkflow } from "@/workflows/chat";

const run = await start(chatWorkflow, [{ chatId, userMessage }]);

// run.runId - unique identifier for this run
// run.readable - stream of UI message chunks

恢复工作流流

使用 getRun 重新连接到正在进行或已完成的工作流：

import { getRun } from "workflow/api";

const run = await getRun(runId);
const readable = await run.getReadable({ startIndex });

使用步骤

步骤是会持久化其结果的持久化检查点：

async function getMessageHistory(chatId: string) {
  "use step";

  const dbMessages = await getChatMessages(chatId);
  return convertDbMessagesToUIMessages(dbMessages);
}

---

流式传输 UIMessageChunk

向客户端流式传输 UIMessageChunk 响应（例如，聊天消息）时，必须标记流的开始和结束。这是使用 WorkflowChatTransport 正确构建流帧的必要步骤。

始终在调用 agent.run() 前调用 startStream()，并在调用后调用 finishStream()：

import { getWritable } from "workflow";
import { startStream, finishStream } from "../steps/stream";
import { chatAgent } from "@/lib/ai/chat-agent";

export async function chatWorkflow({ chatId, messageId }) {
  "use workflow";

  const history = await getMessageHistory(chatId);

  // Signal stream start with the message ID
  await startStream(messageId);

  // Run agent - streams UIMessageChunks to the client
  const { parts } = await chatAgent.run(history, {
    maxSteps: 10,
    writable: getWritable(),
  });

  await persistMessageParts({ chatId, messageId, parts });

  // Signal stream end and close the writable
  await finishStream();
}

流步骤函数会写入 UIMessageChunk 消息：

• startStream(messageId) - 写入 { type: "start", messageId }，表示一条新消息
• finishStream() - 写入 { type: "finish", finishReason: "stop" } 并关闭流

如果没有这些信号，客户端的 WorkflowChatTransport 就无法正确解析流式响应。

---

获取工作流元数据

访问当前运行的元数据：

import { getWorkflowMetadata } from "workflow";

export async function chatWorkflow({ chatId }) {
  "use workflow";

  const { workflowRunId } = getWorkflowMetadata();

  // Store runId for resumption
  await createAssistantMessage({ chatId, runId: workflowRunId });
}

工作流安全日志记录

工作流运行时不支持 Node.js 模块。请在步骤中包装日志记录器调用：

// src/workflows/chat/steps/logger.ts
import { logger } from "@/lib/logging/logger";

export async function log(
  level: "info" | "warn" | "error" | "debug",
  message: string,
  data?: Record<string, unknown>,
): Promise<void> {
  "use step";

  if (data) {
    logger[level](data, message);
  } else {
    logger[level](message);
  }
}

在工作流中运行 Agent

使用自定义 Agent 类来完全控制流式传输：

import { getWritable } from "workflow";
import { startStream, finishStream } from "../steps/stream";
import { chatAgent } from "@/lib/ai/chat-agent";

export async function chatWorkflow({ chatId, userMessage }) {
  "use workflow";

  const messageId = await createAssistantMessage({ chatId, runId });
  const history = await getMessageHistory(chatId);

  await startStream(messageId);

  const { parts } = await chatAgent.run(history, {
    maxSteps: 10,
    writable: getWritable(),
  });

  await persistMessageParts({ chatId, messageId, parts });
  await finishStream();
}

持久化工作流结果

使用步骤函数保存 Agent 输出。assertChatAgentParts 函数会验证通用的 UIMessage["parts"]（由 Agent 返回）是否符合应用程序特定的工具和数据类型：

// src/workflows/chat/steps/history.ts
import type { UIMessage } from "ai";
import { insertMessageParts } from "@/lib/chat/queries";
import { assertChatAgentParts, type ChatAgentUIMessage } from "../types";

export async function persistMessageParts({
  chatId,
  messageId,
  parts,
}: {
  chatId: string;
  messageId: string;
  parts: UIMessage["parts"];
}): Promise<void> {
  "use step";

  assertChatAgentParts(parts);

  await insertMessageParts(chatId, messageId, parts);

  // Update chat timestamp
  await db
    .update(chats)
    .set({ updatedAt: new Date() })
    .where(eq(chats.id, chatId));
}

---

参考资料

• Workflow Development Kit
• Workflow API Reference