← Создание агентов на Cloudflare / agents / communication-channels / chat

Автономные ответы

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

Обзор

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

Ключевые примитивы:

Примитив	Роль
`saveMessages`	Вставьте сообщение и запустите LLM: серверный эквивалент `sendMessage`
`submitMessages`	Устойчиво примите ход Think для асинхронного выполнения и проверьте его позже
`startFiber`	Устойчиво принимайте побочные эффекты, принадлежащие приложению, вокруг данного хода
`persistMessages`	Сохраняйте сообщения без вызова ответа: для незаметного добавления контекста
`onChatResponse`	Реагируйте на завершение любого ответа, включая те, которые вы не инициировали
`isServerStreaming`	Флаг на стороне клиента: `true` когда активен поток, инициированный сервером

`saveMessages` в сравнении с `persistMessages`

saveMessages сохраняет сообщения в SQLite и запускает onChatMessage для нового ответа LLM. Его можно дождаться через await: после этого LLM уже ответила, а сообщение сохранено.

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

`saveMessages` в сравнении с `submitMessages`

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

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

const submission = await this.submitMessages(
	[
		{
			id: crypto.randomUUID(),
			role: "user",
			parts: [
				{ type: "text", text: `Webhook event: ${JSON.stringify(payload)}` },
			],
		},
	],
	{ idempotencyKey: payload.id },
);

return Response.json({
	submissionId: submission.submissionId,
	status: submission.status,
	accepted: submission.accepted,
});

const submission = await this.submitMessages(
	[
		{
			id: crypto.randomUUID(),
			role: "user",
			parts: [
				{ type: "text", text: `Webhook event: ${JSON.stringify(payload)}` },
			],
		},
	],
	{ idempotencyKey: payload.id },
);

return Response.json({
	submissionId: submission.submissionId,
	status: submission.status,
	accepted: submission.accepted,
});

submitMessages() сначала сохраняет отложенную работу и добавляет сообщения в Session беседы только тогда, когда отправка начинает выполняться. Он принимает сериализуемые UIMessage[] значения, а не функциональную форму, поддерживаемую saveMessages((messages) => ...).

Используйте startFiber() вне Think, когда устойчивой единицей является охватывающая задача приложения, например однократный приём webhook, восстановление состояния провайдера, публикация видимого ответа и запись политики восстановления. submitMessages() отвечает за приём разговора в Think; управляемые fiber отвечают за внешние побочные эффекты вокруг этого хода.

Полное описание API Think см. в submitMessages().

Когда использовать `saveMessages` в сравнении с `onChatResponse`

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

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

`waitUntilStable`

Всегда вызывайте waitUntilStable() перед чтением this.messages или вызов saveMessages из колбэков расписания, вебхуков, обработчиков электронной почты или других точек входа, не связанных с чатом.

waitUntilStable() дожидается полной стабилизации разговора:

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

Возвращает true когда стабильно, или false если время ожидания истечёт до того, как будет обработано ожидающее взаимодействие. Если ничего не ожидает обработки, возврат происходит немедленно.

const stable = await this.waitUntilStable({ timeout: 30_000 });
if (!stable) {
	// The conversation is blocked on a user interaction or an in-flight
	// stream that did not complete within 30 seconds.
	console.warn("Conversation not stable, skipping server-driven message");
	return;
}
// Safe to read this.messages and call saveMessages.

const stable = await this.waitUntilStable({ timeout: 30_000 });
if (!stable) {
	// The conversation is blocked on a user interaction or an in-flight
	// stream that did not complete within 30 seconds.
	console.warn("Conversation not stable, skipping server-driven message");
	return;
}
// Safe to read this.messages and call saveMessages.

Без этой защиты вы рискуете прочитать устаревшие сообщения или столкнуться с наложением на ещё активный поток.

Паттерны триггеров

Расписание cron

Агент ежедневной сводки, который каждое утро подводит итоги активности. Cron-расписания по умолчанию идемпотентны, поэтому вызов schedule() в onStart безопасен: он не создаёт дубликатов при перезапусках Durable Object.

import { AIChatAgent } from "@cloudflare/ai-chat";

export class DigestAgent extends AIChatAgent {
	async onChatMessage() {
		// ... your LLM call
	}

	async onStart() {
		await this.schedule("0 9 * * *", "dailyDigest");
	}

