AI 应用监控告警：生产环境的健康保障

普通 Web 服务的监控关注 HTTP 状态码、响应时间、QPS。AI 应用需要在此基础上监控一批特有指标：LLM API 的调用成功率、token 消耗趋势、Embedding 延迟、RAG 检索质量、以及 Agent 的工具调用失败率。这些指标和成本直接挂钩，也是定位问题的关键线索。

没有监控的 AI 应用是黑盒。上线之后发生了什么，只有等用户投诉才知道。本章介绍如何用 Prometheus（开源监控系统，负责收集和存储各类指标数据）+ Grafana（开源可视化平台，将 Prometheus 数据绘制成图表和仪表盘）构建完整的 AI 应用监控告警体系。ELK（Elasticsearch + Logstash + Kibana，一套专门用于日志收集、存储、搜索和可视化的开源工具栈）则常用于日志分析场景。

1.1 需要监控什么

AI应用监控体系图
AI应用监控告警体系——指标采集、存储聚合、可视化告警、通知响应的完整链路

AI 应用的监控指标分四个层次：

基础设施层：CPU、内存这些常规指标，GPU 利用率（如果有本地模型）。

应用服务层：FastAPI 的 HTTP 指标，Celery Worker 的任务队列状态。这些和普通 Web 服务一样，Prometheus 有现成的 Exporter。

AI 业务层：这是 AI 应用特有的，需要自己埋点。LLM API 的成功率和延迟波动，往往比你的服务本身更不稳定。OpenAI 有时会出现间歇性超时，自己不监控就不知道是 API 问题还是自己的 bug。

AI 特有的监控指标（这是和普通 Web 服务最大的区别）：

首 Token 延迟（TTFT, Time to First Token）：用户感知的响应速度，比总延迟更重要。即使完整响应需要 30 秒，如果 1 秒内出了第一个字，用户感觉是"快速响应"。
Token 消耗量：每次调用消耗多少 input/output token，以及随时间的变化趋势。突然激增可能是 Prompt 膨胀、上下文管理出问题。
RAG 检索质量：检索命中率（有多少次检索到了相关文档）、检索召回率（相关文档有没有被召回）、答案相关性分数。
幻觉率：用户明确标记为"答案不对"或"编造了不存在信息"的比率。这个需要用户反馈来度量。
缓存命中率：相似查询命中缓存的比例，直接反映成本节省效果。
工具调用失败率（针对 Agent）：Agent 调用外部工具时的错误率，高错误率说明工具集成有问题或 LLM 的工具调用质量下降。

成本层：token 消耗和成本是最终关心的业务指标。一个异常的 prompt 模板可能让 token 消耗翻倍，必须及时发现。

1.2 Prometheus + Grafana 方案

Prometheus 定期抓取应用暴露的 /metrics 端点，将指标存为时序数据。Grafana 连接 Prometheus 画图表。Alertmanager 负责告警路由和去重。

1.3 在 FastAPI 中集成 Prometheus 指标

bash

pip install prometheus-client prometheus-fastapi-instrumentator

python

# metrics.py
from prometheus_client import Counter, Histogram, Gauge, Summary
import time

# Counter：只增不减，适合统计调用次数、成功/失败次数
llm_requests_total = Counter(
    "llm_requests_total",
    "LLM API 调用总次数",
    # label 用于区分不同维度，Prometheus 会为每种 label 组合单独存储一条时序
    labelnames=["model", "status"],  # status: success / error / timeout
)

llm_tokens_total = Counter(
    "llm_tokens_total",
    "LLM Token 消耗总数",
    labelnames=["model", "token_type"],  # token_type: prompt / completion
)

# Histogram：记录值的分布，自动计算 P50/P95/P99
# 桶（bucket）的范围应覆盖实际延迟区间，这里按 LLM 特性设置较大范围
llm_latency_seconds = Histogram(
    "llm_latency_seconds",
    "LLM API 调用延迟（秒）",
    labelnames=["model"],
    buckets=[0.5, 1, 2, 5, 10, 20, 30, 60, 120],
)

# Gauge：可增可减，适合当前状态值（队列长度、活跃连接数）
celery_queue_length = Gauge(
    "celery_queue_length",
    "Celery 任务队列中等待执行的任务数",
    labelnames=["queue_name"],
)

rag_cache_hits = Counter(
    "rag_cache_hits_total",
    "RAG 缓存命中次数",
    labelnames=["cache_type"],  # exact / semantic / miss
)

llm_cost_dollars = Counter(
    "llm_cost_dollars_total",
    "LLM API 调用估算成本（美元）",
    labelnames=["model"],
)

# 每种模型的 token 价格（美元/百万 token）
TOKEN_PRICES = {
    "gpt-4o": {"input": 2.5, "output": 10.0},
    "gpt-4o-mini": {"input": 0.15, "output": 0.60},
    "gpt-4-turbo": {"input": 10.0, "output": 30.0},
}

