AI应用性能监控与告警完全教程

1. AI应用监控概述与挑战

传统Web应用的监控体系已经非常成熟——CPU、内存、磁盘、网络、请求延迟、错误率，这些指标足以覆盖绝大多数场景。但AI应用引入了全新的监控维度，传统指标远远不够。

AI应用特有的监控挑战：

• 非确定性输出：同一个请求可能产生不同结果，"错误"的定义变得模糊
• 推理成本高昂：一次LLM调用可能消耗数千Token，直接影响费用
• 延迟分布异常：首Token延迟(TTFT)与总完成时间(TCT)是两个完全不同的体验指标
• 质量难以量化：模型输出是否"正确"需要语义层面的评估，而非简单的状态码
• 长链路依赖：RAG、Agent等架构涉及多次模型调用、外部检索、工具执行，链路复杂

一个完整的AI应用监控体系需要覆盖五个层次：

┌─────────────────────────────────────┐
│         用户体验层                    │
│   满意度 / 反馈 / 端到端延迟          │
├─────────────────────────────────────┤
│         模型质量层                    │
│   幻觉率 / 相关性 / 准确性            │
├─────────────────────────────────────┤
│         推理性能层                    │
│   TTFT / TCT / 吞吐量 / 队列等待      │
├─────────────────────────────────────┤
│         资源与成本层                  │
│   Token用量 / GPU利用率 / API费用     │
├─────────────────────────────────────┤
│         基础设施层                    │
│   CPU / 内存 / 网络 / 磁盘            │
└─────────────────────────────────────┘

2. LLM推理延迟与吞吐量监控

LLM推理延迟是用户体验最直接的感知指标。与传统API不同，LLM输出是流式的，需要区分多个延迟阶段。

关键延迟指标：

Python埋点示例（基于OpenAI兼容接口）：

import time
import asyncio
from dataclasses import dataclass
from prometheus_client import Histogram, Counter, Gauge

# 定义Prometheus指标
llm_request_duration = Histogram(
    'llm_request_duration_seconds',
    'Total LLM request duration',
    ['model', 'status'],
    buckets=[0.1, 0.5, 1, 2, 5, 10, 30, 60, 120]
)

llm_ttft = Histogram(
    'llm_time_to_first_token_seconds',
    'Time to first token in streaming mode',
    ['model'],
    buckets=[0.05, 0.1, 0.2, 0.5, 1, 2, 5, 10]
)

llm_tps = Histogram(
    'llm_tokens_per_second',
    'Token generation speed',
    ['model'],
    buckets=[5, 10, 20, 50, 100, 200, 500]
)

llm_tokens_total = Counter(
    'llm_tokens_total',
    'Total tokens consumed',
    ['model', 'type']  # type: prompt / completion
)

llm_active_requests = Gauge(
    'llm_active_requests',
    'Number of currently active LLM requests',
    ['model']
)


@dataclass
class LLMCallMetrics:
    request_start: float
    first_token_time: float = 0.0
    completion_time: float = 0.0
    prompt_tokens: int = 0
    completion_tokens: int = 0


async def tracked_llm_call(client, model: str, messages: list, **kwargs):
    """带完整监控埋点的LLM调用封装"""
    metrics = LLMCallMetrics(request_start=time.monotonic())
    llm_active_requests.labels(model=model).inc()

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

        full_content = []
        async for chunk in response:
            if chunk.choices and chunk.choices[0].delta.content:
                if metrics.first_token_time == 0.0:
                    metrics.first_token_time = time.monotonic()
                    llm_ttft.labels(model=model).observe(
                        metrics.first_token_time - metrics.request_start
                    )
                full_content.append(chunk.choices[0].delta.content)

        metrics.completion_time = time.monotonic()
        total_duration = metrics.completion_time - metrics.request_start

        # 记录总延迟
        llm_request_duration.labels(model=model, status='success').observe(total_duration)

        # 计算并记录TPS
        if hasattr(response, 'usage') and response.usage:
            metrics.completion_tokens = response.usage.completion_tokens
            metrics.prompt_tokens = response.usage.prompt_tokens
        else:
            # 估算Token数（简化处理）
            metrics.completion_tokens = len(''.join(full_content)) // 4

        if metrics.completion_tokens > 0 and total_duration > 0:
            tps = metrics.completion_tokens / total_duration
            llm_tps.labels(model=model).observe(tps)

        # 记录Token用量
        llm_tokens_total.labels(model=model, type='prompt').inc(metrics.prompt_tokens)
        llm_tokens_total.labels(model=model, type='completion').inc(metrics.completion_tokens)

        return ''.join(full_content)

    except Exception as e:
        llm_request_duration.labels(model=model, status='error').observe(
            time.monotonic() - metrics.request_start
        )
        raise
    finally:
        llm_active_requests.labels(model=model).dec()

