可观测性:Traces、Spans 与 OpenTelemetry

为什么 LLM 应用格外需要可观测性

传统 Web 服务的 APM(New Relic / Datadog)关心 CPU、内存、QPS、p95 延迟。这些对 LLM 应用仍然重要,但不够——LLM 独有的失败方式,传统监控看不到:

• 成本爆炸:一次意外的长上下文可能花掉 $2;一天下来是百美元的区别
• 幻觉:输出语法没错但事实错,日志里全是 HTTP 200
• 工具链失败:Agent 走了 15 步才答一个简单问题,没人知道
• Prompt 注入:用户输入诱导模型泄露系统 prompt,需要专门侦测
• 漂移:上游模型静默升级(GPT-4o → GPT-4o-2025-03),行为突变

Trace / Span / Event:三个核心概念

一个典型 Agent Trace 长什么样

OpenTelemetry:可观测性的事实标准

OpenTelemetry(OTel)是 CNCF 毕业项目,provides 通用的 trace/metric/log 规范和 SDK。2024-2025 年它发布了专门的 GenAI Semantic Conventions,定义了 LLM 调用应该上报的标准属性。

GenAI 核心属性(官方语义约定)

最小接入:Python + OpenTelemetry

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint="http://otel-collector:4318/v1/traces"))
)
tracer = trace.get_tracer("my-llm-app")

def call_llm(messages, model="gpt-4o-mini"):
    with tracer.start_as_current_span("chat") as span:
        span.set_attribute("gen_ai.system", "openai")
        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("gen_ai.operation.name", "chat")

        rsp = client.chat.completions.create(model=model, messages=messages)

        span.set_attribute("gen_ai.response.model", rsp.model)
        span.set_attribute("gen_ai.usage.input_tokens", rsp.usage.prompt_tokens)
        span.set_attribute("gen_ai.usage.output_tokens", rsp.usage.completion_tokens)
        span.set_attribute("gen_ai.response.finish_reasons", [rsp.choices[0].finish_reason])
        return rsp

自动插桩(更推荐)

大部分 LLM SDK 已经有自动插桩包,接入两行就完事,不用手写 span:

pip install opentelemetry-instrumentation-openai
# 或 anthropic / langchain / llama-index

from opentelemetry.instrumentation.openai import OpenAIInstrumentor
OpenAIInstrumentor().instrument()

# 之后所有 openai 调用都会自动产出合规的 span
rsp = client.chat.completions.create(...)

三大 LLM 观测后端对比

关键监控维度

① 延迟(Latency)

LLM 延迟比普通 API 更复杂,至少要看 3 个维度:

• TTFT(Time To First Token):首 token 到达时间。流式场景直接决定用户感知
• TPOT(Time Per Output Token):每 token 平均生成耗时。模型吞吐核心指标
• 端到端延迟:请求到最后一个 token,含 retriever / tool / post-processing

import time

