タスクをキューする

Agents SDK には、タスクを非同期実行向けにスケジュールできる組み込みキューがあります。バックグラウンド処理、遅延操作、すぐ実行しなくてよいワークロードの管理に使えます。

概要

キューはベースの Agent クラスに組み込まれています。タスクは SQLite テーブルに保存され、FIFO（First In, First Out）順で自動処理されます。

`QueueItem` 型

type QueueItem<T> = {
	id: string; // Unique identifier for the queued task
	payload: T; // Data to pass to the callback function
	callback: keyof Agent; // Name of the method to call
	created_at: number; // Timestamp when the task was created
	retry?: RetryOptions; // Retry options for this task
};

主要メソッド

`queue()`

将来の実行向けに、タスクをキューへ追加します。

async queue<T>(
  callback: keyof this,
  payload: T,
  options?: { retry?: RetryOptions }
): Promise<string>

パラメーター:

callback - タスク処理時に呼ぶメソッド名
payload - コールバックメソッドへ渡すデータ
options - 任意の設定:
- retry - コールバック実行の再試行オプション。コールバックが throw すると、指数バックオフで再試行します。RetryOptions の詳細は再試行を参照してください

戻り値: キューに入れたタスクの一意な ID

例:

class MyAgent extends Agent {
	async processEmail(data) {
		// Process the email
		console.log(`Processing email: ${data.subject}`);
	}

	async onMessage(message) {
		// Queue an email processing task
		const taskId = await this.queue("processEmail", {
			email: "user@example.com",
			subject: "Welcome!",
		});

		console.log(`Queued task with ID: ${taskId}`);
	}
}

class MyAgent extends Agent {
	async processEmail(data: { email: string; subject: string }) {
		// Process the email
		console.log(`Processing email: ${data.subject}`);
	}

	async onMessage(message: string) {
		// Queue an email processing task
		const taskId = await this.queue("processEmail", {
			email: "user@example.com",
			subject: "Welcome!",
		});

		console.log(`Queued task with ID: ${taskId}`);
	}
}

`dequeue()`

ID を指定して、キューから特定のタスクを削除します。このメソッドは同期です。

dequeue(id: string): void

パラメーター:

id - 削除するタスクの ID

例:

// Remove a specific task
agent.dequeue("abc123def");

// Remove a specific task
agent.dequeue("abc123def");

`dequeueAll()`

キューからすべてのタスクを削除します。このメソッドは同期です。

dequeueAll(): void

例:

// Clear the entire queue
agent.dequeueAll();

// Clear the entire queue
agent.dequeueAll();

`dequeueAllByCallback()`

特定のコールバックメソッドに一致するタスクをすべて削除します。このメソッドは同期です。

dequeueAllByCallback(callback: string): void

パラメーター:

callback - コールバックメソッド名

例:

// Remove all email processing tasks
agent.dequeueAllByCallback("processEmail");

// Remove all email processing tasks
agent.dequeueAllByCallback("processEmail");

`getQueue()`

ID を指定して、キュー内の特定タスクを取得します。このメソッドは同期です。

getQueue<T>(id: string): QueueItem<T> | undefined

パラメーター:

id - 取得するタスクの ID

戻り値: パース済みペイロード付きの QueueItem。見つからない場合は undefined

ペイロードは返す前に、JSON から自動でパースされます。

例:

const task = agent.getQueue("abc123def");
if (task) {
	console.log(`Task callback: ${task.callback}`);
	console.log(`Task payload:`, task.payload);
}

const task = agent.getQueue("abc123def");
if (task) {
	console.log(`Task callback: ${task.callback}`);
	console.log(`Task payload:`, task.payload);
}

`getQueues()`

ペイロード内の特定のキーと値に一致する、キュー内の全タスクを取得します。このメソッドは同期です。

getQueues<T>(key: string, value: string): QueueItem<T>[]

パラメーター:

key - ペイロード内で絞り込むキー
value - 一致させる値

戻り値: 一致した QueueItem オブジェクトの配列