自托管模型的vLLM监控：

vLLM内置了Prometheus指标导出，配置后可直接采集：

# 启动vLLM时开启Prometheus指标
python -m vllm.entrypoints.openai.api_server \
    --model meta-llama/Llama-3-8B-Instruct \
    --enable-metrics \
    --metrics-port 9090

vLLM暴露的关键指标包括：

• vllm:request_success — 请求成功计数
• vllm:time_to_first_token_seconds — TTFT分布
• vllm:e2e_request_latency_seconds — 端到端延迟
• vllm:num_requests_running — 正在运行的请求数
• vllm:gpu_cache_usage_perc — GPU缓存使用率

3. Token使用量与成本追踪

Token是AI应用的基本计费单位。精确的Token追踪是成本控制的前提。

构建Token成本追踪中间件：

import datetime
from decimal import Decimal
from prometheus_client import Counter
import redis

# 每个模型的单价（美元/百万Token）
MODEL_PRICING = {
    'gpt-4o': {'input': Decimal('2.50'), 'output': Decimal('10.00')},
    'gpt-4o-mini': {'input': Decimal('0.15'), 'output': Decimal('0.60')},
    'claude-3-5-sonnet': {'input': Decimal('3.00'), 'output': Decimal('15.00')},
    'deepseek-v3': {'input': Decimal('0.27'), 'output': Decimal('1.10')},
}

token_cost_total = Counter(
    'llm_cost_usd_total',
    'Total LLM API cost in USD',
    ['model', 'cost_type', 'team']
)

r = redis.Redis(host='localhost', port=6379, decode_responses=True)


def record_token_usage(
    model: str,
    prompt_tokens: int,
    completion_tokens: int,
    team: str = 'default',
    request_id: str = ''
):
    """记录Token使用量并计算成本"""
    pricing = MODEL_PRICING.get(model)
    if not pricing:
        return

    input_cost = (Decimal(prompt_tokens) / Decimal('1000000')) * pricing['input']
    output_cost = (Decimal(completion_tokens) / Decimal('1000000')) * pricing['output']
    total_cost = input_cost + output_cost

    # Prometheus指标
    token_cost_total.labels(model=model, cost_type='input', team=team).inc(float(input_cost))
    token_cost_total.labels(model=model, cost_type='output', team=team).inc(float(output_cost))

    # 写入Redis用于成本报表
    today = datetime.date.today().isoformat()
    cost_key = f"llm:cost:{today}:{team}"
    r.hincrby(cost_key, 'total_tokens', prompt_tokens + completion_tokens)
    r.hincrbyfloat(cost_key, 'total_cost_usd', float(total_cost))
    r.hincrby(cost_key, 'request_count', 1)
    r.expire(cost_key, 90 * 86400)  # 保留90天

    # 按模型细分
    model_key = f"llm:cost:{today}:{team}:{model}"
    r.hincrby(model_key, 'prompt_tokens', prompt_tokens)
    r.hincrby(model_key, 'completion_tokens', completion_tokens)
    r.hincrbyfloat(model_key, 'cost_usd', float(total_cost))
    r.expire(model_key, 90 * 86400)

    return float(total_cost)


def get_daily_cost_report(team: str, date: str = None) -> dict:
    """获取每日成本报告"""
    date = date or datetime.date.today().isoformat()
    cost_key = f"llm:cost:{date}:{team}"
    data = r.hgetall(cost_key)
    return {
        'date': date,
        'team': team,
        'total_tokens': int(data.get('total_tokens', 0)),
        'total_cost_usd': round(float(data.get('total_cost_usd', 0)), 4),
        'request_count': int(data.get('request_count', 0)),
    }

