← Создание агентов на Cloudflare / agents / runtime / execution

Задачи очереди

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 - Параметры повторных попыток для выполнения callback. Если callback выбрасывает исключение, попытка повторяется с экспоненциальной отсрочкой. См. Повторы для подробностей о 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: "[email protected]",
			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: "[email protected]",
			subject: "Welcome!",
		});

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

`dequeue()`

Удаляет конкретную задачу из очереди по ID. Этот метод синхронный.

dequeue(id: string): void

Параметры:

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 - Имя метода 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 - Идентификатор задачи, которую нужно получить

Возвращает: Заголовок QueueItem с разобранной полезной нагрузкой или undefined если не найдено

Payload автоматически разбирается из 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()`

Извлекает все задачи из очереди, соответствующие определённой паре ключ-значение в их payload. Этот метод синхронный.

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

Параметры:

key - Ключ для фильтрации в payload
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 временная метка).
Сохранение контекста: Каждая задача из очереди выполняется в том же контексте агента (соединение, запрос, 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 опцию вместо ручной логики повторной постановки в очередь. Если колбэк выбрасывает исключение, задача автоматически повторяется с экспоненциальной задержкой:

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 попытки, базовая задержка 100 мс, максимальная задержка 3 с). См. Повторы для полных подробностей.

Интеграция с другими функциями

Система очередей работает вместе с другими функциями Agent SDK:

Управление состоянием: Доступ к состоянию агента внутри поставленных в очередь колбэков.
Планирование: Сочетайте с schedule() для обработки очереди по времени.
Контекст: Задачи в очереди сохраняют исходный контекст запроса.
База данных: Использует ту же базу данных, что и остальные данные агента.

Ограничения

Задачи обрабатываются последовательно, а не параллельно.
Без системы приоритетов (только FIFO).
Обработка очереди происходит во время выполнения агента, а не в виде отдельных фоновых задач.

Queue и Schedule

Используйте очередь когда задачи должны выполняться по порядку как можно быстрее. Используйте расписание когда задачи нужно выполнять в конкретное время или на регулярной основе.

Возможность	Queue	Запланировать
Время выполнения	Немедленный (FIFO)	Конкретное время или cron
Сценарий использования	Фоновая обработка	Отложенные или периодические задачи
Хранилище	`cf_agents_queues` таблица	`cf_agents_schedules` таблица

Задачи очереди

Обзор

`QueueItem` тип

Основные методы

`queue()`

`dequeue()`

`dequeueAll()`

`dequeueAllByCallback()`

`getQueue()`

`getQueues()`

Как работает обработка очереди

Методы колбэков очереди

Сценарии использования

Фоновая обработка

Пакетные операции

Обработка ошибок

Рекомендации

Интеграция с другими функциями

Ограничения

Queue и Schedule

Следующие шаги

Agents API

Планирование задач

Запуск Workflows

Задачи очереди

Обзор

QueueItem тип

Основные методы

queue()

dequeue()

dequeueAll()

dequeueAllByCallback()

getQueue()

getQueues()

Как работает обработка очереди

Методы колбэков очереди

Сценарии использования

Фоновая обработка

Пакетные операции

Обработка ошибок

Рекомендации

Интеграция с другими функциями

Ограничения

Queue и Schedule

Следующие шаги

Agents API

Планирование задач

Запуск Workflows

`QueueItem` тип

`queue()`

`dequeue()`

`dequeueAll()`

`dequeueAllByCallback()`

`getQueue()`

`getQueues()`