AI 应用监控体系：从模型质量到基础设施

┌─────────────────────────────────────────────────┐
│  Layer 3: 模型质量监控                             │
│  - 回答正确率 / 幻觉率 / 毒性检测 / 偏见            │
│  - 用户满意度反馈                                  │
├─────────────────────────────────────────────────┤
│  Layer 2: 应用性能监控                             │
│  - 首字延迟 (TTFT) / 完整延迟 / 吞吐量              │
│  - Token 消耗 / 成本 / 缓存命中率                   │
├─────────────────────────────────────────────────┤
│  Layer 1: 基础设施监控                             │
│  - GPU 利用率 / 内存 / 网络 / 磁盘                  │
│  - API 错误率 / 限流次数 / 超时                    │
└─────────────────────────────────────────────────┘

from prometheus_client import Counter, Histogram, Gauge, Summary

# === 模型质量指标 ===
llm_quality_score = Gauge(
    "llm_quality_score",
    "Average answer quality score (0-1)",
    ["model", "task_type"]
)
llm_hallucination_rate = Gauge(
    "llm_hallucination_rate",
    "Hallucination rate detected",
    ["model"]
)
llm_user_feedback = Counter(
    "llm_user_feedback_total",
    "User feedback counts",
    ["model", "feedback"]  # feedback: positive/negative
)

# === 应用性能指标 ===
llm_ttft = Histogram(
    "llm_time_to_first_token_seconds",
    "Time to first token in seconds",
    ["model"],
    buckets=[0.1, 0.3, 0.5, 1, 2, 5, 10]
)
llm_total_latency = Histogram(
    "llm_total_latency_seconds",
    "Total request latency in seconds",
    ["model"],
    buckets=[0.5, 1, 2, 5, 10, 20, 30, 60]
)
llm_tokens_total = Counter(
    "llm_tokens_total",
    "Total tokens consumed",
    ["model", "direction"]  # direction: input/output
)
llm_cost_total = Counter(
    "llm_cost_usd_total",
    "Total cost in USD",
    ["model"]
)
llm_cache_hit_rate = Gauge(
    "llm_cache_hit_rate",
    "Cache hit rate",
    ["cache_type"]  # cache_type: exact/semantic
)

# === 基础设施指标 ===
gpu_utilization = Gauge(
    "gpu_utilization_percent",
    "GPU utilization percentage",
    ["gpu_id", "node"]
)
api_error_rate = Gauge(
    "llm_api_error_rate",
    "LLM API error rate",
    ["provider", "error_type"]
)

import time
from contextlib import asynccontextmanager

class LLMInstrumentation:
    def __init__(self, registry):
        self.registry = registry

    @asynccontextmanager
    async def trace_request(self, model: str, task_type: str):
        request_id = generate_id()
        start = time.monotonic()
        first_token_time = None

        async def on_first_token():
            nonlocal first_token_time
            first_token_time = time.monotonic()

        try:
            yield {"request_id": request_id, "on_first_token": on_first_token}
            # 请求成功
            elapsed = time.monotonic() - start
            llm_total_latency.labels(model=model).observe(elapsed)
            if first_token_time:
                ttft = first_token_time - start
                llm_ttft.labels(model=model).observe(ttft)
        except Exception as e:
            api_error_rate.labels(
                provider=model, error_type=type(e).__name__
            ).inc()
            raise

    def record_tokens(self, model, input_tokens, output_tokens):
        llm_tokens_total.labels(
            model=model, direction="input"
        ).inc(input_tokens)
        llm_tokens_total.labels(
            model=model, direction="output"
        ).inc(output_tokens)
        # 成本计算
        cost = self._calc_cost(model, input_tokens, output_tokens)
        llm_cost_total.labels(model=model).inc(cost)

    def _calc_cost(self, model, in_tok, out_tok):
        prices = {
            "gpt-4o": (0.0025, 0.01),
            "gpt-4o-mini": (0.00015, 0.0006),
            "claude-sonnet": (0.003, 0.015),
        }
        in_price, out_price = prices.get(model, (0, 0))
        return (in_tok * in_price + out_tok * out_price) / 1000

instrument = LLMInstrumentation(registry)

