AI应用可观测性实践：用OpenTelemetry追踪链路与Token成本

普通接口通常可以依靠状态码和结构化日志定位问题，但 LLM 调用还伴随延迟波动、按 Token 计费、输出不可预测以及限额触发等情况。要让 AI 应用稳定运行，需要把每次请求的调用路径、资源消耗和结果质量串联起来，建立覆盖日志、链路与指标的可观测体系。

AI 应用可观测性实战：OpenTelemetry 链路追踪与成本监控

AI 应用上线后两眼一抹黑：LLM 调用耗时多久、成本多少、为什么返回乱码、哪次请求把 Token 限额打满了，统统不知道。

这篇讲清楚四件事：可观测性三要素（日志/链路/指标）、LLM 调用链路追踪实现、Token 成本实时监控、质量指标（延迟/错误率/召回率）统计，最后给排错表和配置清单。

---

目录

• 可观测性三要素：日志、链路、指标
• LLM 调用链路追踪实现
• Token 成本实时监控
• LLM 输出质量评估
• 生产级监控仪表盘
• 快速排错表
• 配置检查清单

---

可观测性三要素：日志、链路、指标

为什么 AI 应用特别需要可观测性

AI 应用和普通微服务不同，有几个特殊挑战：

维度	普通微服务	AI 应用
成本	稳定可预测	按 Token 计费，单次请求成本差异可达100倍
延迟	相对稳定	LLM 调用 1-30 秒不等，波动大
调试	有结构化日志	输出是自然语言，难以自动化判断
可靠性	明确的错误码	LLM 可能返回乱码、截断、有害内容
Token 限额	无	有 RPM/TPM 限制，容易触发 429

三支柱模型

可观测性
├── ? 日志（Logs）：发生了什么，事件级别的原始记录
├── ? 链路（Traces）：怎么发生的，请求的完整调用路径
└── ? 指标（Metrics）：发生了多少，聚合的数值统计

---

LLM 调用链路追踪实现

Trace ID 生成与传递

import uuid
import time
import json
import logging
from typing import Optional, Dict, Any, List
from dataclasses import dataclass, field
from contextvars import ContextVar

# 设置日志
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

# 上下文变量：Trace ID 跨异步传递
current_trace_id: ContextVar[Optional[str]] = ContextVar('trace_id', default=None)

@dataclass
class Span:
    """链路中的一个 Span（操作单元）"""
    span_id: str
    trace_id: str
    parent_span_id: Optional[str]
    operation_name: str
    start_time: float
    end_time: Optional[float] = None
    status: str = "ok"  # ok / error
    attributes: Dict[str, Any] = field(default_factory=dict)
    events: List[Dict] = field(default_factory=list)

    @property
    def duration_ms(self) -> float:
        if self.end_time is None:
            return -1
        return (self.end_time - self.start_time) * 1000

    def add_attribute(self, key: str, value: Any):
        self.attributes[key] = value

    def add_event(self, name: str, attributes: Optional[Dict] = None):
        self.events.append({
            "name": name,
            "timestamp": time.time(),
            "attributes": attributes or {}
        })

    def finish(self, status: str = "ok"):
        self.end_time = time.time()
        self.status = status

class LLMLocalTracer:
    """轻量级本地链路追踪器（适合单机/小规模，生产建议用 OpenTelemetry）"""

    def __init__(self):
        self.spans: List[Span] = []
        self._enabled = True

    def start_span(self, operation_name: str, parent_span_id: Optional[str] = None) -> Span:
        """启动一个新的 Span"""
        trace_id = current_trace_id.get()
        if not trace_id:
            trace_id = str(uuid.uuid4())
            current_trace_id.set(trace_id)

        span_id = str(uuid.uuid4())[:16]
        span = Span(
            span_id=span_id,
            trace_id=trace_id,
            parent_span_id=parent_span_id,
            operation_name=operation_name,
            start_time=time.time()
        )
        self.spans.append(span)
        return span

    def get_trace_summary(self) -> Dict[str, Any]:
        """获取当前 Trace 的汇总信息"""
        if not self.spans:
            return {"total_spans": 0}

        total_time = 0
        error_count = 0
        for span in self.spans:
            if span.end_time:
                total_time += span.duration_ms
            if span.status == "error":
                error_count += 1

        return {
            "trace_id": self.spans[0].trace_id,
            "total_spans": len(self.spans),
            "total_duration_ms": round(total_time, 2),
            "error_count": error_count,
            "spans": [
                {
                    "name": s.operation_name,
                    "duration_ms": round(s.duration_ms, 2) if s.end_time else -1,
                    "status": s.status,
                    "attributes": s.attributes
                }
                for s in self.spans
            ]
        }