python

# main.py
from fastapi import FastAPI, Request
from prometheus_fastapi_instrumentator import Instrumentator
from prometheus_client import make_asgi_app
from metrics import (
    llm_requests_total, llm_tokens_total, llm_latency_seconds,
    llm_cost_dollars, rag_cache_hits, TOKEN_PRICES
)
import time
from openai import OpenAI

app = FastAPI()
client = OpenAI()

# 自动为所有 HTTP 路由添加 Prometheus 指标（延迟、QPS、状态码分布）
Instrumentator().instrument(app).expose(app)

# 挂载 Prometheus metrics 端点（Prometheus 会定期来抓取这个端点）
metrics_app = make_asgi_app()
app.mount("/metrics", metrics_app)


def monitored_llm_call(model: str, messages: list, **kwargs) -> dict:
    """
    带监控的 LLM 调用封装。
    所有需要调 LLM 的地方都通过这个函数，保证指标收集的一致性。
    """
    start_time = time.time()
    status = "success"

    try:
        response = client.chat.completions.create(
            model=model,
            messages=messages,
            **kwargs,
        )

        # 记录 token 消耗
        usage = response.usage
        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)

        # 估算成本并记录
        if model in TOKEN_PRICES:
            price = TOKEN_PRICES[model]
            cost = (
                usage.prompt_tokens * price["input"] +
                usage.completion_tokens * price["output"]
            ) / 1_000_000  # 价格单位是每百万 token
            llm_cost_dollars.labels(model=model).inc(cost)

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

    except Exception as e:
        # 区分超时和其他错误，便于告警规则精细化
        if "timeout" in str(e).lower():
            status = "timeout"
        else:
            status = "error"
        raise

    finally:
        # 无论成功失败都记录延迟和请求计数
        elapsed = time.time() - start_time
        llm_latency_seconds.labels(model=model).observe(elapsed)
        llm_requests_total.labels(model=model, status=status).inc()


@app.post("/api/ask")
async def ask(request: Request):
    body = await request.json()
    question = body.get("question")

    # 记录缓存命中情况（实际项目中在缓存层记录）
    rag_cache_hits.labels(cache_type="miss").inc()

    result = monitored_llm_call(
        model="gpt-4o",
        messages=[{"role": "user", "content": question}],
    )
    return result

1.4 定期采集异步指标

Celery 队列长度、Redis 内存等需要主动采集，不是请求触发的指标。用后台线程定期更新 Gauge：

python

# metrics_collector.py
import threading
import time
import redis
from metrics import celery_queue_length

redis_client = redis.Redis(host="localhost", port=6379, db=0)


def collect_celery_metrics():
    """
    每 15 秒采集一次 Celery 队列长度。
    队列积压是 Worker 处理能力不足的早期信号，应该在用户感知到慢之前就告警。
    """
    while True:
        try:
            for queue_name in ["high_priority", "low_priority", "default"]:
                length = redis_client.llen(queue_name)
                celery_queue_length.labels(queue_name=queue_name).set(length)
        except Exception as e:
            print(f"采集队列指标失败: {e}")
        time.sleep(15)


# 在应用启动时在后台线程中运行
collector_thread = threading.Thread(target=collect_celery_metrics, daemon=True)
collector_thread.start()

1.5 告警规则设计

在 Prometheus 的告警规则文件中定义：

yaml

# ai_alerts.yml
groups:
  - name: ai_application
    rules:
      # LLM 错误率超过 5%（5 分钟内）
      - alert: LLMHighErrorRate
        expr: |
          rate(llm_requests_total{status=~"error|timeout"}[5m])
          /
          rate(llm_requests_total[5m]) > 0.05
        for: 2m  # 持续 2 分钟才触发，避免瞬间抖动误报
        labels:
          severity: warning
        annotations:
          summary: "LLM API 错误率过高"
          description: "模型 {{ $labels.model }} 的错误率达到 {{ $value | humanizePercentage }}"

      # P99 延迟超过 30 秒
      - alert: LLMHighLatency
        expr: histogram_quantile(0.99, rate(llm_latency_seconds_bucket[5m])) > 30
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "LLM 响应延迟过高"
          description: "P99 延迟达到 {{ $value }}s，可能影响用户体验"

      # 每小时 Token 消耗异常（超过基线的 3 倍）
      - alert: TokenConsumptionAnomaly
        expr: |
          increase(llm_tokens_total[1h]) >
          3 * avg_over_time(increase(llm_tokens_total[1h])[24h:1h])
        for: 10m
        labels:
          severity: critical
        annotations:
          summary: "Token 消耗异常激增"
          description: "当前小时 Token 消耗是过去 24 小时均值的 3 倍以上，请排查是否有异常请求"

      # Celery 队列积压超过 100 个任务
      - alert: CeleryQueueBacklog
        expr: celery_queue_length{queue_name="high_priority"} > 100
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "高优先级队列积压"
          description: "队列 {{ $labels.queue_name }} 中有 {{ $value }} 个任务等待处理"

