workflow-best-practices

工作流最佳实践

使用步骤、流式传输和 Agent 执行，构建持久可靠的工作流。

前置条件

请先完成以下设置教程：

• 工作流开发工具包设置

文件夹结构

每个工作流都会在 src/workflows/ 下拥有一个子文件夹。index.ts 负责编排（"use workflow"）；steps/ 用于存放持久检查点（"use step"）；共享步骤则放在顶层的 steps/ 中。

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

创建工作流

编排函数带有 "use workflow" 指令，并调用各个步骤。运行 Agent 前务必调用 startStream(messageId)，运行后调用 finishStream()；WorkflowChatTransport 需要开始和结束帧才能解析响应。

// 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();

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

  // runId lets clients resume this stream later
  const messageId = await createAssistantMessage({
    chatId,
    runId: workflowRunId,
  });
  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();

  await removeRunId(messageId);
}

startStream 会写入 { type: "start", messageId }；finishStream 会写入 { type: "finish", finishReason: "stop" }，并关闭可写流。

步骤

步骤是可持久化其结果的持久检查点。

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

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

工作流运行时无法导入 Node 模块，因此请将日志记录器调用封装在步骤中。

// 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);
  }
}

启动和恢复

使用 workflow/api 中的 start 启动；使用 getRun 重新连接到正在运行或已完成的运行。

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

const run = await start(chatWorkflow, [{ chatId, userMessage }]);
// run.runId       - unique id for this run
// run.readable    - stream of UI message chunks

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

持久化结果

在步骤中保存 Agent 输出。插入前，assertChatAgentParts 会将通用的 UIMessage["parts"] 收窄为应用的工具／数据类型。

// src/workflows/chat/steps/history.ts
import type { UIMessage } from "ai";
import { insertMessageParts } from "@/lib/chat/queries";
import { assertChatAgentParts } 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);

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

---

参考资料

• 工作流开发工具包
• 工作流 API 参考