4. 模型质量监控

性能指标只能告诉你"系统快不快"，质量指标才能告诉你"系统好不好"。

幻觉检测方案：

from enum import Enum
from pydantic import BaseModel
import openai

class QualityDimension(str, Enum):
    HALLUCINATION = 'hallucination'
    RELEVANCE = 'relevance'
    FAITHFULNESS = 'faithfulness'
    COMPLETENESS = 'completeness'

class QualityScore(BaseModel):
    dimension: QualityDimension
    score: float  # 0.0 - 1.0
    reasoning: str

class ModelQualityEvaluator:
    """使用LLM-as-Judge进行质量评估"""

    def __init__(self, judge_model: str = 'gpt-4o-mini'):
        self.judge_model = judge_model
        self.client = openai.AsyncOpenAI()

    async def evaluate_hallucination(
        self, question: str, answer: str, context: str = ''
    ) -> QualityScore:
        """评估回答中是否存在幻觉"""
        prompt = f"""你是一个严格的质量评估专家。判断以下回答是否存在幻觉（即编造不存在的信息）。

问题：{question}

参考上下文：{context if context else '无参考上下文'}

待评估回答：{answer}

评估标准：
- 0.0: 严重幻觉，大量编造信息
- 0.3: 较多幻觉，部分信息不准确
- 0.5: 少量幻觉，大部分信息正确
- 0.8: 基本准确，仅有微小偏差
- 1.0: 完全准确，无任何幻觉

请以JSON格式返回：{{"score": <0-1>, "reasoning": "<评估理由>"}}"""

        response = await self.client.chat.completions.create(
            model=self.judge_model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0
        )

        result = QualityScore(
            dimension=QualityDimension.HALLUCINATION,
            **eval(response.choices[0].message.content)
        )
        return result

    async def evaluate_relevance(self, question: str, answer: str) -> QualityScore:
        """评估回答与问题的相关性"""
        prompt = f"""评估以下回答与问题的相关性。

问题：{question}
回答：{answer}

评分标准：
- 0.0: 完全无关
- 0.5: 部分相关
- 1.0: 高度相关，直接回答了问题

返回JSON：{{"score": <0-1>, "reasoning": "<理由>"}}"""

        response = await self.client.chat.completions.create(
            model=self.judge_model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
            temperature=0
        )

        return QualityScore(
            dimension=QualityDimension.RELEVANCE,
            **eval(response.choices[0].message.content)
        )


# 采样评估（不评估每一条，按比例采样以控制成本）
import random
from prometheus_client import Histogram

