INTEGRITY Документация

← Создание агентов на Cloudflare / agents / runtime / operations / observability

Диагностические каналы

Agents публикуют структурированные события в диагностические каналы для каждой значимой операции: вызовов RPC, изменений состояния, выполнения расписаний, переходов workflow, подключений MCP и прочего. Публикация не создаёт накладных расходов, если никто не слушает.

Структура события

У каждого события есть следующие поля:

{
  type: "rpc",                        // what happened
  agent: "MyAgent",                   // which agent class emitted it
  name: "user-123",                   // which agent instance (Durable Object name)
  payload: { method: "getWeather" },  // details
  timestamp: 1758005142787            // when (ms since epoch)
}

agent и name идентифицируют исходного агента: agent это имя класса, а name это имя экземпляра Durable Object.

Каналы

События маршрутизируются по именованным каналам в зависимости от их типа:

Канал Типы событий Описание
agents:state state:update События синхронизации состояния
agents:rpc rpc, rpc:error Вызовы RPC-методов и сбои
agents:message message:request, message:response, message:clear, message:cancel, message:error, tool:result, tool:approval, submission:create, submission:status, submission:error Жизненный цикл отправки сообщения чата, инструмента и Think
agents:chat chat:request:failed, chat:recovery:*, chat:stream:stalled, chat:context:compacted Жизненный цикл запроса чата, восстановления, зависания потока и компактизации контекста
agents:transcript chat:transcript:repaired События восстановления транскрипта
agents:fiber fiber:run:*, fiber:recovery:* Жизненный цикл устойчивого fiber
agents:agent_tool agent_tool:recovery:* Восстановление agent-tool у родителя и потомка
agents:schedule schedule:create, schedule:execute, schedule:cancel, schedule:retry, schedule:error, schedule:duplicate_warning, queue:create, queue:retry, queue:error Жизненный цикл запланированных и поставленных в очередь задач
agents:lifecycle connect, disconnect, destroy Подключение и отключение Agent
agents:workflow workflow:start, workflow:event, workflow:approved, workflow:rejected, workflow:terminated, workflow:paused, workflow:resumed, workflow:restarted Переходы состояний Workflow
agents:mcp mcp:client:preconnect, mcp:client:connect, mcp:client:authorize, mcp:client:discover Операции MCP-клиента
agents:email email:receive, email:reply, email:send Обработка электронной почты

Подписка на события

Типизированный помощник для подписки

subscribe() функцию из agents/observability предоставляет типобезопасный доступ к событиям на конкретном канале:

import { subscribe } from "agents/observability";

const unsub = subscribe("rpc", (event) => {
	if (event.type === "rpc") {
		console.log(`RPC call: ${event.payload.method}`);
	}
	if (event.type === "rpc:error") {
		console.error(
			`RPC failed: ${event.payload.method} — ${event.payload.error}`,
		);
	}
});

// Clean up when done
unsub();
import { subscribe } from "agents/observability";

const unsub = subscribe("rpc", (event) => {
	if (event.type === "rpc") {
		console.log(`RPC call: ${event.payload.method}`);
	}
	if (event.type === "rpc:error") {
		console.error(
			`RPC failed: ${event.payload.method} — ${event.payload.error}`,
		);
	}
});

// Clean up when done
unsub();

Колбэк полностью типизирован: event сужается только до типов событий, которые проходят через этот канал.

Типизированный helper использует ключи в camelCase, поэтому восстановление agent-tool subscribe("agentTool", ...). Подписчикам необработанного диагностического канала следует использовать имя канала, указанное в событии, agents:agent_tool.

Низкоуровневый diagnostics_channel

Вы также можете подписаться напрямую с помощью Node.js API:

import { subscribe } from "node:diagnostics_channel";

subscribe("agents:schedule", (event) => {
	console.log(event);
});
import { subscribe } from "node:diagnostics_channel";

subscribe("agents:schedule", (event) => {
	console.log(event);
});

Tail Workers (в продакшене)

В продакшене все сообщения диагностического канала автоматически пересылаются в Tail Workers. Код подписки в самом агенте не нужен: подключите Tail Worker и получайте доступ к событиям через event.diagnosticsChannelEvents:

export default {
	async tail(events) {
		for (const event of events) {
			for (const msg of event.diagnosticsChannelEvents) {
				// msg.channel is "agents:rpc", "agents:workflow", etc.
				// msg.message is the typed event payload
				console.log(msg.timestamp, msg.channel, msg.message);
			}
		}
	},
};
export default {
	async tail(events) {
		for (const event of events) {
			for (const msg of event.diagnosticsChannelEvents) {
				// msg.channel is "agents:rpc", "agents:workflow", etc.
				// msg.message is the typed event payload
				console.log(msg.timestamp, msg.channel, msg.message);
			}
		}
	},
};

