作为一名后端开发，我曾经对 API 可观测性一无所知。直到有一天凌晨两点，线上 AI 对话服务突然超时，用户反馈"机器人不回答了"。我翻遍日志才找到问题根源——原来是某个模型供应商的 API 响应时间从平时的 800ms 暴涨到了 15 秒。那一刻我意识到，没有监控的 API 调用就像蒙着眼睛开车。
今天我要手把手教你使用 OpenTelemetry 为 AI API 构建完整的可观测性体系。我们将监控延迟、错误率、token 消耗等关键指标，而所有示例都将基于 HolySheep AI 的 API endpoint（国内访问延迟低于 50ms，无需科学上网）。
什么是 OpenTelemetry？为什么 AI API 需要它？
OpenTelemetry（简称 OTel）是 CNCF 旗下的可观测性框架，它能自动追踪你的代码"从哪里来、到哪里去、花了多长时间"。对于 AI API 调用来说，这意味着你可以：
- 记录每次 API 调用的响应时间，识别慢查询
- 捕获错误信息，快速定位问题根因
- 统计 Token 消耗，优化成本
- 在 Grafana 或 Jaeger 中可视化所有调用链路
准备工作：一分钟注册 HolySheheep AI
在开始之前，你需要准备一个 API Key。如果你还没有，建议使用 HolySheep AI，因为它对中国开发者非常友好：
- 微信/支付宝直接充值，汇率 1 元 = 1 美元（官方是 7.3 元换 1 美元，节省超过 85%）
- 国内服务器直连，延迟低于 50ms
- 注册即送免费额度
- 支持 GPT-4.1、Claude Sonnet 4.5、Gemini 2.5 Flash 等 2026 年主流模型
图 1：登录后进入控制台 → API Keys → 创建新密钥（请妥善保管，不要泄露）
第一步：安装必要的依赖
我们使用 Python 来演示，因为它的生态最成熟。打开终端，执行以下命令：
pip install opentelemetry-api \
opentelemetry-sdk \
opentelemetry-exporter-otlp \
opentelemetry-instrumentation \
opentelemetry-instrumentation-requests \
requests这些包的作用分别是：opentelemetry-api 定义追踪接口，opentelemetry-sdk 是核心实现，exporter 负责导出数据到后端，而 instrumentation-requests 能自动监控 HTTP 请求。
第二步：初始化 OpenTelemetry 并挂载全局追踪器
创建一个 otel_setup.py 文件，这是所有监控的基础。我在这里配置了数据导出到本地 Jaeger（你也可以换成其他后端）：
import os
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.sdk.resources import Resource, SERVICE_NAME
定义服务信息
resource = Resource(attributes={
SERVICE_NAME: "ai-api-monitor"
})
创建追踪器提供者
provider = TracerProvider(resource=resource)
输出到控制台（调试用，生产环境建议用 OTLPExporter 发送到 Jaeger/Grafana）
processor = BatchSpanProcessor(ConsoleSpanExporter())
provider.add_span_processor(processor)
设为全局默认
trace.set_tracer_provider(provider)
def get_tracer(name: str):
"""获取指定名称的追踪器"""
return trace.get_tracer(name)
print("✅ OpenTelemetry 初始化完成！")运行这个脚本，如果看到"✅ OpenTelemetry 初始化完成！"，说明基础配置成功。
第三步：封装 AI API 客户端并添加自定义属性
这是本文最核心的部分。我创建了一个包装类，能自动记录每次调用的关键指标：
import time
import requests
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
class HolySheepAIClient:
"""带 OpenTelemetry 监控的 HolySheep AI 客户端"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url.rstrip('/')
self.tracer = trace.get_tracer("holysheep-client")
def chat_completion(self, messages: list, model: str = "gpt-4.1", **kwargs):
"""
发送聊天请求并自动追踪所有指标
Args:
messages: 消息列表，格式同 OpenAI
model: 模型名称，如 gpt-4.1, claude-sonnet-4.5, gemini-2.5-flash
**kwargs: 其他参数如 temperature, max_tokens
Returns:
dict: API 响应
"""
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("ai.chat_completion") as span:
# 设置标签，方便后续过滤
span.set_attribute("ai.model", model)
span.set_attribute("ai.provider", "holysheep")
span.set_attribute("ai.messages_count", len(messages))
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": messages,
**kwargs
}
start_time = time.time()
try:
response = requests.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload,
timeout=30
)
elapsed_ms = (time.time() - start_time) * 1000
# 记录延迟
span.set_attribute("ai.latency_ms", round(elapsed_ms, 2))
span.set_attribute("http.status_code", response.status_code)
if response.status_code == 200:
data = response.json()
# 提取 token 使用量（关键成本指标）
usage = data.get("usage", {})
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
total_tokens = usage.get("total_tokens", 0)
span.set_attribute("ai.prompt_tokens", prompt_tokens)
span.set_attribute("ai.completion_tokens", completion_tokens)
span.set_attribute("ai.total_tokens", total_tokens)
# 计算预估成本（以 GPT-4.1 为例）
if model == "gpt-4.1":
# GPT-4.1 input: $2/MTok, output: $8/MTok
cost = (prompt_tokens / 1_000_000 * 2) + (completion_tokens / 1_000_000 * 8)
elif model == "deepseek-v3.2":
# DeepSeek V3.2: input $