quality_hallucination_score = Histogram(
    'llm_quality_hallucination_score',
    'Hallucination quality score',
    ['model'],
    buckets=[0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
)

quality_relevance_score = Histogram(
    'llm_quality_relevance_score',
    'Relevance quality score',
    ['model'],
    buckets=[0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
)

SAMPLE_RATE = 0.05  # 5%采样率


async def maybe_evaluate_quality(
    evaluator: ModelQualityEvaluator,
    model: str,
    question: str,
    answer: str,
    context: str = ''
):
    """按采样率触发质量评估"""
    if random.random() > SAMPLE_RATE:
        return

    hallucination = await evaluator.evaluate_hallucination(question, answer, context)
    relevance = await evaluator.evaluate_relevance(question, answer)

    quality_hallucination_score.labels(model=model).observe(hallucination.score)
    quality_relevance_score.labels(model=model).observe(relevance.score)

    # 低分告警
    if hallucination.score < 0.3:
        await send_quality_alert(model, question, answer, hallucination)

5. 用户满意度与反馈收集

技术指标之外，用户反馈是质量的最终验证。

反馈收集API设计：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from enum import Enum
import redis
import json
from prometheus_client import Counter, Histogram

app = FastAPI()
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

class FeedbackType(str, Enum):
    THUMBS_UP = 'thumbs_up'
    THUMBS_DOWN = 'thumbs_down'
    RATING = 'rating'  # 1-5星

class FeedbackRequest(BaseModel):
    request_id: str
    feedback_type: FeedbackType
    rating: int | None = None  # 1-5
    comment: str | None = None
    category: str | None = None  # 'inaccurate', 'harmful', 'irrelevant', 'other'

# Prometheus指标
user_feedback_total = Counter(
    'llm_user_feedback_total',
    'Total user feedback',
    ['model', 'feedback_type', 'category']
)

user_satisfaction_ratio = Histogram(
    'llm_user_satisfaction_ratio',
    'User satisfaction ratio (positive / total)',
    ['model'],
    buckets=[0, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]
)


@app.post('/feedback')
async def submit_feedback(feedback: FeedbackRequest):
    # 从请求日志中获取原始请求信息
    request_info = r.hgetall(f'llm:request:{feedback.request_id}')
    if not request_info:
        raise HTTPException(404, 'Request not found')

    model = request_info.get('model', 'unknown')

    # 存储反馈
    feedback_data = {
        **feedback.model_dump(),
        'model': model,
        'question': request_info.get('question', ''),
        'answer': request_info.get('answer', ''),
    }
    r.lpush('llm:feedback:queue', json.dumps(feedback_data))
    r.hincrby(f'llm:feedback:stats:{model}', feedback.feedback_type.value, 1)
    if feedback.category:
        r.hincrby(f'llm:feedback:stats:{model}', f'cat:{feedback.category}', 1)

    # 更新Prometheus
    user_feedback_total.labels(
        model=model,
        feedback_type=feedback.feedback_type.value,
        category=feedback.category or 'none'
    ).inc()

    # 计算满意度
    stats = r.hgetall(f'llm:feedback:stats:{model}')
    positive = int(stats.get('thumbs_up', 0))
    total = positive + int(stats.get('thumbs_down', 0))
    if total > 0:
        user_satisfaction_ratio.labels(model=model).observe(positive / total)

    return {'status': 'ok'}

前端反馈组件（React）：

import { useState } from 'react';

interface FeedbackWidgetProps {
  requestId: string;
  apiUrl: string;
}

export function FeedbackWidget({ requestId, apiUrl }: FeedbackWidgetProps) {
  const [submitted, setSubmitted] = useState(false);
  const [showDetail, setShowDetail] = useState(false);

  const sendFeedback = async (type: string, category?: string) => {
    await fetch(`${apiUrl}/feedback`, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify({
        request_id: requestId,
        feedback_type: type,
        category,
      }),
    });
    setSubmitted(true);
  };

  if (submitted) {
    return <span className="text-sm text-green-600">感谢反馈！</span>;
  }

  return (
    <div className="flex gap-2 items-center">
      <button onClick={() => sendFeedback('thumbs_up')} className="hover:text-green-600">👍</button>
      <button onClick={() => setShowDetail(!showDetail)} className="hover:text-red-600">👎</button>
      {showDetail && (
        <div className="flex gap-1 text-xs">
          <button onClick={() => sendFeedback('thumbs_down', 'inaccurate')}>不准确</button>
          <button onClick={() => sendFeedback('thumbs_down', 'irrelevant')}>不相关</button>
          <button onClick={() => sendFeedback('thumbs_down', 'harmful')}>有害</button>
        </div>
      )}
    </div>
  );
}

6. Prometheus + Grafana 监控方案

Prometheus配置（prometheus.yml）：

global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - 'ai_app_alerts.yml'

scrape_configs:
  # AI应用服务
  - job_name: 'ai-app'
    static_configs:
      - targets: ['ai-app:8000']
    metrics_path: '/metrics'

  # vLLM推理服务
  - job_name: 'vllm'
    static_configs:
      - targets: ['vllm-server:9090']
    metrics_path: '/metrics'

  # 向量数据库
  - job_name: 'qdrant'
    static_configs:
      - targets: ['qdrant:6333']
    metrics_path: '/metrics'

  # Redis缓存
  - job_name: 'redis'
    static_configs:
      - targets: ['redis-exporter:9121']

Docker Compose编排：

version: '3.8'
services:
  prometheus:
    image: prom/prometheus:latest
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml
      - ./ai_app_alerts.yml:/etc/prometheus/ai_app_alerts.yml
      - prometheus_data:/prometheus
    ports:
      - "9091:9090"
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--storage.tsdb.retention.time=30d'

  grafana:
    image: grafana/grafana:latest
    volumes:
      - grafana_data:/var/lib/grafana
      - ./grafana/dashboards:/etc/grafana/provisioning/dashboards
      - ./grafana/datasources:/etc/grafana/provisioning/datasources
    ports:
      - "3000:3000"
    environment:
      - GF_SECURITY_ADMIN_PASSWORD=admin

  alertmanager:
    image: prom/alertmanager:latest
    volumes:
      - ./alertmanager.yml:/etc/alertmanager/alertmanager.yml
    ports:
      - "9093:9093"

volumes:
  prometheus_data:
  grafana_data:

7. 自定义指标与告警规则

告警规则文件（ai_app_alerts.yml）：

groups:
  - name: ai_app_performance
    rules:
      # TTFT过高告警
      - alert: HighTTFT
        expr: histogram_quantile(0.95, rate(llm_time_to_first_token_seconds_bucket[5m])) > 5
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "P95 TTFT超过5秒"
          description: "当前P95 TTFT为 {{ $value }}秒，影响用户体验"

      # 推理吞吐量下降
      - alert: LowThroughput
        expr: rate(llm_tokens_per_second_sum[5m]) / rate(llm_tokens_per_second_count[5m]) < 10
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "推理吞吐量异常下降"
          description: "平均TPS降至 {{ $value }}，可能需要检查GPU负载"

      # 错误率告警
      - alert: HighLLMErrorRate
        expr: |
          rate(llm_request_duration_seconds_count{status="error"}[5m])
          / rate(llm_request_duration_seconds_count[5m]) > 0.05
        for: 3m
        labels:
          severity: critical
        annotations:
          summary: "LLM请求错误率超过5%"
          description: "当前错误率 {{ $value | humanizePercentage }}"

      # Token成本告警（日预算）
      - alert: DailyCostExceeded
        expr: increase(llm_cost_usd_total[24h]) > 100
        labels:
          severity: critical
        annotations:
          summary: "日Token成本超过$100预算"
          description: "过去24小时成本 ${{ $value }}"

      # 幻觉率告警
      - alert: HighHallucinationRate
        expr: |
          histogram_quantile(0.5, rate(llm_quality_hallucination_score_bucket[1h])) < 0.5
        for: 30m
        labels:
          severity: warning
        annotations:
          summary: "模型幻觉率偏高"
          description: "中位幻觉评分降至 {{ $value }}，需检查模型输出质量"

      # 活跃请求积压
      - alert: RequestBacklog
        expr: llm_active_requests > 50
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "LLM请求积压严重"
          description: "当前活跃请求数 {{ $value }}，系统可能过载"

  - name: ai_app_quality
    rules:
      # 用户满意度下降
      - alert: LowUserSatisfaction
        expr: |
          histogram_quantile(0.5, rate(llm_user_satisfaction_ratio_bucket[6h])) < 0.6
        for: 1h
        labels:
          severity: warning
        annotations:
          summary: "用户满意度低于60%"
          description: "需调查模型输出质量问题"

Alertmanager配置（alertmanager.yml）：

global:
  resolve_timeout: 5m

route:
  group_by: ['alertname', 'severity']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 4h
  receiver: 'default'

  routes:
    - match:
        severity: critical
      receiver: 'pagerduty'
      repeat_interval: 1h

receivers:
  - name: 'default'
    webhook_configs:
      - url: 'http://alert-relay:8080/webhook'
        send_resolved: true

  - name: 'pagerduty'
    pagerduty_configs:
      - service_key: 'YOUR_PAGERDUTY_KEY'
    webhook_configs:
      - url: 'http://alert-relay:8080/webhook'
        send_resolved: true

  - name: 'feishu'
    webhook_configs:
      - url: 'http://feishu-bot-relay:8080/send'
        send_resolved: true

8. 分布式追踪（OpenTelemetry集成）

RAG和Agent场景下，一次用户请求可能触发多步操作：检索→重排→生成→工具调用→再生成。分布式追踪是理解链路的关键。

OpenTelemetry集成示例：

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.sdk.resources import Resource, SERVICE_NAME
from opentelemetry.trace import SpanKind
import functools

# 初始化Tracer
resource = Resource.create({SERVICE_NAME: "ai-rag-app"})
provider = TracerProvider(resource=resource)
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://otel-collector:4317"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("ai-rag-app")


class RAGPipelineTracer:
    """RAG链路追踪器"""

    def __init__(self, user_query: str, session_id: str):
        self.user_query = user_query
        self.session_id = session_id

    async def traced_retrieval(self, retriever, query: str, top_k: int = 5):
        """追踪检索阶段"""
        with tracer.start_as_current_span(
            "rag.retrieval",
            kind=SpanKind.CLIENT,
            attributes={
                "rag.query": query,
                "rag.top_k": top_k,
                "rag.retriever.type": type(retriever).__name__,
            }
        ) as span:
            results = await retriever.search(query, top_k=top_k)
            span.set_attribute("rag.results.count", len(results))
            span.set_attribute("rag.results.top_score", results[0].score if results else 0)
            return results

    async def traced_reranking(self, reranker, query: str, documents: list):
        """追踪重排序阶段"""
        with tracer.start_as_current_span(
            "rag.reranking",
            attributes={
                "rag.reranker.input_count": len(documents),
            }
        ) as span:
            reranked = await reranker.rerank(query, documents)
            span.set_attribute("rag.reranker.output_count", len(reranked))
            return reranked

    async def traced_generation(self, llm_client, model: str, messages: list):
        """追踪生成阶段"""
        with tracer.start_as_current_span(
            "rag.generation",
            kind=SpanKind.CLIENT,
            attributes={
                "llm.model": model,
                "llm.messages.count": len(messages),
            }
        ) as span:
            response = await llm_client.chat.completions.create(
                model=model, messages=messages
            )
            usage = response.usage
            span.set_attribute("llm.tokens.prompt", usage.prompt_tokens)
            span.set_attribute("llm.tokens.completion", usage.completion_tokens)
            span.set_attribute("llm.finish_reason", response.choices[0].finish_reason)
            return response

    async def execute_pipeline(self, retriever, reranker, llm_client, model: str):
        """执行完整RAG Pipeline并生成追踪"""
        with tracer.start_as_current_span(
            "rag.pipeline",
            attributes={
                "rag.session_id": self.session_id,
                "rag.user_query": self.user_query,
            }
        ) as root_span:
            # Step 1: 检索
            docs = await self.traced_retrieval(retriever, self.user_query)

            # Step 2: 重排序
            reranked = await self.traced_reranking(reranker, self.user_query, docs)

            # Step 3: 构造Prompt并生成
            context = "\n\n".join([d.content for d in reranked[:3]])
            messages = [
                {"role": "system", "content": f"基于以下上下文回答问题：\n{context}"},
                {"role": "user", "content": self.user_query},
            ]
            response = await self.traced_generation(llm_client, model, messages)

            root_span.set_attribute("rag.pipeline.success", True)
            return response.choices[0].message.content

Jaeger/Grafana Tempo中的链路视图：

每条Trace会展示完整的调用树：

rag.pipeline [total: 3.2s]
├── rag.retrieval [450ms]
│   ├── vector_db.search [420ms]
│   └── embedding.encode [25ms]
├── rag.reranking [180ms]
└── rag.generation [2.5s]
    ├── llm.ttft [0.8s]
    └── llm.tokens [1.7s, 850 tokens]

9. 日志聚合与异常检测

结构化日志规范：

import structlog
import json
from datetime import datetime, timezone

structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.JSONRenderer()
    ],
    wrapper_class=structlog.BoundLogger,
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)

