📚 AI 工程与智能体教程站

第 13 章 · 遥测与错误处理 ​

本章目标：

  • 理解 AI SDK 基于 OpenTelemetry 的遥测体系，学会启用与定制
  • 掌握 registerTelemetry 全局注册与按调用配置 telemetry 参数
  • 学会编写自定义 Telemetry integration 收集生成生命周期事件
  • 掌握普通错误、流式错误、流中止（abort）三类错误处理模式

13.1 遥测体系概览 ​

AI SDK 使用 OpenTelemetry 收集遥测数据。OpenTelemetry 是一个开源可观测性框架，为采集遥测数据提供标准化的埋点规范。

你可以查看官方文档的 AI SDK Observability Integrations 页面，了解哪些厂商为 AI SDK 应用提供监控与追踪能力。

13.2 启用遥测 ​

第一步：注册 OpenTelemetry 集成 ​

OpenTelemetry span 采集需要安装 @ai-sdk/otel 包，并在应用启动时注册一次：

bash
pnpm install @ai-sdk/otel
ts
import { registerTelemetry } from 'ai';
import { OpenTelemetry } from '@ai-sdk/otel';

registerTelemetry(new OpenTelemetry());

对于 Next.js 应用，在项目根目录创建 instrumentation.ts 文件，把 AI SDK 的遥测集成和 OpenTelemetry provider 一起注册：

ts
import { registerOTel } from '@vercel/otel';
import { registerTelemetry } from 'ai';
import { OpenTelemetry } from '@ai-sdk/otel';

export function register() {
  registerOTel({
    serviceName: 'my-ai-app',
  });

  registerTelemetry(new OpenTelemetry());
}

更多在 Next.js 中设置 OpenTelemetry 的细节，参见 Next.js 官方的 OpenTelemetry 指南。对于不带 Next.js 的 Node.js 应用，在入口文件顶层注册即可。

第二步：开启遥测 ​

一旦注册了遥测集成，所有 AI SDK 调用默认都会发出遥测事件。你仍然可以传入 telemetry 参数附加元数据（比如 functionId），或对某次调用选择退出。

ts
import { generateText, createGateway } from 'ai';
import 'dotenv/config';

// 方式一：Vercel AI Gateway（推荐默认）；同样适用于自定义 provider（见第 3 章）
const gateway = createGateway({ apiKey: process.env.AI_GATEWAY_API_KEY ?? '' });

const result = await generateText({
  model: gateway('openai/gpt-5'),
  prompt: 'Write a short story about a cat.',
  telemetry: {
    functionId: 'story-agent',
  },
});

默认情况下输入与输出都会被记录。将 recordInputs 和 recordOutputs 设为 false 可以关闭它们——这对隐私、数据传输量与性能都有帮助。例如当输入包含敏感信息时，可以只关闭输入记录。

选择退出 ​

遥测是 opt-out（默认开启）的设计。要对特定调用关闭遥测，设置 isEnabled: false：

ts
const result = await generateText({
  model: gateway('openai/gpt-5'),
  prompt: 'Write a short story about a cat.',
  telemetry: { isEnabled: false },
});

想全局禁用遥测的话，不通过 registerTelemetry() 注册任何遥测集成即可。

13.3 遥测元数据 ​

你可以提供 functionId 标识遥测数据对应的函数，并通过 runtimeContext 在遥测数据中附带额外信息：

ts
const result = await generateText({
  model: gateway('openai/gpt-5'),
  prompt: 'Write a short story about a cat.',
  runtimeContext: {
    userId: 'user_123',
    requestId: 'req_abc',
  },
  telemetry: {
    functionId: 'my-awesome-function',
  },
});

运行时上下文的遥测包含控制 ​

runtimeContext 中可能包含应用内部有用但不该发送给遥测厂商的值（如用户标识、租户 ID 或凭证）。使用 telemetry.includeRuntimeContext 标记哪些顶层属性允许进入遥测：

