การแจ้งเตือนเวลาแฝง ต้นทุน และคุณภาพที่ลดลง

เหตุใดการแจ้งเตือนจึงสำคัญต่อระบบ LLM

สามประเภทของการแจ้งเตือน

from dataclasses import dataclass

@dataclass
class AlertThresholds:
    # Latency (milliseconds)
    p50_latency_ms: int = 2000    # median should be under 2s
    p99_latency_ms: int = 10000   # 99th percentile under 10s
    # Cost (USD)
    max_cost_per_request: float = 0.05  # alert if one request costs > 5 cents
    max_daily_spend: float = 50.00      # alert if daily spend exceeds $50
    # Quality (0.0 to 1.0 scale)
    min_quality_score: float = 0.75     # alert if rolling avg drops below 75%
    min_faithfulness: float = 0.80      # alert if RAGAS faithfulness drops below 80%

DEFAULT_THRESHOLDS = AlertThresholds()

การรวบรวมตัวชี้วัดเวลาแฝง

import time
from collections import deque
from statistics import quantiles

class LatencyTracker:
    def __init__(self, window_size=100):
        self.window = deque(maxlen=window_size)  # rolling window of latencies

    def record(self, latency_ms: float):
        self.window.append(latency_ms)

    def p50(self) -> float:
        if not self.window:
            return 0
        return quantiles(self.window, n=100)[49]

    def p99(self) -> float:
        if not self.window:
            return 0
        return quantiles(self.window, n=100)[98]

    def check_alert(self, thresholds: AlertThresholds) -> list[str]:
        alerts = []
        if self.p99() > thresholds.p99_latency_ms:
            alerts.append(f'p99 latency {self.p99():.0f}ms exceeds {thresholds.p99_latency_ms}ms')
        if self.p50() > thresholds.p50_latency_ms:
            alerts.append(f'p50 latency {self.p50():.0f}ms exceeds {thresholds.p50_latency_ms}ms')
        return alerts

latency_tracker = LatencyTracker()

การติดตามและแจ้งเตือนค่าใช้จ่าย

from datetime import datetime, date
import threading

class CostTracker:
    def __init__(self):
        self._lock = threading.Lock()
        self._daily_spend = {}  # date -> total USD
        self._per_request = []  # list of (timestamp, cost)

    def record(self, cost_usd: float) -> list[str]:
        today = date.today().isoformat()
        alerts = []
        
        with self._lock:
            # Track daily spend
            self._daily_spend[today] = self._daily_spend.get(today, 0) + cost_usd
            self._per_request.append((datetime.now(), cost_usd))
        
        # Alert on expensive single requests
        if cost_usd > DEFAULT_THRESHOLDS.max_cost_per_request:
            alerts.append(f'Expensive request: ${cost_usd:.4f} (threshold: ${DEFAULT_THRESHOLDS.max_cost_per_request})')
        
        # Alert on daily budget exceeded
        daily_total = self._daily_spend[today]
        if daily_total > DEFAULT_THRESHOLDS.max_daily_spend:
            alerts.append(f'Daily budget exceeded: ${daily_total:.2f} > ${DEFAULT_THRESHOLDS.max_daily_spend}')
        
        return alerts

cost_tracker = CostTracker()

การแจ้งเตือนคะแนนคุณภาพ

import random
from openai import OpenAI

client = OpenAI()

def evaluate_response_quality(question: str, answer: str) -> float:
    judge_prompt = f'''Rate the quality of this AI assistant response on a scale from 0 to 1.

Question: {question}
Answer: {answer}

Return only a JSON object: {{"score": 0.85, "reason": "brief reason"}}

Scoring guide:
1.0 = Perfect, accurate, helpful
0.75 = Good, minor issues
0.5 = Acceptable but incomplete
0.25 = Poor, major gaps
0.0 = Completely wrong or harmful'''
    
    response = client.chat.completions.create(
        model='gpt-4o-mini',  # cheaper model for evaluation
        messages=[{'role': 'user', 'content': judge_prompt}],
        response_format={'type': 'json_object'}
    )
    import json
    result = json.loads(response.choices[0].message.content)
    return float(result['score'])

def maybe_evaluate(question: str, answer: str, sample_rate=0.1):
    if random.random() < sample_rate:
        score = evaluate_response_quality(question, answer)
        quality_tracker.record(score)
        return score
    return None

ค่าเฉลี่ยเคลื่อนที่สำหรับตรวจจับแนวโน้ม