このメソッドはキュー項目をすべて取得し、各ペイロードをパースして、指定キーが値と一致するかをメモリ上でフィルターします。

例:

// Find all tasks for a specific user
const userTasks = agent.getQueues("userId", "12345");

// Find all tasks for a specific user
const userTasks = agent.getQueues("userId", "12345");

キュー処理の仕組み

検証: queue() を呼ぶと、コールバックがエージェント上の関数として存在するか検証します。
自動処理: キュー投入後、システムは自動でキューのフラッシュを試みます。
FIFO 順: タスクは作成順（created_at タイムスタンプ）で処理されます。
コンテキストの保持: 各キュータスクは、同じエージェントコンテキスト（connection、request、email）で実行されます。
自動デキュー: 正常に実行されたタスクは、キューから自動で削除されます。
エラー処理: 実行時にコールバックメソッドが存在しない場合、エラーを記録してタスクをスキップします。
永続化: タスクは cf_agents_queues SQL テーブルに保存され、エージェントの再起動後も残ります。

キューのコールバックメソッド

キュータスク用のコールバックメソッドは、次のシグネチャにしてください。

async callbackMethod(payload: unknown, queueItem: QueueItem): Promise<void>

例:

class MyAgent extends Agent {
	async sendNotification(payload, queueItem) {
		console.log(`Processing task ${queueItem.id}`);
		console.log(
			`Sending notification to user ${payload.userId}: ${payload.message}`,
		);

		// Your notification logic here
		await this.notificationService.send(payload.userId, payload.message);
	}

	async onUserSignup(userData) {
		// Queue a welcome notification
		await this.queue("sendNotification", {
			userId: userData.id,
			message: "Welcome to our platform!",
		});
	}
}

class MyAgent extends Agent {
	async sendNotification(
		payload: { userId: string; message: string },
		queueItem: QueueItem<{ userId: string; message: string }>,
	) {
		console.log(`Processing task ${queueItem.id}`);
		console.log(
			`Sending notification to user ${payload.userId}: ${payload.message}`,
		);

		// Your notification logic here
		await this.notificationService.send(payload.userId, payload.message);
	}

	async onUserSignup(userData: any) {
		// Queue a welcome notification
		await this.queue("sendNotification", {
			userId: userData.id,
			message: "Welcome to our platform!",
		});
	}
}

ユースケース

バックグラウンド処理

class DataProcessor extends Agent {
	async processLargeDataset(data) {
		const results = await this.heavyComputation(data.datasetId);
		await this.notifyUser(data.userId, results);
	}

	async onDataUpload(uploadData) {
		// Queue the processing instead of doing it synchronously
		await this.queue("processLargeDataset", {
			datasetId: uploadData.id,
			userId: uploadData.userId,
		});

		return { message: "Data upload received, processing started" };
	}
}

class DataProcessor extends Agent {
	async processLargeDataset(data: { datasetId: string; userId: string }) {
		const results = await this.heavyComputation(data.datasetId);
		await this.notifyUser(data.userId, results);
	}

	async onDataUpload(uploadData: any) {
		// Queue the processing instead of doing it synchronously
		await this.queue("processLargeDataset", {
			datasetId: uploadData.id,
			userId: uploadData.userId,
		});

		return { message: "Data upload received, processing started" };
	}
}

バッチ操作

class BatchProcessor extends Agent {
	async processBatch(data) {
		for (const item of data.items) {
			await this.processItem(item);
		}
		console.log(`Completed batch ${data.batchId}`);
	}

	async onLargeRequest(items) {
		// Split large requests into smaller batches
		const batchSize = 10;
		for (let i = 0; i < items.length; i += batchSize) {
			const batch = items.slice(i, i + batchSize);
			await this.queue("processBatch", {
				items: batch,
				batchId: `batch-${i / batchSize + 1}`,
			});
		}
	}
}

class BatchProcessor extends Agent {
	async processBatch(data: { items: any[]; batchId: string }) {
		for (const item of data.items) {
			await this.processItem(item);
		}
		console.log(`Completed batch ${data.batchId}`);
	}

