INTEGRITY Документация

← Создание агентов на Cloudflare / agents / harnesses / think

Workflows

ThinkWorkflow связывает Think с Cloudflare Workflows, когда устойчивой задаче нужен один шаг рассуждения, управляемый моделью.

Используйте это, если процессом управляет Workflow:

Оформляйте повторяющиеся запросы как запланированные задачи, а простые одноразовые фоновые ходы оставьте на submitMessages(). Workflows предназначены для задач, где важны отдельные шаги.

API

Импортируйте из @cloudflare/think/workflows:

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

Расширить ThinkWorkflow и вызвать step.prompt() внутри 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);
		});
	}
}

Запустите Workflow изнутри вашего Think Agent с помощью 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() создаёт экземпляр Workflow и внедряет идентификатор Agent, который ThinkWorkflow необходимо переподключиться к this.agent внутри run(). Предпочитайте этот способ прямому вызову привязки Workflows:

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

Используйте sendWorkflowEvent() из Agent, когда ожидающему Workflow требуется внешний сигнал, например, подтверждение человеком:

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

step.prompt() принимает строку запроса и объектную схему Zod. Перед тем как Workflow вызывает Agent, схема преобразуется в JSON Schema. Затем Think выполняет полный агентный ход: Agent может использовать свои инструменты в несколько шагов и возвращает структурированный результат, вызывая внутренний final_answer инструмент, аргументы которого соответствуют схеме. Здесь используется обычный вызов инструмента, а не потоковый response_format, чтобы это работало у всех провайдеров, которые поддерживает Think, включая Workers AI, который отклоняет ответы в формате JSON Schema при потоковых запросах. Когда Workflow возобновляется, полезная нагрузка снова проверяется по исходной схеме Zod, прежде чем будет возвращено типизированное значение.

Неподдерживаемые функции Zod, которые нельзя представить как JSON Schema, приводят к ошибке при создании шага промпта. Think не исправляет некорректный вывод модели незаметно. Если модель не выдаёт корректный final_answer вызова заявка переходит в конечное состояние ошибки, и step.prompt() выбрасывает исключение.

Замечания о поведении

Как это выполняется

Вызов выглядит как блокирующий шаг, но на деле не удерживает долгоживущий RPC к Durable Object открытым.

  1. step.do("<name>:submit", ...) создаёт или находит идемпотентную отправку Think.
  2. Think прогоняет отправленный ход через обычную очередь отправки.
  3. Когда отправка достигает completed, error, aborted, или skipped, Think записывает ожидающее уведомление workflow.
  4. Think очищает outbox уведомлений с помощью sendWorkflowEvent() и будильники Durable Object, пока доставка не завершится успешно.
  5. step.waitForEvent("<name>:wait", ...) возобновляет Workflow.
  6. step.prompt() проверяет структурированный вывод или выбрасывает типизированную ошибку.

Машиночитаемый вывод передаётся в payload ожидающего уведомления и в payload события Workflow. Think не хранит отдельный output_json столбец в реестре отправок и очищает payload уведомления после доставки. После доставки устойчивый результат принадлежит Workflow.

Идемпотентность

По умолчанию step.prompt() выводит ключ идемпотентности из идентичности Workflow и имени шага:

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

Для циклов передайте строку key чтобы различать повторные использования одного и того же имени шага:

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

Текст промпта не входит в выводимый ключ, но Think сохраняет метаданные workflow и отпечаток промпта и конфигурации для диагностики.

Тайм-ауты

Передайте timeout чтобы управлять тем, как долго Workflow ожидает терминальное событие. Если время ожидания истекает, step.prompt() по умолчанию отменяет отправку Think и выбрасывает ThinkPromptTimeoutError.

Задайте cancelOnTimeout: false когда вы намеренно хотите, чтобы отправка Think продолжалась после того, как Workflow перестанет ждать.

Граница с другими примитивами

Используйте getScheduledTasks() для повторяющихся отправок промптов или детерминированных запланированных обработчиков:

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

Используйте submitMessages() для устойчивых одноразовых ходов, когда вызывающая сторона может позже проверить статус отправки.

Используйте startFiber() для идемпотентных заданий Agent, принадлежащих приложению и требующих восстановления внутри Agent. Доставка уведомлений workflow в Think не использует fiber: вместо этого применяется приватный outbox, поскольку событие нужно хранить до тех пор, пока доставка не завершится успешно.

Используйте Workflows, если процесс включает несколько детерминированных шагов, долгие ожидания или подтверждение от человека.