logger = structlog.get_logger()


def log_llm_request(
    request_id: str,
    model: str,
    question: str,
    answer: str,
    prompt_tokens: int,
    completion_tokens: int,
    latency_ms: float,
    ttft_ms: float,
    status: str,
    user_id: str = '',
    session_id: str = '',
    error: str = '',
):
    """结构化LLM请求日志"""
    log_data = {
        'event': 'llm_request',
        'request_id': request_id,
        'model': model,
        'user_id': user_id,
        'session_id': session_id,
        'question_preview': question[:200],
        'answer_preview': answer[:200],
        'metrics': {
            'prompt_tokens': prompt_tokens,
            'completion_tokens': completion_tokens,
            'latency_ms': round(latency_ms, 2),
            'ttft_ms': round(ttft_ms, 2),
        },
        'status': status,
    }
    if error:
        log_data['error'] = error

    if status == 'error':
        logger.error(**log_data)
    elif latency_ms > 10000:
        logger.warning('slow_request', **log_data)
    else:
        logger.info(**log_data)

异常检测 — 基于滑动窗口的延迟异常：

import numpy as np
from collections import deque
import threading

class LatencyAnomalyDetector:
    """基于Z-Score的延迟异常检测"""

    def __init__(self, window_size: int = 1000, z_threshold: float = 3.0):
        self.window_size = window_size
        self.z_threshold = z_threshold
        self.latencies = deque(maxlen=window_size)
        self.lock = threading.Lock()

    def record(self, latency_ms: float):
        with self.lock:
            self.latencies.append(latency_ms)

    def is_anomaly(self, latency_ms: float) -> bool:
        with self.lock:
            if len(self.latencies) < 100:
                return False
            arr = np.array(self.latencies)
            mean = arr.mean()
            std = arr.std()
            if std == 0:
                return False
            z_score = abs(latency_ms - mean) / std
            return z_score > self.z_threshold

    def get_stats(self) -> dict:
        with self.lock:
            if not self.latencies:
                return {}
            arr = np.array(self.latencies)
            return {
                'count': len(arr),
                'mean': round(float(arr.mean()), 2),
                'p50': round(float(np.percentile(arr, 50)), 2),
                'p95': round(float(np.percentile(arr, 95)), 2),
                'p99': round(float(np.percentile(arr, 99)), 2),
                'std': round(float(arr.std()), 2),
            }

