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

Использование Agents с Workflows

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

У Agents и Workflows взаимно дополняющие сильные стороны:

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

Используйте только Agents для:

• Приложения для чата и обмена сообщениями
• Быстрые вызовы API и ответы на них
• Функции совместной работы в реальном времени
• Задачи короче 30 секунд
• Один устойчивый ход чата Think с submitMessages()

Используйте Agents вместе с Workflows для:

• Пайплайны обработки данных
• Генерация отчётов
• Потоки подтверждения с участием человека
• Задачи, требующие гарантированной доставки
• Многошаговые операции с требованиями повторных попыток

Используйте только Workflows для:

• Фоновые задачи с подтверждением пользователя или без него
• Запланированная синхронизация данных
• Событийно-ориентированные конвейеры обработки

AgentWorkflow класс (импортированный из agents/workflows) обеспечивает двустороннюю связь между Workflow и их исходным Agent.

Workflows могут взаимодействовать с Agents несколькими способами:

• RPC-вызовы: Вызывайте методы Agent напрямую с полной типобезопасностью через this.agent
• Отчёты о прогрессе: Отправляйте обновления о ходе выполнения через this.reportProgress() которые запускают колбэки Agent
• Обновления состояния: Изменяйте состояние Agent через step.updateAgentState() или step.mergeAgentState(), которая рассылается подключённым клиентам
• Клиентские рассылки: Отправляйте сообщения всем клиентам WebSocket через 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 могут взаимодействовать с выполняющимися Workflows следующими способами:

• Запуск workflow: Запускайте новые экземпляры workflow с помощью runWorkflow()
• Отправка событий: Отправляйте события с помощью sendWorkflowEvent()
• Подтверждение/отклонение: Отвечайте на запросы на одобрение с помощью approveWorkflow() / rejectWorkflow()
• Управление Workflow: Приостанавливайте, возобновляйте, завершайте или перезапускайте workflow
• Запросы статуса: Проверяйте прогресс workflow с помощью getWorkflow() / getWorkflows()

Понимание durability является ключевым для эффективного использования workflow:

Эти операции лёгкие и подходят для частых обновлений, но могут выполняться несколько раз, если workflow повторяет попытку:

• this.reportProgress() : отчёты о ходе выполнения
• this.broadcastToClients() : рассылки через WebSocket
• Прямые вызовы RPC к this.agent

Эти операции используют step параметр и гарантированно выполняются ровно один раз:

• step.do() : выполнение устойчивых шагов
• step.reportComplete() / step.reportError() : Отчетность о завершении
• step.sendEvent() : Пользовательские события
• step.updateAgentState() / step.mergeAgentState() : синхронизация состояния

Workflows обеспечивают устойчивость благодаря пошаговому выполнению:

1. Завершение шага необратимо : после завершения шага он не выполняется повторно, даже если workflow перезапускается
2. Автоматические повторные попытки : неудачные шаги повторяются с настраиваемым интервалом отката
3. Сохранение событий : workflows могут ожидать события до одного года
4. Восстановление состояния : состояние workflow сохраняется при сбоях инфраструктуры

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

Когда Agent запускает workflow с помощью runWorkflow(), workflow автоматически отслеживается во внутренней базе данных агента. Это позволяет:

• Запрос статуса workflow по ID, имени или метаданным с курсорной пагинацией
• Отслеживание прогресса через колбэки жизненного цикла (onWorkflowProgress, onWorkflowComplete, onWorkflowError)
• Управление Workflow: пауза, возобновление, завершение, перезапуск
• Очистка завершённых записей Workflows с помощью deleteWorkflow() / deleteWorkflows()
• Сопоставление workflow с пользователями или сессиями через метаданные

Agent получает запрос, запускает Workflow для ресурсоёмкой обработки и рассылает подключённым клиентам обновления о ходе выполнения по мере того, как Workflow выполняет каждый шаг.

// 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 подготавливает запрос и приостанавливается, ожидая подтверждения с помощью waitForApproval(), а Agent предоставляет пользовательский интерфейс, в котором пользователи могут подтвердить или отклонить через approveWorkflow() / rejectWorkflow(). Workflow возобновляется или выбрасывает WorkflowRejectedError на основе решения.

Workflow оборачивает обращения к внешнему API в устойчивые шаги с логикой повторов. Если API возвращает ошибку или workflow перезапускается, завершённые вызовы не повторяются, а неудачные повторяются автоматически.

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 обновляет состояние Agent на ключевых вехах с помощью step.updateAgentState() или step.mergeAgentState(). Эти изменения состояния рассылаются всем подключённым клиентам, поддерживая синхронизацию интерфейсов без опроса.