# 全局追踪器实例
tracer = LLMLocalTracer()

def call_llm_with_trace(model: str, prompt: str, temperature: float = 0.7,
                        max_tokens: int = 1000) -> Dict[str, Any]:
    """带链路追踪的 LLM 调用"""
    from openai import OpenAI
    import os

    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    # 1. 创建请求 Span
    req_span = tracer.start_span("llm.request")
    req_span.add_attribute("model", model)
    req_span.add_attribute("temperature", temperature)
    req_span.add_attribute("max_tokens", max_tokens)

    try:
        # 2. 发送请求
        start = time.time()
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=temperature,
            max_tokens=max_tokens
        )
        latency = time.time() - start

        # 3. 记录响应信息
        req_span.add_attribute("latency_ms", round(latency * 1000, 2))
        req_span.add_attribute("usage.input_tokens", response.usage.prompt_tokens)
        req_span.add_attribute("usage.output_tokens", response.usage.completion_tokens)
        req_span.add_attribute("usage.total_tokens", response.usage.total_tokens)
        req_span.add_attribute("finish_reason", response.choices[0].finish_reason)

        req_span.finish(status="ok")
        logger.info(f"[Trace] {response.usage.total_tokens} tokens, {latency:.2f}s")

        return {
            "content": response.choices[0].message.content,
            "usage": {
                "prompt_tokens": response.usage.prompt_tokens,
                "completion_tokens": response.usage.completion_tokens,
                "total_tokens": response.usage.total_tokens
            },
            "latency": latency,
            "finish_reason": response.choices[0].finish_reason
        }

    except Exception as e:
        req_span.add_event("error", {"message": str(e), "type": type(e).__name__})
        req_span.finish(status="error")
        logger.error(f"[Trace] LLM 调用失败: {e}")
        raise

OpenTelemetry 集成（生产推荐）

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor, ConsoleSpanExporter
from opentelemetry.trace import Status, StatusCode

# 初始化 OpenTelemetry（一次性配置）
def setup_opentelemetry(service_name: str):
    provider = TracerProvider()
    # 本地开发：输出到控制台
    processor = BatchSpanProcessor(ConsoleSpanExporter())
    provider.add_span_processor(processor)
    trace.set_tracer_provider(provider)
    return trace.get_tracer(service_name)

tracer = setup_opentelemetry("ai-chat-service")

def call_llm_with_otel(model: str, prompt: str, temperature: float = 0) -> str:
    """使用 OpenTelemetry 追踪 LLM 调用"""
    from openai import OpenAI
    import os

    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    with tracer.start_as_current_span("llm.call") as span:
        span.set_attribute("llm.model", model)
        span.set_attribute("llm.temperature", temperature)

        try:
            response = client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": prompt}],
                temperature=temperature
            )

            usage = response.usage
            span.set_attribute("llm.usage.prompt_tokens", usage.prompt_tokens)
            span.set_attribute("llm.usage.completion_tokens", usage.completion_tokens)
            span.set_attribute("llm.usage.total_tokens", usage.total_tokens)

            content = response.choices[0].message.content
            span.set_status(Status(StatusCode.OK))
            return content

        except Exception as e:
            span.set_status(Status(StatusCode.ERROR, str(e)))
            span.record_exception(e)
            raise

---

Token 成本实时监控

成本追踪器实现

import time
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

@dataclass
class CostRecord:
    """单次 API 调用成本记录"""
    timestamp: datetime
    model: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    cost_usd: float
    latency_ms: float
    trace_id: Optional[str]

# 主流模型价格表（美元/1K tokens，截至 2026 年 9 月）
MODEL_PRICING = {
    "gpt-4o": {"input": 0.005, "output": 0.015, "cached_input": 0.00125},
    "gpt-4o-mini": {"input": 0.00015, "output": 0.0006, "cached_input": 0.000075},
    "gpt-4-turbo": {"input": 0.01, "output": 0.03, "cached_input": 0.0015},
    "claude-3-5-sonnet": {"input": 0.003, "output": 0.015, "cached_input": 0.0003},
    "deepseek-v3": {"input": 0.00027, "output": 0.0011, "cached_input": 0.00007},
    "qwen-plus": {"input": 0.001, "output": 0.006, "cached_input": None},
}

