← 返回首页
🧠

LLM指标监控

📂 llm ⏱ 2 min 321 words

--- title: "LLM指标监控" description: "深入讲解LLM应用的关键监控指标,包括性能指标、质量指标和成本指标的定义与采集方法。" tags: ["指标", "LLM", "监控"] category: "llm" icon: "🧠"

LLM指标监控

指标监控的重要性

指标监控是LLM可观测性的核心组成部分。通过定义和追踪关键指标,团队可以实时了解系统健康状况,及时发现问题,并做出数据驱动的决策。

对于LLM应用,指标监控不仅关注传统的系统性能,还需要考虑模型特有的维度,如token使用量、推理质量和成本效益。

核心性能指标

延迟指标

延迟是衡量LLM响应速度的关键指标:

import time
from dataclasses import dataclass
from typing import List

@dataclass
class LatencyMetrics:
    ttft: float  # 首token延迟
    total_time: float  # 总响应时间
    tokens_generated: int  # 生成的token数
    
    @property
    def generation_speed(self) -> float:
        """每秒生成的token数"""
        return self.tokens_generated / self.total_time if self.total_time > 0 else 0

class LatencyTracker:
    def __init__(self):
        self.records: List[LatencyMetrics] = []
    
    def record(self, metrics: LatencyMetrics):
        self.records.append(metrics)
    
    def get_percentile(self, percentile: float) -> float:
        sorted_times = sorted(r.total_time for r in self.records)
        index = int(len(sorted_times) * percentile / 100)
        return sorted_times[min(index, len(sorted_times) - 1)]

吞吐量指标

吞吐量反映系统的处理能力:

质量指标

输出质量评估

LLM输出质量需要从多个维度评估:

from enum import Enum
from typing import Optional

class QualityLevel(Enum):
    EXCELLENT = 5
    GOOD = 4
    ACCEPTABLE = 3
    POOR = 2
    UNACCEPTABLE = 1

class QualityEvaluator:
    def __init__(self):
        self.evaluation_prompt = """
        评估以下LLM输出的质量(1-5分):
        问题:{query}
        输出:{response}
        
        请从相关性、准确性、完整性三个维度打分。
        """
    
    def evaluate(self, query: str, response: str) -> dict:
        # 使用评估模型或规则引擎进行评分
        return {
            "relevance": self._score_relevance(query, response),
            "accuracy": self._score_accuracy(response),
            "completeness": self._score_completeness(query, response)
        }

成本指标

Token成本追踪

LLM API通常按token计费,成本监控至关重要:

@dataclass
class CostMetrics:
    input_tokens: int
    output_tokens: int
    model: str
    
    # 价格表(每1K token)
    PRICING = {
        "gpt-4": {"input": 0.03, "output": 0.06},
        "gpt-3.5-turbo": {"input": 0.001, "output": 0.002}
    }
    
    def calculate_cost(self) -> float:
        prices = self.PRICING.get(self.model, {"input": 0.01, "output": 0.02})
        input_cost = (self.input_tokens / 1000) * prices["input"]
        output_cost = (self.output_tokens / 1000) * prices["output"]
        return input_cost + output_cost

指标聚合与分析

时间序列聚合

将指标按时间窗口聚合,观察趋势变化:

from datetime import datetime, timedelta
from collections import defaultdict

class MetricsAggregator:
    def __init__(self, window_size: timedelta = timedelta(minutes=5)):
        self.window_size = window_size
        self.buckets = defaultdict(list)
    
    def add_metric(self, metric_name: str, value: float, timestamp: datetime):
        bucket_key = timestamp.replace(
            minute=(timestamp.minute // 5) * 5,
            second=0, microsecond=0
        )
        self.buckets[(metric_name, bucket_key)].append(value)
    
    def get_average(self, metric_name: str, start: datetime, end: datetime) -> float:
        values = []
        current = start
        while current <= end:
            key = (metric_name, current)
            values.extend(self.buckets.get(key, []))
            current += self.window_size
        return sum(values) / len(values) if values else 0

告警策略

设置合理的告警阈值,确保及时发现问题:

指标 警告阈值 严重阈值 检查频率
P99延迟 >3秒 >10秒 每分钟
错误率 >1% >5% 每分钟
每分钟成本 >$10 >$50 每小时

通过系统化的指标监控,你可以全面掌握LLM应用的运行状况,为优化决策提供数据支撑。