← Vytvářejte agenty na Cloudflare / agents / concepts

Používání Agents s Workflows

Cloudflare Workflows poskytují trvalé, vícekrokové provádění úloh, které musí přežít selhání, automaticky se opakovat a čekat na externí události. Při integraci s Agents se Workflows starají o dlouhotrvající zpracování na pozadí, zatímco Agents zajišťují komunikaci v reálném čase.

Agents a Workflows mají vzájemně se doplňující silné stránky:

Agents mohou vytvářet smyčky, větvit se a přímo interagovat s uživateli. Workflows provádějí kroky sekvenčně se zaručeným doručením a mohou se na dny pozastavit a čekat na schválení nebo externí data.

Používejte Agents samostatně pro:

• Aplikace pro chat a zasílání zpráv
• Rychlá volání API a odpovědi
• Funkce pro spolupráci v reálném čase
• Úlohy kratší než 30 sekund
• Jedno trvalé kolo chatu Think s submitMessages()

Používejte Agents spolu s Workflows pro:

• Pipeline pro zpracování dat
• Generování reportů
• Schvalovací postupy s lidským zásahem
• Úlohy vyžadující zaručené doručení
• Vícekrokové operace s požadavky na opakované pokusy

Používejte Workflows samostatně pro:

• Úlohy na pozadí se schválením uživatele i bez něj
• Naplánovaná synchronizace dat
• Procesní pipeline řízené událostmi

AgentWorkflow třída (importovaná z agents/workflows) poskytuje obousměrnou komunikaci mezi Workflow a jejich zdrojovým Agentem.

Workflows mohou s Agents komunikovat několika způsoby:

• Volání RPC: Volejte metody Agent přímo, s plnou typovou bezpečností, prostřednictvím this.agent
• Hlášení postupu: Odesílejte aktualizace průběhu pomocí this.reportProgress() které spouštějí callbacky Agenta
• Aktualizace stavu: Upravujte stav Agent prostřednictvím step.updateAgentState() nebo step.mergeAgentState(), který vysílá připojeným klientům
• Vysílání klienta: Odesílejte zprávy všem klientům WebSocket pomocí this.broadcastToClients()

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

Agents mohou s běžícími Workflows interagovat těmito způsoby:

• Spouštění workflows: Spouštějte nové instance workflow pomocí runWorkflow()
• Odesílání událostí: Odesílejte události pomocí sendWorkflowEvent()
• Schválení/zamítnutí: Reagujte na žádosti o schválení pomocí approveWorkflow() / rejectWorkflow()
• Řízení workflow: Pozastavte, obnovte, ukončete nebo restartujte workflow
• Dotazy na stav: Zkontrolujte průběh workflow pomocí getWorkflow() / getWorkflows()

Pochopení trvanlivosti je klíčové pro efektivní používání workflow:

Tyto operace jsou odlehčené a vhodné pro časté aktualizace, ale pokud workflow provede nový pokus, mohou se spustit vícekrát:

• this.reportProgress() : Hlášení průběhu
• this.broadcastToClients() : Vysílání přes WebSocket
• Přímá volání RPC do this.agent

Tyto operace využívají step parametr a je zaručeno, že se provedou přesně jednou:

• step.do() : Provádění trvalých kroků
• step.reportComplete() / step.reportError() : Hlášení o dokončení
• step.sendEvent() : Vlastní události
• step.updateAgentState() / step.mergeAgentState() : Synchronizace stavu

Workflows zajišťují odolnost pomocí spouštění po jednotlivých krocích:

1. Dokončení kroku je trvalé : Jakmile je krok dokončen, znovu se nespustí, ani pokud dojde k restartu workflow
2. Automatická opakování : Neúspěšné kroky se opakují s nastavitelnou prodlevou mezi pokusy
3. Perzistence událostí : Workflows mohou čekat na události až jeden rok
4. Obnovení stavu : Stav workflow přežije selhání infrastruktury

Tento model trvalosti znamená, že workflows se dobře hodí pro úlohy, u kterých je třeba zachovat částečné dokončení, například vícestupňové zpracování dat nebo transakce probíhající napříč více systémy.

Když Agent spustí workflow pomocí runWorkflow(), workflow se automaticky sleduje v interní databázi agenta. To umožňuje:

• Dotazování na stav workflow podle ID, názvu nebo metadat se stránkováním založeným na kurzoru
• Sledování průběhu prostřednictvím callbacků životního cyklu (onWorkflowProgress, onWorkflowComplete, onWorkflowError)
• Řízení workflow: pozastavení, pokračování, ukončení, restart
• Čištění dokončených záznamů Workflows pomocí deleteWorkflow() / deleteWorkflows()
• Korelace workflows s uživateli nebo relacemi pomocí metadat

Agent přijme požadavek, spustí Workflow pro náročné zpracování a v průběhu jednotlivých kroků Workflow vysílá aktualizace o průběhu připojeným klientům.

// 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 připraví požadavek a pozastaví se, aby počkal na schválení pomocí waitForApproval(), a Agent poskytuje uživatelské rozhraní, ve kterém uživatelé mohou schválit nebo zamítnout přes approveWorkflow() / rejectWorkflow(). Workflow pokračuje nebo vyvolá WorkflowRejectedError na základě rozhodnutí.

Workflow obaluje volání externího API do trvalých kroků s logikou opakování. Pokud API selže nebo se workflow restartuje, dokončená volání se neopakují a neúspěšná volání se opakují automaticky.

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 aktualizuje stav Agenta v klíčových milnících pomocí step.updateAgentState() nebo step.mergeAgentState(). Tyto změny stavu se rozešlou všem připojeným klientům, takže se uživatelská rozhraní synchronizují bez nutnosti dotazování (pollingu).