LLM 应用监控与可观测性 · 运维手册

适用场景：生产环境部署的 LLM 应用（聊天机器人、RAG 系统、Agent），需要监控性能、质量和成本。

---

1. 监控指标体系

1.1 核心指标分类

类别	指标	含义	告警阈值
性能	P50 延迟	50% 请求的响应时间	> 2s
	P99 延迟	99% 请求的响应时间	> 5s
	吞吐量	每秒处理请求数	< 10 req/s
	Token/sec	生成速度	< 20 tok/s
质量	幻觉率	输出与事实不符的比例	> 15%
	回答相关性	答案与问题的匹配度	< 0.7
	用户满意度	点赞/点踩比例	< 80%
成本	Token 消耗	每次请求的 Token 数	-
	API 费用	累计调用成本	> 预算
业务	请求成功率	200 响应比例	< 99%
	错误率	4xx/5xx 比例	> 5%

1.2 监控架构图

┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  LLM 服务    │ ──▶ │  OpenTelemetry│ ──▶ │  Prometheus  │
│  (API)      │     │  (SDK)       │     │  (存储)      │
└─────────────┘     └─────────────┘     └──────┬──────┘
                                               │
                    ┌──────────────────────────┘
                    │
        ┌───────────┼───────────┬───────────┐
        ▼           ▼           ▼           ▼
   ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐
   │ Grafana │ │ 告警    │ │ 日志    │ │ 链路追踪 │
   │ Dashboard│ │ (PagerDuty)│ │ (Loki) │ │ (Jaeger) │
   └─────────┘ └─────────┘ └─────────┘ └─────────┘

---

2. OpenTelemetry 集成

2.1 SDK 初始化

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.prometheus import PrometheusMetricsExporter
from opentelemetry.instrumentation.requests import RequestsInstrumentor

# 初始化 Tracer
provider = TracerProvider()
trace.set_tracer_provider(provider)

# 导出到 Prometheus
metrics_exporter = PrometheusMetricsExporter()
processor = BatchSpanProcessor(metrics_exporter)
provider.add_span_processor(processor)

# 自动 instrumentation
RequestsInstrumentor().instrument()

2.2 自定义指标

from prometheus_client import Counter, Histogram, Gauge

# 请求计数器
LLM_REQUESTS = Counter(
    'llm_requests_total',
    'LLM request count',
    ['model', 'endpoint', 'status']
)

# 延迟直方图
LLM_LATENCY = Histogram(
    'llm_request_latency_seconds',
    'LLM request latency',
    buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
)

# 并发 Gauge
LLM_CONCURRENT = Gauge(
    'llm_concurrent_requests',
    'Current concurrent requests'
)

# Token 消耗计数器
LLM_TOKENS = Counter(
    'llm_tokens_total',
    'Token consumption',
    ['type']  # prompt, completion
)

2.3 使用示例

import time

@LLM_LATENCY.time()
def call_llm(prompt: str) -> str:
    LLM_REQUESTS.labels(model='qwen2.5', endpoint='/chat', status='success').inc()
    LLM_CONCURRENT.inc()

    start = time.time()
    result = llm.invoke(prompt)
    elapsed = time.time() - start

    # 记录 Token 消耗
    prompt_tokens = len(prompt.split()) // 2  # 近似
    completion_tokens = len(result.split()) // 2
    LLM_TOKENS.labels(type='prompt').inc(prompt_tokens)
    LLM_TOKENS.labels(type='completion').inc(completion_tokens)

    LLM_CONCURRENT.dec()
    return result

---

3. 质量监控

3.1 幻觉检测

from langchain.evaluation import load_evaluator

# 加载评估器
hallucination_evaluator = load_evaluator("labeled_hallucination")

def check_hallucination(question: str, context: str, answer: str) -> dict:
    """检测幻觉"""
    result = hallucination_evaluator.evaluate_strings(
        prediction=answer,
        input=question,
        reference=context
    )
    return {
        "is_hallucinated": result["score"] < 0.5,
        "confidence": result["score"],
        "reasoning": result["reasoning"]
    }

3.2 答案相关性评估

from langchain.evaluation import load_evaluator

relevance_evaluator = load_evaluator("qa")

def evaluate_relevance(question: str, answer: str) -> float:
    """评估答案相关性"""
    result = relevance_evaluator.evaluate_strings(
        input=question,
        prediction=answer
    )
    return result["score"]  # 0-1 分

3.3 用户反馈收集

@app.post("/feedback")
def submit_feedback(
    request_id: str,
    rating: int,  # 1-5
    comment: str = None
):
    """收集用户反馈"""
    # 存储到数据库
    db.execute(
        "INSERT INTO feedback (request_id, rating, comment, created_at) VALUES (?, ?, ?, ?)",
        request_id, rating, comment, datetime.now()
    )

    # 更新指标
    FEEDBACK_RATING.summary(rating).record()

    return {"status": "received"}