Это даёт вам структурированную, фильтруемую наблюдаемость в production без какой-либо дополнительной нагрузки на hot path агента.

Пользовательская наблюдаемость

Вы можете переопределить реализацию по умолчанию, предоставив собственную Observability интерфейс:

import { Agent } from "agents";

const myObservability = {
	emit(event) {
		// Send to your logging service, filter events, etc.
		if (event.type === "rpc:error") {
			console.error(event.payload.method, event.payload.error);
		}
	},
};

class MyAgent extends Agent {
	observability = myObservability;
}
import { Agent } from "agents";
import type { Observability } from "agents/observability";

const myObservability: Observability = {
	emit(event) {
		// Send to your logging service, filter events, etc.
		if (event.type === "rpc:error") {
			console.error(event.payload.method, event.payload.error);
		}
	},
};

class MyAgent extends Agent {
	override observability = myObservability;
}

Задайте observability к undefined чтобы отключить отправку всех событий:

import { Agent } from "agents";

class MyAgent extends Agent {
	observability = undefined;
}
import { Agent } from "agents";

class MyAgent extends Agent {
	override observability = undefined;
}

Справочник событий

События RPC

Тип Полезная нагрузка Когда
rpc { method, streaming? } A @callable метод вызывается
rpc:error { method, error } A @callable метод выбрасывает исключение

События состояния

Тип Полезная нагрузка Когда
state:update {} setState() вызывается

События сообщений, инструментов и отправки

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

Тип Полезная нагрузка Когда
message:request {} Сообщение чата получено
message:response {} Поток ответа чата завершается
message:clear {} История чата очищается
message:cancel { requestId } Потоковый запрос отменяется
message:error { error } Поток чата завершается ошибкой
tool:result { toolCallId, toolName } Получен результат клиентского инструмента
tool:approval { toolCallId, approved } Вызов инструмента подтверждается или отклоняется
submission:create { submissionId } Отправка Think принимается
submission:status { submissionId, status } Статус отправки Think изменяется
submission:error { submissionId, error } Отправка Think завершается ошибкой

События восстановления чата

Тип Полезная нагрузка Когда
chat:request:failed { requestId?, stage, messagesPersisted?, error } Запрос чата Think завершается ошибкой на этапе разбора, сохранения, выполнения или потоковой передачи
chat:recovery:detected { incidentId, requestId, attempt, maxAttempts, recoveryKind } Прерванный fiber чата обнаруживается первым
chat:recovery:attempt { incidentId, requestId, attempt, maxAttempts, recoveryKind } Framework начинает попытку восстановления
chat:recovery:scheduled { incidentId, requestId, attempt, maxAttempts, recoveryKind } Планируется колбэк повтора или продолжения
chat:recovery:completed { incidentId, requestId, attempt, maxAttempts, recoveryKind } Восстановление успешно завершено
chat:recovery:skipped { incidentId, requestId, attempt, maxAttempts, recoveryKind, reason? } Восстановление было пропущено, потому что разговор изменился или стал невосстановимым
chat:recovery:failed { incidentId, requestId, attempt, maxAttempts, recoveryKind, reason? } Восстановление выполнилось, но завершилось неудачей
chat:recovery:exhausted { incidentId, requestId, attempt, maxAttempts, recoveryKind, reason } Восстановление превысило настроенный бюджет попыток
chat:stream:stalled { requestId, timeoutMs } Watchdog бездействия сработал, поскольку ни один фрагмент потока не поступил в течение chatStreamStallTimeoutMs. Ход направляется в устойчивое восстановление

recoveryKind это "retry" когда восстановление повторно воспроизводит ход пользователя, оставшийся без ответа, и "continue" когда он продолжает частичный ход ассистента.

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

Тип Полезная нагрузка Когда
chat:context:compacted { reason, shortened, requestId?, attempt? } Think компактизирует сессию, чтобы справиться с переполнением контекстного окна. reason это "proactive" ( contextOverflow.proactive guard, сработавший перед шагом) или "reactive" (contextOverflow.reactive запускается после переполнения). shortened состоит в том, привела ли компактизация к реальному сокращению истории: false означает, что повторная попытка снова приведёт к переполнению. См. Восстановление после переполнения контекстного окна.

События транскрипта

Тип Полезная нагрузка Когда
chat:transcript:repaired { requestId?, removedToolCalls, normalizedInputs, toolCallIds? } Перед отправкой провайдеру Think исправляет сохранённый транскрипт разговора. removedToolCalls считает исправленные осиротевшие вызовы инструментов; normalizedInputs считает исправленные преобразованные в строку или отсутствующие входные данные инструментов

