← Vytvářejte agenty na Cloudflare / agents / communication-channels / chat

Autonomní odpovědi

Odesílejte zprávy a spouštějte odpovědi LLM ze serveru bez zásahu člověka. Použijte to pro plánované navazující zprávy, zpracování front, odpovědi vyvolané e-mailem a autonomní pracovní postupy agentů.

V typickém průběhu chatu uživatel odešle zprávu a agent odpoví. Agenti ale často potřebují jednat sami od sebe: spustí se naplánovaná připomínka, přijde webhook, dokončí se workflow, nebo se agent po prozkoumání vlastní odpovědi rozhodne pokračovat.

Klíčová primitiva:

saveMessages ukládá zprávy do SQLite a spouští onChatMessage pro novou odpověď LLM. Dá se na ni čekat (await), po jejím návratu už LLM odpověděl a zpráva je uložena.

persistMessages uchovává zprávy a odesílá je připojeným klientům, ale ne spustí tah modelu. Použijte ho, pokud chcete do konverzace vložit kontext (například systémovou zprávu nebo podkladová data), aniž byste spustili odpověď.

Použijte saveMessages() když volající může počkat na dokončení kola modelu.

Použijte submitMessages() s Think, když volající potřebuje rychlé trvalé potvrzení o přijetí, idempotentní opakování a pozdější kontrolu stavu. Hodí se to pro handlery webhooků, volající RPC a nadřazené Workery s přísnými časovými limity:

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() nejprve uloží čekající práci a zprávy do konverzačního Session přidá až ve chvíli, kdy se odeslání začne provádět. Přijímá serializovatelné UIMessage[] hodnoty, nikoli tvar funkce podporovaný saveMessages((messages) => ...).

Použijte startFiber() mimo Think, když je trvalou jednotkou okolní úloha aplikace, například jednorázové přijetí webhooku, obnovení stavu poskytovatele, odeslání viditelné odpovědi a zaznamenání zásady zotavení. submitMessages() řídí přijímání konverzace pro Think; spravovaná vlákna řídí externí vedlejší efekty kolem daného kola.

Úplné API Think najdete v submitMessages().

Použijte saveMessages pokud máte spouštěč pod kontrolou : callbacky plánovače, webhooky, handlery e-mailů nebo jakoukoli jinou metodu, kde sami rozhodujete, kdy zprávu vložit.

Použijte onChatResponse když potřebujete reagovat na odpovědi, které jste nevyvolali : zprávy iniciované uživatelem, automatická pokračování po schválení nástrojů nebo jakékoli kolo, které framework spustil vaším jménem.

Vždy zavolejte waitUntilStable() před čtením this.messages nebo volání saveMessages z callbacků plánovače, webhooků, handlerů e-mailů nebo jiných vstupních bodů mimo chat.

waitUntilStable() počká, dokud konverzace zcela nestabilizuje:

• Aktuálně neprobíhá žádný stream LLM
• Žádné čekající interakce klient-nástroj (výsledky nástrojů nebo schválení, která uživatel zatím neposkytl)
• Žádná zařazená pokračovací kola

Vrací true jakmile je stabilní, nebo false pokud časový limit vyprší dříve, než se nevyřízená interakce dokončí. Pokud nic nečeká na vyřízení, vrátí se okamžitě.

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.

Bez této pojistky riskujete čtení zastaralých zpráv nebo překryv s právě probíhajícím streamem.

Agent pro denní přehled, který každé ráno shrnuje aktivitu. Plány cron jsou ve výchozím nastavení idempotentní, takže volání schedule() v onStart je bezpečné: nevytváří duplicity napříč restarty 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.
	}
}

Funkční podoba saveMessages, saveMessages((messages) => [...]) : čte nejnovější uložené zprávy v okamžiku spuštění. Tím se předchází zastaralým výchozím stavům, když se do fronty nahromadí více volání najednou (například při rychlém příchodu webhooků). Viz Plánování úloh pro více informací o schedule() a syntaxe cron.

Pokud spouštěč ovládáte sami, je nejsrozumitelnějším vzorem jednoduchá smyčka:

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 = [];
}

Nejsou potřeba žádné speciální hooky: saveMessages se vrátí až po dokončení celého kola.

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

Pokud poskytovatel webhooku očekává rychlou odpověď, použijte submitMessages() místo toho. Poskytovateli to dává trvalé potvrzení a umožňuje mu bezpečně opakovat pokus se stejným idempotentním klíčem:

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

Použijte persistMessages pro přidání zpráv, které LLM uvidí při dalším tahu, aniž by se tah spouštěl hned:

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 se spustí po každý dokončené kolo: zprávy iniciované uživatelem, saveMessages volání a automatická pokračování. Použijte ho, když potřebujete sledovat odpovědi nebo na ně reagovat bez ohledu na to, jak byly vyvolány.

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

Agent může zkontrolovat vlastní odpověď a rozhodnout, zda pokračovat. To funguje i u zpráv iniciovaných uživatelem: nelze předvídat, na co se uživatel zeptá, ale lze reagovat na to, co agent řekl.

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(),
			},
		]);
	}
}