---

4. 成本监控

4.1 Token 成本计算

PRICING = {
    "gpt-4o": {"input": 2.50 / 1M, "output": 10.00 / 1M},
    "gpt-4o-mini": {"input": 0.15 / 1M, "output": 0.60 / 1M},
    "qwen2.5:7b": {"input": 0, "output": 0},  # 本地免费
}

def calculate_cost(model: str, prompt_tokens: int, completion_tokens: int) -> float:
    """计算成本"""
    if model not in PRICING:
        return 0

    pricing = PRICING[model]
    cost = (prompt_tokens * pricing["input"] + 
            completion_tokens * pricing["output"]) / 1_000_000
    return cost

4.2 成本告警

DAILY_BUDGET = 100.0  # 每日预算
cost_tracker = CostTracker()

@app.on_event("shutdown")
def check_budget():
    """关闭时检查预算"""
    daily_cost = cost_tracker.get_daily_cost()
    if daily_cost > DAILY_BUDGET:
        send_alert(f"今日 LLM 成本已达 ${daily_cost:.2f}，超出预算 ${DAILY_BUDGET}")

---

5. Grafana Dashboard

5.1 关键面板

面板	查询	展示方式
请求延迟分布	histogram_quantile(0.99, rate(llm_request_latency_seconds_bucket[5m]))	折线图
每秒请求数	rate(llm_requests_total[5m])	面积图
Token 消耗	rate(llm_tokens_total[5m])	柱状图
错误率	rate(llm_requests_total{status=~"4..|5.."}[5m]) / rate(llm_requests_total[5m])	仪表盘
日均成本	sum(rate(llm_cost_total[24h])) * 86400	数字面板

5.2 告警规则

groups:
  - name: llm_alerts
    rules:
      - alert: LLMHighLatency
        expr: histogram_quantile(0.99, rate(llm_request_latency_seconds_bucket[5m])) > 5
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "LLM P99 延迟超过 5 秒"

      - alert: LLMBadErrorRate
        expr: rate(llm_requests_total{status=~"5.."}[5m]) / rate(llm_requests_total[5m]) > 0.05
        for: 10m
        labels:
          severity: critical
        annotations:
          summary: "LLM 5xx 错误率超过 5%"

---

6. 日志与链路追踪

6.1 结构化日志

import logging
import uuid

logger = logging.getLogger(__name__)

def log_llm_call(prompt: str, response: str, latency: float):
    trace_id = str(uuid.uuid4())[:8]
    logger.info({
        "event": "llm_call",
        "trace_id": trace_id,
        "prompt_length": len(prompt),
        "response_length": len(response),
        "latency_ms": latency * 1000,
        "timestamp": datetime.now().isoformat()
    })

6.2 Jaeger 链路追踪

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# 配置 Jaeger 导出
jaeger_exporter = JaegerExporter(
    agent_endpoint="localhost:6831"
)

provider = TracerProvider()
provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))
trace.set_tracer_provider(provider)

---

7. 常见问题排查

问题 1：延迟突然升高

排查步骤： 1. 检查 GPU 利用率（nvidia-smi） 2. 查看并发请求数是否激增 3. 检查网络延迟（到 GPU 服务器） 4. 分析慢查询（哪些 prompt 最慢）

---

问题 2：Token 消耗异常

排查步骤： 1. 检查是否有超长 prompt 未截断 2. 查看历史消息是否累积过多 3. 分析用户输入模式变化 4. 设置 Token 上限告警

---

问题 3：质量下降

排查步骤： 1. 检查模型版本是否变更 2. 分析用户反馈趋势 3. 对比不同 prompt 的质量 4. 检查 RAG 知识库是否过期

---

8. 延伸阅读

• OpenTelemetry Python
• LangChain Evaluation
• Grafana LLM Dashboard

---

FAQ

LLM 应用需要监控哪些指标？ 核心四类：① 性能指标（延迟、吞吐、并发数）② 质量指标（Token 消耗、幻觉率、答案相关性）③ 成本指标（API 调用费用、Token 单价）④ 业务指标（请求成功率、用户满意度）。

如何检测 LLM 幻觉？ 双模型校验：用另一个 LLM 验证输出事实性；RAG 场景检查上下文来源是否匹配；设置置信度阈值，低于阈值标记为"可能幻觉"。

Token 消耗如何优化？ ① 减少历史消息长度 ② 用 Prompt 压缩 ③ 选择合适的模型大小 ④ 缓存相同问题的响应 ⑤ 开启 streaming 减少等待。

LLM 服务降级策略是什么？ 三级降级：① 主模型故障 → 切换备用模型 ② 备用故障 → 返回缓存答案 ③ 全故障 → 提示"服务暂不可用"。设置熔断阈值（错误率 > 10%）。

---