Skip to main content

06. 可观测性:Metrics、Logs、Traces、SLO

读完这篇你能:

  • 给 FastAPI 加结构化日志、Prometheus 指标、OpenTelemetry trace
  • 接 Grafana Cloud,配监控面板
  • 定义 SLO 和告警,系统出问题你比用户先知道
  • 监控 LLM 特有的指标(成本、token、反馈率)

对应架构层:可观测层

前置知识:01-05 篇

预计用时:3 小时(含动手)

0. 为什么算法工程师要学这个​

你部署了应用,开始有用户。然后某天老板说:

"用户反馈说应用很慢。"

你打开终端,看 docker logs,发现一堆日志混在一起,完全看不出哪条对哪个用户。你重启服务,好像好了,但第二天又慢。

问题:你的系统对你是个黑盒。你不知道:

  • 哪个接口慢?p99 延迟多少?
  • 错误率多少?是 0.1% 还是 50%?
  • LLM 成本一天烧多少?哪个用户最贵?
  • 一个请求经过哪些服务,卡在哪一段?

没有可观测性,生产事故只能"用户告诉你"。这就像训练模型不开 TensorBoard——你根本不知道训练发生了什么。

算法工程师类比:可观测性 = 生产系统的 TensorBoard + W&B + alert。没有它,你就是在盲飞。

1. 核心概念(最少必要理论)​

1.1 可观测性三支柱(Three Pillars)​

业界共识,可观测性由三部分组成:

支柱是什么回答什么问题工具
Metrics(指标)数字看板(QPS、延迟、错误率)系统健不健康?Prometheus、Grafana
Logs(日志)事件流(谁、什么时候、做了什么)出了什么问题?ELK、Loki
Traces(追踪)一个请求经过所有服务的链路慢在哪里?OpenTelemetry、Jaeger

三者协作:

告警触发(Grafana 看到 Metrics 异常)
   ↓
看哪些请求出错(Metrics → 服务 + 时间窗)
   ↓
查这些请求的 Logs(找错误堆栈)
   ↓
看 Trace 找瓶颈(是 LLM 慢?还是 DB?)

记忆法:

  • Metrics 告诉你有问题(数字异常)
  • Logs 告诉你出了什么错(详细信息)
  • Traces 告诉你问题在哪(精确定位)

1.2 为什么不能只用 print()​

算法工程师习惯 print()。生产环境完全不够:

问题print结构化日志
格式自由文本,难解析JSON,可按字段查询
关联没有 request_id有,跨服务追踪
级别没分级DEBUG/INFO/WARN/ERROR
聚合不能可入 ELK 集中查
上下文没 user_id / path有
# ❌ print
print(f"user {user_id} did something at {time}")
# 1 万条这样的日志混在一起,什么都看不出来

# ✅ 结构化日志
logger.info("user_action",
    user_id=user_id,
    action="summarize",
    duration_ms=1234,
    request_id=req_id)
# ELK 里能搜:user_action AND user_id=123,按 duration_ms 排序

1.3 指标(Metrics)类型​

Prometheus 定义四种指标类型:

类型用途例子
Counter只增不减请求总数、错误总数
Gauge可增可减当前连接数、队列长度
Histogram分布(分位数)延迟分布(p50/p90/p99)
Summary分位数(客户端算)同 Histogram,但服务端不能聚合

算法工程师类比:

  • Counter = epoch 数(只增)
  • Gauge = 当前 loss(可增可减)
  • Histogram = 不同 batch 大小的分布

关键分位数:

  • p50(中位数):50% 的请求比这快
  • p90:90% 的请求比这快
  • p99:99% 的请求比这快(慢请求的尾巴)
  • p99.9: Thousand

为什么看 p99 不看平均? 100 个请求,99 个 100ms + 1 个 10s——平均 200ms 看着挺好,但有 1 个用户体验极差。p99 = 10s,暴露问题。

1.4 SLO / SLA / SLI / Error Budget​

Google SRE 的概念,定义"系统够不够用"。

概念含义例子
SLI(Service Level Indicator)指标"成功请求数 / 总请求数"
SLO(Service Level Objective)目标"SLI ≥ 99.9%"(一个月内)
SLA(Service Level Agreement)合同承诺"SLI < 99%,退款"
Error Budget容忍的出错额度一个月允许 43 分钟不可用(99.9% 的反面)

