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

RPC дочернего агента и программные ходы

Think работает как агент верхнего уровня и одновременно как дочерний агент. При использовании в качестве дочернего агента chat() метод выполняет полный ход и передаёт события через колбэк.

Информацию об устойчивом приёме с идемпотентными повторами и последующей проверкой статуса см. в Программные отправки. Для восстановления после вытеснения см. Устойчивое восстановление.

чат

async chat(
	userMessage: string | UIMessage,
	callback: StreamCallback,
	options?: ChatOptions,
): Promise<void>

StreamCallback

Метод	Когда это срабатывает
`onStart(event)`	До начала работы; предоставляет ID запроса для отмены
`onEvent(json)`	Для каждого потокового чанка (сериализованного в JSON `UIMessageChunk`)
`onDone()`	После завершения хода и сохранения сообщения ассистента
`onError(message)`	При ошибке во время хода
`onInterrupted()`	Необязательно. Попытка была прервана, и запланированное продолжение (в более позднем isolate) отвечает за итоговый результат: не завершено, но и не терминальная ошибка. По умолчанию ничего не делает

onInterrupted важно для chat()-управляемый ход, который прерывается и восстанавливается: промис RPC разрешается корректно (изолят всё ещё жив), поэтому потребитель, ориентирующийся только на чистое разрешение, может ошибочно принять это за успех и финализировать любую частичную часть, которую он уже передал потоком. Рассматривайте это как «не завершено, но и не провалено: ответом занимается последующее продолжение»: держите канал открытым, показывайте состояние восстановления или переподключайтесь, вместо того чтобы финализировать частичный результат. Прерывание из-за развёртывания или вытеснения уничтожает изолят раньше, чем это успевает произойти (вместо этого вызывающая сторона видит обрыв транспорта); onInterrupted охватывает путь от зависания к восстановлению внутри isolate.

ChatOptions

Поле	Описание
`signal`	`AbortSignal` чтобы отменить ход в середине потока

Инструменты принадлежат дочернему агенту. Постоянные возможности определяйте с помощью дочернего агента getTools(), расширения, инструменты MCP или схемы клиентских инструментов. Устаревшие вызывающие стороны, которые передают options.tools к chat() получат предупреждение, а значение будет проигнорировано.

Пример: родительский агент вызывает дочерний

import { Think } from "@cloudflare/think";

export class ParentAgent extends Think {
	getModel() {
		/* ... */
	}

	async delegateToChild(task) {
		const child = await this.subAgent(ChildAgent, "child-1");

		const chunks = [];
		await child.chat(task, {
			onStart: (event) => {
				console.log("Child started:", event.requestId);
			},
			onEvent: (json) => {
				chunks.push(json);
			},
			onDone: () => {
				console.log("Child completed");
			},
			onError: (error) => {
				console.error("Child failed:", error);
			},
		});

		return chunks;
	}
}

export class ChildAgent extends Think {
	getModel() {
		/* ... */
	}

	getSystemPrompt() {
		return "You are a research assistant. Analyze data and report findings.";
	}
}

import { Think } from "@cloudflare/think";
import type { StreamCallback } from "@cloudflare/think";

export class ParentAgent extends Think<Env> {
	getModel() {
		/* ... */
	}

	async delegateToChild(task: string) {
		const child = await this.subAgent(ChildAgent, "child-1");

		const chunks: string[] = [];
		await child.chat(task, {
			onStart: (event) => {
				console.log("Child started:", event.requestId);
			},
			onEvent: (json) => {
				chunks.push(json);
			},
			onDone: () => {
				console.log("Child completed");
			},
			onError: (error) => {
				console.error("Child failed:", error);
			},
		});

		return chunks;
	}
}

export class ChildAgent extends Think<Env> {
	getModel() {
		/* ... */
	}

	getSystemPrompt() {
		return "You are a research assistant. Analyze data and report findings.";
	}
}

Отмена хода дочернего агента

Используйте onStart и cancelChat() для безопасной с точки зрения RPC отмены на границе дочернего агента:

let requestId;

const callback = {
	onStart(event) {
		requestId = event.requestId;
	},
	onEvent(json) {
		// Forward stream chunks.
	},
	onDone() {},
	onError(error) {
		console.error(error);
	},
};

const turn = child.chat("Long analysis task", callback);

// Later, from another RPC call or failure handler:
if (requestId) {
	await child.cancelChat(requestId, "client disconnected");
}

await turn;

let requestId: string | undefined;

const callback: StreamCallback = {
	onStart(event) {
		requestId = event.requestId;
	},
	onEvent(json) {
		// Forward stream chunks.
	},
	onDone() {},
	onError(error) {
		console.error(error);
	},
};

const turn = child.chat("Long analysis task", callback);

