Alerty dotyczące opóźnień, kosztów i spadku jakości
Zdefiniuj progi alertów dla opóźnienia p99, kosztu pojedynczego żądania i automatycznych ocen jakości oraz kieruj alerty do Slack lub PagerDuty, gdy jakość potoku LLM się pogarsza.
Alerty dotyczące opóźnień, kosztów i spadku jakości to bezpłatna lekcja AI Engineering Academy na CoddyKit. To lekcja 4 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Engineering Academy, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Engineering Academy zawiera 4 lekcji w sumie.
Dlaczego alerty są ważne w systemach LLM
Aplikacje LLM zawodzą w sposób subtelny i stopniowy. Zmiana promptu może zwiększyć średnie opóźnienie o 30%, ulepszenie mechanizmu pobierania danych może nieznacznie obniżyć jakość odpowiedzi, a nagły wzrost użycia może podnieść dzienne koszty do pięciokrotności budżetu. Bez proaktywnych alertów problemy te zostaną wykryte dopiero wtedy, gdy użytkownicy zaczną narzekać lub nadejdzie miesięczny rachunek. Alerty zmieniają reaktywne gaszenie pożarów w proaktywne działania operacyjne.
Trzy kategorie alertów
Alerty w aplikacjach LLM dzielą się na trzy kategorie. Alerty dotyczące opóźnień uruchamiają się, gdy czas odpowiedzi przekroczy próg jakości obsługi użytkownika (np. p99 > 10 sekund). Alerty dotyczące kosztów uruchamiają się, gdy koszt pojedynczego żądania lub dzienne wydatki przekroczą ustalone limity budżetowe, zapobiegając nieoczekiwanie wysokim rachunkom. Alerty dotyczące jakości uruchamiają się, gdy automatyczne metryki jakości (oceny LLM-as-judge, wskaźniki zadowolenia użytkowników, oceny wierności) spadną poniżej akceptowalnego poziomu. Każda kategoria wymaga innego oprzyrządowania i innych kanałów alertów.
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()Zbieranie metryk opóźnień
Aby tworzyć alerty dotyczące opóźnień, należy najpierw zbierać dane w spójny sposób. Należy osobno mierzyć trzy składowe opóźnienia: czas do pierwszego tokenu (TTFT, kluczowy dla obsługi strumieniowania), całkowity czas odpowiedzi oraz opóźnienia poszczególnych kroków pobierania danych i generowania. Należy przechowywać je jako dane szeregów czasowych (jeden punkt danych na żądanie), aby można było obliczać percentyle i średnie kroczące z ostatnich N minut lub godzin.
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()Śledzenie kosztów i alerty dotyczące kosztów
Monitorowanie kosztów wymaga śledzenia wydatków na kilku poziomach szczegółowości: kosztu pojedynczego żądania (aby wykrywać kosztowne zapytania odstające), sum godzinowych i dziennych (aby wykrywać skoki użycia) oraz kosztu poszczególnych funkcji (aby ustalić, która część aplikacji jest najdroższa). Należy natychmiast reagować na anomalie kosztu pojedynczych żądań, a do dziennych limitów budżetowych używać zaplanowanych kontroli.
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()Alerty dotyczące ocen jakości
Alerty dotyczące jakości są najbardziej złożoną kategorią, ponieważ jakości nie można mierzyć wyłącznie na podstawie metryk infrastruktury — wymaga ona oceny semantycznej. Najlepiej skalowalnym podejściem jest automatyczna ocena próbkowanych danych: dla losowej próbki żądań produkcyjnych (5–10%) należy asynchronicznie uruchomić ewaluator LLM-as-judge, zapisać oceny i obliczyć średnią kroczącą. Alert należy uruchomić, gdy średnia krocząca spadnie poniżej ustalonego minimalnego poziomu jakości.
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Średnie kroczące do wykrywania trendów
Pojedyncza nieudana odpowiedź nie oznacza problemu systemowego. Należy używać średnich kroczących w określonym przedziale czasu (np. z ostatniej godziny lub dla ostatnich 100 żądań), aby wykrywać trendy. Średnia krocząca, która przekroczy próg i utrzyma się powyżej niego przez ponad 10 minut, wskazuje na rzeczywisty problem, podczas gdy krótkotrwały skok może być szumem. Wykładniczo ważone średnie kroczące (EWMA) szybciej reagują na ostatnie zmiany niż zwykłe średnie kroczące.
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()Kierowanie alertów: Slack i PagerDuty
Alerty są przydatne tylko wtedy, gdy docierają do właściwej osoby właściwym kanałem. Należy kierować alerty według ważności: alerty o pogorszeniu jakości i przekroczeniu budżetu kosztów powinny trafiać do kanału Slack, który zespół monitoruje w godzinach pracy. Alerty dotyczące opóźnień powyżej krytycznych progów (np. p99 > 30 sekund) oraz nagłe skoki kosztów (np. 10-krotność normy w ciągu 5 minut) powinny trafiać do PagerDuty w celu natychmiastowej eskalacji do osoby dyżurującej.
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)Pętla oceny alertów
Logika alertów powinna działać w zaplanowanej pętli, a nie w ramach przetwarzania żądań. Asynchroniczne uruchamianie kontroli alertów zapobiega dodawaniu opóźnień do żądań produkcyjnych. W przypadku większości potrzeb związanych z alertami wystarczy wątek działający w tle lub zaplanowane zadanie uruchamiane co 60 sekund. Metryki należy zbierać w ścieżce obsługi żądania, a progi oceniać w pętli działającej w tle.
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()Zmęczenie alertami i dostrajanie progów
Zmęczenie alertami występuje, gdy alerty uruchamiają się tak często, że zespół zaczyna je ignorować. Aby temu zapobiec, należy: rozpocząć od ostrożnych (wysokich) progów i zaostrzać je w miarę poznawania wartości bazowych, wymagać utrzymania się warunku alertu przez kilka cykli oceny przed jego uruchomieniem, grupować powiązane alerty w pojedyncze powiadomienia oraz regularnie przeglądać i wycofywać alerty, które nigdy nie prowadzą do znaczących działań.
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)Monitorowanie anomalii kosztów metodami statystycznymi
Proste alerty progowe nie wykrywają subtelnych anomalii kosztów, takich jak stopniowy wzrost kosztów o 50% w ciągu dwóch dni. Należy używać statystycznego wykrywania anomalii: obliczać średnią kroczącą i odchylenie standardowe kosztów godzinowych oraz uruchamiać alert, gdy koszt w bieżącej godzinie jest większy od średniej o więcej niż N odchyleń standardowych (alerty na podstawie wyniku Z). Metoda ta automatycznie dostosowuje się do naturalnych wzorców użycia, takich jak ruch w dni robocze i weekendy.
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 NoneTworzenie pulpitu operacyjnego
Alerty są warstwą reaktywną, a pulpit operacyjny na żywo — warstwą proaktywną. Należy utworzyć prosty pulpit pokazujący metryki w czasie rzeczywistym: bieżące opóźnienie p50/p99, liczbę żądań na minutę, bieżącą ocenę jakości (krocząca EWMA), dzienny koszt do tej pory w porównaniu z budżetem oraz 5 najdroższych żądań z ostatniej godziny. Zapewnia to zespołowi szybki wgląd w stan systemu bez konieczności czekania na uruchomienie alertu.
Szybkie sprawdzenie
Sprawdź swoją wiedzę na temat alertów dotyczących metryk aplikacji LLM z tego modułu.
Podsumowanie modułu
W tej lekcji nauczyli się Państwo, że trzy kategorie alertów obejmują systemy LLM — opóźnienie, koszt i jakość — a każda z nich wymaga innego sposobu instrumentacji; średnie kroczące i EWMA lepiej wykrywają stopniowe pogarszanie jakości niż sprawdzanie progów dla pojedynczych żądań, natomiast debouncing alertów zapobiega zmęczeniu alertami, wymagając utrzymywania się warunków przez wiele cykli ewaluacji przed wygenerowaniem alertu. W dalszej części zajmiemy się atakami typu prompt injection i bezpieczeństwem AI.
Często zadawane pytania
Czy lekcja „Alerty dotyczące opóźnień, kosztów i spadku jakości” jest bezpłatna?
Tak — pełny tekst „Alerty dotyczące opóźnień, kosztów i spadku jakości” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Engineering Academy, przejdź na CoddyKit PRO. Kurs AI Engineering Academy zawiera 4 lekcji w sumie.
Co nauczysz się w „Alerty dotyczące opóźnień, kosztów i spadku jakości”?
Zdefiniuj progi alertów dla opóźnienia p99, kosztu pojedynczego żądania i automatycznych ocen jakości oraz kieruj alerty do Slack lub PagerDuty, gdy jakość potoku LLM się pogarsza. Ćwiczysz AI Engineering Academy z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.
Czy potrzebuję doświadczenia, aby zacząć AI Engineering Academy?
Nie wymagamy żadnego doświadczenia. AI Engineering Academy w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 4 z 4.
Ile czasu zajmuje lekcja „Alerty dotyczące opóźnień, kosztów i spadku jakości”?
Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.
Czy mogę pisać i uruchamiać kod w tej lekcji AI Engineering Academy?
Tak. Każda lekcja AI Engineering Academy zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.
Wszystkie lekcje w tym kursie
- Dlaczego aplikacje LLM trudno debugować
- Śledzenie za pomocą LangSmith
- Langfuse do obserwowalności niezależnej od modelu
- Alerty dotyczące opóźnień, kosztów i spadku jakości