class CostTracker:
    """Token 成本追踪器"""

    def __init__(self):
        self.records: list[CostRecord] = []
        self.daily_budget_usd: float = 10.0  # 默认每日限额

    def record(self, model: str, prompt_tokens: int, completion_tokens: int,
               latency_ms: float, trace_id: Optional[str] = None) -> CostRecord:
        """记录一次调用的成本"""
        total_tokens = prompt_tokens + completion_tokens
        pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})

        input_cost = prompt_tokens / 1000 * pricing["input"]
        output_cost = completion_tokens / 1000 * pricing["output"]
        total_cost = input_cost + output_cost

        record = CostRecord(
            timestamp=datetime.now(),
            model=model,
            prompt_tokens=prompt_tokens,
            completion_tokens=completion_tokens,
            total_tokens=total_tokens,
            cost_usd=round(total_cost, 6),
            latency_ms=latency_ms,
            trace_id=trace_id
        )
        self.records.append(record)
        return record

    def get_today_cost(self) -> float:
        """计算今日累计成本"""
        today = datetime.now().date()
        return sum(
            r.cost_usd for r in self.records
            if r.timestamp.date() == today
        )

    def get_summary(self) -> dict:
        """获取成本汇总统计"""
        today = datetime.now().date()
        today_records = [r for r in self.records if r.timestamp.date() == today]

        if not today_records:
            return {"date": str(today), "total_cost_usd": 0, "total_calls": 0}

        return {
            "date": str(today),
            "total_cost_usd": round(sum(r.cost_usd for r in today_records), 4),
            "total_calls": len(today_records),
            "total_tokens": sum(r.total_tokens for r in today_records),
            "avg_latency_ms": round(
                sum(r.latency_ms for r in today_records) / len(today_records), 2
            ),
            "budget_usd": self.daily_budget_usd,
            "budget_usage_pct": round(
                sum(r.cost_usd for r in today_records) / self.daily_budget_usd * 100, 1
            )
        }

    def check_budget(self) -> bool:
        """检查是否超过预算，返回 True 表示安全，False 表示超限"""
        return self.get_today_cost() < self.daily_budget_usd

# 全局实例
cost_tracker = CostTracker()

# 集成到 LLM 调用
def call_llm_with_cost_tracking(model: str, prompt: str, trace_id: Optional[str] = None) -> str:
    """带成本追踪的 LLM 调用"""
    from openai import OpenAI
    import os

    # 预算检查
    if not cost_tracker.check_budget():
        raise RuntimeError(
            f"每日预算 ${cost_tracker.daily_budget_usd} 已超限！"
            f" 当前: ${cost_tracker.get_today_cost():.4f}"
        )

    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    start = time.time()
    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    )
    latency_ms = (time.time() - start) * 1000

    # 记录成本
    record = cost_tracker.record(
        model=model,
        prompt_tokens=response.usage.prompt_tokens,
        completion_tokens=response.usage.completion_tokens,
        latency_ms=latency_ms,
        trace_id=trace_id
    )

    # 超预算 80% 时告警
    usage_pct = record.cost_usd / cost_tracker.daily_budget_usd * 100
    if usage_pct > 80:
        print(f"⚠️ 预算预警: {usage_pct:.1f}% 已使用 (${record.cost_usd:.6f})")

    return response.choices[0].message.content

---

LLM 输出质量评估

自动化质量指标

from dataclasses import dataclass
from typing import Optional

@dataclass
class QualityMetrics:
    """LLM 输出质量指标"""
    response_length: int
    contains_code: bool
    has_refusal: bool  # 是否拒绝回答
    has_mention_of_context: bool  # 是否引用了上下文
    latency_ms: float
    tokens_per_second: float
    overall_score: Optional[float] = None

def evaluate_response(response: str, latency_ms: float,
                     total_tokens: int, prompt_tokens: int) -> QualityMetrics:
    """评估单次 LLM 输出的质量"""
    response_length = len(response)

    # 基础指标
    tokens_per_second = (total_tokens / latency_ms * 1000) if latency_ms > 0 else 0
    contains_code = "```" in response or "def " in response or "class " in response
    has_refusal = any(kw in response for kw in ["无法", "不知道", "无法回答", "no information", "I don't know"])
    has_mention_of_context = any(kw in response for kw in ["根据", "基于", "context", "参考资料", "根据上文"])

    # 简单质量评分（生产环境建议用 LLM-as-Judge）
    score = 0.0
    if response_length > 50:
        score += 30  # 有实质内容
    if contains_code:
        score += 20  # 有代码块
    if has_mention_of_context:
        score += 20  # 引用了上下文
    if not has_refusal:
        score += 15  # 正常回答
    if 50 < response_length < 2000:
        score += 15  # 长度适中

    return QualityMetrics(
        response_length=response_length,
        contains_code=contains_code,
        has_refusal=has_refusal,
        has_mention_of_context=has_mention_of_context,
        latency_ms=latency_ms,
        tokens_per_second=round(tokens_per_second, 2),
        overall_score=score
    )