为什么定义 SLO:

  • 给团队统一目标(不是"尽量稳定",而是"99.9%")
  • 平衡"稳定性"和"迭代速度"(Error Budget 用完,优先修 bug 不上新功能)
  • 给用户明确承诺

LLM 应用的常见 SLO:

  • 可用性 ≥ 99.5%(允许每月 3.6 小时宕机)
  • p99 延迟 < 5 秒(LLM 调用本来就慢)
  • 错误率 < 1%

1.5 告警设计原则​

反模式:

  • 告警太多 → 告警疲劳,关键告警被忽略
  • 告警太宽(只要错误率 > 0%)→ 半夜被叫醒
  • 告警没用("CPU 80%")→ 你知道了能干啥?

好告警的特征(可执行的):

  1. 基于 SLO(不是基于 CPU、内存等中间指标)
  2. 有明确动作(收到告警你知道该干啥)
  3. 分级:
    • P1(立即处理,会叫人):生产宕机、错误率 > 5%
    • P2(工作时间处理):p99 延迟涨了
    • P3(周报里看):日活下降

告警 vs 仪表盘:

  • 告警:异常时通知到人(电话、Slack)
  • 仪表盘:正常时人主动看

2. 实战:给应用加完整可观测性​

2.1 装 Prometheus 客户端 + 结构化日志 + OTel​

pip install prometheus-fastapi-instrumentator structlog opentelemetry-distro opentelemetry-instrumentation-fastapi opentelemetry-exporter-otlp
opentelemetry-bootstrap -a install

2.2 结构化日志(structlog)​

新建 logging_config.py:

import structlog
import logging

def setup_logging(env: str = "dev"):
    log_level = logging.DEBUG if env == "dev" else logging.INFO

    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer()  # 生产 JSON
            if env == "prod"
            else structlog.dev.ConsoleRenderer(),  # 开发彩色
        ],
        wrapper_class=structlog.make_filtering_bound_logger(log_level),
        cache_logger_on_first_use=True,
    )

logger = structlog.get_logger()

在 main.py 里:

from logging_config import setup_logging, logger
setup_logging(settings.env)

@app.middleware("http")
async def logging_middleware(request: Request, call_next):
    request_id = request.headers.get("X-Request-ID") or str(uuid.uuid4())
    ctx = structlog.contextvars.bind_contextvars(
        request_id=request_id,
        path=request.url.path,
        method=request.method,
    )
    start = time.time()
    try:
        response = await call_next(request)
        duration_ms = int((time.time() - start) * 1000)
        logger.info("request_completed",
            status=response.status_code,
            duration_ms=duration_ms)
        response.headers["X-Request-ID"] = request_id
        return response
    except Exception:
        logger.exception("request_failed")
        raise
    finally:
        ctx.clear()

所有 log 自动带 request_id、path、method,日志聚合工具能完美串联。

2.3 Prometheus 指标​

pip install prometheus-fastapi-instrumentator
from prometheus_fastapi_instrumentator import Instrumentator

Instrumentator().instrument(app).expose(app, endpoint="/metrics")

默认指标:

  • http_requests_total(Counter):请求总数,带 status、method、handler 标签
  • http_request_duration_seconds(Histogram):延迟分布
  • http_requests_in_progress(Gauge):当前并发

打开 http://localhost:8000/metrics 看到原始格式。

自定义业务指标:

from prometheus_client import Counter, Histogram, Gauge

# 业务指标
llm_tokens_used = Counter(
    "llm_tokens_used_total",
    "Total tokens consumed",
    ["model", "user_tier"]  # 标签
)

llm_request_duration = Histogram(
    "llm_request_duration_seconds",
    "LLM call duration",
    ["model"],
    buckets=[0.5, 1, 2, 5, 10, 30, 60]  # 自定义分桶
)

active_users = Gauge("active_users", "Currently active users")

# 在 summarize 里用
@app.post("/api/summarize")
async def summarize(req, user=Depends(get_current_user)):
    start = time.time()
    resp = await client.chat.completions.create(...)

    llm_tokens_used.labels(
        model="gpt-4o-mini",
        user_tier=user.tier
    ).inc(resp.usage.total_tokens)

    llm_request_duration.labels(model="gpt-4o-mini").observe(time.time() - start)
    ...