Když saveMessages se volá zevnitř onChatResponse, vnitřní kolo proběhne až do konce a saveMessages vrátí. Po dokončení aktuálního onChatResponse volání vrátí výsledek, framework spustí onChatResponse znovu pro vnitřní odpověď. Takto to pokračuje, dokud ve frontě nezbyde žádná další práce. Framework nikdy nezanořuje onChatResponse volání: výsledky se odčerpávají postupně.

Když mohou být položky fronty kdykoli přidány externími událostmi (zprávy uživatelů, webhooky), onChatResponse umožňuje vyprázdnit frontu po každé odpovědi bez ohledu na to, kdo ji vyvolal:

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(),
			},
		]);
	}
}

Když server spustí stream přes saveMessages, z AI SDK status zůstává "ready" protože požadavek nezahájil klient. useAgentChat hook pro tento účel poskytuje dva další příznaky:

Použijte isStreaming pro většinu záležitostí UI (deaktivaci tlačítka odeslat, zobrazení indikátoru načítání). Použijte isServerStreaming pouze tehdy, když potřebujete rozlišit streamy iniciované uživatelem a serverem (například pro zobrazení jiného indikátoru, jako je "Agent pracuje na pozadí...").

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

Když serverem iniciovaná odpověď dorazí ve chvíli, kdy je uživatel neaktivní, připojení klienti uvidí nové zprávy v reálném čase. isStreaming příznak přechází z false na true na false v průběhu streamování, takže se prvky uživatelského rozhraní, jako tlačítko pro odeslání, automaticky deaktivují a znovu aktivují.

messageConcurrency nastavení na AIChatAgent řídí, jak se chovají překrývající se odeslání uživatele ("queue", "latest", "merge", "drop", "debounce"). Toto nastavení platí pouze pro sendMessage() : zprávy iniciované uživatelem z klienta.

saveMessages() vždy používá serializované (frontové) chování bez ohledu na messageConcurrency nastavení. Zprávy iniciované serverem se tak nikdy nezahodí, neslučují ani nedebouncují: vždy se zařadí do fronty a provedou v pořadí.

Předejte AbortSignal když tentýž Durable Object kolo spouští a řídí:

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() přijímá stejné options.signal argument. AbortSignal objekty nemohou překročit hranice RPC Durable Object a signál existuje pouze v paměti. Pokud Durable Object v polovině tahu hibernuje, trvalé zotavení obvykle pokračuje bez původního signálu. U přerušení před zahájením streamu může zotavení zopakovat poslední nezodpovězenou zprávu uživatele. Přerušení vyvolané po restartu nemá na obnovený tah žádný vliv.

Uložte záměr zrušení, pokud musí zrušení přežít restart. Tento stav si přečtěte v onChatRecovery() a vrátí { continue: false } k zamezení dalšího volání modelu.

Použijte cancelSubmission(submissionId) pro trvalé zrušení, pokud byla práce přijata pomocí submitMessages() nebo když zrušení musí překročit hranice RPC mezi Workerem a Durable Object.

Použijte cancelFiber(fiberId) když byla trvalá jednotka přijata s startFiber() a zrušení by se mělo týkat obklopující aplikační úlohy, nikoli kola typu Think.

• saveMessages lze použít s await. Po jeho návratu už LLM odpověděl a zpráva je uložena. Použijte, pokud máte spouštěč pod kontrolou.
• Použijte formu funkce saveMessages. saveMessages((messages) => [...messages, newMsg]) čte nejnovější uložené zprávy v okamžiku spuštění, čímž se vyhne zastaralým výchozím stavům, když se ve frontě nahromadí více volání.
• persistMessages nevyvolá odpověď. Použijte ho k tichému vložení kontextu nebo systémových zpráv.
• onChatResponse slouží k reakci na kola, která jste sami nezahájili. Použijte ho pro zprávy iniciované uživatelem, automatická pokračování nebo jakýkoli tah, ve kterém jste nevolali saveMessages sami.
• onChatResponse se nevnořuje. Když saveMessages se volá zevnitř onChatResponse, vnitřní kolo se dokončí a onChatResponse se spustí znovu sekvenčně, ne rekurzivně.
• Zprávy jsou uloženy před onChatResponse se spustí. Pokud se Durable Object uvolní z paměti během hooku, konverzace zůstává v bezpečí v SQLite, ztratí se pouze callback hooku.
• waitUntilStable() před vložením. Volejte to vždy z callbacků plánovače, webhooků nebo jiných vstupních bodů mimo chat, abyste se vyhnuli překrytí s probíhajícím streamem nebo nevyřízenou interakcí s nástrojem.
• Klient vidí dokončenou odpověď dříve, než onChatResponse se spustí. Hook na straně serveru klienta nezdržuje.
• messageConcurrency neovlivňuje saveMessages. Zprávy iniciované serverem se vždy řadí do fronty a provádějí v pořadí, v jakém přišly.