ts
const result = await generateText({
  model: gateway('openai/gpt-5'),
  prompt: 'Write a short story about a cat.',
  runtimeContext: {
    userId: 'user_123',
    requestId: 'req_abc',
  },
  telemetry: {
    includeRuntimeContext: {
      requestId: true,
    },
  },
});

这个例子里，遥测集成收到的 runtimeContext 只有 { requestId: 'req_abc' }；设为 false 或省略的属性会被排除。若完全省略 telemetry.includeRuntimeContext，则不包含任何运行时上下文属性。该选项受 generateText、streamText 与 ToolLoopAgent 支持。

注意：telemetry.includeRuntimeContext 只过滤遥测集成。生命周期回调与返回结果仍会收到完整的 runtimeContext。

工具上下文 ​

工具上下文同样可能包含执行期间有用但不该进入遥测的值（如 API key、access token）。使用 telemetry.includeToolsContext 选择性纳入工具 context 的顶层属性：

ts
const weatherTool = tool({
  inputSchema: z.object({
    location: z.string(),
  }),
  contextSchema: z.object({
    weatherApiKey: z.string(),
    defaultUnit: z.enum(['celsius', 'fahrenheit']),
  }),
  execute: async ({ location }, { context }) =>
    fetchWeather(location, context.weatherApiKey, context.defaultUnit),
});

const result = await generateText({
  model: gateway('openai/gpt-5'),
  tools: { weather: weatherTool },
  toolsContext: {
    weather: {
      weatherApiKey: 'weather-123',
      defaultUnit: 'fahrenheit',
    },
  },
  prompt: 'What is the weather in San Francisco?',
  telemetry: {
    includeToolsContext: {
      weather: {
        defaultUnit: true,
      },
    },
  },
});

此例中遥测事件里的 toolContext 只会是 { defaultUnit: 'fahrenheit' }，weatherApiKey 被排除在外。规则与 runtimeContext 一致：设为 false 或省略即排除；省略整个选项时不包含任何工具上下文属性。

13.4 自定义 Tracer ​

如果你想让 trace 使用 @opentelemetry/api 单例之外的 TracerProvider，可以向 OpenTelemetry 构造函数传入自定义 Tracer：

ts
import { registerTelemetry } from 'ai';
import { OpenTelemetry } from '@ai-sdk/otel';

const tracerProvider = new NodeTracerProvider();
registerTelemetry(
  new OpenTelemetry({
    tracer: tracerProvider.getTracer('gen_ai'),
  }),
);

13.5 遥测集成（Telemetry Integrations） ​

遥测集成让你挂钩生成生命周期来构建自定义可观测性——日志、分析、DevTools 或任何其他监控系统。你只需实现一次 Telemetry 接口，然后全局注册或通过 telemetry.integrations 传入，而不必在每次调用上单独挂回调。

全局注册集成 ​

使用 registerTelemetry 为所有 AI SDK 调用注册一次集成：

ts
import { registerTelemetry } from 'ai';
import { OpenTelemetry } from '@ai-sdk/otel';

registerTelemetry(new OpenTelemetry());

也可以在一次调用里注册多个集成，它们都会收到相同的生命周期事件：

ts
import { registerTelemetryIntegration } from 'ai';
import { OpenTelemetry } from '@ai-sdk/otel';
import { DevToolsTelemetry } from '@ai-sdk/devtools';

registerTelemetryIntegration(new OpenTelemetry(), DevToolsTelemetry());

按调用传入集成 ​

你也可以给单次 generateText 或 streamText 调用传入集成。提供了按调用集成时，它会替换该次调用的全局注册集成：

ts
import { streamText, createGateway } from 'ai';
import { DevToolsTelemetry } from '@ai-sdk/devtools';

const gateway = createGateway({ apiKey: process.env.AI_GATEWAY_API_KEY ?? '' });

const result = streamText({
  model: gateway('openai/gpt-5'),
  prompt: 'Hello!',
  telemetry: {
    integrations: [DevToolsTelemetry()],
  },
});

多个集成可以组合使用；集成内部抛出的错误会被捕获，不会中断生成流程。

Tracing Channel ​