// Later, from another RPC call or failure handler:
if (requestId) {
	await child.cancelChat(requestId, "client disconnected");
}

await turn;

Если вызывающий и вызываемый объекты не разделены Workers RPC, можно также передать AbortSignal чтобы отменить в середине потока:

const controller = new AbortController();
setTimeout(() => controller.abort(), 30_000);

await child.chat("Long analysis task", callback, {
	signal: controller.signal,
});

const controller = new AbortController();
setTimeout(() => controller.abort(), 30_000);

await child.chat("Long analysis task", callback, {
	signal: controller.signal,
});

cancelChat(requestId, reason?) не выполняет никаких действий, если ход уже завершён или идентификатор запроса неизвестен. При прерывании частичное сообщение ассистента всё равно сохраняется.

saveMessages

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

async saveMessages(
	messages:
		| UIMessage[]
		| ((current: UIMessage[]) => UIMessage[] | Promise<UIMessage[]>),
	options?: SaveMessagesOptions,
): Promise<SaveMessagesResult>

Возвращает { requestId, status, error? } где status это "completed", "error", "skipped", или "aborted".

`status`	Когда
`"completed"`	Ход выполнен до конца.
`"error"`	Ход начался, но поток сообщил об ошибке. `error` содержит сообщение об ошибке потока, если оно доступно.
`"skipped"`	Ход признан недействительным на лету, например из-за `chat-clear`; сообщение пользователя сохранено, модель не запускалась.
`"aborted"`	Ход отменён до завершения через `options.signal` или `chat-request-cancel`. Частичные чанки ассистента всё равно сохраняются.

Передайте options.signal чтобы отменить программный ход из Durable Object, который его запускает. AbortSignal не может пересекать границы RPC Durable Object, а сигнал не сохраняется при гибернации.

Статические сообщения

await this.saveMessages([
	{
		id: crypto.randomUUID(),
		role: "user",
		parts: [{ type: "text", text: "Time for your daily summary." }],
	},
]);

await this.saveMessages([
	{
		id: crypto.randomUUID(),
		role: "user",
		parts: [{ type: "text", text: "Time for your daily summary." }],
	},
]);

Форма функции

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

await this.saveMessages((current) => [
	...current,
	{
		id: crypto.randomUUID(),
		role: "user",
		parts: [{ type: "text", text: "Continue your analysis." }],
	},
]);

await this.saveMessages((current) => [
	...current,
	{
		id: crypto.randomUUID(),
		role: "user",
		parts: [{ type: "text", text: "Continue your analysis." }],
	},
]);

Запланированные ответы

Запускайте повторяющийся ход с промптом с помощью getScheduledTasks():

export class MyAgent extends Think {
	getModel() {
		/* ... */
	}

	getScheduledTasks() {
		return {
			dailyReport: {
				schedule: "every day at 09:00",
				timezone: "UTC",
				prompt: "Generate the daily report.",
			},
		};
	}
}

export class MyAgent extends Think<Env> {
	getModel() {
		/* ... */
	}

	getScheduledTasks() {
		return {
			dailyReport: {
				schedule: "every day at 09:00",
				timezone: "UTC",
				prompt: "Generate the daily report.",
			},
		};
	}
}

Цепочка от onChatResponse

Запустите следующий ход после завершения текущего:

async onChatResponse(result: ChatResponseResult) {
	if (result.status === "completed" && this.needsFollowUp(result.message)) {
		await this.saveMessages([{
			id: crypto.randomUUID(),
			role: "user",
			parts: [{ type: "text", text: "Now summarize what you found." }],
		}]);
	}
}

continueLastTurn

Запустите ещё один вызов модели после последнего сообщения ассистента, не добавляя новое сообщение пользователя. Think сохраняет результат как новое сообщение ассистента с continuation: true; он не добавляет фрагменты к существующему сообщению ассистента.

protected async continueLastTurn(
	body?: Record<string, unknown>,
	options?: SaveMessagesOptions,
): Promise<SaveMessagesResult>

Возвращает { requestId, status: "skipped" } если последнее сообщение не является сообщением ассистента. Необязательный body параметр переопределяет сохранённое тело для этого продолжения. Передайте options.signal чтобы отменить продолжение во время его выполнения.

abortRequest и abortAllRequests

Отмените выполняющиеся ходы чата изнутри Durable Object:

protected abortRequest(requestId: string, reason?: unknown): void
protected abortAllRequests(): void

Используйте abortRequest() когда вам известен ID запроса. Используйте abortAllRequests() для узкоспециализированных вспомогательных функций, которые должны отменять текущий выполняющийся ход. Предпочтительнее использовать SaveMessagesOptions.signal для программных ходов, когда сигнал можно передать прямо в месте вызова.