INTEGRITY Dokumentace

← Vytvářejte agenty na Cloudflare / agents / harnesses / think

Workflows

ThinkWorkflow propojuje Think s Cloudflare Workflows, když trvalá úloha potřebuje jeden krok uvažování řízený modelem.

Použijte ho, když proces vlastní Workflow:

Opakující se prompty uchovávejte jako naplánované úlohy, a jednoduchá jednorázová kola na pozadí ponechte na submitMessages(). Workflows jsou určené pro úlohy, u kterých záleží na jednotlivých krocích.

API

Importujte z @cloudflare/think/workflows:

import { ThinkWorkflow } from "@cloudflare/think/workflows";

Rozšiřte ThinkWorkflow a volejte step.prompt() uvnitř run():

import { z } from "zod";
import { ThinkWorkflow } from "@cloudflare/think/workflows";

const draftSchema = z.object({
	title: z.string(),
	summary: z.string(),
	labels: z.array(z.string()),
});

export class TriageWorkflow extends ThinkWorkflow {
	async run(event, step) {
		const draft = await step.prompt("triage-issue", {
			prompt: `Triage issue #${event.payload.issueNumber}`,
			output: draftSchema,
			timeout: "3 days",
		});

		await step.do("apply-labels", async () => {
			await this.agent.applyLabels(draft.labels);
		});
	}
}
import { z } from "zod";
import { ThinkWorkflow } from "@cloudflare/think/workflows";
import type { ThinkWorkflowStep } from "@cloudflare/think/workflows";
import type { AgentWorkflowEvent } from "agents/workflows";

const draftSchema = z.object({
	title: z.string(),
	summary: z.string(),
	labels: z.array(z.string()),
});

export class TriageWorkflow extends ThinkWorkflow<TriageAgent, Params> {
	async run(event: AgentWorkflowEvent<Params>, step: ThinkWorkflowStep) {
		const draft = await step.prompt("triage-issue", {
			prompt: `Triage issue #${event.payload.issueNumber}`,
			output: draftSchema,
			timeout: "3 days",
		});

		await step.do("apply-labels", async () => {
			await this.agent.applyLabels(draft.labels);
		});
	}
}

Spusťte Workflow uvnitř svého Think Agent pomocí runWorkflow():

export class TriageAgent extends Think {
	async triageIssue(issueNumber) {
		return this.runWorkflow(
			"TRIAGE_WORKFLOW",
			{ issueNumber },
			{ metadata: { issueNumber } },
		);
	}
}
export class TriageAgent extends Think<Env> {
	async triageIssue(issueNumber: number): Promise<string> {
		return this.runWorkflow(
			"TRIAGE_WORKFLOW",
			{ issueNumber },
			{ metadata: { issueNumber } },
		);
	}
}

runWorkflow() vytvoří instanci Workflow a vloží identitu Agenta, kterou ThinkWorkflow se potřebuje znovu připojit k this.agent uvnitř run(). Upřednostněte ji před přímým voláním vazby Workflows:

// Avoid this for Agent workflows. It does not include Agent context.
await this.env.TRIAGE_WORKFLOW.create({ params: { issueNumber } });

Použijte sendWorkflowEvent() z Agenta, když čekající Workflow potřebuje externí signál, například schválení člověkem:

await this.sendWorkflowEvent("TRIAGE_WORKFLOW", workflowId, {
	type: "approval",
	payload: { approved: true },
});

step.prompt() přijímá řetězec s promptem a objektové schéma Zod. Toto schéma se před voláním Agentu z Workflow převede na JSON Schema. Think poté spustí celý agentní tah: Agent může v rámci více kroků použít své nástroje a strukturovaný výsledek vrátí voláním interní final_answer nástroj, jehož argumenty odpovídají schématu. Používá se běžné volání nástrojů místo streamovacího response_format, takže to funguje u všech poskytovatelů, které Think podporuje, včetně Workers AI, který u streamovaných požadavků odmítá odpovědi ve formátu JSON Schema. Když se Workflow obnoví, payload se znovu ověří podle původního schématu Zod, než se vrátí typovaná hodnota.

Nepodporované funkce Zod, které nelze vyjádřit jako JSON Schema, způsobí selhání při vytváření kroku promptu. Think tiše neopravuje neplatný výstup modelu. Pokud model nevytvoří platný final_answer volání se odeslání dostane do konečného chybového stavu a step.prompt() vyvolá výjimku.

Poznámky k chování

Jak to běží

Volání vypadá jako blokující krok, ale ve skutečnosti nedrží dlouhodobě otevřené RPC k Durable Object.

  1. step.do("<name>:submit", ...) vytvoří nebo najde idempotentní odeslání Think.
  2. Think spouští odeslané kolo přes běžnou frontu odeslání.
  3. Když podání dosáhne completed, error, aborted, nebo skipped, Think zaznamená čekající oznámení o workflow.
  4. Think vyprazdňuje frontu odchozích oznámení pomocí sendWorkflowEvent() a alarmy Durable Object, dokud se doručení neprovede úspěšně.
  5. step.waitForEvent("<name>:wait", ...) obnoví Workflow.
  6. step.prompt() ověří strukturovaný výstup, nebo vyvolá typovanou výjimku.

Strojově čitelný výstup se přenáší v payloadu čekajícího oznámení a v payloadu události Workflow. Think neukládá samostatný output_json sloupec v evidenci odeslání a po doručení vymaže payload notifikace. Po doručení vlastní trvalý výsledek Workflow.

Idempotence

Standardně step.prompt() odvodí klíč idempotence z identity Workflow a názvu kroku:

think-workflow:<workflowName>:<workflowId>:<stepName>

Pro smyčky předejte řetězec key k rozlišení opakovaného použití stejného názvu kroku:

await step.prompt("summarize-file", {
	key: file.path,
	prompt: `Summarize ${file.path}`,
	output: summarySchema,
});

Text promptu není součástí odvozeného klíče, Think ale pro diagnostiku ukládá metadata workflow a otisk promptu/konfigurace.

Časové limity

Předejte timeout pro řízení toho, jak dlouho Workflow čeká na konečnou událost. Pokud čekání vyprší, step.prompt() ve výchozím nastavení zruší odeslání Think a vyvolá ThinkPromptTimeoutError.

Nastavte cancelOnTimeout: false když záměrně chcete, aby odeslání Think pokračovalo i poté, co Workflow přestane čekat.

Hranice vůči ostatním primitivům

Použijte getScheduledTasks() pro opakovaná odeslání promptů nebo deterministické naplánované obslužné rutiny:

getScheduledTasks() {
	return {
		dailySummary: {
			schedule: "every day at 09:00",
			timezone: "UTC",
			prompt: "Generate the daily report."
		},
		dailyWorkflow: {
			schedule: "every day at 09:00",
			timezone: "UTC",
			retry: { maxAttempts: 3 },
			handler: async ({ idempotencyKey, scheduledFor, timezone }) => {
				await this.env.REPORT_WORKFLOW.create({
					id: idempotencyKey,
					params: { scheduledFor, timezone }
				});
			}
		}
	};
}

Použijte submitMessages() pro trvalá jednorázová kola, u nichž volající může stav odeslání zkontrolovat později.

Použijte startFiber() pro Agentem vlastněné idempotentní úlohy, které vyžadují zotavení uvnitř Agenta. Doručování oznámení workflow ve službě Think nepoužívá fibery, ale privátní outbox, protože potřebuje uchovat událost, dokud se úspěšně nedoručí.

Použijte Workflows, pokud proces zahrnuje více deterministických kroků, dlouhá čekání nebo schválení člověkem.