---

生产级监控仪表盘

Prometheus + Grafana 集成

from prometheus_client import Counter, Histogram, Gauge, generate_latest
import time

# 定义 Prometheus 指标
llm_requests_total = Counter(
    'llm_requests_total',
    'Total LLM API requests',
    ['model', 'status']  # 按模型和状态标签
)

llm_latency_seconds = Histogram(
    'llm_latency_seconds',
    'LLM API latency in seconds',
    ['model']
)

llm_tokens_total = Counter(
    'llm_tokens_total',
    'Total tokens used',
    ['model', 'token_type']  # prompt / completion
)

llm_cost_usd = Counter(
    'llm_cost_usd',
    'Total cost in USD',
    ['model']
)

llm_budget_usage = Gauge(
    'llm_daily_budget_usage_pct',
    'Daily budget usage percentage'
)

def call_llm_with_prometheus(model: str, prompt: str) -> str:
    """带 Prometheus 指标的 LLM 调用"""
    from openai import OpenAI
    import os

    client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
    start = time.time()

    try:
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}]
        )
        latency = time.time() - start
        content = response.choices[0].message.content
        usage = response.usage

        # 记录指标
        llm_requests_total.labels(model=model, status="success").inc()
        llm_latency_seconds.labels(model=model).observe(latency)
        llm_tokens_total.labels(model=model, token_type="prompt").inc(usage.prompt_tokens)
        llm_tokens_total.labels(model=model, token_type="completion").inc(usage.completion_tokens)

        # 成本
        pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})
        cost = (usage.prompt_tokens / 1000 * pricing["input"] +
                usage.completion_tokens / 1000 * pricing["output"])
        llm_cost_usd.labels(model=model).inc(cost)

        # 预算使用率
        usage_pct = cost_tracker.get_today_cost() / cost_tracker.daily_budget_usd * 100
        llm_budget_usage.set(usage_pct)

        return content

    except Exception as e:
        llm_requests_total.labels(model=model, status="error").inc()
        raise

# 健康检查端点
def metrics_endpoint():
    """返回 Prometheus 格式的指标（供 /metrics 端点调用）"""
    return generate_latest()

---

快速排错表

问题	可能原因	解决方法
链路只追踪到一半	异常导致 Span 未 finish	用 try/finally 确保 Span 在所有路径下都 finish
429 错误频发	RPM/TPM 限额超	加速率限制退避，实现请求队列
成本异常高	Prompt 过长 / 循环调用	检查 Token 统计，检查是否有重复调用
输出为空或截断	max_tokens 太小 / 模型截断	增大 max_tokens，检查 finish_reason
链路跨服务断掉	Trace ID 未传递	确保 Trace ID 通过 HTTP header 传递到下游
Prometheus 指标为 0	指标未注册到同一 Registry	多进程共享 Registry，或用 Pushgateway
预算预警一直响	日预算设置太低	按实际用量调整 daily_budget_usd
Embedding 耗时比 LLM 还长	向量库查询慢 / 网络问题	检查向量库连接池配置

配置检查清单

检查项	推荐做法
Trace ID 传递	用 ContextVar 跨协程传递，或通过 HTTP header X-Trace-ID 传递
Span 命名	统一命名规范：{service}.{operation}，如 llm.call, vectorstore.query
采样策略	生产环境用尾采样（只保留错误/慢请求完整链路）
成本追踪	每次调用记录 prompt_tokens / completion_tokens，累加到 Redis
预算告警	80% 时警告，100% 时熔断拒绝新请求
指标导出	Prometheus + Grafana，指标加 model / status / operation 标签
日志格式	结构化 JSON，含 trace_id / user_id / model / cost / latency
慢查询	P99 延迟超 10 秒的请求单独告警
敏感数据	日志中脱敏 user query / API key / token 用量
上线前	灰度发布，第一天盯紧成本曲线和错误率