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

Устойчивое восстановление

Think всегда оборачивает ходы чата в восстанавливаемые fiber. Если Durable Object вытесняется в середине потоковой передачи, Think восстанавливает все буферизованные чанки. Он сохраняет частичный результат и планирует продолжение или повторную попытку.

Ходы WebSocket, дочерний агент chat() ходов, устойчивые submitMessages() выполнений, автоматических продолжений, saveMessages(), а также continueLastTurn() оборачиваются в runFiber.

Прерывание watchdog из-за зависания потока (chatStreamStallTimeoutMs) использует тот же ограниченный путь восстановления. SDK сохраняет устоявшийся частичный результат и планирует продолжение. Временное зависание восстанавливается автоматически. Поставщик, зависший навсегда, исчерпывает бюджет тем же путём, что и при развёртывании или вытеснении. SDK вызывает onExhausted, генерирует chat:recovery:exhausted, и показывает настроенный terminalMessage.

Настройте ограниченное восстановление, задав chatRecovery в объект:

export class MyAgent extends Think {
	chatRecovery = {
		maxAttempts: 6,
		stableTimeoutMs: 10_000,
		terminalMessage: "The assistant was interrupted and could not recover.",
		async onExhausted(ctx) {
			console.warn("Chat recovery exhausted", ctx.incidentId);
		},
	};

	getModel() {
		/* ... */
	}
}

export class MyAgent extends Think<Env> {
	override chatRecovery = {
		maxAttempts: 6,
		stableTimeoutMs: 10_000,
		terminalMessage: "The assistant was interrupted and could not recover.",
		async onExhausted(ctx) {
			console.warn("Chat recovery exhausted", ctx.incidentId);
		},
	};

	getModel() {
		/* ... */
	}
}

Те же события восстановления доступны через agents/observability на chat канал; исправления транскрипта отправляются в transcript канал. См. Диагностические каналы.

Переопределить onChatRecovery когда нужно восстановление, специфичное для провайдера, например получение сохранённого результата OpenAI Responses вместо нового вызова модели:

export class MyAgent extends Think {
	chatRecovery = {
		maxAttempts: 10,
		terminalMessage: "The assistant was interrupted. Please try again.",
	};

	async onChatRecovery(ctx) {
		console.log("Recovering chat turn", ctx.incidentId, ctx.attempt);
		return {}; // persist partial output and continue/retry when possible
	}
}

import type {
	ChatRecoveryContext,
	ChatRecoveryOptions,
} from "@cloudflare/think";

export class MyAgent extends Think<Env> {
	override chatRecovery = {
		maxAttempts: 10,
		terminalMessage: "The assistant was interrupted. Please try again.",
	};

	override async onChatRecovery(
		ctx: ChatRecoveryContext,
	): Promise<ChatRecoveryOptions> {
		console.log("Recovering chat turn", ctx.incidentId, ctx.attempt);
		return {}; // persist partial output and continue/retry when possible
	}
}

С persist: true, частичное сообщение сохраняется. При continue: true, Think вызывает continueLastTurn() после того как агент достигает стабильного состояния.

Для прерываний до начала потоковой передачи, когда ctx.streamId === "" и ctx.partialText === "" но последнее сохранённое сообщение по-прежнему остаётся сообщением пользователя без ответа, Think автоматически повторяет этот ход, если только continue это false.

onChatRecovery(ctx: ChatRecoveryContext): ChatRecoveryOptions {
	if (!ctx.streamId && !ctx.partialText) {
		console.log("Recovering a pre-stream interruption");
	}
	return {};
}

Используйте ctx.createdAt чтобы пропускать устаревшие восстановления. Например, если прерванный ход старше нескольких минут, верните { continue: false } чтобы частичный ответ сохранялся без запуска старого continuation.

Устойчивое ведение учёта остаётся активным, когда автоматическое продолжение неуместно. Верните { continue: false } чтобы предотвратить повторный вызов модели. Сведения об отмене, побочных эффектах и контроле затрат см. в Управление автоматическим продолжением.

Назначьте chatRecovery объект для настройки лимитов восстановления и терминального поведения. Выполняющийся ход переживает повторные прерывания, пока остаётся в пределах maxRecoveryWork лимит. Следующие параметры определяют, когда восстановление останавливается:

export class MyAgent extends Think {
	chatRecovery = {
		maxAttempts: 10,
		noProgressTimeoutMs: 5 * 60 * 1000,
		maxRecoveryWork: 1_000,
		terminalMessage: "The assistant was interrupted and could not recover.",
		// Consulted from the second recovery attempt onward. Return false to stop.
		// Called as `config.shouldKeepRecovering(ctx)`, so it is NOT bound to the
		// agent instance — track real token/cost spend in your own store keyed by
		// `ctx.recoveryRootRequestId`.
		async shouldKeepRecovering(ctx) {
			return (await getSpendForTurn(ctx.recoveryRootRequestId)) < MAX_SPEND;
		},
		async onExhausted(ctx) {
			console.warn("Recovery exhausted", ctx.incidentId, ctx.reason);
		},
	};
}

export class MyAgent extends Think<Env> {
	override chatRecovery = {
		maxAttempts: 10,
		noProgressTimeoutMs: 5 * 60 * 1000,
		maxRecoveryWork: 1_000,
		terminalMessage: "The assistant was interrupted and could not recover.",
		// Consulted from the second recovery attempt onward. Return false to stop.
		// Called as `config.shouldKeepRecovering(ctx)`, so it is NOT bound to the
		// agent instance — track real token/cost spend in your own store keyed by
		// `ctx.recoveryRootRequestId`.
		async shouldKeepRecovering(ctx) {
			return (await getSpendForTurn(ctx.recoveryRootRequestId)) < MAX_SPEND;
		},
		async onExhausted(ctx) {
			console.warn("Recovery exhausted", ctx.incidentId, ctx.reason);
		},
	};
}

ctx.reason для исчерпанного хука принимает одно из следующих значений: no_progress_timeout (застряло), max_attempts_exceeded (цикл alarm без прогресса), work_budget_exceeded (неконтролируемый), recovery_aborted (ваш shouldKeepRecovering возвращено false), out_of_memory (бюджет повторов при ограничении памяти), или stable_timeout (экстремальная флуктуация). См. Восстановление потока для полного общего справочника. Think и @cloudflare/ai-chat используют одинаковую конфигурацию восстановления.

Если ход прерывается на середине выполнения, в транскрипте может остаться вызов инструмента без итогового результата. Перед следующим обращением к провайдеру Think исправляет такие вызовы, чтобы модель не перезапустила их незаметно, а провайдер не отклонил транскрипт с AI_MissingToolResultsError. По умолчанию прерванный вызов переводится в результат инструмента с ошибкой, поэтому запись сохраняется, а преобразование по-прежнему получает результат инструмента для него.

Переопределить repairInterruptedToolPart чтобы настроить восстановленную форму. Типичный случай: инструмент, который разрешается на стороне клиента, например ask_user вопрос, у которого нет серверного execute и обычно получает ответ в следующем сообщении пользователя. Преобразование его в обычную текстовую часть позволяет модели воспринимать это как обычный разговор, а не как ошибку инструмента, и сохраняет вопрос дословно при компакции:

export class MyAgent extends Think {
	repairInterruptedToolPart(part) {
		const record = part;
		if (record.type === "tool-ask_user") {
			const input = record.input;
			if (input?.prompt) {
				return { type: "text", text: input.prompt };
			}
		}
		return super.repairInterruptedToolPart(part);
	}
}

import type { UIMessage } from "ai";

export class MyAgent extends Think<Env> {
	protected override repairInterruptedToolPart(
		part: UIMessage["parts"][number],
	): UIMessage["parts"][number] {
		const record = part as Record<string, unknown>;
		if (record.type === "tool-ask_user") {
			const input = record.input as { prompt?: string } | undefined;
			if (input?.prompt) {
				return { type: "text", text: input.prompt };
			}
		}
		return super.repairInterruptedToolPart(part);
	}
}

Это выполняется во время восстановления транскрипта, ещё до того, как восстановленный транскрипт будет сохранён и отправлен модели, поэтому преобразование формирует текущий ход, а не только следующий. Этот input уже нормализован до корректного объекта. Возвращённый tool part должен нести согласованный результат (output-available, output-error, или output-denied); также допустимо вернуть часть, не являющуюся инструментом, например текст.

Компактизация проверяется между ходами, compactAfter() выполняется после каждого appendMessage(). Но один длинный ход с большим количеством вызовов инструментов постепенно увеличивает промпт внутри одного streamText цикл и может превысить контекстное окно модели посреди хода, ещё до следующей проверки перед ходом. После этого поставщик отклоняет запрос ("prompt is too long", context_length_exceeded), иначе ход завершился бы окончательной ошибкой.