	async dailyDigest() {
		const stable = await this.waitUntilStable({ timeout: 30_000 });
		if (!stable) {
			console.warn("Conversation not stable, skipping daily digest");
			return;
		}

		await this.saveMessages((messages) => [
			...messages,
			{
				id: crypto.randomUUID(),
				role: "user",
				parts: [
					{
						type: "text",
						text: "Summarize what happened since your last digest.",
					},
				],
				createdAt: new Date(),
			},
		]);
		// At this point the LLM has responded and the message is persisted.
	}
}

import { AIChatAgent } from "@cloudflare/ai-chat";

export class DigestAgent extends AIChatAgent {
	async onChatMessage() {
		// ... your LLM call
	}

	async onStart() {
		await this.schedule("0 9 * * *", "dailyDigest");
	}

	async dailyDigest() {
		const stable = await this.waitUntilStable({ timeout: 30_000 });
		if (!stable) {
			console.warn("Conversation not stable, skipping daily digest");
			return;
		}

		await this.saveMessages((messages) => [
			...messages,
			{
				id: crypto.randomUUID(),
				role: "user",
				parts: [
					{
						type: "text",
						text: "Summarize what happened since your last digest.",
					},
				],
				createdAt: new Date(),
			},
		]);
		// At this point the LLM has responded and the message is persisted.
	}
}

Функциональная форма saveMessages, saveMessages((messages) => [...]) : считывает последние сохранённые сообщения на момент выполнения. Это предотвращает использование устаревших исходных данных, когда в очередь одновременно попадает несколько вызовов (например, при быстром поступлении вебхуков). См. Планирование задач подробнее о schedule() и синтаксис cron.

Обработка очереди

Если вы сами управляете триггером, проще всего использовать простой цикл:

async processQueue() {
	for (const task of this.taskQueue) {
		const stable = await this.waitUntilStable({ timeout: 30_000 });
		if (!stable) {
			console.warn("Conversation not stable, stopping queue processing");
			break;
		}

		await this.saveMessages((messages) => [
			...messages,
			{
				id: crypto.randomUUID(),
				role: "user",
				parts: [{ type: "text", text: task }],
				createdAt: new Date(),
			},
		]);
		// LLM has responded. this.messages is updated. Next iteration.
	}
	this.taskQueue = [];
}

Специальные хуки не нужны: saveMessages возвращается после завершения полного turn.

Запускается по email

async onEmail(email: AgentEmail) {
	const stable = await this.waitUntilStable({ timeout: 30_000 });
	if (!stable) {
		console.warn("Conversation not stable, cannot process email");
		return;
	}

	const subject = email.headers.get("subject") ?? "(no subject)";
	const body = await new Response(email.raw).text();

	await this.saveMessages((messages) => [
		...messages,
		{
			id: crypto.randomUUID(),
			role: "user",
			parts: [
				{
					type: "text",
					text: `Email from ${email.from}: ${subject}\n\n${body}`,
				},
			],
			createdAt: new Date(),
		},
	]);
}

Запускается вебхуком

async onRequest(request: Request): Promise<Response> {
	const url = new URL(request.url);

	if (url.pathname.endsWith("/webhook") && request.method === "POST") {
		const stable = await this.waitUntilStable({ timeout: 30_000 });
		if (!stable) {
			return new Response("Agent is busy", { status: 503 });
		}

		const payload = await request.json();
		try {
			await this.saveMessages((messages) => [
				...messages,
				{
					id: crypto.randomUUID(),
					role: "user",
					parts: [
						{
							type: "text",
							text: `Webhook event: ${JSON.stringify(payload)}`,
						},
					],
					createdAt: new Date(),
				},
			]);
			return new Response("ok");
		} catch (error) {
			console.error("Failed to process webhook:", error);
			return new Response("Internal error", { status: 500 });
		}
	}

	return super.onRequest(request);
}

Если провайдер вебхука ожидает быстрый ответ, используйте submitMessages() вместо этого. Это даёт провайдеру устойчивое подтверждение получения и позволяет ему безопасно повторить попытку с тем же ключом идемпотентности:

async onRequest(request: Request): Promise<Response> {
	if (request.method !== "POST") return super.onRequest(request);

	const payload = await request.json<{ id: string }>();
	const submission = await this.submitMessages(
		[
			{
				id: crypto.randomUUID(),
				role: "user",
				parts: [
					{ type: "text", text: `Webhook event: ${JSON.stringify(payload)}` },
				],
			},
		],
		{ idempotencyKey: payload.id },
	);

	return Response.json({
		submissionId: submission.submissionId,
		accepted: submission.accepted,
		status: submission.status,
	});
}

Внедрение контекста без вызова ответа

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

async addBackgroundContext(data: string) {
	const stable = await this.waitUntilStable({ timeout: 30_000 });
	if (!stable) return;

	await this.persistMessages([
		...this.messages,
		{
			id: crypto.randomUUID(),
			role: "user",
			parts: [{ type: "text", text: `[Background context]: ${data}` }],
			createdAt: new Date(),
		},
	]);
	// Message is stored and broadcast to clients, but no LLM call happens.
}

Реакция на ответы, которые вы не инициировали

onChatResponse срабатывает после каждый завершённый ход: сообщения, инициированные пользователем, saveMessages вызовы и автоматические продолжения. Используйте это, когда нужно отслеживать ответы или реагировать на них независимо от того, как они были вызваны.

Рассылка состояния

import { AIChatAgent } from "@cloudflare/ai-chat";

export class ChatAgent extends AIChatAgent {
	async onChatMessage() {
		// ... your LLM call
	}

	async onChatResponse(result) {
		if (result.status === "completed") {
			this.broadcast(JSON.stringify({ streaming: false }));
		}
	}
}

import { AIChatAgent, type ChatResponseResult } from "@cloudflare/ai-chat";

export class ChatAgent extends AIChatAgent {
	async onChatMessage() {
		// ... your LLM call
	}

	protected async onChatResponse(result: ChatResponseResult) {
		if (result.status === "completed") {
			this.broadcast(JSON.stringify({ streaming: false }));
		}
	}
}

Аналитика

protected async onChatResponse(result: ChatResponseResult) {
	try {
		await fetch("https://analytics.example.com/event", {
			method: "POST",
			body: JSON.stringify({
				requestId: result.requestId,
				status: result.status,
				continuation: result.continuation,
			}),
		});
	} catch (error) {
		console.error("Analytics reporting failed:", error);
	}
}

Цепочка рассуждений

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

protected async onChatResponse(result: ChatResponseResult) {
	if (result.status !== "completed") return;

	const lastText = result.message.parts
		.filter((p) => p.type === "text")
		.map((p) => p.text)
		.join("");

	if (lastText.includes("[NEEDS_MORE_RESEARCH]")) {
		await this.saveMessages((messages) => [
			...messages,
			{
				id: crypto.randomUUID(),
				role: "user",
				parts: [{ type: "text", text: "Continue your research." }],
				createdAt: new Date(),
			},
		]);
	}
}

Когда saveMessages вызывается изнутри onChatResponse, внутренний ход доходит до завершения, и saveMessages возвращается. После текущего onChatResponse вызов возвращается, фреймворк вызывает onChatResponse снова для внутреннего ответа. Это продолжается, пока в очереди остаётся работа. Фреймворк никогда не вкладывает onChatResponse вызовы: результаты вытягиваются последовательно.

Реактивная обработка очереди

Если элементы очереди могут добавляться внешними событиями (сообщениями пользователей, вебхуками) в любой момент, onChatResponse позволяет опустошать очередь после каждого ответа независимо от того, кто её инициировал:

protected async onChatResponse(result: ChatResponseResult) {
	if (result.status === "completed" && this.taskQueue.length > 0) {
		const next = this.taskQueue.shift()!;
		await this.saveMessages((messages) => [
			...messages,
			{
				id: crypto.randomUUID(),
				role: "user",
				parts: [{ type: "text", text: next }],
				createdAt: new Date(),
			},
		]);
	}
}

`ChatResponseResult` поля

Поле	Тип	Описание
`message`	`UIMessage`	Финализированное сообщение ассистента
`requestId`	`string`	Уникальный ID для этого хода
`continuation`	`boolean`	`true` если это было автоматическое продолжение
`status`	`"completed" \| "error" \| "aborted"`	Как завершился ход
`error`	`string \| undefined`	Подробности ошибки, когда статус равен `"error"`

На стороне клиента: обнаружение потоков, инициированных сервером