def stream_with_metrics(messages):
    with tracer.start_as_current_span("chat_stream") as span:
        start = time.time()
        ttft, n_tokens = None, 0

        stream = client.chat.completions.create(
            model="gpt-4o-mini", messages=messages, stream=True,
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                if ttft is None:
                    ttft = time.time() - start
                    span.add_event("first_token", attributes={"ttft_ms": int(ttft * 1000)})
                n_tokens += 1
                yield chunk.choices[0].delta.content

        total = time.time() - start
        tpot = (total - ttft) / max(n_tokens - 1, 1)
        span.set_attribute("llm.latency.ttft_ms", int(ttft * 1000))
        span.set_attribute("llm.latency.tpot_ms", int(tpot * 1000))
        span.set_attribute("llm.latency.total_ms", int(total * 1000))

② 成本(Cost)

按模型单价把 token 折算成钱,维度上报。这是 LLM 特有且最关键的 SLO。

PRICING = {  # 美元 / 1K tokens,2026 年报价
    "gpt-4o-mini":       {"in": 0.00015, "out": 0.00060},
    "gpt-4o":            {"in": 0.00250, "out": 0.01000},
    "claude-sonnet-4-6": {"in": 0.00300, "out": 0.01500},
    "claude-opus-4-7":   {"in": 0.01500, "out": 0.07500},
}

def annotate_cost(span, model, in_tokens, out_tokens):
    p = PRICING.get(model)
    if not p: return
    cost = (in_tokens * p["in"] + out_tokens * p["out"]) / 1000
    span.set_attribute("gen_ai.usage.cost_usd", cost)

③ Token 使用

输入/输出 token 分别看。常见异常模式:

• 输入 token 突增:prompt 注入了脏数据,或 RAG 检索返回过大 chunk
• 输出 token 拉满到 max_tokens:模型想说更多,被截断——答案可能不完整
• 输出 token 异常长:可能是模型陷入重复生成(loop)或拒答长篇大论

④ 质量信号

质量指标也要打到 span 上,和成本/延迟关联看:

• Judge 评分(采样 1-5%)
• 用户反馈(👍/👎/regenerate)
• Guardrail 判决(pass / blocked / escalated)
• Retry 次数 / Fallback 触发

告警规则设计

LLM 应用的告警比传统服务多了一批"内容相关"的维度。一份合理的告警基线:

Prompt 版本与实验维度

Trace 上一定要打 prompt_version / experiment / variant,否则你无法做"不同版本在线上的真实表现对比":

span.set_attribute("app.prompt.version", "v7")
span.set_attribute("app.prompt.hash", "sha256:abc123...")
span.set_attribute("app.experiment.name", "tone-tweak")
span.set_attribute("app.variant", "B")
span.set_attribute("app.user.segment", "vip")

这些属性和系统指标组合,能直接在观测后端答出:"VIP 用户在 v7 版本的 p95 延迟和成本如何?"

Session/Trace 关联用户反馈

反馈回流必须用 session_id/trace_id 关联,否则"thumbs down"是孤立的点,无法映射到具体 prompt 版本:

@app.post("/feedback")
async def record_feedback(payload: FeedbackIn):
    # trace_id 由前端通过 response header 收到并回传
    langfuse.score(
        trace_id=payload.trace_id,
        name="user_feedback",
        value=1 if payload.thumb == "up" else 0,
        comment=payload.comment,
    )
    return {"ok": True}

敏感信息处理

默认记录完整 prompt/response 很危险——用户隐私、PII、公司机密全进了你的观测后端。三种处理策略:

import re

PII_PATTERNS = [
    (re.compile(r"\b[\w.+-]+@[\w-]+\.[\w.-]+\b"),  "[EMAIL]"),
    (re.compile(r"\b1[3-9]\d{9}\b"),                "[PHONE]"),
    (re.compile(r"\b\d{17}[\dXx]\b"),              "[IDCARD]"),
    (re.compile(r"\b\d{13,19}\b"),                  "[CARD]"),
]

def redact(text: str) -> str:
    for pat, repl in PII_PATTERNS:
        text = pat.sub(repl, text)
    return text

span.set_attribute("gen_ai.prompt", redact(prompt_text)[:2000])

事故回溯流程(Post-Mortem)

一次线上"回答错误"的投诉,靠 trace 如何 5 分钟定位原因?标准操作:

1. 拿到 trace_id:用户投诉附带 session_id,从中找到 trace
2. 查看完整 span 树:看 LLM 调用次数、tool 调用序列
3. 对比基线:同类 query 的典型 trace 长什么样
4. 抓 prompt_version 和 model:是不是升了新版本?是不是命中了 A/B
5. 检查输入:RAG 拿到了什么上下文?有没有污染
6. 检查 guardrail:有没有被过滤,为什么放行
7. 记录根因:写进事故库,加进评估集,避免再犯

Langfuse 最小接入示例

from langfuse import Langfuse
from langfuse.openai import openai  # drop-in 替换

lf = Langfuse(
    public_key="pk-...",
    secret_key="sk-...",
    host="https://cloud.langfuse.com",
)

# 普通 openai 调用,自动 trace
rsp = openai.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "hello"}],
    name="greet-user",                    # span name
    metadata={"prompt_version": "v7"},   # 自定义维度
    user_id="user_42",                     # 用户聚合
    session_id="sess_abc",                 # session 关联
)

# 业务维度打分回流
lf.score(trace_id="...", name="task_success", value=1)

把 Evals 和 Traces 打通

这是闭环的关键:线上 trace → 评估集 → 离线评估 → 新版本 → 再上线。一个推荐架构:

成熟度自检表

本章小结