← Vytvářejte agenty na Cloudflare / agents / runtime / execution

Zařazování úloh do fronty

Agents SDK poskytuje vestavěný systém front, který umožňuje plánovat úlohy pro asynchronní spuštění. To se hodí pro zpracování na pozadí, zpožděné operace a správu úloh, které nevyžadují okamžité spuštění.

Přehled

Systém front je zabudovaný do základní Agent třída. Úlohy se ukládají do tabulky SQLite a automaticky se zpracovávají v pořadí FIFO (First In, First Out).

`QueueItem` typ

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

Základní metody

`queue()`

Přidá úlohu do fronty pro pozdější spuštění.

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

Parametry:

callback - Název metody, která se má volat při zpracování úlohy
payload - Data, která se předají metodě callbacku
options - Volitelná konfigurace:
- retry - Možnosti opakování pro provedení callbacku. Pokud callback vyvolá výjimku, opakuje se s exponenciálním prodlužováním intervalu. Viz Opakování pro podrobnosti o RetryOptions

Vrací: Jedinečné ID úlohy zařazené do fronty

Příklad:

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()`

Odebere konkrétní úlohu z fronty podle ID. Tato metoda je synchronní.

dequeue(id: string): void

Parametry:

id - ID úlohy, která se má odebrat

Příklad:

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

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

`dequeueAll()`

Odebere všechny úlohy z fronty. Tato metoda je synchronní.

dequeueAll(): void

Příklad:

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

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

`dequeueAllByCallback()`

Odebere všechny úlohy, které odpovídají konkrétní metodě callbacku. Tato metoda je synchronní.

dequeueAllByCallback(callback: string): void

Parametry:

callback - Název callback metody

Příklad:

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

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

`getQueue()`

Načte konkrétní zařazenou úlohu podle ID. Tato metoda je synchronní.

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

Parametry:

id - ID úlohy, která se má načíst

Vrací: QueueItem s parsovaným payloadem nebo undefined pokud nebylo nalezeno

Payload se před vrácením automaticky parsuje z formátu JSON.

Příklad:

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()`

Načte všechny zařazené úlohy, jejichž payload odpovídá konkrétnímu páru klíč-hodnota. Tato metoda je synchronní.

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

Parametry:

key - Klíč, podle kterého se filtruje v payloadu
value - Hodnota, která se má porovnat

Vrací: Pole odpovídajících QueueItem objekty

Tato metoda načte všechny položky fronty a filtruje je v paměti: rozparsuje každý payload a ověří, zda zadaný klíč odpovídá hodnotě.

Příklad:

// 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");

Jak funguje zpracování fronty

Validace: Při volání queue(), metoda ověří, že daný callback existuje jako funkce agenta.
Automatické zpracování: Po zařazení do fronty se systém automaticky pokusí frontu vyprázdnit.
Pořadí FIFO: Úlohy se zpracovávají v pořadí, v jakém byly vytvořeny (created_at časové razítko).
Zachování kontextu: Každá úloha zařazená do fronty běží se stejným kontextem agenta (připojení, požadavek, e-mail).
Automatické odstranění z fronty: Úspěšně provedené úlohy se automaticky odstraní z fronty.
Zpracování chyb: Pokud metoda callbacku v okamžiku spuštění neexistuje, zaznamená se chyba a úloha se přeskočí.
Perzistence: Úlohy se ukládají do cf_agents_queues SQL tabulce a přežívají restart agenta.

Metody callbacku fronty

Při definování metod callbacků pro úlohy zařazené do fronty musí mít tuto signaturu:

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

Příklad:

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!",
		});
	}
}

Případy použití

Zpracování na pozadí

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" };
	}
}

Dávkové operace

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}`,
			});
		}
	}
}

Zpracování chyb

Použijte vestavěný retry místo ruční logiky opětovného zařazení do fronty. Když callback vyvolá výjimku, úloha se automaticky opakuje s exponential backoff:

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

Pokud žádný retry možnost je zadána, výchozí hodnoty na úrovni třídy z static options.retry se používají (3 pokusy, základní zpoždění 100 ms, maximální zpoždění 3 s). Podrobnosti najdete v Opakování pro úplné podrobnosti.

Osvědčené postupy

Udržujte payloady malé: Payloady se serializují do JSON a ukládají do databáze.
Idempotentní operace: Navrhněte metody callbacků tak, aby bylo bezpečné je opakovat.
Zpracování chyb: Zahrňte do metod callbacků řádné ošetření chyb.
Monitoring: Pomocí logování sledujte zpracování fronty.
Vyčištění: Podle potřeby pravidelně čistěte dokončené nebo neúspěšné úlohy.

Integrace s dalšími funkcemi

Systém front spolupracuje s dalšími funkcemi Agent SDK:

Správa stavu: Přistupte ke stavu agenta v rámci callbacků zařazených do fronty.
Plánování: Kombinujte s schedule() pro časově řízené zpracování fronty.
Kontext: Úlohy zařazené do fronty si zachovávají původní kontext požadavku.
Databáze: Používá stejnou databázi jako ostatní data agenta.

Omezení

Úlohy se zpracovávají postupně, ne paralelně.
Žádný systém priorit (pouze FIFO).
Zpracování fronty probíhá během běhu agenta, nikoli jako samostatné úlohy na pozadí.

Zařazování do fronty oproti plánování

Použijte fronta když chcete, aby se úlohy spouštěly co nejdříve v daném pořadí. Použijte plán když potřebujete, aby se úlohy spouštěly v konkrétní časy nebo opakovaně.

Funkce	Queue	Naplánovat
Časování spuštění	Okamžité (FIFO)	Konkrétní čas nebo cron
Případ použití	Zpracování na pozadí	Odložené nebo opakující se úlohy
Úložiště	`cf_agents_queues` tabulka	`cf_agents_schedules` tabulka

Zařazování úloh do fronty

Přehled

`QueueItem` typ

Základní metody

`queue()`

`dequeue()`

`dequeueAll()`

`dequeueAllByCallback()`

`getQueue()`

`getQueues()`

Jak funguje zpracování fronty

Metody callbacku fronty

Případy použití

Zpracování na pozadí

Dávkové operace

Zpracování chyb

Osvědčené postupy

Integrace s dalšími funkcemi

Omezení

Zařazování do fronty oproti plánování

Další kroky

Agents API

Plánování úloh

Spouštění Workflows

Zařazování úloh do fronty

Přehled

QueueItem typ

Základní metody

queue()

dequeue()

dequeueAll()

dequeueAllByCallback()

getQueue()

getQueues()

Jak funguje zpracování fronty

Metody callbacku fronty

Případy použití

Zpracování na pozadí

Dávkové operace

Zpracování chyb

Osvědčené postupy

Integrace s dalšími funkcemi

Omezení

Zařazování do fronty oproti plánování

Další kroky

Agents API

Plánování úloh

Spouštění Workflows

`QueueItem` typ

`queue()`

`dequeue()`

`dequeueAll()`

`dequeueAllByCallback()`

`getQueue()`

`getQueues()`