Когда сервер запускает поток через saveMessages, из AI SDK status остаётся "ready" потому что клиент не инициировал запрос. useAgentChat хук предоставляет для этого два дополнительных флага:

Флаг	Что отслеживается
`status`	Жизненный цикл AI SDK: `"submitted"`, `"streaming"`, `"ready"`, `"error"` : только для запросов, инициированных клиентом
`isServerStreaming`	`true` когда активен поток, инициированный сервером
`isStreaming`	`true` когда активен поток на стороне клиента или сервера: используйте это для универсального индикатора

Используйте isStreaming для большинства задач UI (отключение кнопки отправки, отображение индикатора загрузки). Используйте isServerStreaming только если вам нужно различать потоки, инициированные пользователем и сервером (например, чтобы показать другой индикатор вроде "Agent работает в фоновом режиме...").

import { useAgent } from "agents/react";
import { useAgentChat } from "@cloudflare/ai-chat/react";

function Chat() {
	const agent = useAgent({ agent: "ChatAgent" });
	const { messages, sendMessage, isStreaming, isServerStreaming } =
		useAgentChat({ agent });

	return (
		<div>
			{messages.map((m) => (
				<div key={m.id}>{/* render message */}</div>
			))}

			{isServerStreaming && <div>Agent is working in the background...</div>}
			{!isServerStreaming && isStreaming && <div>Agent is responding...</div>}

			<form
				onSubmit={(e) => {
					e.preventDefault();
					const input = e.currentTarget.elements.namedItem(
						"input",
					) as HTMLInputElement;
					sendMessage({ text: input.value });
					input.value = "";
				}}
			>
				<input name="input" placeholder="Type a message..." />
				<button type="submit" disabled={isStreaming}>
					Send
				</button>
			</form>
		</div>
	);
}

Когда ответ, инициированный сервером, приходит в момент бездействия пользователя, подключённые клиенты видят, как новые сообщения появляются в реальном времени. isStreaming флаг переходит из false к true к false по мере выполнения потока, поэтому элементы интерфейса, такие как кнопка отправки, автоматически отключаются и включаются обратно.

Взаимодействие с `messageConcurrency`

messageConcurrency параметр в AIChatAgent определяет, как ведут себя пересекающиеся отправки пользователя ("queue", "latest", "merge", "drop", "debounce"). Эта настройка применяется только к sendMessage() : сообщения, инициированные пользователем со стороны клиента.

saveMessages() всегда использует сериализованное (поставленное в очередь) поведение независимо от messageConcurrency параметр. Это означает, что сообщения, инициированные сервером, никогда не отбрасываются, не объединяются и не задерживаются: они всегда встают в очередь и выполняются по порядку.

Сочетание с другими примитивами Agent

Примитив	Как комбинировать
`schedule()`	Запланируйте callback, который вызывает `saveMessages` : см. пример с cron выше
`queue()`	Поставьте в очередь метод, который вызывает `saveMessages` для отложенной обработки
`startFiber()`	Устойчиво примите и проверьте работу, принадлежащую приложению, вокруг хода сообщения
`runWorkflow()`	Запустите Workflow; используйте `AgentWorkflow.agent` RPC для вызова метода, который запускает `saveMessages` или `submitMessages`
`onEmail()`	Преобразуйте содержимое письма в сообщение чата и вызовите `saveMessages`
`onRequest()`	Обрабатывайте вебхуки и вызывайте `saveMessages` или `submitMessages`
`this.broadcast()`	Рассылает пользовательское состояние из `onChatResponse`

Отмена хода, управляемого сервером

Передайте AbortSignal когда один и тот же Durable Object запускает ход и управляет им:

const controller = new AbortController();
const result = await this.saveMessages(
	[
		{
			id: crypto.randomUUID(),
			role: "user",
			parts: [{ type: "text", text: "Run the long analysis." }],
		},
	],
	{ signal: controller.signal },
);

if (result.status === "aborted") {
	// Partial chunks already streamed are persisted.
}

const controller = new AbortController();
const result = await this.saveMessages(
	[
		{
			id: crypto.randomUUID(),
			role: "user",
			parts: [{ type: "text", text: "Run the long analysis." }],
		},
	],
	{ signal: controller.signal },
);

if (result.status === "aborted") {
	// Partial chunks already streamed are persisted.
}

