目录

LLM 可观测性:监控 Prompt 和 Response 的实战方法

LLM 服务上线后,用户开始抱怨输出质量下降。你打开监控面板——CPU 正常,内存正常,P99 延迟在 SLA 范围内。服务从基础设施角度看完全健康,但你根本不知道模型在返回什么。这就是 LLM 可观测性的缺口:传统 APM 工具能告诉你系统好不好,但告诉不了你 AI 好不好。

LLM 可观测性的三个维度

LLM 的可观测性不同于普通服务,需要在三个层面同时监控:

运维指标(Operational Metrics)

这些是传统 APM 覆盖的部分,但 LLM 有一些独特指标:

指标 说明 告警阈值(参考)
latency_p99 端到端延迟 P99 > 10s
ttft (Time to First Token) 首个 token 延迟 > 3s
tokens_per_second 生成速度 < 10 TPS
input_tokens / output_tokens Token 用量(直接影响成本) 超出预算
error_rate 按错误类型分组 > 1%

质量指标(Quality Metrics)

这是 LLM 独有的挑战——如何量化「回答好不好」:

  • Faithfulness(忠实度):回答是否基于检索到的上下文(RAG 场景)
  • Relevance(相关性):回答是否回应了用户的问题
  • Coherence(连贯性):回答是否前后一致、逻辑清晰
  • LLM-as-Judge 评分:用另一个 LLM 对输出打分,适合批量评估

业务指标(Business Metrics)

最终要回答的问题:用户有没有得到他们想要的?

  • 任务完成率(用户是否需要多次追问)
  • 用户满意度(显式评分或隐式行为信号)
  • 对话放弃率(用户中途退出的比例)

工具选型

选一个合适的 LLM 可观测性工具能省大量重复造轮子的时间:

Langfuse

开源、可自托管,有 Python 和 JavaScript SDK。最大优势是对 LangChain、LlamaIndex 有原生集成,几行代码接入。支持 prompt 版本管理、A/B 测试追踪。适合需要完全数据主权的团队。

Helicone

代理模式接入——只改一行 API 地址,零代码侵入。把 OpenAI 请求路由到 Helicone 代理,自动记录所有 prompt/response。适合快速接入、不想改代码的场景。

Phoenix by Arize

ML 可观测性背景,对 embedding、检索质量的分析更专业。有内置的 Hallucination Detector 和 QA 评估器。适合 RAG 系统的深度评估。

OpenTelemetry

如果你的团队已经有 OTel 基础设施,可以用标准协议自建。社区维护了 opentelemetry-instrumentation-openai 等包。灵活但需要更多自建工作。

实战:用 Langfuse 追踪 Prompt/Response

import os
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
import openai

langfuse = Langfuse(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
    host="https://cloud.langfuse.com",  # 或自托管地址
)

openai_client = openai.OpenAI()

@observe()  # 自动追踪这个函数的调用
def answer_question(user_question: str, user_id: str) -> str:
    # 关联用户 ID,便于后续过滤
    langfuse_context.update_current_trace(
        user_id=user_id,
        tags=["production", "v2"],
    )

    # 使用 prompt 模板(在 Langfuse 控制台版本化管理)
    prompt = langfuse.get_prompt("support-assistant")
    compiled = prompt.compile(question=user_question)

    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=compiled,
    )
    answer = response.choices[0].message.content

    langfuse_context.update_current_observation(
        output=answer,
        metadata={"model": "gpt-4o", "user_id": user_id},
    )
    return answer

# 用户给出反馈后,追加评分
def record_user_feedback(trace_id: str, score: float, comment: str = ""):
    langfuse.score(
        trace_id=trace_id,
        name="user_satisfaction",
        value=score,  # 0.0 - 1.0
        comment=comment,
    )

运行后,在 Langfuse 控制台可以看到:每次调用的 prompt 内容、response 内容、Token 用量、延迟、用户 ID 和评分趋势。

关键指标与告警

配置告警时,建议分层设置:

P1(立即响应)

  • 错误率 > 5%(按 error_type 分组:context_length_exceededrate_limitcontent_filter
  • P99 延迟 > 30s

P2(1 小时内响应)

  • P95 延迟 > 10s
  • 日 Token 消耗超出预算的 80%
  • Faithfulness 评分连续 1 小时低于 0.75

趋势监控(每日 review)

  • 平均 input/output token 长度趋势(突然增长可能是 prompt 泄漏或攻击)
  • 质量分数周环比变化
  • 新增 error type 类型

Prometheus 告警规则示例:

# prometheus-rules.yml
groups:
  - name: llm_alerts
    rules:
      - alert: LLMHighErrorRate
        expr: >
          rate(llm_requests_total{status="error"}[5m])
          / rate(llm_requests_total[5m]) > 0.05
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "LLM error rate {{ $value | humanizePercentage }}"

      - alert: LLMHighLatency
        expr: histogram_quantile(0.99, rate(llm_latency_seconds_bucket[5m])) > 30
        for: 5m
        labels:
          severity: warning

数据存储与查询

LLM trace 数据量大,需要专门考虑存储策略:

存储选型

  • ClickHouse:高写入吞吐、列式存储,适合海量 trace 查询。Langfuse 自托管版本使用 ClickHouse。
  • PostgreSQL:数据量不大时(< 100 万条/天)够用,查询更灵活。
  • S3 + Athena:冷数据归档,成本最低,查询延迟高。

采样策略

不需要存储所有请求,按重要性采样:

import random

def should_trace(request_context: dict) -> bool:
    # 错误请求:100% 记录
    if request_context.get("error"):
        return True

    # 用户有反馈的请求:100% 记录
    if request_context.get("has_feedback"):
        return True

    # 低质量评分(< 0.6):100% 记录
    if request_context.get("quality_score", 1.0) < 0.6:
        return True

    # 正常请求:10% 随机采样
    return random.random() < 0.10

RETENTION_POLICY = {
    "full_fidelity_days": 90,   # 最近 90 天全量保存
    "compressed_days": 365,     # 90-365 天压缩保存(只保留关键字段)
    "archive_days": 730,        # 1-2 年归档到冷存储
}

延伸阅读