detector = LatencyAnomalyDetector()

# 在请求处理中使用
async def handle_request(request):
    start = time.monotonic()
    response = await call_llm(request)
    latency_ms = (time.monotonic() - start) * 1000

    detector.record(latency_ms)
    if detector.is_anomaly(latency_ms):
        logger.warning('latency_anomaly', latency_ms=latency_ms, stats=detector.get_stats())

10. 实战案例：构建AI应用监控看板

Grafana Dashboard JSON（核心面板定义）：

{
  "dashboard": {
    "title": "AI应用监控看板",
    "panels": [
      {
        "title": "请求量 QPS",
        "type": "timeseries",
        "targets": [
          {
            "expr": "rate(llm_request_duration_seconds_count[5m])",
            "legendFormat": "{{model}}"
          }
        ],
        "gridPos": {"x": 0, "y": 0, "w": 8, "h": 6}
      },
      {
        "title": "P95 TTFT (秒)",
        "type": "timeseries",
        "targets": [
          {
            "expr": "histogram_quantile(0.95, rate(llm_time_to_first_token_seconds_bucket[5m]))",
            "legendFormat": "{{model}}"
          }
        ],
        "gridPos": {"x": 8, "y": 0, "w": 8, "h": 6}
      },
      {
        "title": "Token消耗速率",
        "type": "timeseries",
        "targets": [
          {
            "expr": "rate(llm_tokens_total[5m])",
            "legendFormat": "{{model}} - {{type}}"
          }
        ],
        "gridPos": {"x": 16, "y": 0, "w": 8, "h": 6}
      },
      {
        "title": "日累计成本 (USD)",
        "type": "stat",
        "targets": [
          {
            "expr": "increase(llm_cost_usd_total[24h])",
            "legendFormat": "{{model}}"
          }
        ],
        "gridPos": {"x": 0, "y": 6, "w": 6, "h": 4}
      },
      {
        "title": "用户满意度",
        "type": "gauge",
        "targets": [
          {
            "expr": "histogram_quantile(0.5, rate(llm_user_satisfaction_ratio_bucket[6h]))",
            "legendFormat": "{{model}}"
          }
        ],
        "fieldConfig": {
          "defaults": {
            "min": 0, "max": 1,
            "thresholds": {
              "steps": [
                {"value": 0, "color": "red"},
                {"value": 0.6, "color": "yellow"},
                {"value": 0.8, "color": "green"}
              ]
            }
          }
        },
        "gridPos": {"x": 6, "y": 6, "w": 6, "h": 4}
      },
      {
        "title": "错误率",
        "type": "timeseries",
        "targets": [
          {
            "expr": "rate(llm_request_duration_seconds_count{status=\"error\"}[5m]) / rate(llm_request_duration_seconds_count[5m])",
            "legendFormat": "{{model}}"
          }
        ],
        "gridPos": {"x": 12, "y": 6, "w": 12, "h": 4}
      },
      {
        "title": "质量评分分布",
        "type": "heatmap",
        "targets": [
          {
            "expr": "rate(llm_quality_hallucination_score_bucket[1h])",
            "legendFormat": "{{le}}"
          }
        ],
        "gridPos": {"x": 0, "y": 10, "w": 12, "h": 6}
      }
    ]
  }
}