continueLastTurn() принимает такой же options.signal аргумент. AbortSignal объекты не могут пересекать границы RPC Durable Object, а сигнал существует только в памяти. Если Durable Object переходит в спящий режим в середине хода, устойчивое восстановление обычно продолжается уже без исходного сигнала. Для прерываний до начала потоковой передачи восстановление может повторить последнее сообщение пользователя, оставшееся без ответа. Прерывание, вызванное после перезапуска, не влияет на восстановленный ход.

Сохраняйте намерение отмены, если отмена должна пережить перезапуск. Читайте это состояние в onChatRecovery() и вернуть { continue: false } чтобы предотвратить повторный вызов модели.

Используйте cancelSubmission(submissionId) для устойчивой отмены, если работа была принята с submitMessages() или когда отмена должна пересекать границы RPC между Worker и Durable Object.

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

Важные замечания

saveMessages поддерживает await. После возврата LLM уже ответила, и сообщение сохранено. Используйте это, когда вы сами управляете триггером.
Используйте функциональную форму saveMessages. saveMessages((messages) => [...messages, newMsg]) считывает последние сохранённые сообщения в момент выполнения, избегая устаревших базовых состояний, когда в очереди накапливается несколько вызовов.
persistMessages не вызывает ответ. Используйте это, чтобы незаметно добавлять контекст или системные сообщения.
onChatResponse предназначен для реакции на ходы, которые вы сами не инициировали. Используйте это для сообщений, инициированных пользователем, автопродолжений или любого хода, в котором вы не вызывали saveMessages самостоятельно.
onChatResponse не вкладывается. Когда saveMessages вызывается изнутри onChatResponse, внутренний ход завершается, и onChatResponse срабатывает снова последовательно, а не рекурсивно.
Сообщения сохраняются перед onChatResponse срабатывает. Если Durable Object вытесняется во время выполнения hook, разговор остаётся в безопасности в SQLite: теряется только сам callback hook.
waitUntilStable() перед внедрением. Всегда вызывайте это из callback-ов расписания, вебхуков или других точек входа, не связанных с чатом, чтобы избежать пересечения с активным потоком или ожидающим взаимодействием с инструментом.
Клиент видит завершённый ответ раньше, чем onChatResponse выполняется. Серверный хук не задерживает клиента.
messageConcurrency не влияет на saveMessages. Сообщения, инициированные сервером, всегда становятся в очередь и выполняются по порядку.

Автономные ответы

Обзор

`saveMessages` в сравнении с `persistMessages`

`saveMessages` в сравнении с `submitMessages`

Когда использовать `saveMessages` в сравнении с `onChatResponse`

`waitUntilStable`

Паттерны триггеров

Расписание cron

Обработка очереди

Запускается по email

Запускается вебхуком

Внедрение контекста без вызова ответа

Реакция на ответы, которые вы не инициировали

Рассылка состояния

Аналитика

Цепочка рассуждений

Реактивная обработка очереди

`ChatResponseResult` поля

На стороне клиента: обнаружение потоков, инициированных сервером

Взаимодействие с `messageConcurrency`

Сочетание с другими примитивами Agent

Отмена хода, управляемого сервером

Важные замечания

Следующие шаги

Чат-агенты

Планирование задач

Webhooks

Маршрутизация электронной почты

Автономные ответы

Обзор

saveMessages в сравнении с persistMessages

saveMessages в сравнении с submitMessages

Когда использовать saveMessages в сравнении с onChatResponse

waitUntilStable

Паттерны триггеров

Расписание cron

Обработка очереди

Запускается по email

Запускается вебхуком

Внедрение контекста без вызова ответа

Реакция на ответы, которые вы не инициировали

Рассылка состояния

Аналитика

Цепочка рассуждений

Реактивная обработка очереди

ChatResponseResult поля

На стороне клиента: обнаружение потоков, инициированных сервером

Взаимодействие с messageConcurrency

Сочетание с другими примитивами Agent

Отмена хода, управляемого сервером

Важные замечания

Следующие шаги

Чат-агенты

Планирование задач

Webhooks

Маршрутизация электронной почты

`saveMessages` в сравнении с `persistMessages`

`saveMessages` в сравнении с `submitMessages`

Когда использовать `saveMessages` в сравнении с `onChatResponse`

`waitUntilStable`

`ChatResponseResult` поля

Взаимодействие с `messageConcurrency`