Think восстанавливается после этого с помощью двух опциональных слоёв, не зависящих от провайдера; оба настраиваются через contextOverflow свойство. Оба по умолчанию отключены, поэтому существующее поведение не меняется. Оба используют функцию компакции вашей сессии, поэтому для них требуется configureSession() с onCompaction() настроено. Оба требуют classifyChatError чтобы сообщить Think, какие ошибки считаются переполнением: в своём ядре Think не содержит логики сопоставления, зависящей от провайдера.

1. Реактивная подстраховка: contextOverflow.reactive. Когда ход завершается ошибкой, которую вы классифицируете как "context_overflow", Think отбрасывает усечённый частичный результат, запускает session.compact(), и перезапускает ход с уже компактизированной историей. Частичный результат не сохраняется: ход запускается заново с нуля, поэтому сохранение оборванного сообщения ассистента сделало бы его осиротевшим рядом с восстановленным ответом. Это ограничено contextOverflow.maxRetries (по умолчанию 1); если компактизация не может сократить историю или бюджет исчерпан, переполнение проявляется как финальная ошибка через onChatError с classification: "context_overflow" : никогда не зацикливается и не завершается молча.

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

export class MyAgent extends Think {
	contextOverflow = { reactive: true };

	// The bundled classifier covers the common providers (Anthropic, OpenAI,
	// Google, Bedrock, …). Assign it directly, or write your own.
	classifyChatError = defaultContextOverflowClassifier;
}

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

export class MyAgent extends Think<Env> {
	override contextOverflow = { reactive: true };

	// The bundled classifier covers the common providers (Anthropic, OpenAI,
	// Google, Bedrock, …). Assign it directly, or write your own.
	override classifyChatError = defaultContextOverflowClassifier;
}

2. Проактивная защита: contextOverflow.proactive. Предотвращает ошибку провайдера ещё до её появления. Перед каждым шагом Think считывает сообщённый моделью на предыдущем шаге usage.inputTokens (независимо от провайдера) и, если он превышает maxInputTokens * (headroom ?? 0.9), компактизирует историю на месте и передаёт заново компактизированную историю в следующий шаг. Если провайдер не указывает inputTokens, он переключается на usage.totalTokens (безопасное, слегка завышенное приближение: компактизация происходит немного раньше, чем пропуск порога). Компактизация выполняется не более proactive.maxCompactions раз за ход (по умолчанию 1), независимо от reactive maxRetries бюджет: поэтому история, которую невозможно сократить, не подвергается компактизации на каждом шаге.

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

export class MyAgent extends Think {
	contextOverflow = {
		reactive: true,
		// Compact mid-turn once a step approaches 90% of a 200K window.
		proactive: { maxInputTokens: 200_000 },
	};

	classifyChatError = defaultContextOverflowClassifier;
}

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

export class MyAgent extends Think<Env> {
	override contextOverflow = {
		reactive: true,
		// Compact mid-turn once a step approaches 90% of a 200K window.
		proactive: { maxInputTokens: 200_000 },
	};

	override classifyChatError = defaultContextOverflowClassifier;
}

Используйте один из уровней по отдельности или оба вместе: проактивная защита предотвращает большинство переполнений, а реактивный резервный механизм перехватывает те, что всё же проскальзывают (например, ход, который начинается уже с превышением бюджета, или единичный результат инструмента настолько большой, что компактизация не помогает, и тогда он корректно переводится в терминальное состояние). Оба уровня применяются на каждом пути входа хода (WebSocket, дочерний агент chat(), и программный saveMessages() / submitMessages()), и оба выдают chat:context:compacted событие диагностического канала.

Готовую к запуску демонстрацию с реальной моделью Workers AI см. в context-overflow-recovery пример ↗.

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

Возвращает true если какое-либо сообщение ассистента содержит невыполненные вызовы инструментов (инструменты без результатов или с ожидающими подтверждениями).

protected hasPendingInteraction(): boolean

Возвращает promise, который разрешается в true когда агент достигает стабильного состояния, или false если время ожидания превышено.

const stable = await this.waitUntilStable({ timeout: 30_000 });
if (stable) {
	await this.saveMessages([
		{
			id: crypto.randomUUID(),
			role: "user",
			parts: [{ type: "text", text: "Now that you are done, summarize." }],
		},
	]);
}

const stable = await this.waitUntilStable({ timeout: 30_000 });
if (stable) {
	await this.saveMessages([
		{
			id: crypto.randomUUID(),
			role: "user",
			parts: [{ type: "text", text: "Now that you are done, summarize." }],
		},
	]);
}