class RollingQualityMonitor:
    def __init__(self, window=50, alert_threshold=0.75, ewma_alpha=0.1):
        self.scores = []
        self.window = window
        self.alert_threshold = alert_threshold
        self.alpha = ewma_alpha  # EWMA decay factor
        self.ewma_score = None

    def record(self, score: float):
        self.scores.append(score)
        if len(self.scores) > self.window:
            self.scores.pop(0)
        
        # Update exponentially weighted moving average
        if self.ewma_score is None:
            self.ewma_score = score
        else:
            self.ewma_score = self.alpha * score + (1 - self.alpha) * self.ewma_score

    def should_alert(self) -> bool:
        if len(self.scores) < 10:  # not enough data yet
            return False
        return self.ewma_score < self.alert_threshold

    def summary(self) -> dict:
        if not self.scores:
            return {}
        return {
            'rolling_avg': sum(self.scores) / len(self.scores),
            'ewma': self.ewma_score,
            'sample_count': len(self.scores),
            'alert': self.should_alert()
        }

quality_tracker = RollingQualityMonitor()

การกำหนดเส้นทางการแจ้งเตือน: Slack และ PagerDuty

import requests

def send_slack_alert(message: str, severity: str, webhook_url: str):
    color = {'critical': '#ff0000', 'warning': '#ff9900', 'info': '#36a64f'}[severity]
    payload = {
        'attachments': [{
            'color': color,
            'title': f'LLM Alert [{severity.upper()}]',
            'text': message,
            'footer': 'AI Pipeline Monitor'
        }]
    }
    requests.post(webhook_url, json=payload)

def send_pagerduty_alert(title: str, details: str, routing_key: str):
    payload = {
        'routing_key': routing_key,
        'event_action': 'trigger',
        'payload': {
            'summary': title,
            'severity': 'critical',
            'source': 'ai-pipeline-monitor',
            'custom_details': {'details': details}
        }
    }
    requests.post('https://events.pagerduty.com/v2/enqueue', json=payload)

def route_alert(alert_text: str, severity: str):
    send_slack_alert(alert_text, severity, SLACK_WEBHOOK_URL)
    if severity == 'critical':
        send_pagerduty_alert(alert_text, alert_text, PAGERDUTY_ROUTING_KEY)

วงรอบการประเมินการแจ้งเตือน

import threading
import time

def alert_evaluation_loop(interval_seconds=60):
    while True:
        try:
            # Check latency
            latency_alerts = latency_tracker.check_alert(DEFAULT_THRESHOLDS)
            for alert in latency_alerts:
                route_alert(f'LATENCY: {alert}', 'warning')
            
            # Check quality
            quality_summary = quality_tracker.summary()
            if quality_summary.get('alert'):
                msg = f'QUALITY DEGRADATION: Rolling avg {quality_summary["ewma"]:.2f} below threshold {DEFAULT_THRESHOLDS.min_quality_score}'
                route_alert(msg, 'warning')
            
            print(f'Alert check complete. Quality: {quality_summary.get("ewma", "N/A")}')
        except Exception as e:
            print(f'Alert evaluation error: {e}')
        
        time.sleep(interval_seconds)

# Start in background thread
alert_thread = threading.Thread(target=alert_evaluation_loop, daemon=True)
alert_thread.start()

ความล้าจากการแจ้งเตือนและการปรับแต่งเกณฑ์

class AlertDebouncer:
    def __init__(self, required_consecutive_fires=3):
        self.required = required_consecutive_fires
        self.fire_counts = {}  # alert_name -> consecutive fires
        self.already_firing = set()  # alert_names currently active

    def should_fire(self, alert_name: str, condition: bool) -> bool:
        if condition:
            self.fire_counts[alert_name] = self.fire_counts.get(alert_name, 0) + 1
            if self.fire_counts[alert_name] >= self.required and alert_name not in self.already_firing:
                self.already_firing.add(alert_name)
                return True  # fire the alert (first time condition sustained)
        else:
            if self.fire_counts.get(alert_name, 0) > 0:
                self.fire_counts[alert_name] = 0
                self.already_firing.discard(alert_name)  # condition cleared
        return False  # either not sustained or already firing (no duplicate alert)

debouncer = AlertDebouncer(required_consecutive_fires=3)

การตรวจสอบความผิดปกติของค่าใช้จ่ายด้วยวิธีการทางสถิติ

import statistics

def compute_z_score(recent_value: float, historical_values: list[float]) -> float:
    if len(historical_values) < 5:
        return 0  # not enough data
    mean = statistics.mean(historical_values)
    stdev = statistics.stdev(historical_values)
    if stdev == 0:
        return 0
    return (recent_value - mean) / stdev

def check_cost_anomaly(current_hour_cost: float, historical_hourly_costs: list[float]) -> str | None:
    z = compute_z_score(current_hour_cost, historical_hourly_costs)
    if z > 3.0:  # more than 3 standard deviations above mean
        mean = statistics.mean(historical_hourly_costs)
        return f'Cost anomaly: ${current_hour_cost:.2f} this hour (normal: ${mean:.2f}, z={z:.1f})'
    return None

การสร้างแดชบอร์ดการปฏิบัติการ

ตรวจสอบความเข้าใจอย่างรวดเร็ว

สรุปบทเรียน