async def chat_completion(messages, model="gpt-4o"):
    async with instrument.trace_request(model, "chat") as ctx:
        response = await llm_client.chat.completions.create(
            model=model,
            messages=messages,
            stream=True,
        )
        async for chunk in response:
            if chunk.choices[0].delta.content:
                if not ctx["first_token_called"]:
                    await ctx["on_first_token"]()
                    ctx["first_token_called"] = True
                yield chunk.choices[0].delta.content

    instrument.record_tokens(model, input_count, output_count)

class QualityMonitor:
    def __init__(self, judge_model="gpt-4o", sample_rate=0.1):
        self.judge_model = judge_model
        self.sample_rate = sample_rate

    async def maybe_evaluate(self, query, response, context=None):
        """按采样率随机评估回答质量"""
        import random
        if random.random() > self.sample_rate:
            return

        score, issues = await self._judge(query, response, context)

        llm_quality_score.labels(
            model=self.judge_model, task_type="auto"
        ).set(score)

        if "hallucination" in issues:
            llm_hallucination_rate.labels(
                model=self.judge_model
            ).inc()

    async def _judge(self, query, response, context):
        prompt = f"""评估以下回答的质量。
问题：{query}
回答：{response}
参考资料：{context or '无'}

请检查：
1. 事实准确性（是否有幻觉）
2. 回答完整性
3. 逻辑一致性

输出 JSON：{{"score": 0.0-1.0, "issues": ["..."]}}"""

        result = await llm_client.chat.completions.create(
            model=self.judge_model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"},
        )
        data = json.loads(result.choices[0].message.content)
        return data["score"], data.get("issues", [])

# TTFT P99
histogram_quantile(0.99, rate(llm_time_to_first_token_seconds_bucket[5m]))

# 总延迟 P50/P90/P99
histogram_quantile(0.50, rate(llm_total_latency_seconds_bucket[5m]))
histogram_quantile(0.90, rate(llm_total_latency_seconds_bucket[5m]))
histogram_quantile(0.99, rate(llm_total_latency_seconds_bucket[5m]))

# 每小时 Token 消耗（按模型）
sum(rate(llm_tokens_total[1h])) by (model, direction)

# 每日成本
sum(increase(llm_cost_usd_total[24h])) by (model)

# 质量评分趋势
avg_over_time(llm_quality_score[1h])

# 幻觉率
avg_over_time(llm_hallucination_rate[1h])

llm_cache_hit_rate{cache_type="exact"}
llm_cache_hit_rate{cache_type="semantic"}

# alerting_rules.yml
groups:
  - name: llm_quality
    rules:
      - alert: QualityScoreDrop
        expr: avg_over_time(llm_quality_score[30m]) < 0.8
        for: 10m
        labels:
          severity: warning
        annotations:
          summary: "LLM quality score below 0.8"
          description: "Quality score is {{ $value }} for model {{ $labels.model }}"

      - alert: HallucinationSpike
        expr: rate(llm_hallucination_rate[5m]) > 0.05
        for: 5m
        labels:
          severity: critical
        annotations:
          summary: "Hallucination rate exceeds 5%"

  - name: llm_performance
    rules:
      - alert: HighLatency
        expr: histogram_quantile(0.99, rate(llm_total_latency_seconds_bucket[5m])) > 15
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "P99 latency > 15s"

      - alert: APIErrorRate
        expr: rate(llm_api_error_rate[5m]) > 0.01
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "API error rate > 1%"

  - name: llm_cost
    rules:
      - alert: CostBudgetExceeded
        expr: sum(increase(llm_cost_usd_total[24h])) > 500
        labels:
          severity: warning
        annotations:
          summary: "Daily cost budget exceeded: ${{ $value }}"

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

async def rag_pipeline(query):
    with tracer.start_as_current_span("rag_pipeline") as span:
        span.set_attribute("query.length", len(query))

        with tracer.start_as_current_span("retrieval"):
            docs = await vector_db.search(query, top_k=5)
            span.set_attribute("retrieval.doc_count", len(docs))

        with tracer.start_as_current_span("generation"):
            response = await llm.complete(query, context=docs)
            span.set_attribute("generation.tokens", len(response) // 4)

        return response