2.4 OpenTelemetry 分布式追踪​

OTel 是 CNCF 标准,厂商无关——你写一次,可以发到 Jaeger、Datadog、Honeycomb、Langfuse 等任何后端。

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor

# 配置 tracer
provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter(
    endpoint="http://localhost:4317"  # Jaeger / OTel collector
))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

# 自动 instrument FastAPI
FastAPIInstrumentor.instrument_app(app)

手动埋点(关键业务逻辑):

tracer = trace.get_tracer(__name__)

@app.post("/api/summarize")
async def summarize(req, user=Depends(get_current_user)):
    with tracer.start_as_current_span("summarize_endpoint") as span:
        span.set_attribute("user.id", user.id)
        span.set_attribute("request.text_length", len(req.text))

        with tracer.start_as_current_span("cache_lookup"):
            cached = cache_get(...)

        if not cached:
            with tracer.start_as_current_span("llm_call") as llm_span:
                resp = await client.chat.completions.create(...)
                llm_span.set_attribute("llm.tokens", resp.usage.total_tokens)

在 Jaeger UI 看到的就是一个请求的瀑布图:cache_lookup(2ms)→ llm_call(2300ms)→ db_write(50ms),一眼看出瓶颈。

2.5 LLM 专属:Langfuse​

LLM 应用有专属的可观测需求,通用 APM 不够:

  • prompt 内容(record + replay)
  • 完整对话上下文
  • 成本归因到用户
  • 模型对比

Langfuse(开源)是主流选择:

pip install langfuse
from langfuse import Langfuse
from langfuse.openai import openai  # ⚠️ 用 langfuse 包装的 openai

langfuse = Langfuse()

# 自动记录所有 OpenAI 调用,带 trace_id, user_id
resp = await openai.chat.completions.create(
    model="gpt-4o-mini",
    messages=[...],
    metadata={
        "trace_id": request_id,
        "user_id": user.id,
        "session_id": session_id,
    }
)

Langfuse UI 里能看到:

  • 每次调用的完整 prompt + response
  • token 数、成本
  • 延迟
  • 用户路径(一个会话内调了哪些接口)

2.6 用 Grafana Cloud 一站式接入​

自己搭 Prometheus + Grafana + Loki + Tempo 太麻烦。Grafana Cloud 有免费层,一键全搞定。

步骤:

  1. 注册 https://grafana.cloud
  2. 创建 stack,拿到:
    • Prometheus remote_write URL + 凭证
    • Loki URL + 凭证
    • Tempo URL + 凭证
  3. 把这些配置塞进应用:
# Prometheus remote write
from prometheus_client import CollectorRegistry, REGISTRY
from prometheus_client import start_http_server
# 通过 Grafana Agent / otel-collector 转发

# Loki 日志
import logging
from logging_loki import LokiHandler
handler = LokiHandler(
    url="https://logs-prod-xxx.grafana.net/loki/api/v1/push",
    tags={"app": "my-llm-app"},
    auth=("USER_ID", "API_KEY")
)
logger.addHandler(handler)

更简单的方案:用 Grafana Agent(一个二进制)收集所有指标/日志/trace,转发到 Grafana Cloud。

3. 进阶:监控 LLM 特有指标​

LLM 应用要监控的专属指标:

指标为什么告警阈值
Token 消耗速率防止单用户烧爆单用户每小时 > 100k token
成本(美元/天)预算控制日成本 > $100
缓存命中率缓存设计是否合理< 20% 告警
错误率LLM 服务不稳定(OpenAI 429/500)> 5%
反馈率(👍/👎)用户满意度差评率 > 10%
平均响应时长用户体验p99 > 10s
被护栏拦截的次数内容安全突然飙升可能是攻击

Token 成本看板(Grafana):

# 每小时成本
rate(llm_tokens_used_total[1h]) * 0.000001 * 1000
# (token 数 × 每千 token 价格)

# 按用户分组的日成本
sum by (user_id) (increase(llm_tokens_used_total[24h])) * 0.0000015

用户反馈监控:

feedback_rating = Histogram(
    "user_feedback_rating",
    "User feedback 1-5 stars",
    buckets=[1, 2, 3, 4, 5]
)

@app.post("/api/feedback")
def feedback(query_id: int, rating: int):
    feedback_rating.observe(rating)
    ...

4. 告警实战​

