AI 应用的监控与可观测性

AI应用监控的特殊性

与传统Web应用不同，AI应用的监控面临独特的挑战。传统应用监控关注的是CPU、内存、请求延迟等基础设施指标，而AI应用还需要监控模型层面的指标：每次调用的token消耗（成本监控）、生成内容的质量（是否有幻觉、是否答非所问）、用户行为指标（满意度、复问率、转人工率）、以及模型自身的状态（API限流、模型版本变更、响应格式变化）。这些指标共同构成了AI应用的可观测性（Observability）——不仅要看到系统"是否在运行"，还要看到系统"运行得好不好"。

没有完善的监控体系，AI应用就像在黑暗中飞行。你不知道用户是否满意、成本是否超预算、模型是否产生了幻觉——直到用户投诉爆发或账单超支。本文将构建一个涵盖基础设施、应用、模型和业务四个层面的AI应用监控体系。

四层监控架构

第一层：基础设施监控。与传统应用监控一致——CPU、内存、网络、磁盘。使用Prometheus + Node Exporter采集，Grafana可视化。对于GPU推理服务，还需监控GPU利用率、显存使用、GPU温度。

第二层：应用层监控。核心指标包括：API请求量（QPS）、请求延迟（P50/P95/P99）、错误率（4xx/5xx）、并发连接数。使用OpenTelemetry进行分布式链路追踪，追踪一个用户请求在ASR→LLM→TTS各环节的耗时分布。

第三层：模型层监控。这是AI应用特有的监控层，也是最关键的一层。核心指标：Token消耗（每次调用的输入/输出token数）、模型延迟（TTFT首token时间、TPOT每token时间）、API限流命中次数、模型响应长度分布。

import time, json
from openai import OpenAI
from dataclasses import dataclass, field
from typing import List

client = OpenAI(api_key="your-deepseek-api-key", base_url="https://api.deepseek.com")

@dataclass
class LLMCallMetrics:
    timestamp: float
    model: str
    input_tokens: int
    output_tokens: int
    ttft_ms: float  # Time to first token
    total_latency_ms: float
    status: str  # success/error/rate_limited
    cost_estimate: float

class AIMonitor:
    def __init__(self):
        self.metrics: List[LLMCallMetrics] = []
        self.alert_thresholds = {
            "p99_latency_ms": 5000,
            "error_rate": 0.05,
            "daily_cost_usd": 50,
            "hallucination_rate": 0.10
        }

    def call_with_monitoring(self, messages, model="deepseek-chat", **kwargs):
        """包装LLM调用，自动采集指标"""
        start = time.time()
        ttft = None
        try:
            stream = client.chat.completions.create(
                model=model, messages=messages, stream=True, **kwargs
            )
            result = ""
            usage = None
            for chunk in stream:
                if ttft is None and chunk.choices[0].delta.content:
                    ttft = (time.time() - start) * 1000
                if chunk.choices[0].delta.content:
                    result += chunk.choices[0].delta.content
                if hasattr(chunk, 'usage') and chunk.usage:
                    usage = chunk.usage

            latency = (time.time() - start) * 1000
            metric = LLMCallMetrics(
                timestamp=time.time(), model=model,
                input_tokens=usage.prompt_tokens if usage else 0,
                output_tokens=usage.completion_tokens if usage else len(result)//2,
                ttft_ms=ttft or 0, total_latency_ms=latency,
                status="success",
                cost_estimate=(usage.prompt_tokens if usage else 0)*0.14/1000000
            )
            self.metrics.append(metric)
            self._check_alerts()
            return result
        except Exception as e:
            latency = (time.time() - start) * 1000
            self.metrics.append(LLMCallMetrics(
                timestamp=time.time(), model=model,
                input_tokens=0, output_tokens=0,
                ttft_ms=0, total_latency_ms=latency,
                status=f"error: {str(e)[:50]}", cost_estimate=0
            ))
            raise

    def _check_alerts(self):
        """检查告警阈值"""
        recent = [m for m in self.metrics if time.time()-m.timestamp < 3600]
        if not recent:
            return
        error_rate = sum(1 for m in recent if m.status!="success") / len(recent)
        if error_rate > self.alert_thresholds["error_rate"]:
            print(f"🚨 告警：错误率 {error_rate:.1%} 超过阈值 {self.alert_thresholds['error_rate']:.1%}")

        latencies = [m.total_latency_ms for m in recent if m.status=="success"]
        if latencies and sorted(latencies)[int(len(latencies)*0.99)] > self.alert_thresholds["p99_latency_ms"]:
            print(f"🚨 告警：P99延迟超过 {self.alert_thresholds['p99_latency_ms']}ms")

    def get_stats(self):
        """获取统计摘要"""
        success = [m for m in self.metrics if m.status=="success"]
        if not success:
            return "暂无数据"
        return {
            "总调用次数": len(self.metrics),
            "成功率": f"{len(success)/len(self.metrics):.1%}",
            "平均延迟": f"{sum(m.total_latency_ms for m in success)/len(success):.0f}ms",
            "总Token消耗": sum(m.input_tokens+m.output_tokens for m in success),
            "估算成本": f"${sum(m.cost_estimate for m in success):.4f}"
        }