11. 告警策略与值班机制

分级告警策略：

┌──────────┬──────────────────────────────┬──────────────────────────┐
│  级别     │  触发条件                      │  响应方式                  │
├──────────┼──────────────────────────────┼──────────────────────────┤
│  P0      │  服务完全不可用                │  立即电话通知值班           │
│  Critical│  错误率>50%                   │  5分钟内响应               │
├──────────┼──────────────────────────────┼──────────────────────────┤
│  P1      │  错误率>5%                    │  飞书/钉钉群通知           │
│  Warning │  P95延迟>10s                  │  15分钟内响应             │
│          │  日成本超预算                   │                          │
├──────────┼──────────────────────────────┼──────────────────────────┤
│  P2      │  质量评分下降                  │  邮件通知                 │
│  Info    │  满意度低于阈值                │  次日处理                 │
│          │  Token用量异常波动              │                          │
└──────────┴──────────────────────────────┴──────────────────────────┘

值班轮换脚本：

from datetime import datetime, timedelta
import json

ONCALL_SCHEDULE = [
    {"name": "张三", "phone": "138****1234", "start": "2024-01-01"},
    {"name": "李四", "phone": "139****5678", "start": "2024-01-08"},
    {"name": "王五", "phone": "137****9012", "start": "2024-01-15"},
]