	async onLargeRequest(items: any[]) {
		// Split large requests into smaller batches
		const batchSize = 10;
		for (let i = 0; i < items.length; i += batchSize) {
			const batch = items.slice(i, i + batchSize);
			await this.queue("processBatch", {
				items: batch,
				batchId: `batch-${i / batchSize + 1}`,
			});
		}
	}
}

エラー処理

手動で再キューする代わりに、組み込みの retry オプションを使います。コールバックが throw すると、タスクは指数バックオフで自動再試行されます。

class RobustAgent extends Agent {
	async reliableTask(payload, queueItem) {
		console.log(`Processing task ${queueItem.id}`);
		const response = await fetch(payload.url);
		if (!response.ok) {
			throw new Error(`Request failed: ${response.status}`);
		}
	}

	async onMessage(connection, message) {
		await this.queue(
			"reliableTask",
			{ url: "https://api.example.com/data" },
			{
				retry: {
					maxAttempts: 5,
					baseDelayMs: 500,
					maxDelayMs: 10_000,
				},
			},
		);
	}
}

class RobustAgent extends Agent {
	async reliableTask(payload: { url: string }, queueItem: QueueItem) {
		console.log(`Processing task ${queueItem.id}`);
		const response = await fetch(payload.url);
		if (!response.ok) {
			throw new Error(`Request failed: ${response.status}`);
		}
	}

	async onMessage(connection: Connection, message: WSMessage) {
		await this.queue(
			"reliableTask",
			{ url: "https://api.example.com/data" },
			{
				retry: {
					maxAttempts: 5,
					baseDelayMs: 500,
					maxDelayMs: 10_000,
				},
			},
		);
	}
}

retry オプションを渡さない場合は、static options.retry のクラスレベルの既定値（3 回、ベース遅延 100ms、最大遅延 3s）が使われます。詳細は再試行を参照してください。

ベストプラクティス

ペイロードは小さく: ペイロードは JSON シリアライズされ、データベースに保存されます。
冪等な操作: コールバックメソッドは再試行しても安全な設計にします。
エラー処理: コールバックメソッドに適切なエラー処理を含めます。
監視: ログでキュー処理を追跡します。
クリーンアップ: 必要なら完了済みや失敗したタスクを定期的に片付けます。

ほかの機能との連携

キューは、ほかの Agent SDK 機能と組み合わせて使えます。

状態管理: キューのコールバック内でエージェント状態にアクセスできます。
スケジュール: 時刻ベースのキュー処理には schedule() と組み合わせます。
コンテキスト: キュータスクは、元のリクエストコンテキストを維持します。
データベース: ほかのエージェントデータと同じデータベースを使います。

制限

タスクは並列ではなく、順次処理されます。
優先度はありません（FIFO のみ）。
キュー処理は、別のバックグラウンドジョブではなく、エージェント実行中に行われます。

キューとスケジュールの比較

できるだけ早く順番どおり実行したいときは キュー を使います。特定時刻や定期実行が必要なときは スケジュール を使います。

機能	キュー	スケジュール
実行タイミング	即時（FIFO）	指定時刻または cron
用途	バックグラウンド処理	遅延または定期タスク
ストレージ	`cf_agents_queues` テーブル	`cf_agents_schedules` テーブル

次のステップ

Agents API

Agents SDK の API リファレンスです。

タスクをスケジュールする

cron と遅延による時刻指定の実行です。

Workflows を実行する

耐久的な複数ステップのバックグラウンド処理です。

タスクをキューする

概要

QueueItem 型

主要メソッド

queue()

dequeue()

dequeueAll()

dequeueAllByCallback()

getQueue()

getQueues()

キュー処理の仕組み

キューのコールバックメソッド

ユースケース

バックグラウンド処理

バッチ操作

エラー処理

ベストプラクティス

ほかの機能との連携

制限

キューとスケジュールの比較

次のステップ

Agents API

タスクをスケジュールする

Workflows を実行する

`QueueItem` 型

`queue()`

`dequeue()`

`dequeueAll()`

`dequeueAllByCallback()`

`getQueue()`

`getQueues()`