1.6 LangSmith / Langfuse：AI 专项可观测性

Prometheus 擅长数值指标，但看不到 AI 应用最重要的内容：每次对话的 prompt、输出、中间推理步骤。LangSmith 和 Langfuse 专门为 LLM 应用设计，提供完整的调用链追踪。

python

# 使用 Langfuse 追踪 LLM 调用链（开源可自部署）
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context

langfuse = Langfuse(
    public_key="your_public_key",
    secret_key="your_secret_key",
    host="https://cloud.langfuse.com",  # 或自部署地址
)


@observe()  # 自动追踪函数的输入输出，记录到 Langfuse
def rag_pipeline(question: str) -> str:
    """
    @observe 装饰器会自动：
    1. 记录函数的调用时间和耗时
    2. 记录输入参数和返回值
    3. 追踪内部的 LLM 调用（需要使用 Langfuse 封装的 client）
    """
    # 标记当前 trace 的用户（用于分析不同用户的使用模式）
    langfuse_context.update_current_trace(
        user_id="user_123",
        tags=["rag", "production"],
    )

    # 检索和生成步骤
    docs = retrieve_relevant_docs(question)
    answer = generate_answer(question, docs)

    # 手动为这次回答打分（也可以让用户打分后异步上报）
    langfuse_context.score_current_trace(
        name="relevance",
        value=0.9,
        comment="检索到的文档与问题高度相关",
    )

    return answer

Langfuse 的核心价值在于"时光机"：某个用户反馈答案不对，可以立刻找到那次对话的完整 trace，看到 prompt 是什么、检索到了哪些文档、LLM 的原始输出是什么，精准定位问题所在。

1.7 结构化日志

结构化日志是监控的补充，适合排查具体错误，而 Prometheus 适合看趋势。

python

# structured_logging.py
import logging
import json
import time
from functools import wraps


class JSONFormatter(logging.Formatter):
    """输出 JSON 格式的日志，便于 ELK/Loki 采集和查询。"""

    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime(record.created)),
            "level": record.levelname,
            "logger": record.name,
            "message": record.getMessage(),
        }
        # 把通过 extra 传入的自定义字段合并进来
        if hasattr(record, "extra"):
            log_data.update(record.extra)
        return json.dumps(log_data, ensure_ascii=False)


def setup_logger(name: str) -> logging.Logger:
    logger = logging.getLogger(name)
    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)
    return logger


ai_logger = setup_logger("ai_service")


def log_llm_call(func):
    """装饰器：自动记录 LLM 调用的结构化日志。"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        try:
            result = func(*args, **kwargs)
            ai_logger.info(
                "LLM call succeeded",
                extra={
                    "extra": {
                        "duration_ms": round((time.time() - start) * 1000),
                        "model": kwargs.get("model"),
                        "tokens": result.get("usage", {}).get("total_tokens"),
                    }
                },
            )
            return result
        except Exception as e:
            ai_logger.error(
                "LLM call failed",
                extra={
                    "extra": {
                        "duration_ms": round((time.time() - start) * 1000),
                        "model": kwargs.get("model"),
                        "error_type": type(e).__name__,
                        "error_message": str(e),
                    }
                },
            )
            raise
    return wrapper

1.8 方案对比

方案	擅长	不擅长	成本
Prometheus + Grafana	数值趋势、告警	日志内容、对话详情	免费，需运维
LangSmith	完整调用链追踪、Prompt 调试	基础设施指标	按 trace 收费
Langfuse	同 LangSmith，可自部署	基础设施指标	开源免费
Datadog LLM Observability	一站式，无需搭建	成本较高	按用量收费
ELK Stack	日志检索、全文搜索	时序指标	免费，运维重

生产环境推荐组合：Prometheus + Grafana（基础指标和告警）+ Langfuse 自部署（AI 调用链追踪），两者互补，覆盖所有监控需求。

1.9 小结

AI 应用的监控不是可选项，而是上线的前提条件。核心指标按优先级：LLM API 错误率 > Token 消耗趋势 > P99 延迟 > 缓存命中率 > 队列积压。

告警规则要避免两个极端：阈值太低导致告警风暴，阈值太高导致等用户投诉才发现。使用 for 字段要求持续触发才告警（而不是瞬间抖动），并且区分 warning 和 critical 级别，critical 才需要人工立即介入。

到这里，AI 服务的生产化部署基本齐全了：API 服务、流式输出、可观测性、评估、成本控制、错误处理、Docker 部署、安全、异步任务、缓存、监控告警，每一块都有了。