using-workflows

ワークフローの操作

ステップ、ストリーミング、エージェント実行を使用して、耐久性のあるワークフローを作成して実行します。ワークフロー結果の開始、再開、永続化について説明します。

ワークフローの操作

ステップ、ストリーミング、エージェント実行を使用して、耐久性のあるワークフローを作成して実行します。ワークフロー結果の開始、再開、永続化について説明します。

参照先：

• リソース：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 クラスを使用します。

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

ワークフロー結果の永続化

ステップ関数を使用してエージェントの出力を保存します。assertChatAgentParts 関数は、汎用の UIMessage["parts"]（エージェントから返される）が、アプリケーション固有のツールおよびデータ型と一致することを検証します。

// 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 リファレンス