跳转到内容
搜索文档
↵选择
↑↓导航
ESC关闭

可观测性

最后更新 查看 MarkdownAgent 设置

Agent 会为每项重要操作发出结构化事件——RPC 调用、状态变更、调度执行、Workflow 转换、MCP 连接等。这些事件发布到 diagnostics channel，默认静默（无人订阅时零开销）。

事件结构

每个事件包含以下字段：

{
  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——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 中的 subscribe() 提供对特定通道上事件的类型安全访问：

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 收窄为该通道流经的事件类型。

类型化辅助函数使用 camelCase 键，因此 agent-tool 恢复为 subscribe("agentTool", ...)。原始 diagnostics channel 订阅者应使用发出的通道名 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（生产环境）

生产环境中，所有 diagnostics channel 消息自动转发到 Tail Workers。Agent 本身无需订阅代码——挂载 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);
			}
		}
	},
};

这样可在生产环境获得结构化、可过滤的可观测性，且 Agent 热路径零开销。

自定义可观测性

可通过提供自定义 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 事件

类型 Payload 触发时机
rpc { method, streaming? } 调用 @callable 方法时
rpc:error { method, error } @callable 方法抛出异常时

状态事件

类型 Payload 触发时机
state:update {} 调用 setState() 时

消息、工具与提交事件

这些事件跟踪聊天消息生命周期、客户端工具交互与 Think 的持久提交。

类型 Payload 触发时机
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 提交失败

聊天恢复事件

类型 Payload 触发时机
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 } 框架开始恢复尝试
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 } 不活动看门狗触发——在 chatStreamStallTimeoutMs 内无流分块。启用 chatRecovery 时，轮次会路由到恢复

recoveryKind 为 "retry" 表示恢复重放未应答的用户轮次；为 "continue" 表示继续部分 assistant 轮次。

聊天上下文事件

类型 Payload 触发时机
chat:context:compacted { reason, shortened, requestId?, attempt? } Think 压缩会话以处理上下文窗口溢出。reason 为 "proactive"（步骤前 contextOverflow.proactive 防护触发）或 "reactive"（溢出后 contextOverflow.reactive 触发）。shortened 表示压缩是否实际缩短历史——false 表示重试仍会溢出。见 上下文窗口溢出恢复。

转录事件

类型 Payload 触发时机
chat:transcript:repaired { requestId?, removedToolCalls, normalizedInputs, toolCallIds? } Think 在发送给提供商前修复已持久化转录。removedToolCalls 统计已修复的孤立工具调用；normalizedInputs 统计已修复的字符串化或缺失工具输入

Fiber 事件

类型 Payload 触发时机
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 } 恢复钩子启动
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? } 恢复钩子失败

Agent-tool 恢复事件

类型 Payload 触发时机
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 } 父恢复意外失败

调度与队列事件

类型 Payload 触发时机
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 } 已入队回调耗尽重试后失败

生命周期事件

类型 Payload 触发时机
connect { connectionId } 建立 WebSocket 连接
disconnect { connectionId, code, reason } WebSocket 连接关闭
destroy {} agent 被销毁

Workflow 事件

类型 Payload 触发时机
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 事件

类型 Payload 触发时机
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 能力发现成功或失败

邮件事件

类型 Payload 触发时机
email:receive { from, to, subject? } 收到邮件
email:reply { from, to, subject? } 发送回复邮件
email:send { from, to, subject? } 发送邮件

后续步骤

配置

wrangler.jsonc 配置与部署。

Tail Workers

将 diagnostics channel 事件转发到 Tail Worker 以进行生产监控。

这篇文档对您有帮助吗？