Agents と Workflows を組み合わせる

Cloudflare Workflows は、失敗を越え、自動リトライし、外部イベントを待つ必要があるタスク向けの、耐久的な複数ステップ実行です。Agents と連携すると、長時間のバックグラウンド処理は Workflows が担当し、リアルタイム通信は Agents が担当します。

Agents と Workflows は、互いに補完する強みを持ちます。

Agents はループ、分岐、ユーザーとの直接対話ができます。Workflows はステップを順に実行し、配信を保証し、承認や外部データを数日待つために一時停止できます。

Agents だけを使う場合:

• チャットとメッセージングアプリ
• 短い API 呼び出しと応答
• リアルタイムの共同作業機能
• 30 秒未満のタスク
• submitMessages() による、1 回の耐久的な Think チャットターン

Agents と Workflows を一緒に使う場合:

• データ処理パイプライン
• レポート生成
• Human-in-the-loop の承認フロー
• 配信保証が必要なタスク
• リトライが必要な複数ステップ操作

Workflows だけを使う場合:

• ユーザー承認の有無を問わないバックグラウンドジョブ
• スケジュールされたデータ同期
• イベント駆動の処理パイプライン

AgentWorkflow クラス（agents/workflows からインポート）は、Workflows と、それを始めたエージェントの双方向通信を提供します。

Workflows は、次の仕組みでエージェントと通信できます。

• RPC 呼び出し: this.agent 経由で、型安全のままエージェントメソッドを直接呼びます
• 進捗報告: this.reportProgress() で進捗を送り、エージェントのコールバックを起動します
• 状態更新: step.updateAgentState() または step.mergeAgentState() でエージェント状態を変更し、接続中のクライアントへブロードキャストします
• クライアントへのブロードキャスト: this.broadcastToClients() ですべての WebSocket クライアントへメッセージを送ります

// Inside a workflow's run() method
await this.agent.updateTaskStatus(taskId, "processing"); // RPC call
await this.reportProgress({ step: "process", percent: 0.5 }); // Progress (non-durable)
this.broadcastToClients({ type: "update", taskId }); // Broadcast (non-durable)
await step.mergeAgentState({ taskProgress: 0.5 }); // State update (durable)

// Inside a workflow's run() method
await this.agent.updateTaskStatus(taskId, "processing"); // RPC call
await this.reportProgress({ step: "process", percent: 0.5 }); // Progress (non-durable)
this.broadcastToClients({ type: "update", taskId }); // Broadcast (non-durable)
await step.mergeAgentState({ taskProgress: 0.5 }); // State update (durable)

実行中の Workflows に対して、エージェントは次ができます。

• Workflow の開始: runWorkflow() で新しい Workflow インスタンスを起動します
• イベント送信: sendWorkflowEvent() でイベントを送ります
• 承認 / 却下: approveWorkflow() / rejectWorkflow() で承認リクエストに応答します
• Workflow 制御: Workflow の一時停止、再開、終了、再起動
• 状態照会: getWorkflow() / getWorkflows() で進捗を確認します

Workflows を効果的に使うには、耐久性の理解が重要です。

軽量で頻繁な更新に向きます。ただし Workflow がリトライすると、複数回実行されることがあります。

• this.reportProgress() — 進捗報告
• this.broadcastToClients() — WebSocket ブロードキャスト
• this.agent への直接 RPC 呼び出し

step パラメータを使い、ちょうど 1 回だけ実行されることが保証されます。

• step.do() — 耐久ステップの実行
• step.reportComplete() / step.reportError() — 完了報告
• step.sendEvent() — カスタムイベント
• step.updateAgentState() / step.mergeAgentState() — 状態同期

Workflows は、ステップ単位の実行で耐久性を提供します。

1. ステップ完了は永続です — 一度完了したステップは、Workflow が再起動しても再実行されません
2. 自動リトライ — 失敗したステップは、設定可能なバックオフでリトライします
3. イベントの永続化 — Workflows は最大 1 年、イベントを待てます
4. 状態の復旧 — Workflow の状態はインフラ障害を越えて残ります

この耐久モデルでは、途中までの完了を残す必要があるタスク、たとえば複数段階のデータ処理や複数システムにまたがるトランザクションに、Workflows が向きます。

エージェントが runWorkflow() で Workflow を始めると、その Workflow はエージェント内部のデータベースで自動追跡されます。これにより次ができます。

• ID、名前、メタデータで Workflow 状態を照会し、カーソルベースのページネーションを使います
• ライフサイクルコールバック（onWorkflowProgress、onWorkflowComplete、onWorkflowError）で進捗を監視します
• Workflow 制御: 一時停止、再開、終了、再起動
• deleteWorkflow() / deleteWorkflows() で完了済みレコードを掃除します
• メタデータで Workflow をユーザーやセッションと対応付けます

エージェントがリクエストを受け、重い処理用に Workflow を始め、Workflow が各ステップを実行するたびに接続中のクライアントへ進捗をブロードキャストします。

// Workflow reports progress after each item
for (let i = 0; i < items.length; i++) {
	await step.do(`process-${i}`, async () => processItem(items[i]));
	await this.reportProgress({
		step: `process-${i}`,
		percent: (i + 1) / items.length,
		message: `Processed ${i + 1}/${items.length}`,
	});
}

// Workflow reports progress after each item
for (let i = 0; i < items.length; i++) {
	await step.do(`process-${i}`, async () => processItem(items[i]));
	await this.reportProgress({
		step: `process-${i}`,
		percent: (i + 1) / items.length,
		message: `Processed ${i + 1}/${items.length}`,
	});
}

Workflow がリクエストを準備し、waitForApproval() で承認待ちに入ります。エージェントは approveWorkflow() / rejectWorkflow() でユーザーが承認または却下できる UI を提供します。判断に応じて Workflow は再開するか、WorkflowRejectedError を投げます。

Workflow は外部 API 呼び出しを、リトライ付きの耐久ステップで包みます。API が失敗しても Workflow が再起動しても、完了済みの呼び出しは繰り返さず、失敗した呼び出しは自動リトライします。

const result = await step.do(
	"call-api",
	{
		retries: { limit: 5, delay: "10 seconds", backoff: "exponential" },
		timeout: "5 minutes",
	},
	async () => {
		const response = await fetch("https://api.example.com/process");
		if (!response.ok) throw new Error(`API error: ${response.status}`);
		return response.json();
	},
);

const result = await step.do(
	"call-api",
	{
		retries: { limit: 5, delay: "10 seconds", backoff: "exponential" },
		timeout: "5 minutes",
	},
	async () => {
		const response = await fetch("https://api.example.com/process");
		if (!response.ok) throw new Error(`API error: ${response.status}`);
		return response.json();
	},
);

Workflow は、主要な節目で step.updateAgentState() または step.mergeAgentState() を使い、エージェント状態を更新します。この状態変更は接続中の全クライアントへブロードキャストされ、ポーリングなしで UI を同期できます。