def get_current_oncall() -> dict:
    """获取当前值班人"""
    today = datetime.now().date()
    schedule_start = datetime.strptime(ONCALL_SCHEDULE[0]["start"], "%Y-%m-%d").date()
    days_since_start = (today - schedule_start).days
    cycle_length = len(ONCALL_SCHEDULE) * 7  # 每人一周
    current_index = (days_since_start % cycle_length) // 7
    return ONCALL_SCHEDULE[current_index]


async def send_alert(level: str, title: str, description: str):
    """发送告警通知"""
    oncall = get_current_oncall()

    message = {
        "level": level,
        "title": title,
        "description": description,
        "oncall": oncall["name"],
        "timestamp": datetime.now().isoformat(),
    }

    if level == "P0":
        # 电话通知
        await send_phone_alert(oncall["phone"], title)
        # 同时发群消息
        await send_group_message(message)
    elif level == "P1":
        await send_group_message(message)
    else:
        await send_email(title, description)


async def handle_alertmanager_webhook(payload: dict):
    """处理Alertmanager的Webhook"""
    for alert in payload.get("alerts", []):
        level = "P0" if alert["labels"].get("severity") == "critical" else "P1"
        await send_alert(
            level=level,
            title=alert["annotations"].get("summary", "Unknown Alert"),
            description=alert["annotations"].get("description", ""),
        )

告警抑制与聚合规则：

# alertmanager.yml - 抑制规则
inhibit_rules:
  # 如果P0告警已触发，抑制同组的P1告警
  - source_match:
      severity: critical
    target_match:
      severity: warning
    equal: ['alertname', 'model']

  # 如果服务不可用，抑制所有性能告警
  - source_match:
      alertname: ServiceDown
    target_match_re:
      alertname: HighTTFT|LowThroughput

以上就是构建AI应用监控与告警系统的完整方案。核心思路是：先埋点采集，再存储聚合，最后可视化与告警。从TTFT、TPS等推理性能指标，到Token成本、幻觉率等AI特有指标，再到用户满意度反馈，形成完整的可观测性闭环。实际落地时，建议从最关键的两三个指标开始，逐步扩展监控覆盖面。