在 Node.js 中，AI SDK 的遥测生命周期与执行事件还会通过 ai:telemetry tracing channel 发出。这让可观测性厂商无需注册独立集成即可订阅 AI SDK 遥测事件：

ts
import { tracingChannel } from 'node:diagnostics_channel';
import {
  AI_SDK_TELEMETRY_TRACING_CHANNEL,
  type TelemetryTracingChannelMessage,
} from 'ai';

tracingChannel(AI_SDK_TELEMETRY_TRACING_CHANNEL).subscribe({
  start(message) {
    const telemetryMessage = message as TelemetryTracingChannelMessage;

    if (telemetryMessage.type === 'onStart') {
      // 检查 telemetryMessage.event 并转发给你的监控服务
    }
  },
});

Tracing-channel 事件遵循与其他遥测集成相同的按调用设置：telemetry: { isEnabled: false } 会同时禁用已注册集成与 tracing-channel 事件。

构建自定义集成 ​

实现 ai 包的 Telemetry 接口即可。所有方法都是可选的——只实现你关心的生命周期事件：

ts
import type { Telemetry } from 'ai';

class MyIntegration implements Telemetry {
  async onStart(event) {
    console.log('Generation started:', event.modelId);
  }

  async onStepEnd(event) {
    console.log(
      `Step ${event.stepNumber} done:`,
      event.usage.totalTokens,
      'tokens',
    );
  }

  async onToolExecutionEnd(event) {
    if (event.toolOutput.type === 'tool-result') {
      console.log(
        `Tool "${event.toolCall.toolName}" took ${event.toolExecutionMs}ms`,
      );
    } else {
      console.error(
        `Tool "${event.toolCall.toolName}" failed:`,
        event.toolOutput.error,
      );
    }
  }

  async onEnd(event) {
    console.log('Done. Total tokens:', event.usage.totalTokens);
  }

  async onAbort(event) {
    console.log('Stream aborted after', event.steps.length, 'finished steps');
  }
}

export function myIntegration(): Telemetry {
  return new MyIntegration();
}

可用的生命周期方法包括：onStart（生成开始）、onStepStart / onStepEnd（每一步 LLM 调用的起止）、onLanguageModelCallStart / onLanguageModelCallEnd（模型调用前后）、onToolExecutionStart / onToolExecutionEnd（工具执行前后）、onEmbedEnd（嵌入调用完成）、onRerankEnd（重排序调用完成）、onEnd（整体生成完成）、onAbort（流被中止）。各方法的事件类型与对应的生命周期回调一致。

13.6 处理普通错误 ​

普通错误会直接抛出，用 try/catch 块处理即可：

ts
import { generateText, createGateway } from 'ai';
import 'dotenv/config';

const gateway = createGateway({ apiKey: process.env.AI_GATEWAY_API_KEY ?? '' });

try {
  const { text } = await generateText({
    model: gateway('openai/gpt-5'),
    prompt: 'Write a vegetarian lasagna recipe for 4 people.',
  });
} catch (error) {
  // handle error
}

不同错误类型的完整清单见官方 Error Types 参考（/docs/reference/ai-sdk-errors），例如 AI_NoSuchModelError、AI_InvalidPromptError 等。

13.7 处理流式错误 ​

简单流的场景 ​

当流不支持 error chunk 时，错误会作为普通异常抛出，同样可以用 try/catch 处理：

ts
import { streamText, createGateway } from 'ai';
import 'dotenv/config';

const gateway = createGateway({ apiKey: process.env.AI_GATEWAY_API_KEY ?? '' });

try {
  const { textStream } = streamText({
    model: gateway('openai/gpt-5'),
    prompt: 'Write a vegetarian lasagna recipe for 4 people.',
  });

  for await (const textPart of textStream) {
    process.stdout.write(textPart);
  }
} catch (error) {
  // handle error
}

支持 error 分片的流 ​

stream 结果本身支持 error part，你可以像处理其他 part 一样处理它。官方建议同时在流外层保留 try-catch，以捕获发生在流式输出之外（如构造参数阶段）的错误：

ts
import { streamText, createGateway } from 'ai';
import 'dotenv/config';

