第 13 章 · Observability 可观测性
本章目标：区分 Flue 的两个事件面，学会用
observe()订阅运行时事件、读取 token 用量，并把遥测导出到 OpenTelemetry、Sentry 与 Braintrust。
13.1 两个事件面
Flue 把"agent 干了什么"暴露为类型化的运行时事件：模型回合、工具调用、结构化日志、压缩（compaction）、结算（settlement）。注意它与聊天 UI 读的消息流是两个不同的面：
text
① 运行时事件面（本章）：observe() 订阅，面向运维与调试
② 会话消息面：Routing + Flue Agent SDK，面向聊天界面的逐条消息typescript
// 面向运维的事件订阅：observe()
import { observe } from '@flue/runtime';
const stop = observe((event) => {
// 每一个事件都是可判别的联合类型，switch 处理
switch (event.type) {
case 'model_turn':
console.log('[model]', event.model, 'tokens:', event.usage);
break;
case 'tool_call':
console.log('[tool]', event.name, '耗时:', event.durationMs, 'ms');
break;
case 'log':
console.log('[log ]', event.level, event.message);
break;
}
});
// 不再需要时停止订阅
// stop();13.2 事件流里有什么
模型回合事件自带 token 用量与 provider 诊断信息，这是成本监控的原始数据源：
typescript
// 成本统计：聚合 model_turn 事件的 usage
import { observe } from '@flue/runtime';
const usageByDay = new Map<string, { input: number; output: number }>();
observe((event) => {
if (event.type !== 'model_turn') return;
const day = new Date().toISOString().slice(0, 10); // 按天聚合
const cur = usageByDay.get(day) ?? { input: 0, output: 0 };
usageByDay.set(day, {
input: cur.input + (event.usage?.inputTokens ?? 0),
output: cur.output + (event.usage?.outputTokens ?? 0),
});
});
// 每小时打印一次用量报表
setInterval(() => {
for (const [day, u] of usageByDay) {
console.log(day, 'input:', u.input, 'output:', u.output);
}
}, 60 * 60 * 1000);13.3 导出到 OpenTelemetry
@flue/opentelemetry 把运行时事件转换为 OTel trace/span，接入任意兼容后端（Jaeger、Grafana Tempo 等）：
bash
npm install @flue/opentelemetry @opentelemetry/sdk-node @opentelemetry/exporter-trace-otlp-httptypescript
// src/otel.ts
// OpenTelemetry 导出适配
import { flueOtel } from '@flue/opentelemetry';
import { NodeSDK } from '@opentelemetry/sdk-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
const sdk = new NodeSDK({
// Flue 提供的适配器把 agent 活动映射为 span
traceExporter: new OTLPTraceExporter({
url: process.env.OTEL_EXPORTER_OTLP_ENDPOINT, // 例如 http://localhost:4318/v1/traces
}),
});
// 注册 Flue 事件 → OTel 的桥接
flueOtel({ sdk });
sdk.start();13.4 Sentry 与 Braintrust
两类现成集成覆盖"报错追踪"与"LLM 评估"两种需求：
typescript
// Sentry：把 agent 异常与上下文送进错误追踪
import * as Sentry from '@sentry/node';
import { observe } from '@flue/runtime';
Sentry.init({ dsn: process.env.SENTRY_DSN });
observe((event) => {
// 工具失败或会话错误时上报，附带会话 id 便于排查
if (event.type === 'error') {
Sentry.captureException(event.error, {
tags: { sessionId: event.sessionId, agent: event.agent },
});
}
});typescript
// Braintrust：把模型回合导出为评估样本
import { braintrustExporter } from '@flue/runtime/observers'; // 以文档导出器为例
import { observe } from '@flue/runtime';
observe(braintrustExporter({
project: 'flue-triage', // Braintrust 项目名
apiKey: process.env.BRAINTRUST_API_KEY!,
}));13.5 自定义 observer 与生产监控
任何 observe() 回调都是一个 observer——写一个写文件的审计日志只需几行：
typescript
// 自定义 observer：审计日志落盘（JSON Lines 格式）
import { observe } from '@flue/runtime';
import { appendFileSync } from 'node:fs';
observe((event) => {
// 每条事件一行 JSON，便于 jq/grep 分析
appendFileSync(
'audit.jsonl',
JSON.stringify({ ts: Date.now(), ...event }) + '\n',
);
});Cloudflare 平台可观测性
部署在 Workers 上时，agent 活动还会自然出现在 Cloudflare 平台自身的观测面板（logs/metrics）中——无需额外接线即可看到基础指标。
生产监控的最小清单：
- 错误率：error 事件 / 总事件，接 Sentry 告警；
- 成本：按天聚合的 token 用量（13.2 的报表）；
- 延迟：model_turn 的首字延迟与 tool_call 耗时分位数。
13.6 本章小结
- 两个事件面：
observe()的运行时事件面（运维）与 Routing/SDK 的会话消息面（聊天 UI）； - 事件是类型化联合：model_turn（含 usage）、tool_call、log、error、settlement 等；
@flue/opentelemetry一行桥接任意 OTel 后端；Sentry 管报错、Braintrust 管评估；- 自定义 observer 就是 observe 回调，JSONL 审计日志是最简单的落地方式。
🧪 随堂测验
点击你认为正确的选项。答错时会展示正确答案与原因解析。
1. Flue 的运行时事件面与会话消息面的区别是？
2. 想统计每个会话的 token 成本，应该从哪类事件取数据？
3. @flue/opentelemetry 包的作用是？
4. 写一个把事件追加到 audit.jsonl 的函数，本质上是在使用什么机制？
🛠️ 动手实践
- 写一个 observer 统计每个 agent 的平均回合数与 token 成本，输出 Top 5 成本会话。
- 本地起一个 Jaeger（docker run jaegertracing/all-in-one），接入 @flue/opentelemetry，截图一次会话的 trace 瀑布图。
- 把 13.5 的 JSONL 审计日志用
jq做一次"昨天所有 tool_call 耗时 > 2s"的查询。