4.1 Grafana 告警规则​

在 Grafana → Alerting → New alert rule:

告警 1:错误率

# 5 分钟内 5xx 错误率
sum(rate(http_requests_total{status=~"5.."}[5m]))
/
sum(rate(http_requests_total[5m]))

阈值:> 0.05(5%),持续 5 分钟。

告警 2:p99 延迟

histogram_quantile(0.99,
  rate(http_request_duration_seconds_bucket[5m])
)

阈值:> 5 秒。

告警 3:LLM 成本

increase(llm_tokens_used_total[1h]) * 0.0000015

阈值:> 50($50/小时,可能是异常调用)。

告警 4:服务宕机

up{job="my-llm-app"}

阈值:== 0,持续 1 分钟。

4.2 告警通知渠道​

Grafana 支持:

  • Slack(团队频道)
  • 飞书 / 企业微信(国内)
  • PagerDuty / Opsgenie(电话叫醒)
  • 邮件

通知模板:

🚨 [{{ .Labels.alertname }}]
{{ .Annotations.summary }}
Value: {{ .Value }}
Runbook: {{ .Annotations.runbook_url }}

关键:每个告警都要有 runbook_url——指向"出问题怎么办"的文档。

4.3 降噪:告警分组和抑制​

  • 分组:同一服务的多个告警合并成一条通知
  • 抑制:父告警触发时,抑制子告警(数据库挂了,所有依赖它的服务都会报错,只报数据库那个就行)

5. 常见踩坑​

坑 1:日志太多,ELK 撑不住​

# ❌ 每个 token 都打日志
for token in stream:
    logger.info("token_generated", token=token)

症状:ES 索引爆炸,查询慢,成本飙升。

解决:

  • DEBUG 日志默认关掉
  • 高频事件用 Metrics(Counter)代替 Logs
  • 日志采样(只记 10% 的成功请求,所有错误都记)

坑 2:Trace 采样率设 100%,成本爆炸​

OTel 默认采样 100%,生产环境不行。

from opentelemetry.sdk.trace.sampling import TraceIdRatioBased

# 只采样 10%
processor = BatchSpanProcessor(
    OTLPSpanExporter(...),
    sampler=TraceIdRatioBased(0.1)
)

经验:错误请求 100% 采样,正常请求 1-10%。

坑 3:Metrics 标签太多(Cardinality Explosion)​

# ❌ 把 user_id 当标签 → 每个 user 一个时间序列
requests_total.labels(user_id=user.id).inc()

10 万用户 = 10 万时间序列,Prometheus 内存爆。

解决:

  • 标签只用低基数值(status、method、tier)
  • user_id 这种高基数值放 Span / Log 里

坑 4:告警没 runbook​

收到告警:"p99 > 10s",然后呢?

解决:每个告警配 runbook_url,指向 wiki / Confluence 页面,说明:

  • 怎么诊断
  • 临时缓解
  • 长期修复

坑 5:监控只看平均​

"平均延迟 500ms"——但 p99 是 10 秒,1% 的用户体验灾难。

永远用分位数(p50/p90/p99),不要用平均。

坑 6:PII 泄露到日志​

# ❌ 用户邮箱、身份证号、密码出现在日志
logger.info("login_attempt", email=user.email, password=password)

解决:

  • 日志里脱敏(只记 user_id)
  • 用专门的审计日志表存敏感操作
  • 启用日志扫描(GCP Sensitive Data Protection)

6. 检查清单​

  • 我能解释 Metrics / Logs / Traces 各解决什么问题
  • 我用 structlog 替换了 print
  • 我的日志带 request_id,能串联一个请求
  • 我接了 Prometheus,/metrics 端点能访问
  • 我有 Grafana 看板,能看到 QPS、延迟、错误率
  • 我用 OTel 加了分布式追踪
  • 我接了 Langfuse 记录 LLM 调用
  • 我定义了 SLO(可用性、延迟、错误率)
  • 我配了 5 个核心告警,都通到 Slack / 飞书
  • 我的告警都有 runbook
  • 我监控了 LLM 成本和 token 消耗
  • 我没用高基数标签

7. 下一步​

可观测层搞定!现在你的系统是"透明"的,任何异常你比用户先知道。

下一站去接入层 + 协调层 + 部署层进阶:

学到那,你就能扛住百万日活了。

进阶学习资源​