События Fiber

Тип Полезная нагрузка Когда
fiber:run:started { fiberId, fiberName, managed? } Устойчивый fiber запускается
fiber:run:completed { fiberId, fiberName, managed?, elapsedMs? } Устойчивый fiber завершается
fiber:run:failed { fiberId, fiberName, managed?, error, elapsedMs? } Устойчивый fiber выбрасывает исключение
fiber:run:interrupted { fiberId, fiberName, managed?, recoveryReason, elapsedMs? } При запуске обнаруживается прерванный fiber
fiber:recovery:detected { fiberId, fiberName, managed?, recoveryReason, elapsedMs? } Восстановление обнаруживает прерванный fiber
fiber:recovery:attempt { fiberId, fiberName, managed?, recoveryReason } Recovery hook запускается
fiber:recovery:handled { fiberId, fiberName, managed?, recoveryReason, status, elapsedMs? } Обработка восстановления завершается
fiber:recovery:skipped { fiberId, fiberName, managed?, reason, elapsedMs? } Скан восстановления пропускает оставшуюся работу
fiber:recovery:failed { fiberId, fiberName, managed?, error, reason?, elapsedMs? } Recovery hook завершается ошибкой

События восстановления агентных инструментов

Тип Полезная нагрузка Когда
agent_tool:recovery:begin { runCount, totalTimeoutMs? } Восстановление родителя начинает сканирование устаревших запусков agent-tool
agent_tool:recovery:row { runId, agentType, status, reason?, elapsedMs? } Один устаревший запуск сверяется
agent_tool:recovery:deadline { runId, agentType, elapsedMs? } Общий срок восстановления истёк до проверки строки
agent_tool:recovery:complete { runCount, elapsedMs? } Восстановление родителя завершает сканирование строк
agent_tool:recovery:failed { error } Восстановление родителя неожиданно завершается ошибкой

Планирование и постановка событий в очередь

Тип Полезная нагрузка Когда
schedule:create { callback, id } Расписание создаётся
schedule:execute { callback, id } Запланированный колбэк запускается
schedule:cancel { callback, id } Расписание отменяется
schedule:retry { callback, id, attempt, maxAttempts } Запланированный колбэк повторяется
schedule:error { callback, id, error, attempts } Запланированный колбэк завершается ошибкой после всех повторов
schedule:duplicate_warning { callback } Неидемпотентное расписание может дублировать работу
queue:create { callback, id } Задача ставится в очередь
queue:retry { callback, id, attempt, maxAttempts } Поставленный в очередь колбэк повторяется
queue:error { callback, id, error, attempts } Поставленный в очередь колбэк завершается ошибкой после всех повторов

События жизненного цикла

Тип Полезная нагрузка Когда
connect { connectionId } Соединение WebSocket устанавливается
disconnect { connectionId, code, reason } Соединение WebSocket закрывается
destroy {} Агент уничтожается

События Workflow

Тип Полезная нагрузка Когда
workflow:start { workflowId, workflowName? } Экземпляр Workflow запущен
workflow:event { workflowId, eventType? } Событие отправляется в workflow
workflow:approved { workflowId, reason? } Workflow одобрен
workflow:rejected { workflowId, reason? } Workflow отклонён
workflow:terminated { workflowId, workflowName? } Workflow прекращён
workflow:paused { workflowId, workflowName? } Workflow приостановлен
workflow:resumed { workflowId, workflowName? } Workflow возобновлён
workflow:restarted { workflowId, workflowName? } Workflow перезапущен

MCP-события

Тип Полезная нагрузка Когда
mcp:client:preconnect { serverId } Перед подключением к MCP-серверу
mcp:client:connect { url, transport, state, error? } Попытка подключения MCP завершается успехом или неудачей
mcp:client:authorize { serverId, authUrl, clientId? } Начинается поток MCP OAuth
mcp:client:discover { url?, state?, error?, capability? } Успешное или неуспешное обнаружение возможностей MCP

События электронной почты

Тип Полезная нагрузка Когда
email:receive { from, to, subject? } Электронное письмо получено
email:reply { from, to, subject? } Отправляется ответное письмо
email:send { from, to, subject? } Электронное письмо отправлено

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

Трассировка

Отслеживайте вызовы модели, запуски инструментов и подтверждения с помощью трассировок Workers.

Конфигурация

wrangler.jsonc: настройка и развёртывание.

Tail Workers

Пересылайте события канала диагностики в Tail Worker для мониторинга в продакшене.