const gateway = createGateway({ apiKey: process.env.AI_GATEWAY_API_KEY ?? '' });

try {
  const { stream } = streamText({
    model: gateway('openai/gpt-5'),
    prompt: 'Write a vegetarian lasagna recipe for 4 people.',
  });

  for await (const part of stream) {
    switch (part.type) {
      // ... handle other part types

      case 'error': {
        const error = part.error;
        // handle error
        break;
      }

      case 'abort': {
        // handle stream abort
        break;
      }

      case 'tool-error': {
        const error = part.error;
        // handle error
        break;
      }
    }
  }
} catch (error) {
  // handle error
}

13.8 处理流中止（Abort） ​

当流被中止（例如用户点击聊天停止按钮）时，你可能需要做清理操作，比如更新 UI 中存储的消息。使用 onAbort 回调处理这类情况：流经 AbortSignal 中止时会触发 onAbort 而不会触发 onEnd，保证你仍能正确更新 UI 状态。

ts
import { streamText, createGateway } from 'ai';
import 'dotenv/config';

const gateway = createGateway({ apiKey: process.env.AI_GATEWAY_API_KEY ?? '' });

const { textStream } = streamText({
  model: gateway('openai/gpt-5'),
  prompt: 'Write a vegetarian lasagna recipe for 4 people.',
  onAbort: ({ steps }) => {
    // 更新已存储的消息或执行清理
    console.log('Stream aborted after', steps.length, 'steps');
  },
  onEnd: ({ steps, totalUsage }) => {
    // 正常完成时才会调用
    console.log('Stream completed normally');
  },
});

for await (const textPart of textStream) {
  process.stdout.write(textPart);
}

onAbort 回调接收一个参数对象，其中 steps 是中止前所有已完成步骤组成的数组。

你也可以直接在流中处理 abort 事件：

ts
const { stream } = streamText({
  model: gateway('openai/gpt-5'),
  prompt: 'Write a vegetarian lasagna recipe for 4 people.',
});

for await (const chunk of stream) {
  switch (chunk.type) {
    case 'abort': {
      // 在流中直接处理中止
      console.log('Stream was aborted');
      break;
    }
    // ... handle other part types
  }
}

本章小结 ​

  • AI SDK 通过 @ai-sdk/otel + registerTelemetry(new OpenTelemetry()) 一行代码接入 OpenTelemetry，Next.js 中配合 instrumentation.ts 注册；
  • 遥测默认开启且 opt-out：telemetry: { isEnabled: false } 可对单次调用退出，不注册集成即全局关闭；
  • recordInputs / recordOutputs 控制输入输出记录，includeRuntimeContext / includeToolsContext 精确控制敏感上下文是否进入遥测；
  • 自定义可观测性只需实现 Telemetry 接口（方法全部可选），可全局注册、也可按调用覆盖；
  • 普通错误用 try/catch；支持 error 分片的流要在 for await 里处理 error / abort / tool-error part；
  • 流中止用 onAbort 回调清理状态（此时 onEnd 不触发），或在流中处理 abort 分片。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. AI SDK 的遥测功能基于哪个可观测性框架？

2. 如何对单个 generateText 调用关闭遥测？

3. 关于 telemetry.includeRuntimeContext，下列说法正确的是？

4. 当流式生成被 AbortSignal 中止时，会发生什么？

🛠️ 动手实践 ​

  1. 给第 5 章的文本生成脚本接入 OpenTelemetry：安装 @ai-sdk/otel，注册集成后发起两次 generateText 调用（一次带 functionId: 'story-agent'，一次 isEnabled: false），观察 span 输出的差异。
  2. 编写一个自定义 Telemetry integration，实现 onStart、onStepEnd、onEnd 三个方法，把每步 token 用量和总用量写入 JSON 文件，然后跑一次带工具调用的多步生成验证记录完整性。
  3. 改造流式脚本：加上 onAbort 回调打印已完成步骤数，并用 AbortController 在 2 秒后主动中止流，验证 onEnd 不会被调用而 onAbort 会触发。