monitor = AIMonitor()
for i in range(5):
    result = monitor.call_with_monitoring(
        [{"role":"user","content":"用一句话介绍DeepSeek"}]
    )
print(json.dumps(monitor.get_stats(), ensure_ascii=False, indent=2))

第四层：业务层监控。最终衡量AI应用价值的是业务指标：用户满意度（点赞率/点踩率）、任务完成率（用户是否完成了他们想做的事情）、复问率（用户是否反复问同一个问题——说明AI没答好）、转人工率（对于客服AI）、日均活跃用户数、会话时长。这些指标不能从系统日志中自动获取，需要在产品中埋点采集。

成本追踪与优化

AI应用的成本主要来自LLM API调用。需要追踪的维度包括：按模型的成本分布（V3 vs R1）、按功能的成本分布（哪个功能消耗最多token）、按用户的成本分布（是否有滥用或大户）、以及成本趋势（日/周/月环比）。建议设置每日预算告警（如超过$50自动通知），并对异常消耗（如某用户单日消耗超过平均值10倍）进行自动熔断。

质量监控

AI生成内容的质量是最难监控但最重要的维度。实用方案：随机抽样（每小时抽取1-5条对话进行人工审核）、用户反馈（利用点赞/点踩数据）、自动评估（用LLM-as-Judge自动评估生成内容的忠实度和相关性）、回归测试（维护一组标准测试用例，每次模型或提示词变更后自动运行，确保关键场景不受影响）。质量监控不需要覆盖100%的流量，覆盖5-10%就足以发现问题趋势。

告警策略设计

有了监控数据，下一步是设计合理的告警策略。告警太少会导致问题被忽视，告警太多会导致"告警疲劳"——人们开始忽略所有告警。推荐的告警设计原则：分层告警——P0（紧急，影响线上用户，5分钟内响应）、P1（重要，需当天处理）、P2（一般，可排入迭代）；复合条件——避免单一指标触发告警，例如"错误率>5%"AND"持续>5分钟"才触发；告警收敛——同一类告警在30分钟内只发送一次，避免告警风暴；告警升级——P1告警30分钟未确认自动升级为P0。对于AI应用的特有告警：Token消耗异常（单日消耗超过前一天同期的150%）、幻觉率异常（自动评估的Faithfulness低于0.8）、模型响应格式异常（JSON解析失败率超过5%）。成本可视化仪表盘：成本监控需要专门的仪表盘，让团队一眼就能看到"钱花在哪了"。推荐的Dashboard布局：左上角显示今日/本周/本月总费用及环比变化；中间显示按模型划分的费用分布（饼图）；右侧显示按功能/API端点划分的费用（柱状图）；底部显示费用趋势（折线图，按天/周/月切换）。同时标注预算线——当前费用接近预算的80%时黄色预警，超过预算时红色预警。仪表盘数据每小时更新一次，确保信息的时效性。

异常检测与智能诊断

除了被动告警，主动的异常检测能让团队在用户投诉之前就发现问题。推荐使用时间序列异常检测算法（如Prophet或Isolation Forest）分析关键指标的历史趋势，当当前值偏离预测值超过3个标准差时自动触发告警。更进一步，可以构建智能诊断系统——当检测到异常时，自动关联分析可能的原因（如"延迟突增"同时"GPU利用率也突增"，可能原因是流量突增而非模型问题），给出初步诊断结论和建议的处理方案，帮助运维人员快速定位问题根因。

总结：AI应用的可观测性是一个持续构建的过程。建议从最关键的几个指标开始（API延迟、错误率、Token消耗），逐步扩展到更全面的监控体系。不要追求一开始就完美——一个简单但实际在用的监控系统，远比一个复杂但从未部署的监控方案有价值。记住监控的终极目的不是收集数据，而是让团队能够在问题影响用户之前发现并解决它。