workflow-best-practices

ワークフローのベストプラクティス

ステップ、ストリーミング、エージェントの実行を活用して、堅牢なワークフローを構築します。

前提条件

まず、次のセットアップ手順を完了してください。

• Workflow Development Kit のセットアップ

フォルダー構成

各ワークフローには、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" を指定し、ステップを呼び出します。エージェントの実行前には必ず 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 });

結果の永続化

エージェントの出力はステップ内に保存します。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));
}

---

参考資料

• Workflow Development Kit
• Workflow API Reference